wsk/deerflow-code

0 Fork 0

Files

T

fengxiangdebaozi a85f1571a0 接口修改

2026-05-23 10:49:16 +08:00

15 KiB

Raw Blame History

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

本文档面向需要维护或扩展"多智能体圆桌研讨"功能后端的开发者。前端 API 速查请看 multi-agent-api.md,此文档关注实现细节、设计取舍、性能特征。

1. 功能定位

把"任务目标 + 用户约束"丢给一个协调智能体,由它通过 agent_orchestration 技能调度 6 个固定职能的子智能体（情报收集 / 环境评估 / 方案设计 / 风险审查 / 执行规划 / 总结陈述）协同产出方案,直到协调智能体认为信息足够、停止派活并输出最终结论为止。

对应前端页面：/page/roundtable/planning 的 Step 2 多智能体研讨。

2. 涉及的后端文件

offline-backend-20260512/backend/
├── app/gateway/
│   ├── app.py                       # 注册 multi_agent.router
│   └── routers/
│       └── multi_agent.py           # 本功能唯一新增的 router
└── .deer-flow/agents/
    ├── roundtable-intelligence/     # 6 个内置子智能体目录
    │   ├── config.yaml              #   ├── id / name / description
    │   └── SOUL.md                  #   └── 角色 prompt
    ├── roundtable-environment/
    ├── roundtable-solution-design/
    ├── roundtable-risk-review/
    ├── roundtable-execution-plan/
    └── roundtable-summary/

零代码改动构造内置 agent：6 个目录通过既有的 _sync_legacy_agents()（app/gateway/routers/agents.py）在下次任意 /api/agents 请求时自动 upsert 进 agents 表,user_id IS NULL 标记为内置。

3. 接口总览

路由前缀 /api/multi-agent，所有接口需要 Authorization: Bearer <JWT>。

方法	路径	用途
POST	`/api/multi-agent/init`	一次性创建协调智能体 + 6 个子席位线程,返回 `thread_ids` 映射
POST	`/api/multi-agent/run/stream`	SSE 流式驱动一次 leader 或 special 调用

字段细节见 multi-agent-api.md,这里不重复。

4. 内部实现详解

4.1 `/init` 内部流程

[阶段1] 并行准备 6 个子智能体席位 (asyncio.gather)
  ├─ GET  /api/agents/{id}         × 6  → 取 name + description
  └─ POST /api/threads             × 6  → 建新 thread

[阶段2] 创建协调智能体
  ├─ POST /api/threads                  → 协调智能体的 thread
  └─ POST /api/agents (with model + skills=[agent_orchestration] + soul)
       └─ soul 里包含可调度的 6 个 agent_name + description 列表

协调智能体 SOUL 是动态拼出来的(multi_agent.py:216-224):

soul = (
    "#核心定位：\n"
    "用途：根据用户的任务目标,调用其他智能体干活,只负责协调汇总,...\n"
    "#调度方法：\n"
    "通过使用agent_orchestration技能调用其他智能体干活,...\n"
    "#可调度智能体：\n"
    f"{descriptions}"  # ← 6 个子席位的 name+description
)

由于 main_agent_name 由前端每次新建（roundtable-coordinator-{timestamp36}）,所以协调智能体本身是一次性的、用完即弃。

4.2 `/run/stream` 内部流程

接收 agent_type 后走两条分支：

`agent_type = "leader"` (协调智能体)

1. broadcast「总控收到用户消息」到所有非自己线程  (asyncio.gather)
2. 拼 RunCreateRequest,转发到 POST /api/threads/{coordinator}/runs/stream
   - excluded_tools: ["web_search"]
   - context.model_name = 用户传的 model
3. 流式透传 LangGraph 事件给前端
4. 流结束后从最后一帧 values 取最末一条 message
5. 解析 message.tool_calls (agent_orchestration 技能调用):
   - 用 _extract_name_task() 提取 [agent_name, task_text]
   - 对每个 [agent_name, task]:
     broadcast「总控给 X 派活,内容为: ...」到所有非派活双方
6. 末尾追加状态帧:
   {
     "status": [["agent_name", "task"], ...],  // 空数组 = 共识达成
     "content": "<leader 最终文本>",
     "agent_name": "<coordinator>"
   }

`agent_type = "special"` (子智能体)

1. 拼 RunCreateRequest,转发到 POST /api/threads/{sub_agent}/runs/stream
   - excluded_tools: ["web_search","ask_clarification","present_files","view_image"]
2. 流式透传 LangGraph 事件
3. 流结束后从最后一帧 values 取交付内容 (content)
4. broadcast「子智能体 X 完成 Y 工作,交付内容为: ...」到所有非自己线程
5. 末尾追加状态帧:
   {
     "status": "子智能体 X 完成 Y 工作",
     "content": "<交付内容>",
     "agent_name": "<sub agent>"
   }

