0 Fork 0

Files

T

2026-05-25 19:38:16 +08:00

38 KiB

Raw Blame History

多智能体圆桌研讨 · 后端开发文档

本文档面向需要维护或扩展「多智能体圆桌研讨」功能后端的开发者。前端 API 速查见 multi-agent-api.md；前端实现见 multi-agent-frontend-dev.md；推荐弹窗专项进度见 recommend-agents-progress.md（若未提供则当作待补文件）。

⚠️ 历史版本曾引用 multi-agent-step1-api.md、multi-agent-step3-api.md，当前仓库未提供这两份文件；Step 1（intent）与 Step 3（artifacts）的接口签名直接参考本页第 4 / 7 节即可。

源码内已有中文模块级注释（app/gateway/routers/intent.py、recommend.py、multi_agent.py、artifacts.py 及 clarification_utils.py），与本文互为补充。

0. 速读

三个圆桌 Router（intent / recommend / multi_agent）全部走 HTTP loopback：在同进程内通过 httpx.AsyncClient 调本机 127.0.0.1 的 /api/threads、/api/agents、/api/threads/{id}/runs/stream。
所有 loopback httpx.AsyncClient 必须 trust_env=False。开发机若设了 HTTP_PROXY / HTTPS_PROXY，httpx 默认会把 loopback 也走代理 → 不可达 → 单次请求多出 40 秒超时（典型症状日志：headers_in=42.160s）。
席位 agent 持久注册（磁盘 .deer-flow/agents/ + _sync_legacy_agents），协调智能体每次新建（命名 roundtable-coordinator-{timestamp36}，DB 累积，需后续 GC）。
init 是 SSE 而非一次性 JSON：用 asyncio.as_completed 按真实完成顺序推送 seat_ready，前端因此能逐个点亮席位。
init_done.thread_ids 包含协调智能体自身（agent_threads[main_agent_name] = coord_thread），前端 Step 3 列文件时需自行过滤。
澄清协议统一：Step 1 与 Step 2 leader 都借 ask_clarification + 终态 clarification 帧驱动同一套选项卡片组件；allow_multiple 默认 false（单选、点击即发送）。
leader 共识 = status: [] 空数组；澄清 = status: "clarification"（字符串）；派活 = status: [[name, task], ...]。这三者互斥不混用。
后端 picks clamp 2–6，前端用户可手动加到 8。

1. 功能定位与三步数据流

对应前端页面：/page/workspace/roundtable/planning（RoundtablePlanningPage.tsx）的三个步骤。

Step 1  任务意图理解
  POST /api/intent/init          → 建 thread,绑定内置 roundtable-intent
  POST /api/intent/stream (多轮)  → 澄清对话,终态 [INTENT_READY] + summary

        ↓ intentReady { objective, constraints, assumptions }

推荐弹窗(Step 1 → Step 2 之间,可选)
  POST /api/recommend/stream      → 从候选池选 2–6 个席位,[RECOMMEND_READY] + picks

        ↓ selectedAgents[]   // 前端可再手动调整到 2-8

Step 2  多智能体圆桌研讨
  POST /api/multi-agent/init (SSE) → 协调智能体 + 各席位 thread(asyncio.as_completed 流式推 seat_ready)
  POST /api/multi-agent/run/stream → leader 派活 / special 执行 / 澄清

        ↓ lastLeaderContent、各席位交付、hasConsensus

Step 3  最终方案展示
  GET   /api/threads/{tid}/artifacts          → 列出 outputs/ 产出文件
  GET   /api/threads/{tid}/artifacts/{path}   → 预览或下载
  PATCH /api/threads/{tid}/artifacts/{path}   → 文本类产物原地编辑(可选)
  (总控结论、席位文本主要来自 Step 2 前端内存缓存,一般不再调 run/stream)

步骤	前端 `currentStep`	专属 Router	内置 Agent
1	1	`intent.py`	`roundtable-intent`（固定 id，复用）
1→2	推荐弹窗	`recommend.py`	`roundtable-recommender`（每请求新建 thread）
2	2	`multi_agent.py`	协调智能体（每次新建）+ 用户选定的 `roundtable-*` 席位
3	3	`artifacts.py`（通用）	无新 agent

所有接口需 Authorization: Bearer <JWT>。Loopback 调用统一 trust_env=False，避免本机 HTTP_PROXY 把 127.0.0.1 指到不可达代理（症状：headers_in=40s+，前端表现为「Step 1 init 卡顿 / Step 2 init 永远不发 seat_ready」）。

2. 涉及的后端文件

