nanobot 深入学习：多渠道接入架构

1. 模块概述

Channel 系统（nanobot/channels/）通过抽象基类 BaseChannel 统一了 15+ 聊天平台的集成接口，由 ChannelManager 协调启动、停止和消息路由。

📍 核心源码：nanobot/nanobot/channels/base.py#L1-L257、nanobot/nanobot/channels/manager.py#L1-L490

1.1 支持的渠道

渠道	模块	流式支持
Telegram	telegram.py	✅ (delta)
Discord	discord.py	✅
Slack	slack.py	✅
微信	weixin.py	❌
飞书	feishu.py	✅ (卡片流式)
钉钉	dingtalk.py	❌
WhatsApp	whatsapp.py	✅ (bridge)
QQ	qq.py	❌
Matrix	matrix.py	❌
Signal	signal.py	❌
企业微信	wecom.py	❌
Email	email.py	❌
MS Teams	msteams.py	❌
MoChat	mochat.py	❌
WebSocket	websocket.py	✅ (WebUI)
NapCat	napcat.py	❌

1.2 在系统中的位置

graph TB
    subgraph "外部平台"
        TG[Telegram API]
        DC[Discord API]
        WX[WeChat API]
    end

    subgraph "Channel 层"
        TGC[TelegramChannel]
        DCC[DiscordChannel]
        WXC[WeChatChannel]
        WS[WebSocketChannel]
    end

    subgraph "管理层"
        CM[ChannelManager]
    end

    subgraph "核心层"
        MB[MessageBus]
        AL[AgentLoop]
    end

    TG <--> TGC
    DC <--> DCC
    WX <--> WXC
    TGC -->|publish_inbound| MB
    DCC -->|publish_inbound| MB
    WXC -->|publish_inbound| MB
    MB -->|consume_outbound| CM
    CM -->|send| TGC
    CM -->|send| DCC
    CM -->|send| WXC
    WS -->|WebSocket| WebUI
    CM -->|start/stop| TGC
    CM -->|start/stop| DCC

2. 命名体系与易混淆函数对比

2.1 命名规律拆解

前缀	操作对象	操作	含义
send_	—	—	发送完整消息（抽象方法）
send_	delta	—	发送流式增量
send_	reasoning_	delta	发送推理增量
send_	reasoning_	end	结束推理流段
send_	reasoning	—	一次性发送完整推理
send_	file_edit_	events	发送文件编辑事件
_handle_	—	message	处理入站消息的统一入口
_send_	once	—	ChannelManager：单次发送
_send_	with_	retry	ChannelManager：带重试发送
_coalesce_	stream_	deltas	ChannelManager：合并流式增量

2.2 易混淆函数对比表

对比维度	send	send_delta
调用场景	完整回复	流式输出的每个文本块
消息标志	无特殊标志	_stream_delta 元数据
是否必须实现	✅ (abstract)	❌ (可选，默认 no-op)
幂等性	渠道自行保证	由 _stream_id 和 _stream_end 保证
一句话区分	send 发送完整消息	send_delta 发送流式片段，需配合 _stream_end 标志

3. API Signatures

graph TB
    subgraph "BaseChannel 抽象接口"
        start["start() [abstract]"]
        stop["stop() [abstract]"]
        send["send(msg) [abstract]"]
        send_delta["send_delta(chat_id, delta)"]
        send_reasoning["send_reasoning(msg)"]
        send_reasoning_delta["send_reasoning_delta(chat_id, delta)"]
        send_reasoning_end["send_reasoning_end(chat_id)"]
        handle_message["_handle_message(sender_id, chat_id, content)"]
        is_allowed["is_allowed(sender_id)"]
    end

    subgraph "ChannelManager"
        start_all["start_all()"]
        stop_all["stop_all()"]
        dispatch_outbound["_dispatch_outbound()"]
        send_with_retry["_send_with_retry()"]
        send_once["_send_once()"]
    end

BaseChannel