4.3 _broadcast 机制

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))

_append_thread_message 对每个目标 thread:

GET  /api/threads/{id}/state           # 拿当前 messages 列表
state.values.messages.append({"type":"human","content":...})
POST /api/threads/{id}/state           # 写回完整 state

这是为了让每个子线程的对话历史里都能看到全局上下文（其他席位说了什么、协调智能体怎么派活的）。子智能体下次被调用时,模型自然能基于这份"群消息回放"做判断。

关键不变量:skip 列表通常包含发起方和接收方,避免发起方收到回声、接收方收到重复消息。

5. 6 个内置 agent SOUL 设计要点

每个 roundtable-* 的 SOUL.md 遵循统一结构：

# {席位名}

## 角色定位
你是"多智能体圆桌研讨"会议的 **{席位}** 席位。会议由"总控协调智能体"统一调度,
你只在被派到任务时发言,不需要主动推动议程。

## 核心职责
- ...具体职责...

## 输出风格
- 每次发言以 {emoji} 开头
- 第一句直接给结论
- 不超过 4 句

## 边界
- 只产出 {本席位输出}
- 不替其他席位做 {其他席位输出}
- 任务结束后停下,等待总控派下一个任务

## 示例输出
> ...

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

为什么不让子智能体自己调度：所有 SOUL 都明文写了"只在被点到时发言,由总控协调智能体派活"。子智能体不装 agent_orchestration 技能。只有协调智能体装。这是单调度源,避免多智能体相互抢调度权造成的死循环。

6. 关键设计决策

6.1 为什么用 HTTP loopback 而不是内部函数调用

multi_agent.py 调用 /api/threads、/api/agents、/api/threads/.../runs/stream 都是用 httpx 走 http://localhost:8001/... 自身循环回来,而不是直接 from app.gateway.services import ...。

好处:

自动复用 Gateway 的鉴权 + CSRF + permission 中间件,不用再手写
复用既有路由的请求验证（Pydantic）和错误响应
Authorization 头透传,被调端自然知道是哪个用户在调

代价: 每次调用都有一次额外的 TCP+JWT 解析开销（~10-50ms）。但因为 init 已经做了并行优化（A）、broadcast 已经省掉（C）,目前总体耗时是可接受的。

如未来需要进一步优化,可改为直接调用 app/gateway/services.py 里的内部函数,绕过 HTTP 一层。

6.2 为什么协调智能体每次新建

前端在每次进入 step 2 时生成 roundtable-coordinator-{Date.now().toString(36)},作为 main_agent_name。

好处:

不同会话/不同用户/同一用户不同任务之间完全隔离
协调智能体的对话历史就是本次研讨的全部上下文,不会被上次任务污染
SOUL 不会被"上次的可调度列表"覆盖（虽然现在 6 个席位是固定的）

代价: 每次都要在 agents 表插一行,.deer-flow/agents/<id>/ 下生成新目录。长期会累积大量协调智能体目录,未来需要清理脚本（标记 description 以 "你是总控控制协调智能体" 开头的 agent 自动 GC）。

6.3 子智能体席位为什么固定 6 个

圆桌的 6 个角色定位是产品级强约束（情报/环境/方案/风险/执行/总结）
写在后端 SOUL.md 一处维护,跟其他内置 agent 同等地位
前端只需要把 6 个 hyphen-case 名作为 agent_names 传过来即可

如果后续要支持"用户自选席位",在前端加 mapping 面板让用户从 GET /api/agents 里挑,不动后端。

6.4 为什么 SSE 末尾还要追加一个 `{status:...}` 帧

LangGraph 原生 SSE 里没有"调度结果"这种语义,前端只能从最后一帧 values 里去翻 tool_calls。如果由前端解析,前后端协议会绑死 LangGraph 内部数据形状,迁移成本高。所以后端在流末尾追加一个协议层的总结帧:

{
  "status": [["agent_name", "task"], ...] | "<完成提示>",
  "content": "<最终 AI 文本>",
  "agent_name": "<本轮 agent>"
}

前端 streamMultiAgent helper 通过 isFinalStatusFrame() 识别该帧（必须同时有 status + content 键）,与其他 LangGraph 帧区分。

7. 性能特征

7.1 已应用的优化

ID	优化	收益
A	`/init` 阶段 1 并行化（`asyncio.gather`）	阶段 1 从 2-4s 降到 0.3-0.7s
C	删除 `/init` 末尾的 intro 广播（SOUL 已含定位）	阶段 3 整段消失,省 1-2s

应用前 /init 耗时 3-8s,应用后 0.5-1.2s。

7.2 仍然存在的开销点

