02 - 渠道抽象层：一个 Agent 服务九个平台

学习目标

理解 CowAgent 如何通过三层渠道抽象层实现一个 Agent 同时服务 Web、微信、飞书、钉钉、企业微信、QQ、公众号、Telegram 九个平台。

项目实践

三层渠道继承体系

各层职责：

• Channel：定义 startup() 和 send() 接口，与具体平台无关的抽象
• ChatChannel：实现通用消息处理流水线，包含队列管理、线程池调度、上下文构建、回复装饰等
• 具体渠道：实现平台特定的消息接收、格式解析、发送逻辑

消息队列消费模型

# ChatChannel.__init__ — 每个实例独立的消息队列和锁
self.futures = {}          # session_id -> Future 列表
self.sessions = {}         # session_id -> [Dequeue, BoundedSemaphore]
self.lock = threading.Lock()

# 启动消费者线程
_thread = threading.Thread(target=self.consume, daemon=True)
_thread.start()

消费循环：

关键设计：

• 每 session 一个 BoundedSemaphore：控制同一 session 最多 1 条消息在处理中（防止乱序）
• ThreadPoolExecutor(max_workers=8)：最多 8 条消息并行处理
• # 前缀命令优先：管理命令优先放入队列左侧（putleft）

/cancel 命令的队列旁路

_BYPASS_QUEUE_COMMANDS = ("/cancel",)

def produce(self, context):
    stripped = context.content.strip().lower()
    if stripped in self._BYPASS_QUEUE_COMMANDS:
        # 同步处理，不入队列
        self._handle_cancel_command(context, session_id)
        return
    # 正常路径：入队列
    self.sessions[session_id][0].put(context)

/cancel 不能排队——如果它排在正在运行的任务后面，就无法取消那个任务。所以它走同步旁路，直接调用取消注册表。

群聊会话隔离

群聊中 session_id 的确定逻辑：

配置	session_id	效果
group_shared_session = True（默认）	group_id	群内所有用户共享一个会话
group_shared_session = False	actual_user_id	群内每个用户独立会话
用户在 group_chat_in_one_session 白名单	group_id	特定群强制共享

问题与规避

问题	CowAgent 的规避策略
多个渠道消费同一个消息队列	每个渠道实例独立的 self.sessions 和 self.lock
同一 session 消息乱序	BoundedSemaphore(1) 保证同一 session 串行
线程池任务异常未捕获	done_callback 捕获异常并记录日志
群聊中非目标消息触发回复	group_name_white_list + @bot 前缀 + group_at_off 三重过滤
用户发送 Bot 自己的消息	trigger_by_self = False 过滤自触发

设计取舍

为什么用 Dequeue + 轮询而非 asyncio.Queue？

CowAgent 整体使用 threading 而非 asyncio。Dequeue 是一个简单的双端队列，支持 putleft（高优先级）和 put（普通）。轮询（sleep 0.2s）虽然不如事件通知高效，但简单可靠，且在消息量不大的场景下延迟可接受。

为什么 Web console 不走消息队列？

Web console 使用独立的 ChatService.run() 直接调用 Agent，不走 ChatChannel 的队列模型。原因是 Web console 支持流式输出（SSE），需要实时将 Agent 的事件翻译为 CHAT protocol chunks，这与 IM 渠道的”消息入队→异步处理→一次性发送”模型不同。

参考来源

• CowAgent ChatChannel 源码
• CowAgent ChannelFactory 源码