learnNote

主题

Day 2 — Channel 设计文档 ​

设计目标：通过抽象基类定义统一的渠道接口，让系统能无感知地支持多种接入方式（CLI、WebChat、Webhook），且新增渠道不需要修改任何已有代码。

一、核心接口设计 ​

1.1 Channel 抽象基类 ​

方法签名说明
startasync start()启动渠道
stopasync stop()停止渠道
receiveasync receive() -> AsyncIterator[GatewayMessage]持续接收消息的异步迭代器
sendasync send(message: GatewayMessage)发送消息到渠道

为什么用 ABC（抽象基类）？

  • 强制所有渠道实现这四个方法，避免遗漏
  • 上层代码面向 Channel 接口编程，不依赖具体实现
  • 新增渠道只需 class NewChannel(Channel) 并实现四个方法

1.2 CLIChannel ​

属性/方法说明
user_idCLI 用户标识，默认 "cli_user"
receive()用 asyncio.get_event_loop().run_in_executor() 在线程池中读取 stdin，避免阻塞事件循环
send()直接 print() 到 stdout

关键设计：标准输入 input() 是阻塞的，在 asyncio 中直接调用会卡住整个事件循环。解决方案是放到线程池执行器中。

1.3 WebChatChannel ​

端点方法说明
/chatPOST接收 JSON {"message": "...", "user_id": "..."}
/wsWebSocket全双工实时通信
/healthGET健康检查，返回 {"status": "ok"}

内部架构：

为什么用 Queue？

  • 生产者（HTTP handler、WebSocket handler）和消费者（receive()）解耦
  • 多个生产者可以同时往 Queue 写入，线程安全
  • Queue 有背压机制——满了会自动阻塞生产者，防止内存溢出

1.4 WebhookChannel ​

配置说明
parsersDict[str, Callable] 平台名 → 解析函数
路由POST /webhook/{platform} → 查找对应 parser

二、关键流程图 ​

消息接收流程（以 WebChatChannel 为例） ​

三、设计决策与权衡 ​

决策 1：receive() 使用 async generator 而非回调 ​

方案优点缺点
async generator（选择）调用方用 async for 自然消费，代码直观需要理解生成器概念
回调函数传统模式，很多人熟悉回调地狱，错误处理复杂
返回消息列表最简单不适合持续接收的场景

决策 2：CLI 用线程池读取输入 ​

input() 是阻塞调用。在 asyncio 中有三种处理方式：

方案优点缺点
run_in_executor（选择）最简单，不阻塞事件循环多了一个线程
aioconsole 库专业的异步控制台库多一个依赖
select/poll无额外线程跨平台兼容性差

决策 3：WebhookChannel 用可插拔解析器 ​

不同平台的 Webhook 数据格式完全不同（飞书 vs 微信 vs Slack），硬编码解析逻辑会导致代码膨胀。

用 register_parser(platform, parser_func) 的方式，每个平台的解析逻辑独立在一个函数里——新增平台只需注册新的解析函数。

四、与前序章节的集成点 ​

  • 依赖 Day 1：所有 Channel 的输出都是 GatewayMessage 格式
  • 被 Day 3 依赖：Session 管理需要从 Channel 获取 user_id 和 channel 名称来路由会话

五、与真实生产系统的对比 ​

维度miniOpenClaw生产级实现
CLI仅用于调试同样仅用于调试和管理
Webaiohttp 单进程Nginx + Gunicorn + FastAPI 多 worker
Webhook简单路由签名验证、重放攻击防护、幂等处理
队列asyncio.QueueRedis Queue / RabbitMQ
消息格式转换简单 JSON 解析完整的 Schema 校验 + 版本适配