class BaseChannel(ABC):
    name: str = "base"                     # 渠道短名
    display_name: str = "Base"             # 显示名
    send_progress: bool = True             # 是否发送进度更新
    send_tool_hints: bool = False          # 是否发送工具提示
    show_reasoning: bool = True            # 是否展示推理内容

    def __init__(self, config, bus: MessageBus):
        # 初始化渠道
        # 场景：ChannelManager._init_channels() 创建

    @abstractmethod
    async def start(self) -> None:
        # 启动渠道，开始监听消息
        # 场景：ChannelManager.start_all()

    @abstractmethod
    async def stop(self) -> None:
        # 停止渠道，清理资源
        # 场景：ChannelManager.stop_all()

    @abstractmethod
    async def send(self, msg: OutboundMessage) -> None:
        # 发送完整消息
        # 场景：ChannelManager._send_once()

    async def send_delta(self, chat_id, delta, metadata=None) -> None:
        # 发送流式增量（默认 no-op）
        # 场景：ChannelManager._send_once() 检测 _stream_delta 标志

    def is_allowed(self, sender_id: str) -> bool:
        # 权限检查：star > allowlist > pairing store > deny
        # 场景：_handle_message() 入口

    async def _handle_message(self, sender_id, chat_id, content, ...):
        # 统一消息处理入口
        # 1. 权限检查（is_allowed）
        # 2. DM 未授权则发送配对码
        # 3. 构建 InboundMessage 发布到 MessageBus

ChannelManager

class ChannelManager:
    def __init__(self, config, bus, *, session_manager=None, ...):
        # 初始化所有启用的渠道
        # 场景：网关启动时创建

    async def start_all(self) -> None:
        # 启动所有渠道和出站分发器
        # 场景：网关启动

    async def stop_all(self) -> None:
        # 停止所有渠道和分发器
        # 场景：网关关闭

4. 数据结构深度解析

4.1 BaseChannel

4.a 结构体存在的理由

BaseChannel 定义了所有渠道必须遵守的契约。如果没有这个抽象基类，每个渠道将自由定义自己的接口，ChannelManager 无法统一管理和路由消息。它同时提供了默认实现（如 send_delta 的 no-op），使渠道只需实现核心方法。

4.b 字段三层分析表

字段	设计动机	反事实	替代方案
send_progress	控制是否向该渠道发送进度更新（如”正在搜索…”）	所有渠道都接收进度更新，某些渠道（如 Email）显示效果差	可以在 ChannelManager 中统一过滤，但渠道级别的控制更灵活
send_tool_hints	控制是否发送工具调用提示（如 read_file("...")）	工具提示对所有渠道可见，可能泄露文件路径	可以用全局配置，但不同渠道的隐私需求不同
show_reasoning	控制是否展示模型推理过程	推理内容对所有渠道可见，增加噪音	可以在消息元数据中标记，但渠道级别的开关更简洁
supports_streaming	动态属性：config 启用 streaming AND 实现了 send_delta	需要单独维护一个 boolean 配置	动态计算确保配置和实现一致性

4.2 ChannelManager

4.a 关键设计

ChannelManager 的出站分发器 _dispatch_outbound 是一个独立的 asyncio 任务，持续从 MessageBus 的出站队列消费消息：

1. 优先级路由：reasoning 消息优先处理（有独立的渲染路径）
2. 进度过滤：根据渠道的 send_progress / send_tool_hints 过滤
3. 流式合并：_coalesce_stream_deltas 将连续的 _stream_delta 合并
4. 重复抑制：通过内容指纹避免重复发送相同回复
5. 重试策略：最多 3 次指数退避（1s, 2s, 4s）

5. 函数逐行精讲

5.1 BaseChannel._handle_message() — 消息处理入口

async def _handle_message(self, sender_id, chat_id, content, media=None, metadata=None, ...):
    # → 所有渠道子类调用此方法将外部消息转发到 MessageBus
    if not self.is_allowed(sender_id):
        # → 权限检查：star > allowlist > pairing store
        if is_dm:
            code = generate_code(self.name, str(sender_id))
            # → 生成配对码
            await self.send(OutboundMessage(..., content=format_pairing_reply(code), ...))
            # → 在 DM 中发送配对码（而非直接拒绝）
            # → 这样未授权的用户可以通过配对码获得访问权限
        else:
            self.logger.warning("Access denied for sender %s...", sender_id)
            # → 群聊中直接忽略，不发送配对码
        return
        # → 权限不足，不发布到 MessageBus

    meta = metadata or {}
    if self.supports_streaming:
        meta = {**meta, "_wants_stream": True}
        # → 标记此消息期望流式回复

    msg = InboundMessage(
        channel=self.name,
        sender_id=str(sender_id),
        chat_id=str(chat_id),
        content=content,
        media=media or [],
        metadata=meta,
        session_key_override=session_key,
    )
    await self.bus.publish_inbound(msg)
    # → 发布到 MessageBus，AgentLoop 将异步处理

