多 Agent 通信与唤醒机制

发表于 2026-05-20 更新于 2026-07-07 分类于 AI

多 Agent 通信与唤醒机制

多 Agent 系统的难点不在「能跑多个 Agent」，而在异步协作时的消息投递与唤醒时机——何时只排队通知、何时触发目标 Agent 开新 turn、子 Agent 完成后如何回传父 Agent。下面分别梳理 Codex（Cursor Agent Runtime）与 AWorld 的实现，并做对照。

结论

维度	Codex（Cursor Agent Runtime）	AWorld
调度载体	`AgentControl + Session(InputQueue) + Task Lifecycle`	`TaskEventRunner + EventManager + InMemoryEventbus`
核心控制位	`InterAgentCommunication.trigger_turn`	`Message.category/receiver` + Swarm 拓扑 Handler
消息缓存	`InputQueue.mailbox_pending_mails`	`PriorityQueue`（按 task_id）+ Memory `pending` 类型
唤醒语义	显式区分 `send_message`（不唤醒）与 `followup_task`（唤醒）	显式 `AgentMessage` 入队即唤醒；或 `pending` 内存延迟投递
完成回传	Completion Watcher → 父 mailbox（`trigger_turn=false`）	Team/Handoff Handler → 向 root/caller 发 `AgentMessage`

Codex：Mailbox + trigger_turn

Codex 的子 Agent 唤醒不是独立调度器进程，而是由运行时三件套协同完成：

AgentControl：解析目标 Agent、路由消息、维护父子 thread 关系
Session / InputQueue：接收并缓存 Agent 间消息（mailbox），发布 mailbox activity
Task Lifecycle：目标 Agent 空闲且存在 trigger_turn=true 的邮件时，创建新 turn 并启动执行
Completion Watcher：监听子 Agent 终态（Completed / Errored / Shutdown），把完成消息回写父 Agent mailbox

通信协议（InterAgentCommunication envelope）

字段	含义
`author`	发送方 `AgentPath`
`recipient`	接收方 `AgentPath`
`content`	消息正文
`trigger_turn`	是否触发目标 Agent 开新 turn

语义区分

send_message：trigger_turn=false，仅入队，不主动唤醒
followup_task：trigger_turn=true，目标 idle 时触发新 turn
目标忙碌时不并发开第二个 turn，消息在 turn 边界点投递

唤醒流程

父 Agent 发消息（工具 → AgentControl）
  → 目标 Session 写入 InputQueue.mailbox_pending_mails
  → 若 trigger_turn=true 且目标无 active_turn
     → maybe_start_turn_for_pending_work → 创建 regular turn
  → 模型处理 mailbox 消息并产出结果

完成通知回传

子 Agent 到达终态 → Completion Watcher 构造 FINAL_ANSWER envelope
通过 InterAgentCommunication(trigger_turn=false) 写回父 thread mailbox
父 Agent 通过 wait_agent 或后续 turn 读取
默认不强制打断父 Agent 当前执行

设计价值：显式区分「通知排队」与「任务唤醒」；父子异步协作、避免无谓抢占；completion 统一走 mailbox，链路可追踪。

AWorld：Event Bus + Swarm Handler

AWorld 的多 Agent 协作建立在 Event-Driven Runtime 之上，没有单独的 mailbox 进程，而是由 TaskEventRunner 的事件循环统一调度。

架构分工

flowchart TB
    subgraph Producer["消息生产者"]
        Parent["父 Agent policy"]
        Tool["Tool Handler"]
        BG["BackgroundTask Handler"]
        Runner["TaskEventRunner\ninit AgentMessage"]
    end
    subgraph Bus["EventManager + InMemoryEventbus"]
        Queue["PriorityQueue\n按 task_id 隔离"]
    end
    subgraph Consumer["消息消费者"]
        RunnerLoop["TaskEventRunner._do_run\nconsume → _common_process"]
        AgentH["DefaultAgentHandler\n路由 Agent / Tool"]
        AgentRun["agent.run / async_run\n注册为 category=agent 的 handler"]
    end
    Parent -->|"AgentMessage\nreceiver=子Agent"| Queue
    Tool -->|"AgentMessage / ToolMessage"| Queue
    BG -->|"trigger_agent=true → AgentMessage\n否则 → Memory pending"| Queue
    Runner --> Queue
    Queue --> RunnerLoop --> AgentH --> AgentRun