调用	单次耗时	主因
`GET /api/agents/{id}`	50-300ms	每次触发 `_sync_legacy_agents()` 扫整个 `.deer-flow/agents/` 目录(目前 30+ 项)
`POST /api/threads`	100-200ms	LangGraph thread 表插入 + 默认 state 初始化
`POST /api/agents`	100-300ms	文件系统写入 SOUL.md + config.yaml + agents 表
`GET/POST /state`	50-150ms × 2	checkpointer 读写
`POST /runs/stream` 启动	50-100ms	RunManager + StreamBridge 初始化
LLM 推理	数百毫秒~数十秒	取决于 model + 上下文长度

LLM 推理才是绝对大头,基础设施开销在十几个百分点以内。

7.3 未来可探索

B(已识别未做): broadcast 跳过 GET-state,直接 POST 新建态。但 POST 需要构造完整的 state shape,改造成本不小。
D(已识别未做): loopback HTTP → 内部函数调用。代码改动大,但能省掉 ~20-30 次 HTTP 解析开销。

8. 依赖与扩展点

8.1 `agent_orchestration` 技能

协调智能体强依赖此技能,SOUL 里硬编码:

通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活

技能的实际派活产物是一个形如 python3 scripts/agent_orchestration.py "agent_name" "task" 的 shell 调用文本,后端用正则提取：

_TOOL_CALL_PATTERN = re.compile(r'\.py\s+"([^"]*)"\s+"([^"]*)"')

若该技能被删除/改名,leader 流程会拿不到 tool_calls,直接走"无派活=共识达成"分支。注意 skills/ 下必须保留此技能。

8.2 模型选择

前端通过 GET /api/models 拿到列表,在 model 字段传 Model.name (与 config.yaml.models[].name 对应)。

/init 的 model 字段 → 协调智能体的默认模型 (写进 agents 配置)
/run/stream 的 model 字段 → 该次具体运行用什么模型 (写进 context.model_name)

前端允许"逐席位覆盖模型",由前端在 special 调用时按 agentConfigs[seat].model 传入。后端不感知"席位"概念,只看到 model 字段。

8.3 自定义新席位

如果要加第 7 个席位（比如"成本测算"）:

.deer-flow/agents/roundtable-cost/ 加 config.yaml + SOUL.md
前端 RoundtablePlanningPage.tsx 的 ROUNDTABLE_ROLE_TEMPLATES + AGENT_NAME_TO_ROLE 加映射
圆桌可视化（圆圈环形图）加一个气泡 + 头像

后端代码完全不需要改。

9. 故障排查

现象	可能原因	排查
`/init` 500 "create coordinator agent failed"	`agent_orchestration` 技能不存在 / config.yaml 写入失败	看后端日志 `multi_agent` logger;检查 `skills/` 目录
`/run/stream` 502 "upstream returned HTTP 4xx"	JWT 过期 / 转发到上游时鉴权挂了	前端检查 Authorization 头是否带上
leader 永远 `status: []` 立即结束	协调智能体没产生 `tool_calls`,直接给文本回答	协调智能体 SOUL 不够强、或模型不支持工具调用,换 model
special 内容为空	子智能体被 `excluded_tools` 限制到无可用工具 / 模型无文本输出	看上游 `runs/stream` 的 values 帧 messages.last.content
进入 step 2 卡住转圈	`/init` 超时或失败	F12 Network 看 `/api/multi-agent/init` 状态码与响应体

服务端日志关键字

# multi_agent 路由的日志
grep "multi_agent" .runtime/logs/*.log

# 派活解析失败
grep "dispatch failed" .runtime/logs/*.log

# 广播失败（通常无害）
grep "update state .* failed" .runtime/logs/*.log

10. 测试入口

后端目前没有针对本功能的单测。建议补：

backend/tests/test_multi_agent.py
  - test_init_creates_threads_for_all_subagents
  - test_init_creates_coordinator_with_orchestration_skill
  - test_init_returns_400_when_agent_names_empty
  - test_stream_leader_emits_status_frame_with_content
  - test_stream_special_emits_status_frame_with_content
  - test_extract_name_task_parses_orchestration_skill_output
  - test_broadcast_skips_listed_agents

可用 httpx.MockTransport 拦截 loopback 请求,不依赖真实 Gateway。

11. 路径速查

资源	路径
路由实现	`offline-backend-20260512/backend/app/gateway/routers/multi_agent.py`
路由注册	`offline-backend-20260512/backend/app/gateway/app.py`(`include_router(multi_agent.router)`)
6 个内置 agent	`offline-backend-20260512/backend/.deer-flow/agents/roundtable-*/`
`agent_orchestration` 技能定义	`skills/public/agent_orchestration/`(若存在)
前端 API helper	`frontend-web/src/roundtable-planning/api/multi-agent.ts`
前端页面	`frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx`
前端 API 速查	`frontend-web/docs/multi-agent-api.md`

15 KiB Raw Blame History Unescape Escape