This commit is contained in:
董宏凯 committed 2026-06-05 11:20:22 +08:00
commit 9c9330114f
14 files changed
+846 -136

No files matched your search

@@ -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-<id>` 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>", jobId, jobStatus, step2RoundtableDialogues:[...], step3:{...}} ]`。
4. **前端验证**:后台挂起 → 等跑完 → 从历史加载该草稿 → Step2 应显示后台研讨完整对话(走现有 `normalizeStep2Runs`/`hydrateStep2Run`)。
---
## 三、遗留问题(下次重点,已按优先级排)
### P0 —— 报告 agent 不写 HTML 文件(结果绘制拿不到产物)
- **现象**:后台报告 run 跑完,日志 `report html: no .html found in ...threads/<id>/user-data/outputs`;
`run.step3.html` 回退成报告 agent 的**纯文字**(不含 `<html>`),第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。
+85 -3
View File
@@ -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",
+1
View File
@@ -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",
+7 -3
View File
@@ -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";
@@ -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 (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="flex max-h-[88vh] flex-col gap-0 sm:max-w-5xl">
<DialogContent className="flex max-h-[90vh] w-[94vw] flex-col gap-0 sm:max-w-[1240px]">
<DialogHeader className="shrink-0">
<DialogTitle className="flex items-center gap-2">
{status === "running" && (
@@ -223,9 +224,9 @@ export function DispatchChainModal({
<div key={d.id} className="flex gap-2.5">
<div className="shrink-0">{getAvatar(avatarForDialogue(d), "h-7 w-7")}</div>
<div className="min-w-0 flex-1">
<div className="text-xs font-semibold text-foreground/80">{d.name}</div>
<div className="mt-0.5 whitespace-pre-wrap break-words text-sm leading-relaxed text-foreground/70">
{d.content || "…"}
<div className="mb-0.5 text-xs font-semibold text-foreground/80">{d.name}</div>
<div className="prose prose-sm max-w-none break-words text-sm leading-relaxed text-foreground/80 dark:prose-invert">
<MarkdownContent>{d.content || "…"}</MarkdownContent>
</div>
</div>
</div>
@@ -289,10 +290,15 @@ export function DispatchChainModal({
停止
</Button>
)}
<Button onClick={onView} disabled={!onView} className="gap-1.5">
<EyeIcon className="size-4" />
{status === "done" ? "查看报告" : "查看"}
</Button>
{/* 只对「运行中」任务隐藏入口——它正在后台真跑,进入会暂停前端编排、
与后台作业双跑或借此重跑。其它状态(排队 / 待澄清 / 已完成 / 已取消 /
出错)都不是正在跑的后台作业,可正常查看。 */}
{status !== "running" && (
<Button onClick={onView} disabled={!onView} className="gap-1.5">
<EyeIcon className="size-4" />
{status === "done" ? "查看报告" : "查看"}
</Button>
)}
</div>
</DialogFooter>
</DialogContent>
@@ -1690,32 +1690,39 @@ function IntentSummaryCard({
{intentReady ? (
<>
<ul className="text-xs text-slate-700 space-y-2.5 font-normal">
<li className="flex items-baseline gap-2 leading-relaxed">
<span className="text-primary text-base leading-none">•</span>
<span>
<li className="flex items-start gap-2 leading-relaxed">
<span className="text-primary text-base leading-none pt-0.5">•</span>
<span
className={cn(
"min-w-0 leading-relaxed",
collapsible && !expanded && "line-clamp-3",
)}
>
<b className="text-slate-900 font-bold">目标</b>: {intentReady.objective}
</span>
</li>
</ul>
{collapsible && (
<button
type="button"
onClick={() => setExpanded((v) => !v)}
className="flex items-center gap-0.5 text-[10px] font-bold text-slate-500 hover:text-primary transition-colors"
aria-expanded={expanded}
>
{expanded ? (
<>
收起
<ChevronUp className="w-3.5 h-3.5" />
</>
) : (
<>
展开
<ChevronDown className="w-3.5 h-3.5" />
</>
)}
</button>
<div className="flex justify-end">
<button
type="button"
onClick={() => setExpanded((v) => !v)}
className="inline-flex items-center gap-0.5 text-xs font-bold text-slate-500 hover:text-primary transition-colors"
aria-expanded={expanded}
>
{expanded ? (
<>
收起
<ChevronUp className="w-3.5 h-3.5" />
</>
) : (
<>
展开
<ChevronDown className="w-3.5 h-3.5" />
</>
)}
</button>
</div>
)}
{showFullDetail && (
<ul className="text-xs text-slate-700 space-y-2.5 font-normal">
@@ -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)
@@ -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:<toolName>``——前端 ``getStepDisplay`` 据此推图标 + 文案;
- ``id`` = ``<dialogueId>-<i>``——对齐前端 ``${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-<tid>``、
工具调用 ``tc-<tid>-<i>``),使加载草稿后点开产物的虚拟 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-<jobId>`` 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)
@@ -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 [])]))
@@ -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(
@@ -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)
@@ -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 必须是完整文档,<head> 最前面声明 <meta charset="utf-8">。'
)
return "\n".join(lines)
@@ -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()
@@ -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