AI Agent Engineer

AI Agent Engineer

Channel 与 Gateway

5 min read

Channel 与 Gateway

对应源码

主题源码
Channel 抽象契约nanobot/channels/base.py::BaseChannel
发现、注册和生命周期nanobot/channels/registry.py、nanobot/channels/manager.py::ChannelManager
Pairing 与访问控制nanobot/pairing/store.py
Gateway 启动和 runtimenanobot/cli/commands.py::_run_gateway、nanobot/gateway/runtime.py、nanobot/gateway/service.py
流式和 reasoning 转换nanobot/channels/manager.py、具体 Channel 实现

学习目标

读完后，你应当能够：

  • 解释 BaseChannel 的最小契约；
  • 说明内置和外部 Channel 如何被发现；
  • 解释 allowFrom、Pairing 和入站发布顺序；
  • 说明 ChannelManager 如何启动、停止和投递 outbound；
  • 理解 Gateway 为什么是长期运行的组装入口。

前置知识

  • 理解 MessageBus；
  • 理解 Inbound/OutboundMessage；
  • 已阅读 Cron 与 Heartbeat。

1. BaseChannel 是适配器契约

位置：

  • nanobot/channels/base.py::BaseChannel

必须实现：

  • start()
  • stop()
  • send(OutboundMessage)

可选覆盖：

  • login()
  • send_delta()
  • send_reasoning_delta() / send_reasoning_end()
  • send_file_edit_events()
  • default_config()

start() 应当是长期运行任务：连接平台、接收消息，并通过 _handle_message() 交给通用入口。

2. 入站消息先过访问控制

BaseChannel._handle_message()：

flowchart TD
    M["平台消息"] --> A{"sender allowed?"}
    A -->|否，DM| P["生成 Pairing code<br/>直接发送说明"]
    A -->|否，非 DM| D["拒绝并记录日志"]
    A -->|是| S{"Channel 支持 streaming?"}
    S -->|是| W["metadata 加 _wants_stream"]
    S -->|否| I["构造 InboundMessage"]
    W --> I
    I --> B["bus.publish_inbound"]

is_allowed() 优先级：

  1. allowFrom 包含 *；
  2. sender ID 与 allowlist 精确匹配；
  3. Pairing store 已批准；
  4. 否则拒绝。

安全含义：

  • allowFrom 条目是 opaque token，精确匹配；
  • 未配置 allowFrom 时可能进入 pairing-only 模式；
  • DM 可以返回配对码；
  • 群聊不应把配对码公开发出。

3. Streaming 能力是配置与实现的交集

supports_streaming 只有在以下两个条件同时满足时为真：

  • Channel 配置开启 streaming；
  • 子类真正覆盖 send_delta()。

仅在配置中写 streaming: true 不会自动获得流式能力。

有状态 Channel 应使用 _stream_id 区分 segment，不能只按 chat_id 累积。

4. Channel 发现

内置 Channel：

  • 使用 pkgutil.iter_modules 列出 nanobot.channels 下模块；
  • 只导入启用项，减少可选依赖失败。

外部 Channel：

  • Python entry point group nanobot.channels。

合并规则：

  • 内置实现优先；
  • 插件不能静默覆盖同名内置 Channel。

相关源码：

  • nanobot/channels/registry.py

5. ChannelManager 初始化

ChannelManager._init_channels()：

  1. 发现候选名称；
  2. 合并 Pydantic extra config 中的插件 key；
  3. 找出 enabled Channel；
  4. 只导入这些类；
  5. 对 WebSocket 构造额外 Gateway services；
  6. 解析 progress/tool-hint/reasoning override；
  7. 验证 allowFrom；
  8. 保存实例。

某个可选 Channel 初始化失败时，会记录 warning，而不是让所有 Channel 都无法启动。

6. 启动和停止

start_all()：

  • 创建 outbound dispatcher Task；
  • 为每个 Channel 创建长期 Task；
  • gather 等待；
  • Channel 的 start() 正常情况下不应立即返回。

stop_all()：

  • 先取消 dispatcher；
  • 等待其取消；
  • 逐个调用 Channel stop()；
  • 尊重真实 Task cancellation。

7. Outbound Dispatcher

_dispatch_outbound() 持续消费 MessageBus outbound，根据 metadata 分流：

  • 普通完整消息；
  • content delta；
  • stream end；
  • reasoning delta/end；
  • file edit event；
  • progress/tool hint。

它还负责：

  • 选择目标 Channel；
  • retry；
  • delta coalescing；
  • 重复消息抑制；
  • 发送失败缓冲和日志。

Channel send() 应在失败时抛出异常，让 Manager 统一应用 retry，而不是静默吞掉。

8. 发送重试

ChannelManager 集中处理 outbound delivery retry。当前配置：

  • channels.sendMaxRetries

具体等待序列和实际次数以 _send_with_retry() 为准。

这类 retry 与 Provider retry 不同：

Provider retry：模型请求失败
Channel retry：最终消息投递失败

9. Gateway 组装

nanobot/cli/commands.py::_run_gateway 创建：

  • MessageBus；
  • RuntimeEventBus；
  • Provider snapshot；
  • SessionManager；
  • CronService；
  • AgentLoop；
  • WebuiTurnCoordinator；
  • ChannelManager。

然后启动 AgentLoop、Cron、Channel 和可选 health server。

Gateway 不是新的 Agent；它是让这些长期组件共享一个 lifecycle 的进程入口。

10. 推荐调试顺序

如果某个 Channel 不回复：

  1. nanobot agent -m 是否成功；
  2. Channel 是否 enabled；
  3. 可选依赖是否导入；
  4. start() 是否仍在运行；
  5. sender 是否通过 allowFrom/Pairing；
  6. _handle_message() 是否发布 inbound；
  7. AgentLoop 是否消费；
  8. outbound 是否到 dispatcher；
  9. send() 是否抛错并 retry；
  10. 平台 API 是否接受消息格式。

源码与测试

  • nanobot/channels/base.py
  • nanobot/channels/registry.py
  • nanobot/channels/manager.py
  • nanobot/pairing/store.py
  • nanobot/cli/commands.py::_run_gateway
  • tests/channels/test_base_channel.py
  • tests/channels/test_channel_plugins.py
  • tests/channels/test_channel_manager_delta_coalescing.py
  • tests/channels/test_channel_manager_reasoning.py
  • tests/pairing/

自检

  1. 为什么 _handle_message() 不应绕过访问控制直接 publish？
  2. 为什么只导入 enabled Channel？
  3. 配置开启 streaming 为什么还不够？
  4. Provider retry 和 Channel retry 为什么必须分开？
  5. Gateway 为什么是组装入口而不是另一套 Agent？

下一步

阅读 HTTP API 与 Python SDK。

Backlinks