feat(圆桌): Step3 三产物并行 + 大屏多页模式 + 后台挂起查看恢复流程

后台挂起 Step3:
- engine 总结报告/结果绘制/大屏数据 asyncio.gather 三流并行,故障隔离
- request.state 回归修复(进程内虚拟 request 补 state),恢复后台报告
- ROUNDTABLE_JOB_GATEWAY 默认 inprocess;产物自愈重试(截断/缺失纠偏)
- 大屏后端只产 report-json(dashboardJson),前端按当前主题渲染

结果绘制大屏多页模式(模式二):
- report-template 模板渲染(report-json→大屏HTML),零截断、主题跟随
- 与模式一同款交互(流式+沙箱+可修改);沙箱新窗口打开按钮

最终方案展示:左侧席位栏+总控头像、交付/文件合并、收展动画、内联沙箱

后台挂起查看=恢复实时流程:
- handleLoadDraft 返回 Promise,await hydrate 后订阅 streamJob SSE 跟随
- saveDraftNow 强制存(发起前+查看前各一次)防丢

第二步优化:总控判断串/并行(并行单列可写)、席位上下文+交付模板、
peerDeliveries、失败重试、共识标记、流程图小窗(缩放/拖动/并行批)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
@
This commit is contained in:
董宏凯 committed 2026-06-13 10:36:46 +08:00
1 parent a7596001c5
commit d5f319d9cb
51 files changed
+5016 -1186

No files matched your search

