From 9bbd04d6cf2ba1ec9bc7aa6be78e598d2327218b Mon Sep 17 00:00:00 2001 From: fengxiangdebaozi <10845145+fengxiangdebaozi@user.noreply.gitee.com> Date: Fri, 5 Jun 2026 10:59:56 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=8E=E5=8F=B0=E6=8C=82=E8=B5=B7=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...agent-background-run-遗留问题与下次继续.md | 124 ++++++++++ frontend-web/package-lock.json | 88 ++++++- frontend-web/package.json | 1 + frontend-web/src/main.tsx | 10 +- .../components/DispatchChainModal.tsx | 22 +- .../pages/RoundtablePlanningPage.tsx | 49 ++-- .../gateway/roundtable_inprocess_gateway.py | 110 ++++++--- .../app/gateway/roundtable_job_executor.py | 219 ++++++++++++++++-- .../app/gateway/roundtable_run_policy.py | 94 ++++++++ .../app/gateway/routers/multi_agent.py | 46 ++-- .../app/gateway/routers/roundtable_jobs.py | 2 + .../agents/roundtable_orchestrator/engine.py | 127 ++++++++-- .../roundtable_orchestrator/progress.py | 84 ++++++- .../agents/roundtable_orchestrator/types.py | 6 + 14 files changed, 846 insertions(+), 136 deletions(-) create mode 100644 frontend-web/docs/multi-agent-background-run-遗留问题与下次继续.md create mode 100644 offline-backend-20260512/backend/app/gateway/roundtable_run_policy.py diff --git a/frontend-web/docs/multi-agent-background-run-遗留问题与下次继续.md b/frontend-web/docs/multi-agent-background-run-遗留问题与下次继续.md new file mode 100644 index 00000000..ba2eefde --- /dev/null +++ b/frontend-web/docs/multi-agent-background-run-遗留问题与下次继续.md @@ -0,0 +1,124 @@ +# 多智能体「后台挂起」—— 当前状态、遗留问题与下次继续指南 + +> 配套设计文档：`multi-agent-background-run-dev.md`（分 Phase 0-9 的总方案）。 +> 本文记录**截至本次会话的实际进展 + 还没解决的遗留问题**，方便下次直接续接。 + +--- + +## 一、整体状态：跑通到哪了 + +「后台挂起」= Step2 把研讨交给后端**进程内驱动真模型**（leader→各席位→共识→报告）， +survive 刷新/关页，对话与报告作为一个 **run 写进草稿**（和手动跑同一存取），可在历史回看。 + +**真网关已在本地验证可跑通**（2026-06-04 日志确认：roundtable-coordinator + 各席位 + +roundtable-report 真调 deepseek、真搜索、跑到 report run success）。但还有几个**内容/产物层面的遗留问题**（见第三节）。 + +### 已完成（可用） +- ✅ 后端持久化：`roundtable_jobs` 表（迁移 `20260604_01` / `20260605_01` dialogues / `20260605_02` step3）。 +- ✅ 编排引擎 `packages/harness/deerflow/agents/roundtable_orchestrator/`（recommend + chain + 报告，纯逻辑可单测）。 +- ✅ 控制面 `app/gateway/routers/roundtable_jobs.py`：start/get/by-draft/stream(SSE)/resume/cancel/admin-cleanup。 +- ✅ 执行器 `app/gateway/roundtable_job_executor.py`：后台任务 + `set_current_user` + 生命周期 + **每轮把对话作为 `job-` run 写进 `draft.step2.runs[]`（同手动存取）** + done 写 step3。 +- ✅ 真网关 `app/gateway/roundtable_inprocess_gateway.py`：`SimpleNamespace(app)` + `start_run` 进程内驱动 agent；复用 `multi_agent._extract_dispatch` 解析派活；checkpointer 读最终 messages。 +- ✅ 模拟网关 `roundtable_orchestrator/simulated_gateway.py`（不调真模型，前端联调用）。 +- ✅ 前端：`api/roundtable-jobs.ts`、`lib/job-types.ts`、Step2「后台挂起」按钮、历史下拉 loading、`DispatchChainModal`/`DispatchChainJobModal`（流程图 + 对话面板 markdown + 加宽 + 澄清作答 + 取消 + 查看）。 +- ✅ 上下文「广播」（inline）：席位/总控 prompt 带上「任务背景 + 前序各席位交付 + 别提问」约束（修「席位拿不到关键信息、反问用户、总控瞎编」）。 +- ✅ 后端测试：`tests/test_roundtable_jobs.py` / `test_roundtable_orchestrator.py` / `test_roundtable_job_executor.py` / `test_roundtable_jobs_api.py` 全过。 + +--- + +## 二、怎么启动 / 验证（每次开工先做） + +1. **开真网关**（默认是模拟网关）：已写进 `offline-backend-20260512/.env`： + ``` + ROUNDTABLE_JOB_GATEWAY=inprocess + ``` + 启动日志应出现 `[roundtable-jobs] gateway = inprocess (real models)`。 + +2. **必须用 `make gateway`（不带 --reload）跑后端！** + `make dev` 的 `--reload` 监视 `backend/` 目录，agent 运行时往 `.deer-flow/` 写文件 + （memory.json / checkpoint / 沙箱输出）会触发热重载，**把长时后台任务杀掉**。 + （改进项：给 uvicorn 加 `--reload-exclude '.deer-flow/*'`，或始终用 `make gateway`。） + +3. **直接查库验证**（最有效，PowerShell 控制台是 GBK，导出到文件看中文）： + ```bash + cd offline-backend-20260512/backend + # 查最新作业的 dialogues / step3 + PYTHONPATH=. uv run python -c "import sqlite3,json; con=sqlite3.connect('.deer-flow/data/deerflow.db'); ..." + ``` + - 作业表：`.deer-flow/data/deerflow.db` → 表 `roundtable_jobs`（`dialogues`/`step3` 列）。 + - 草稿表：`roundtable_drafts` → `step2` 列里应有 `runs[ {id:"job-", jobId, jobStatus, step2RoundtableDialogues:[...], step3:{...}} ]`。 + +4. **前端验证**：后台挂起 → 等跑完 → 从历史加载该草稿 → Step2 应显示后台研讨完整对话（走现有 `normalizeStep2Runs`/`hydrateStep2Run`）。 + +--- + +## 三、遗留问题（下次重点，已按优先级排） + +### P0 —— 报告 agent 不写 HTML 文件（结果绘制拿不到产物） +- **现象**：后台报告 run 跑完，日志 `report html: no .html found in ...threads//user-data/outputs`； + `run.step3.html` 回退成报告 agent 的**纯文字**（不含 ``），第3步画不出东西。 +- **已做**：`build_report_prompt`（`roundtable_orchestrator/engine.py`）已对齐前端 + `useStep3Report.buildReportPrompt`，明确「用 write_file 写入 /mnt/user-data/outputs/方案总览.html」； + `_read_report_html`（`roundtable_inprocess_gateway.py`）已改为扫 outputs 目录任意 `.html`。 +- **仍没解决**：报告 agent（deepseek-chat）还是没调 `write_file`。 +- **下次怎么查**： + 1. 对比**前端那条能写出 HTML 的报告 run** 的确切调用（`useStep3Report.ts` 里 `streamMultiAgent` + 调 `/api/multi-agent/run/stream` 的 body：`agent_type`/`excluded_tools`/`context` 等）， + 和后台 `_run_agent_turn`（`roundtable_inprocess_gateway.py`，agent_name=roundtable-report， + `excluded_tools=_REPORT_EXCLUDED`）逐项比对，找出差异（怀疑 reasoning_effort / thinking / + 某个 context 字段 / recursion 步数导致它没走到 write_file）。 + 2. 看报告 run 的最终 messages（checkpoint）里有没有 write_file 的 tool_call（被 dangling 截断？）。 + 3. 报告 agent 的 SOUL：`app/gateway/routers/_roundtable_seed/roundtable-report/SOUL.md`——确认它默认行为。 + +### P1 —— 内容质量（本次刚加 inline 广播，待真机验证） +- **已做**：席位/总控 prompt 注入「任务背景 + 前序交付 + 别提问」。 +- **待验证**：真机跑一次，看席位是否基于真实上下文产出（不再反问/凭记忆瞎编）。 +- **若 inline 不够**：考虑实现**真·thread 广播**（像 `multi_agent._append_thread_message` 那样把 + 「用户意图 / 派活 / 各席位交付」追加进各 thread 历史）。进程内要避开 HTTP loopback 鉴权， + 得用 checkpointer 直接 append human 消息（参考 `threads` 路由 POST `/state` 的内部逻辑）。这块较重，先看 inline 够不够。 + +### P2 —— 对话「没有步骤（steps）」 +- **现象**：手动跑的气泡带工具调用步骤卡（`onPhaseChange` → `appendStepIfNew` → `Step2Dialogue.steps`）； + 后台对话只存了最终正文，没有 steps。 +- **下次怎么做**：后台 `_run_agent_turn` 跑完后，从 checkpoint 的 messages 里抽出 tool_calls， + 映射成 `Step2Dialogue.steps`（`{id,key,label,ts,args}`），在执行器 `_to_step2_dialogue` 里带上。 + 纯展示增强，不影响内容/报告。 + +### P3 —— 双跑浪费 + --reload 坑（工程健壮性） +- 进 Step2 时前端会自动跑一遍本地研讨（真模型、真线程）；点后台挂起后前端那条 run 在后端还会跑完 + （`stopOrchestration` 只断客户端、停不了后端 run）→ 和后台作业短暂双跑、双倍 token。 + - 优化方向：进 Step2 改成「显式开始」而非自动跑；或后台挂起时真正 cancel 前端那条 run。 +- `make dev` 的 `--reload` 会杀后台任务（见第二节）；改进项：`--reload-exclude '.deer-flow/*'`。 + +--- + +## 四、关键文件清单（下次定位用） + +**后端** +| 文件 | 作用 | +|------|------| +| `app/gateway/roundtable_inprocess_gateway.py` | 真网关：进程内驱动 leader/seat/re 编排循环 + `build_report_prompt` + inline 广播（`_seat_message`/`_deliveries_block`） | +| `.../roundtable_orchestrator/progress.py` | 进度落库 + `on_dialogue` 每轮回调 | +| `.../persistence/roundtable_jobs/` | job 表 model/sql | +| `app/gateway/deps.py` | `ROUNDTABLE_JOB_GATEWAY` 选网关 + 注入 executor（draft_store） | + +**前端** +| 文件 | 作用 | +|------|------| +| `src/roundtable-planning/api/roundtable-jobs.ts` | 作业 API + SSE | +| `src/roundtable-planning/lib/job-types.ts` | 类型（RoundtableJob/JobDialogue/...） | +| `src/roundtable-planning/components/DispatchChainModal.tsx` / `DispatchChainJobModal.tsx` | 流程图弹窗 + 对话面板 | +| `src/roundtable-planning/pages/RoundtablePlanningPage.tsx` | 后台挂起按钮、历史下拉、**runs[] 多运行模型**（`hydrateStep2Run`/`getStep2SnapshotWithRuns`/`normalizeStep2Runs`） | +| `src/roundtable-planning/api/drafts.ts` | `DraftStep2RunSnapshot`（带 jobId/jobStatus/step3） | + +--- + +## 五、关键架构事实（避免重复踩坑） + +1. **真网关进程内驱动**：`start_run(body, thread_id, SimpleNamespace(app=app))` + `set_current_user`， + 不走 HTTP（loopback 调 `/api/multi-agent/*` 不可行：`_auth_headers` 只转发用户 token + owner check）。 +2. **后台作业首启自建全新线程**（`StartParams.is_resume`），绝不复用前端本地研讨的线程（会撞 409）。 +3. **存取统一**：后台作业 = `draft.step2.runs[]` 里一个带 jobId 的 run，和手动跑同一存取/hydrate 路径， + **不要再开 observe/回显分支**（之前试过，被否）。 +4. **聊天记录确实存上了**（job.dialogues + draft 的 run）；之前"没存上"是测了旧代码 / 看的是前端本地残缺数据。 +5. 多 worker / SSE：状态进 DB、SSE 走 DB 轮询；真网关进程内无 loopback。 diff --git a/frontend-web/package-lock.json b/frontend-web/package-lock.json index 0437d4af..8dc58311 100644 --- a/frontend-web/package-lock.json +++ b/frontend-web/package-lock.json @@ -15,6 +15,10 @@ "@codemirror/lang-markdown": "^6.5.0", "@codemirror/lang-python": "^6.2.1", "@codemirror/language-data": "^6.5.2", + "@dnd-kit/core": "^6.3.1", + "@dnd-kit/sortable": "^10.0.0", + "@dnd-kit/utilities": "^3.2.2", + "@fontsource/noto-sans-sc": "^5.2.9", "@langchain/core": "^1.1.15", "@langchain/langgraph-sdk": "^1.5.3", "@radix-ui/react-avatar": "^1.1.11", @@ -55,9 +59,12 @@ "cmdk": "^1.1.1", "codemirror": "^6.0.2", "date-fns": "^4.1.0", + "dayjs": "^1.11.21", + "dnd-kit": "^0.0.2", "docx": "^9.5.1", "dotenv": "^17.2.3", "embla-carousel-react": "^8.6.0", + "file-saver": "^2.0.5", "gsap": "^3.13.0", "hast": "^1.0.0", "katex": "^0.16.28", @@ -90,6 +97,7 @@ }, "devDependencies": { "@tailwindcss/postcss": "^4.0.15", + "@types/file-saver": "^2.0.7", "@types/gsap": "^3.0.0", "@types/node": "^20.14.10", "@types/react": "^19.0.0", @@ -858,6 +866,55 @@ "w3c-keyname": "^2.2.4" } }, + "node_modules/@dnd-kit/accessibility": { + "version": "3.1.1", + "resolved": "https://registry.npmjs.org/@dnd-kit/accessibility/-/accessibility-3.1.1.tgz", + "integrity": "sha512-2P+YgaXF+gRsIihwwY1gCsQSYnu9Zyj2py8kY5fFvUM1qm2WA2u639R6YNVfU4GWr+ZM5mqEsfHZZLoRONbemw==", + "dependencies": { + "tslib": "^2.0.0" + }, + "peerDependencies": { + "react": ">=16.8.0" + } + }, + "node_modules/@dnd-kit/core": { + "version": "6.3.1", + "resolved": "https://registry.npmjs.org/@dnd-kit/core/-/core-6.3.1.tgz", + "integrity": "sha512-xkGBRQQab4RLwgXxoqETICr6S5JlogafbhNsidmrkVv2YRs5MLwpjoF2qpiGjQt8S9AoxtIV603s0GIUpY5eYQ==", + "dependencies": { + "@dnd-kit/accessibility": "^3.1.1", + "@dnd-kit/utilities": "^3.2.2", + "tslib": "^2.0.0" + }, + "peerDependencies": { + "react": ">=16.8.0", + "react-dom": ">=16.8.0" + } + }, + "node_modules/@dnd-kit/sortable": { + "version": "10.0.0", + "resolved": "https://registry.npmjs.org/@dnd-kit/sortable/-/sortable-10.0.0.tgz", + "integrity": "sha512-+xqhmIIzvAYMGfBYYnbKuNicfSsk4RksY2XdmJhT+HAC01nix6fHCztU68jooFiMUB01Ky3F0FyOvhG/BZrWkg==", + "dependencies": { + "@dnd-kit/utilities": "^3.2.2", + "tslib": "^2.0.0" + }, + "peerDependencies": { + "@dnd-kit/core": "^6.3.0", + "react": ">=16.8.0" + } + }, + "node_modules/@dnd-kit/utilities": { + "version": "3.2.2", + "resolved": "https://registry.npmjs.org/@dnd-kit/utilities/-/utilities-3.2.2.tgz", + "integrity": "sha512-+MKAJEOfaBe5SmV6t34p80MMKhjvUz0vRrvVJbPT0WElzaOJ/1xs+D+KDv+tD/NE5ujfrChEcshd4fLn0wpiqg==", + "dependencies": { + "tslib": "^2.0.0" + }, + "peerDependencies": { + "react": ">=16.8.0" + } + }, "node_modules/@esbuild/aix-ppc64": { "version": "0.25.12", "resolved": "https://registry.npmmirror.com/@esbuild/aix-ppc64/-/aix-ppc64-0.25.12.tgz", @@ -1334,6 +1391,14 @@ "resolved": "https://registry.npmjs.org/@floating-ui/utils/-/utils-0.2.11.tgz", "integrity": "sha512-RiB/yIh78pcIxl6lLMG0CgBXAZ2Y0eVHqMPYugu+9U0AeT6YBeiJpf7lbdJNIugFP5SIjwNRgo4DhR1Qxi26Gg==" }, + "node_modules/@fontsource/noto-sans-sc": { + "version": "5.2.9", + "resolved": "https://registry.npmjs.org/@fontsource/noto-sans-sc/-/noto-sans-sc-5.2.9.tgz", + "integrity": "sha512-bTUIWGBgJDpwi5qAr+x0/lcgv80IHTB9vl6s2f6EymZEa7qYV99yNRBZuKFT+SYDKVunZrjCEhWtpxqmbXWl5Q==", + "funding": { + "url": "https://github.com/sponsors/ayuhito" + } + }, "node_modules/@iconify/types": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/@iconify/types/-/types-2.0.0.tgz", @@ -4995,6 +5060,12 @@ "@types/estree": "*" } }, + "node_modules/@types/file-saver": { + "version": "2.0.7", + "resolved": "https://registry.npmjs.org/@types/file-saver/-/file-saver-2.0.7.tgz", + "integrity": "sha512-dNKVfHd/jk0SkR/exKGj2ggkB45MAkzvWCaqLUUgkyjITkGNzH8H+yUwr+BLJUBjZOe9w8X3wgmXhZDRg1ED6A==", + "dev": true + }, "node_modules/@types/geojson": { "version": "7946.0.16", "resolved": "https://registry.npmjs.org/@types/geojson/-/geojson-7946.0.16.tgz", @@ -6146,9 +6217,9 @@ } }, "node_modules/dayjs": { - "version": "1.11.20", - "resolved": "https://registry.npmjs.org/dayjs/-/dayjs-1.11.20.tgz", - "integrity": "sha512-YbwwqR/uYpeoP4pu043q+LTDLFBLApUP6VxRihdfNTqu4ubqMlGDLd6ErXhEgsyvY0K6nCs7nggYumAN+9uEuQ==" + "version": "1.11.21", + "resolved": "https://registry.npmjs.org/dayjs/-/dayjs-1.11.21.tgz", + "integrity": "sha512-98IT+HOahAisibz/yjKbzuOBwYcjJ7BCLPzARyHiyEBmRz4fatF+KPJszEHXsGYjUG234aH/cOjW1wwTbKUZlA==" }, "node_modules/debug": { "version": "4.4.3", @@ -6228,6 +6299,12 @@ "url": "https://github.com/sponsors/wooorm" } }, + "node_modules/dnd-kit": { + "version": "0.0.2", + "resolved": "https://registry.npmjs.org/dnd-kit/-/dnd-kit-0.0.2.tgz", + "integrity": "sha512-d8AYd6I7D2b5u882+QNVGw0slBAt851/LWZ2j/pU+onf5/TGEKXeb47sCyhPYKEAUXp4oLfvWfNCqfkU03R1lw==", + "deprecated": "Package no longer supported. Contact Support at https://www.npmjs.com/support for more info." + }, "node_modules/docx": { "version": "9.6.1", "resolved": "https://registry.npmjs.org/docx/-/docx-9.6.1.tgz", @@ -6470,6 +6547,11 @@ } } }, + "node_modules/file-saver": { + "version": "2.0.5", + "resolved": "https://registry.npmjs.org/file-saver/-/file-saver-2.0.5.tgz", + "integrity": "sha512-P9bmyZ3h/PRG+Nzga+rbdI4OEpNDzAVyy74uVO9ATgzLK6VtAsYybF/+TOCvrc0MO793d6+42lLyZTw7/ArVzA==" + }, "node_modules/format": { "version": "0.2.2", "resolved": "https://registry.npmjs.org/format/-/format-0.2.2.tgz", diff --git a/frontend-web/package.json b/frontend-web/package.json index f7eed944..c977bb51 100644 --- a/frontend-web/package.json +++ b/frontend-web/package.json @@ -22,6 +22,7 @@ "@dnd-kit/core": "^6.3.1", "@dnd-kit/sortable": "^10.0.0", "@dnd-kit/utilities": "^3.2.2", + "@fontsource/noto-sans-sc": "^5.2.9", "@langchain/core": "^1.1.15", "@langchain/langgraph-sdk": "^1.5.3", "@radix-ui/react-avatar": "^1.1.11", diff --git a/frontend-web/src/main.tsx b/frontend-web/src/main.tsx index 924e3e76..b7b08da6 100644 --- a/frontend-web/src/main.tsx +++ b/frontend-web/src/main.tsx @@ -1,6 +1,10 @@ -import "./styles/globals.css"; -import "tdesign-react/es/style/index.css"; -import "katex/dist/katex.min.css"; +import "@fontsource/noto-sans-sc/chinese-simplified-400.css"; +import "@fontsource/noto-sans-sc/chinese-simplified-500.css"; +import "@fontsource/noto-sans-sc/chinese-simplified-600.css"; +import "@fontsource/noto-sans-sc/chinese-simplified-700.css"; +import "./styles/globals.css"; +import "tdesign-react/es/style/index.css"; +import "katex/dist/katex.min.css"; import React from "react"; import ReactDOM from "react-dom/client"; diff --git a/frontend-web/src/roundtable-planning/components/DispatchChainModal.tsx b/frontend-web/src/roundtable-planning/components/DispatchChainModal.tsx index 41a441c5..400bc7e9 100644 --- a/frontend-web/src/roundtable-planning/components/DispatchChainModal.tsx +++ b/frontend-web/src/roundtable-planning/components/DispatchChainModal.tsx @@ -29,6 +29,7 @@ import { import { cn } from "@/lib/utils"; import { getAvatar } from "./Avatars"; +import { MarkdownContent } from "./MessageBubble"; import { chainTagClasses } from "../lib/business-chain"; import { getAvatarTypeFor } from "../lib/roundtable-constants"; import type { @@ -140,7 +141,7 @@ export function DispatchChainModal({ return ( - + {status === "running" && ( @@ -223,9 +224,9 @@ export function DispatchChainModal({
{getAvatar(avatarForDialogue(d), "h-7 w-7")}
-
{d.name}
-
- {d.content || "…"} +
{d.name}
+
+ {d.content || "…"}
@@ -289,10 +290,15 @@ export function DispatchChainModal({ 停止 )} - + {/* 只对「运行中」任务隐藏入口——它正在后台真跑,进入会暂停前端编排、 + 与后台作业双跑或借此重跑。其它状态（排队 / 待澄清 / 已完成 / 已取消 / + 出错）都不是正在跑的后台作业,可正常查看。 */} + {status !== "running" && ( + + )}
diff --git a/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx b/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx index e2582397..726380f7 100644 --- a/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx +++ b/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx @@ -1642,32 +1642,39 @@ function IntentSummaryCard({ {intentReady ? ( <>
    -
  • - • - +
  • + • + 目标: {intentReady.objective}
{collapsible && ( - +
+ +
)} {showFullDetail && (
    diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py b/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py index 4353da31..af624952 100644 --- a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py +++ b/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py @@ -22,6 +22,12 @@ from typing import Any from langgraph.checkpoint.base import empty_checkpoint +from app.gateway.roundtable_run_policy import roundtable_run_policy +from app.gateway.routers._roundtable_seed import ( + COORDINATOR_AGENT_ID, + REPORT_AGENT_ID, + ensure_roundtable_functional_agents, +) from deerflow.agents.roundtable_orchestrator.types import ( LeaderTurn, ReportResult, @@ -31,22 +37,14 @@ from deerflow.agents.roundtable_orchestrator.types import ( from deerflow.config.paths import get_paths from deerflow.runtime.serialization import serialize_channel_values -from app.gateway.routers._roundtable_seed import ( - COORDINATOR_AGENT_ID, - REPORT_AGENT_ID, - ensure_roundtable_functional_agents, -) - logger = logging.getLogger(__name__) _ROUNDTABLE_THREAD_METADATA: dict[str, Any] = {"thread_type": "roundtable", "system": True} _REPORT_FILE = "方案总览.html" -# 各角色的工具裁剪（镜像 multi_agent 的 _leader_run / _special_run）。 -_LEADER_EXCLUDED = ["web_search"] -_LEADER_SKILL_STOP = ["agent_orchestration"] -_SEAT_EXCLUDED = ["ask_clarification", "present_files", "view_image", "agent_orchestration"] -_REPORT_EXCLUDED = [*_SEAT_EXCLUDED, "bash", "str_replace"] +# 各角色的工具裁剪 / thinking / reasoning 全部取自单一数据源 roundtable_run_policy +# （与 multi_agent 的 _leader_run / _special_run 同源），不再在本文件硬编码——历史上 +# 这里漏禁了 web_search，导致后台席位 / 报告能联网搜索而前端不能，现由 policy 统一。 # ── 进程内底层操作 ─────────────────────────────────────────────────────────── @@ -101,6 +99,56 @@ def _message_text(message: dict[str, Any] | None) -> str: return "" +def _latest_ai_text(messages: list[dict[str, Any]]) -> str: + """取**最后一条有正文**的 AI 消息文本。 + + 席位常以「调一堆工具」的 AIMessage 收尾（正文为空、只有 tool_calls，甚至 dangling）， + 直接取最后一条 AI 消息会拿到空串 → 报告就拿不到席位产物。这里向前找到真正有文字的那条。 + """ + for m in reversed(messages): + if not isinstance(m, dict): + continue + if (m.get("type") or m.get("role")) not in {"ai", "assistant"}: + continue + text = _message_text(m) + if text.strip(): + return text + return "" + + +def _normalize_tool_calls(tool_calls: Any) -> list[dict[str, Any]]: + """把 checkpoint 里的 tool_calls 归一化成 {name, args, id}（丢掉无名项）。""" + out: list[dict[str, Any]] = [] + for tc in tool_calls or []: + if not isinstance(tc, dict): + continue + name = tc.get("name") or "" + if not name: + continue + out.append({"name": name, "args": tc.get("args") or {}, "id": tc.get("id") or ""}) + return out + + +def _tool_calls_since_last_human(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + """收集「最后一条 human 消息之后」各 AI 消息的 tool_calls（即**本轮**调用），保序。 + + 席位 thread 跨轮累积历史，直接扫全量会把往轮的工具调用也算进来；以最后一条 human + （本轮 run 的输入）为界，只取其后的 AI 工具调用，对应前端「这一条气泡的步骤卡」。 + """ + last_human = -1 + for i, m in enumerate(messages): + if isinstance(m, dict) and (m.get("type") or m.get("role")) in {"human", "user"}: + last_human = i + out: list[dict[str, Any]] = [] + for m in messages[last_human + 1 :]: + if not isinstance(m, dict): + continue + if (m.get("type") or m.get("role")) not in {"ai", "assistant"}: + continue + out.extend(_normalize_tool_calls(m.get("tool_calls"))) + return out + + async def _run_agent_turn( app, *, @@ -216,27 +264,31 @@ class InProcessRoundtableGateway: if self._first_leader and self._seat_intro: msg = self._seat_intro + "\n\n" + message self._first_leader = False + policy = roundtable_run_policy("leader") messages = await _run_agent_turn( self._app, thread_id=thread_ids[coordinator_name], agent_name=coordinator_name, message=msg, model=model or self._model, - excluded_tools=_LEADER_EXCLUDED, - skill_stop_names=_LEADER_SKILL_STOP, - thinking_enabled=False, - reasoning_effort="low", + excluded_tools=policy.excluded_tools, + skill_stop_names=policy.skill_stop_names, + thinking_enabled=policy.thinking_enabled, + reasoning_effort=policy.reasoning_effort, ) ai = _latest_ai_message(messages) - final_text = _message_text(ai) + # tool_calls 取最后一条 AI 消息（派活就在这条）；正文取最后一条有文字的 AI 消息。 + final_text = _latest_ai_text(messages) tool_calls = (ai.get("tool_calls") if ai else None) or [] + # 本轮工具调用（派活 / 澄清），带进 LeaderTurn 供草稿 run 重建步骤卡。 + turn_tool_calls = _normalize_tool_calls(tool_calls) # 澄清优先 clar = next((tc for tc in tool_calls if (tc.get("name") or "").lower() == "ask_clarification"), None) if clar is not None: cargs = clar.get("args") or {} question = str(cargs.get("question") or "").strip() or "总控需要补充信息，请补充后继续。" - return LeaderTurn(final_text=final_text, clarification=question) + return LeaderTurn(final_text=final_text, clarification=question, tool_calls=turn_tool_calls) # 派活解析（复用 multi_agent 纯函数） from app.gateway.routers.multi_agent import _extract_dispatch @@ -249,38 +301,40 @@ class InProcessRoundtableGateway: dispatched.append((name, task)) except Exception: # noqa: BLE001 logger.warning("dispatch parse failed for tool_call", exc_info=True) - return LeaderTurn(final_text=final_text, dispatched=dispatched) + return LeaderTurn(final_text=final_text, dispatched=dispatched, tool_calls=turn_tool_calls) async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn: + policy = roundtable_run_policy("seat") messages = await _run_agent_turn( self._app, thread_id=thread_ids[agent_id], agent_name=agent_id, message=task, model=model or self._model, - excluded_tools=_SEAT_EXCLUDED, - skill_stop_names=[], - thinking_enabled=True, - reasoning_effort="medium", + excluded_tools=policy.excluded_tools, + skill_stop_names=policy.skill_stop_names, + thinking_enabled=policy.thinking_enabled, + reasoning_effort=policy.reasoning_effort, ) - return SeatTurn(final_text=_message_text(_latest_ai_message(messages))) + return SeatTurn(final_text=_latest_ai_text(messages), tool_calls=_tool_calls_since_last_human(messages)) async def create_report_thread(self) -> str: ensure_roundtable_functional_agents() return await _create_thread(self._app) async def run_report(self, *, report_thread_id, prompt, model) -> ReportResult: + policy = roundtable_run_policy("report") messages = await _run_agent_turn( self._app, thread_id=report_thread_id, agent_name=REPORT_AGENT_ID, message=prompt, model=model or self._model, - excluded_tools=_REPORT_EXCLUDED, - skill_stop_names=[], - thinking_enabled=True, - reasoning_effort="medium", + excluded_tools=policy.excluded_tools, + skill_stop_names=policy.skill_stop_names, + thinking_enabled=policy.thinking_enabled, + reasoning_effort=policy.reasoning_effort, ) - summary = _message_text(_latest_ai_message(messages)) + summary = _latest_ai_text(messages) html = _read_report_html(self._app, report_thread_id, self._user_id) or summary return ReportResult(html=html, summary=summary, model=model or self._model) diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py b/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py index b08c9ab4..e7530b64 100644 --- a/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py +++ b/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py @@ -17,7 +17,8 @@ from __future__ import annotations import asyncio import logging from dataclasses import dataclass, field -from typing import Callable +from datetime import UTC, datetime +from typing import Any, Callable from deerflow.agents.roundtable_orchestrator import ( JobProgressWriter, @@ -36,6 +37,115 @@ class _JobUser: self.id = user_id +# 头像配色：与前端 getAvatarTypeFor 对齐（内置圆桌席位固定色，其余按序轮换）。 +_BUILTIN_AVATAR = { + "roundtable-intelligence": "purple", + "roundtable-environment": "emerald", + "roundtable-solution-design": "human", + "roundtable-risk-review": "amber", + "roundtable-execution-plan": "pink", + "roundtable-summary": "blue", +} +_FALLBACK_AVATAR = ["blue", "purple", "emerald", "amber", "pink"] + + +def _avatar_for(agent_id: str, index: int) -> str: + return _BUILTIN_AVATAR.get(agent_id) or _FALLBACK_AVATAR[index % len(_FALLBACK_AVATAR)] + + +# 工具步骤卡的回退文案。前端 getStepDisplay 对 tool 步骤是从 key+args 推图标/文案的, +# 存的 label 基本不展示——这里给个合理回退即可（与前端默认文案对齐）。 +_STEP_LABEL = { + "agent_orchestration": "派活给子智能体", + "write_file": "写入文件", + "str_replace": "写入文件", + "read_file": "读取文件", + "ls": "列出文件夹", + "bash": "执行命令", + "web_search": "搜索相关信息", + "ask_clarification": "需要你的协助", + "present_files": "展示文件", +} + + +def _steps_from_tool_calls(tool_calls: list[dict], dialogue_id: object, ts: str) -> list[dict]: + """把一轮 tool_calls 映射成前端 ``Step2Dialogue.steps``（工具步骤卡）。 + + - ``key`` = ``tool_calling:``——前端 ``getStepDisplay`` 据此推图标 + 文案; + - ``id`` = ``-``——对齐前端 ``${bubbleId}-${stepIndex}``; + - ``args`` 原样带上（write_file 的 path/content、派活的 agent_name/task 等）。 + """ + steps: list[dict] = [] + for i, tc in enumerate(tool_calls or []): + name = tc.get("name") or "" + if not name: + continue + steps.append( + { + "id": f"{dialogue_id}-{i}", + "key": f"tool_calling:{name}", + "label": _STEP_LABEL.get(name, f"使用 “{name}” 工具"), + "ts": ts, + "args": tc.get("args") or {}, + } + ) + return steps + + +def _seat_stub_messages(dialogues: list[dict]) -> dict[str, list[dict]]: + """从各对话的 write_file 调用重建 ``seatStubMessages``（threadId → [stub ai message]）。 + + 对齐前端 ``upsertSeatToolCall`` 的 stub 形状与 id 约定（消息 ``roundtable-stub-``、 + 工具调用 ``tc--``），使加载草稿后点开产物的虚拟 URL 能正确解析到内容。 + 同一 thread 跨轮写多次按 path 去重（后写覆盖），与前端一致。 + """ + by_tid: dict[str, dict[str, dict]] = {} + for d in dialogues: + tid = d.get("threadId") + if not tid: + continue + for tc in d.get("toolCalls") or []: + if tc.get("name") != "write_file": + continue + path = str((tc.get("args") or {}).get("path") or "") + by_tid.setdefault(tid, {})[path] = tc # 后写覆盖 + out: dict[str, list[dict]] = {} + for tid, by_path in by_tid.items(): + tool_calls = [ + {"id": f"tc-{tid}-{i}", "name": "write_file", "args": tc.get("args") or {}} + for i, tc in enumerate(by_path.values()) + ] + out[tid] = [{"id": f"roundtable-stub-{tid}", "type": "ai", "content": "", "tool_calls": tool_calls}] + return out + + +def _to_step2_dialogue(d: dict, idx_by_agent: dict[str, int]) -> dict: + """把作业的中性对话记录映射成前端 Step2Dialogue 形状（含 avatarType / steps / + agentThreadId / time），使写进草稿 step2.runs 后，前端 hydrateStep2Run 能像手动跑 + 一样直接渲染——步骤卡、时间戳、点开产物全部可用。""" + dialogue_id = d.get("id") + time_str = d.get("time") or "" + base: dict = { + "id": dialogue_id, + "content": d.get("content") or "", + "time": time_str, + "streaming": False, + "steps": _steps_from_tool_calls(d.get("toolCalls") or [], dialogue_id, time_str), + } + thread_id = d.get("threadId") + if thread_id: + base["agentThreadId"] = thread_id + if d.get("role") == "leader": + return {**base, "sender": "总控协调", "avatarType": "blue", "confidence": "派活说明"} + aid = d.get("agentId") or "" + return { + **base, + "sender": d.get("name") or aid, + "avatarType": _avatar_for(aid, idx_by_agent.get(aid, 0)), + "confidence": "子智能体交付", + } + + @dataclass class StartParams: job_id: str @@ -52,6 +162,8 @@ class StartParams: # 是否为「续跑」（resume 澄清）。首启=False → 真网关建全新线程（不复用前端线程， # 避免与前端本地研讨的活跃 run 撞 409）；续跑=True → 复用作业自己的线程。 is_resume: bool = False + # chain 模式来源链条（写进草稿 run 用，展示）。 + chain: dict[str, Any] | None = None # 网关工厂：给定一次启动参数，返回一个 RoundtableGateway 实例。 @@ -94,7 +206,19 @@ class RoundtableJobExecutor: except Exception: # noqa: BLE001 logger.warning("persist thread_ids for job %s failed", params.job_id, exc_info=True) - progress = JobProgressWriter(self._store, job_id=params.job_id, agents=params.agents) + # 每轮模型返回后，把对话作为一个带 jobId 的 run 写进草稿 step2.runs[] + # —— 和手动跑同一存取路径（draft_store.update_draft），加载草稿即可显示。 + async def _on_dialogue(dialogues: list[dict], consensus: int) -> None: + await self._write_run_to_draft( + params, dialogues=dialogues, status="running", consensus=consensus, thread_ids=thread_ids + ) + + progress = JobProgressWriter( + self._store, + job_id=params.job_id, + agents=params.agents, + on_dialogue=_on_dialogue if (self._draft_store and params.draft_id) else None, + ) result = await run_orchestration( gateway=gateway, progress=progress, @@ -106,26 +230,21 @@ class RoundtableJobExecutor: mode=params.mode, intent_text=params.intent_text, ) - # 完成且有报告： - # 1) 把 step3 挂到 job 上 —— 前端「多运行」模型从单个 job 即可拿到完整结果 - # （对话 dialogues + 报告 step3）组装一个 run； - # 2) 同时写回关联草稿的顶层 step3 + 推进 furthest_step=3，兼容既有 - # 「查看报告 → 加载草稿 → Step3」路径。 + # 终态：把最终对话 + step3 + jobStatus 落进同一个 run（含 furthest_step=3）。 + await self._write_run_to_draft( + params, + dialogues=result.dialogues, + status=result.status, + consensus=100 if result.status == "done" else 95, + step3=result.step3, + thread_ids=thread_ids, + ) + # step3 也挂 job（API / 弹窗用）。 if result.status == "done" and result.step3: try: await self._store.update_progress(params.job_id, step3=result.step3) except Exception: # noqa: BLE001 logger.warning("persist step3 to job %s failed", params.job_id, exc_info=True) - if params.draft_id and self._draft_store is not None: - try: - await self._draft_store.update_draft( - params.draft_id, - params.user_id, - step3=result.step3, - furthest_step=3, - ) - except Exception: # noqa: BLE001 — 写报告失败不影响作业本身已 done - logger.warning("persist step3 to draft %s failed", params.draft_id, exc_info=True) except asyncio.CancelledError: # 干净停止即可，**不在被取消的任务里写 DB**：取消态下二次 await 会再抛 # CancelledError、并可能打断进行中的 sqlite 连接。``cancelled`` 终态由 @@ -137,6 +256,72 @@ class RoundtableJobExecutor: finally: reset_current_user(token) + async def _write_run_to_draft( + self, + params: StartParams, + *, + dialogues: list[dict], + status: str, + consensus: int, + step3: dict | None = None, + thread_ids: dict[str, str] | None = None, + ) -> None: + """把本次后台作业作为一个带 jobId 的 run 写进草稿 step2.runs[]（同手动跑存取）。 + + 读现有 step2 → 按 run_id=``job-`` upsert（保留其它手动 run）→ 设为 activeRun + → update_draft。前端加载草稿即可像手动跑一样 hydrate 出对话。 + """ + if self._draft_store is None or not params.draft_id: + return + idx_by_agent = {a.agent_id: i for i, a in enumerate(params.agents)} + s2_dialogues = [_to_step2_dialogue(d, idx_by_agent) for d in dialogues] + last_leader = next( + (d.get("content") for d in reversed(dialogues) if d.get("role") == "leader"), "" + ) + run_id = f"job-{params.job_id}" + try: + draft = await self._draft_store.get_draft(params.draft_id, params.user_id) + except Exception: # noqa: BLE001 + draft = None + step2 = dict((draft or {}).get("step2") or {}) + runs = list(step2.get("runs") or []) + idx = next((i for i, r in enumerate(runs) if r.get("id") == run_id), -1) + created = runs[idx].get("createdAt") if idx >= 0 else datetime.now(UTC).isoformat() + prev_step3 = runs[idx].get("step3") if idx >= 0 else None + run = { + "id": run_id, + "title": "后台研讨", + "createdAt": created, + "source": params.mode, + "selectedAgents": [{"agent_id": a.agent_id, "name": a.name} for a in params.agents], + # 后台真建的 leader/seat 线程是真实持久的——存进去，前端加载后能像手动跑一样 + # 按 thread 打开各席位产物（HTTP outputs 兜底）。hydrate 是纯展示、绝不 resume， + # 不会触发重跑（init 闸门 + isRoundTableRunning=false 双重拦截）。空则存 None。 + "threadIds": (thread_ids or None), + "coordinatorName": params.coordinator_name, + "step2RoundtableDialogues": s2_dialogues, + "lastLeaderContent": last_leader or "", + "hasConsensus": status == "done", + "consensusPercentage": int(consensus), + "budgetLimit": 0, + # 从各席位 write_file 调用重建 stub，使加载草稿后点开产物的虚拟预览可用。 + "seatStubMessages": _seat_stub_messages(dialogues), + "orchestrationMode": params.mode, + "chain": params.chain, + "step3": step3 if step3 is not None else prev_step3, + "jobId": params.job_id, + "jobStatus": status, + } + runs = [run if i == idx else r for i, r in enumerate(runs)] if idx >= 0 else [*runs, run] + step2["runs"] = runs + step2["activeRunId"] = run_id + try: + await self._draft_store.update_draft( + params.draft_id, params.user_id, step2=step2, furthest_step=3 if status == "done" else 2 + ) + except Exception: # noqa: BLE001 — 写草稿失败不影响作业本身 + logger.warning("write run to draft %s failed", params.draft_id, exc_info=True) + def cancel(self, job_id: str) -> bool: """请求取消一个在跑的作业。返回是否命中在跑任务。""" task = self._tasks.get(job_id) diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_run_policy.py b/offline-backend-20260512/backend/app/gateway/roundtable_run_policy.py new file mode 100644 index 00000000..8843a618 --- /dev/null +++ b/offline-backend-20260512/backend/app/gateway/roundtable_run_policy.py @@ -0,0 +1,94 @@ +"""圆桌各角色「单轮 run 参数」的**单一数据源**。 + +历史问题：前端手动路径（``multi_agent.py`` 的 ``_leader_run`` / ``_special_run``） +和后台进程内网关（``roundtable_inprocess_gateway.py``）各自硬编码了一份「每角色 +工具裁剪 + thinking + reasoning」，结果**漂移**出差异（后台席位 / 报告少禁了 +``web_search``，能联网搜索而前端不能）。 + +本模块把这套策略收敛成唯一定义，两条路径都从这里取，保证「后台挂起」与「前端 +手动调用」传参**逐字一致**，且以后改一处即可，不再漂移。 + +角色（``role``）： + - ``leader`` —— 总控派活：短决策，关 thinking + low reasoning（首字节快）； + 只禁 ``web_search``；并配合 ``SkillStopMiddleware`` 在 ``agent_orchestration`` + 调用后停图，便于网关从 AIMessage.tool_calls 解析派活列表。 + - ``seat`` —— 子智能体交付：开 thinking + medium reasoning；禁派活 / + 澄清 / 展示文件 / 看图 / 联网搜索（与前端 ``_special_run`` 一致）。 + - ``report`` —— Step3 报告绘制：在 ``seat`` 基础上**额外禁 ``bash`` / + ``str_replace``**： + * 禁 ``bash`` → 杜绝它写 Python 脚本再跑出 HTML（脚本产物落在 outputs， + 但前端虚拟预览只抓 ``write_file`` 的 ``args.content``，抓不到脚本产物 → + 预览停在半截 ``write_file``，正是「没写完就去生成 py 文件」的现象）； + * 禁 ``str_replace`` → 强制返工用整文件 ``write_file`` 重写 + （``str_replace`` 没有 ``content`` 字段，会让前端虚拟预览读到 undefined + 而清空沙箱）。 +""" + +from __future__ import annotations + +from dataclasses import dataclass + +# ── 每角色基础工具裁剪（唯一定义处）───────────────────────────────────────────── +# +# 注意：``seat`` / ``report`` 都包含 ``web_search`` —— 圆桌研讨里联网搜索一律禁用 +# （与前端 ``_special_run`` 的 excluded_tools 对齐）。leader 也禁 ``web_search``。 +_LEADER_EXCLUDED: list[str] = ["web_search"] +_SEAT_EXCLUDED: list[str] = ["web_search", "ask_clarification", "present_files", "view_image", "agent_orchestration"] +_REPORT_EXCLUDED: list[str] = [*_SEAT_EXCLUDED, "bash", "str_replace"] + +# leader 内禀的 skill_stop：``agent_orchestration`` 调用后停图。调用方可在此基础上 +# 再叠加自己的 skill_stop（见 ``merge_skill_stop_names``）。 +_LEADER_SKILL_STOP: list[str] = ["agent_orchestration"] + + +@dataclass(frozen=True) +class RoundtableRunPolicy: + """一个角色单轮 run 的归一化参数。 + + 与 ``multi_agent._run_payload`` 的入参一一对应：``excluded_tools`` / + ``skill_stop_names`` / ``thinking_enabled`` / ``reasoning_effort``。 + """ + + excluded_tools: list[str] + skill_stop_names: list[str] + thinking_enabled: bool + reasoning_effort: str + + +def roundtable_run_policy(role: str) -> RoundtableRunPolicy: + """返回某角色（``leader`` / ``seat`` / ``report``）的单轮 run 参数。 + + 返回值里的列表是**新副本**，调用方可安全地再 append/merge 而不污染本模块常量。 + 未知角色抛 ``ValueError``（早失败，避免悄悄退化到错误的工具面）。 + """ + if role == "leader": + return RoundtableRunPolicy( + excluded_tools=list(_LEADER_EXCLUDED), + skill_stop_names=list(_LEADER_SKILL_STOP), + thinking_enabled=False, + reasoning_effort="low", + ) + if role == "seat": + return RoundtableRunPolicy( + excluded_tools=list(_SEAT_EXCLUDED), + skill_stop_names=[], + thinking_enabled=True, + reasoning_effort="medium", + ) + if role == "report": + return RoundtableRunPolicy( + excluded_tools=list(_REPORT_EXCLUDED), + skill_stop_names=[], + thinking_enabled=True, + reasoning_effort="medium", + ) + raise ValueError(f"unknown roundtable role: {role!r} (expected leader/seat/report)") + + +def merge_skill_stop_names(policy_names: list[str], caller_names: list[str] | None) -> list[str]: + """把角色内禀 skill_stop 与调用方追加的 skill_stop 合并去重（保序）。 + + 取代历史上 ``list({"agent_orchestration", *skill_stop_names})`` 的 set 写法 + （set 顺序不确定）；这里用 dict.fromkeys 保序去重，结果确定、便于断言。 + """ + return list(dict.fromkeys([*policy_names, *(caller_names or [])])) 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 199c366e..3f5faf4c 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py +++ b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py @@ -57,18 +57,20 @@ import uuid from typing import Any, AsyncIterator import httpx +from fastapi import APIRouter, Body, HTTPException, Request +from fastapi.responses import StreamingResponse +from langgraph.checkpoint.base import empty_checkpoint + +from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable_run_policy from app.gateway.routers._roundtable_seed import ( COORDINATOR_AGENT_ID, REPORT_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 +from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple from deerflow.utils.time import now_iso -from fastapi import APIRouter, Body, HTTPException, Request -from fastapi.responses import StreamingResponse -from langgraph.checkpoint.base import empty_checkpoint logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/multi-agent", tags=["multi-agent"]) @@ -1008,19 +1010,20 @@ async def _leader_run( _broadcast(client, base, headers, "总控智能体收到用户消息:" + new_message, agent_threads, [agent_name]) ) thread_id = agent_threads[agent_name] - effective_stop_names = list({"agent_orchestration", *skill_stop_names}) + # 角色参数取自单一数据源 roundtable_run_policy（leader：禁 web_search、停 + # agent_orchestration 图、关 thinking + low reasoning——短决策、首字节快）。 + # 调用方可再叠加自己的 skill_stop。后台进程内网关用同一份 policy，保证传参一致。 + policy = roundtable_run_policy("leader") + effective_stop_names = merge_skill_stop_names(policy.skill_stop_names, skill_stop_names) body = _run_payload( agent_name=agent_name, model_name=model_name, thread_id=thread_id, new_message=new_message, - excluded_tools=["web_search"], + excluded_tools=policy.excluded_tools, skill_stop_names=effective_stop_names, - # 总控派活是"下一步派谁、什么任务"的短决策，不需要 CoT。 - # 关掉 thinking + low reasoning 让"调度中"几乎瞬间转入"正在派活"， - # 子智能体仍保留 thinking（在 _special_run 里默认 True），保证交付质量。 - thinking_enabled=False, - reasoning_effort="low", + thinking_enabled=policy.thinking_enabled, + reasoning_effort=policy.reasoning_effort, ) t_payload_built = time.perf_counter() logger.info( @@ -1163,23 +1166,20 @@ async def _special_run( """ t0 = time.perf_counter() thread_id = agent_threads[agent_name] - excluded_tools = ["web_search", "ask_clarification", "present_files", "view_image", "agent_orchestration"] - if agent_name == REPORT_AGENT_ID: - # Step 3「结果绘制」的报告智能体只需**直接 write_file 写出完整 HTML**。 - # 额外禁用 bash / str_replace： - # - 禁 bash → 杜绝它写 Python/脚本再跑出 HTML（脚本产物落在 outputs 里， - # 但前端虚拟预览只抓 write_file 的 args.content，抓不到脚本产物 → 预览停在 - # 那个半截 write_file 上，正是用户看到的「没写完就去生成 py 文件」）； - # - 禁 str_replace → 强制返工用整文件 write_file 重写（str_replace 没有 content - # 字段，会让前端虚拟预览读到 undefined 而清空沙箱）。 - excluded_tools = [*excluded_tools, "bash", "str_replace"] + # 角色参数取自单一数据源 roundtable_run_policy（seat / report）。report 在 seat + # 基础上额外禁 bash / str_replace（理由见 roundtable_run_policy 模块注释）。 + # 后台进程内网关用同一份 policy，保证「后台挂起」与「前端手动调用」传参一致。 + role = "report" if agent_name == REPORT_AGENT_ID else "seat" + policy = roundtable_run_policy(role) body = _run_payload( agent_name=agent_name, model_name=model_name, thread_id=thread_id, new_message=new_message, - excluded_tools=excluded_tools, - skill_stop_names=skill_stop_names, + excluded_tools=policy.excluded_tools, + skill_stop_names=merge_skill_stop_names(policy.skill_stop_names, skill_stop_names), + thinking_enabled=policy.thinking_enabled, + reasoning_effort=policy.reasoning_effort, ) t_payload_built = time.perf_counter() logger.info( diff --git a/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py b/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py index 4e3d4b9e..6a6e0af4 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py +++ b/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py @@ -203,6 +203,7 @@ async def start_job(request: Request, body: StartJobRequest) -> RoundtableJobRes thread_ids=body.threadIds, coordinator_name=body.coordinatorName, draft_id=body.draftId, + chain=body.chain, ) ) return _to_response(row) @@ -287,6 +288,7 @@ async def resume_job(request: Request, job_id: str, body: ResumeJobRequest) -> R coordinator_name=row.get("coordinator_name") or "roundtable-coordinator", draft_id=row.get("draft_id"), is_resume=True, + chain=row.get("chain"), ) ) return _to_response(updated or row) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py index 9f21212e..17aaa51f 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py @@ -47,6 +47,7 @@ class OrchestrationResult: pending_clarification: str | None = None error: str | None = None last_leader_content: str = "" + dialogues: list = field(default_factory=list) @dataclass @@ -85,12 +86,13 @@ async def run_orchestration( mode: "recommend" | "chain"。 intent_text: 任务意图，用于报告 prompt（缺省退回 seed_message）。 """ + intent = (intent_text or seed_message or "").strip() try: await progress.begin() if mode == "chain": - outcome = await _run_chain(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message) + outcome = await _run_chain(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent) else: - outcome = await _run_recommend(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message) + outcome = await _run_recommend(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent) if outcome.kind == "clarification": question = outcome.question or "总控需要补充信息，请补充后继续。" @@ -99,6 +101,7 @@ async def run_orchestration( status="awaiting_input", pending_clarification=question, last_leader_content=outcome.last_leader_content, + dialogues=progress.dialogues(), ) # consensus 或 max_cycles：均收口 → 报告绘制 @@ -118,14 +121,48 @@ async def run_orchestration( consensus=True, step3=step3, last_leader_content=outcome.last_leader_content, + dialogues=progress.dialogues(), ) except Exception as exc: # noqa: BLE001 — 顶层兜底：标记 error 并返回 logger.exception("roundtable orchestration failed") await progress.error(str(exc)) - return OrchestrationResult(status="error", error=str(exc)) + return OrchestrationResult(status="error", error=str(exc), dialogues=progress.dialogues()) -# ── recommend 模式 ─────────────────────────────────────────────────────────── +# ── 上下文「广播」（inline 实现）──────────────────────────────────────────────── +# +# 进程内网关没有 multi_agent 的 thread 广播，席位/总控会看不到「原始任务背景 + 各席位 +# 前序交付」→ 席位反问用户、总控靠记忆瞎编。这里把这些信息**直接拼进 prompt**： +# - 每个席位的任务带上：任务背景 + 前序各席位交付 + 「直接产出、别提问」硬约束； +# - 总控的续轮 prompt 带上：各席位本轮交付，让它据实派活/收口。 + +_SEAT_NO_ASK = ( + "注意：这是自动化圆桌研讨，没有用户在线应答——**请勿向用户提问或等待确认**，" + "直接基于已知信息做出合理假设并产出完整的结构化交付（Markdown）。" +) + + +def _deliveries_block(deliveries: dict[str, str], name_by_id: dict[str, str]) -> str: + parts: list[str] = [] + for aid, text in deliveries.items(): + t = (text or "").strip() + if not t: + continue + parts.append(f"### {name_by_id.get(aid, aid)}\n{t}") + return "\n\n".join(parts) + + +def _seat_message(intent_text: str, deliveries: dict[str, str], name_by_id: dict[str, str], task: str) -> str: + """把任务背景 + 前序交付 + 防提问约束拼进席位任务（替代 thread 广播）。""" + parts: list[str] = [] + if intent_text.strip(): + parts.append(f"【本次研讨任务背景】\n{intent_text.strip()}") + block = _deliveries_block(deliveries, name_by_id) + if block: + parts.append("【前序各席位已完成的交付，请在此基础上承接、补充、不要重复或推翻】\n" + block) + parts.append(f"【你的本轮具体任务】\n{task}") + parts.append(_SEAT_NO_ASK) + return "\n\n---\n\n".join(parts) async def _run_recommend( @@ -136,6 +173,7 @@ async def _run_recommend( coordinator_name: str, model: str | None, seed_message: str, + intent_text: str, ) -> _LoopOutcome: deliveries: dict[str, str] = {} last_leader = "" @@ -152,24 +190,32 @@ async def _run_recommend( ) if turn.final_text: last_leader = turn.final_text - await progress.add_leader_dialogue(turn.final_text, cycle) + await progress.add_leader_dialogue( + turn.final_text, cycle, tool_calls=turn.tool_calls, thread_id=thread_ids.get(coordinator_name) + ) if turn.is_clarification: return _LoopOutcome("clarification", question=turn.clarification, last_leader_content=last_leader, deliveries=deliveries) if turn.is_consensus: return _LoopOutcome("consensus", last_leader_content=last_leader, deliveries=deliveries) - # **顺序**派活 —— 不要并行 + # **顺序**派活 —— 不要并行；每个席位带上「背景 + 前序交付」上下文 await progress.dispatching(cycle) for agent_id, task in turn.dispatched: await progress.seat_active(agent_id) - seat = await gateway.run_seat(thread_ids=thread_ids, agent_id=agent_id, task=task, model=model) + seat_msg = _seat_message(intent_text, deliveries, name_by_id, task) + seat = await gateway.run_seat(thread_ids=thread_ids, agent_id=agent_id, task=seat_msg, model=model) deliveries[agent_id] = seat.final_text - await progress.add_seat_dialogue(agent_id, name_by_id.get(agent_id, agent_id), seat.final_text, cycle) + await progress.add_seat_dialogue( + agent_id, name_by_id.get(agent_id, agent_id), seat.final_text, cycle, + tool_calls=seat.tool_calls, thread_id=thread_ids.get(agent_id), + ) await progress.seat_done(agent_id, cycle) await progress.bump_consensus() - message = CONTINUE_PROMPT + # 续轮：把各席位本轮交付喂给总控，让它据实判断（总控 SOUL 依赖看到交付） + block = _deliveries_block(deliveries, name_by_id) + message = CONTINUE_PROMPT + (f"\n\n各席位最新交付如下：\n{block}" if block else "") # 达到最大轮次：强制收口（对齐前端 toast「已达到最大研讨轮次」） logger.info("roundtable recommend loop hit MAX_CYCLES=%d, forcing consensus", MAX_CYCLES) @@ -187,13 +233,15 @@ async def _run_chain( coordinator_name: str, model: str | None, seed_message: str, + intent_text: str, ) -> _LoopOutcome: deliveries: dict[str, str] = {} last_leader = "" total = len(agents) + name_by_id = {a.agent_id: a.name for a in agents} for i, seat in enumerate(agents): - # (a) 总控按链条顺序只给当前席位派活 + # (a) 总控按链条顺序只给当前席位派活；带上前序各席位交付，让它据实派活 await progress.leader_thinking(i + 1) leader_task = build_chain_leader_task( seed_message=seed_message, @@ -203,6 +251,9 @@ async def _run_chain( seat_name=seat.name, is_first=(i == 0), ) + block = _deliveries_block(deliveries, name_by_id) + if block: + leader_task += f"\n\n前序各席位已完成的交付如下，请据此为本席位分配承接性任务：\n{block}" turn = await gateway.run_leader( thread_ids=thread_ids, coordinator_name=coordinator_name, @@ -211,34 +262,44 @@ async def _run_chain( ) if turn.final_text: last_leader = turn.final_text - await progress.add_leader_dialogue(turn.final_text, i + 1) + await progress.add_leader_dialogue( + turn.final_text, i + 1, tool_calls=turn.tool_calls, thread_id=thread_ids.get(coordinator_name) + ) if turn.is_clarification: return _LoopOutcome("clarification", question=turn.clarification, last_leader_content=last_leader, deliveries=deliveries) task = pick_chain_task(turn.dispatched, turn.final_text, seat.agent_id, seat.name) - # (b) 当前席位交付（顺序，不并行） + # (b) 当前席位交付（顺序，不并行）；带上「背景 + 前序交付 + 别提问」上下文 await progress.seat_active(seat.agent_id) - seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=task, model=model) + seat_msg = _seat_message(intent_text, deliveries, name_by_id, task) + seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model) deliveries[seat.agent_id] = seat_turn.final_text - await progress.add_seat_dialogue(seat.agent_id, seat.name, seat_turn.final_text, i + 1) + await progress.add_seat_dialogue( + seat.agent_id, seat.name, seat_turn.final_text, i + 1, + tool_calls=seat_turn.tool_calls, thread_id=thread_ids.get(seat.agent_id), + ) await progress.seat_done(seat.agent_id, i + 1) if total > 0: await progress.set_consensus_pct(round((i + 1) / total * 90)) - # (c) 总控收口：综合全链交付，产出最终结论 + # (c) 总控收口：综合全链交付，产出最终结论（带上全部席位交付） if total > 0: await progress.leader_thinking(total + 1) + block = _deliveries_block(deliveries, name_by_id) + closeout_msg = CHAIN_CLOSEOUT_PROMPT + (f"\n\n各席位完整交付如下：\n{block}" if block else "") closeout = await gateway.run_leader( thread_ids=thread_ids, coordinator_name=coordinator_name, - message=CHAIN_CLOSEOUT_PROMPT, + message=closeout_msg, model=model, ) if closeout.final_text: last_leader = closeout.final_text - await progress.add_leader_dialogue(closeout.final_text, total + 1) + await progress.add_leader_dialogue( + closeout.final_text, total + 1, tool_calls=closeout.tool_calls, thread_id=thread_ids.get(coordinator_name) + ) return _LoopOutcome("consensus", last_leader_content=last_leader, deliveries=deliveries) @@ -279,27 +340,43 @@ def build_report_prompt( 席位交付按 ``agents`` 顺序列出（缺交付的席位跳过）。报告智能体据此用 write_file 写出自包含的 ``outputs/方案总览.html``（具体绘制规则在 roundtable-report 的 SOUL）。 """ + # 与前端 useStep3Report.buildReportPrompt 对齐 —— 关键是**明确叫报告 agent 用 + # write_file 写到 /mnt/user-data/outputs/方案总览.html**，否则它只会在对话里回文字、 + # 不落 HTML 文件（_read_report_html 就读不到）。 + name_by_id = {a.agent_id: a.name for a in agents} lines: list[str] = [ - "【圆桌最终成果汇总 —— 请据此绘制方案总览 HTML 看板】", + "请基于以下圆桌多智能体研讨的成果，撰写一份图文并茂、以文字叙述为主线的 HTML「方案研讨汇报」，" + "用 write_file 工具写入 /mnt/user-data/outputs/方案总览.html。", + "目标：让用户能轻松读懂这群智能体讨论出了什么方案、为什么是它。请逐个席位提炼其核心观点与交付写成文字，" + "再综合成最终方案；图表用来点睛（风险矩阵、时间线、资源占比、贡献雷达等），详略得当——" + "缺明确数字时也要据研讨内容合理估算出图并标注「定性/估算」，不要通篇只有文字。", "", - "## 任务意图", + "# 一、任务意图", (intent_text or "").strip() or "（未提供）", "", - "## 各席位交付", + "# 二、各席位最终交付", ] - name_by_id = {a.agent_id: a.name for a in agents} has_delivery = False - for a in agents: + for i, a in enumerate(agents): text = (deliveries.get(a.agent_id) or "").strip() if not text: continue has_delivery = True - lines.append(f"### {name_by_id.get(a.agent_id, a.agent_id)}") + lines.append(f"## {i + 1}. {name_by_id.get(a.agent_id, a.agent_id)}") lines.append(text) lines.append("") if not has_delivery: - lines.append("（各席位暂无可用交付，请基于任务意图与总控结论绘制。）") + lines.append("（各席位暂无文字交付，请基于任务意图与总控结论合理展开成方案。）") lines.append("") - lines.append("## 总控最终结论") + lines.append("# 三、总控最终共识结论") lines.append((consensus_text or "").strip() or "（未提供）") + lines.append("") + lines.append( + "请严格遵循你的职责设定：自包含离线（纯 inline SVG 图表、禁止任何外链 CDN）、研讨叙事型结构" + "（结论+KPI → 背景目标 → 各席位观点 → 综合分析 → 最终共识方案 → 执行路线）、" + "文字为主图表点睛、浅红主题（主色 #c25555）；写完文件后只用一两句话总结，不要把 HTML 源码贴进对话。" + ) + lines.append( + '硬性要求：① 全程使用简体中文；② HTML 必须是完整文档， 最前面声明 。' + ) return "\n".join(lines) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/progress.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/progress.py index 1ecf06c7..ad6ba229 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/progress.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/progress.py @@ -10,12 +10,21 @@ from __future__ import annotations import logging +from datetime import datetime, timedelta, timezone from typing import Any, Awaitable, Callable from .types import SeatRef logger = logging.getLogger(__name__) +# 北京时区（UTC+8，固定无 DST）——对话气泡时间戳用，对齐前端 nowTime() +# = new Date().toLocaleTimeString("zh-CN", {hour12:false}) 的 "HH:MM:SS" 形状。 +_BEIJING_TZ = timezone(timedelta(hours=8)) + + +def _now_hms() -> str: + return datetime.now(_BEIJING_TZ).strftime("%H:%M:%S") + # 节点状态机：pending → active → done。 _PENDING = "pending" _ACTIVE = "active" @@ -42,11 +51,15 @@ class JobProgressWriter: job_id: str, agents: list[SeatRef], event_sink: EventSink = None, + on_dialogue: Any | None = None, ) -> None: self._store = store self._job_id = job_id self._agents = list(agents) self._event_sink = event_sink + # 每轮 leader/seat 模型返回后回调（dialogues, consensus）—— 执行器据此把对话 + # 写进草稿（和手动跑同一存取路径）。 + self._on_dialogue = on_dialogue self._state: dict[str, str] = {a.agent_id: _PENDING for a in agents} self._spoke_cycle: dict[str, int] = {} self._consensus = 0 @@ -147,20 +160,75 @@ class JobProgressWriter: # ── 对话记录 ───────────────────────────────────────────────────────────── - def _add(self, role: str, agent_id: str | None, name: str, content: str, cycle: int) -> None: + def _add( + self, + role: str, + agent_id: str | None, + name: str, + content: str, + cycle: int, + *, + tool_calls: list[dict] | None = None, + thread_id: str | None = None, + ) -> None: self._dlg_seq += 1 self._dialogues.append( - {"id": self._dlg_seq, "role": role, "agentId": agent_id, "name": name, "content": content, "cycle": cycle} + { + "id": self._dlg_seq, + "role": role, + "agentId": agent_id, + "name": name, + "content": content, + "cycle": cycle, + # 工具步骤卡数据源（每项 {name, args, id}）；映射成前端 Step2Dialogue.steps。 + "toolCalls": list(tool_calls or []), + # 该气泡对应的 leader/seat 线程——映射成 agentThreadId，供点开产物 / 反查 stub。 + "threadId": thread_id, + # 气泡时间戳，**创建时一次性盖章**（不随后续 flush 抖动），对齐前端 nowTime()。 + "time": _now_hms(), + } ) - async def add_leader_dialogue(self, content: str, cycle: int) -> None: - """总控一轮发言（派活说明 / 共识结论）。空内容跳过。""" + def dialogues(self) -> list[dict[str, Any]]: + return list(self._dialogues) + + async def _flush_dialogue(self) -> None: + if self._on_dialogue is None: + return + try: + await self._on_dialogue(self.dialogues(), self._consensus) + except Exception: # noqa: BLE001 — 写草稿失败不打断编排 + logger.warning("roundtable job dialogue → draft flush failed", exc_info=True) + + async def add_leader_dialogue( + self, + content: str, + cycle: int, + *, + tool_calls: list[dict] | None = None, + thread_id: str | None = None, + ) -> None: + """总控一轮发言（派活说明 / 共识结论）。空内容跳过。 + + ``tool_calls`` / ``thread_id`` 用于重建工具步骤卡（派活 chip）+ agentThreadId。 + """ if not (content or "").strip(): return - self._add("leader", None, "总控协调", content, cycle) + self._add("leader", None, "总控协调", content, cycle, tool_calls=tool_calls, thread_id=thread_id) await self._persist() + await self._flush_dialogue() - async def add_seat_dialogue(self, agent_id: str, name: str, content: str, cycle: int) -> None: - """某席位一轮交付。""" - self._add("seat", agent_id, name, content or "", cycle) + async def add_seat_dialogue( + self, + agent_id: str, + name: str, + content: str, + cycle: int, + *, + tool_calls: list[dict] | None = None, + thread_id: str | None = None, + ) -> None: + """某席位一轮交付。``tool_calls`` / ``thread_id`` 用于步骤卡 + 产物预览。""" + self._add("seat", agent_id, name, content or "", cycle, tool_calls=tool_calls, thread_id=thread_id) await self._persist() + await self._flush_dialogue() diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py index 6c011f7a..3525493f 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py @@ -52,6 +52,9 @@ class LeaderTurn: dispatched: list[tuple[str, str]] = field(default_factory=list) # 命中澄清时的问题文本；非 None 即 clarification 分支。 clarification: str | None = None + # 本轮总控调用的工具（按出现顺序，每项 {name, args, id}）——通常是 agent_orchestration + # 派活 / ask_clarification。用于在草稿 run 里重建「工具步骤卡」，与前端手动跑一致。 + tool_calls: list[dict] = field(default_factory=list) @property def is_clarification(self) -> bool: @@ -67,6 +70,9 @@ class SeatTurn: """某席位一轮交付的归一化结果（由 gateway.run_seat 产出）。""" final_text: str = "" + # 本轮席位调用的工具（按出现顺序，每项 {name, args, id}）——用于在草稿 run 里 + # 重建「工具步骤卡」+ write_file stub（点开产物），与前端手动跑一致。 + tool_calls: list[dict] = field(default_factory=list) @dataclass