From de597b602a1c2aa065c6d8ae14e44882b77ad19b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Thu, 28 May 2026 23:23:40 +0800 Subject: [PATCH] =?UTF-8?q?=E5=9C=86=E6=A1=8C=20Step2=EF=BC=9A=E5=9B=BA?= =?UTF-8?q?=E5=AE=9A=E5=8D=8F=E8=B0=83=E5=91=98=E5=8D=95=E4=BE=8B=E3=80=81?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E8=B0=83=E5=BA=A6=E6=80=A7=E8=83=BD=E3=80=81?= =?UTF-8?q?=E4=BF=AE=E6=AD=A3=E7=A0=94=E8=AE=A8=E4=B8=AD=E7=8A=B6=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将 roundtable-coordinator 改为内置单例避免重复创建；legacy sync 加 TTL 减少 runs/stream 延迟；前端仅当前发言者显示研讨中。 Co-authored-by: Cursor --- frontend-web/docs/multi-agent-backend-dev.md | 101 +++--- frontend-web/docs/multi-agent-frontend-dev.md | 8 +- .../src/roundtable-planning/api/recommend.ts | 17 +- .../hooks/useStep2Orchestration.ts | 30 +- .../lib/roundtable-constants.ts | 21 +- .../backend/app/gateway/app.py | 6 + .../app/gateway/routers/_roundtable_seed.py | 125 +++++++- .../app/gateway/routers/multi_agent.py | 192 ++++-------- .../backend/app/gateway/services.py | 287 ++++++++++++------ 9 files changed, 507 insertions(+), 280 deletions(-) diff --git a/frontend-web/docs/multi-agent-backend-dev.md b/frontend-web/docs/multi-agent-backend-dev.md index e759d19d..4492f3e5 100644 --- a/frontend-web/docs/multi-agent-backend-dev.md +++ b/frontend-web/docs/multi-agent-backend-dev.md @@ -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)` 工厂统一构造。 - 历史包袱：早期实现走 `httpx.AsyncClient(trust_env=False)` + `127.0.0.1:`，遇到开发机 `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）。 -- **`.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。 +- **席位 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 目录在部署机上**不会存在**。但功能型 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_done.thread_ids` 包含协调智能体自身**（`agent_threads[main_agent_name] = coord_thread`），前端 Step 3 列文件时需自行过滤。 - **澄清协议统一**：Step 1 与 Step 2 leader 都借 `ask_clarification` + 终态 `clarification` 帧驱动同一套选项卡片组件；`allow_multiple` 默认 false（单选、点击即发送）。 @@ -282,33 +282,50 @@ offline-backend-20260512/backend/ 请求体: { "agent_names": ["roundtable-intelligence", ...], // 用户选的席位 id - "main_agent_name": "roundtable-coordinator-xxx", // 当次总控 id,前端生成 - "model": "deepseek-chat" // 可选,写入 coordinator agent + "main_agent_name": "<可选,已废弃>", // 历史字段;2026-05 改造后后端忽略,统一用 COORDINATOR_AGENT_ID + "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）： - `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()` 后比较 / 存储 +- `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) 每完成一个 → 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-) +协调智能体(单例): + main_agent_name = COORDINATOR_AGENT_ID # 固定常量,不再每次新建 agent + coord_thread = _create_thread_direct(...) # 仍然每次新 thread,会话独立 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":"init_done","main_agent_name","thread_ids":{...}} @@ -316,28 +333,26 @@ asyncio.as_completed(各席位 _prepare_sub_agent) 失败 → 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_orchestration 技能调用其他智能体干活,使用技能时传入 -agent_name 和需要做的活。 -⚠️ agent_name 必须严格使用下方列出的英文 id(例如 roundtable-intelligence), - 不要把括号里的中文显示名拼进 agent_name 里。 - -#可调度智能体: -- agent_name: roundtable-intelligence(情报收集): -- agent_name: roundtable-environment(环境评估): +【圆桌系统初始化】本次研讨可调度的席位清单如下,请严格按 agent_name 通过 agent_orchestration 技能派活,不要派给不在清单内的 agent: +- agent_name: roundtable-intelligence（情报收集）： +- agent_name: roundtable-environment（环境评估）： ... ``` -中文显示名通过 `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`） @@ -609,13 +624,20 @@ httpx.AsyncClient( **已废止的 `trust_env=False` workaround**：旧版本走 `127.0.0.1:` 真实 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 澄清协议统一 @@ -705,15 +727,22 @@ offline-backend-20260512/backend/.deer-flow/* | Agent | 角色 | 是否必须打包发布 | |-------|------|------------------| -| `roundtable-intent` | **功能型**：Step 1 任务理解，所有部署都用它 | ✅ **必须**（任何环境） | -| `roundtable-recommender` | **功能型**：Step 1→2 之间从候选池里挑推荐，所有部署都用它 | ✅ **必须**（任何环境） | +| `roundtable-intent` | **功能型**：Step 1 任务理解，所有部署都用它 | ✅ seeder 自动落地（无需手工打包） | +| `roundtable-recommender` | **功能型**：Step 1→2 之间从候选池里挑推荐，所有部署都用它 | ✅ seeder 自动落地（无需手工打包） | +| `roundtable-coordinator` | **功能型单例**：Step 2 总控协调，2026-05 起为单例（id 固定） | ✅ seeder 自动落地（无需手工打包） | | `roundtable-intelligence` | 示例研讨席位（情报收集） | ⚠️ **互联网 demo 用**，内网不需要 | | `roundtable-environment` | 示例研讨席位（环境评估） | ⚠️ 同上 | | `roundtable-solution-design` | 示例研讨席位（方案设计） | ⚠️ 同上 | | `roundtable-risk-review` | 示例研讨席位（风险审查） | ⚠️ 同上 | | `roundtable-execution-plan` | 示例研讨席位（执行规划） | ⚠️ 同上 | | `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 + 推荐流程跑不起来，跟客户业务无关 diff --git a/frontend-web/docs/multi-agent-frontend-dev.md b/frontend-web/docs/multi-agent-frontend-dev.md index 88da48a8..89ee2963 100644 --- a/frontend-web/docs/multi-agent-frontend-dev.md +++ b/frontend-web/docs/multi-agent-frontend-dev.md @@ -16,6 +16,7 @@ - **不使用 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` 配自家模型即可，前后端均无需改动**。 - **席位区间硬约束**：推荐弹窗要求 **2 ≤ 选中数 ≤ 8**（`MIN_SELECTED` / `MAX_SELECTED`）。 +- **协调智能体单例（2026-05）**：`COORDINATOR_AGENT_ID = "roundtable-coordinator"` 是固定常量，不再每会话生成新 id。后端 seeder 自动落地，「本次席位列表」改由后端通过 thread 历史 human 消息注入，前端无感。`newCoordinatorName()` 仅作为兼容性壳保留。 - **总控编排循环上限**：`MAX_CYCLES = 8`，超出后强制停止并 toast。 - **席位派活策略**：**顺序**（不是并行）。旧实现并行触发多路 SSE 会让浏览器卡顿，现已改为顺序。 - **调试开关**：`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` Persona 覆盖；下一轮该席位的派活会用此 model | | `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-` 形式。只要那个旧 agent 目录还在磁盘上（即运维没手动删过），加载后继续在 Step 2 干活就能正常工作；如果旧 agent 被清理掉了，需要让用户「重新起草」从头跑一遍 Step 1 → Step 2。 ### 5.4 编排循环 `runOrchestration(seedMessage, ids, coord)` diff --git a/frontend-web/src/roundtable-planning/api/recommend.ts b/frontend-web/src/roundtable-planning/api/recommend.ts index db73b4ed..01fb53bb 100644 --- a/frontend-web/src/roundtable-planning/api/recommend.ts +++ b/frontend-web/src/roundtable-planning/api/recommend.ts @@ -68,13 +68,20 @@ export interface RecommendCandidate { * Agent ids that must never enter the candidate pool. Centralized here so * every caller (recommend dialog, future tools) applies the same rule. * - * - `roundtable-intent` — Step 1 clarification specialist, no role in research. - * - `roundtable-recommender`— the recommender itself, can't recommend itself. - * - `roundtable-coordinator-*` — disposable Step 2 coordinator shells (DB - * accumulates one per session); meaningless as roundtable participants. + * - `roundtable-intent` — Step 1 clarification specialist, no role in research. + * - `roundtable-recommender` — the recommender itself, can't recommend itself. + * - `roundtable-coordinator` — Step 2 coordinator singleton (2026-05 改造后固定 id, + * ensure 自动落地;它是"协调者",不能反过来作为席位被协调)。 + * - `roundtable-coordinator-*` 前缀 — 历史残留:2026-05 之前每次 init 会创建 + * `roundtable-coordinator-` 形式的临时 agent,DB 仍可能保留这些遗留记录。 + * 保留前缀过滤可以在管理员清理之前先把它们隐藏起来,避免出现在推荐弹窗候选池里。 */ const EXCLUDED_AGENT_ID_PREFIX = "roundtable-coordinator-"; -const EXCLUDED_AGENT_IDS = new Set(["roundtable-intent", "roundtable-recommender"]); +const EXCLUDED_AGENT_IDS = new Set([ + "roundtable-intent", + "roundtable-recommender", + "roundtable-coordinator", +]); /** * Drop excluded ids from a raw `GET /api/agents` list and project each entry diff --git a/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts b/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts index 0a20610d..6c9ac6fa 100644 --- a/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts +++ b/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts @@ -833,16 +833,19 @@ export function useStep2Orchestration( const markSpeakerStreaming = useCallback( (speakerId: number) => { - setActiveSpeakerSet((prev) => { - if (prev.has(speakerId)) return prev; - const next = new Set(prev); - next.add(speakerId); - return next; - }); + // 编排循环已经改成**顺序**派 special(详见 §5.4 / 文件顶部注释), + // 任意时刻只有一个子智能体在跑。因此这里既要把当前发言席位置为"研讨中", + // 又要把上一轮 / 历史残留下来的"研讨中"统一清成"在线",保证右侧"参与角色" + // 面板任何时刻**只有一个** sub agent 显示研讨中,与对话区实际在 stream 的 + // 那一个一致。activeSpeakerSet 沿用 Set 形态(虽然实际只放 1 个)以兼容外部 + // 消费方,真正的"高亮排他"靠 roles 的 status reset。 + setActiveSpeakerSet(() => new Set([speakerId])); setRoles((prev) => - prev.map((item) => - item.id === speakerId ? { ...item, status: "研讨中" } : item, - ), + prev.map((item) => { + if (item.id === speakerId) return { ...item, status: "研讨中" }; + if (item.status === "研讨中") return { ...item, status: "在线" }; + return item; + }), ); }, [setRoles], @@ -912,7 +915,14 @@ export function useStep2Orchestration( safetyCounter < MAX_CYCLES ) { 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 流,导致总控写文件时沙箱不自动打开。 // 这里把 leader thread id 绑到气泡 + artifact handler 上,与 special 同样的行为。 diff --git a/frontend-web/src/roundtable-planning/lib/roundtable-constants.ts b/frontend-web/src/roundtable-planning/lib/roundtable-constants.ts index 1b5d76f4..4bde0b96 100644 --- a/frontend-web/src/roundtable-planning/lib/roundtable-constants.ts +++ b/frontend-web/src/roundtable-planning/lib/roundtable-constants.ts @@ -130,8 +130,27 @@ export function nowTime(): string { return new Date().toLocaleTimeString("zh-CN", { hour12: false }); } +/** + * 圆桌总控协调智能体的**固定单例 id**。后端 `_roundtable_seed.COORDINATOR_AGENT_ID` + * 与本常量必须保持一致(2026-05 改造前是 `roundtable-coordinator-` 形式, + * 每次会话生成新 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 { - return `roundtable-coordinator-${Date.now().toString(36)}`; + return COORDINATOR_AGENT_ID; } /** diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index a993f732..50319e12 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/offline-backend-20260512/backend/app/gateway/app.py @@ -512,11 +512,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: _mark_legacy_sync_fresh, _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) if agent_store is not None: await _sync_legacy_agents(agent_store) _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") except Exception: logger.exception("Legacy agent sync failed at startup (non-fatal)") diff --git a/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed.py b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed.py index f4bd40df..1b365426 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed.py +++ b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed.py @@ -2,21 +2,35 @@ 适用场景 -------- -``roundtable-intent`` / ``roundtable-recommender`` 是圆桌 Step 1 + 推荐链路的 -**功能型** 内置 agent,任何部署都必须存在。但仓库根 ``.gitignore`` 把整个 -``.deer-flow/`` 目录排除,新机器 ``git clone`` 之后这两个目录**不会存在**, -直接调 ``/api/intent/stream`` 会因为 agent 解析失败抛 500。 +``roundtable-intent`` / ``roundtable-recommender`` / ``roundtable-coordinator`` +是圆桌 Step 1 + 推荐 + Step 2 总控链路的**功能型** 内置 agent,任何部署都必须 +存在。但仓库根 ``.gitignore`` 把整个 ``.deer-flow/`` 目录排除,新机器 +``git clone`` 之后这些目录**不会存在**,直接调对应 endpoint 会因为 agent +解析失败抛 500。 -本模块在三个路由入口(``intent.init``、``intent.stream``、``recommend.stream``) -被调用,检查磁盘文件是否齐全;缺失则用下方嵌入的标准模板写回去。 -``_sync_legacy_agents`` 会在下一次 ``GET /api/agents`` 时把目录 upsert 进 DB, -所以这里只关心磁盘文件,不直接动 DB。 +本模块在四个路由入口(``intent.init``、``intent.stream``、``recommend.stream``、 +``multi_agent.init``)被调用,检查磁盘文件是否齐全;缺失则用下方嵌入的标准模板 +写回去。``_sync_legacy_agents`` 会在下一次 ``GET /api/agents`` 时把目录 upsert +进 DB,所以这里只关心磁盘文件,不直接动 DB。 为什么嵌入字符串而不是用 importlib.resources 资源文件 ------------------------------------------------------ -SOUL 体量不大(两个加起来 ~200 行),嵌入到 Python 文件里是**单一可信源**——改 +SOUL 体量不大(三个加起来 ~400 行),嵌入到 Python 文件里是**单一可信源**——改 SOUL 走代码 review、commit 走 git 历史,且不会因为 wheel 打包漏配 ``package-data`` 导致部署机拿不到。 + +为什么 ``roundtable-coordinator`` 是单例而不是每次会话新建 +------------------------------------------------------------ +历史实现每次 Step 2 init 都用 ``roundtable-coordinator-`` 这种 +带时间戳的临时 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 @@ -476,17 +490,91 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员 """ -_BUILTIN_FUNCTIONAL_AGENTS: tuple[tuple[str, str, str, str], ...] = ( - (INTENT_AGENT_ID, INTENT_AGENT_NAME, INTENT_AGENT_DESCRIPTION, INTENT_AGENT_SOUL), - (RECOMMENDER_AGENT_ID, RECOMMENDER_AGENT_NAME, RECOMMENDER_AGENT_DESCRIPTION, RECOMMENDER_AGENT_SOUL), +COORDINATOR_AGENT_ID = "roundtable-coordinator" +COORDINATOR_AGENT_NAME = "总控协调" +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. Returns True if files were just created, False if the directory already 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) 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 部署时与 # 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( - {"id": agent_id, "name": name, "description": description}, + config_payload, default_flow_style=False, allow_unicode=True, ) @@ -520,9 +613,9 @@ def ensure_roundtable_functional_agents() -> list[str]: actionable error if the agent really cannot be resolved. """ 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: - if _ensure_one(agent_id, name, description, soul): + if _ensure_one(agent_id, name, description, soul, skills): created.append(agent_id) except Exception: logger.exception("Failed to seed roundtable functional agent '%s'", agent_id) diff --git a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py index 57114fb8..1e31a70c 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py +++ b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py @@ -57,6 +57,10 @@ import uuid from typing import Any, AsyncIterator 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.tools.builtins.clarification_utils import resolve_allow_multiple from deerflow.runtime.user_context import get_effective_user_id @@ -443,82 +447,11 @@ async def _get_agents_batch_direct( return out -async def _create_coordinator_direct( - request: Request, - *, - agent_id: str, - name: str, - 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) +# NOTE: 历史上这里曾有 _create_coordinator_direct(),每次 init 都创建一个带时间戳的 +# roundtable-coordinator-。2026-05 改造为单例(见 _roundtable_seed.py 与本 +# 模块 init_multi_agent 的 docstring),不再需要"每次创建"的直调辅助;改造后由 +# ensure_roundtable_functional_agents() 在路由入口落地静态 SOUL,席位列表通过 +# _append_thread_message 注入到 coordinator thread 的对话历史中。 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 - "main_agent_name": "roundtable-coordinator-xxx", // 当次总控 id，前端生成 - "model": "<可选>" // 可选,写入 coordinator agent;不传由 lead_agent 回退到 config.yaml models[0] + "main_agent_name": "<可选,已废弃>", // 历史兼容字段,新代码忽略,统一用 COORDINATOR_AGENT_ID + "model": "<可选>" // 可选;不传由 lead_agent 回退到 config.yaml models[0] } SSE 事件（每条 ``data: {...}\\n\\n``）:: seat_ready — 某一席位 thread 已创建（``asyncio.as_completed`` 顺序，先完先发） - coordinator_ready — 总控 agent + thread 已就绪 + coordinator_ready — 总控 thread 就绪 + 已注入本次席位列表 init_done — ``thread_ids`` 全量映射，前端存 ``threadIds`` 并开始编排 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-`` 残留)。 + + 本次圆桌的「可调度席位列表」由 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: 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") + # main_agent_name 是历史字段,2026-05 改造后忽略,统一用单例 id。保留接收但不再用前端值, + # 避免下游再次创建带时间戳的临时 coordinator。 + main_agent_name = COORDINATOR_AGENT_ID if not agent_names: raise HTTPException(status_code=400, detail="agent_names is required") for name in agent_names: _validate_agent_id(name, "agent_names entry") - _validate_agent_id(main_agent_name, "main_agent_name") def _frame(payload: dict[str, Any]) -> str: 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() 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}" for sid, _, display_name, description in prepared ] - # 动态 SOUL：把本次 init 的席位列表写进总控 prompt，供 agent_orchestration 派活。 - soul = ( - "#核心定位：\n" - "用途：根据用户的任务目标,调用其他智能体干活,只负责协调汇总," - "当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n" - "#调度方法：\n" - "通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活。" - "注意：agent_name 必须严格使用下方列出的英文 id（例如 roundtable-intelligence," - "或形如 02714cbe6e64406a8a4659f0e3d9c0b8 这样的 UUID）," - "不要把括号里的中文显示名拼进 agent_name 里。\n" - "#关键约束(必读)：\n" - "1. **不要对同一子智能体重复派同样的活**：每次派活前回顾对话历史," - "如果某个子智能体已经针对你正要派的任务完成过交付(" - "看到形如\"子智能体[名称]完成XX工作,交付内容为:...\"的 human 消息)," - "就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题。\n" - "2. **看到交付要会用**：子智能体的交付内容会作为 human 消息追加到你的对话历史里," - "你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具)," - "让讨论收敛。\n" - "3. **避免无意义循环**：如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论。\n" - "#可调度智能体：\n" - + "\n".join(descriptions) + # 协调智能体改造为**单例**(id = roundtable-coordinator): + # - SOUL 已经由 ensure_roundtable_functional_agents() 在路由入口落地, + # 通用协调规则常驻磁盘,无需每次重写。 + # - 本次"可调度席位清单"以 human 消息形式 append 到 coordinator + # thread 历史里,leader run 每轮都能在对话历史最前面看到。 + # 这样席位列表随会话动态变化,但磁盘上不再累积 roundtable-coordinator-。 + # + # 注意:model 字段在 lead_agent run 时通过 context.model_name 注入, + # 不需要写回到 agent config.yaml(否则会污染单例的配置)。 + seat_introduction = ( + "【圆桌系统初始化】本次研讨可调度的席位清单如下,请严格按 agent_name " + "通过 agent_orchestration 技能派活,不要派给不在清单内的 agent:\n" + + "\n".join(seat_lines) ) - # 3) coordinator thread + agent 并行创建。thread 与 agent 之间无依赖, - # 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 + coord_thread = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA) 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: - yield _frame({"event": "error", "detail": f"create coordinator agent failed: {result}"}) - return + + # 把席位列表注入 coordinator thread 的对话历史。HTTP loopback 调用 + # 走 /api/threads/{tid}/state(读+写),开销 ~10-30ms,在 init 整体 + # 几百 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() logger.info( "[init-direct] meta=%.3fs seats=%.3fs coord=%.3fs total=%.3fs seats_count=%d", diff --git a/offline-backend-20260512/backend/app/gateway/services.py b/offline-backend-20260512/backend/app/gateway/services.py index 46965a2c..1a08ea3e 100644 --- a/offline-backend-20260512/backend/app/gateway/services.py +++ b/offline-backend-20260512/backend/app/gateway/services.py @@ -11,6 +11,7 @@ import asyncio import json import logging import re +import time from collections.abc import Mapping from typing import Any @@ -100,14 +101,14 @@ _DEFAULT_ASSISTANT_ID = "lead_agent" # Built-in LangGraph graph IDs registered in langgraph.json. # These are NOT stored in the agents DB and must bypass the custom-agent access check. -_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent", "ai_writing"}) - -_RAG_CITATION_PROMPT = """当前对话已开启参考文献模式。 -如果本轮调用了搜索、知识库、检索类技能或工具，请基于本轮所有检索结果整体回答。 -你必须把同一轮内多次检索得到的资料视为同一批参考文献，按出现顺序从 [1] 开始连续编号，不要让每个检索词各自从 [1] 重新编号。 -回答正文中的关键结论必须标注对应编号，例如 [1]、[2]，多个来源可写为 [1][3][5]。 -不要引用不存在的编号，不要使用技能内部自带的编号，不要把参考文献完整列表堆在正文末尾。 -如果没有任何可用检索结果，请直接说明未检索到可靠参考资料。""" +_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent", "ai_writing"}) + +_RAG_CITATION_PROMPT = """当前对话已开启参考文献模式。 +如果本轮调用了搜索、知识库、检索类技能或工具，请基于本轮所有检索结果整体回答。 +你必须把同一轮内多次检索得到的资料视为同一批参考文献，按出现顺序从 [1] 开始连续编号，不要让每个检索词各自从 [1] 重新编号。 +回答正文中的关键结论必须标注对应编号，例如 [1]、[2]，多个来源可写为 [1][3][5]。 +不要引用不存在的编号，不要使用技能内部自带的编号，不要把参考文献完整列表堆在正文末尾。 +如果没有任何可用检索结果，请直接说明未检索到可靠参考资料。""" def _prepend_prompt_prefix(graph_input: dict[str, Any], prefix: str) -> dict[str, Any]: @@ -136,7 +137,7 @@ def _prepend_prompt_prefix(graph_input: dict[str, Any], prefix: str) -> dict[str return {**graph_input, "messages": new_messages} -def _prompt_prefix_opted_in(body: Any) -> bool: +def _prompt_prefix_opted_in(body: Any) -> bool: """Detect whether the caller asked for the prompt prefix to be applied.""" context = getattr(body, "context", None) if isinstance(context, Mapping) and context.get("prompt_prefix_enabled"): @@ -144,44 +145,44 @@ def _prompt_prefix_opted_in(body: Any) -> bool: metadata = getattr(body, "metadata", None) if isinstance(metadata, Mapping) and metadata.get("prompt_prefix_enabled"): return True - return False - - -def _rag_mode_requested(body: Any) -> bool: - context = getattr(body, "context", None) - if isinstance(context, Mapping) and context.get("rag_mode_enabled"): - return True - metadata = getattr(body, "metadata", None) - if isinstance(metadata, Mapping) and ( - metadata.get("rag_mode_enabled") or metadata.get("rag_mode") - ): - return True - return False - - -def _thread_rag_mode_enabled(record: Any) -> bool: - if not isinstance(record, Mapping): - return False - metadata = record.get("metadata") - return isinstance(metadata, Mapping) and bool( - metadata.get("rag_mode_enabled") or metadata.get("rag_mode") - ) - - -def _append_invisible_prompt(graph_input: dict[str, Any], prompt: str) -> dict[str, Any]: - messages = graph_input.get("messages") - if not isinstance(messages, list) or not messages: - return graph_input - new_messages = list(messages) - for idx, msg in enumerate(new_messages): - if not isinstance(msg, HumanMessage): - continue - extra = dict(getattr(msg, "additional_kwargs", None) or {}) - existing = str(extra.get("prompt_prefix") or "").strip() - extra["prompt_prefix"] = f"{existing}\n\n{prompt}".strip() if existing else prompt - new_messages[idx] = msg.model_copy(update={"additional_kwargs": extra}) - break - return {**graph_input, "messages": new_messages} + return False + + +def _rag_mode_requested(body: Any) -> bool: + context = getattr(body, "context", None) + if isinstance(context, Mapping) and context.get("rag_mode_enabled"): + return True + metadata = getattr(body, "metadata", None) + if isinstance(metadata, Mapping) and ( + metadata.get("rag_mode_enabled") or metadata.get("rag_mode") + ): + return True + return False + + +def _thread_rag_mode_enabled(record: Any) -> bool: + if not isinstance(record, Mapping): + return False + metadata = record.get("metadata") + return isinstance(metadata, Mapping) and bool( + metadata.get("rag_mode_enabled") or metadata.get("rag_mode") + ) + + +def _append_invisible_prompt(graph_input: dict[str, Any], prompt: str) -> dict[str, Any]: + messages = graph_input.get("messages") + if not isinstance(messages, list) or not messages: + return graph_input + new_messages = list(messages) + for idx, msg in enumerate(new_messages): + if not isinstance(msg, HumanMessage): + continue + extra = dict(getattr(msg, "additional_kwargs", None) or {}) + existing = str(extra.get("prompt_prefix") or "").strip() + extra["prompt_prefix"] = f"{existing}\n\n{prompt}".strip() if existing else prompt + new_messages[idx] = msg.model_copy(update={"additional_kwargs": extra}) + break + return {**graph_input, "messages": new_messages} # Whitelist of run-context keys that the langgraph-compat layer forwards from @@ -196,9 +197,9 @@ _CONTEXT_CONFIGURABLE_KEYS: frozenset[str] = frozenset( "model_name", "mode", "thinking_enabled", - "reasoning_effort", - "rag_mode_enabled", - "writing_mode", + "reasoning_effort", + "rag_mode_enabled", + "writing_mode", "writing_artifact_path", "canvas_artifact_title", "canvas_artifact_type", @@ -356,7 +357,17 @@ def _collect_requested_agent_ids(body: Any) -> set[str]: 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 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: agent_ids = _collect_requested_agent_ids(body) 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 store = get_agent_store(request) - await _sync_legacy_agent_records(store) + await _maybe_sync_legacy_agent_records(store) user_id = get_effective_user_id() for agent_id in agent_ids: try: @@ -417,14 +491,27 @@ async def start_run( Target thread. request : Request 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) run_mgr = get_run_manager(request) run_ctx = get_run_context(request) disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_ + t_pre_enforce = time.perf_counter() await _enforce_agent_access(body, request) + t_post_enforce = time.perf_counter() try: record = await run_mgr.create_or_reject( @@ -439,33 +526,34 @@ async def start_run( raise HTTPException(status_code=409, detail=str(exc)) from exc except UnsupportedStrategyError as 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, # even for threads that were never explicitly created via POST /threads # (e.g. stateless runs). - requested_rag_mode = _rag_mode_requested(body) - effective_rag_mode = requested_rag_mode - try: - existing = await run_ctx.thread_store.get(thread_id) - effective_rag_mode = requested_rag_mode or _thread_rag_mode_enabled(existing) - if existing is None: - metadata = dict(body.metadata or {}) - if effective_rag_mode: - metadata["rag_mode_enabled"] = True - metadata["rag_mode"] = True - await run_ctx.thread_store.create( - thread_id, - assistant_id=body.assistant_id, - metadata=metadata, - ) - else: - await run_ctx.thread_store.update_status(thread_id, "running") - if requested_rag_mode and not _thread_rag_mode_enabled(existing): - await run_ctx.thread_store.update_metadata( - thread_id, - {"rag_mode_enabled": True, "rag_mode": True}, - ) - effective_rag_mode = True + requested_rag_mode = _rag_mode_requested(body) + effective_rag_mode = requested_rag_mode + try: + existing = await run_ctx.thread_store.get(thread_id) + effective_rag_mode = requested_rag_mode or _thread_rag_mode_enabled(existing) + if existing is None: + metadata = dict(body.metadata or {}) + if effective_rag_mode: + metadata["rag_mode_enabled"] = True + metadata["rag_mode"] = True + await run_ctx.thread_store.create( + thread_id, + assistant_id=body.assistant_id, + metadata=metadata, + ) + else: + await run_ctx.thread_store.update_status(thread_id, "running") + if requested_rag_mode and not _thread_rag_mode_enabled(existing): + await run_ctx.thread_store.update_metadata( + thread_id, + {"rag_mode_enabled": True, "rag_mode": True}, + ) + effective_rag_mode = True # SDK 的 ``client.threads.create()`` 不传 assistant_id —— 先建出来空的 thread， # 真正提交 run 时才知道用哪个 assistant。这里回填一次，避免历史列表 / 按 # assistant 筛选时拿不到关联。 @@ -473,6 +561,7 @@ async def start_run( await run_ctx.thread_store.update_assistant_id(thread_id, body.assistant_id) except Exception: 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) graph_input = normalize_input(body.input) @@ -485,21 +574,21 @@ async def start_run( # applies once. We intentionally do NOT gate on thread-row existence: the # thread_meta row is frequently pre-created, which previously skipped the # prefix for genuine first turns. - if _prompt_prefix_opted_in(body): - prefix_settings = load_system_settings().prompt_prefix - if prefix_settings.enabled and prefix_settings.prompt_prefix.strip(): - graph_input = _prepend_prompt_prefix(graph_input, prefix_settings.prompt_prefix.strip()) - if effective_rag_mode: - graph_input = _append_invisible_prompt(graph_input, _RAG_CITATION_PROMPT) - config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id) + if _prompt_prefix_opted_in(body): + prefix_settings = load_system_settings().prompt_prefix + if prefix_settings.enabled and prefix_settings.prompt_prefix.strip(): + graph_input = _prepend_prompt_prefix(graph_input, prefix_settings.prompt_prefix.strip()) + if effective_rag_mode: + graph_input = _append_invisible_prompt(graph_input, _RAG_CITATION_PROMPT) + config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id) # Merge DeerFlow-specific context overrides into both ``configurable`` and ``context``. # The ``context`` field is a custom extension for the langgraph-compat layer # that carries agent configuration (model_name, thinking_enabled, etc.). # Only agent-relevant keys are forwarded; unknown keys (e.g. thread_id) are ignored. - body_context = dict(getattr(body, "context", None) or {}) - if effective_rag_mode: - body_context["rag_mode_enabled"] = True + body_context = dict(getattr(body, "context", None) or {}) + if effective_rag_mode: + body_context["rag_mode_enabled"] = True # Promote whitelisted top-level body fields into the context map so # callers can pass them either nested under ``context`` or as flat # top-level fields (``excluded_tools``, ``skill_stop_names``, …). @@ -511,6 +600,7 @@ async def start_run( stream_modes = normalize_stream_modes(body.stream_mode) + t_pre_task = time.perf_counter() task = asyncio.create_task( run_agent( bridge, @@ -535,6 +625,33 @@ async def start_run( # title from the checkpoint and calls thread_store.update_display_name # 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