当前位置：首页 > news >正文

【Nanobot】README09_LEVEL4 添加新聊天渠道

news 2026/6/5 4:27:40

【Nanobot】README09_LEVEL4 添加新聊天渠道

源码地址：https://github.com/HKUDS/nanobot

🎯 目标

指导如何为 nanobot 添加新的聊天渠道（如 Signal、Matrix、Line 等）。

📋 添加新 Channel 的步骤

步骤 1：创建 Channel 类

文件：nanobot/channels/signal.py

fromnanobot.channels.baseimportBaseChannelfromnanobot.bus.eventsimportInboundMessage,OutboundMessagefromnanobot.bus.queueimportMessageBusfromtypingimportAnyclassSignalChannel(BaseChannel):""" Signal 消息渠道实现。 特性： 1. 接收 Signal 消息 2. 发送消息到 Signal 3. 支持流式输出（可选） """name="signal"display_name="Signal"def__init__(self,config:Any,bus:MessageBus):super().__init__(config,bus)# 初始化 Signal 客户端importsignalbot self.bot=signalbot.Bot(config.phone_number)self._running=Falseasyncdefstart(self)->None:"""启动 Signal Bot。"""self._running=True# 注册消息处理器@self.bot.handler()asyncdefon_message(message):awaitself._handle_message(sender_id=str(message.source),chat_id=str(message.chat_id),content=message.text,media=message.attachmentsor[],)# 启动 Botawaitself.bot.start()asyncdefstop(self)->None:"""停止 Signal Bot。"""self._running=Falseawaitself.bot.stop()asyncdefsend(self,msg:OutboundMessage)->None:"""发送消息到 Signal。"""awaitself.bot.send_message(chat_id=msg.chat_id,text=msg.content,quote_reply_to=msg.reply_to,)

步骤 2：实现流式输出（可选）

如果 Channel 支持编辑消息（如 Telegram），可以实现流式输出：

asyncdefsend_delta(self,chat_id:str,delta:str,metadata:dict[str,Any]|None=None,)->None:""" 交付流式文本块。 实现： 1. 提取 stream_id 2. 第一条消息 → 发送新消息 3. 后续消息 → 编辑已有消息 4. _stream_end → 清理缓冲区 """stream_id=metadata.get("_stream_id")ifmetadataelseNoneifnotstream_id:returnifmetadata.get("_stream_delta"):# 增量内容ifstream_idnotinself._edit_buffers:# 发送第一条消息sent_msg=awaitself.bot.send_message(chat_id=chat_id,text=delta,)self._edit_buffers[stream_id]=sent_msgelse:# 编辑已有消息old_msg=self._edit_buffers[stream_id]new_content=old_msg.text+deltaawaitold_msg.edit(text=new_content)elifmetadata.get("_stream_end"):# 流式结束ifstream_idinself._edit_buffers:delself._edit_buffers[stream_id]

步骤 3：添加配置字段

文件：nanobot/config/schema.py

在ChannelsConfig中添加字段：

classChannelsConfig(Base):telegram:TelegramConfig|None=Nonediscord:DiscordConfig|None=Nonesignal:SignalConfig|None=None# 新增classSignalConfig(Base):enabled:bool=Falsephone_number:str=""allow_from:list[str]=["*"]# 访问控制列表

步骤 4：配置新 Channel

文件：~/.nanobot/config.json

{"channels":{"signal":{"enabled":true,"phoneNumber":"+1234567890","allowFrom":["*"]}}}

步骤 5：测试新 Channel

# 启动 gatewaynanobot gateway# 在 Signal 中发送消息给 Bot# 应该收到回复

🔧 高级：插件化 Channel

如果希望将 Channel 作为独立插件发布（不修改核心代码），可以使用entry_points。

项目结构

nanobot-signal/ ├── nanobot_signal/ │ └── __init__.py └── pyproject.toml

`pyproject.toml`

[project] name = "nanobot-signal" version = "0.1.0" [project.entry-points."nanobot.channels"] signal = "nanobot_signal:SignalChannel" [project.dependencies] nanobot-ai = ">=0.1.0" signalbot = "*"

`nanobot_signal/init.py`

fromnanobot.channels.baseimportBaseChannelfromnanobot.bus.queueimportMessageBusfromtypingimportAnyclassSignalChannel(BaseChannel):name="signal"display_name="Signal"def__init__(self,config:Any,bus:MessageBus):super().__init__(config,bus)# ... 实现代码 ...

安装和使用

# 安装插件pipinstallnanobot-signal# 配置（不需要修改核心代码）cat>~/.nanobot/config.json<<EOF { "channels": { "signal": { "enabled": true } } } EOF# 启动nanobot gateway

📝 检查清单

实现BaseChannel接口
- start()- 启动监听
- send()- 发送消息
- stop()- 停止监听
处理消息
- 接收消息并调用_handle_message()
- 支持媒体文件（图片、文档等）
- 支持音频转录（可选）
实现流式输出（可选）
- send_delta()- 增量更新
- 消息编辑缓冲区
添加配置字段
- 在schema.py中添加配置类
- 在config.json中启用
测试
- 基本对话功能
- 工具调用
- 流式输出（如果支持）
- 访问控制（allow_from）

🎯 完整示例：添加 Email Channel

Email Channel 已经内置，参考实现：

文件：nanobot/channels/email.py

classEmailChannel(BaseChannel):"""电子邮件渠道。"""name="email"display_name="Email"asyncdefstart(self):"""启动 IMAP 监听。"""importimaplib self.imap=imaplib.IMAP4_SSL(self.config.imap_server)self.imap.login(self.config.email,self.config.password)self.imap.select("INBOX")whileself._running:# 轮询新邮件awaitself._check_new_emails()awaitasyncio.sleep(30)asyncdefsend(self,msg:OutboundMessage):"""发送邮件。"""importsmtplibfromemail.messageimportEmailMessage email=EmailMessage()email["From"]=self.config.email email["To"]=msg.chat_id email.set_content(msg.content)withsmtplib.SMTP_SSL(self.config.smtp_server)assmtp:smtp.login(self.config.email,self.config.password)smtp.send_message(email)