offline-backend-20260512/backend/
├── app/gateway/
│   ├── app.py                          # include_router: multi_agent, intent, recommend (artifacts 已存在)
│   └── routers/
│       ├── multi_agent.py              # Step 2:init(SSE) + run/stream + 共享的 loopback 工具
│       ├── intent.py                   # Step 1:init + stream(import multi_agent 工具)
│       ├── recommend.py                # Step 1→2:推荐 stream(import multi_agent 工具)
│       └── artifacts.py                # Step 3:列出/下载/编辑 thread 产出(通用,非圆桌专用)
├── packages/harness/deerflow/tools/builtins/
│   └── clarification_utils.py          # resolve_allow_multiple + coerce_bool(Step1/Step2 共享)
└── .deer-flow/agents/
    ├── roundtable-intent/              # Step 1 任务理解(SOUL 含完整工作流,可不挂 Skill)
    ├── roundtable-recommender/         # 推荐智能体
    ├── roundtable-intelligence/        # 默认可选席位(6 个,前端可替换)
    ├── roundtable-environment/
    ├── roundtable-solution-design/
    ├── roundtable-risk-review/
    ├── roundtable-execution-plan/
    ├── roundtable-summary/
    └── roundtable-coordinator-<id>/    # 协调智能体壳,每次会话新建,长期累积

内置 agent 注册：各 roundtable-* 目录通过 _sync_legacy_agents()（app/gateway/routers/agents.py）在下次 /api/agents 请求时 upsert 进 agents 表，user_id IS NULL 表示内置。recommend.py 不依赖此表，前端会过滤 roundtable-coordinator-* 出候选池。

Router 间共享导出：intent.py 与 recommend.py 从 multi_agent.py import _auth_headers / _create_thread / _flatten_content / _loopback_base / _parse_last_messages / _run_payload / _stream_upstream 等底层工具，不要在 intent / recommend 重复实现 loopback 逻辑。

共享工具逻辑：clarification_utils.py 的 resolve_allow_multiple(args) 与 coerce_bool(value, default) 被 intent.py / multi_agent.py / ClarificationMiddleware 共用：

coerce_bool 把 LLM tool args 里可能出现的 bool / 数字 / 字符串（"true"、"1"、"yes"、"y"、"on" / "false"、"0"、"no"、"n"、"off"）统一成 bool；
resolve_allow_multiple 仅 args["allow_multiple"] 显式为真时返回 True，缺省 / null 都视为 False（单选）。

3. 接口总览

方法	路径	步骤	用途
POST	`/api/intent/init`	1	创建意图澄清 thread
POST	`/api/intent/stream`	1	SSE，一轮用户对话
POST	`/api/recommend/stream`	1→2	SSE，一次性推荐席位
POST	`/api/multi-agent/init`	2	SSE，按完成顺序推 `seat_ready`，最后 `init_done`
POST	`/api/multi-agent/run/stream`	2	SSE，leader 或 special 一轮
GET	`/api/threads/{thread_id}/artifacts`	3	列出该 thread 的 outputs 文件
GET	`/api/threads/{thread_id}/artifacts/{path}`	3	预览/下载单个产出（含 `.skill/` 内文件提取）
PATCH	`/api/threads/{thread_id}/artifacts/{path}`	3	编辑文本类产物（白名单后缀）

字段级请求/响应见各 step 的 API 文档；下文只写后端实现语义。

4. Step 1：任务理解（`intent.py`）

4.1 定位

轻量包装 Lead Agent 运行时，固定跑内置 roundtable-intent（INTENT_AGENT_ID 常量）。不做 broadcast、不派活；每轮用户消息对应一次 /stream。

4.2 `POST /api/intent/init`

_create_thread() → LangGraph thread
返回固定 agent_id: "roundtable-intent"（不每次新建 agent）

请求体：{"model": "deepseek-chat"} — model 可选，此处不写入 thread，每轮 /stream 自己带 model。

与 Step 2 /init 的区别：无协调智能体、无 agent_names 列表。

4.3 `POST /api/intent/stream`

入参：{thread_id, message, model?}，缺 thread_id 或 message → 400。

上游 run_payload：见 §6.6；excluded_tools 屏蔽除 ask_clarification 外的几乎全部工具，硬清单：

[
  "web_search",
  "present_files",
  "view_image",
  "agent_orchestration",
  "bash", "ls", "read_file", "write_file", "str_replace",
  "memory", "hindsight_recall", "hindsight_reflect", "hindsight_retain",
  "task", "write_todos",
]

写死黑名单而不是「只允许 ask_clarification」的原因：SOUL 文字约束不可靠，硬排除更稳；防止模型偏离去写方案文档或派活。

流式：透传 LangGraph SSE → _parse_last_messages(captured) 从倒数 values 快照拿 messages。

终态帧判定顺序（重要，勿调整）：