5.2 ChannelManager._send_once() — 单次发送

@staticmethod
async def _send_once(channel, msg):
    # → 根据消息元数据选择不同的发送方法
    if msg.metadata.get("_reasoning_end"):
        await channel.send_reasoning_end(msg.chat_id, msg.metadata)
        # → 推理流段结束
    elif msg.metadata.get("_reasoning_delta"):
        await channel.send_reasoning_delta(msg.chat_id, msg.content, msg.metadata)
        # → 推理流式增量
    elif msg.metadata.get("_reasoning"):
        await channel.send_reasoning(msg)
        # → 一次性推理（向后兼容）
    elif msg.metadata.get("_file_edit_events"):
        await channel.send_file_edit_events(msg.chat_id, edits, msg.metadata)
        # → 文件编辑事件
    elif msg.metadata.get("_stream_delta") or msg.metadata.get("_stream_end"):
        await channel.send_delta(msg.chat_id, msg.content, msg.metadata)
        # → 流式文本增量
    elif not msg.metadata.get("_streamed"):
        await channel.send(msg)
        # → 完整消息

6. 设计决策分析

6.1 为什么流式消息使用元数据标志而不是独立的消息类型？

决策：使用 _stream_delta、_stream_end、_reasoning_delta 等元数据标志区分消息类型，而不是创建多个 OutboundMessage 子类。

权衡：

• ✅ 所有消息走同一条 MessageBus 管道，简化架构
• ✅ 渠道可以选择性地处理（或不处理）不同类型的消息
• ✅ 新增消息类型不需要修改 MessageBus 接口
• ❌ 元数据字典是弱类型的，拼写错误只能在运行时发现
• ❌ 消息类型的语义隐藏在字符串键中，不如类型系统直观

6.2 为什么配对码机制设计在 Channel 层而不是 Agent 层？

决策：DM 中的未授权用户收到配对码而不是直接被拒绝，配对码存储在 pairing/store.py。

权衡：

• ✅ 用户可以通过任何渠道自行授权，无需修改配置文件
• ✅ 配对码有有效期限制，安全性可控
• ✅ 不需要 Agent 参与授权决策，减少 LLM 调用
• ❌ 配对码存储在本地文件中，多实例部署时需要共享存储

7. 学习检查点

📝 本章小结

1. BaseChannel 是抽象基类，定义 start/stop/send 三个必须实现的抽象方法
2. 流式支持是可选的，渠道实现 send_delta 并在 config 中启用 streaming 即可
3. ChannelManager 的出站分发器是独立 asyncio 任务，负责消息路由、流式合并和重试
4. 配对码机制允许用户自助授权，无需修改配置文件
5. 权限检查链：*（全允许）> allowlist > pairing store > deny

🤔 思考题

1. 为什么 _coalesce_stream_deltas 只在遇到第一个非匹配消息时就停止合并，而不是继续扫描队列中的后续匹配消息？

   参考答案

   这是为了保持消息的时序完整性（nanobot/nanobot/channels/manager.py#L407-L431）。如果跳过非匹配消息继续合并后续的流式增量，会导致消息乱序——中间的非流式消息（如 _stream_end 标记另一个流段结束）被延迟处理。只合并连续的 delta 确保了流的原子性：一个流段的所有 delta 合并后，队列中的下一条消息（无论类型）按原始顺序处理。

2. send_reasoning 默认实现为什么通过 send_reasoning_delta + send_reasoning_end 实现，而不是直接调用 send？

   参考答案

   这是为了保持单一渲染路径（nanobot/nanobot/channels/base.py#L157-L174）。无论推理内容是流式到达（通过 delta）还是一次性到达（通过 send_reasoning），最终都通过 delta/end 对渲染。渠道只需要覆盖 send_reasoning_delta 和 send_reasoning_end 就能同时支持两种模式，无需额外实现 send_reasoning。