圆桌 Step2:固定协调员单例、优化调度性能、修正研讨中状态

将 roundtable-coordinator 改为内置单例避免重复创建;legacy sync 加 TTL 减少 runs/stream 延迟;前端仅当前发言者显示研讨中。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
董宏凯andCursor committed 2026-05-28 23:23:40 +08:00
1 parent 2371364911
commit de597b602a
9 files changed
+424 -197

No files matched your search

+65 -36
View File
@@ -12,8 +12,8 @@
- 三个圆桌 Router(`intent / recommend / multi_agent`)**通过 httpx ASGITransport 进程内直调本机 ASGI app**,复用既有 `/api/threads`、`/api/agents`、`/api/threads/{id}/runs/stream` 路由——**不走 TCP / 不经过网络栈**。所有 client 都由 `multi_agent._make_loopback_client(request)` 工厂统一构造。 - 三个圆桌 Router(`intent / recommend / multi_agent`)**通过 httpx ASGITransport 进程内直调本机 ASGI app**,复用既有 `/api/threads`、`/api/agents`、`/api/threads/{id}/runs/stream` 路由——**不走 TCP / 不经过网络栈**。所有 client 都由 `multi_agent._make_loopback_client(request)` 工厂统一构造。
- 历史包袱:早期实现走 `httpx.AsyncClient(trust_env=False)` + `127.0.0.1:<port>`,遇到开发机 `HTTP_PROXY` / `HTTPS_PROXY` 时仍可能把回环走代理(症状 `headers_in=42.160s`)。**改用 ASGITransport 后此问题不再存在**,相关 `trust_env=False` workaround 全部移除。 - 历史包袱:早期实现走 `httpx.AsyncClient(trust_env=False)` + `127.0.0.1:<port>`,遇到开发机 `HTTP_PROXY` / `HTTPS_PROXY` 时仍可能把回环走代理(症状 `headers_in=42.160s`)。**改用 ASGITransport 后此问题不再存在**,相关 `trust_env=False` workaround 全部移除。
- **席位 agent 持久注册**(磁盘 `.deer-flow/agents/` + `_sync_legacy_agents`),**协调智能体每次新建**(命名 `roundtable-coordinator-{timestamp36}`,DB 累积,需后续 GC)。 - **席位 agent 持久注册**(磁盘 `.deer-flow/agents/` + `_sync_legacy_agents`),**协调智能体也是持久单例**(id 固定为 `roundtable-coordinator`,与 intent/recommender 同样走 `_roundtable_seed.ensure_roundtable_functional_agents()` 缺失自动落地)。2026-05 之前曾按 `roundtable-coordinator-{timestamp36}` 每会话新建并把动态席位列表写进 SOUL,导致 `.deer-flow/agents/` 累积大量壳目录;改造后 SOUL 是静态通用版本,「本次可调度席位列表」由 `multi_agent.init` 在 coordinator thread 创建后以 human 消息追加到对话历史,leader run 每轮都能看到。
- **`.deer-flow/` 整个目录被 git ignore**(仓库根 `.gitignore` 第 64 / 74 行),`git clone` 后内置 agent 目录在部署机上**不会存在**——任何环境都必须把 `roundtable-intent` + `roundtable-recommender` 两个**功能型 agent** 持久化进发布物,否则 `/api/intent/stream` 报 500。**互联网 demo 还需要 6 个示例研讨席位**(`roundtable-intelligence` 等),**内网部署用客户自有 agent 替代**这些席位,无需打包。详见 §11.7。 - **`.deer-flow/` 整个目录被 git ignore**(仓库根 `.gitignore` 第 64 / 74 行),`git clone` 后内置 agent 目录在部署机上**不会存在**。但功能型 agent 现在通过 `_roundtable_seed.ensure_roundtable_functional_agents()` **自动落地**(首次任何 intent/recommend/multi-agent 路由调用都会触发),因此**新部署不再需要手动准备** `roundtable-intent` / `roundtable-recommender` / `roundtable-coordinator` 三个目录——它们都是 seeder 自动写盘的。**互联网 demo 仍需要 6 个示例研讨席位**(`roundtable-intelligence` 等),**内网部署用客户自有 agent 替代**这些席位,无需打包。详见 §11.7。
- **`init` 是 SSE 而非一次性 JSON**:用 `asyncio.as_completed` 按真实完成顺序推送 `seat_ready`,前端因此能逐个点亮席位。 - **`init` 是 SSE 而非一次性 JSON**:用 `asyncio.as_completed` 按真实完成顺序推送 `seat_ready`,前端因此能逐个点亮席位。
- **`init_done.thread_ids` 包含协调智能体自身**(`agent_threads[main_agent_name] = coord_thread`),前端 Step 3 列文件时需自行过滤。 - **`init_done.thread_ids` 包含协调智能体自身**(`agent_threads[main_agent_name] = coord_thread`),前端 Step 3 列文件时需自行过滤。
- **澄清协议统一**:Step 1 与 Step 2 leader 都借 `ask_clarification` + 终态 `clarification` 帧驱动同一套选项卡片组件;`allow_multiple` 默认 false(单选、点击即发送)。 - **澄清协议统一**:Step 1 与 Step 2 leader 都借 `ask_clarification` + 终态 `clarification` 帧驱动同一套选项卡片组件;`allow_multiple` 默认 false(单选、点击即发送)。
@@ -282,33 +282,50 @@ offline-backend-20260512/backend/
请求体: 请求体:
{ {
"agent_names": ["roundtable-intelligence", ...], // 用户选的席位 id "agent_names": ["roundtable-intelligence", ...], // 用户选的席位 id
"main_agent_name": "roundtable-coordinator-xxx", // 当次总控 id,前端生成 "main_agent_name": "<可选,已废弃>", // 历史字段;2026-05 改造后后端忽略,统一用 COORDINATOR_AGENT_ID
"model": "deepseek-chat" // 可选,写入 coordinator agent "model": "deepseek-chat" // 可选;不传由 lead_agent 回退到 config.yaml models[0]
} }
``` ```
**协调智能体改造说明(2026-05)**:以前 `main_agent_name` 由前端用时间戳生成
(`roundtable-coordinator-${Date.now().toString(36)}`),后端 `_create_coordinator_direct`
把动态拼接的席位列表 SOUL 写盘并入 DB,导致 `.deer-flow/agents/` 越用越多。改造后:
- **id 固定**:`COORDINATOR_AGENT_ID = "roundtable-coordinator"`(前后端常量必须保持一致);
- **缺失自动落地**:路由入口调 `ensure_roundtable_functional_agents()`,与 intent/recommender 共用同一套 seeder;
- **席位列表运行时注入**:init 创建好 coordinator thread 之后,立刻通过 `_append_thread_message` 把「【圆桌系统初始化】本次可调度席位:…」作为 human 消息追加到 thread 历史最前面,leader run 每轮都能看到;
- **`main_agent_name` 入参保留为向后兼容**:旧 client 仍可传,后端只是日志里看到不会用它创建新 agent。
**早期校验**(同步抛 HTTP 4xx,**不**进 SSE): **早期校验**(同步抛 HTTP 4xx,**不**进 SSE):
- `agent_names` 必须非空 - `agent_names` 必须非空
- 每个 `agent_names` 元素 + `main_agent_name` 都需通过 `_AGENT_ID_PATTERN = ^[A-Za-z0-9_-]+$`(用 `_validate_agent_id` 提前 400;上游若收到不合规 id 会抛 422,提前拦更友好) - 每个 `agent_names` 元素需通过 `_AGENT_ID_PATTERN = ^[A-Za-z0-9_-]+$`(用 `_validate_agent_id` 提前 400;上游若收到不合规 id 会抛 422,提前拦更友好)
- 所有 id 都 `.lower()` 后比较 / 存储 - 所有 id 都 `.lower()` 后比较 / 存储
- `main_agent_name` 入参不再做校验(被忽略,统一用 `COORDINATOR_AGENT_ID` 常量)
**SSE 内部流程**: **SSE 内部流程**(2026-05 单例改造后):
``` ```
ensure_roundtable_functional_agents()
// 路由入口:保证 .deer-flow/agents/roundtable-coordinator/ 存在
// 缺失时由 _roundtable_seed.COORDINATOR_AGENT_SOUL 落地静态 SOUL
asyncio.as_completed(各席位 _prepare_sub_agent) asyncio.as_completed(各席位 _prepare_sub_agent)
每完成一个 → data: {"event":"seat_ready","agent_name","thread_id","display_name","description"} 每完成一个 → data: {"event":"seat_ready","agent_name","thread_id","display_name","description"}
(按真实完成顺序,先完成先发,前端因此能逐个点亮) (按真实完成顺序,先完成先发,前端因此能逐个点亮)
异常时:取消所有未完成的 task,避免泄露到 AsyncClient close 之后 异常时:取消所有未完成的 task,避免泄露到 AsyncClient close 之后
协调智能体: 协调智能体(单例):
POST /api/threads → coord_thread main_agent_name = COORDINATOR_AGENT_ID # 固定常量,不再每次新建 agent
POST /api/agents coord_thread = _create_thread_direct(...) # 仍然每次新 thread,会话独立
skills=["agent_orchestration"]
soul=动态拼接(本次 init 的席位列表 + 派活规则)
id=main_agent_name(前端生成的 roundtable-coordinator-<ts36>)
agent_threads[main_agent_name] = coord_thread // ⚠️ thread_ids 含 coordinator 自身 agent_threads[main_agent_name] = coord_thread // ⚠️ thread_ids 含 coordinator 自身
# 通过 HTTP loopback append 一条 human 消息,把本次席位清单写入对话历史:
# "【圆桌系统初始化】本次研讨可调度的席位清单如下..."
# leader run 每轮都能从对话历史里读到,等效于旧版动态 SOUL 的作用。
_append_thread_message(seed_client, ..., coord_thread, seat_introduction)
→ data: {"event":"coordinator_ready","main_agent_name","thread_id"} → data: {"event":"coordinator_ready","main_agent_name","thread_id"}
→ data: {"event":"init_done","main_agent_name","thread_ids":{...}} → data: {"event":"init_done","main_agent_name","thread_ids":{...}}
@@ -316,28 +333,26 @@ asyncio.as_completed(各席位 _prepare_sub_agent)
失败 → data: {"event":"error","detail":"..."} 失败 → data: {"event":"error","detail":"..."}
``` ```
**协调智能体创建幂等**:若 `_create_agent` 返回 (False, "exists ...")(同名 agent 已存在)则继续往下走,不报错。注意这里的「已存在」会复用历史 agent 的 SOUL(与本次动态拼接的不同),通常只在前端用同一个时间戳跑过两遍时触发。 **协调智能体单例 SOUL**(磁盘上唯一一份,见 `_roundtable_seed.COORDINATOR_AGENT_SOUL`):
**协调智能体 SOUL**(动态拼接,与席位列表一起写入): 通用协调规则,**不含**席位列表。模型派活时从对话历史最前面的「【圆桌系统初始化】…」消息读取本次可用 agent_name。仍然强调英文 id(避免把中文显示名拼进 `agent_name`),靠后端 `_strip_display_suffix` 兜底(详见 §6.4)。
**席位清单注入消息形状**(init 阶段写入 coordinator thread):
``` ```
#核心定位: 【圆桌系统初始化】本次研讨可调度的席位清单如下,请严格按 agent_name 通过 agent_orchestration 技能派活,不要派给不在清单内的 agent:
用途:根据用户的任务目标,调用其他智能体干活,只负责协调汇总, - agent_name: roundtable-intelligence(情报收集):<description>
当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果 - agent_name: roundtable-environment(环境评估):<description>
#调度方法:
通过使用 agent_orchestration 技能调用其他智能体干活,使用技能时传入
agent_name 和需要做的活。
⚠️ agent_name 必须严格使用下方列出的英文 id(例如 roundtable-intelligence),
不要把括号里的中文显示名拼进 agent_name 里。
#可调度智能体:
- agent_name: roundtable-intelligence(情报收集):<description>
- agent_name: roundtable-environment(环境评估):<description>
... ...
``` ```
中文显示名通过 `GET /api/agents/{id}` 取 `name`、描述取 `description`。**强调英文 id** 是因为部分模型会把整段 `roundtable-x(中文)` 当 `agent_name` 塞回来,靠后端 `_strip_display_suffix` 兜底(详见 §6.4)。 中文显示名通过 `_get_agents_batch_direct` 一次 DB 查询取 `name`、描述取 `description`。
**单例 agent 的几个语义保障**:
1. **agent 落地幂等**:`_ensure_one` 检查目录是否存在,存在就 no-op,不覆盖运维手改的 SOUL;
2. **席位列表 append 失败不阻断**:`_append_thread_message` 包在 try/except 里,失败只记 warning,leader run 仍能跑(只是看不到席位清单,可能派活到错的 id,由 `_validate_agent_id` 在 run/stream 里报错);
3. **`_create_coordinator_direct` 已删除**:旧函数已移除(2026-05 改造),避免后续维护者误调;如果你正在 review 旧 PR 看到这个函数,那是 pre-rebase 版本。
### 6.2 `/run/stream` — `agent_type = "leader"`(`_leader_run`) ### 6.2 `/run/stream` — `agent_type = "leader"`(`_leader_run`)
@@ -609,13 +624,20 @@ httpx.AsyncClient(
**已废止的 `trust_env=False` workaround**:旧版本走 `127.0.0.1:<port>` 真实 HTTP,需要 `trust_env=False` 阻止 httpx 读 `HTTP_PROXY` 把回环也走代理(症状日志 `headers_in=42.160s`)。改 ASGITransport 后这个问题在物理上已不存在——根本没走网络栈。 **已废止的 `trust_env=False` workaround**:旧版本走 `127.0.0.1:<port>` 真实 HTTP,需要 `trust_env=False` 阻止 httpx 读 `HTTP_PROXY` 把回环也走代理(症状日志 `headers_in=42.160s`)。改 ASGITransport 后这个问题在物理上已不存在——根本没走网络栈。
### 9.2 协调智能体每次新建 ### 9.2 协调智能体单例(2026-05 改造)
`main_agent_name` 形如 `roundtable-coordinator-{timestamp36}`,会话隔离;长期会累积 agent 目录(实际仓库已有数十个 `roundtable-coordinator-mp*`),需后续 GC 策略: 历史实现 `main_agent_name = roundtable-coordinator-{timestamp36}`,会话隔离,但长期会累积 agent 目录(仓库里曾积累几十个 `roundtable-coordinator-mp*` 壳目录)。2026-05 改造选用上方决策 §9.2 选项 C 的本质:**把 coordinator 实例化为「内置 + 静态 SOUL + 运行时注入席位列表」**,固定 id `roundtable-coordinator`,与 `roundtable-intent` / `roundtable-recommender` 同样走 `ensure_roundtable_functional_agents()` 缺失自动落地。
- 选项 A:定时任务清理 N 天前无活跃 thread 的 coordinator agent 实现要点:
- 选项 B:每次创建前 LRU 清理(注意:跨服务实例难协调)
- 选项 C:把 coordinator 实例化为「内置 + 动态 SOUL」复用同一个 id - **磁盘 SOUL 是通用版本**:不含本次席位列表,只规定协调规则、工具用法、终止条件;
- **席位列表运行时注入**:init 创建 coordinator thread 后立刻 `_append_thread_message` 写入一条 human 消息("【圆桌系统初始化】…"),位置在对话历史最前面,leader run 每轮自然读到;
- **`_create_coordinator_direct` 已删除**:不再有"每次创建"路径。
**遗留 `roundtable-coordinator-*` 目录的处理**:仓库可能仍有几十个历史时间戳壳目录。改造后**不会**再生成新的,但旧目录目前不自动清理(避免破坏可能仍存在的旧草稿引用)。运维可通过:
- 在 admin agents 管理页逐个 DELETE;
- 或写一次性脚本扫描 `.deer-flow/agents/roundtable-coordinator-*` 并删除目录 + DB 记录(注意不要删除 `roundtable-coordinator` 本体)。
### 9.3 澄清协议统一 ### 9.3 澄清协议统一
@@ -705,15 +727,22 @@ offline-backend-20260512/backend/.deer-flow/*
| Agent | 角色 | 是否必须打包发布 | | Agent | 角色 | 是否必须打包发布 |
|-------|------|------------------| |-------|------|------------------|
| `roundtable-intent` | **功能型**:Step 1 任务理解,所有部署都用它 | ✅ **必须**(任何环境) | | `roundtable-intent` | **功能型**:Step 1 任务理解,所有部署都用它 | ✅ seeder 自动落地(无需手工打包) |
| `roundtable-recommender` | **功能型**:Step 1→2 之间从候选池里挑推荐,所有部署都用它 | ✅ **必须**(任何环境) | | `roundtable-recommender` | **功能型**:Step 1→2 之间从候选池里挑推荐,所有部署都用它 | ✅ seeder 自动落地(无需手工打包) |
| `roundtable-coordinator` | **功能型单例**:Step 2 总控协调,2026-05 起为单例(id 固定) | ✅ seeder 自动落地(无需手工打包) |
| `roundtable-intelligence` | 示例研讨席位(情报收集) | ⚠️ **互联网 demo 用**,内网不需要 | | `roundtable-intelligence` | 示例研讨席位(情报收集) | ⚠️ **互联网 demo 用**,内网不需要 |
| `roundtable-environment` | 示例研讨席位(环境评估) | ⚠️ 同上 | | `roundtable-environment` | 示例研讨席位(环境评估) | ⚠️ 同上 |
| `roundtable-solution-design` | 示例研讨席位(方案设计) | ⚠️ 同上 | | `roundtable-solution-design` | 示例研讨席位(方案设计) | ⚠️ 同上 |
| `roundtable-risk-review` | 示例研讨席位(风险审查) | ⚠️ 同上 | | `roundtable-risk-review` | 示例研讨席位(风险审查) | ⚠️ 同上 |
| `roundtable-execution-plan` | 示例研讨席位(执行规划) | ⚠️ 同上 | | `roundtable-execution-plan` | 示例研讨席位(执行规划) | ⚠️ 同上 |
| `roundtable-summary` | 示例研讨席位(总结陈述) | ⚠️ 同上 | | `roundtable-summary` | 示例研讨席位(总结陈述) | ⚠️ 同上 |
| `roundtable-coordinator-*` | 协调智能体壳,每次会话动态新建 | ❌ **不打包**(运行时生成) | | `roundtable-coordinator-*`(带后缀) | 2026-05 之前每次会话动态新建的临时壳 | ❌ **遗留**(仓库可能仍有几十个旧目录,新版本不再生成;运维按需清理) |
**关键变化(2026-05)**:三个功能型 agent(intent / recommender / coordinator)现在都由
``_roundtable_seed.ensure_roundtable_functional_agents()`` **自动落地**——任何 intent /
recommend / multi-agent 路由的入口都会触发,首次请求时缺什么补什么。所以**新部署的
机器 `git clone` 后既不需要手工拷目录、也不需要走下文的 `.gitignore` 白名单方案**,
SOUL 改动跟着 Python 代码走 git review 即可。
**关键区分**: **关键区分**:
- `roundtable-intent` / `roundtable-recommender` 是**圆桌功能本身的一部分**——没有它们,整个 Step 1 + 推荐流程跑不起来,跟客户业务无关 - `roundtable-intent` / `roundtable-recommender` 是**圆桌功能本身的一部分**——没有它们,整个 Step 1 + 推荐流程跑不起来,跟客户业务无关
@@ -16,6 +16,7 @@
- **不使用 LangGraph SDK**:所有 API 都是自研的 `apiFetch` + `ReadableStream` SSE 解析(见 `api/intent.ts`、`api/recommend.ts`、`api/multi-agent.ts`)。 - **不使用 LangGraph SDK**:所有 API 都是自研的 `apiFetch` + `ReadableStream` SSE 解析(见 `api/intent.ts`、`api/recommend.ts`、`api/multi-agent.ts`)。
- **不写死模型名**:与 `/page/canvas/ai-writing` 同样的策略 —— 全局选中模型由 `selectedModel` state 维护,首次 `useModels()` 完成后初始化为 `availableModels[0].name`,顶栏「模型」下拉可切换。所有 API 调用统一传 `model: selectedModel || undefined`;后端缺省由 `lead_agent._resolve_model_name()` 回退到 `config.yaml` 的 `models[0]`。**内网部署只需在 `config.yaml` 配自家模型即可,前后端均无需改动**。 - **不写死模型名**:与 `/page/canvas/ai-writing` 同样的策略 —— 全局选中模型由 `selectedModel` state 维护,首次 `useModels()` 完成后初始化为 `availableModels[0].name`,顶栏「模型」下拉可切换。所有 API 调用统一传 `model: selectedModel || undefined`;后端缺省由 `lead_agent._resolve_model_name()` 回退到 `config.yaml` 的 `models[0]`。**内网部署只需在 `config.yaml` 配自家模型即可,前后端均无需改动**。
- **席位区间硬约束**:推荐弹窗要求 **2 ≤ 选中数 ≤ 8**(`MIN_SELECTED` / `MAX_SELECTED`)。 - **席位区间硬约束**:推荐弹窗要求 **2 ≤ 选中数 ≤ 8**(`MIN_SELECTED` / `MAX_SELECTED`)。
- **协调智能体单例(2026-05)**:`COORDINATOR_AGENT_ID = "roundtable-coordinator"` 是固定常量,不再每会话生成新 id。后端 seeder 自动落地,「本次席位列表」改由后端通过 thread 历史 human 消息注入,前端无感。`newCoordinatorName()` 仅作为兼容性壳保留。
- **总控编排循环上限**:`MAX_CYCLES = 8`,超出后强制停止并 toast。 - **总控编排循环上限**:`MAX_CYCLES = 8`,超出后强制停止并 toast。
- **席位派活策略**:**顺序**(不是并行)。旧实现并行触发多路 SSE 会让浏览器卡顿,现已改为顺序。 - **席位派活策略**:**顺序**(不是并行)。旧实现并行触发多路 SSE 会让浏览器卡顿,现已改为顺序。
- **调试开关**:`window.__MULTI_AGENT_INIT_TIMING__` 与 `window.__MULTI_AGENT_DEBUG__`(**不是** `__MULTI_AGENT_STREAM_DEBUG__`,文档历史版本有误)。 - **调试开关**:`window.__MULTI_AGENT_INIT_TIMING__` 与 `window.__MULTI_AGENT_DEBUG__`(**不是** `__MULTI_AGENT_STREAM_DEBUG__`,文档历史版本有误)。
@@ -341,7 +342,12 @@ React 18 StrictMode 开发环境下,effect 会 mount → cleanup → mount 两
| `agentConfigs` | `Map<seatId, {model, weight, prompt, temp}>` Persona 覆盖;下一轮该席位的派活会用此 model | | `agentConfigs` | `Map<seatId, {model, weight, prompt, temp}>` Persona 覆盖;下一轮该席位的派活会用此 model |
| `isUpperPanelCollapsed` | 是否折叠圆桌可视化 + 调控台,给对话区让出垂直空间 | | `isUpperPanelCollapsed` | 是否折叠圆桌可视化 + 调控台,给对话区让出垂直空间 |
**协调智能体命名**:`roundtable-coordinator-${Date.now().toString(36)}`,每次 Step 2 会话新建。后端把这类壳 agent 留在 DB 中,因此 `filterRecommendCandidates` 默认过滤掉。 **协调智能体命名(2026-05 改造)**:现在是**固定单例 id** `roundtable-coordinator`(见 `lib/roundtable-constants.ts::COORDINATOR_AGENT_ID`)。`newCoordinatorName()` 函数为兼容性保留,内部直接返回这个常量字符串。后端 `ensure_roundtable_functional_agents()` 缺失自动落地,不再每次新建。`filterRecommendCandidates` 在 `EXCLUDED_AGENT_IDS` 集合里同时排除:
- `roundtable-coordinator`(新版单例 id,永远过滤);
- `roundtable-coordinator-*` 前缀(历史残留,管理员清理之前先在 UI 隐藏)。
> 历史草稿兼容性:用户加载 2026-05 前保存的草稿时,存储的 `coordinatorName` 仍是 `roundtable-coordinator-<ts36>` 形式。只要那个旧 agent 目录还在磁盘上(即运维没手动删过),加载后继续在 Step 2 干活就能正常工作;如果旧 agent 被清理掉了,需要让用户「重新起草」从头跑一遍 Step 1 → Step 2。
### 5.4 编排循环 `runOrchestration(seedMessage, ids, coord)` ### 5.4 编排循环 `runOrchestration(seedMessage, ids, coord)`
@@ -68,13 +68,20 @@ export interface RecommendCandidate {
* Agent ids that must never enter the candidate pool. Centralized here so * Agent ids that must never enter the candidate pool. Centralized here so
* every caller (recommend dialog, future tools) applies the same rule. * every caller (recommend dialog, future tools) applies the same rule.
* *
* - `roundtable-intent` — Step 1 clarification specialist, no role in research. * - `roundtable-intent` — Step 1 clarification specialist, no role in research.
* - `roundtable-recommender`— the recommender itself, can't recommend itself. * - `roundtable-recommender` — the recommender itself, can't recommend itself.
* - `roundtable-coordinator-*` — disposable Step 2 coordinator shells (DB * - `roundtable-coordinator` — Step 2 coordinator singleton (2026-05 改造后固定 id,
* accumulates one per session); meaningless as roundtable participants. * ensure 自动落地;它是"协调者",不能反过来作为席位被协调)。
* - `roundtable-coordinator-*` 前缀 — 历史残留:2026-05 之前每次 init 会创建
* `roundtable-coordinator-<ts36>` 形式的临时 agent,DB 仍可能保留这些遗留记录。
* 保留前缀过滤可以在管理员清理之前先把它们隐藏起来,避免出现在推荐弹窗候选池里。
*/ */
const EXCLUDED_AGENT_ID_PREFIX = "roundtable-coordinator-"; const EXCLUDED_AGENT_ID_PREFIX = "roundtable-coordinator-";
const EXCLUDED_AGENT_IDS = new Set<string>(["roundtable-intent", "roundtable-recommender"]); const EXCLUDED_AGENT_IDS = new Set<string>([
"roundtable-intent",
"roundtable-recommender",
"roundtable-coordinator",
]);
/** /**
* Drop excluded ids from a raw `GET /api/agents` list and project each entry * Drop excluded ids from a raw `GET /api/agents` list and project each entry
@@ -833,16 +833,19 @@ export function useStep2Orchestration(
const markSpeakerStreaming = useCallback( const markSpeakerStreaming = useCallback(
(speakerId: number) => { (speakerId: number) => {
setActiveSpeakerSet((prev) => { // 编排循环已经改成**顺序**派 special(详见 §5.4 / 文件顶部注释),
if (prev.has(speakerId)) return prev; // 任意时刻只有一个子智能体在跑。因此这里既要把当前发言席位置为"研讨中",
const next = new Set(prev); // 又要把上一轮 / 历史残留下来的"研讨中"统一清成"在线",保证右侧"参与角色"
next.add(speakerId); // 面板任何时刻**只有一个** sub agent 显示研讨中,与对话区实际在 stream 的
return next; // 那一个一致。activeSpeakerSet 沿用 Set 形态(虽然实际只放 1 个)以兼容外部
}); // 消费方,真正的"高亮排他"靠 roles 的 status reset。
setActiveSpeakerSet(() => new Set([speakerId]));
setRoles((prev) => setRoles((prev) =>
prev.map((item) => prev.map((item) => {
item.id === speakerId ? { ...item, status: "研讨中" } : item, if (item.id === speakerId) return { ...item, status: "研讨中" };
), if (item.status === "研讨中") return { ...item, status: "在线" };
return item;
}),
); );
}, },
[setRoles], [setRoles],
@@ -912,7 +915,14 @@ export function useStep2Orchestration(
safetyCounter < MAX_CYCLES safetyCounter < MAX_CYCLES
) { ) {
safetyCounter += 1; safetyCounter += 1;
highlightSpeaker(2); // Leader(协调智能体)正在派活,**不应该**让任何 sub agent 显示"研讨中"——
// 协调者本身不在右侧"参与角色"面板里(COMMANDER_SEAT_ID=1 是人类主持人,
// sub agent 从 id=2 起)。历史代码这里写的是 `highlightSpeaker(2)`,会
// 错把第一个 sub agent 误标为研讨中,叠加 special 派活时另一个 sub 也被
// markSpeakerStreaming,就出现"两个 sub 同时研讨中"的视觉 bug。
// 改成 null 后:leader 跑的阶段所有 sub 都是"在线",special 跑哪个就只
// 高亮哪个。activeSpeakerId 同步置空,圆桌可视化也不会高亮错的位置。
highlightSpeaker(null);
// 总控也可能直接 write_file —— 之前只接了 special 流,导致总控写文件时沙箱不自动打开。 // 总控也可能直接 write_file —— 之前只接了 special 流,导致总控写文件时沙箱不自动打开。
// 这里把 leader thread id 绑到气泡 + artifact handler 上,与 special 同样的行为。 // 这里把 leader thread id 绑到气泡 + artifact handler 上,与 special 同样的行为。
@@ -130,8 +130,27 @@ export function nowTime(): string {
return new Date().toLocaleTimeString("zh-CN", { hour12: false }); return new Date().toLocaleTimeString("zh-CN", { hour12: false });
} }
/**
* 圆桌总控协调智能体的**固定单例 id**。后端 `_roundtable_seed.COORDINATOR_AGENT_ID`
* 与本常量必须保持一致(2026-05 改造前是 `roundtable-coordinator-<ts36>` 形式,
* 每次会话生成新 id 导致 `.deer-flow/agents/` 累积大量壳目录)。
*
* 改造后:
* - 后端 ensure 机制保证磁盘上始终存在 `.deer-flow/agents/roundtable-coordinator/`;
* - 本次会话的「可调度席位列表」由后端 init 通过 human 消息注入 coordinator
* thread 历史,不再写入 SOUL,所以单例 SOUL 长期稳定;
* - 前端 `streamMultiAgent({ coordinatorName })` 等位置不需要任何改动,因为
* `coordinatorName` 仍然是一个**字符串**,只是值不再每次变化。
*/
export const COORDINATOR_AGENT_ID = "roundtable-coordinator";
/**
* 兼容性保留:历史上叫 `newCoordinatorName()`,意为"生成新的一次性 coordinator id"。
* 改造为单例后实际只返回常量;命名不再准确,但留下避免触发跨文件级联改名。
* 新代码请直接使用 {@link COORDINATOR_AGENT_ID}。
*/
export function newCoordinatorName(): string { export function newCoordinatorName(): string {
return `roundtable-coordinator-${Date.now().toString(36)}`; return COORDINATOR_AGENT_ID;
} }
/** /**
@@ -512,11 +512,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
_mark_legacy_sync_fresh, _mark_legacy_sync_fresh,
_sync_legacy_agents, _sync_legacy_agents,
) )
from app.gateway.services import _mark_legacy_sync_fresh as _services_mark_sync_fresh
agent_store = getattr(app.state, "agent_store", None) agent_store = getattr(app.state, "agent_store", None)
if agent_store is not None: if agent_store is not None:
await _sync_legacy_agents(agent_store) await _sync_legacy_agents(agent_store)
_mark_legacy_sync_fresh() _mark_legacy_sync_fresh()
# ``services.py`` keeps an independent TTL gate (used by
# ``start_run`` → ``_enforce_agent_access``). Without this
# ack, the first ``/runs/stream`` after startup would
# re-run the heavy sync even though we just did it.
_services_mark_sync_fresh()
logger.info("Legacy folder-backed agents synced at startup") logger.info("Legacy folder-backed agents synced at startup")
except Exception: except Exception:
logger.exception("Legacy agent sync failed at startup (non-fatal)") logger.exception("Legacy agent sync failed at startup (non-fatal)")
@@ -2,21 +2,35 @@
适用场景 适用场景
-------- --------
``roundtable-intent`` / ``roundtable-recommender`` 是圆桌 Step 1 + 推荐链路的 ``roundtable-intent`` / ``roundtable-recommender`` / ``roundtable-coordinator``
**功能型** 内置 agent,任何部署都必须存在。但仓库根 ``.gitignore`` 把整个 是圆桌 Step 1 + 推荐 + Step 2 总控链路的**功能型** 内置 agent,任何部署都必须
``.deer-flow/`` 目录排除,新机器 ``git clone`` 之后这两个目录**不会存在**, 存在。但仓库根 ``.gitignore`` 把整个 ``.deer-flow/`` 目录排除,新机器
直接调 ``/api/intent/stream`` 会因为 agent 解析失败抛 500。 ``git clone`` 之后这些目录**不会存在**,直接调对应 endpoint 会因为 agent
解析失败抛 500。
本模块在三个路由入口(``intent.init``、``intent.stream``、``recommend.stream``) 本模块在四个路由入口(``intent.init``、``intent.stream``、``recommend.stream``、
被调用,检查磁盘文件是否齐全;缺失则用下方嵌入的标准模板写回去。 ``multi_agent.init``)被调用,检查磁盘文件是否齐全;缺失则用下方嵌入的标准模板
``_sync_legacy_agents`` 会在下一次 ``GET /api/agents`` 时把目录 upsert 进 DB, 写回去。``_sync_legacy_agents`` 会在下一次 ``GET /api/agents`` 时把目录 upsert
所以这里只关心磁盘文件,不直接动 DB。 进 DB,所以这里只关心磁盘文件,不直接动 DB。
为什么嵌入字符串而不是用 importlib.resources 资源文件 为什么嵌入字符串而不是用 importlib.resources 资源文件
------------------------------------------------------ ------------------------------------------------------
SOUL 体量不大(两个加起来 ~200 行),嵌入到 Python 文件里是**单一可信源**——改 SOUL 体量不大(三个加起来 ~400 行),嵌入到 Python 文件里是**单一可信源**——改
SOUL 走代码 review、commit 走 git 历史,且不会因为 wheel 打包漏配 SOUL 走代码 review、commit 走 git 历史,且不会因为 wheel 打包漏配
``package-data`` 导致部署机拿不到。 ``package-data`` 导致部署机拿不到。
为什么 ``roundtable-coordinator`` 是单例而不是每次会话新建
------------------------------------------------------------
历史实现每次 Step 2 init 都用 ``roundtable-coordinator-<timestamp36>`` 这种
带时间戳的临时 id 创建新 agent,把"本次可调度席位列表"写进动态拼接的 SOUL。
长期跑下来 ``.deer-flow/agents/`` 累积了几十上百个完全相同的 coordinator
壳,管理页非常杂乱。
改造后 coordinator 是 **单例**(id 固定为 ``roundtable-coordinator``):
- SOUL 是**通用协调规则**,不含本次席位列表;
- 本次圆桌可调度的席位清单由 ``multi_agent.init`` 在创建好 coordinator
thread 后,以 human 消息的形式 append 到 thread 历史,让 leader run 自然
读到。这样 SOUL 长期稳定,席位列表随会话动态注入。
""" """
from __future__ import annotations from __future__ import annotations
@@ -476,17 +490,91 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
""" """
_BUILTIN_FUNCTIONAL_AGENTS: tuple[tuple[str, str, str, str], ...] = ( COORDINATOR_AGENT_ID = "roundtable-coordinator"
(INTENT_AGENT_ID, INTENT_AGENT_NAME, INTENT_AGENT_DESCRIPTION, INTENT_AGENT_SOUL), COORDINATOR_AGENT_NAME = "总控协调"
(RECOMMENDER_AGENT_ID, RECOMMENDER_AGENT_NAME, RECOMMENDER_AGENT_DESCRIPTION, RECOMMENDER_AGENT_SOUL), COORDINATOR_AGENT_DESCRIPTION = (
"圆桌总控协调智能体,基于「任务目标 + 当前会话已选席位列表」决定派活给哪些子智能体、"
"汇总它们的交付,达成共识后输出最终方案。本身不调研、不写文档,只负责协调与综合。"
)
# Coordinator SOUL 是**通用版本**——只规定协调规则、工具用法、终止条件,
# **不**包含本次会话的席位列表。席位列表由 multi_agent.init 在创建 coordinator
# thread 时以 human 消息的形式 append 进对话历史,leader run 每次都能看到。
# 这样 SOUL 单例 + 席位列表动态注入,避免每次 init 重写 SOUL 文件。
COORDINATOR_AGENT_SOUL = """# 圆桌总控协调智能体
## 核心定位
你是多智能体圆桌研讨 **Step 2** 的「总控协调智能体」。你不亲自调研、不写文档、不联网搜索,你的**唯一职责**是:
1. **理解用户任务目标**(由 Step 1 的任务理解专员整理交给你);
2. **从本次圆桌已上线的席位中挑合适的派活**(通过 `agent_orchestration` 技能);
3. **汇总子智能体交付**,达成共识后输出最终结论。
## 调度方法
通过 `agent_orchestration` 技能调用其他智能体干活,使用技能时传入 `agent_name` 和需要做的活。
**⚠️ 关于 `agent_name` 的硬性规则**:
- 必须严格使用**本次圆桌已上线席位的英文 id**(例如 `roundtable-intelligence`,或形如 `02714cbe6e64406a8a4659f0e3d9c0b8` 这样的 UUID)
- 本次可调度的席位清单会以「【圆桌系统初始化】本次可调度席位:...」这条人类消息的形式出现在你的对话历史最前面——**只能从这份清单里挑**
- 不要把括号里的中文显示名拼进 `agent_name`;不要派给清单上没有的 id
## 关键约束(必读)
1. **不要对同一子智能体重复派同样的活**:每次派活前回顾对话历史,如果某个子智能体已经针对你正要派的任务完成过交付(看到形如「子智能体[名称]完成XX工作,交付内容为:...」的 human 消息),就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题
2. **看到交付要会用**:子智能体的交付内容会作为 human 消息追加到你的对话历史里,你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具),让讨论收敛
3. **避免无意义循环**:如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论
4. **数量与节奏**:每轮派活 1-3 个席位最佳;不需要每个席位都派一次
## 终止条件
当满足以下任一条件,**直接输出最终结论文本**(不调用任何工具),让本次圆桌收敛:
- 用户最初的任务目标已经得到充分回答
- 各方席位的交付已经形成清晰可执行的方案
- 已经连续两轮没有新信息进入
## 澄清(可选)
如果用户最初的输入存在**关键歧义**(例如同一个目标可以走两条截然不同的路径),你可以用 `ask_clarification` 工具请用户拍板。**但请克制使用**——大多数情况下你应该基于现有信息开始派活,而不是反过来追问用户。
## 输出风格
- 派活说明:简短(1-2 句话)告诉子智能体你期望它做什么、关注哪些点
- 最终综合:结构化,明确各席位的核心结论、冲突点、最终建议
- 适度 emoji:📋 汇总、⚡ 派活、✅ 完成
## 禁止事项
- ❌ 亲自调研、写文档、写代码、联网搜索(这些是子智能体的活)
- ❌ 派活给本次清单**没有**的 agent_name
- ❌ 把括号里的中文显示名拼进 `agent_name`
- ❌ 已经达成共识后仍然继续派活
"""
# (id, name, description, soul, skills) — skills 为 None 表示不写 skills 字段。
_BUILTIN_FUNCTIONAL_AGENTS: tuple[tuple[str, str, str, str, list[str] | None], ...] = (
(INTENT_AGENT_ID, INTENT_AGENT_NAME, INTENT_AGENT_DESCRIPTION, INTENT_AGENT_SOUL, None),
(RECOMMENDER_AGENT_ID, RECOMMENDER_AGENT_NAME, RECOMMENDER_AGENT_DESCRIPTION, RECOMMENDER_AGENT_SOUL, None),
(
COORDINATOR_AGENT_ID,
COORDINATOR_AGENT_NAME,
COORDINATOR_AGENT_DESCRIPTION,
COORDINATOR_AGENT_SOUL,
["agent_orchestration"],
),
) )
def _ensure_one(agent_id: str, name: str, description: str, soul: str) -> bool: def _ensure_one(
agent_id: str,
name: str,
description: str,
soul: str,
skills: list[str] | None = None,
) -> bool:
"""Create the on-disk files for one functional agent if missing. """Create the on-disk files for one functional agent if missing.
Returns True if files were just created, False if the directory already Returns True if files were just created, False if the directory already
existed (and we didn't touch it — respect any local edits). existed (and we didn't touch it — respect any local edits).
``skills`` 写入 config.yaml 顶层 ``skills`` 字段,与 routers/agents.py 的
``_write_agent_config`` 输出形状对齐;coordinator 需要 ``agent_orchestration``
才能派活,intent / recommender 不需要任何 skill(传 None 即可)。
""" """
agent_dir = get_paths().agent_dir(agent_id) agent_dir = get_paths().agent_dir(agent_id)
if agent_dir.exists(): if agent_dir.exists():
@@ -496,8 +584,13 @@ def _ensure_one(agent_id: str, name: str, description: str, soul: str) -> bool:
# newline="\n" 强制 LF;否则 Windows 下 open() 把 \n 翻成 \r\n,跨 OS 部署时与 # newline="\n" 强制 LF;否则 Windows 下 open() 把 \n 翻成 \r\n,跨 OS 部署时与
# Linux 部署机字节级不一致(功能上 YAML/Markdown 都兼容,但 git diff / 哈希对不齐)。 # Linux 部署机字节级不一致(功能上 YAML/Markdown 都兼容,但 git diff / 哈希对不齐)。
config_payload: dict[str, object] = {"id": agent_id, "name": name, "description": description}
if skills:
# 与 routers/agents.py::_write_agent_config 行为一致:None 不写、空列表也不写,
# 仅当确有 skill 名时才追加顶层 skills 字段。
config_payload["skills"] = list(skills)
config_yaml = yaml.dump( config_yaml = yaml.dump(
{"id": agent_id, "name": name, "description": description}, config_payload,
default_flow_style=False, default_flow_style=False,
allow_unicode=True, allow_unicode=True,
) )
@@ -520,9 +613,9 @@ def ensure_roundtable_functional_agents() -> list[str]:
actionable error if the agent really cannot be resolved. actionable error if the agent really cannot be resolved.
""" """
created: list[str] = [] created: list[str] = []
for agent_id, name, description, soul in _BUILTIN_FUNCTIONAL_AGENTS: for agent_id, name, description, soul, skills in _BUILTIN_FUNCTIONAL_AGENTS:
try: try:
if _ensure_one(agent_id, name, description, soul): if _ensure_one(agent_id, name, description, soul, skills):
created.append(agent_id) created.append(agent_id)
except Exception: except Exception:
logger.exception("Failed to seed roundtable functional agent '%s'", agent_id) logger.exception("Failed to seed roundtable functional agent '%s'", agent_id)
@@ -57,6 +57,10 @@ import uuid
from typing import Any, AsyncIterator from typing import Any, AsyncIterator
import httpx import httpx
from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID,
ensure_roundtable_functional_agents,
)
from deerflow.config import get_app_config from deerflow.config import get_app_config
from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
from deerflow.runtime.user_context import get_effective_user_id from deerflow.runtime.user_context import get_effective_user_id
@@ -443,82 +447,11 @@ async def _get_agents_batch_direct(
return out return out
async def _create_coordinator_direct( # NOTE: 历史上这里曾有 _create_coordinator_direct(),每次 init 都创建一个带时间戳的
request: Request, # roundtable-coordinator-<ts36>。2026-05 改造为单例(见 _roundtable_seed.py 与本
*, # 模块 init_multi_agent 的 docstring),不再需要"每次创建"的直调辅助;改造后由
agent_id: str, # ensure_roundtable_functional_agents() 在路由入口落地静态 SOUL,席位列表通过
name: str, # _append_thread_message 注入到 coordinator thread 的对话历史中。
description: str,
soul: str,
model: str | None,
skills: list[str] | None,
) -> tuple[bool, dict[str, Any] | str]:
"""同进程版的协调 agent 创建。
完整复刻 ``routers/agents.create_agent_endpoint`` 的关键副作用:
- ``agents_api.enabled`` master switch
- 显示名校验(空白则 422 → 这里转 (False, msg))
- 重复 id 检查(409 → (False, msg))
- 写 ``config.yaml`` + ``SOUL.md``(用 ``asyncio.to_thread`` 防同步 IO 阻塞)
- ``agent_store.create`` 入表
返回 ``(True, record)`` 或 ``(False, error_text)``;调用方决定要不要发 SSE error 帧。
"""
from app.gateway.deps import get_agent_store
from app.gateway.routers.agents import (
_require_agents_api_enabled,
_write_agent_config,
_write_agent_soul,
)
from deerflow.config.paths import get_paths
_require_agents_api_enabled()
normalized_name = (name or "").strip()
if not normalized_name:
return False, "Agent name cannot be empty"
store = get_agent_store(request)
if await store.get_any(agent_id) is not None:
return False, f"Agent id '{agent_id}' already exists"
agent_dir = get_paths().agent_dir(agent_id)
if agent_dir.exists():
return False, f"Agent id '{agent_id}' already exists"
try:
# config.yaml / SOUL.md 是同步 IO —— 全部丢线程池,不阻塞 init SSE 的发帧节奏。
await asyncio.to_thread(
_write_agent_config,
agent_id,
name=normalized_name,
description=description or "",
skills=skills,
model=model,
tool_groups=None,
)
await asyncio.to_thread(_write_agent_soul, agent_id, soul)
record = await store.create(
{
"id": agent_id,
"name": normalized_name,
"description": description or "",
"user_id": _resolve_user_id(request),
"published": False,
}
)
return True, record
except Exception as exc:
# 与 routers/agents 失败回滚一致 —— 文件已写出但 DB 失败时清掉 dir,避免下次
# `agent_dir.exists()` 误判为"已存在"导致后续 init 永远 409。
try:
if agent_dir.exists():
import shutil
await asyncio.to_thread(shutil.rmtree, agent_dir)
except Exception:
logger.exception("[init-direct] cleanup after coordinator create failure")
logger.exception("[init-direct] coordinator create failed")
return False, str(exc)
async def _append_thread_message( async def _append_thread_message(
@@ -718,31 +651,44 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> Strea
{ {
"agent_names": ["roundtable-intelligence", ...], // 用户选的席位 id "agent_names": ["roundtable-intelligence", ...], // 用户选的席位 id
"main_agent_name": "roundtable-coordinator-xxx", // 当次总控 id,前端生成 "main_agent_name": "<可选,已废弃>", // 历史兼容字段,新代码忽略,统一用 COORDINATOR_AGENT_ID
"model": "<可选>" // 可选,写入 coordinator agent;不传由 lead_agent 回退到 config.yaml models[0] "model": "<可选>" // 可选;不传由 lead_agent 回退到 config.yaml models[0]
} }
SSE 事件(每条 ``data: {...}\\n\\n``):: SSE 事件(每条 ``data: {...}\\n\\n``)::
seat_ready — 某一席位 thread 已创建(``asyncio.as_completed`` 顺序,先完先发) seat_ready — 某一席位 thread 已创建(``asyncio.as_completed`` 顺序,先完先发)
coordinator_ready — 总控 agent + thread 已就绪 coordinator_ready — 总控 thread 就绪 + 已注入本次席位列表
init_done — ``thread_ids`` 全量映射,前端存 ``threadIds`` 并开始编排 init_done — ``thread_ids`` 全量映射,前端存 ``threadIds`` 并开始编排
error — 失败详情(同步校验失败则直接 HTTP 4xx,不进 SSE) error — 失败详情(同步校验失败则直接 HTTP 4xx,不进 SSE)
协调智能体 SOUL 在运行时根据各席位 ``GET /api/agents/{id}`` 的 name/description 拼接, 协调智能体改造说明
强调 ``agent_name`` 必须用英文 id(``roundtable-intelligence``),禁止带中文括号后缀。 ----------------
协调智能体现在是 **单例**(id 固定为 ``roundtable-coordinator``,SOUL 见
``_roundtable_seed.COORDINATOR_AGENT_SOUL``),由 ``ensure_roundtable_functional_agents()``
在首次调用时自动落地。每次 init **不再创建新的 coordinator agent**(避免
``.deer-flow/agents/`` 累积 ``roundtable-coordinator-<ts36>`` 残留)。
本次圆桌的「可调度席位列表」由 init 在创建好 coordinator thread 之后,以一条
human 消息追加到 thread 历史里,leader run 每轮都能从对话历史里读到。这样
SOUL 长期稳定,席位列表随会话动态注入。前端为兼容旧 client 仍可传
``main_agent_name`` 字段,但后端会忽略并使用 ``COORDINATOR_AGENT_ID``。
""" """
# 调用前自检:与 intent/recommend 一样,缺失时就地补建,避免 /run/stream 报 500。
ensure_roundtable_functional_agents()
agent_names_in = payload.get("agent_names") or [] agent_names_in = payload.get("agent_names") or []
agent_names: list[str] = [str(name).lower() for name in agent_names_in] agent_names: list[str] = [str(name).lower() for name in agent_names_in]
main_agent_name: str = str(payload.get("main_agent_name") or "main-agent").lower()
model: str | None = payload.get("model") model: str | None = payload.get("model")
# main_agent_name 是历史字段,2026-05 改造后忽略,统一用单例 id。保留接收但不再用前端值,
# 避免下游再次创建带时间戳的临时 coordinator。
main_agent_name = COORDINATOR_AGENT_ID
if not agent_names: if not agent_names:
raise HTTPException(status_code=400, detail="agent_names is required") raise HTTPException(status_code=400, detail="agent_names is required")
for name in agent_names: for name in agent_names:
_validate_agent_id(name, "agent_names entry") _validate_agent_id(name, "agent_names entry")
_validate_agent_id(main_agent_name, "main_agent_name")
def _frame(payload: dict[str, Any]) -> str: def _frame(payload: dict[str, Any]) -> str:
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n" return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
@@ -794,56 +740,50 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> Strea
t_seats_done = time.perf_counter() t_seats_done = time.perf_counter()
agent_threads: dict[str, str] = {sid: thread_id for sid, thread_id, _, _ in prepared} agent_threads: dict[str, str] = {sid: thread_id for sid, thread_id, _, _ in prepared}
descriptions: list[str] = [ seat_lines: list[str] = [
f"- agent_name: {sid}({display_name}):{description}" f"- agent_name: {sid}({display_name}):{description}"
for sid, _, display_name, description in prepared for sid, _, display_name, description in prepared
] ]
# 动态 SOUL:把本次 init 的席位列表写进总控 prompt,供 agent_orchestration 派活。 # 协调智能体改造为**单例**(id = roundtable-coordinator):
soul = ( # - SOUL 已经由 ensure_roundtable_functional_agents() 在路由入口落地,
"#核心定位:\n" # 通用协调规则常驻磁盘,无需每次重写。
"用途:根据用户的任务目标,调用其他智能体干活,只负责协调汇总," # - 本次"可调度席位清单"以 human 消息形式 append 到 coordinator
"当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n" # thread 历史里,leader run 每轮都能在对话历史最前面看到。
"#调度方法:\n" # 这样席位列表随会话动态变化,但磁盘上不再累积 roundtable-coordinator-<ts36>。
"通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活。" #
"注意:agent_name 必须严格使用下方列出的英文 id(例如 roundtable-intelligence," # 注意:model 字段在 lead_agent run 时通过 context.model_name 注入,
"或形如 02714cbe6e64406a8a4659f0e3d9c0b8 这样的 UUID)," # 不需要写回到 agent config.yaml(否则会污染单例的配置)。
"不要把括号里的中文显示名拼进 agent_name 里。\n" seat_introduction = (
"#关键约束(必读):\n" "【圆桌系统初始化】本次研讨可调度的席位清单如下,请严格按 agent_name "
"1. **不要对同一子智能体重复派同样的活**:每次派活前回顾对话历史," "通过 agent_orchestration 技能派活,不要派给不在清单内的 agent:\n"
"如果某个子智能体已经针对你正要派的任务完成过交付(" + "\n".join(seat_lines)
"看到形如\"子智能体[名称]完成XX工作,交付内容为:...\"的 human 消息),"
"就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题。\n"
"2. **看到交付要会用**:子智能体的交付内容会作为 human 消息追加到你的对话历史里,"
"你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具),"
"让讨论收敛。\n"
"3. **避免无意义循环**:如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论。\n"
"#可调度智能体:\n"
+ "\n".join(descriptions)
) )
# 3) coordinator thread + agent 并行创建。thread 与 agent 之间无依赖, coord_thread = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
# asyncio.gather 比串行省一个直调往返的开销。
coord_thread_task = asyncio.create_task(
_create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
)
coord_agent_task = asyncio.create_task(
_create_coordinator_direct(
request,
agent_id=main_agent_name,
name=main_agent_name,
description="你是总控控制协调智能体,负责协调其他智能体干活",
soul=soul,
model=model,
skills=["agent_orchestration"],
)
)
coord_thread = await coord_thread_task
agent_threads[main_agent_name] = coord_thread agent_threads[main_agent_name] = coord_thread
ok, result = await coord_agent_task
if not ok and isinstance(result, str) and "exists" not in result: # 把席位列表注入 coordinator thread 的对话历史。HTTP loopback 调用
yield _frame({"event": "error", "detail": f"create coordinator agent failed: {result}"}) # 走 /api/threads/{tid}/state(读+写),开销 ~10-30ms,在 init 整体
return # 几百 ms 的尺度上微不足道,但比起手撕 langgraph checkpoint 内部结构
# 风险低、与 _broadcast / _append_thread_message 一套语义,长期可维护。
try:
base = _loopback_base(request)
headers = _auth_headers(request)
async with _make_loopback_client(request) as seed_client:
await _append_thread_message(
seed_client,
base,
headers,
coord_thread,
seat_introduction,
)
except Exception:
# 注入失败不阻断 init:leader run 仍能跑,只是看不到席位清单,
# 可能派活给错的 id。会被 _strip_display_suffix / _validate_agent_id
# 兜底,前端能看到错误帧;运维通过 backend 日志找到根因再修复。
logger.exception("[init-direct] failed to seed coordinator thread with seat list")
t_coord_done = time.perf_counter() t_coord_done = time.perf_counter()
logger.info( logger.info(
"[init-direct] meta=%.3fs seats=%.3fs coord=%.3fs total=%.3fs seats_count=%d", "[init-direct] meta=%.3fs seats=%.3fs coord=%.3fs total=%.3fs seats_count=%d",
@@ -11,6 +11,7 @@ import asyncio
import json import json
import logging import logging
import re import re
import time
from collections.abc import Mapping from collections.abc import Mapping
from typing import Any from typing import Any
@@ -356,7 +357,17 @@ def _collect_requested_agent_ids(body: Any) -> set[str]:
async def _sync_legacy_agent_records(store) -> None: async def _sync_legacy_agent_records(store) -> None:
"""Import existing folder-backed agents as built-in records.""" """Import existing folder-backed agents as built-in records.
Heavy: ``list_custom_agents()`` scans ``.deer-flow/agents/`` and reads
``config.yaml`` + ``SOUL.md`` per directory, then ``store.ensure_builtin``
does one DB round-trip per agent. On a machine with N agent dirs this is
O(N) file IO + O(N) DB calls — historically the worst per-request hotspot
in ``start_run`` (observed 30s+ on machines with 100+ residual coordinator
shells + slow MySQL). **Never call this directly from request handlers** —
use :func:`_maybe_sync_legacy_agent_records` which applies the same TTL
gating as ``routers/agents._maybe_sync_legacy_agents``.
"""
from deerflow.config.agents_config import list_custom_agents from deerflow.config.agents_config import list_custom_agents
for agent_cfg in list_custom_agents(): for agent_cfg in list_custom_agents():
@@ -373,6 +384,69 @@ async def _sync_legacy_agent_records(store) -> None:
) )
# Throttle per-request legacy-sync. Matches ``routers/agents._LEGACY_SYNC_TTL_SECONDS``
# semantics — the real sync runs once at startup (lifespan); this is a safety
# net for environments that bypass the lifespan hook or for freshly dropped
# agent directories. Lifespan registers itself via the same state by calling
# the agents router's ``_mark_legacy_sync_fresh()`` after its boot-time sync,
# so this module's TTL state is independent (a separate gate kept here so we
# don't import from ``routers/agents`` and create a circular import).
_LEGACY_SYNC_TTL_SECONDS = 3600.0
_legacy_sync_state: dict[str, Any] = {"last_at": 0.0, "lock": None}
# Anything slower than this triggers the WARNING-level breakdown in
# ``start_run``. 500ms gives plenty of headroom for cold caches + a few DB
# round-trips on healthy systems while still catching the pathological
# multi-second cases that brought us here.
_START_RUN_SLOW_THRESHOLD_SECONDS = 0.5
def _mark_legacy_sync_fresh() -> None:
"""Tell this module's TTL gate that a sync just completed elsewhere.
Called by the Gateway lifespan hook after it runs the boot-time legacy
agent sync (``routers/agents._sync_legacy_agents``). Without this, the
first request after startup would race past the TTL gate here and trigger
a redundant rescan even though the data is already fresh.
"""
_legacy_sync_state["last_at"] = time.monotonic()
async def _maybe_sync_legacy_agent_records(store) -> None:
"""TTL-gated wrapper around :func:`_sync_legacy_agent_records`.
Steady-state behavior: returns in microseconds inside the TTL window — no
filesystem walk, no DB round-trips. Outside the TTL window we serialize
callers under an :class:`asyncio.Lock` so a burst of concurrent
``/runs/stream`` requests fans into a single rescan instead of N parallel
fan-outs. Pattern mirrors ``routers/agents._maybe_sync_legacy_agents``.
"""
now = time.monotonic()
if now - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS:
return
lock = _legacy_sync_state["lock"]
if lock is None:
lock = asyncio.Lock()
_legacy_sync_state["lock"] = lock
async with lock:
# Re-check under the lock — a sibling coroutine may have just refreshed.
if time.monotonic() - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS:
return
t_sync_start = time.perf_counter()
await _sync_legacy_agent_records(store)
elapsed = time.perf_counter() - t_sync_start
_legacy_sync_state["last_at"] = time.monotonic()
# When this fires it's a "first request of the hour" warmup — log it
# so operators can see the spike isn't an isolated bug. Anything
# over 2s usually means a lot of agent directories + slow IO/DB.
logger.info(
"[start-run-timing] legacy-agent sync warmed cache in %.3fs (next sync in %.0fs)",
elapsed,
_LEGACY_SYNC_TTL_SECONDS,
)
async def _enforce_agent_access(body: Any, request: Request) -> None: async def _enforce_agent_access(body: Any, request: Request) -> None:
agent_ids = _collect_requested_agent_ids(body) agent_ids = _collect_requested_agent_ids(body)
if not agent_ids: if not agent_ids:
@@ -383,7 +457,7 @@ async def _enforce_agent_access(body: Any, request: Request) -> None:
from deerflow.runtime.user_context import get_effective_user_id from deerflow.runtime.user_context import get_effective_user_id
store = get_agent_store(request) store = get_agent_store(request)
await _sync_legacy_agent_records(store) await _maybe_sync_legacy_agent_records(store)
user_id = get_effective_user_id() user_id = get_effective_user_id()
for agent_id in agent_ids: for agent_id in agent_ids:
try: try:
@@ -417,14 +491,27 @@ async def start_run(
Target thread. Target thread.
request : Request request : Request
FastAPI request — used to retrieve singletons from ``app.state``. FastAPI request — used to retrieve singletons from ``app.state``.
Timing observability
--------------------
Each phase records ``perf_counter`` deltas. When the total wallclock from
function entry to ``run_agent`` task creation exceeds
``_START_RUN_SLOW_THRESHOLD_SECONDS`` we emit a ``[start-run-timing]``
breakdown at WARNING level. Stays at DEBUG when fast, so steady-state
logs are not polluted. Investigate spikes by grep'ing
``[start-run-timing]`` together with ``[upstream-timing]`` (multi-agent
router) and ``[leader-timing] / [special-timing]`` (圆桌 leader/special).
""" """
t_entry = time.perf_counter()
bridge = get_stream_bridge(request) bridge = get_stream_bridge(request)
run_mgr = get_run_manager(request) run_mgr = get_run_manager(request)
run_ctx = get_run_context(request) run_ctx = get_run_context(request)
disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_ disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_
t_pre_enforce = time.perf_counter()
await _enforce_agent_access(body, request) await _enforce_agent_access(body, request)
t_post_enforce = time.perf_counter()
try: try:
record = await run_mgr.create_or_reject( record = await run_mgr.create_or_reject(
@@ -439,6 +526,7 @@ async def start_run(
raise HTTPException(status_code=409, detail=str(exc)) from exc raise HTTPException(status_code=409, detail=str(exc)) from exc
except UnsupportedStrategyError as exc: except UnsupportedStrategyError as exc:
raise HTTPException(status_code=501, detail=str(exc)) from exc raise HTTPException(status_code=501, detail=str(exc)) from exc
t_post_create = time.perf_counter()
# Upsert thread metadata so the thread appears in /threads/search, # Upsert thread metadata so the thread appears in /threads/search,
# even for threads that were never explicitly created via POST /threads # even for threads that were never explicitly created via POST /threads
@@ -473,6 +561,7 @@ async def start_run(
await run_ctx.thread_store.update_assistant_id(thread_id, body.assistant_id) await run_ctx.thread_store.update_assistant_id(thread_id, body.assistant_id)
except Exception: except Exception:
logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id)) logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
t_post_thread_upsert = time.perf_counter()
agent_factory = resolve_agent_factory(body.assistant_id) agent_factory = resolve_agent_factory(body.assistant_id)
graph_input = normalize_input(body.input) graph_input = normalize_input(body.input)
@@ -511,6 +600,7 @@ async def start_run(
stream_modes = normalize_stream_modes(body.stream_mode) stream_modes = normalize_stream_modes(body.stream_mode)
t_pre_task = time.perf_counter()
task = asyncio.create_task( task = asyncio.create_task(
run_agent( run_agent(
bridge, bridge,
@@ -535,6 +625,33 @@ async def start_run(
# title from the checkpoint and calls thread_store.update_display_name # title from the checkpoint and calls thread_store.update_display_name
# after the run completes. # after the run completes.
total = t_pre_task - t_entry
if total >= _START_RUN_SLOW_THRESHOLD_SECONDS:
# Slow path: emit a phase breakdown so on-call can identify the
# culprit without re-running with extra instrumentation. The most
# common cause is a stale ``_maybe_sync_legacy_agent_records`` TTL
# window firing on a machine with hundreds of agent directories.
logger.warning(
"[start-run-timing] thread=%s total=%.3fs enforce_access=%.3fs "
"run_create=%.3fs thread_upsert=%.3fs build_payload=%.3fs",
sanitize_log_param(thread_id),
total,
t_post_enforce - t_pre_enforce,
t_post_create - t_post_enforce,
t_post_thread_upsert - t_post_create,
t_pre_task - t_post_thread_upsert,
)
else:
logger.debug(
"[start-run-timing] thread=%s total=%.3fs enforce=%.3fs create=%.3fs upsert=%.3fs build=%.3fs",
sanitize_log_param(thread_id),
total,
t_post_enforce - t_pre_enforce,
t_post_create - t_post_enforce,
t_post_thread_upsert - t_post_create,
t_pre_task - t_post_thread_upsert,
)
return record return record