[INTENT_READY] 优先：从最新消息向前扫描每条 AIMessage，正则 _INTENT_READY_PATTERN = /\[INTENT_READY\]\s*```json\s*(\{.*?\})\s*```/s 命中后 json.loads 解析
- 命中 → status: "done"，summary 为 JSON、content 为含标记的整段
- 优先于 ask_clarification：模型可能在较早的 AIMessage 里已 emit [INTENT_READY]，但本轮残留 ask_clarification 的 tool_call；不先扫 READY 前端会卡在澄清 UI
否则用 _find_dispatch_message(messages) 找最后一条带 tool_calls 的 AIMessage，检查是否含 ask_clarification
- 命中 → status: "clarification"，附结构化字段：
```
{
  "status": "clarification",
  "content": "<follow-up 或 preamble 或 question>",
  "question": "...",
  "clarification_type": "missing_info",
  "clarification_context": "...",
  "options": ["..."],
  "allow_custom": true,
  "allow_multiple": false,
  "summary": null
}
```
- content 取 _find_trailing_ai_text(messages, exclude_text=preamble)：ClarificationMiddleware 常在 tool 后还有一条纯文本 AIMessage，比 tool_call 旁的短 preamble 更适合作为气泡正文
- allow_multiple 由 resolve_allow_multiple(cargs) 决定（缺省 false）
否则 → status: "asking"，content 为 _flatten_content(dispatch_msg.content)，前端继续 chat

异常：HTTPException / 其它异常都返回 {"status": "error", "content": "...", "summary": null}，前端展示红色错误气泡。

summary 形状（done 时）：

{
  "objective": "一句话目标",
  "constraints": ["硬约束 1", "..."],
  "assumptions": ["可选假设"]
}

4.4 内置 Agent：`roundtable-intent`

配置：.deer-flow/agents/roundtable-intent/config.yaml + SOUL.md
SOUL 内写清：工作流、ask_clarification 用法、[INTENT_READY] 格式、禁止派活/写文件
可不配置 skills；若配置 Skill，SOUL 要求不得偏离澄清与结束协议

5. 推荐智能体（`recommend.py`，Step 1 → 2）

5.1 定位

无状态单次调用：每次请求内部 _create_thread()，不暴露 /init。前端传入 Step 1 的 intent + 过滤后的 candidates（剔除 roundtable-intent、roundtable-recommender、历史 roundtable-coordinator-*）。

5.2 `POST /api/recommend/stream`

入参校验：

intent.objective 必填（trim 后非空）
candidates 必须是非空数组
至少一个有效 agent_id（trim 后非空）
三者缺一 → 400

Prompt 拼装（_build_prompt）：

任务目标:<objective>
约束:<constraint1>; <constraint2>; ...
关键假设:<assumption1>; ...

候选智能体池:
- <agent_id> (<name>): <description>
- ...

请按 SOUL 中规定的格式输出推荐结果(自然语言思路 + [RECOMMEND_READY] + json 块)。

格式刻意贴近 SOUL 示例，让模型有清晰的复刻模板。

上游 run_payload：agent_name=RECOMMENDER_AGENT_ID、excluded_tools=["web_search", "present_files", "view_image", "agent_orchestration", "ask_clarification"] — 纯判断，不调任何工具。

终态分支：

取最后一条非空 AIMessage 的文本（last_ai_content）
正则 _RECOMMEND_READY_PATTERN = /\[RECOMMEND_READY\]\s*```json\s*(\{.*?\})\s*```/s 匹配 picks JSON
_validate_picks(parsed.picks, candidate_ids)：
- 跳过非 dict / 缺 agent_id / 不在候选池 / 已 seen 的项
- 至多保留 _MAX_PICKS = 6 个
- 不足 _MIN_PICKS = 2 → 返回 []，触发 fallback
有效 picks ≥ 2 → status: "done"，picks 有值
否则 → status: "asking"（不是模型在追问用户，而是「网关未能产出合法 picks」，前端走 FALLBACK 默认 6 席）

{
  "status": "done",
  "content": "<自然语言推荐理由>",
  "picks": [{"agent_id": "roundtable-risk-review", "reason": "..."}]
}

picks 数量上下限的不一致（需要心里有数）：

边界	后端	前端弹窗
下限	`_MIN_PICKS = 2`（不足返 asking）	`MIN_SELECTED = 2`（确认按钮校验）
上限	`_MAX_PICKS = 6`（截断模型输出）	`MAX_SELECTED = 8`（允许用户手加 2 个）

后端只对模型输出做 clamp，用户可手动追加候选池里其它 agent 到 8 个。

前端流式展示：自然语言理由与 JSON 由 [RECOMMEND_READY] 分界；TitleMiddleware 产生的标题增量会在客户端过滤（langgraph_node 含 TitleMiddleware / tags 含 middleware:title），避免污染推荐理由区。

5.3 与 SOUL 的契约

模型须在自然语言思路后输出：

