01 - 架构设计：从聊天机器人到超级 Agent

学习目标

理解 CowAgent 如何从原始的聊天机器人架构演进为支持多 session 隔离的超级 Agent，以及 Bridge 模式在其中扮演的关键角色。

前置知识

本章涉及 Agent 核心循环与架构模式，建议先阅读 Agent 核心循环设计。

项目实践

架构演进

CowAgent 经历了从 chatgpt-on-wechat（COW）到超级 Agent 的架构升级：

维度	COW（原始架构）	CowAgent（新架构）
核心能力	一问一答的 LLM 聊天	多轮推理 + 工具调用 + 规划 + 记忆
会话模型	全局单 session	多 session 隔离（per-user）
启动方式	直接启动	ChannelManager 多线程并发启动多渠道
模型接入	Bridge 直接调用 bot	AgentBridge → AgentLLMModel → Bot 三层
扩展能力	插件系统	插件 + 工具 + Skills 三层扩展

Bridge 单例模式

Bridge 是整个系统的核心路由器，使用单例模式确保全局唯一：

@singleton
class Bridge(object):
    def __init__(self):
        self.btype = {
            "chat": const.OPENAI,          # 聊天模型
            "voice_to_text": "...",        # ASR 提供商
            "text_to_voice": "...",        # TTS 提供商
            "translate": "...",            # 翻译提供商
        }
        self.bots = {}                      # 缓存各类型的 bot 实例
        self._agent_bridge = None           # 懒加载 AgentBridge

Bridge 负责四大能力的路由：聊天、语音转文字、文字转语音、翻译。每种能力的 bot 实例按需创建并缓存。

AgentBridge 懒加载

Agent 相关的桥接通过 get_agent_bridge() 懒加载：

def get_agent_bridge(self):
    if self._agent_bridge is None:
        from bridge.agent_bridge import AgentBridge
        self._agent_bridge = AgentBridge(self)
    return self._agent_bridge

这种设计确保了非 Agent 模式（传统聊天模式）的用户不会加载 Agent 相关的重量级组件。

ChannelManager 多线程启动

CowAgent 使用 ChannelManager 管理多渠道并发运行：

关键设计点：

1. Web 优先启动：Web console 先启动，其余渠道 0.1s 间隔依次启动，避免日志交错
2. MCP 预热：在渠道启动前后台加载 MCP 工具，减少首条消息延迟
3. 调度器预热：Eager-init AgentBridge 确保 cron 任务在第一条用户消息到达前就可运行

多 Session 隔离

AgentBridge 维护 agents: dict[session_id -> Agent] 映射：

• 每个用户（session_id）拥有独立的 Agent 实例
• 独立的 messages（对话历史）、skill_manager、memory_manager
• 通过 _init_agent_for_session() 按需创建
• clear_session() 支持清理单个用户会话

问题与规避

问题	CowAgent 的规避策略
Bridge 单例在 reload 时状态残留	reset_bot() 重新初始化全部 bots
AgentBridge 首次加载阻塞首条消息	懒加载 + MCP 预热 + 调度器预热提前初始化
多渠道启动顺序导致日志混乱	Web 先启动 + 0.1s 间隔
session 过多导致内存膨胀	clear_session() 支持按需清理

设计取舍

为什么使用 Bridge 单例而非 DI 容器？

COW 项目定位为轻量级、易部署的个人项目。单例模式简单直接，无需额外的 DI 框架。代价是测试隔离性较差（需要手动重置单例状态）。

为什么不直接使用 asyncio 重写全部？

CowAgent 的多渠道 SDK（微信、飞书、钉钉等）大多是同步阻塞式 API。如果全面转向 asyncio，需要为每个渠道做 async 适配层，工程量大且维护成本高。使用 threading + ThreadPoolExecutor 是务实的选择。

参考来源

• CowAgent Bridge 源码
• CowAgent AgentBridge 源码
• CowAgent app.py 入口源码
• Agent 核心循环设计