agent-Specialization/docs/multi_agent_mode/04_message_routing.md

多智能体模式消息路由机制

消息路由由接收方决定：根据接收方当前状态，选择 inline 插入工具结果后，或作为 user 消息触发新一轮任务。
参考现有后台任务通知池机制实现，保证消息不丢失、不错乱、可批量处理。

---

1. 核心思想

1.1 接收方决定插入方式

当一条消息需要发送给某个子智能体（或主智能体）时，路由层只关心接收方当前处于什么状态：

接收方状态	插入方式	效果
正在运行中，且当前在某次工具调用中阻塞等待	把消息作为该工具调用的结果返回	子智能体在当前轮次继续执行，立即看到消息
正在运行中，但不在阻塞等待状态	把消息 inline 插入到当前对话上下文末尾	子智能体下一轮输出时自然看到
空闲状态（本轮已自然结束）	把消息作为普通 user 消息插入	触发子智能体新一轮运行

1.2 参考现有后台通知池

现有后台任务通知机制：

• 运行期间（inline）：server/chat_flow_tool_loop.py 中 execute_tool_calls 末尾调用 process_sub_agent_updates(..., inline=True)，把已完成的子智能体任务一次性插入当前 messages，不触发新任务。
• 停止输出后（polling）：server/chat_flow_task_main.py 中 handle_task_with_sender 结尾启动 run_completion_poll，统一轮询子智能体 + 后台命令完成通知，批量插入 user 消息，只触发一个后续任务。

多智能体模式的消息路由借鉴此机制：

• 每个子智能体有自己的「待处理消息队列」。
• 子智能体每次进入可接收消息的状态时，从队列中取出所有消息，按顺序处理。
• 避免逐条消息触发多次「工作 → 停止 → 再工作」循环。

---

2. 消息路由状态机

2.1 子智能体状态

          ┌─────────────┐
          │   idle      │ 空闲
          └──────┬──────┘
                 │ create_sub_agent / send_message_to_sub_agent
                 ▼
          ┌─────────────┐
          │  running    │ 运行中
          │  (normal)   │ 正常输出
          └──────┬──────┘
                 │ 调用 ask_master / ask_other_agent
                 ▼
          ┌─────────────┐
          │  waiting    │ 阻塞等待回答
          │  (asking)   │
          └──────┬──────┘
                 │ 收到 answer / 自然结束
                 ▼
          ┌─────────────┐
          │  completed  │ 本轮结束（上下文保留）
          └─────────────┘

2.2 路由决策

def route_message(target_agent_id, message):
    agent = get_agent(target_agent_id)

    if agent.state == "waiting" and agent.pending_tool_call:
        # 目标正在阻塞等待回答：直接返回到工具结果
        agent.resolve_pending_tool_call(message)
        return "resolved"

    if agent.state == "running":
        # 目标正在运行：inline 插入到当前对话上下文末尾
        agent.inject_inline_message(message)
        return "injected_inline"

    if agent.state in ("idle", "completed"):
        # 目标空闲：作为 user 消息插入，触发新一轮运行
        agent.inject_user_message(message)
        return "triggered_new_turn"

---

3. 关键场景分析

3.1 子智能体正在输出，主智能体要引导

场景：

• UI Operator_1 正在运行，输出了「接下来我会创建新 API...」
• Team Leader 要立即干预：「先不要创建 API，先确认现有 auth 模块是否可复用。」

处理：

• UI Operator_1 状态为 running（正常输出，不在阻塞等待）
• send_message_to_sub_agent 的消息 inline 插入到 UI Operator_1 的 messages 末尾
• UI Operator_1 下一轮模型调用时会看到这条 user 消息

3.2 子智能体正在等待回答

场景：

• Full-Stack Engineer_1 调用了 ask_master，等待 Team Leader 回答
• Team Leader 调用 answer_sub_agent_question

处理：

• Full-Stack Engineer_1 状态为 waiting
• 回答直接返回到 ask_master 工具结果中
• Full-Stack Engineer_1 继续执行