[RECOMMEND_READY]
```json
{"picks": [{"agent_id": "...", "reason": "..."}, ...]}


`agent_id` 必须是候选池内已有的 id，`_validate_picks` 会做白名单校验。

---

## 6. Step 2：多智能体研讨（`multi_agent.py`）

### 6.1 `POST /api/multi-agent/init` — SSE，非一次性 JSON

请求体: { "agent_names": ["roundtable-intelligence", ...], // 用户选的席位 id "main_agent_name": "roundtable-coordinator-xxx", // 当次总控 id,前端生成 "model": "deepseek-chat" // 可选,写入 coordinator agent }


**早期校验**（同步抛 HTTP 4xx，**不**进 SSE）：

- `agent_names` 必须非空
- 每个 `agent_names` 元素 + `main_agent_name` 都需通过 `_AGENT_ID_PATTERN = ^[A-Za-z0-9_-]+$`（用 `_validate_agent_id` 提前 400；上游若收到不合规 id 会抛 422，提前拦更友好）
- 所有 id 都 `.lower()` 后比较 / 存储

**SSE 内部流程**：

asyncio.as_completed(各席位 _prepare_sub_agent) 每完成一个 → data: {"event":"seat_ready","agent_name","thread_id","display_name","description"} (按真实完成顺序,先完成先发,前端因此能逐个点亮)

异常时:取消所有未完成的 task,避免泄露到 AsyncClient close 之后

协调智能体: POST /api/threads → coord_thread POST /api/agents skills=["agent_orchestration"] soul=动态拼接(本次 init 的席位列表 + 派活规则) id=main_agent_name(前端生成的 roundtable-coordinator-) agent_threads[main_agent_name] = coord_thread // ⚠️ thread_ids 含 coordinator 自身 → data: {"event":"coordinator_ready","main_agent_name","thread_id"}

→ data: {"event":"init_done","main_agent_name","thread_ids":{...}}

失败 → data: {"event":"error","detail":"..."}


**协调智能体创建幂等**：若 `_create_agent` 返回 (False, "exists ...")（同名 agent 已存在）则继续往下走，不报错。注意这里的「已存在」会复用历史 agent 的 SOUL（与本次动态拼接的不同），通常只在前端用同一个时间戳跑过两遍时触发。

**协调智能体 SOUL**（动态拼接，与席位列表一起写入）：

#核心定位: 用途:根据用户的任务目标,调用其他智能体干活,只负责协调汇总, 当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果

#调度方法: 通过使用 agent_orchestration 技能调用其他智能体干活,使用技能时传入 agent_name 和需要做的活。 ⚠️ agent_name 必须严格使用下方列出的英文 id(例如 roundtable-intelligence), 不要把括号里的中文显示名拼进 agent_name 里。

#可调度智能体:

agent_name: roundtable-intelligence(情报收集):
agent_name: roundtable-environment(环境评估): ...


中文显示名通过 `GET /api/agents/{id}` 取 `name`、描述取 `description`。**强调英文 id** 是因为部分模型会把整段 `roundtable-x（中文）` 当 `agent_name` 塞回来，靠后端 `_strip_display_suffix` 兜底（详见 §6.4）。

### 6.2 `/run/stream` — `agent_type = "leader"`（`_leader_run`）

并行启动 pre_broadcast_task(不 await):向除自己外所有 thread 追加 "总控智能体收到用户消息:" human 消息
_run_payload + _stream_upstream(POST /api/threads/{coordinator}/runs/stream)
- excluded_tools: ["web_search"] // 只禁联网,其余工具(含 agent_orchestration)保留
- skill_stop_names: {"agent_orchestration", *callers_extra} // 网关自动并入 agent_orchestration,让 SkillStopMiddleware 在工具调用后停图, // 便于本路由从 AIMessage.tool_calls 解析派活列表
透传上游 SSE,_parse_last_messages 拿 messages
await pre_broadcast_task(LLM 通常已跑完,这里 await 接近 0 成本)
_find_dispatch_message(messages) 找带 tool_calls 的 AIMessage
分三支:

分支 A — ask_clarification: {"status": "clarification", "content": follow_up 或 question 或 leader_text, "question", "clarification_type", "clarification_context", "options", "allow_custom", "allow_multiple", "agent_name"} ⚠️ 空 dispatched 不等于共识,前端展示澄清卡片并暂停自动派活循环

分支 B — 有 tool_calls(派活): _extract_dispatch(tool_call)解析每个 tool_call → (sub_name, task_text) 并行 _broadcast("总控智能体给 X 派活,内容为:Y", skip=[leader, sub_name]) 等所有派活广播完成(asyncio.gather) {"status": [[sub_name, task_text], ...], "content": leader_text, "agent_name": agent_name}

分支 C — 无 tool_calls: {"status": [], "content": leader_text, "agent_name": agent_name} // 共识达成,前端 hasConsensus = true


**`_extract_dispatch` 双格式支持**：

1. **结构化**：`{"name": "agent_orchestration", "args": {"agent_name": "...", "task": "..."}}`
   — 大多数 LLM 在 SOUL 引导下走这条
2. **Legacy bash argv**：`{"name": "bash", "args": {"command": '... .py "X" "Y" ...'}}`
   — 正则 `_TOOL_CALL_PATTERN = /\.py\s+"([^"]*)"\s+"([^"]*)"/`，兼容旧 skill；不可移除

两条路径都会过 `_strip_display_suffix`：用 `re.split(r"[^A-Za-z0-9_-]", agent_name, maxsplit=1)[0]` 截断第一个非合法 id 字符之后的内容，把 `roundtable-x（中文）` 还原为 `roundtable-x`；剥离后若仍违规，原样返回让上游 422。

**时序日志**（grep `[leader-timing]`）：

payload_built — 构造请求体耗时(broadcast 已经在后台跑) first_upstream_line — 上游首行(LangGraph 启动 + LLM 首字节) upstream_stream_complete — 整条 SSE 透传完成 pre_broadcast_await — 必须 await 才能 yield 终态;>0 表示广播比 LLM 还慢 dispatch_broadcasts_await — 派活广播并行 gather 耗时


### 6.3 `/run/stream` — `agent_type = "special"`（`_special_run`）

_run_payload + _stream_upstream(POST /api/threads/{sub}/runs/stream)
- excluded_tools: ["web_search", "ask_clarification", "present_files", "view_image", "agent_orchestration"] // 子智能体禁派活/澄清/文件呈现/图片;只做交付
透传 SSE,_parse_last_messages 拿最后一条 AIMessage 的 content
启动 post_broadcast_task("子智能体 X 完成 Y 工作,交付内容为:..."),不 await
yield {"status": "子智能体X完成X工作", "content": "<交付正文>", "agent_name": "..."} // ⚠️ 在 yield 之后再 await broadcast,让前端先收到终态、关 loading
await post_broadcast_task(SSE 连接稍微多保持一会;前端早就拿到结果了)


**时序日志**（grep `[special-timing]`）：

payload_built first_upstream_line upstream_stream_complete post_broadcast_total — yield 之后 await broadcast 的总耗时


### 6.4 `_broadcast` 与 `_append_thread_message` 机制

```python
async def _append_thread_message(client, base, headers, thread_id, content):
    # GET /api/threads/{tid}/state → state
    # state["values"]["messages"].append({"type": "human", "content": content})
    # POST /api/threads/{tid}/state
    # 失败仅打 warning,不阻断主流程

async def _broadcast(client, base, headers, message, agent_threads, skip):
    targets = [tid for name, tid in agent_threads.items() if name not in skip]
    await asyncio.gather(*(_append_thread_message(...) for tid in targets),
                        return_exceptions=True)

目的：各 thread 对话历史里保留「谁派活、谁交付」的群聊上下文；skip 通常含发起方与接收方，避免回声。

三种触发：

触发点	内容	skip
leader pre_broadcast	`总控智能体收到用户消息:<msg>`	`[leader]`
leader 派活	`总控智能体给 X 派活,内容为:<task>`	`[leader, sub_name]`
special 后广播	`子智能体 X 完成 Y 工作,交付内容为:<content>`	`[sub]`

注意：广播是 best-effort，对方 LLM 真正干活仍需前端显式 /run/stream 调到对应席位。

6.5 终态帧协议（leader / special / clarification）

前端 streamMultiAgent 用 isFinalStatusFrame() 识别：同时存在 status 与 content。

`status` 类型	含义
`[["agent","task"],...]`	派活列表（数组）；`[]` = 共识
`"clarification"`	总控追问，需用户回复后再 leader
`"子智能体 X 完成..."`	special 完成（字符串）
`"error"`	异常

与 Step 1 的 clarification 帧字段对齐（question、options、allow_multiple 等），便于复用选项卡片组件。

6.6 `_run_payload` 完整字段

{
  "input": {"messages": [{"type": "human",
                          "content": [{"type": "text", "text": new_message}],
                          "additional_kwargs": {}}]},
  "config": {"recursion_limit": 1000},
  "context": {
    "agent_name": <coordinator 或 sub agent id>,
    "model_name": "deepseek-chat",
    "mode": "pro",
    "reasoning_effort": "medium",
    "thinking_enabled": True,
    "is_plan_mode": False,
    "subagent_enabled": False,
    "thread_id": <thread_id>,
  },
  "stream_mode": ["messages-tuple", "values"],
  "stream_subgraphs": True,
  "stream_resumable": True,
  "assistant_id": "lead_agent",
  "on_disconnect": "continue",
  "excluded_tools": <按 router 不同>,
  "skill_stop_names": <leader 强制并入 agent_orchestration>,
}

字段含义：

recursion_limit: 1000 — 防止 LangGraph 循环深度兜底（圆桌实际很难达到）
thinking_enabled: True — 开启模型的扩展思考（CoT），需 model 支持
is_plan_mode: False、subagent_enabled: False — 圆桌内禁用 TodoList / subagent 中间件，避免与多智能体协调冲突
stream_mode: ["messages-tuple", "values"] — 同时需要逐 token 增量与最后一次完整 values 快照
assistant_id: "lead_agent" — 走 Lead Agent 工厂
on_disconnect: "continue" — 客户端断连后上游继续跑完（不丢 token 但客户端拿不到）

6.7 上游 SSE 透传与 `_parse_last_messages`

_stream_upstream 逐行 yield 上游 /api/threads/{tid}/runs/stream 的 SSE。返回元组：

(line, None)        — 原样转发给浏览器（含 event: / data: 行）
(None, captured)    — 流结束，captured 是所有 data: 行的字符串列表

日志 [upstream-timing] <agent> headers_in=<> / first_line=<>：排查首 token 慢、代理导致 loopback 超时（见 trust_env=False）。

_parse_last_messages(captured) 从倒数往前扫，跳过 trailing data: null end 帧，找第一条解析为 dict 且含 messages key 的 frame；取 messages 列表。无可解析帧抛 502。

6.8 三个 AIMessage 搜索辅助

函数	用途
`_find_dispatch_message(messages)`	从后向前找带 `tool_calls` 的 AIMessage（跳过 SkillStop / Clarification 注入的 ToolMessage）；都没有则返回最后一条 AIMessage
`_find_trailing_ai_text(messages, exclude_text)`	从后向前找无 tool_calls 的 AIMessage 的纯文本（用于澄清后的 follow-up，排除等于 preamble 的项）
`_flatten_content(content)`	把 `str` / 结构化 content blocks `[{"text": ...}]` 拍平成单串

7. Step 3：产出文件（`artifacts.py`，通用 Thread API）

Step 3 没有单独的 multi_agent 路由；使用既有 artifacts 能力，圆桌只是消费者之一。

7.1 `GET /api/threads/{thread_id}/artifacts`

鉴权：@require_permission("threads", "read", owner_check=True)
扫描：get_paths().sandbox_outputs_dir(thread_id, user_id=get_effective_user_id())
- 解析为 .deer-flow/users/{user_id}/threads/{thread_id}/user-data/outputs/
collect_outputs_listing(outputs_dir)：
- 递归 rglob("*")，过滤非文件
- 软链接安全检查（resolved.relative_to(outputs_root)，越界丢弃）
- 按 modified_at 倒序
目录不存在 → files: []（非 404）

{
  "thread_id": "...",
  "files": [
    {
      "path": "mnt/user-data/outputs/report.md",
      "name": "report.md",
      "size_bytes": 12480,
      "mime_type": "text/markdown",
      "modified_at": "2026-05-23T14:32:30.123Z"
    }
  ]
}

前端 listSessionArtifacts(threadId) 封装此接口；Step 3 对协调 thread 与各席位 thread 分别拉取，合并为下载清单（实际只列子席位）。

7.2 `GET /api/threads/{thread_id}/artifacts/{path}`

鉴权同上
虚拟路径前缀 mnt/user-data/outputs/...，经 resolve_thread_virtual_path 转物理路径并防穿越
特殊：.skill/ 压缩包透视：若 path 含 .skill/，从 ZIP 提取内部文件（如 xxx.skill/SKILL.md）。_extract_file_from_skill_archive 同时支持「直接路径命中」与「带顶层目录前缀」两种压缩包结构。结果带 Cache-Control: private, max-age=300 头避免重复解压
ACTIVE_CONTENT_MIME_TYPES = {"text/html", "application/xhtml+xml", "image/svg+xml"} 始终强制 Content-Disposition: attachment（防止 XSS 与脚本执行）
文本类（mime 以 text/ 开头）→ PlainTextResponse
后缀判定不到、但 8KB 内无 null 字节 → 当文本返回（is_text_file_by_content）
其它 → FileResponse 或 Response 内嵌
?download=true 对所有类型强制附件下载

7.3 `PATCH /api/threads/{thread_id}/artifacts/{path}`

鉴权：@require_permission("threads", "write", owner_check=True)
请求体：{"content": "<新内容>"}
限制：
- .skill/ ZIP 内文件只读（400）
- ACTIVE_CONTENT_MIME_TYPES 拒绝（400 active artifacts are read-only）
- 文件后缀需在 EDITABLE_TEXT_SUFFIXES = {.md, .markdown, .txt, .json, .yaml, .yml, .csv, .xml, .log} 之内，或 mime 以 text/ 开头
用 UTF-8 覆写原文件，返回 PlainTextResponse(body.content)

圆桌 Step 3 当前主要消费 GET，PATCH 用于「在线修订交付稿」类场景，前端默认未启用。

7.4 与 Step 2 内存的关系

总控结论、各席位最终段落：主要来自 Step 2 前端缓存的 lastLeaderContent、step2RoundtableDialogues（filter confidence === "子智能体交付"），不为 Step 3 新增聚合接口；如后续需要后端聚合，可加 sessions/summary 草案路由。

8. 内置席位 SOUL 设计要点

每个 roundtable-*（除 intent / recommender）遵循统一结构：角色定位、核心职责、输出风格（emoji + 简短结论）、边界（只产出本席位、等总控派活）。

席位	agent_id	emoji	核心职责
情报收集	`roundtable-intelligence`	📢	同类案例 + 历史风险
环境评估	`roundtable-environment`	🌍	硬件容量 + 承载力
方案设计	`roundtable-solution-design`	📐	主备方案 + 关键路径
风险审查	`roundtable-risk-review`	⚠️	合规红线 + 收敛建议
执行规划	`roundtable-execution-plan`	📅	里程碑 + 资源
总结陈述	`roundtable-summary`	🎤	归纳 + 冲突标记

子智能体不安装 agent_orchestration；仅协调智能体安装，保证单调度源。

9. 关键设计决策

9.1 HTTP loopback + `trust_env=False`

三个 router 均通过 httpx 调本机 Gateway（127.0.0.1），复用鉴权与 LangGraph 路由，代价是额外 HTTP 开销；init 已用 asyncio.as_completed 并行化席位准备。

每个 httpx.AsyncClient(timeout=None, trust_env=False) 都是必须的：

timeout=None — SSE 长连接需要
trust_env=False — 否则 httpx 会读取 HTTP_PROXY / HTTPS_PROXY 环境变量，把 loopback 127.0.0.1 也走代理。开发机常配的「校园网代理 / 公司代理」会把回环地址丢弃，导致单次请求 40+ 秒超时（日志典型症状：headers_in=42.160s）

9.2 协调智能体每次新建

main_agent_name 形如 roundtable-coordinator-{timestamp36}，会话隔离；长期会累积 agent 目录（实际仓库已有数十个 roundtable-coordinator-mp*），需后续 GC 策略：

选项 A：定时任务清理 N 天前无活跃 thread 的 coordinator agent
选项 B：每次创建前 LRU 清理（注意：跨服务实例难协调）
选项 C：把 coordinator 实例化为「内置 + 动态 SOUL」复用同一个 id

9.3 澄清协议统一

Step 1 与 Step 2 leader 均通过 ask_clarification + 终态 clarification 帧驱动选项卡片；[INTENT_READY] / 派活列表 / [] 各自表示阶段结束语义，互不混用。

9.4 选项单选默认

resolve_allow_multiple()：未传则 false（单选、点击发送）；仅显式 allow_multiple: true 时多选。降低模型把 missing_info 全标成多选导致 UX 混乱的概率。coerce_bool 容忍 LLM 返回的字符串/数字 bool。

9.5 SSE 末尾状态帧

LangGraph 原生流不含「派活结果 / 意图 summary / 推荐 picks」；各 router 在流末追加 JSON data: 帧，前端用 isFinalStatusFrame() / isFinalIntentFrame() / isFinalRecommendFrame() 与 LangGraph 帧区分解析。

9.6 SkillStopMiddleware 配合 leader 派活

leader run 自动并入 skill_stop_names = {"agent_orchestration", ...}，让 SkillStopMiddleware 在 agent_orchestration 工具调用后立刻停图。这样网关能从 AIMessage.tool_calls 直接读出派活列表，不必等 LLM 真把任务执行完。

9.7 init `agent_threads` 含 coordinator 自身

init_done.thread_ids 是 {agent_name: thread_id} 形式，其中 main_agent_name 自己的 thread 也在里面。前端 Step 3 列文件时需自行过滤掉 coordinator thread，否则会多调一次 listSessionArtifacts（虽然结果会是空，但有冗余 HTTP）。

10. 性能特征

阶段	已做优化	典型耗时
Step 2 `/init`	席位 `asyncio.as_completed` + 流式 `seat_ready` + 去掉 intro 广播	约 0.5–1.2s（视席位数）
LLM `/run/stream`	—	数百 ms~数十秒（主因）
leader pre_broadcast	与上游 LLM run 并发启动	通常已在 LLM 完成前结束
leader 派活广播	`asyncio.gather` 并行	< 200ms（同 thread state R/W）
special post_broadcast	yield 之后才 await，前端先关 loading	不阻塞 UX
Step 1/推荐	单 thread，无 broadcast	与模型相关

时序日志关键字（开发自带）：

grep "init-timing"      .runtime/logs/*.log   # 仅前端打,作参考
grep "upstream-timing"  .runtime/logs/*.log   # _stream_upstream 内 headers_in / first_line
grep "leader-timing"    .runtime/logs/*.log   # payload_built / first_upstream_line / dispatch_broadcasts_await ...
grep "special-timing"   .runtime/logs/*.log   # payload_built / first_upstream_line / post_broadcast_total

仍存在的开销：GET /api/agents/{id} 触发 legacy sync 扫目录、GET/POST /state 广播等。未来可考虑 loopback 改内部函数调用。

11. 依赖与扩展点

11.1 `agent_orchestration` 技能

协调智能体派活依赖此技能；正则 _TOOL_CALL_PATTERN 从 tool 输出提取 agent_name 与 task。技能缺失时 leader 易变成「无 tool_calls → 空 status → 前端误判共识」。

11.2 模型

model 写入 run 的 context.model_name；前端可按席位覆盖 special 调用的模型（见前端文档 §5.5 Persona 抽屉）。

11.3 新增席位

.deer-flow/agents/roundtable-xxx/ 增加 config.yaml + SOUL.md
前端候选池 / 推荐逻辑纳入该 id
Step 2 agent_names 传入即可 — multi_agent 路由无需改

11.4 新增 Step 1 工具

若需新工具，须同步修改 intent.py 的 excluded_tools 黑名单（当前默认禁 15 个工具，仅 ask_clarification 实际可用）。

11.5 改回旧 bash 派活协议

后端已双格式兼容，前端无需变化；只要 SOUL 改成生成 bash + .py "X" "Y" argv，_extract_dispatch 仍会解析正确。但建议保持新的结构化 agent_orchestration，prompt 更稳定。

11.6 把广播改为推送

_append_thread_message 通过 GET state → mutate → POST state 实现，是为了与 LangGraph checkpoint 一致；如果将来 Gateway 提供 POST /api/threads/{id}/messages/append 这类原子接口，可换上，降低并发竞争（虽然圆桌当前不会并发广播同一 thread）。

12. 故障排查

现象	可能原因	排查
Step 1 一直有澄清、无法进入 Step 2	未产出 `[INTENT_READY]` 或终态被 `clarification` 覆盖	看 SOUL、最后一帧 `status`；确认 intent 先扫 READY（§4.3 顺序）
选项全是多选	旧前端默认 `allow_multiple !== false`	确认 `allow_multiple` 字段与 `resolve_allow_multiple`（缺省 false）
Step 2 init SSE `error`	协调 agent 创建失败 / agent_orchestration 缺失 / `_validate_agent_id` 命中	Network + 后端 `multi_agent` 日志
Step 2 init 完全卡住 ~40 秒	`httpx` 走了 `HTTP_PROXY`	确认 `trust_env=False` 仍在；grep `headers_in=` 看时长
leader 立即 `status: []`	未调用 orchestration 技能 / SkillStopMiddleware 没拦到 / SOUL 没声明派活	换模型或加强 SOUL；检查 `skill_stop_names` 是否包含 `agent_orchestration`
dispatch 失败但 leader 文本看着正常	模型把 `roundtable-x（中文）` 整段塞进 `agent_name`	`_strip_display_suffix` 会兜底；若仍报 400，强化 SOUL 的「严格英文 id」
推荐 picks 为空	未输出 `[RECOMMEND_READY]` 或 picks 不在候选池	看 recommend 终态帧；检查 `candidate_ids` 集合
Step 3 无文件	子 agent 未写 outputs / 错路径	`GET .../artifacts` 对子 thread_id 逐个查；确认 sandbox outputs 路径
`seat_ready` 顺序不固定	`asyncio.as_completed` 按真实完成顺序推送	这是设计行为，不是 bug；前端不应按顺序依赖 agent_name

日志关键字

grep "intent stream"       .runtime/logs/*.log
grep "multi_agent\|multi-agent"  .runtime/logs/*.log
grep "recommend stream"    .runtime/logs/*.log
grep "dispatch failed"     .runtime/logs/*.log
grep "upstream-timing"     .runtime/logs/*.log
grep "leader-timing"       .runtime/logs/*.log
grep "special-timing"      .runtime/logs/*.log
grep "headers_in="         .runtime/logs/*.log   # 走代理症状

13. 测试建议

backend/tests/test_intent.py          # init、INTENT_READY 优先级、clarification 字段
backend/tests/test_recommend.py       # picks 校验、候选池过滤、clamp 2-6
backend/tests/test_multi_agent.py     # init SSE 事件顺序、leader/special 终态、clarification

可用 httpx.MockTransport 模拟 loopback，不依赖真实 Gateway：

def mock_handler(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/api/threads":
        return httpx.Response(200, json={"thread_id": "t-1"})
    ...

transport = httpx.MockTransport(mock_handler)
client = httpx.AsyncClient(transport=transport, base_url=...)

针对 SSE 上游可用 respx 或直接 mock _stream_upstream 的字符串列表。

14. 路径速查

资源	路径
Step 1 路由	`offline-backend-20260512/backend/app/gateway/routers/intent.py`
推荐路由	`offline-backend-20260512/backend/app/gateway/routers/recommend.py`
Step 2 路由	`offline-backend-20260512/backend/app/gateway/routers/multi_agent.py`
Step 3 列出/下载/编辑	`offline-backend-20260512/backend/app/gateway/routers/artifacts.py`
路由注册	`offline-backend-20260512/backend/app/gateway/app.py`
澄清单选/多选	`offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/clarification_utils.py`
任务理解 Agent	`offline-backend-20260512/backend/.deer-flow/agents/roundtable-intent/`
推荐 Agent	`offline-backend-20260512/backend/.deer-flow/agents/roundtable-recommender/`
圆桌席位	`offline-backend-20260512/backend/.deer-flow/agents/roundtable-*/`
协调智能体壳（累积）	`offline-backend-20260512/backend/.deer-flow/agents/roundtable-coordinator-*/`
前端 Step 1 API	`frontend-web/src/roundtable-planning/api/intent.ts`
前端推荐 API	`frontend-web/src/roundtable-planning/api/recommend.ts`
前端 Step 2 API	`frontend-web/src/roundtable-planning/api/multi-agent.ts`
前端页面	`frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx`
前端开发文档	`frontend-web/docs/multi-agent-frontend-dev.md`
API 速查	`frontend-web/docs/multi-agent-api.md`

38 KiB Raw Blame History Unescape Escape