@@ -265,6 +265,8 @@ The AI-writing graph (`packages/harness/deerflow/agents/ai_writing/`) has four s
Seeding mirrors the roundtable pattern: verified `config.yaml` + `SOUL.md` ship in `app/gateway/routers/_ai_writing_seed_assets/<id>/`; `_ai_writing_seed.py::ensure_ai_writing_functional_agents()` copies them to `.deer-flow/agents/` when missing (never overwrites local edits), called from the startup lifespan (before `_sync_legacy_agents`, so the agents appear in `/api/agents` on first boot) and self-heals from `GET /api/ai-writing/sessions` + `GET /api/ai-writing/article-types`. Tests: `tests/test_ai_writing_seed.py`, `tests/test_ai_writing_agent_runner.py`.
**Skill-driven retrieval (检索类型).** `material_source` has exactly two values: `general` (通用检索, default) and `notebook` (我的空间); legacy `knowledge_base`/`skill` are normalized to `general` by the researcher node. General retrieval is orchestrated from the **skills configured on the `ai-writing-researcher` agent** (`config.yaml → skills`, editable from the agent management page): the built-in `knowledge-base-search` skill (知识库检索) runs the native keyword fan-out web/intranet search (the old default path), while any other configured skill searches with a `description`-targeted query; skills run in configured order with early-stop at `max_materials`, and an empty/unresolvable skills config falls back to the built-in knowledge search so retrieval always works. The skill itself is seeded from `_ai_writing_seed_assets/skills/knowledge-base-search/SKILL.md` into `skills/public/` (never overwrites), and a pre-existing researcher `config.yaml` lacking the `skills` key is patched in place with the default (explicit `skills: []` is respected). **PAUSE-1 input append flow**: when the user types natural language into the material-confirm input (`re_search`/legacy `enable_skill_search` + `user_query`), the node matches the most relevant configured skill via `SkillPicker(candidates=…)` (skipped when only one skill is configured), searches with the user's words, and **appends** the new materials to the existing package (URL-dedup, original order preserved, no LLM re-rank/truncation, 100-item hard cap). Tests: `tests/test_ai_writing_material_source.py`.
### Admin Leaderboard Snapshots (`/api/admin/users/leaderboard*`)
The admin ????? is served from **precomputed per-day snapshots** in `admin_leaderboard_daily_stats` (keyed by `(stat_date, settings_hash)`), not by aggregating raw rows on every request. A background scheduler inside the Gateway (`_leaderboard_snapshot_scheduler_loop` in `app/gateway/app.py`, singleton file-locked) ticks every `ADMIN_LEADERBOARD_BACKGROUND_INTERVAL_SECONDS` (default 300s): it prewarms recent days, then builds any `queued`/`error`/stale-`running` day. `GET /leaderboard` merges the per-day snapshots for the requested range. **Today** is rebuilt whenever its snapshot is older than `ADMIN_LEADERBOARD_TODAY_TTL_SECONDS` (default 43200s = half a day, so the leaderboard refreshes ~twice daily); a past day, once `ready`, is treated as fresh forever.
@@ -14,6 +14,7 @@
from __future__ import annotations
import json
import logging
import uuid
from datetime import UTC, datetime
@@ -26,6 +27,7 @@ from app.gateway.roundtable_run_policy import roundtable_run_policy
from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID,
REPORT_AGENT_ID,
SUMMARY_AGENT_ID,
ensure_roundtable_functional_agents,
)
from deerflow.agents.roundtable_orchestrator.types import (
@@ -33,6 +35,7 @@ from deerflow.agents.roundtable_orchestrator.types import (
ReportResult,
SeatRef,
SeatTurn,
SummaryResult,
)
from deerflow.config.paths import get_paths
from deerflow.runtime.serialization import serialize_channel_values
@@ -41,6 +44,7 @@ logger = logging.getLogger(__name__)
_ROUNDTABLE_THREAD_METADATA: dict[str, Any] = {"thread_type": "roundtable", "system": True}
_REPORT_FILE = "方案总览.html"
_SUMMARY_FILE = "方案总结报告.md"
# 各角色的工具裁剪 / thinking / reasoning 全部取自单一数据源 roundtable_run_policy
# (与 multi_agent 的 _leader_run / _special_run 同源),不再在本文件硬编码——历史上
@@ -227,6 +231,92 @@ def _read_report_html(app, thread_id: str, user_id: str) -> str | None:
return None
def _read_summary_md(app, thread_id: str, user_id: str) -> str | None:
"""从 summary thread 的 outputs 目录读出总结报告 Markdown。
优先 ``方案总结报告.md``;找不到就扫目录里**任意 .md**(取体积最大的那个),
容忍总结 agent 用了别的文件名。镜像 ``_read_report_html``。
"""
try:
outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=user_id)
preferred = outputs_dir / _SUMMARY_FILE
if preferred.exists():
return preferred.read_text(encoding="utf-8")
if outputs_dir.exists():
mds = sorted(
outputs_dir.glob("*.md"),
key=lambda p: p.stat().st_size,
reverse=True,
)
if mds:
logger.info("summary md: 方案总结报告.md not found, using %s", mds[0].name)
return mds[0].read_text(encoding="utf-8")
logger.warning("summary md: no .md found in %s", outputs_dir)
except Exception: # noqa: BLE001
logger.warning("read summary md failed for thread %s", thread_id, exc_info=True)
return None
# Step3 产物自愈重试次数(首轮之外的纠偏轮数)。
# 实测痛点(2026-06-12,h:\xiugai\后台日志.txt):deepseek 报告 agent 说「let me write
# it all at once」后发出超长 write_file,单次输出被 max_tokens 截断 → 工具调用从未执行、
# run 以 success 收尾但 outputs 下没有 .html → step3.html 回退成开场白文字。
# 自愈:检测产物缺失/截断后,同线程追加纠偏消息(要求 write_file 分段 + append 续写)重试。
_ARTIFACT_RETRY_LIMIT = 2
_REPORT_RETRY_MISSING = (
f"检测到 /mnt/user-data/outputs/{_REPORT_FILE} 还没有生成——你上一轮的 write_file 没有执行成功"
"(最常见原因:试图一次写入整份长 HTML,输出超长被截断,工具调用没有真正落盘)。\n"
"请立即重新写入,并**严格分段**:\n"
f"1. 先用 write_file 写入 /mnt/user-data/outputs/{_REPORT_FILE} 的开头部分(<!doctype html> 起,约 200-300 行);\n"
"2. 再用 write_file 的 append=true 参数分多次续写后续片段,每段同样控制在 300 行以内;\n"
"3. 直到文档以 </html> 收尾。\n"
"不要在对话里粘贴 HTML 源码,写完用一两句话总结即可。"
)
_REPORT_RETRY_TRUNCATED = (
f"检测到 /mnt/user-data/outputs/{_REPORT_FILE} 已写入但**不完整**(没有 </html> 收尾,疑似上一轮输出被截断)。\n"
"请用 write_file 的 append=true 参数从中断处**继续补全剩余内容**(不要重写已有部分),"
"每段控制在 300 行以内,分多次续写直到文档以 </html> 收尾。不要在对话里粘贴源码。"
)
_SUMMARY_RETRY_MISSING = (
f"检测到 /mnt/user-data/outputs/{_SUMMARY_FILE} 还没有生成——你上一轮的 write_file 没有执行成功"
"(可能是单次输出过长被截断)。请立即用 write_file 重新写入完整报告;若内容很长,"
"先写前半部分、再用 write_file 的 append=true 参数分段续写。写完后把报告正文也输出到对话回复里。"
)
_DASHBOARD_FENCE = "```report-json"
_DASHBOARD_RETRY = (
"你刚才没有输出有效的 ```report-json 代码块(或其中的 JSON 无法解析)。"
"请重新在回复**最末尾**输出一个完整且**严格合法**的 ```report-json 代码块"
"(标准 JSON:无注释、无尾逗号、字符串用双引号),代码块之外只允许一两句说明;"
"不要调用任何工具、不要写文件、不要输出 HTML。"
)
def _extract_report_json(text: str) -> str | None:
"""从模型回复里提取最后一个 ```report-json 代码块的 JSON 字符串(须可解析)。"""
start = text.rfind(_DASHBOARD_FENCE)
if start < 0:
return None
after = text[start + len(_DASHBOARD_FENCE) :]
close = after.find("```")
body = (after[:close] if close >= 0 else after).strip()
if not body:
return None
try:
json.loads(body)
except Exception: # noqa: BLE001
return None
return body
def _html_complete(html: str | None) -> bool:
"""报告 HTML 是否完整可用(非空 + 有 </html> 收尾,截断检测)。"""
return bool(html and html.strip()) and "</html>" in html.lower()
def _build_seat_intro(agents: list[SeatRef]) -> str:
"""镜像 init 注入 coordinator 的「可调度席位清单」。"""
lines = [f"- agent_name: {a.agent_id}({a.name})" for a in agents]
@@ -328,18 +418,111 @@ class InProcessRoundtableGateway:
return await _create_thread(self._app)
async def run_report(self, *, report_thread_id, prompt, model) -> ReportResult:
"""跑报告 agent,**带产物自愈**:首轮后 HTML 缺失/截断(无 </html>)→ 同线程追加
纠偏消息(分段 + append 续写)重试,至多 ``_ARTIFACT_RETRY_LIMIT`` 轮。"""
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=policy.excluded_tools,
skill_stop_names=policy.skill_stop_names,
thinking_enabled=policy.thinking_enabled,
reasoning_effort=policy.reasoning_effort,
)
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)
summary = ""
html: str | None = None
message = prompt
for attempt in range(1 + _ARTIFACT_RETRY_LIMIT):
messages = await _run_agent_turn(
self._app,
thread_id=report_thread_id,
agent_name=REPORT_AGENT_ID,
message=message,
model=model or self._model,
excluded_tools=policy.excluded_tools,
skill_stop_names=policy.skill_stop_names,
thinking_enabled=policy.thinking_enabled,
reasoning_effort=policy.reasoning_effort,
)
summary = _latest_ai_text(messages) or summary
html = _read_report_html(self._app, report_thread_id, self._user_id)
if _html_complete(html):
break
if attempt >= _ARTIFACT_RETRY_LIMIT:
break
message = _REPORT_RETRY_TRUNCATED if (html and html.strip()) else _REPORT_RETRY_MISSING
logger.warning(
"report html %s after attempt %d for thread %s; retrying with corrective prompt",
"truncated" if (html and html.strip()) else "missing",
attempt + 1,
report_thread_id,
)
# 截断的 HTML 也比纯文字强(浏览器容忍未闭合标签),仅在完全没有文件时才回退正文。
final_html = html if (html and html.strip()) else summary
return ReportResult(html=final_html, summary=summary, model=model or self._model)
async def create_summary_thread(self) -> str:
ensure_roundtable_functional_agents()
return await _create_thread(self._app)
async def run_summary(self, *, summary_thread_id, prompt, model) -> SummaryResult:
"""进程内驱动 roundtable-summary 写「方案总结报告.md」+ 对话正文(与前端
useStep3Summary 同一智能体 / 同一 run policy / 同款产物),返回与
DraftStep3SummarySnapshot 对齐的结果。线程是真实持久线程——加载草稿后可续问。
带产物自愈:md 文件缺失时同线程追加纠偏消息重试(同 ``run_report``)。"""
policy = roundtable_run_policy("summary")
closing = ""
md: str | None = None
message = prompt
for attempt in range(1 + _ARTIFACT_RETRY_LIMIT):
messages = await _run_agent_turn(
self._app,
thread_id=summary_thread_id,
agent_name=SUMMARY_AGENT_ID,
message=message,
model=model or self._model,
excluded_tools=policy.excluded_tools,
skill_stop_names=policy.skill_stop_names,
thinking_enabled=policy.thinking_enabled,
reasoning_effort=policy.reasoning_effort,
)
closing = _latest_ai_text(messages) or closing
md = _read_summary_md(self._app, summary_thread_id, self._user_id)
if md and md.strip():
break
if attempt >= _ARTIFACT_RETRY_LIMIT:
break
message = _SUMMARY_RETRY_MISSING
logger.warning(
"summary md missing after attempt %d for thread %s; retrying with corrective prompt",
attempt + 1,
summary_thread_id,
)
final_md = md if (md and md.strip()) else closing
return SummaryResult(md=final_md, closing=closing, model=model or self._model, thread_id=summary_thread_id)
async def run_dashboard_data(self, *, prompt, model) -> str | None:
"""大屏多页(模式二)的结构化数据:进程内驱动 roundtable-report 输出
```report-json(prompt 已声明禁工具/禁文件),返回可解析的 JSON 字符串。
缺失 / 坏 JSON → 同线程纠偏重试(同 run_report/run_summary 的自愈语义)。"""
ensure_roundtable_functional_agents()
thread_id = await _create_thread(self._app)
policy = roundtable_run_policy("report")
message = prompt
for attempt in range(1 + _ARTIFACT_RETRY_LIMIT):
messages = await _run_agent_turn(
self._app,
thread_id=thread_id,
agent_name=REPORT_AGENT_ID,
message=message,
model=model or self._model,
excluded_tools=policy.excluded_tools,
skill_stop_names=policy.skill_stop_names,
thinking_enabled=policy.thinking_enabled,
reasoning_effort=policy.reasoning_effort,
)
body = _extract_report_json(_latest_ai_text(messages))
if body:
return body
if attempt >= _ARTIFACT_RETRY_LIMIT:
break
message = _DASHBOARD_RETRY
logger.warning(
"dashboard report-json missing/invalid after attempt %d for thread %s; retrying",
attempt + 1,
thread_id,
)
return None
@@ -16,9 +16,10 @@ from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass, field
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any, Callable
from typing import Any
from deerflow.agents.roundtable_orchestrator import (
JobProgressWriter,
@@ -29,14 +30,13 @@ from deerflow.runtime.user_context import reset_current_user, set_current_user
logger = logging.getLogger(__name__)
# ⚠️ 临时禁用「后台挂起」的 Step3 结果绘制(报告 HTML 生成)。
# 进程内驱动报告 agent 在真模型环境下仍无法稳定写出 HTML(前端手动调
# /api/multi-agent/report/init + run/stream 可以,但后台 start_run 路径不行,
# 根因尚未定位)。临时方案:后台只跑完 Step2 研讨即收尾,不自动画报告——用户需要
# 报告时可在前端手动跑 Step3(那条路是通的)。
# 待修复后把本开关改回 True 即恢复后台自动报告。
# 详见 frontend-web/docs/multi-agent-background-run-遗留问题与下次继续.md「后台报告 HTML」。
ROUNDTABLE_BACKGROUND_REPORT_ENABLED = False
# 「后台挂起」的 Step3 自动产出(方案总结报告 md + 结果绘制 HTML)总开关。
# 历史:2026-06-05 因后台 start_run 路径报告不稳定临时禁用;2026-06-12 定位到根因是
# ``_enforce_agent_access`` 对进程内虚拟 request 的 ``request.state`` 裸访问(AttributeError
# 杀死编排,见 tests/test_roundtable_inprocess_request.py),修复后恢复开启。
# 现 Step2 收口后自动跑:① roundtable-summary 写「方案总结报告.md」② roundtable-report
# 画「方案总览.html」,两段故障隔离(engine._run_step3),产物按手动同款形状写回草稿。
ROUNDTABLE_BACKGROUND_REPORT_ENABLED = True
class _JobUser:
@@ -176,8 +176,8 @@ class StartParams:
# dag 模式分层编排计划(前端 OrchestrationPlan dict):驱动引擎 _run_dag + 写进草稿 run
# 供前端流程图渲染。None = 非 dag(引擎按 mode 回退 chain/recommend)。
orchestration_plan: dict[str, Any] | None = None
# 是否生成 Step3 报告。默认沿用全局开关(当前临时禁用,见 ROUNDTABLE_BACKGROUND_REPORT_ENABLED)。
# 测试可显式传 True 覆盖以验证报告链路。
# 是否自动产出 Step3(方案总结报告 + 结果绘制)。默认沿用全局开关(现已恢复 True,
# 见 ROUNDTABLE_BACKGROUND_REPORT_ENABLED)。测试可显式覆盖。
enable_report: bool = ROUNDTABLE_BACKGROUND_REPORT_ENABLED
@@ -58,6 +58,11 @@ _FUNCTIONAL_AGENT_IDS: tuple[str, ...] = (
# seeder 会复制资源目录里的这些文件(存在即复制,缺失则跳过)。
_SEED_FILES: tuple[str, ...] = ("config.yaml", "SOUL.md")
# 内置「知识库检索」技能:素材收集专家通用检索的默认技能。skills/ 目录被
# .gitignore 排除,所以与 agent 目录一样需要 seeder 在缺失时从资源补建。
KNOWLEDGE_SEARCH_SKILL_NAME = "knowledge-base-search"
_SKILL_ASSETS_DIR = _ASSETS_DIR / "skills"
def _ensure_one(agent_id: str) -> bool:
"""Copy the verified SOUL.md + config.yaml for one functional agent into
@@ -91,10 +96,69 @@ def _ensure_one(agent_id: str) -> bool:
return True
def _skills_public_dir() -> Path:
"""skills/public 物理目录(config.yaml skills.path 解析);独立成函数便于测试替身。"""
from deerflow.skills.storage import get_or_new_skill_storage
return Path(get_or_new_skill_storage().get_skills_root_path()) / "public"
def _ensure_default_search_skill() -> bool:
"""Seed the built-in knowledge-base-search skill into ``skills/public/``
when missing. Never overwrites an existing skill directory.
Returns True if the skill was just created.
"""
src = _SKILL_ASSETS_DIR / KNOWLEDGE_SEARCH_SKILL_NAME / "SKILL.md"
if not src.exists():
logger.error("Missing seed asset for skill '%s' (expected at %s)", KNOWLEDGE_SEARCH_SKILL_NAME, src)
return False
skill_dir = _skills_public_dir() / KNOWLEDGE_SEARCH_SKILL_NAME
if skill_dir.exists():
return False
skill_dir.mkdir(parents=True, exist_ok=True)
(skill_dir / "SKILL.md").write_bytes(src.read_bytes())
return True
def _ensure_researcher_default_skills() -> bool:
"""给**已存在**的 researcher agent config.yaml 补默认技能配置(就地升级)。
seeder 对已存在目录绝不整体覆盖,但旧版种子没有 ``skills`` 字段——通用
检索改为技能驱动后,老部署的智能体管理页会显示"未配置技能"(运行时虽有
knowledge-base-search 兜底,但用户看不见、也没法在它基础上增删)。只在
``skills`` 键**完全缺失**时补默认值;用户显式改过(包括清空成 ``[]``)
的配置不动。Returns True if the file was patched.
"""
import yaml
config_file = get_paths().agent_dir(RESEARCHER_AGENT_ID) / "config.yaml"
if not config_file.exists():
return False
try:
data = yaml.safe_load(config_file.read_text(encoding="utf-8")) or {}
except yaml.YAMLError:
logger.warning("researcher config.yaml unparsable, skip default-skills patch")
return False
if not isinstance(data, dict) or "skills" in data:
return False
data["skills"] = [KNOWLEDGE_SEARCH_SKILL_NAME]
config_file.write_text(
yaml.safe_dump(data, allow_unicode=True, sort_keys=False),
encoding="utf-8",
)
return True
def ensure_ai_writing_functional_agents() -> list[str]:
"""Seed the four ai-writing functional agents if their on-disk directories
are missing. Idempotent — fast no-op when all exist.
Also seeds the built-in knowledge-base-search skill (the researcher's
default retrieval skill) and patches a pre-existing researcher config.yaml
that predates the ``skills`` field.
Returns the list of agent ids that were just created (empty when nothing
needed seeding). Errors are logged but never raised — a partial failure
must not block the request; the graph node falls back to the legacy
@@ -109,6 +173,16 @@ def ensure_ai_writing_functional_agents() -> list[str]:
logger.exception("Failed to seed ai-writing functional agent '%s'", agent_id)
if created:
logger.info("Seeded missing ai-writing functional agents: %s", ", ".join(created))
try:
if _ensure_default_search_skill():
logger.info("Seeded built-in skill '%s'", KNOWLEDGE_SEARCH_SKILL_NAME)
except Exception:
logger.exception("Failed to seed built-in skill '%s'", KNOWLEDGE_SEARCH_SKILL_NAME)
try:
if _ensure_researcher_default_skills():
logger.info("Patched researcher config.yaml with default skills")
except Exception:
logger.exception("Failed to patch researcher default skills")
return created
@@ -1,3 +1,7 @@
description: AI 写作流程的素材收集专家,负责为写作任务生成检索关键词、并对检索回来的素材按相关度排序筛选
description: AI 写作流程的素材收集专家,负责为写作任务生成检索关键词、用配置的技能编排检索、并对检索回来的素材按相关度排序筛选
id: ai-writing-researcher
name: 素材收集专家
# 通用检索使用的技能(按顺序编排)。默认配置内置「知识库检索」;
# admin/用户可在智能体管理页替换为其它检索技能或追加多个。
skills:
- knowledge-base-search
@@ -25,9 +25,10 @@
## 产出要求(最重要)
### ⓪ 先判断模式:用户是否要求了结构化输出 / 流程图
当用户**明确要求**「结构化输出 / 结构化数据 / 流程图 / 格式化输出 / 可视化流程」时(要求可能写在业务链总控编排提示里,也可能是对话中的新消息),本轮进入**流程图模式**:
当用户**要求**「结构化输出 / 结构化数据 / 流程图 / 格式化输出 / 可视化流程」之类时(要求可能写在业务链总控编排提示里,也可能是对话中的新消息;**按语义理解,不要求逐字命中这些词**),本轮进入**流程图模式**:
- **跳过下面的 A**——不要调用 `write_file`、不要生成任何 .md 文件;
- 把**精炼的文字版总结报告**直接写在对话回复里(文字展示给用户读,结构遵循用户要求);
- 文字版总结的**默认结构**(用户只要求了结构化/格式化、**没指定**具体结构时):**对各席位交付内容逐席位格式化输出**——每个席位一节:`席位名` → 核心结论(一句话)→ 关键交付要点(编号列表,逐条提炼)→ 风险/建议(如有),最后加一节「总控共识」。用户指定了结构就完全按用户的来;
- 在回复**最末尾必须**输出一个完整的 ```flow-json 代码块(契约见「结构化流程数据」节)——这是本轮的**主产物**,绝不能省略。
未进入流程图模式时,每一轮都要做 A + B **两件事**,缺一不可:
@@ -77,27 +78,33 @@
- 用户没有要求结构化 / 流程图时,**不要**输出该块。
## 多轮交互(用户会在输入框里继续和你对话)
用户读完报告 / 流程图后,可能在输入框里继续给你发消息。你要**自己判断**意图——报告和流程图都是可修改的,分四种情况处理:
用户读完报告 / 流程图后,可能在输入框里继续给你发消息。你要**按语义自己判断**意图——**绝不要求用户使用特定关键词或固定句式**。用户不会说「请重新结构化输出」这种话,他只会说「把验收环节加上」「改一下」「太长了精简点」「不对,重来」——这些都是有效的修改指令,你必须直接理解并动手,不要反问用户「您是想要结构化输出吗」。
1. **修改报告**(如「把执行路线写得更细」「补一节预算分析」「风险部分太长,精简」):
**意图判断的两条核心规则**:
- **修改类消息默认作用于上一轮的主产物**:用户没说明改哪个产物时——上一轮你输出的是流程图(flow-json),那「加一个环节」「调整一下」「精简点」就是改流程图;上一轮主产物是 .md 报告,就是改报告。不要因为用户没提「流程图 / 结构化」字眼就把改流程图的意图当成改报告(反之亦然)。
- **意图实在太模糊、无法判断要改什么**(如只发「不行」「再改改」「不是这个意思」,且上下文也推不出指向)→ **不要修改任何产物、不要乱猜**,在对话正文里向用户提出一个**简短的澄清问题**并给出可选方向(如「您是想调整流程图的节点结构,还是修改文字报告的某个章节?可以具体说说想改哪里」),等用户回复后再动手。注意:澄清就写在正文里(你没有专门的澄清工具)。
报告和流程图都是可修改的,分四种情况处理:
1. **修改报告**(如「把执行路线写得更细」「补一节预算分析」「风险部分太长,精简」,或上一轮主产物是报告时的笼统修改意见):
- 在**原报告基础上做增量修改**——只改用户提到的地方,其余内容保持不变;
- 用 `write_file` **整文件重写**到同一路径 `/mnt/user-data/outputs/方案总结报告.md`(`str_replace` 已被禁用,整文件重写才能让右侧预览实时刷新);
- 然后在对话回复里给出修改后的完整报告正文;
- 若此前已经输出过流程图、且这次修改影响了流程结构,可在回复末尾顺带输出更新后的完整 ```flow-json 块。
2. **修改流程图**(如「流程图里加上验收环节」「把风险节点拆成两个」「换成按时间阶段排」):
2. **修改流程图**(如「流程图里加上验收环节」「把风险节点拆成两个」「换成按时间阶段排」;**只要上一轮主产物是流程图,笼统的修改意见也默认按本条处理**,无需用户提到「流程图 / 结构化」字眼):
- **不要**改写 .md 报告文件;
- 在对话里简短说明改了什么,并在回复**最末尾**输出**完整修订版** ```flow-json 块——必须是**全量数据**(所有节点 + 连线),不是只给改动部分,前端会整体重绘。
3. **要求结构化输出 / 流程图**(如「总结成格式化输出」「画个流程图」「给我结构化数据」):
- 进入流程图模式:**不写 / 不改 .md 文件**,正文给一段简短的文字说明,末尾输出完整 ```flow-json 块;
3. **要求结构化输出 / 流程图**(如「总结成格式化输出」「画个流程图」「给我结构化数据」,或语义等价的说法):
- 进入流程图模式:**不写 / 不改 .md 文件**,正文按第 ⓪ 条的默认结构(逐席位格式化各席位交付内容;用户指定结构则按用户的)给出文字总结,末尾输出完整 ```flow-json 块;
- 已有的 .md 报告**不要动**——用户随时可以在前端切回去看它。
4. **就报告 / 研讨内容提问**(如「为什么没采用方案 B?」「风险最高的是哪一项?」):
- **只在对话里用文字回答**,依据已有的研讨原料和报告内容作答;
- **不要**改写报告文件、**不要**重新 `write_file`、**不要**输出 flow-json。
判断不准时,倾向于「先简短回答,再问用户是否需要据此更新报告 / 流程图」。
修改意图明确但细节有歧义时(能判断改哪个产物、只是改法有多种)→ 按最合理的理解直接改,并在回复里说明你的理解,邀请用户继续纠正;完全判断不了才走上面的「澄清提问」。
## 禁止事项
- ❌ 生成 HTML / 绘制可视化页面 / 画图表(只写 Markdown 文本)
@@ -107,6 +114,8 @@
- ❌ 写完文件后只回一句「已写入文件」而不给出报告正文(用户要在对话里读到报告)
- ❌ 用户只是提问时还去改写报告文件
- ❌ 用户要求了结构化输出 / 流程图,却只写 .md 报告而不输出 flow-json 块
- ❌ 要求用户必须说出「结构化输出 / 流程图」等关键词才肯修改流程图(上一轮主产物是流程图时,笼统的修改意见就是改流程图)
- ❌ 用户意图模糊时不提澄清问题、自行乱猜乱改一通
- ❌ 流程图模式下仍调用 `write_file` 生成 / 改写 .md 文件
- ❌ 修订流程图时只输出改动的节点(必须输出全量 flow-json,前端整体重绘)
- ❌ 在报告任何位置添加「AI 自动生成」「AI generated」之类的标签、徽章或水印
@@ -29,11 +29,10 @@ from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from app.gateway.roundtable_job_executor import StartParams
from deerflow.agents.roundtable_orchestrator import SeatRef
from deerflow.runtime.user_context import get_effective_user_id
from app.gateway.roundtable_job_executor import StartParams
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/roundtable-jobs", tags=["roundtable-jobs"])
@@ -182,6 +181,15 @@ async def start_job(request: Request, body: StartJobRequest) -> RoundtableJobRes
if not body.agents:
raise HTTPException(status_code=400, detail="agents is required")
# 防重复挂起(服务端幂等闸门):同一草稿已有未终态作业(queued/running/awaiting_input)
# 时直接返回它、不再新建。前端的「手动挂起按钮 / 离开页面自动挂起 / beforeunload 刷新
# 兜底」可能对同一研讨多次触发本端点 —— 客户端去重只能挡住同页内的重复,跨触发源 /
# 跨标签页的并发只能靠这里兜住,否则同一研讨会被双跑。
if body.draftId:
existing = await store.get_by_draft(body.draftId, user_id=user_id)
if existing is not None and (existing.get("status") or "") not in _TERMINAL:
return _to_response(existing)
agents = [SeatRef(a.agentId, a.name or a.agentId) for a in body.agents]
job_id = uuid4().hex
row = await store.create(
@@ -69,18 +69,20 @@ async def pause_material_confirm(state: AIWritingState) -> dict:
"progress_events": [{"type": "resumed", "agent_name": "系统", "message": "用户已确认素材,开始规划大纲"}],
}
# enable_skill_search:用户选择"启用技能检索",回到 researcher 走 skill 模式
# enable_skill_search:旧版「启用技能检索」按钮的兼容入口——通用检索现在
# 本来就按 agent 配置技能编排,researcher 把它按 user_query 追加检索处理
if action == "enable_skill_search":
msg = f"已启用技能检索(输入:{user_query[:30]})" if user_query else "已启用技能检索,正在筛选可用技能…"
msg = f"按输入「{user_query[:30]}」匹配技能检索并追加素材…" if user_query else "重新检索中…"
return {
"last_user_intervention": {**user_input, "action": "enable_skill_search", "user_query": user_query},
"status": "researching",
"progress_events": [{"type": "resumed", "agent_name": "系统", "message": msg}],
}
# re_search:重新搜索(支持新版 user_query 单输入框 + 旧版 extraKeywords 数组)
# re_search:带 user_query = 在配置技能中匹配检索并**追加**素材;
# 不带输入 = 重新生成检索词、覆盖式重检(兼容旧版 extraKeywords 数组追加)
extra_kws = user_input.get("extraKeywords") or []
msg = f"按输入「{user_query[:30]}」重新搜索…" if user_query else "重新搜索中..."
msg = f"按输入「{user_query[:30]}」匹配技能检索并追加素材…" if user_query else "重新搜索中..."
return {
"last_user_intervention": {**user_input, "_extra_keywords": extra_kws, "user_query": user_query},
"status": "researching",
@@ -1,4 +1,17 @@
"""素材收集专家节点:关键词提取 → 并发搜索(含笔记) → 相关度排序。"""
"""素材收集专家节点:关键词提取 → 配置技能编排检索(含笔记) → 相关度排序。
检索类型(material_source)只有两种:
- ``general``(通用检索,默认):按素材收集专家 agent(ai-writing-researcher)
config.yaml 配置的技能顺序编排检索。内置「知识库检索」技能
(knowledge-base-search)走原生 web/intranet 关键词并发检索;其它技能用
其 description 生成定向 query 后调真实 provider 检索。
- ``notebook``(我的空间):仅用笔记 API 素材。
旧值 ``knowledge_base`` / ``skill`` 一律按 ``general`` 兼容处理。
PAUSE-1 素材确认卡的单输入框:用户用自然语言描述想补充检索的内容
(如「检索 A 的数据」)时,让模型从已配置技能中按相关度匹配,逐个检索,
把新素材**追加**到现有素材列表(URL 去重、保持原有顺序、不重排不丢弃)。
"""
from __future__ import annotations
import asyncio
@@ -6,8 +19,7 @@ import logging
import os
import re
import time
import uuid
from typing import List
from types import SimpleNamespace
from langchain_core.runnables import RunnableConfig
@@ -71,11 +83,63 @@ _MAX_MAX_MATERIALS = 100
_DEFAULT_KEYWORD_COUNT = 6
_MAX_KEYWORD_COUNT = 10
# 单个技能的 web 检索超时(秒)。技能检索是顺序循环、最多 5 个技能,
# 单个技能的 web 检索超时(秒)。技能检索是顺序循环,
# DuckDuckGo 等 provider 偶发限速/挂起会把整段"技能检索过程"拖到几十秒看起来卡死;
# 超时即放弃该技能、emit 0 条结果继续下一个,保证时间线持续推进。
_SKILL_SEARCH_TIMEOUT_SECONDS = 25
# 内置「知识库检索」技能名:素材收集专家 agent 的默认技能。researcher 对它
# 特殊处理——不用 description 拼 query,而是对 Step 1 生成的全部检索词做
# 原生 web/intranet 并发检索(即旧版 knowledge_base 模式的行为)。
KNOWLEDGE_SEARCH_SKILL_NAME = "knowledge-base-search"
_KNOWLEDGE_SKILL_DISPLAY = "知识库检索"
_KNOWLEDGE_SKILL_FALLBACK_DESC = "对写作意图提炼的检索词执行 web/intranet 知识库检索,获取相关素材"
def _fallback_knowledge_skill() -> SimpleNamespace:
"""技能目录里找不到 knowledge-base-search 时的内存兜底(原生路径不依赖 SKILL.md)。"""
return SimpleNamespace(name=KNOWLEDGE_SEARCH_SKILL_NAME, description=_KNOWLEDGE_SKILL_FALLBACK_DESC)
def _load_configured_skills() -> list:
"""解析素材收集专家 agent(ai-writing-researcher)配置的技能列表。
config.yaml 的 ``skills`` 字段是权威来源(admin/用户在智能体管理页可改);
未配置 / 解析失败 / 技能文件全部缺失时,一律回退为内置「知识库检索」,
保证通用检索永远可用。返回的元素带 ``name`` / ``description`` 属性。
"""
names: list[str] = []
try:
from deerflow.config.agents_config import load_agent_config
cfg = load_agent_config(RESEARCHER_AGENT_ID)
if cfg is not None:
names = [n.strip() for n in (cfg.skills or []) if isinstance(n, str) and n.strip()]
except Exception:
logger.warning("researcher: 读取 agent 配置技能失败,回退知识库检索", exc_info=True)
if not names:
return [_fallback_knowledge_skill()]
by_name: dict[str, object] = {}
try:
from deerflow.skills.storage import get_or_new_skill_storage
by_name = {s.name: s for s in get_or_new_skill_storage().load_skills(enabled_only=False)}
except Exception:
logger.warning("researcher: 加载技能目录失败", exc_info=True)
resolved: list = []
for n in names:
sk = by_name.get(n)
if sk is None and n == KNOWLEDGE_SEARCH_SKILL_NAME:
sk = _fallback_knowledge_skill()
if sk is None:
logger.warning("researcher: 配置的技能 '%s' 不存在于技能目录,已跳过", n)
continue
resolved.append(sk)
return resolved or [_fallback_knowledge_skill()]
def _resolve_keyword_count(raw: object) -> int:
"""把用户传入的检索关键词数量规整到 [1, _MAX_KEYWORD_COUNT],非法值回退默认 6。"""
@@ -247,32 +311,24 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
# 检查干预模式(PAUSE-1 素材确认 / 素材不足求助)
intervention = state.get("last_user_intervention") or {}
extra_keywords: list[str] = intervention.get("_extra_keywords") or intervention.get("extra_keywords") or []
# PAUSE-1 单输入框:用户写的自然语言(技能名 / 来源 / 检索意图),优先级最高
# PAUSE-1 单输入框:用户写的自然语言(想补充检索的内容),优先级最高
user_query: str = (intervention.get("user_query") or "").strip()
action = intervention.get("action")
# 素材不足求助点触发的「补充检索」:保留原大纲,检索完直接回写作、不走素材确认/大纲
is_supplement = action == "supplement_search"
# PAUSE-1 素材确认卡用户点「按输入重新搜索」并填了 user_query:
# PAUSE-1 素材确认卡用户在输入框填了自然语言(如「检索 A 的数据」):
# - 跳过 LLM 重新生成检索词(节省一次调用 + 尊重用户意图)
# - 直接用 user_query 作为唯一检索词
# - 保留 existing material_package,按 URL 去重合并
is_user_query_search = action == "re_search" and bool(user_query)
# - 从 agent 配置的技能中匹配最相关的技能,逐个检索
# - **追加**:保留 existing material_package,新素材按 URL 去重后追加在末尾
# 旧版「启用技能检索」按钮(action=enable_skill_search)按同一语义兼容处理。
is_user_query_search = action in ("re_search", "enable_skill_search") and bool(user_query)
# 兼容旧版"追加关键词"调用方(如果还有 client 传 extra_keywords 数组而非 user_query)
is_append_only = action == "re_search" and not user_query and bool(extra_keywords)
# 用户在写作表单选择的素材来源:knowledge_base(默认 web/intranet 检索)/ skill(技能检索)。
# 首次检索(没有用户干预)时按它决定检索方式;缺省 knowledge_base。
material_source = (state.get("material_source") or "knowledge_base").strip()
is_first_run = not intervention
is_initial_skill_source = is_first_run and material_source == "skill"
# 技能检索触发条件(任一即可):
# - 首次检索且表单选了「技能检索」来源
# - PAUSE-1 素材确认卡用户点「启用技能检索」按钮
# 技能检索流程:
# - 不再走 web/intranet 检索,改用 SkillSearchProvider
# - 顺序调用候选技能 + 早停(第一个技能够就停,不够再加下一个)
# - 旧 material_package 完全丢弃(覆盖式替换,让用户对比 web vs skill 结果)
is_skill_search = action == "enable_skill_search" or is_initial_skill_source
# 用户在写作表单选择的检索类型:general(通用检索,按 agent 配置技能编排)/
# notebook(我的空间)。旧值 knowledge_base / skill 一律按 general 兼容。
raw_source = (state.get("material_source") or "general").strip()
material_source = "notebook" if raw_source == "notebook" else "general"
# 用户指定的检索关键词数量(1~10),决定生成多少个检索词、检索多少个方向
keyword_count = _resolve_keyword_count(state.get("keyword_count"))
@@ -291,7 +347,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
max_materials = _resolve_max_materials(ai_writing_config.get("max_materials"))
# 内置 agent 模式(config.yaml ai_writing.use_builtin_agents.researcher):
# 检索编排(并发搜索 / 技能编排 / 笔记检索)仍由节点控制,只把两个 LLM 子任务
# 检索编排(技能编排 / 笔记检索)仍由节点控制,只把两个 LLM 子任务
# (提取检索词 / 素材排序)换成跑 ai-writing-researcher agent(SOUL 可被 admin
# 编辑;模型优先级 agent.model > keyword_model/rank_model > 表单主模型)。
use_agent = use_builtin_agent("researcher")
@@ -308,14 +364,11 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
writer({"type": "research_progress", "message": message})
# Step 1: 提取关键词
kw_result: dict = {} # 走"用户输入 / 追加 / 技能"分支时跳过 LLM 调用,notice 拼接需要兜底值
if is_skill_search:
# 技能模式:每个 skill 会自己拼 query(user_query + skill.description),这里不用 LLM 生成关键词
keywords: list[str] = [user_query or intent]
elif is_user_query_search:
kw_result: dict = {} # 走"用户输入 / 追加"分支时跳过 LLM 调用,notice 拼接需要兜底值
if is_user_query_search:
# 用户输入分支:把整段自然语言作为唯一检索词,尊重原意,不再做关键词拆分
keywords = [user_query]
_emit(f"按你的输入「{user_query[:30]}」检索…")
keywords: list[str] = [user_query]
_emit(f"按你的输入「{user_query[:30]}」匹配技能检索…")
elif is_append_only:
# 兼容旧版追加模式:直接用用户填的关键词,不再调 LLM
keywords = _clean_keywords(extra_keywords, intent)
@@ -359,7 +412,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
# Step 1.5: 笔记检索(如果传入了 notebook_search_space_id)
notebook_search_space_id: int | None = state.get("notebook_search_space_id")
notebook_materials: list[Material] = []
if notebook_search_space_id:
notebook_source_ids: list[int] = state.get("notebook_source_ids") or []
notebook_token: str = cfg.get("notebook_token") or ""
@@ -373,12 +426,13 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
if notebook_materials:
_emit(f"从笔记中获取到 {len(notebook_materials)} 条素材")
# Step 2: 搜索
# 技能检索:技能不是素材,是"检索策略 / 视角"。流程:
# a. SkillPicker 选出按相关度排序的 skills(推 list_skills + select_skills 事件)
# b. 按顺序遍历 skills:每个 skill 用其 description 让 LLM 生成针对性关键词
# c. 调真实 web/intranet provider 执行检索,推 invoke_skill + skill_result 事件
# d. 累积素材到 max_materials 即 break,推 stop_early
# Step 2: 检索
# 通用检索 = 配置技能编排。技能不是素材,是"检索策略 / 视角":
# a. 从 ai-writing-researcher agent config.yaml 读取配置技能(默认知识库检索)
# b. 「知识库检索」技能 → 对 Step 1 检索词做原生并发 web/intranet 检索
# c. 其它技能 → 用 skill.description 拼定向 query 调真实 provider
# d. user_query 模式(PAUSE-1 输入框)→ 先让模型按用户输入匹配技能再检索
# e. 累积素材到 max_materials 即 break,推 stop_early
skill_steps: list[dict] = []
search_errors: list[str] = []
@@ -390,188 +444,208 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
time.monotonic(), step.get("type"), step.get("skill_name") or "")
writer({"type": "skill_search_step", "agent_name": researcher_label, **step})
if is_skill_search:
# ── 技能编排检索 ──
# 技能选择走流式 LLM(推 skill_select_chunk)+ <think> 分离,前端打字机展示
picker = SkillPicker(
config=config,
model_name=keyword_model,
emit_step=_emit_skill_step,
max_select=5,
user_query=user_query,
)
selected = await picker.pick(user_intent=intent)
search_errors.extend(picker.errors)
# 本轮新检索到的原始结果(统一在下方合并段转 Material、与旧素材去重)
new_results: list[SearchResult] = []
provider = None
# 真实 web/intranet provider(按 config.yaml 配置)
web_provider = get_search_provider(ai_writing_config)
# 每个 skill 尽量多取(不分摊),第一个就达到 max_materials 即可早停;
# 这才贴近"第一个技能够了就停,否则下一个补充"的语义。
per_skill = max(1, max_materials)
raw_materials_pre: list[Material] = []
seen_urls_pre: set[str] = set()
stopped_early = False
for entry in selected:
sk = entry["skill"]
skill_query = f"{user_query} {sk.description or ''}".strip() or intent
# invoke_skill 步骤带上实际检索词(截断防过长),让用户在每次长 await 之前
# 就看到"这一步在用技能 X 检索什么",配合前端 spinner 体现进行中。
_emit_skill_step({
"type": "invoke_skill",
"message": f"使用技能「{sk.name}」检索:{skill_query[:60]}",
"skill_name": sk.name,
})
try:
# 加超时兜底:单个技能检索最多等 _SKILL_SEARCH_TIMEOUT_SECONDS 秒,
# 超时不再阻塞后续技能,避免整段过程看起来"冻结"。
res = await asyncio.wait_for(
web_provider.search(skill_query, max_results=per_skill),
timeout=_SKILL_SEARCH_TIMEOUT_SECONDS,
)
except asyncio.TimeoutError:
logger.warning("skill[%s] search timed out after %ss", sk.name, _SKILL_SEARCH_TIMEOUT_SECONDS)
search_errors.append(f"skill[{sk.name}]: 检索超时(>{_SKILL_SEARCH_TIMEOUT_SECONDS}s)")
res = []
except Exception as e:
logger.warning("skill[%s] search failed: %s", sk.name, e)
search_errors.append(f"skill[{sk.name}]: {e}")
res = []
# 去重累积
new_count = 0
for r in res or []:
if r.url and r.url in seen_urls_pre:
continue
if r.url:
seen_urls_pre.add(r.url)
raw_materials_pre.append(Material(
id=f"m{len(raw_materials_pre)+1:03d}",
title=r.title,
content=r.content[:500],
source=f"{r.source}(via 技能 {sk.name})",
published_at=r.published_at,
relevance_score=r.relevance_score,
source_type=r.source_type,
url=r.url,
recUuid=r.recUuid,
))
new_count += 1
_emit_skill_step({
"type": "skill_result",
"message": f"技能「{sk.name}」检索到 {new_count} 条新素材",
"skill_name": sk.name,
"matched_count": new_count,
})
# 早停:累积素材已达上限
if len(raw_materials_pre) >= max_materials:
stopped_early = True
_emit_skill_step({
"type": "stop_early",
"message": f"已获取 {len(raw_materials_pre)} 条素材,达到上限,停止调用后续技能",
"matched_count": len(raw_materials_pre),
})
break
# 技能模式不再走下面的并发/合并逻辑,all_results_nested 留空、跳过 web 通用搜索
all_results_nested: list = []
# 把已收集到的素材挂在 raw_materials_pre,下方合并阶段直接挪过去
elif material_source == "notebook" and notebook_search_space_id:
if material_source == "notebook" and notebook_search_space_id:
# ── 纯笔记模式 ──
# 用户在表单选了「我的空间」作为素材来源 → 跳过 web/intranet 检索,
# 用户在表单选了「我的空间」作为检索类型 → 跳过技能编排,
# 仅使用 Step 1.5 已拉取的 notebook_materials。
# 与 UI 文案"从空间中检索素材"对齐。
_emit(f"仅使用笔记素材,跳过 web/intranet 检索({len(notebook_materials)} 条)")
all_results_nested = []
else:
# ── 常规 web/intranet 模式 ──
_emit(f"仅使用笔记素材,跳过技能检索({len(notebook_materials)} 条)")
elif is_append_only:
# ── 兼容旧版追加关键词模式 ──
# 直接对用户给的关键词做原生并发检索(不走技能编排),结果追加到旧素材。
provider = get_search_provider(ai_writing_config)
# 按用户指定的关键词数量截断。
# 用户输入 / 追加模式:keywords 本身就是用户填的,全部检索不截断。
if is_user_query_search or is_append_only:
search_keywords = keywords
else:
search_limit = keyword_count
if extra_keywords:
search_limit = max(search_limit, len(keywords))
search_keywords = keywords[:search_limit]
total = len(search_keywords)
per_keyword = _per_keyword_limit(max_materials, total)
_emit(f"检索词已生成,开始检索 {total} 个方向…")
completed = 0
async def _search_one(kw: str) -> list[SearchResult]:
nonlocal completed
try:
return await provider.search(kw, max_results=per_keyword)
finally:
completed += 1
if completed < total:
_emit(f"已完成 {completed}/{total} 次检索,正在继续检索相关内容…")
else:
_emit(f"已完成全部 {total} 次检索")
search_tasks = [_search_one(kw) for kw in search_keywords]
all_results_nested = await asyncio.gather(*search_tasks, return_exceptions=True)
# provider 内部吞掉的检索异常(如内网 ES 连接失败),一并取出用于提示用户
search_errors.extend(getattr(provider, "errors", []))
# 合并去重
# - 首次检索 / 普通 re_search(不带 extra_keywords,用户只想 LLM 重出):旧素材丢弃
# - 素材不足求助 supplement_search:保留旧素材 + 追加新检索结果
# - PAUSE-1 追加检索 is_append_only:保留旧素材 + 仅检索新关键词
# - 技能检索 is_skill_search:覆盖式替换,旧素材完全丢弃(编排循环已经把素材累积到 raw_materials_pre)
existing_pkg = state.get("material_package")
if is_skill_search:
# 技能编排循环已经完成素材累积,直接用 raw_materials_pre 当 raw_materials
# 笔记素材追加在技能素材之后
raw_materials: list[Material] = raw_materials_pre # type: ignore[name-defined]
raw_materials.extend(notebook_materials)
else:
seen_urls: set[str] = set()
# 笔记素材优先排在最前面,赋予高相关度
raw_materials = list(notebook_materials)
for m in notebook_materials:
if m.get("url"):
seen_urls.add(m["url"])
preserve_existing = existing_pkg and (is_supplement or is_append_only)
if preserve_existing:
for m in existing_pkg.get("materials", []):
seen_urls.add(m.get("url", ""))
raw_materials.append(m)
for results in all_results_nested:
per_keyword = _per_keyword_limit(max_materials, len(keywords))
_emit(f"开始追加检索 {len(keywords)} 个方向…")
gathered = await asyncio.gather(
*[provider.search(kw, max_results=per_keyword) for kw in keywords],
return_exceptions=True,
)
for results in gathered:
if isinstance(results, Exception):
logger.warning("search failed: %s", results)
search_errors.append(f"{type(results).__name__}: {results}")
continue
for r in results:
if r.url and r.url in seen_urls:
new_results.extend(results or [])
search_errors.extend(getattr(provider, "errors", []))
else:
# ── 通用检索:配置技能编排 ──
configured = _load_configured_skills()
provider = get_search_provider(ai_writing_config)
per_skill = max(1, max_materials)
ordered = configured
if is_user_query_search and len(configured) > 1:
# 用户输入了自然语言:让模型在配置技能中按相关度匹配排序
# (技能选择走流式 LLM,推 skill_select_chunk,前端打字机展示)
picker = SkillPicker(
config=config,
model_name=keyword_model,
emit_step=_emit_skill_step,
max_select=5,
user_query=user_query,
candidates=configured,
)
ranked = await picker.pick(user_intent=intent)
search_errors.extend(picker.errors)
ordered = [entry["skill"] for entry in ranked] or configured
else:
_emit_skill_step({
"type": "list_skills",
"message": f"素材收集专家已配置 {len(configured)} 个检索技能",
"skill_names": [s.name for s in configured],
})
seen_urls_new: set[str] = set()
def _collect(results: list[SearchResult]) -> int:
added = 0
for r in results or []:
if r.url and r.url in seen_urls_new:
continue
if r.url:
seen_urls.add(r.url)
raw_materials.append(Material(
id=f"m{len(raw_materials)+1:03d}",
title=r.title,
content=r.content[:500],
source=r.source,
published_at=r.published_at,
relevance_score=r.relevance_score,
source_type=r.source_type,
url=r.url,
recUuid=r.recUuid,
))
seen_urls_new.add(r.url)
new_results.append(r)
added += 1
return added
# Step 3: 相关度排序(如素材较多,让 LLM 排序)
_emit("检索完成,正在筛选排序素材…")
async def _timed_search(skill_name: str, query: str, limit: int) -> list[SearchResult]:
# 加超时兜底:单个技能检索最多等 _SKILL_SEARCH_TIMEOUT_SECONDS 秒,
# 超时不再阻塞后续技能,避免整段过程看起来"冻结"。
try:
return await asyncio.wait_for(
provider.search(query, max_results=limit),
timeout=_SKILL_SEARCH_TIMEOUT_SECONDS,
) or []
except asyncio.TimeoutError:
logger.warning("skill[%s] search timed out after %ss", skill_name, _SKILL_SEARCH_TIMEOUT_SECONDS)
search_errors.append(f"skill[{skill_name}]: 检索超时(>{_SKILL_SEARCH_TIMEOUT_SECONDS}s)")
return []
except Exception as e:
logger.warning("skill[%s] search failed: %s", skill_name, e)
search_errors.append(f"skill[{skill_name}]: {e}")
return []
for sk in ordered:
is_knowledge = sk.name == KNOWLEDGE_SEARCH_SKILL_NAME
if is_knowledge and not is_user_query_search:
# 原生知识库检索:对全部检索词并发检索(旧版默认路径)。
# 用户带 extra_keywords 时全部检索,否则按 keyword_count 截断。
search_keywords = keywords if extra_keywords else keywords[:keyword_count]
total = len(search_keywords)
per_keyword = _per_keyword_limit(max_materials, total)
_emit(f"检索词已生成,开始检索 {total} 个方向…")
_emit_skill_step({
"type": "invoke_skill",
"message": f"使用技能「{_KNOWLEDGE_SKILL_DISPLAY}」检索 {total} 个方向:{('、'.join(search_keywords[:3]))[:60]}",
"skill_name": sk.name,
})
completed = 0
async def _search_one(kw: str) -> list[SearchResult]:
nonlocal completed
try:
return await provider.search(kw, max_results=per_keyword)
finally:
completed += 1
if completed < total:
_emit(f"已完成 {completed}/{total} 次检索,正在继续检索相关内容…")
else:
_emit(f"已完成全部 {total} 次检索")
gathered = await asyncio.gather(*[_search_one(kw) for kw in search_keywords], return_exceptions=True)
added = 0
for results in gathered:
if isinstance(results, Exception):
logger.warning("search failed: %s", results)
search_errors.append(f"{type(results).__name__}: {results}")
continue
added += _collect(results)
search_errors.extend(getattr(provider, "errors", []))
else:
# 定向技能检索:user_query 模式优先用用户原话;非知识库技能再
# 附上 description 让 query 贴合该技能的领域。
if is_user_query_search:
skill_query = user_query if is_knowledge else f"{user_query} {sk.description or ''}".strip()
else:
skill_query = f"{intent} {sk.description or ''}".strip() or intent
_emit_skill_step({
"type": "invoke_skill",
"message": f"使用技能「{sk.name}」检索:{skill_query[:60]}",
"skill_name": sk.name,
})
added = _collect(await _timed_search(sk.name, skill_query, per_skill))
_emit_skill_step({
"type": "skill_result",
"message": f"技能「{_KNOWLEDGE_SKILL_DISPLAY if is_knowledge else sk.name}」检索到 {added} 条新素材",
"skill_name": sk.name,
"matched_count": added,
})
# 早停:累积素材已达上限
if len(new_results) >= max_materials:
_emit_skill_step({
"type": "stop_early",
"message": f"已获取 {len(new_results)} 条素材,达到上限,停止调用后续技能",
"matched_count": len(new_results),
})
break
# 合并去重
# - 首次检索 / 普通 re_search(无输入,用户只想重出):旧素材丢弃,覆盖式替换
# - 素材不足求助 supplement_search:保留旧素材 + 追加新检索结果
# - PAUSE-1 输入框 is_user_query_search:保留旧素材 + **追加**新素材(核心交互)
# - 旧版追加关键词 is_append_only:保留旧素材 + 仅检索新关键词
existing_pkg = state.get("material_package")
preserve_existing = bool(existing_pkg) and (is_supplement or is_append_only or is_user_query_search)
seen_urls: set[str] = set()
# 笔记素材优先排在最前面,赋予高相关度
raw_materials: list[Material] = list(notebook_materials)
for m in notebook_materials:
if m.get("url"):
seen_urls.add(m["url"])
if preserve_existing:
for m in existing_pkg.get("materials", []):
seen_urls.add(m.get("url", ""))
raw_materials.append(m)
appended_count = 0
for r in new_results:
if r.url and r.url in seen_urls:
continue
if r.url:
seen_urls.add(r.url)
raw_materials.append(Material(
id=f"m{len(raw_materials)+1:03d}",
title=r.title,
content=r.content[:500],
source=r.source,
published_at=r.published_at,
relevance_score=r.relevance_score,
source_type=r.source_type,
url=r.url,
recUuid=r.recUuid,
))
appended_count += 1
# Step 3: 相关度排序
# 用户主动追加(输入框 / 追加关键词):保持旧素材原有顺序,新素材追加在
# 末尾,不调 LLM 重排、不按上限丢弃旧素材(只做 100 条硬上限防爆)。
rank_error = ""
if len(raw_materials) > max_materials:
is_user_append = is_user_query_search or is_append_only
if is_user_append:
materials = raw_materials[:_MAX_MAX_MATERIALS]
dropped = len(raw_materials) - len(materials)
summary = f"追加检索到 {appended_count} 条新素材,当前共 {len(materials)} 条"
if dropped > 0:
summary += f"(超出 {_MAX_MAX_MATERIALS} 条硬上限,丢弃末尾 {dropped} 条)"
elif len(raw_materials) > max_materials:
_emit("检索完成,正在筛选排序素材…")
materials_list = "\n".join(
f"- ID: {m['id']} 标题: {m['title']} 摘要: {m['content'][:100]}"
for m in raw_materials
@@ -608,26 +682,29 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
materials = sorted_materials[:max_materials]
summary = rank_result.get("summary", f"共搜索到{len(raw_materials)}条素材,精选{len(materials)}条")
else:
_emit("检索完成,正在筛选排序素材…")
materials = sorted(raw_materials, key=lambda m: m.get("relevance_score", 0), reverse=True)[:max_materials]
summary = f"共搜索到{len(raw_materials)}条素材,精选{len(materials)}条"
# 空结果兜底:未检索到任何素材时给出明确文案,前端据此弹出「无数据继续」选项
if not materials:
summary = "未检索到相关素材,可重新搜索或在无素材的情况下继续写作"
elif is_user_append and appended_count == 0:
summary = f"本次未检索到新素材(已有 {len(materials)} 条保持不变),可换个说法再试"
# 追加模式下,pkg.keywords 体现到目前为止累积的检索词(旧 + 新追加),
# 让前端列表 / 历史回显能看到完整的搜索路径。
pkg_keywords = keywords
if is_append_only and existing_pkg:
if is_user_append and existing_pkg:
old_kws = list(existing_pkg.get("keywords") or [])
pkg_keywords = list(dict.fromkeys(old_kws + keywords))
pkg = MaterialPackage(keywords=pkg_keywords, materials=materials, summary=summary)
# 技能检索时把过程步骤一并放进 pkg,前端在素材确认卡里能展示"使用了哪些技能"
if is_skill_search:
# 技能编排过程步骤放进 pkg,前端在素材确认卡里能展示"使用了哪些技能"
if skill_steps:
pkg["skill_search_used"] = True
pkg["skill_search_steps"] = skill_steps
elif existing_pkg and existing_pkg.get("skill_search_used") and (is_supplement or is_append_only):
# 保留之前一次技能检索的痕迹(不污染纯 web 检索)
elif existing_pkg and existing_pkg.get("skill_search_used") and preserve_existing:
# 保留之前一次技能检索的痕迹
pkg["skill_search_used"] = existing_pkg.get("skill_search_used", False)
pkg["skill_search_steps"] = existing_pkg.get("skill_search_steps", [])
@@ -640,9 +717,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
if search_errors:
uniq = list(dict.fromkeys(search_errors))
# 解析失败时收集到的接口原始响应体,前端折叠展示便于定位
# (技能模式下 provider 局部变量没暴露,所以 raw_dumps 兜底为空列表)
provider_obj = locals().get("provider") or locals().get("web_provider")
raw_dumps = list(dict.fromkeys(getattr(provider_obj, "raw_responses", []) or []))
raw_dumps = list(dict.fromkeys(getattr(provider, "raw_responses", []) or [])) if provider is not None else []
notices.append(_notice(
"ES 波动导致检索异常",
";".join(uniq[:3]),
@@ -650,13 +725,13 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
agent_name=researcher_label,
))
# 进度事件文案:用户输入 / 追加模式 / 技能模式下让用户看到差异化提示
if is_skill_search:
step_start_msg = f"启用技能检索({len(skill_steps)} 步)"
elif is_user_query_search:
step_start_msg = f"按你的输入检索:{user_query[:40]}"
# 进度事件文案:用户输入 / 追加模式下让用户看到差异化提示
if is_user_query_search:
step_start_msg = f"按你的输入匹配技能检索并追加:{user_query[:40]}"
elif is_append_only:
step_start_msg = f"追加检索关键词:{', '.join(keywords[:3])}"
elif skill_steps:
step_start_msg = f"通用检索(技能编排 {len(skill_steps)} 步):{', '.join(keywords[:3])}"
else:
step_start_msg = f"搜索关键词:{', '.join(keywords[:3])}"
return {
@@ -70,6 +70,8 @@ class SkillPicker:
emit_step: 推送时间线步骤事件(list_skills / select_skills)
max_select: LLM 单次最多挑几个技能
user_query: 用户在素材确认卡输入框写的自然语言
candidates: 候选技能池。None = 全部启用的技能(旧行为);
传入列表(如素材收集专家 agent 配置的技能)则只在其中匹配排序
"""
def __init__(
@@ -80,12 +82,14 @@ class SkillPicker:
emit_step: Optional[EmitStepFn] = None,
max_select: int = 5,
user_query: str = "",
candidates: Optional[list] = None,
) -> None:
self._config = config
self._model_name = model_name
self._emit_step = emit_step or (lambda _step: None)
self._max_select = max_select
self._user_query = (user_query or "").strip()
self._candidates = candidates
self._errors: list[str] = []
@property
@@ -93,6 +97,14 @@ class SkillPicker:
return list(self._errors)
def _load_enabled_skills(self) -> list:
if self._candidates is not None:
skills = list(self._candidates)
self._emit_step({
"type": "list_skills",
"message": f"素材收集专家已配置 {len(skills)} 个检索技能",
"skill_names": [s.name for s in skills],
})
return skills
try:
from deerflow.skills import get_or_new_skill_storage
storage = get_or_new_skill_storage()
@@ -135,8 +135,9 @@ class AIWritingState(TypedDict):
keyword_count: int
# 写作模式:strict=严格依据素材 / loose=可结合通用知识(默认 loose)
writing_mode: str
# 素材来源:knowledge_base=知识库(默认 web/intranet 检索)/ skill=技能检索。
# 用户在写作表单选择,首次检索时由 researcher 节点据此决定检索方式(缺省 knowledge_base)。
# 检索类型:general=通用检索(按素材收集专家 agent 配置的技能编排,默认)/
# notebook=我的空间(仅笔记检索)。旧值 knowledge_base / skill 由 researcher
# 节点按 general 兼容处理。
material_source: str
# 用户在写作表单选定的主模型。大纲/写作/审核严格使用它(不回退默认)。
# 由 intent_parser 在首次 invocation 从 configurable.model_name 捕获进 state,
@@ -13,7 +13,13 @@ Phase 3 会提供具体的 ``RoundtableGateway`` 实现(复用 multi_agent.py
from __future__ import annotations
from .engine import OrchestrationResult, build_report_prompt, run_orchestration
from .engine import (
OrchestrationResult,
build_dashboard_prompt,
build_report_prompt,
build_summary_prompt,
run_orchestration,
)
from .progress import JobProgressWriter
from .simulated_gateway import SimulatedRoundtableGateway
from .types import (
@@ -27,6 +33,7 @@ from .types import (
RoundtableGateway,
SeatRef,
SeatTurn,
SummaryResult,
build_chain_leader_task,
build_dag_stage_leader_task,
parse_dag_plan,
@@ -38,12 +45,15 @@ __all__ = [
"run_orchestration",
"OrchestrationResult",
"build_report_prompt",
"build_summary_prompt",
"build_dashboard_prompt",
"JobProgressWriter",
"SimulatedRoundtableGateway",
"RoundtableGateway",
"LeaderTurn",
"SeatTurn",
"ReportResult",
"SummaryResult",
"SeatRef",
"DagStage",
"DagPlan",
@@ -32,6 +32,7 @@ from .types import (
ReportResult,
RoundtableGateway,
SeatRef,
SummaryResult,
build_chain_leader_task,
build_dag_stage_leader_task,
parse_dag_plan,
@@ -94,9 +95,8 @@ async def run_orchestration(
intent_text: 任务意图,用于报告 prompt(缺省退回 seed_message)。
plan: dag 模式的分层编排计划(前端 ``OrchestrationPlan`` dict)。mode=="dag" 且
可解析出非空 stages 时走 ``_run_dag``;否则(dag 但 plan 缺失/为空)回退 recommend。
enable_report: 是否在收口后执行 Step3 报告绘制。False 时只跑完 Step2 研讨即收尾、
不生成报告(step3=None)。当前「后台挂起」临时禁用报告(进程内驱动报告 agent
在真模型下仍无法稳定写出 HTML,详见 multi-agent-background-run-遗留问题与下次继续.md)。
enable_report: 是否在收口后执行 Step3(方案总结报告 + 结果绘制)。False 时只跑完
Step2 研讨即收尾(step3=None)。两段产物各自故障隔离,详见 ``_run_step3``。
"""
intent = (intent_text or seed_message or "").strip()
try:
@@ -120,11 +120,11 @@ async def run_orchestration(
dialogues=progress.dialogues(),
)
# consensus 或 max_cycles:均收口。报告绘制按 enable_report 开关执行。
# consensus 或 max_cycles:均收口。Step3(总结报告 + 结果绘制)按 enable_report 开关执行。
await progress.consensus(100)
step3 = None
if enable_report:
step3 = await _run_report(
step3 = await _run_step3(
gateway,
progress,
agents=agents,
@@ -132,6 +132,7 @@ async def run_orchestration(
intent_text=intent_text or seed_message,
consensus_text=outcome.last_leader_content,
deliveries=outcome.deliveries,
coordinator_prompt=(dag_plan.coordinator_prompt if dag_plan is not None else None),
)
await progress.done()
return OrchestrationResult(
@@ -428,7 +429,96 @@ async def _run_dag(
return _LoopOutcome("consensus", last_leader_content=last_leader, deliveries=deliveries)
# ── 报告阶段 ─────────────────────────────────────────────────────────────────
# ── Step3 阶段(总结报告 + 结果绘制)────────────────────────────────────────────
async def _run_step3(
gateway: RoundtableGateway,
progress: JobProgressWriter,
*,
agents: list[SeatRef],
model: str | None,
intent_text: str,
consensus_text: str,
deliveries: dict[str, str],
coordinator_prompt: str | None = None,
) -> dict | None:
"""Step3 双产物:「方案总结报告」(roundtable-summary 写 md)与「结果绘制」
(roundtable-report 画 HTML)**并行执行**(两段无依赖,2026-06-13 由串行改并行,
Step3 总耗时 ≈ max(两段) 而非两段之和)。
返回与前端 ``DraftStep3Snapshot`` 同形状的 dict:
``{summaryReport?: {md,threadId,generatedAt,model,closing}, html?, generatedAt?, model?, summary?}``。
**故障隔离**:两段各自 try/except —— 一段失败不拖垮另一段,跑成多少交付多少;
两段都失败时抛出第一个异常(由上层标记 job error)。两段各跑在独立线程上
(gateway 内部各建独立 thread),进程内并行安全(与 ``_run_dag`` 批内并行同理)。"""
step3: dict = {}
errors: list[Exception] = []
async def _summary_task() -> None:
# ① 方案总结报告(Markdown)—— 与前端 useStep3Summary 同源 prompt / 同款快照形状。
try:
await progress.summary_writing()
summary_thread_id = await gateway.create_summary_thread()
summary_prompt = build_summary_prompt(
intent_text, agents, deliveries, consensus_text, coordinator_prompt=coordinator_prompt
)
summary: SummaryResult = await gateway.run_summary(
summary_thread_id=summary_thread_id, prompt=summary_prompt, model=model
)
if (summary.md or "").strip() or (summary.closing or "").strip():
step3["summaryReport"] = {
"md": summary.md or "",
"threadId": summary.thread_id or summary_thread_id,
"generatedAt": summary.generated_at or datetime.now(UTC).isoformat(),
"model": summary.model or (model or ""),
"closing": summary.closing or "",
}
except Exception as exc: # noqa: BLE001 — 总结失败不阻塞结果绘制
logger.exception("roundtable step3 summary failed")
errors.append(exc)
async def _report_task() -> None:
# ② 结果绘制(HTML 看板)。
try:
report = await _run_report(
gateway,
progress,
agents=agents,
model=model,
intent_text=intent_text,
consensus_text=consensus_text,
deliveries=deliveries,
)
step3.update(report)
except Exception as exc: # noqa: BLE001 — 绘制失败时保留已写出的总结报告
logger.exception("roundtable step3 report failed")
errors.append(exc)
async def _dashboard_task() -> None:
# ③ 大屏多页(模式二)的结构化数据(report-json 字符串)。后端**只产数据**,
# 前端查看时用 renderDashboardHtml 按当前主题渲染(与手动并行绘制同源)。
# 旧网关没有 run_dashboard_data 则静默跳过(向后兼容)。
try:
run_dash = getattr(gateway, "run_dashboard_data", None)
if run_dash is None:
return
prompt = build_dashboard_prompt(intent_text, agents, deliveries, consensus_text)
data_json = await run_dash(prompt=prompt, model=model)
if data_json and data_json.strip():
step3["dashboardJson"] = data_json
except Exception as exc: # noqa: BLE001 — 大屏失败不拖垮另两段
logger.exception("roundtable step3 dashboard failed")
errors.append(exc)
await asyncio.gather(_summary_task(), _report_task(), _dashboard_task())
if not step3:
if errors:
raise errors[0]
return None
return step3
async def _run_report(
@@ -453,6 +543,108 @@ async def _run_report(
}
def build_dashboard_prompt(
intent_text: str,
agents: list[SeatRef],
deliveries: dict[str, str],
consensus_text: str,
) -> str:
"""拼装「大屏多页(模式二)」的 report-json 数据提炼 prompt —— 对齐前端
``report-template.buildDashboardPrompt``:模型只输出 1-2KB 结构化数据
(不调工具、不写文件、不输出 HTML),前端按当前主题模板渲染。"""
name_by_id = {a.agent_id: a.name for a in agents}
lines: list[str] = [
"【结构化数据模式】本轮**不要调用任何工具、不要写任何文件、不要输出 HTML**。",
"请通读以下圆桌研讨成果,把它们提炼成一份大屏展示数据,在回复**最末尾**输出一个完整的 ```report-json 代码块(契约见下),代码块之外只允许一两句简短说明。",
"",
"# 一、任务意图",
(intent_text or "").strip() or "(未提供)",
"",
"# 二、各席位最终交付",
]
idx = 0
for a in agents:
text = (deliveries.get(a.agent_id) or "").strip()
if not text:
continue
idx += 1
lines.append(f"## {idx}. {name_by_id.get(a.agent_id, a.agent_id)}")
lines.append(text)
lines.append("")
lines.append("# 三、总控最终共识结论")
lines.append((consensus_text or "").strip() or "(未提供)")
lines.append("")
lines.append(
"\n".join(
[
"```report-json 契约(严格 JSON,全程简体中文):",
'{"title":"大屏主标题(≤20字)","subtitle":"一句话定位",',
' "kpis":[{"label":"指标名","value":"数值","unit":"单位","trend":"up|down|flat"}], // 3-6 个核心指标,从研讨内容提取或合理估算',
' "consensus":{"headline":"一句话最终结论","points":["结论要点,3-6 条"]},',
' "charts":[{"type":"bar|donut|radar","title":"图表标题","items":[{"label":"维度","value":数字}]}], // 1-3 个总览图表(风险分布/资源占比/方案评分等),3-6 项/图',
' "timeline":[{"label":"阶段名","desc":"一句话说明"}], // 执行路线 3-6 阶段',
' "seats":[{"id":"seat-1","name":"席位名(与交付席位一致)","summary":"该席位结论一两句",',
' "sections":[{"heading":"小节标题","bullets":["要点,每条≤40字"]}], // 2-4 节,提炼该席位交付',
' "metrics":[{"label":"量化指标名","value":0到100}], // 0-3 个',
' "risks":[{"label":"风险点","level":"high|mid|low"}], // 0-4 个',
' "dependsOn":["它承接哪些席位的产物(席位id)"]}]} // 每个交付席位一项;依据交付内容判断依赖关系',
"硬性要求:seats 覆盖**每一个**有交付的席位;id 用 seat-1..seat-N;dependsOn 只引用本清单内的 id;JSON 必须可被 JSON.parse 解析(无注释、无尾逗号)。",
]
)
)
return "\n".join(lines)
def build_summary_prompt(
intent_text: str,
agents: list[SeatRef],
deliveries: dict[str, str],
consensus_text: str,
*,
coordinator_prompt: str | None = None,
) -> str:
"""拼装 Step3「方案总结报告」prompt —— 对齐前端 ``useStep3Summary.buildSummaryPrompt``
(非流程图分支)+ ``buildMaterials`` 的原料结构,使后台产出与手动调用一致。"""
name_by_id = {a.agent_id: a.name for a in agents}
lines: list[str] = [
"请基于以下圆桌多智能体研讨的成果,撰写一份以文字叙述为主线的 Markdown「方案总结报告」,"
"用 write_file 工具写入 /mnt/user-data/outputs/方案总结报告.md,"
"并把报告完整正文同时输出到对话回复里供用户直接阅读。",
"",
]
cp = (coordinator_prompt or "").strip()
if cp:
lines += [
"# 业务链总控编排提示(最高优先级,按它的关注点/结构/口径来组织报告)",
cp,
"",
]
lines += ["# 一、任务意图", (intent_text or "").strip() or "(未提供)", ""]
lines.append("# 二、各席位最终交付")
has_delivery = False
for i, a in enumerate(agents):
text = (deliveries.get(a.agent_id) or "").strip()
if not text:
continue
has_delivery = True
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((consensus_text or "").strip() or "(未提供)")
lines.append("")
lines.append(
"请严格遵循你的职责设定:纯 Markdown 文本(不画 HTML / 图表)、全程简体中文。"
"报告结构:默认用叙事型结构(一句话结论 → 背景目标 → 各席位观点 → 综合分析 → 最终共识方案 → 执行路线);"
"但若上面的「业务链总控编排提示」对最终总结提出了要求(输出结构/格式、重点席位、口径、详略),"
"那些要求优先于默认结构——用户指定的结构可以完全替换默认章节。"
)
return "\n".join(lines)
def build_report_prompt(
intent_text: str,
agents: list[SeatRef],
@@ -501,6 +693,9 @@ def build_report_prompt(
"文字为主图表点睛、浅红主题(主色 #c25555);写完文件后只用一两句话总结,不要把 HTML 源码贴进对话。"
)
lines.append(
'硬性要求:① 全程使用简体中文;② HTML 必须是完整文档,<head> 最前面声明 <meta charset="utf-8">。'
'硬性要求:① 全程使用简体中文;② HTML 必须是完整文档,<head> 最前面声明 <meta charset="utf-8">;'
"③ 整份 HTML 较长时**不要试图一次性写入**——先用 write_file 写入开头部分(约 200-300 行),"
"再用 write_file 的 append=true 参数分多次续写(每段 ≤300 行),直到文档以 </html> 收尾"
"(单次超长输出会被截断,导致文件根本写不进去)。"
)
return "\n".join(lines)
@@ -10,8 +10,9 @@
from __future__ import annotations
import logging
from collections.abc import Awaitable, Callable
from datetime import datetime, timedelta, timezone
from typing import Any, Awaitable, Callable
from typing import Any
from .types import SeatRef
@@ -140,6 +141,10 @@ class JobProgressWriter:
self._consensus = int(pct)
await self._persist(phase="consensus", active_agent_id=None)
async def summary_writing(self) -> None:
"""Step3 第一段:方案总结报告(roundtable-summary 写 Markdown)撰写中。"""
await self._persist(phase="summary_writing", active_agent_id=None)
async def report_drawing(self) -> None:
await self._persist(phase="report_drawing", active_agent_id=None)
@@ -13,8 +13,9 @@ multi_agent 的纯解析函数 + ``set_current_user`` 进程内调 agent runtime
from __future__ import annotations
import asyncio
import json
from .types import LeaderTurn, ReportResult, SeatRef, SeatTurn
from .types import LeaderTurn, ReportResult, SeatRef, SeatTurn, SummaryResult
class SimulatedRoundtableGateway:
@@ -63,3 +64,44 @@ class SimulatedRoundtableGateway:
"<body><h1>方案总览(模拟)</h1><p>这是模拟网关生成的占位报告,用于前端联调。</p></body></html>"
)
return ReportResult(html=html, summary="(模拟)方案已生成", model=model)
async def create_summary_thread(self) -> str:
return "sim-summary-thread"
async def run_summary(self, *, summary_thread_id, prompt, model) -> SummaryResult:
await self._tick()
md = "# 方案总结报告(模拟)\n\n这是模拟网关生成的占位总结报告,用于前端联调。"
return SummaryResult(md=md, closing="(模拟)总结报告已生成", model=model, thread_id=summary_thread_id)
async def run_dashboard_data(self, *, prompt, model) -> str | None:
await self._tick()
seats = [
{
"id": f"seat-{i + 1}",
"name": a.name,
"summary": f"(模拟){a.name} 的结论摘要。",
"sections": [{"heading": "要点", "bullets": ["模拟要点一", "模拟要点二"]}],
"metrics": [{"label": "完成度", "value": 90}],
"risks": [],
"dependsOn": [f"seat-{i}"] if i > 0 else [],
}
for i, a in enumerate(self._agents)
]
return json.dumps(
{
"title": "方案研讨大屏(模拟)",
"subtitle": "模拟网关占位数据",
"kpis": [{"label": "席位数", "value": str(len(self._agents)), "unit": "个"}],
"consensus": {"headline": "(模拟)推荐方案已达成共识", "points": ["要点一", "要点二"]},
"charts": [
{
"type": "bar",
"title": "席位贡献(模拟)",
"items": [{"label": a.name, "value": 60 + i * 10} for i, a in enumerate(self._agents)],
}
],
"timeline": [{"label": "阶段一", "desc": "模拟阶段说明"}],
"seats": seats,
},
ensure_ascii=False,
)
@@ -85,6 +85,22 @@ class ReportResult:
generated_at: str | None = None
@dataclass
class SummaryResult:
"""Step3「方案总结报告」结果(由 gateway.run_summary 产出)。
与前端 ``DraftStep3SummarySnapshot`` 一一对应:``md`` = 写入
``outputs/方案总结报告.md`` 的完整 Markdown;``closing`` = 智能体在对话里
回显的报告正文/收尾语;``thread_id`` = 真总结线程 id(加载草稿后可续问)。
"""
md: str = ""
closing: str | None = None
model: str | None = None
thread_id: str | None = None
generated_at: str | None = None
# ── 网关协议 ────────────────────────────────────────────────────────────────
@@ -130,6 +146,23 @@ class RoundtableGateway(Protocol):
model: str | None,
) -> ReportResult: ...
async def create_summary_thread(self) -> str: ...
async def run_summary(
self,
*,
summary_thread_id: str,
prompt: str,
model: str | None,
) -> SummaryResult: ...
async def run_dashboard_data(
self,
*,
prompt: str,
model: str | None,
) -> str | None: ...
# ── chain 模式 prompt 构造(移植自前端 buildChainLeaderTask / pickChainTask)────