3.3 子智能体已经完成，主智能体追加任务

场景：

• Full-Stack Engineer_1 已经自然结束输出，状态为 completed
• Team Leader 要追加新任务

处理：

• send_message_to_sub_agent 的消息作为普通 user 消息插入
• 触发 Full-Stack Engineer_1 新一轮运行

3.4 边界情况：子智能体正在进行最后一轮输出

场景：

• UI Operator_1 正在输出最后一段话，后面没有工具调用了
• Team Leader 此时调用 send_message_to_sub_agent

处理：

• 如果消息到达时 UI Operator_1 还在运行：尝试 inline 插入
• 但由于这是最后一轮，后面没有工具调用了，inline 的消息不会被模型看到
• 因此需要在 UI Operator_1 本轮任务结束后，把这条消息作为触发新一轮任务的 user 消息发送

实现要点：

• 路由层维护每个子智能体的「待处理消息队列」
• 子智能体任务自然结束时，检查队列
• 如果有待处理消息，立即作为 user 消息触发新一轮运行

3.5 子智能体 A 问 B，B 正在运行

场景：

• UI Operator_1 调用 ask_other_agent(target=2) 问 Full-Stack Engineer_1
• Full-Stack Engineer_1 正在运行中

处理：

• 如果 Full-Stack Engineer_1 处于 running 状态：inline 插入 user 消息
• Full-Stack Engineer_1 下一轮输出时看到问题
• 如果 Full-Stack Engineer_1 调用 answer_other_agent：回答返回到 UI Operator_1 的 ask_other_agent 工具结果

---

4. 待处理消息队列

每个子智能体维护一个待处理消息队列 pending_messages。

@dataclass
class PendingMessage:
    id: str
    source_display_name: str
    source_agent_id: Optional[int]
    target_agent_id: int
    message_type: str  # task / output / ask / message / answer
    content: str
    question_id: Optional[str]  # 用于 answer 匹配
    created_at: float

4.1 队列消费时机

1. 子智能体每次模型调用前，先检查队列，把待处理消息合并到 messages 中
2. 子智能体从 waiting 状态恢复时，优先消费回答类消息
3. 子智能体自然结束时，如果有剩余待处理消息，立即触发新一轮运行

4.2 批量消费

参考现有通知池，每次消费时尽可能一次性取出所有可消费消息：

def consume_pending_messages(agent):
    messages = agent.pending_messages.drain_all()
    for msg in messages:
        formatted = format_message(msg)
        agent.messages.append({"role": "user", "content": formatted})
    return len(messages)

---

5. 与现有通知池的对比

维度	现有后台通知池	多智能体消息路由
触发源	子智能体/后台命令完成	子智能体间/主智能体向子智能体发消息
接收方	主智能体对话	子智能体对话或主智能体对话
插入方式	inline / 触发新任务	inline / 返回到工具结果 / 触发新任务
批量处理	_collect_pending_completion_notices 一次性取多条	每个子智能体维护自己的待处理队列
持久化	直接插入对话历史	先进入队列，再按状态消费并持久化

---

6. 消息路由实现位置

• 核心路由逻辑：modules/multi_agent/message_router.py
• 待处理队列：MultiAgentSubAgentTask.pending_messages
• 状态管理：MultiAgentSubAgentTask.state
• 工具调用等待：MultiAgentSubAgentTask.pending_tool_calls

---

7. 防丢失机制

1. 每条消息都有唯一 id
2. 消息进入队列时立即持久化到子智能体 metadata
3. 消费完成后从队列移除并持久化
4. 子智能体恢复时从 metadata 加载未消费消息
5. 回答类消息通过 question_id / message_id 精确匹配

---

8. 关键代码参考

• server/chat_flow_task_support.py：inject_runtime_user_message、process_sub_agent_updates
• server/chat_flow_task_main.py：_collect_pending_completion_notices、poll_completion_notifications、_dispatch_completion_user_notice
• server/chat_flow_tool_loop.py：execute_tool_calls 末尾的 process_sub_agent_updates(..., inline=True)