589 lines
20 KiB
Markdown
589 lines
20 KiB
Markdown
# 异步子 Agent 方案设计
|
||
|
||
> 版本: v8 (最终版)
|
||
> 日期: 2026-08-12
|
||
> 状态: 设计定稿
|
||
|
||
## 一、核心思路
|
||
|
||
子 agent 执行与父 agent 流程解耦，通过**双通道**实现异步通信：
|
||
|
||
- **子代理结果**走独立队列 `sub_done_q`（直达 wait，不绕 SQLite → bus → reload）
|
||
- **用户消息**走现有 `bus`（保持现有机制，零 channel 适配器改动）
|
||
|
||
wait 工具通过 **释放锁 + select! 等待 + 重新获取锁** 实现真等待，bus 消息能在 wait 期间通过 `is_waiting` 分流注入 history 并唤醒 wait。
|
||
|
||
### 核心机制
|
||
|
||
```
|
||
sub_done_q（独立队列） ← 后台子代理完成时注入 → wait select! 直达消费
|
||
bus（现有机制） ← ws.rs 用户消息注入 → process_one 分流注入 + wakeup 唤醒 wait
|
||
```
|
||
|
||
## 二、关键设计决策
|
||
|
||
| 决策点 | 选择 | 理由 |
|
||
|---|---|---|
|
||
| 子 agent 执行方式 | 异步（tokio::spawn） | 消除最大阻塞源 |
|
||
| 子代理结果通道 | sub_done_q 独立队列 | 直达 wait，不绕 SQLite → bus → reload |
|
||
| 用户消息通道 | 复用 bus | 现有机制，channel 零改动 |
|
||
| wait 语义 | 释放锁 + select! + 重获取锁 | 真等待，无死锁，无 TOCTOU |
|
||
| is_waiting 判断 | 持锁后判断 | 原子操作，无 TOCTOU |
|
||
| 完成回调内容 | 当前结果 + 未完成子代理 id 列表 | LLM 据此决策下一步 |
|
||
| task tool_result | running + 引导提示（只提 wait） | LLM 知道应调 wait |
|
||
| 嵌套层 | 仅顶层异步 | 避免复杂度爆炸 |
|
||
| 模式1（中断） | 不做 | 复杂度高，收益低 |
|
||
|
||
## 三、数据结构
|
||
|
||
### 1. SessionHistory 扩展（核心）
|
||
|
||
```rust
|
||
// src/gateway/session_history.rs
|
||
pub(crate) struct SessionHistory {
|
||
// 现有字段
|
||
topic_histories: HashMap<String, Vec<ChatMessage>>,
|
||
chat_topic_ids: HashMap<String, String>,
|
||
compression_in_flight: HashSet<String>,
|
||
topic_serial_locks: HashMap<String, Arc<tokio::sync::Mutex<()>>>,
|
||
conversations: Arc<dyn ConversationRepository>,
|
||
skill_events: Arc<dyn SkillEventRepository>,
|
||
|
||
// 新增：per-topic 子代理完成队列
|
||
sub_done_queues: HashMap<String, mpsc::Sender<SubagentResult>>,
|
||
// 新增：per-topic wait 唤醒信号
|
||
wait_wakeups: HashMap<String, Arc<tokio::sync::Notify>>,
|
||
// 新增：per-topic 等待状态
|
||
waiting_flags: HashMap<String, bool>,
|
||
}
|
||
```
|
||
|
||
**生命周期与 topic_serial_locks 完全一致**：
|
||
- 创建：`topic_serial_lock(topic_id)` 时同步创建
|
||
- 驱逐：`evict_inactive_if_needed` 时同步移除
|
||
- 清理：topic 删除时同步清理
|
||
|
||
### 2. SubagentResult 结构
|
||
|
||
```rust
|
||
struct SubagentResult {
|
||
task_id: String,
|
||
status: SubagentStatus, // completed/failed/timeout/cancelled
|
||
output: String, // 当前子代理的输出（与 task 工具返回格式一致）
|
||
pending_task_ids: Vec<String>, // 未完成的子代理 id 列表
|
||
}
|
||
```
|
||
|
||
**关键设计**：每个完成回调都带上 `pending_task_ids`，LLM 据此判断全局进度。
|
||
|
||
### 3. pending_subagents 表（新增）
|
||
|
||
```sql
|
||
CREATE TABLE pending_subagents (
|
||
task_id TEXT PRIMARY KEY,
|
||
parent_session_id TEXT NOT NULL,
|
||
parent_topic_id TEXT NOT NULL,
|
||
parent_chat_id TEXT NOT NULL,
|
||
parent_channel TEXT NOT NULL,
|
||
def_name TEXT,
|
||
spawned_at INTEGER NOT NULL,
|
||
status TEXT NOT NULL DEFAULT 'running' -- running/completed/failed/interrupted/cancelled
|
||
);
|
||
```
|
||
|
||
## 四、执行流程
|
||
|
||
### 阶段 ① 发起 task
|
||
|
||
```
|
||
用户消息 → bus → process_one → lock(topic) → agent.process()
|
||
→ LLM 调用 task 工具
|
||
→ TaskTool::execute():
|
||
① INSERT pending_subagents (status='running')
|
||
② tokio::spawn 子 agent（后台执行）
|
||
③ 立即返回 tool_result:
|
||
"running, task_id=t1。
|
||
调用 wait_for_subagents(timeout_secs) 等待子代理完成并获取结果。"
|
||
→ LLM 看到 running + 引导，自主决策下一步
|
||
```
|
||
|
||
**LLM 的 3 种可能行为**：
|
||
- 调用 wait_for_subagents（主动等待）— 主路径
|
||
- 调用其他工具（继续工作）— 罕见
|
||
- 生成最终回答（不调 wait）— 后端兜底
|
||
|
||
### 阶段 ② 后台子 agent 执行
|
||
|
||
```
|
||
tokio::spawn 独立运行:
|
||
→ SubAgentEmitter 实时推消息流给前端
|
||
→ Semaphore 限流（默认 8 并发）
|
||
→ 执行完成或失败或超时
|
||
```
|
||
|
||
### 阶段 ③ 子 agent 完成回调（直达 sub_done_q）
|
||
|
||
```
|
||
后台 task 结束:
|
||
① 查询 pending_subagents WHERE parent_topic_id=? AND status='running'
|
||
→ 得到未完成列表 ["t2", "t3"]
|
||
② sub_done_q.send(SubagentResult{
|
||
task_id: "t1",
|
||
status: "completed",
|
||
output: "{与 task 工具一致的输出格式}",
|
||
pending_task_ids: ["t2", "t3"]
|
||
})
|
||
③ UPDATE pending_subagents SET status='completed' WHERE task_id='t1'
|
||
```
|
||
|
||
**关键**：子代理结果**不写 SQLite、不入 bus**，直接通过 sub_done_q 传递给 wait。
|
||
|
||
**顺序保证**：查询未完成 → send → UPDATE
|
||
|
||
### 阶段 ④ wait 工具（释放锁 + select! + 重获取锁）
|
||
|
||
```
|
||
LLM 调用 wait_for_subagents(timeout_secs):
|
||
① 查询 pending_subagents
|
||
→ 无 running → 返回 "无需等待"，继续循环
|
||
→ 有 running → 进入等待
|
||
|
||
② 释放锁，标记等待
|
||
drop(lock_guard)
|
||
session.set_waiting(topic_id, true)
|
||
|
||
③ select! 等待（不持锁）
|
||
select! {
|
||
result = sub_done_q.recv() => Event::Subagent(result),
|
||
_ = wakeup.notified() => Event::UserMessage,
|
||
_ = timeout(60s) => Event::Timeout,
|
||
}
|
||
|
||
④ 重新获取锁，清除标记（顺序重要：先获取锁，再清除）
|
||
let lock_guard = lock(topic_id).await
|
||
session.set_waiting(topic_id, false)
|
||
|
||
⑤ 返回 tool_result
|
||
Subagent(r) => "子代理 {task_id} 完成: {output}。未完成: {pending}"
|
||
UserMessage => "有新用户消息到达"
|
||
Timeout => "等待超时。未完成子代理: {pending}"
|
||
```
|
||
|
||
### 阶段 ⑤ process_one 分流（wait 期间注入用户消息）
|
||
|
||
```rust
|
||
async fn process_one(&self, inbound: InboundMessage) {
|
||
let guard = lock(topic_id).await;
|
||
|
||
if session.is_waiting(topic_id) {
|
||
// wait 期间：只注入消息到 history，不启动 process
|
||
session.inject_to_history(inbound).await; // INSERT + 更新内存 history
|
||
drop(guard);
|
||
session.wakeup(topic_id); // 唤醒 wait
|
||
return;
|
||
}
|
||
|
||
// 正常处理（topic 空闲或 agent 已退出）
|
||
agent.process().await;
|
||
}
|
||
```
|
||
|
||
### 阶段 ⑥ process 退出兜底
|
||
|
||
```
|
||
process 退出前:
|
||
if waited=true:
|
||
→ 不发 ExecutionCompleted（等 bus 消息触发新轮）
|
||
else if pending_subagents 有 running:
|
||
→ 不发 ExecutionCompleted（LLM 没调 wait 但有 pending）
|
||
else:
|
||
→ 发 ExecutionCompleted
|
||
```
|
||
|
||
## 五、wait 工具的完整设计
|
||
|
||
### 工具定义
|
||
|
||
```rust
|
||
// 工具名：wait_for_subagents
|
||
// 参数：timeout_secs（可选，默认 60s）
|
||
// 描述：等待子代理完成或用户消息。用于有 pending 子代理时进入等待状态。
|
||
|
||
wait_for_subagents(timeout_secs: Option<u64>) -> ToolResult {
|
||
let timeout = Duration::from_secs(timeout_secs.unwrap_or(60));
|
||
let pending = query_pending_task_ids(topic_id);
|
||
|
||
if pending.is_empty() {
|
||
return ToolResult {
|
||
output: "无 pending 子代理，无需等待".to_string(),
|
||
metadata: { wait_marker: false }
|
||
};
|
||
}
|
||
|
||
// 释放锁，标记等待
|
||
drop(lock_guard);
|
||
session.set_waiting(topic_id, true);
|
||
|
||
// 等待（不持锁）
|
||
let event = select! {
|
||
result = sub_done_q.recv() => Event::Subagent(result),
|
||
_ = wakeup.notified() => Event::UserMessage,
|
||
_ = sleep(timeout) => Event::Timeout,
|
||
};
|
||
|
||
// 重新获取锁，清除标记（顺序重要）
|
||
let lock_guard = lock(topic_id).await;
|
||
session.set_waiting(topic_id, false);
|
||
|
||
// 先 drain sub_done_q 积压（处理多子代理同时完成）
|
||
match event {
|
||
Event::Subagent(result) => {
|
||
format_subagent_result(result)
|
||
}
|
||
Event::UserMessage => {
|
||
"有新用户消息到达".to_string()
|
||
}
|
||
Event::Timeout => {
|
||
let pending = query_pending_task_ids(topic_id);
|
||
format!("等待超时。未完成子代理: {:?}", pending)
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
### wait 的三种返回场景
|
||
|
||
| 事件 | 返回内容 | LLM 行为 |
|
||
|---|---|---|
|
||
| 子代理完成 | "子代理 t1 完成: {output}。未完成: [t2, t3]" | 处理结果，看 pending 决定继续 wait 或综合 |
|
||
| 用户消息到达 | "有新用户消息到达" | LLM 下一轮看到注入的 user 消息 |
|
||
| 超时 | "等待超时。未完成子代理: [t1, t2]" | 决定继续等或放弃 |
|
||
|
||
## 六、两个通道的职责（核心设计）
|
||
|
||
### sub_done_q：wait 专属消费
|
||
|
||
```
|
||
后台子代理完成 → sub_done_q.send(result) ← 不获取锁，直达 wait
|
||
↓
|
||
wait select! → 立即收到 → 返回给 LLM
|
||
```
|
||
|
||
**不经过 SQLite、不经过 bus**，直接队列传递。
|
||
|
||
### bus：用户消息，wait 期间分流注入
|
||
|
||
```
|
||
用户消息 → bus.publish_inbound(UserMessage)
|
||
→ process_one → lock(topic)
|
||
├─ is_waiting=true → 注入 history + wakeup 唤醒 wait
|
||
└─ is_waiting=false → 正常启动 process
|
||
```
|
||
|
||
**wait 期间**：用户消息持锁注入 history，wakeup 唤醒 wait。
|
||
**非 wait 期间**：正常走 process_one。
|
||
|
||
## 七、LLM 看到的信息流
|
||
|
||
| 时机 | LLM 看到的内容 |
|
||
|---|---|
|
||
| task 调用后 | `tool_result("running, task_id=t1。调用 wait_for_subagents 等待...")` |
|
||
| wait 调用后 | `tool_result("已进入等待，2 个子代理运行中: [t1,t2]")` |
|
||
| t1 完成（wait 返回） | `tool_result("子代理 t1 完成: {output}。未完成: [t2]")` |
|
||
| 用户消息（wait 返回） | `tool_result("有新用户消息到达")` + 下轮 `user("用户消息")` |
|
||
| t2 完成（wait 返回） | `tool_result("子代理 t2 完成: {output}。未完成: []")` |
|
||
|
||
LLM 通过 `pending_task_ids` 能判断：
|
||
- `[]` 空列表 → 全部完成，综合回答
|
||
- `["t3"]` 非空 → 还有未完成的，继续调 wait
|
||
|
||
## 八、消息的完整路径
|
||
|
||
### 子代理结果
|
||
|
||
```
|
||
后台 task 完成 → sub_done_q.send(SubagentResult)
|
||
→ wait select! 立即收到 → 返回 tool_result 给 LLM
|
||
```
|
||
**直达，不绕路**。
|
||
|
||
### 用户消息
|
||
|
||
```
|
||
所有 channel 消息 → bus.publish_inbound(UserMessage)
|
||
→ process_one → lock(topic)
|
||
├─ is_waiting=true → 注入 history + wakeup 唤醒 wait
|
||
│ → wait 返回 "有新用户消息到达"
|
||
│ → LLM 下一轮看到注入的 user 消息
|
||
└─ is_waiting=false → 正常启动 process
|
||
```
|
||
|
||
**两条路径职责清晰，互不干扰**。
|
||
|
||
## 九、三个边界问题的解法
|
||
|
||
### 边界 1：wakeup vs timeout 竞态
|
||
|
||
```
|
||
timeout 和 wakeup 同时触发 → select! 随机选一个
|
||
如果 timeout 赢：
|
||
- 用户消息已注入 history（process_one 在锁内完成注入）
|
||
- wait 返回 "超时"
|
||
- LLM 下一轮 reload history 会看到用户消息
|
||
- 不会丢失
|
||
```
|
||
**解法**：无需特殊处理，消息已持久化，reload 能读到。
|
||
|
||
### 边界 2：is_waiting 清除时机
|
||
|
||
```
|
||
错误顺序：清除 is_waiting → 获取锁
|
||
→ 中间窗口 process_one 看到 is_waiting=false → 启动 process → 死锁
|
||
|
||
正确顺序：获取锁 → 清除 is_waiting
|
||
→ 持锁后才清除，process_one 在锁外看到 is_waiting=true → 走注入路径
|
||
```
|
||
**解法**：`let guard = lock().await; set_waiting(false);` 顺序保证。
|
||
|
||
### 边界 3：多条用户消息
|
||
|
||
```
|
||
wait 释放锁后，3 条用户消息依次到达：
|
||
msg1 → lock → 注入 → drop → wakeup
|
||
msg2 → lock → 注入 → drop → wakeup
|
||
msg3 → lock → 注入 → drop → wakeup
|
||
|
||
wakeup 是 Notify，3 次 notify 只存储 1 个
|
||
wait 被唤醒 1 次，但 history 已有 3 条消息
|
||
wait 返回 "有新用户消息到达"
|
||
LLM 下一轮看到 3 条 user 消息
|
||
```
|
||
**解法**：Notify 合并是正确行为，history 完整。
|
||
|
||
## 十、时序场景表现
|
||
|
||
### 场景 1：多子代理并发完成
|
||
|
||
```
|
||
① task(t1, t2, t3) → running
|
||
② wait → 释放锁 → select!
|
||
③ t1, t2, t3 几乎同时完成 → sub_done_q: [t1, t2, t3]
|
||
④ wait select 收到 t1 → 返回 "t1 完成。未完成: [t2, t3]"
|
||
⑤ LLM 处理 t1 → 再调 wait
|
||
wait 先 try_recv → 立即拿到 t2 → 返回 "t2 完成。未完成: [t3]"
|
||
⑥ LLM 处理 t2 → 再调 wait
|
||
wait 先 try_recv → 立即拿到 t3 → 返回 "t3 完成。未完成: []"
|
||
⑦ LLM 综合 t1+t2+t3 回答
|
||
```
|
||
**3 轮 wait，但无延迟（try_recv 立即返回积压）**。
|
||
|
||
### 场景 2：wait 期间用户发消息
|
||
|
||
```
|
||
① task(t1) → running → wait → 释放锁 → select!
|
||
② 用户发消息 → bus → process_one → lock（wait 已释放，获取成功）
|
||
→ is_waiting=true → 注入 history + wakeup
|
||
③ wait 被 wakeup 唤醒 → 重新获取锁 → 返回 "有新用户消息到达"
|
||
④ LLM 看到 [wait_result, user("用户消息")] → 处理用户消息
|
||
```
|
||
**用户消息即时响应，wait 被唤醒**。
|
||
|
||
### 场景 3：子代理完成 + 用户消息同时到达
|
||
|
||
```
|
||
① task(t1) → running → wait → 释放锁 → select!
|
||
② t1 完成 → sub_done_q
|
||
用户发消息 → bus → process_one → 注入 + wakeup
|
||
③ select! 随机选一个：
|
||
- 选 sub_done_q → 返回 "t1 完成" → LLM 下一轮看到 user 消息
|
||
- 选 wakeup → 返回 "有新用户消息" → LLM 下一轮看到 t1 结果（try_recv）
|
||
```
|
||
**两种情况都不丢失消息**。
|
||
|
||
### 场景 4：父 agent 生成最终回答（不调 wait）
|
||
|
||
```
|
||
① task(t1) → running
|
||
② LLM 生成最终回答（无 tool_calls）→ process 退出
|
||
③ 退出前检查: pending 有 running → 不发 ExecutionCompleted
|
||
④ t1 完成 → sub_done_q.send（无人消费，留在队列）
|
||
→ 触发新轮 process_one（通过 bus 或定期检查）
|
||
⑤ 新轮 LLM 看到 [自己之前的回答] → try_recv sub_done_q → 处理 t1 结果
|
||
```
|
||
|
||
**注意**：此场景需要额外机制触发新轮 process（sub_done_q 有残留时）。可通过：
|
||
- 后台 task 完成后同时 publish_inbound(bus) 作为触发信号
|
||
- 或 wait 退出兜底检查 sub_done_q
|
||
|
||
### 场景 5：wait 超时
|
||
|
||
```
|
||
① task(t1) → running → wait(60s) → 释放锁 → select!
|
||
② 60s 内无事件 → timeout 触发
|
||
③ wait 重新获取锁 → 返回 "等待超时。未完成: [t1]"
|
||
④ LLM 决定继续等（再调 wait）或放弃
|
||
```
|
||
|
||
## 十一、关键机制
|
||
|
||
### 1. 队列与状态生命周期管理
|
||
|
||
```rust
|
||
// session_history.rs
|
||
|
||
/// 获取或创建该 topic 的 sub_done_q sender
|
||
pub(crate) fn sub_done_queue(&mut self, topic_id: &str) -> mpsc::Sender<SubagentResult> {
|
||
self.sub_done_queues
|
||
.entry(topic_id.to_string())
|
||
.or_insert_with(|| {
|
||
let (tx, _rx) = mpsc::channel(32);
|
||
tx
|
||
})
|
||
.clone()
|
||
}
|
||
|
||
/// 获取或创建该 topic 的 wakeup
|
||
pub(crate) fn wait_wakeup(&mut self, topic_id: &str) -> Arc<Notify> {
|
||
self.wait_wakeups
|
||
.entry(topic_id.to_string())
|
||
.or_insert_with(|| Arc::new(Notify::new()))
|
||
.clone()
|
||
}
|
||
|
||
/// 设置等待状态
|
||
pub(crate) fn set_waiting(&mut self, topic_id: &str, waiting: bool) {
|
||
self.waiting_flags.insert(topic_id.to_string(), waiting);
|
||
}
|
||
|
||
/// 检查等待状态
|
||
pub(crate) fn is_waiting(&self, topic_id: &str) -> bool {
|
||
self.waiting_flags.get(topic_id).copied().unwrap_or(false)
|
||
}
|
||
|
||
// evict_inactive_if_needed 中同步清理
|
||
fn evict_inactive_if_needed(&mut self) {
|
||
// ... 现有逻辑
|
||
if let Some(tid) = to_evict.cloned() {
|
||
self.topic_histories.remove(&tid);
|
||
self.topic_serial_locks.remove(&tid);
|
||
self.sub_done_queues.remove(&tid); // 新增
|
||
self.wait_wakeups.remove(&tid); // 新增
|
||
self.waiting_flags.remove(&tid); // 新增
|
||
}
|
||
}
|
||
```
|
||
|
||
### 2. pending_subagents 状态管理
|
||
|
||
| 状态 | 含义 |
|
||
|---|---|
|
||
| running | 后台执行中 |
|
||
| completed | 正常完成 |
|
||
| failed | 执行失败 |
|
||
| interrupted | 进程崩溃，启动扫描标记 |
|
||
| cancelled | 用户取消 |
|
||
|
||
### 3. 崩溃恢复
|
||
|
||
**启动扫描**：
|
||
```sql
|
||
UPDATE pending_subagents SET status='interrupted' WHERE status='running';
|
||
```
|
||
|
||
**history 加载时**：
|
||
- 遇到 `tool_result("running, task_id=...")` → 查 pending_subagents
|
||
- status=interrupted → 替换为"子代理因重启中断，请决定是否重新发起"
|
||
|
||
**两步操作之间崩溃**（查询未完成 → send → UPDATE）：
|
||
- 崩溃在任何点：启动扫描标记 interrupted，history 加载时替换占位
|
||
- 最坏情况：sub_done_q 消息丢失（wait 未消费），但 pending 表状态可恢复
|
||
|
||
### 4. 取消传播
|
||
|
||
- `task_id → CancellationToken` 映射
|
||
- 用户取消 → 遍历 pending → 触发 cancel_token
|
||
- 子代理退出 → sub_done_q.send(SubagentResult{status:"cancelled"})
|
||
|
||
### 5. 去重
|
||
|
||
- DefaultSubAgentRuntime 维护 `completed_tasks: HashSet<task_id>`
|
||
- 入队前检查，已入队的不再入
|
||
|
||
### 6. topic_serial_lock 一致性
|
||
|
||
- 用户消息走 bus，用 topic_id 作 lock_key
|
||
- 子代理结果走 sub_done_q，不需要 lock（直达 wait）
|
||
- wait 释放锁后，bus 的 process_one 能获取锁注入消息
|
||
|
||
## 十二、改动范围
|
||
|
||
| 文件 | 改动 |
|
||
|---|---|
|
||
| `session_history.rs` | 新增 sub_done_q + wait_wakeup + waiting_flags（per-topic）+ 生命周期 |
|
||
| `runtime.rs` | spawn 异步化 + 完成回调（含 pending_task_ids）注入 sub_done_q |
|
||
| `task_tool.rs`（新） | TaskTool 返回 running + 引导提示 + spawn |
|
||
| `wait_tool.rs`（新） | wait 释放锁 + select! + 重获取锁 + 返回 pending 信息 |
|
||
| `agent_loop.rs` | 传递 lock_guard 给 wait 工具（支持释放/重获取） |
|
||
| `processor.rs` | process_one 分流：is_waiting 检查 + 注入 + wakeup |
|
||
| `db.rs` | pending_subagents 表 + 启动扫描 |
|
||
| `history.rs` | running 占位替换 |
|
||
| `cancel.rs` | 取消传播 |
|
||
|
||
**预估总改动：~800 行**
|
||
|
||
## 十三、与之前方案的对比
|
||
|
||
| 维度 | v6（bus + break） | v8（双通道 + wait 释放锁） |
|
||
|---|---|---|
|
||
| 子代理结果通道 | bus（绕 SQLite → reload） | **sub_done_q 直达** ✅ |
|
||
| 用户消息通道 | bus | bus |
|
||
| wait 语义 | break 退出 | **select! 真等待** ✅ |
|
||
| wait 期间用户消息 | 等锁（agent 退出后） | **即时注入 + wakeup** ✅ |
|
||
| is_waiting 判断 | 不需要 | 需要（持锁后判断，无 TOCTOU） |
|
||
| TOCTOU 风险 | 无 | **无**（持锁后判断） |
|
||
| 死锁风险 | 无 | **无**（wait 释放锁） |
|
||
| LLM 调用次数 | 2（batch）/ N+2 | **2-N** |
|
||
| process 退出 | 每次都退出 reload | wait 不退出，连续推理 |
|
||
| 上下文连贯 | 每轮 reload | **同 process 内连续** ✅ |
|
||
| 实现复杂度 | 低 | 中（~20 行新增 + 边界解法） |
|
||
| channel 适配器改动 | 零 | **零** ✅ |
|
||
|
||
## 十四、不做的事
|
||
|
||
- ❌ 模式1（中断插入）
|
||
- ❌ user_msg_q（用户消息走 bus）
|
||
- ❌ 嵌套层异步
|
||
- ❌ 子代理结果走 bus（改回 sub_done_q）
|
||
- ❌ wait 不持锁（改为释放锁 + select + 重获取）
|
||
|
||
## 十五、方案定位总结
|
||
|
||
| 机制 | 定位 | 通道 |
|
||
|---|---|---|
|
||
| wait 释放锁 + select! | **主路径**（真等待，子代理直达） | sub_done_q |
|
||
| process_one 分流 + wakeup | **用户消息路径**（注入 + 唤醒） | bus |
|
||
| process 退出兜底（pending 检查） | **安全网**（LLM 不调 wait 时） | - |
|
||
|
||
三机制分工清晰：
|
||
- wait select! 是核心（子代理结果直达 + 用户消息 wakeup）
|
||
- process_one 分流是用户消息路径（wait 期间注入 + 唤醒）
|
||
- 退出兜底是安全网（防止 LLM 不调 wait 时丢消息）
|
||
|
||
## 十六、配置说明
|
||
|
||
```json
|
||
{
|
||
"subagent": {
|
||
"max_concurrent": 8,
|
||
"spawn_timeout_secs": 300,
|
||
"wait_default_timeout_secs": 60
|
||
}
|
||
}
|
||
```
|
||
|
||
| 配置项 | 默认值 | 说明 |
|
||
|---|---|---|
|
||
| max_concurrent | 8 | 并发子 agent 数量（Semaphore） |
|
||
| spawn_timeout_secs | 300 | 子 agent 执行超时 |
|
||
| wait_default_timeout_secs | 60 | wait 工具默认超时 |
|