第 23 章：多智能体协作

发表于 更新于 分类于 阅读次数：
MultiAgentManager 多 Agent 管理与 ACP 跨进程通信协议——内部 Agent 间的消息传递、分布式 Agent 的发现与连接、共享工作空间的协作模式与任务分发机制。
个人公众号
1
2
卷四：纵深
[20] 配置 -> [21] 记忆 -> [22] 自治任务 -> [23] 多智能体协作 <- you are here

一个 Agent 不够用时怎么办？让多个 Agent 协作。这一章看 MultiAgentManager 怎么管理多个 Agent、Agent 之间怎么通信、ACP 协议怎么连接外部 Agent。

三种协作模式

1
2
3
4
5
6
7
8
9
10
11
模式一：内部 Agent 间通信
  chat_with_agent() / submit_to_agent()
  通过 HTTP API 在两个 QwenPaw Agent 之间传递消息

模式二：外部 Agent 委托
  delegate_external_agent()
  通过 ACP 协议连接外部 Agent（如 Claude Code、Cursor）

模式三：层级委托
  Mission Phase 2 的 worker-verifier 模式
  主 Agent 派遣 worker 执行子任务

MultiAgentManager——多 Agent 的调度中心

MultiAgentManager 管理所有 Agent 工作区的生命周期：

1
2
3
4
5
6
7
8
9
class MultiAgentManager:
    agents: Dict[str, Workspace]  # agent_id -> Workspace

    async def start_all_configured_agents(self):
        # 真正的并行启动！
        results = await asyncio.gather(*[
            self.get_agent(agent_id)
            for agent_id in enabled_agents
        ])

关键设计：并行加载——多个 Agent 同时初始化，不串行等待。get_agent() 使用”抢占或等待”模式——第一个请求者启动 Agent，后续请求者等待同一个 Event。

零停机重载

1
2
3
4
5
6
7
8
async def reload_agent(self, agent_id):
    # 1. 创建新 Workspace
    new_ws = await self._create_workspace(agent_id)
    # 2. 原子替换
    old_ws = self.agents[agent_id]
    self.agents[agent_id] = new_ws
    # 3. 优雅停止旧实例（延迟清理，等待活跃任务）
    await self._delayed_cleanup(old_ws)

蓝绿部署模式应用在单个 Agent 上——新版本就绪后再替换，旧版本等活跃任务完成后清理。

内部 Agent 通信——HTTP API

Agent 之间通过 QwenPaw 的 HTTP API 通信：

同步通信（等待回复）：

1
2
3
4
5
async def chat_with_agent(to_agent, text, timeout=300):
    # POST /api/agent/process
    # SSE 流式读取，等待最终回复
    response = await collect_final_response(...)
    return ToolResponse(content=[TextBlock(text=response)])

异步通信（后台执行）：

1
2
3
4
5
6
7
8
9
async def submit_to_agent(to_agent, text):
    # POST /api/agent/process/task
    task_id = await submit_background_task(...)
    return ToolResponse(content=[TextBlock(text=f"Task {task_id} submitted")])

async def check_agent_task(task_id):
    # GET /api/agent/process/task/{task_id}
    status = await poll_status(task_id)
    return ToolResponse(content=[TextBlock(text=f"Status: {status}")])

消息格式自动加前缀 [Agent {caller_id} requesting]——接收 Agent 知道消息来自另一个 Agent。

ACP——连接外部 Agent

ACP（Agent Communication Protocol）是连接外部 Agent 的标准协议。QwenPaw 既是 ACP 客户端（委托外部 Agent）也是 ACP 服务端（被其他系统调用）。

作为客户端——委托外部 Agent

1
2
3
4
5
async def delegate_external_agent(
    action: str,       # "start" / "message" / "respond" / "close"
    runner: str,       # 外部 Agent 名称（如 "claude_code"）
    message: str,
):

生命周期：

1
2
3
4
5
1. action="start"   -> 启动外部 Agent 进程（stdio JSON-RPC）
2. action="message" -> 发送消息，接收流式回复
3. 权限请求？       -> 暂停，返回权限选项给用户
4. action="respond" -> 用户选择后继续
5. action="close"   -> 关闭会话

预配置的外部 Agent：opencode、qwen_code、claude_code、codex。

作为服务端——被其他系统调用

QwenPawACPAgent 暴露 QwenPaw 为 ACP 兼容的 Agent，通过 stdio JSON-RPC 接收请求。

协作编排模式

顺序模式

1
2
3
Agent A: chat_with_agent("agent-b", "分析数据")
  -> 等待 Agent B 完成
Agent A: chat_with_agent("agent-c", "基于分析结果生成报告")

并行模式

1
2
3
4
5
Agent A: submit_to_agent("agent-b", "任务1") -> task_1
Agent A: submit_to_agent("agent-c", "任务2") -> task_2
Agent A: check_agent_task(task_1)
Agent A: check_agent_task(task_2)
Agent A: 汇总两个结果

层级模式（Mission）

1
2
3
4
主 Agent（控制器）
  -> submit_to_agent("worker", "实现功能 X")
  -> submit_to_agent("worker", "实现功能 Y")
  -> 验证结果

工程权衡

为什么用 HTTP API 而非直接调用？

Agent 运行在不同的工作区、有不同的配置和记忆。HTTP API 保持了 Agent 之间的隔离——每个 Agent 有自己的 Runner、Session、Provider。直接调用会破坏这种隔离。

为什么 ACP 用 stdio 而非网络？

外部 Agent（如 Claude Code）是独立的进程。stdio 管道简单、安全、不需要端口管理。JSON-RPC over stdio 是轻量级的进程间通信。

自检

  • 知道 MultiAgentManager 并行加载多个 Agent
  • 知道内部 Agent 通过 HTTP API 通信
  • 知道 ACP 协议连接外部 Agent（如 Claude Code）
  • 知道三种编排模式：顺序、并行、层级

下一章我们看插件系统——Plugin 和 Skill 有什么区别？