EventManager：封装 emit_message / consume，消息持久化到 InmemoryStorage 便于重放与观测
InMemoryEventbus：按 task_id 维护 PriorityQueue，publish 入队、consume 阻塞取消息
TaskEventRunner：主循环 while True: message = await consume() → _common_process()，注册各 Agent 的 run 为 handler
DefaultAgentHandler：根据 Swarm 拓扑（Team / Handoff / Workflow）决定下一步向谁发 AgentMessage

Message 协议（核心 envelope）

@dataclass
class Message:
    session_id: str
    payload: Any          # Observation | List[ActionModel] | TaskResponse ...
    sender: str           # 发送方 agent/tool id
    receiver: str         # 接收方（p2p 路由）
    caller: str           # 上一跳调用方（handoff 回传链）
    category: str         # "agent" | "tool" | "task" | "background_task" ...
    topic: str            # FINISHED | RERUN | GROUP_ACTIONS ...
    headers: Dict         # context / agent_as_tool / trigger_agent ...

路由规则：get_handlers(category) → 按 message.receiver 或 message.topic 匹配已注册的 Agent / Tool handler。

三种 Swarm 拓扑的唤醒模式

拓扑	类	唤醒方式	完成回传
Coordination（Team）	`TeamSwarm`	Orchestrator 通过 `handoffs` 把子 Agent 当工具调用；`_agent()` 向 `receiver=子Agent` 发 `AgentMessage(Observation)`	子 Agent `finished` → `_team_stop_check` 向 `root_agent` 发 Observation
Collaborative（Handoff）	`HandoffSwarm`	Agent 决策 handoff → `_agent()` 唤醒目标 Agent	执行完成 → `_handoff_stop_check` 向 `caller` 或 `communicate_agent` 回传
Workflow	`WorkflowSwarm`	前驱 Agent 完成 → `_workflow_stop_check` 检查 DAG 前驱全部 finished 后唤醒后继	无后继且 swarm finished → `TopicType.FINISHED`

Team 模式关键代码路径（DefaultAgentHandler._agent）：

cur_agent._finished = False
observation = Observation(content=con, observer=agent.id(), from_agent_name=agent.id())
yield Message(
    category=Constants.AGENT,
    payload=observation,
    sender=action.agent_name,
    receiver=action.tool_name,   # 目标子 Agent
    headers=headers,
)

两种子 Agent 执行模式

模式	触发条件	行为	类比 Codex
同步 sub_task	`wait_tool_result=True`（Team 默认）	`exec_agent(sub_task=True)` 阻塞等待子 Task 完成，结果 inline 返回父 Agent	类似父 Agent 同步 `wait_agent`
异步 Event 唤醒	Handoff / Workflow / 普通 AgentMessage	消息入 Event Queue，Runner 循环 dispatch	类似 `followup_task(trigger_turn=true)`

Background Task 的延迟唤醒

后台任务完成时，DefaultBackgroundTaskHandler 支持两种合并策略：

headers["trigger_agent"]=True：Hot-Merge + 立即唤醒，直接 send_message(AgentMessage(receiver=agent_id))
默认：结果写入 Memory，memory_type="pending"；目标 Agent 在 async_post_run / async_policy 开头检测 pending 项，hold 住 finished 状态，下一轮 turn 加载 pending 消息后继续

这与 Codex 的 send_message(trigger_turn=false) + 后续 turn 读取 mailbox 语义接近。

pending 消息加载（llm_agent.async_policy）

pending_filters = self._build_memory_filters(message.context)
pending_filters['memory_type'] = 'pending'
pending_items = memory.memory_store.get_all(pending_filters)
if pending_items:
    # 转为 message 类型，注入当前 turn 上下文

对照与设计启示

共同点

都区分「仅通知」与「主动唤醒」两种语义
子 Agent 完成结果都通过统一消息通道回传父 Agent，而非共享可变状态
目标 Agent 忙碌时避免并发 turn（Codex 显式 active_turn 检查；AWorld 单 Task 单 Runner 循环顺序消费）

差异点

点	Codex	AWorld
控制粒度	单一 `trigger_turn` 布尔位	多种机制：`AgentMessage` 入队、`trigger_agent` header、`pending` Memory
拓扑表达	Thread 树 + AgentPath	Swarm Graph（Team / Handoff / Workflow）+ `handoffs` 列表
同步等待	`wait_agent` 工具	`wait_tool_result` + `exec_agent(sub_task=True)`
可观测性	mailbox + thread 路径	EventManager 存储 + Trajectory + Trace

AWorld 计划方向

完善 Runtime 模块间 Event 机制——当前 Background Task 的 trigger_agent 与 Memory pending 双路径可进一步统一为显式 trigger_turn 语义，与 Codex 对齐后可降低多 Agent 协作的心智负担。