223 lines
8.4 KiB
Markdown
223 lines
8.4 KiB
Markdown
# 多智能体模式消息路由机制
|
||
|
||
> 消息路由由**接收方**决定：根据接收方当前状态，选择 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 路由决策
|
||
|
||
```python
|
||
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`。
|
||
|
||
```python
|
||
@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 批量消费
|
||
|
||
参考现有通知池，每次消费时尽可能一次性取出所有可消费消息：
|
||
|
||
```python
|
||
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)`
|