From 8859c13cfb7db5baf0b4a4d6d587d23f9bb3414f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Tue, 8 Sep 2026 09:40:31 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BB=A3=E7=A0=81=E5=90=88=E5=B9=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...e-多智能体报告协作工作台-交互重设计方案.md | 186 ++++++++ .../components/ReportCollaborationHeader.tsx | 15 +- .../components/ReportFlowCanvas.tsx | 26 +- .../components/RequirementSummaryCard.tsx | 52 +++ .../pages/ReportCollaborationPage.tsx | 93 ++-- .../state/message-adapter.ts | 17 + .../styles/report-collaboration.css | 15 + .../gateway/routers/report_collaboration.py | 26 +- .../agentscope_runtime/model_adapter.py | 86 ++++ .../execution/executor.py | 13 +- .../report_collaboration/execution/ledger.py | 164 ++++++- .../execution/member_runner.py | 7 +- .../interventions/service.py | 1 + .../planning/planner_agent.py | 422 ++++++++++++++++++ .../report_collaboration/planning/service.py | 72 ++- .../requirement/__init__.py | 4 +- .../requirement/intent_router.py | 90 +--- .../requirement/resolver.py | 209 ++++++--- offline-backend-20260512/backend/config.yaml | 1 + .../config/report_collaboration_config.py | 4 + .../tests/test_report_collaboration_api.py | 39 +- .../tests/test_report_collaboration_models.py | 29 +- ...test_report_collaboration_planner_agent.py | 181 ++++++++ .../test_report_collaboration_planning.py | 51 ++- .../test_report_collaboration_requirement.py | 155 +++---- .../tests/test_report_collaboration_runner.py | 83 ++++ 26 files changed, 1707 insertions(+), 334 deletions(-) create mode 100644 docs/agentscope-多智能体报告协作工作台-交互重设计方案.md create mode 100644 frontend-web/src/report-collaboration/components/RequirementSummaryCard.tsx create mode 100644 offline-backend-20260512/backend/app/report_collaboration/planning/planner_agent.py create mode 100644 offline-backend-20260512/backend/tests/test_report_collaboration_planner_agent.py diff --git a/docs/agentscope-多智能体报告协作工作台-交互重设计方案.md b/docs/agentscope-多智能体报告协作工作台-交互重设计方案.md new file mode 100644 index 00000000..8fc89ed0 --- /dev/null +++ b/docs/agentscope-多智能体报告协作工作台-交互重设计方案.md @@ -0,0 +1,186 @@ +# 报告协作工作台 — 交互重设计方案（v2） + +> 日期：2026-09-07 +> 状态：**待确认**（确认后按 §7 分期实施） +> 取代：现行「规则澄清 + 三张模板方案卡 + 画布为主」的交互 +> 原则：回到最初设计 —— **协调者智能体全程与用户对话的主线** + +--- + +## 1. 设计目标与总体形态 + +用户视角只有一条主线：**和一个规划师智能体对话 → 它摸清意图并规划工作流 → 确认后执行、过程在流程图上可见 → 随时插话干预、指定智能体重跑 → 最终成稿、继续对话式修改**。 + +``` +┌─────────────────────────────────────────────────────────┐ +│ 阶段1 计划模式 阶段2/3 执行+干预 阶段4 成稿 │ +│ ┌───────────┐ ┌───────────┐ ┌──────────┐ │ +│ │ 规划师对话 │ 确认 │ 聊天时间线 │ 审稿过 │ 对话改稿 │ │ +│ │ 需求摘要卡 │ ─────→ │ +成员Tab │ ─────→ │ 版本/diff│ │ +│ │ 工作流卡 │ 冻结 │ +流程图动画│ 报告卡 │ 上游重跑 │ │ +│ └───────────┘ └───────────┘ └──────────┘ │ +│ ↑ 回到计划模式（replan）随时可触发 ↑ │ +└─────────────────────────────────────────────────────────┘ +``` + +与现行的核心差异： + +| 维度 | 现行（做偏的） | v2 | +|------|----------------|-----| +| 需求澄清 | 规则模板问句秒回 | 规划师智能体（LLM）领域化提问 | +| 工作流来源 | 3 张确定性模板 | LLM 按意图生成自定义 DAG，对话迭代修改 | +| 执行展示 | 画布为主、聊天为辅 | **聊天时间线为主**（成员 Tab 切换保留），**执行状态在流程图上动画呈现** | +| 干预 | 命令确认卡体系 | 输入框常开 → 协调者智能体分类；支持**指定智能体重跑** | +| 报告修改 | 改写候选（纯文字层） | 意图分流：润色走改写；**事实性问题触发上游节点重跑** | + +--- + +## 2. 阶段 1：计划模式 — 需求梳理与工作流规划 + +### 2.1 计划模式状态 + +- 用户发出第一条需求消息起进入**计划模式**，Header 显示徽章「📋 计划模式」，输入框提示语切换为「梳理需求中，你可以随时补充或纠正…」。 +- 计划模式在**工作流确认冻结**时结束；执行期大改可通过 replan 回到计划模式（见 §5.4）。 + +### 2.2 规划师对话（LLM，替换现行规则路由） + +- 模型：`report_collaboration.planner_model`（回退 `coordinator_model` → 默认模型，复用现有 model_resolver 冻结机制）。 +- 提问是领域化的：写市场分析就问「关注整车还是供应链 / 要不要竞品对比 / 给谁看 / 篇幅」，而非固定模板问句。 +- 澄清轮次上限沿用 `max_clarification_rounds`；用户随时可以说「直接开始」按默认假设推进。 + +### 2.3 需求摘要卡 + +信息足够后，规划师在对话中输出一张**结构化需求摘要卡**（业务卡插槽渲染）：主题 / 目的 / 读者 / 角度清单 / 篇幅与重点 / 默认假设。用户逐项看，可在对话里继续改（「读者改成投资人」→ 卡片更新出新版）。 + +### 2.4 工作流生成与迭代 + +- 用户确认摘要后，规划师生成**一份**工作流方案（不是三选一）：以工作流卡（成员构成、各节点使命、预计时长/成本）+ 右侧流程图预览呈现。 +- 生成约束（保证可执行）：**节点类型受词表约束** — `research / verify / analyze / write / review`；自由的是结构（几条线、并行还是串行、核验深度）与每个节点的使命/角度/工具。 +- 对话式修改：「把价格策略单独拆一条线」「加一个政策风险分析」→ 规划师修订后出新版工作流卡（带变更说明），再确认。 +- 「开始执行」→ 冻结需求 + 工作流（含模型/预算快照，沿用现有 freeze 机制）。 + +--- + +## 3. 阶段 2：执行 — 过程在流程图上可见 + +### 3.1 流程图动画规格（执行状态的可视化主载体） + +节点状态 → 样式（`ReportFlowCanvas` / `ReportFlowNode`，状态源 `selectNodeStatusMap`，事件源 `node.status.changed` SSE）： + +| 状态 | 节点样式 | 连入边 | +|------|----------|--------| +| planned / queued | 灰色常态 | 静态 | +| **running** | 主色描边 + **呼吸脉冲动画**（ring）+ 角标 spinner | **流动动画**（@xyflow 内置 `animated: true` 的 dash 流动） | +| **completed** | 绿色 + ✓ 角标，常驻 | 恢复静态实线 | +| failed | 红色 + 错误角标 | 红色虚线 | +| awaiting_input | 橙色 + ？角标 | 橙色静态 | +| interrupted / superseded | 半透明降级 | 灰 | + +- 语义：**边流动 = 数据正在流入该节点**（该节点执行中）；节点完成即停动画换完成样式。 +- 实现成本：@xyflow 原生支持 animated 边与自定义节点 className，纯 CSS 补呼吸动画即可，无需新依赖。 + +### 3.2 聊天区（保留现有 Tab 切换） + +- 并行成员沿用 **AgentStageSelector Tab 切换**（维持现状，不改交错时间线）。 +- 每个 Tab 内是该成员的完整流式输出：思考、工具调用卡、检索过程、正文 —— 与普通智能体问答同观感（message-adapter 已具备映射能力）。 +- 协调者播报（「研究线 A 完成，核心发现 X，开始交叉核验」）进入总时间线 Tab。 + +--- + +## 4. 阶段 3：干预 — 对话即干预 + 指定重跑 + +### 4.1 输入框常开，协调者智能体分流 + +执行期输入框永远可用，所有输入先到协调者（LLM 分类，规则高精度直判兜底——复用 RC-BE-014 IntentRouter 骨架，把「规则优先」反转为「LLM 优先 + 规则快路径」）： + +| 用户意图 | 协调者行为 | 生效时机 | +|----------|-----------|----------| +| 提问（「查到什么了？」） | 直接回答 | 立即，不打断 | +| 修正方向（「重点看下沉市场」） | 小改：在安全点注入 overlay，并回复说明 | 下个安全点 | +| **指定重跑**（见 §4.2） | 出确认卡 → 重开节点 attempt | 确认后 | +| 大改（砍线/加线） | 提议 replan → 回计划模式 | 用户确认 | +| 停止 | 直接取消 | 立即 | + +### 4.2 指定智能体重跑（新增核心交互） + +**场景**：用户发现某节点传给下游的信息不对（如研究线 B 的结论有误）。 + +两个入口，同一后端动作： + +1. **对话**：「研究线 B 的结果不对，重新查，重点看价格战」→ 协调者解析出 `node_id` + 新指令； +2. **画布**：点节点 → 抽屉（NodeInspectorDrawer）加「↻ 重新执行」按钮。 + +**级联规则**（复用并补全 ledger 的 targeted rework）： + +``` +重跑节点 N（attempt+1，带用户新指令作为 mission overlay） + → N 完成 → 所有直接/间接依赖 N 产物的下游节点标记 superseded + → 按 DAG 依赖序逐波重跑（边动画重新流动，用户全程可见） + → 受影响章节重新生成 +``` + +- 确认卡必须展示**影响面**：「将级联重跑 核验B / 分析 / 写作 共 3 个节点，预计 +8 分钟」。 +- 成本保护：级联重跑计入 run 预算（现有 budget 机制），预算不足时明确报因。 + +--- + +## 5. 阶段 4：报告成稿与对话式修改 + +### 5.1 成稿 + +审稿通过 → 报告卡出现在时间线（点开 ReportPreviewDialog 全文），协调者播报总结（各线核心发现 → 结论如何形成）。 + +### 5.2 修改意图分流（协调者 LLM 判定） + +| 用户说 | 判定 | 动作 | +|--------|------|------| +| 「把第二章结论写得更笃定」 | 文字润色 | 改写候选（现有 rewrite API）→ 对比预览 → apply → 新版本 | +| 「这个数据不对 / 重新收集一下」 | **事实性问题** | 定位来源节点 → **触发该节点重跑（可带新角度）→ 级联 §4.2 → 受影响章节重生成 → 新版本** | +| 「再从供应链角度补一节」 | 结构性新增 | replan：回计划模式改工作流（只增不删已完成部分） | +| 「恢复到上一版」 | 版本回退 | 现有 restore API | + +- 版本链：每次 apply / restore / 级联重跑产出均产生新 head，历史完整保留（现有机制）。 +- 修改对话里协调者必须先**复述理解**（「你是想让研究线 B 换个角度重查，然后重写第三章，对吗？」）再执行，避免误触发昂贵的级联。 + +--- + +## 6. 工程改造清单 + +### 6.1 保留（本次联调已验证） + +- 持久化 13 张表 + durable SSE + `after_seq` 断线续传；幂等键 / 乐观锁 / 预算框架。 +- MessageList 复用链路（message-adapter → LangGraph Message → 业务卡插槽）。 +- AgentStageSelector Tab、commands 安全点骨架（RC-BE-014）、报告版本三件套（RC-BE-015）、前端 hooks/reducer/selectors 主体。 +- 「清空历史」「生成候选方案」等本次新增入口。 + +### 6.2 后端重做 / 新建 + +| 项 | 说明 | +|----|------| +| 规划师智能体 | LLM 多轮澄清（替换 `requirement/intent_router.py` 规则路由）+ 需求摘要结构化 + **工作流 LLM 生成**（受 §2.4 词表约束，替换 `planning/controller.py` 三模板）+ 对话式工作流修订 | +| 协调者智能体升级 | 干预分类 LLM 优先；修改意图分流（§5.2）；执行前复述确认 | +| 指定重跑 + 级联 | ledger targeted rework 补全级联 superseded/重跑波次；确认卡带影响面计算 | +| AgentScope 适配器修复 | `count_tokens` 协议缺失（上下文压缩必崩）；attempt 耗尽终态化（防僵尸 run） | +| 事件补齐 | 边流动所需的节点状态事件已有；补「级联影响预告」事件 | + +### 6.3 前端重做 / 新建 + +| 项 | 说明 | +|----|------| +| 计划模式 UI | Header 徽章、composer 提示切换、需求摘要卡、工作流卡（替换现三张模板卡） | +| 流程图动画 | running 节点呼吸动画 + 入边 `animated`、completed ✓ 样式、级联重跑时边动画重流（§3.1） | +| 节点抽屉 | 加「重新执行」按钮 → 确认卡 | +| 干预交互 | 执行期输入框常开；意图确认卡（含影响面）；修改对话复述确认 | +| 布局 | 聊天为主、画布为辅的默认比例微调（不动整体两栏结构） | + +--- + +## 7. 实施分期 + +| 期 | 内容 | 验收 | +|----|------|------| +| **P0 地基** | AgentScope `count_tokens` 修复；attempt 耗尽终态化；级联重跑的 ledger 补全 | 执行阶段不崩、无僵尸 run、单节点重跑级联正确（后端测试） | +| **P1 计划模式 + 执行可视化** | 规划师 LLM 对话、需求摘要卡、工作流 LLM 生成与迭代、计划模式 UI、流程图动画 | 从新会话到确认工作流全程对话完成；执行时动画正确随状态流转 | +| **P2 干预 + 成稿修改** | 协调者 LLM 分类、指定重跑（两入口 + 影响面确认卡）、报告修改意图分流、上游级联重跑 | 四类干预各自走通；「数据不对→重跑→新章节→新版本」全链路 | + +每期结束跑 RC 全量测试（当前基线 233 通过）+ 前端 26 单测，不引入对其他模块的影响。 diff --git a/frontend-web/src/report-collaboration/components/ReportCollaborationHeader.tsx b/frontend-web/src/report-collaboration/components/ReportCollaborationHeader.tsx index 0d2a5567..034b109d 100644 --- a/frontend-web/src/report-collaboration/components/ReportCollaborationHeader.tsx +++ b/frontend-web/src/report-collaboration/components/ReportCollaborationHeader.tsx @@ -1,9 +1,12 @@ "use client"; -import { ArrowLeft, FileText, MessageSquarePlus, Square } from "lucide-react"; +import { ArrowLeft, ClipboardList, FileText, MessageSquarePlus, Square } from "lucide-react"; import type { PagePhase } from "@/report-collaboration/state/selectors"; +/** 计划模式覆盖的阶段：第一条消息起到工作流确认执行前。 */ +const PLAN_MODE_PHASES = new Set(["clarifying", "planning", "proposal_ready"]); + const PHASE_BADGES: Record = { empty: { label: "新会话", color: "var(--rc-text-secondary)", bg: "transparent" }, clarifying: { label: "澄清需求", color: "var(--rc-waiting)", bg: "rgba(217, 119, 6, 0.1)" }, @@ -51,6 +54,16 @@ export function ReportCollaborationHeader({ 报告协作 / {title} + {PLAN_MODE_PHASES.has(phase) ? ( + + + 计划模式 + + ) : null} {badge.label} diff --git a/frontend-web/src/report-collaboration/components/ReportFlowCanvas.tsx b/frontend-web/src/report-collaboration/components/ReportFlowCanvas.tsx index ba6e30f0..5101d83c 100644 --- a/frontend-web/src/report-collaboration/components/ReportFlowCanvas.tsx +++ b/frontend-web/src/report-collaboration/components/ReportFlowCanvas.tsx @@ -156,8 +156,23 @@ function CanvasInner({ plan, nodeStates, selectedNodeId, onSelectNode, emptyHint const edges: Edge[] = useMemo(() => { if (!plan) return []; return plan.edges.map((edge, index) => { - const targetState = nodeStates[edge.target]; - const active = targetState?.status === "running" || targetState?.status === "validating"; + // 边语义 = 数据正在流入 target 节点：target 执行中 → 流动动画； + // 失败红虚线 / 等待橙 / 已被替代降灰，其余静态。 + const targetStatus = nodeStates[edge.target]?.status; + let className = "rc-edge"; + let stroke = "var(--rc-flow-edge)"; + if (targetStatus === "running" || targetStatus === "validating") { + className = "rc-edge-active"; + stroke = "var(--rc-running)"; + } else if (targetStatus === "failed") { + className = "rc-edge-failed"; + stroke = "var(--rc-danger)"; + } else if (targetStatus === "awaiting_input") { + className = "rc-edge-waiting"; + stroke = "var(--rc-waiting)"; + } else if (targetStatus === "superseded" || targetStatus === "interrupted" || targetStatus === "cancelled") { + className = "rc-edge-dimmed"; + } return { id: `${edge.source}->${edge.target}-${index}`, source: edge.source, @@ -165,11 +180,8 @@ function CanvasInner({ plan, nodeStates, selectedNodeId, onSelectNode, emptyHint label: edge.label ?? undefined, type: "default", animated: false, - className: active ? "rc-edge-active" : "rc-edge", - style: { - stroke: active ? "var(--rc-running)" : "var(--rc-flow-edge)", - strokeWidth: 2, - }, + className, + style: { stroke, strokeWidth: 2 }, }; }); }, [plan, nodeStates]); diff --git a/frontend-web/src/report-collaboration/components/RequirementSummaryCard.tsx b/frontend-web/src/report-collaboration/components/RequirementSummaryCard.tsx new file mode 100644 index 00000000..32a2cd0e --- /dev/null +++ b/frontend-web/src/report-collaboration/components/RequirementSummaryCard.tsx @@ -0,0 +1,52 @@ +"use client"; + +import { ClipboardCheck } from "lucide-react"; + +import type { ReportRequirementSnapshot } from "@/report-collaboration/api/types"; + +interface RequirementSummaryCardProps { + requirement: ReportRequirementSnapshot; + /** 需求仍可修改的提示（生成方案前）；方案已出后为 false。 */ + editable?: boolean; +} + +function Field({ label, value }: { label: string; value: string | null | undefined }) { + if (!value) return null; + return ( +
+ {label} + {value} +
+ ); +} + +/** 计划模式 · 需求摘要卡：结构化展示规划师确认的需求快照。 */ +export function RequirementSummaryCard({ requirement, editable = true }: RequirementSummaryCardProps) { + const angles = (requirement.required_angles ?? []).filter(Boolean).join("、"); + return ( +
+
+ + 需求摘要 · v{requirement.revision} + + 计划模式 + +
+
+ + + + + + + + +
+ {editable ? ( +

+ 我正在基于这份需求为你规划工作流，稍候即可在对话中查看方案；想调整可直接在下方补充。 +

+ ) : null} +
+ ); +} diff --git a/frontend-web/src/report-collaboration/pages/ReportCollaborationPage.tsx b/frontend-web/src/report-collaboration/pages/ReportCollaborationPage.tsx index 4db6c59a..005d243d 100644 --- a/frontend-web/src/report-collaboration/pages/ReportCollaborationPage.tsx +++ b/frontend-web/src/report-collaboration/pages/ReportCollaborationPage.tsx @@ -4,7 +4,7 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useLocation, useNavigate } from "react-router-dom"; import type { Message } from "@langchain/langgraph-sdk"; import { toast } from "sonner"; -import { FileText, Loader2, Sparkles } from "lucide-react"; +import { FileText, Sparkles } from "lucide-react"; import { ArtifactsProvider } from "@/components/workspace/artifacts"; import { ThreadContext } from "@/components/workspace/messages/context"; @@ -26,6 +26,7 @@ import { } from "@/report-collaboration/components/ReportPreviewDialog"; import { ReportSandboxOverlay } from "@/report-collaboration/components/ReportSandboxOverlay"; import { ReportSpeakerShell } from "@/report-collaboration/components/ReportSpeakerShell"; +import { RequirementSummaryCard } from "@/report-collaboration/components/RequirementSummaryCard"; import { SessionHistoryPanel } from "@/report-collaboration/components/SessionHistoryPanel"; import { useReportPanelLayout } from "@/report-collaboration/hooks/useReportPanelLayout"; import { useReportCollaborationMessages } from "@/report-collaboration/hooks/useReportCollaborationMessages"; @@ -35,6 +36,7 @@ import { buildCommandConfirmCard, buildPlanGroupCard, buildReportAvailableCard, + buildRequirementReadyCard, readReportCollaborationCard, } from "@/report-collaboration/state/message-adapter"; import { resolveReportSpeaker } from "@/report-collaboration/state/speaker"; @@ -247,11 +249,23 @@ export function ReportCollaborationSessionPage({ sessionId }: { sessionId: strin if (next) setFocusedAgentRunId(next); }, [agentTabs, focusPinned, focusedAgentRunId, tabIds]); - // 虚拟业务卡：方案组（已出方案且 run 未开始）/ 命令确认 / 报告入口。 + // 虚拟业务卡：需求摘要（计划模式）/ 方案组（已出方案且 run 未开始）/ 命令确认 / 报告入口。 // 采用方案后卡片保留，「开始协作」按钮随选定态启用；run 开始后消失。 const runStarted = Boolean(state.run && state.run.plan_id); const extraCards = useMemo(() => { const cards: Message[] = []; + if ( + state.requirement && + state.plans.length === 0 && + !runStarted && + Object.values(state.messagesById).some( + (message) => + message.role === "ai" && + (message.metadata as Record | null | undefined)?.ready_to_plan === true, + ) + ) { + cards.push(buildRequirementReadyCard(state.requirement)); + } if ( state.plans.length > 0 && !runStarted && @@ -341,39 +355,16 @@ export function ReportCollaborationSessionPage({ sessionId }: { sessionId: strin [sessionId, session], ); - // 澄清完成（协调者已写入需求版本）但还没生成过方案：给出显式「发起规划」入口。 - const requirementReady = useMemo( - () => - Object.values(state.messagesById).some( - (message) => - message.role === "ai" && - (message.metadata as Record | null | undefined)?.ready_to_plan === true, - ), - [state.messagesById], - ); - const showPlanRequestEntry = - state.session?.status === "clarifying" && state.plans.length === 0 && requirementReady; - const [planRequestBusy, setPlanRequestBusy] = useState(false); - const handleRequestPlans = useCallback(async () => { - setPlanRequestBusy(true); - try { - await getReportCollaborationDriver().requestPlans( - sessionId, - state.session?.requirement_revision ?? null, - ); - await session.refreshSnapshot(); - toast.success("已生成候选方案，可在对话中选择"); - } catch (err) { - toast.error(err instanceof Error ? err.message : String(err)); - } finally { - setPlanRequestBusy(false); - } - }, [sessionId, session, state.session?.requirement_revision]); + // 方案生成由后端在规划师判定 ready 后自动衔接（同一次请求内规划）， + // 前端无手动「生成候选方案」按钮；此处仅保留摘要卡的注入判定。 const businessCardSlot = useCallback( (message: Message) => { const card = readReportCollaborationCard(message); if (!card) return null; + if (card.kind === "requirement_ready") { + return ; + } if (card.kind === "plan_group") { return ( - {showPlanRequestEntry ? ( -
- -
- ) : null} - setComposerTarget(null)} - onSend={(text) => - void session.send(text, { - nodeId: composerTarget?.nodeId ?? null, + setComposerTarget(null)} + onSend={(text) => + void session.send(text, { + nodeId: composerTarget?.nodeId ?? null, agentRunId: composerTarget?.agentRunId ?? null, }) } focusSignal={focusSignal} /> - } /> } diff --git a/frontend-web/src/report-collaboration/state/message-adapter.ts b/frontend-web/src/report-collaboration/state/message-adapter.ts index 32fa3b95..f69831fa 100644 --- a/frontend-web/src/report-collaboration/state/message-adapter.ts +++ b/frontend-web/src/report-collaboration/state/message-adapter.ts @@ -21,6 +21,7 @@ import type { CollaborationCommand, CollaborationMessage, ReportPlanCandidate, + ReportRequirementSnapshot, ReportVersion, } from "@/report-collaboration/api/types"; @@ -113,6 +114,11 @@ export type ReportCollaborationCardPayload = selectedPlanId: string | null; requirementRevision: number; } + | { + kind: "requirement_ready"; + requirement: ReportRequirementSnapshot; + revision: number; + } | { kind: "command_confirm"; command: CollaborationCommand; @@ -163,6 +169,17 @@ export function buildPlanGroupCard( }); } +/** 需求摘要卡（计划模式：需求已就绪、尚未生成方案）。 */ +export function buildRequirementReadyCard( + requirement: ReportRequirementSnapshot, +): Message { + return buildCardMessage(`rc-card-requirement-r${requirement.revision}`, { + kind: "requirement_ready", + requirement, + revision: requirement.revision, + }); +} + /** 命令确认卡（command.confirmation.required 之后）。 */ export function buildCommandConfirmCard(command: CollaborationCommand): Message { return buildCardMessage(`rc-card-confirm-${command.command_id}`, { diff --git a/frontend-web/src/report-collaboration/styles/report-collaboration.css b/frontend-web/src/report-collaboration/styles/report-collaboration.css index 21e8f8d7..f46de066 100644 --- a/frontend-web/src/report-collaboration/styles/report-collaboration.css +++ b/frontend-web/src/report-collaboration/styles/report-collaboration.css @@ -850,6 +850,21 @@ animation: rc-edge-flow 0.5s linear infinite; } +/* P1.4：失败红虚线 / 等待用户橙色 / 已替代·中断降灰 */ +.rc-edge-failed .react-flow__edge-path { + stroke: var(--rc-danger); + stroke-dasharray: 4 4; +} + +.rc-edge-waiting .react-flow__edge-path { + stroke: var(--rc-waiting); +} + +.rc-edge-dimmed .react-flow__edge-path { + stroke: var(--rc-flow-edge); + opacity: 0.45; +} + @keyframes rc-edge-flow { from { stroke-dashoffset: 10; diff --git a/offline-backend-20260512/backend/app/gateway/routers/report_collaboration.py b/offline-backend-20260512/backend/app/gateway/routers/report_collaboration.py index 2a9da612..d62ae255 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/report_collaboration.py +++ b/offline-backend-20260512/backend/app/gateway/routers/report_collaboration.py @@ -302,7 +302,29 @@ async def send_message( agent_run_id=body.agent_run_id, node_run_id=body.target_node_id, ) - await RequirementResolver(store).handle_user_message(session_id, user_message=created, idempotency_key=key) + decision = await RequirementResolver( + store, + config=get_app_config().report_collaboration, + planner_llm_factory=getattr(request.app.state, "report_collaboration_planner_llm", None), + ).handle_user_message(session_id, user_message=created, idempotency_key=key) + # 确认完意图直接规划工作流（设计 v2 衔接）：ready 且尚无方案时， + # 同一次请求内自动发起规划，用户不用再手动点「生成候选方案」。 + if decision is not None and decision.action == "ready": + record = await store.get_session_record(session_id) + plans = await store.list_plans(session_id) + if not plans and record and not record.get("selected_plan_id"): + command = await store.enqueue_plan_request( + session_id, + idempotency_key=f"{key}:autoplan", + expected_revision=decision.requirement.revision, + ) + if command.get("status") == "pending": + await PlanProposalService( + store, + agent_store=getattr(request.app.state, "agent_store", None), + config=get_app_config().report_collaboration, + llm_factory=getattr(request.app.state, "report_collaboration_planner_llm", None), + ).fulfill(session_id, user_id=user_id, command_id=command["command_id"], idempotency_key=f"{key}:autoplan") return created except Exception as exc: return _map_store_error(exc) @@ -333,6 +355,8 @@ async def request_plans( await PlanProposalService( store, agent_store=getattr(request.app.state, "agent_store", None), + config=get_app_config().report_collaboration, + llm_factory=getattr(request.app.state, "report_collaboration_planner_llm", None), ).fulfill(session_id, user_id=user_id, command_id=command["command_id"], idempotency_key=key) return Response(status_code=204) except Exception as exc: diff --git a/offline-backend-20260512/backend/app/report_collaboration/agentscope_runtime/model_adapter.py b/offline-backend-20260512/backend/app/report_collaboration/agentscope_runtime/model_adapter.py index 4e4a3e1e..f82472d2 100644 --- a/offline-backend-20260512/backend/app/report_collaboration/agentscope_runtime/model_adapter.py +++ b/offline-backend-20260512/backend/app/report_collaboration/agentscope_runtime/model_adapter.py @@ -91,6 +91,62 @@ class AttemptUsage: error_type: str | None = None error_message: str | None = None + +# AgentScope 的 Agent.compress_context 会读 model.context_size 并调 count_tokens。 +_DEFAULT_CONTEXT_SIZE = 128_000 +_CJK_RE = re.compile(r"[\u4e00-\u9fff\u3040-\u30ff\uac00-\ud7af]") + + +def estimate_tokens_for_text(text: str) -> int: + """CJK-aware local token estimate: ~1 token per CJK char, ~4 bytes otherwise.""" + + if not text: + return 0 + cjk = len(_CJK_RE.findall(text)) + other = len(text) - cjk + return cjk + other // 4 + (1 if other % 4 else 0) + + +def _collect_countable_text(messages: list[Any]) -> str: + parts: list[str] = [] + for msg in messages or []: + blocks: Any = None + getter = getattr(msg, "get_content_blocks", None) + if callable(getter): + try: + blocks = getter() + except Exception: + blocks = None + if blocks is not None: + for block in blocks: + if isinstance(block, str): + parts.append(block) + elif isinstance(block, dict): + for key in ("text", "thinking", "hint"): + value = block.get(key) + if isinstance(value, str): + parts.append(value) + elif isinstance(value, list): + parts.extend(str(item) for item in value) + else: + text = getattr(block, "text", None) or getattr(block, "thinking", None) + if text: + parts.append(str(text)) + continue + if isinstance(msg, dict): + parts.append(str(msg.get("content") or "")) + else: + raw = getattr(msg, "content", "") + if isinstance(raw, list): + for block in raw: + if isinstance(block, dict): + parts.append(str(block.get("text") or "")) + else: + parts.append(str(getattr(block, "text", block) if not isinstance(block, str) else block)) + else: + parts.append(str(raw or "")) + return "\n".join(parts) + def as_dict(self) -> dict[str, Any]: return redact_mapping( { @@ -244,6 +300,21 @@ def to_lc_messages(messages: list[Any]) -> list[Any]: return converted +def _resolve_context_size(app_config: AppConfig | None, config_name: str) -> int: + """Best-effort context window from model config extras; sane default otherwise.""" + + try: + cfg = (app_config or get_app_config()).get_model_config(config_name) + extra = cfg.model_dump() if cfg is not None else {} + for key in ("context_size", "context_length", "context_window", "max_context_tokens"): + value = extra.get(key) + if isinstance(value, (int, float)) and int(value) > 0: + return int(value) + except Exception: + pass + return _DEFAULT_CONTEXT_SIZE + + class DeerFlowChatModelAdapter: """Frozen DeerFlow-backed chat model. Construct once per node attempt from a snapshot.""" @@ -266,6 +337,21 @@ class DeerFlowChatModelAdapter: self._llm_factory = llm_factory self._node_run_id = node_run_id self.last_usage = AttemptUsage(model_name=snapshot.config_name, node_run_id=node_run_id) + # AgentScope compress_context 契约：读 context_size 判断是否压缩。 + self.context_size = _resolve_context_size(app_config, snapshot.config_name) + + async def count_tokens(self, messages: list[Any], tools: list[dict[str, Any]] | None = None) -> int: + """AgentScope 契约方法（compress_context 调用）。本地估算，不打 API。""" + + import json as _json + + total = estimate_tokens_for_text(_collect_countable_text(messages)) + if tools: + try: + total += estimate_tokens_for_text(_json.dumps(tools, ensure_ascii=False, default=str)) + except Exception: + total += sum(estimate_tokens_for_text(str(item)) for item in tools) + return total def _config(self) -> AppConfig: return self._app_config or get_app_config() diff --git a/offline-backend-20260512/backend/app/report_collaboration/execution/executor.py b/offline-backend-20260512/backend/app/report_collaboration/execution/executor.py index cb82a22a..c77102ae 100644 --- a/offline-backend-20260512/backend/app/report_collaboration/execution/executor.py +++ b/offline-backend-20260512/backend/app/report_collaboration/execution/executor.py @@ -47,7 +47,18 @@ class ReportCollaborationExecutor: existing = self._tasks.get(params.run_id) if existing is not None and not existing.done(): return False - self._tasks[params.run_id] = asyncio.create_task(self.execute(params), name=f"rc-run-{params.run_id}") + task = asyncio.create_task(self.execute(params), name=f"rc-run-{params.run_id}") + + def _log_crash(finished: asyncio.Task[str]) -> None: + if not finished.cancelled() and finished.exception() is not None: + logger.error( + "rc executor task crashed for run %s", + params.run_id, + exc_info=finished.exception(), + ) + + task.add_done_callback(_log_crash) + self._tasks[params.run_id] = task return True async def drain(self) -> None: diff --git a/offline-backend-20260512/backend/app/report_collaboration/execution/ledger.py b/offline-backend-20260512/backend/app/report_collaboration/execution/ledger.py index fba3d216..6d1fcc3d 100644 --- a/offline-backend-20260512/backend/app/report_collaboration/execution/ledger.py +++ b/offline-backend-20260512/backend/app/report_collaboration/execution/ledger.py @@ -10,7 +10,7 @@ from app.report_collaboration.execution.board import ArtifactBoard from app.report_collaboration.execution.completion import late_output_may_write_completed, require_can_complete_node from app.report_collaboration.execution.quality_gate import QualityGate, QualityGateContext, requirement_from_session_record from deerflow.persistence.report_collaboration.base import ReportCollaborationStore -from deerflow.persistence.report_collaboration.codec import iso_now, make_node_run_id +from deerflow.persistence.report_collaboration.codec import _TERMINAL_RUN_STATUSES, iso_now, make_node_run_id TaskResultKind = Literal["artifact_draft", "clarification_request", "repair_request", "failure", "advice"] @@ -314,6 +314,7 @@ class TaskLedger: command_id: str, instruction: str, node_ids: list[str], + cascade_downstream: bool = False, ) -> list[str]: """Supersede affected attempts and persist an immutable command overlay. @@ -321,6 +322,10 @@ class TaskLedger: ``superseded`` and is therefore stored only as superseded evidence. User-directed rework is not counted against automatic quality-repair limits; it creates the next monotonic attempt number. + + ``cascade_downstream=True`` additionally supersedes every transitive + dependent of ``node_ids`` and re-plans it, so the wave loop re-runs + the whole affected sub-DAG in dependency order (设计 §4.2 指定重跑). """ plan, run, latest, artifacts = await self._load(run_id) @@ -407,8 +412,112 @@ class TaskLedger: active_dedupe_key=str(run["session_id"]), session_status="running", ) + if cascade_downstream and applied: + applied.extend( + await self._cascade_downstream( + run, + plan, + latest, + artifacts, + roots=[node_id for node_id in node_ids if node_id in applied], + command_id=command_id, + ) + ) return applied + async def _cascade_downstream( + self, + run: dict[str, Any], + plan: dict[str, Any], + latest: dict[str, dict[str, Any]], + artifacts: list[dict[str, Any]], + *, + roots: list[str], + command_id: str, + ) -> list[str]: + """Supersede transitive dependents of ``roots`` and re-plan them. + + Planned/reopened dependents are left alone — they have produced + nothing yet and pick up the fresh upstream artifacts when queued. + Like the rework target itself, user-directed cascade re-planning is + not budget-limited (monotonic attempt number). + """ + + deps = _dependency_map(plan) + specs = _plan_nodes(plan) + dependents: set[str] = set() + changed = True + while changed: + changed = False + for node_id, parents in deps.items(): + if node_id in dependents or node_id in roots: + continue + if any(parent in dependents or parent in roots for parent in parents): + dependents.add(node_id) + changed = True + cascaded: list[str] = [] + for node_id in sorted(dependents): + node = latest.get(node_id) + spec = specs.get(node_id) + if node is None or spec is None: + continue + status = str(node.get("status") or "") + if status in {"planned", "reopened", "superseded", "cancelled", "cancel_requested", ""}: + continue + require_transition(status, "superseded") + node["status"] = "superseded" + node["ended_at"] = iso_now() + node["awaiting_input"] = None + await self._store.save_node_run(node) + for artifact in artifacts: + if ( + artifact.get("node_id") == node_id + and int(artifact.get("attempt") or 0) == int(node.get("attempt") or 0) + and not artifact.get("superseded_by") + ): + await self._board.mark(artifact, validation_status="superseded", superseded_by=f"command:{command_id}") + await self._store.append_event( + session_id=str(run["session_id"]), + run_id=str(run["id"]), + event_type="node.status.changed", + data={"node_id": node_id, "node_run_id": node["node_run_id"], "status": "superseded", "reason": "downstream_cascade"}, + ) + nxt = int(node.get("attempt") or 0) + 1 + await self._store.save_node_run( + { + "node_run_id": make_node_run_id(str(run["id"]), node_id, nxt), + "session_id": run["session_id"], + "run_id": run["id"], + "node_id": node_id, + "attempt": nxt, + "status": "planned", + "label": node.get("label") or spec.get("label") or node_id, + "role_display_name": node.get("role_display_name") or spec.get("role_display_name"), + "angle": node.get("angle") or spec.get("angle"), + "contract_json": node.get("contract_json"), + "input_artifact_ids": [], + "command_overlay_revision": int(node.get("command_overlay_revision") or 0), + "agent_snapshot": dict(node.get("agent_snapshot") or {}), + "error": None, + "started_at": None, + "ended_at": None, + } + ) + await self._store.append_event( + session_id=str(run["session_id"]), + run_id=str(run["id"]), + event_type="node.attempt.created", + data={ + "node_id": node_id, + "node_run_id": make_node_run_id(str(run["id"]), node_id, nxt), + "attempt": nxt, + "reason": "downstream_cascade", + "command_id": command_id, + }, + ) + cascaded.append(node_id) + return cascaded + @staticmethod def _add_command_overlay(node: dict[str, Any], *, command_id: str, instruction: str) -> None: snapshot = dict(node.get("agent_snapshot") or {}) @@ -622,8 +731,18 @@ class TaskLedger: return False return True + async def refresh_run_terminal(self, run_id: str) -> str: + """Public terminal-state evaluation; safe to call repeatedly.""" + + return await self._refresh_run_terminal(run_id) + async def _refresh_run_terminal(self, run_id: str) -> str: - _plan, run, latest, _artifacts = await self._load(run_id) + plan, run, latest, _artifacts = await self._load(run_id) + run_status = str(run.get("status") or "running") + if run_status in _TERMINAL_RUN_STATUSES: + # Idempotency guard: never re-finalize (the finalizer would run + # twice when the executor's wave loop re-checks after completion). + return run_status statuses = [item["status"] for item in latest.values()] if statuses and all(status == "completed" for status in statuses): from app.report_collaboration.reporting.finalizer import ReportFinalizationError, ReportFinalizer @@ -636,11 +755,32 @@ class TaskLedger: await self._store.patch_run(run_id, status="completed", active_dedupe_key=None, session_status="completed") return "completed" if any(status in _BLOCKING or status == "queued" for status in statuses): - return str(run.get("status") or "running") + return run_status if any(status == "failed" for status in statuses): - await self._store.patch_run(run_id, status="failed", active_dedupe_key=None, session_status="failed") + await self._fail_run(run_id, run, reason="node_failed") return "failed" - return str(run.get("status") or "running") + # Zombie guard: an ``interrupted`` node whose attempt budget is + # exhausted can never run again — the ledger only queues + # planned/reopened attempts and recovery only opens a new attempt + # while attempt < max_attempts. Without this, the run would stay + # "running" forever while the dispatcher keeps re-claiming it. + specs = _plan_nodes(plan) + for node_id, item in latest.items(): + if str(item.get("status")) != "interrupted": + continue + if int(item.get("attempt") or 1) >= _node_max_attempts(specs.get(node_id), item): + await self._fail_run(run_id, run, reason="node_attempts_exhausted", node_id=node_id) + return "failed" + return run_status + + async def _fail_run(self, run_id: str, run: dict[str, Any], *, reason: str, node_id: str | None = None) -> None: + await self._store.patch_run(run_id, status="failed", active_dedupe_key=None, session_status="failed") + await self._store.append_event( + session_id=str(run.get("session_id") or ""), + run_id=run_id, + event_type="run.status.changed", + data={"status": "failed", "reason": reason, **({"node_id": node_id} if node_id else {})}, + ) async def _load(self, run_id: str) -> tuple[dict[str, Any], dict[str, Any], dict[str, dict[str, Any]], list[dict[str, Any]]]: run = await self._store.get_run(run_id) @@ -666,6 +806,20 @@ def _plan_nodes(plan: dict[str, Any]) -> dict[str, dict[str, Any]]: return {str(item["id"]): item for item in (plan.get("nodes") or []) if item.get("id")} +def _node_max_attempts(spec: dict[str, Any] | None, node_record: dict[str, Any]) -> int: + """Same precedence as ``_queue_node``: node contract, then plan spec, then 3.""" + + contract = node_record.get("contract_json") or {} + if isinstance(contract, str): + try: + import json as _json + + contract = _json.loads(contract) + except Exception: + contract = {} + return int(contract.get("max_attempts") or (spec or {}).get("max_attempts") or 3) + + def _dependency_map(plan: dict[str, Any]) -> dict[str, list[str]]: deps: dict[str, list[str]] = {node_id: list(spec.get("depends_on") or []) for node_id, spec in _plan_nodes(plan).items()} for edge in plan.get("edges") or []: diff --git a/offline-backend-20260512/backend/app/report_collaboration/execution/member_runner.py b/offline-backend-20260512/backend/app/report_collaboration/execution/member_runner.py index 26d95629..b01045ac 100644 --- a/offline-backend-20260512/backend/app/report_collaboration/execution/member_runner.py +++ b/offline-backend-20260512/backend/app/report_collaboration/execution/member_runner.py @@ -195,8 +195,11 @@ class WaveExecutor: await self._command_processor.process_safe_point(run_id) wave = await self._ledger.ready_assignments(run_id) if not wave: - run = await self._store.get_run(run_id) - return str((run or {}).get("status") or "queued") + # Nothing runnable. Evaluate terminal state: all-completed + # finalizes, exhausted-interrupted nodes fail the run (zombie + # guard). Idempotent when the run already reached a terminal + # status via apply_result. + return await self._ledger.refresh_run_terminal(run_id) for assignment in wave: self.dispatched.append(assignment.node_id) result = await self._runner.run(assignment) diff --git a/offline-backend-20260512/backend/app/report_collaboration/interventions/service.py b/offline-backend-20260512/backend/app/report_collaboration/interventions/service.py index 515dd2b5..0f34fd43 100644 --- a/offline-backend-20260512/backend/app/report_collaboration/interventions/service.py +++ b/offline-backend-20260512/backend/app/report_collaboration/interventions/service.py @@ -386,6 +386,7 @@ class RuntimeCommandService: command_id=command["command_id"], instruction=instruction, node_ids=impact_nodes, + cascade_downstream=True, ) if not changed: raise ReportCollaborationValidationError("COMMAND_NO_EFFECT", "目标节点已过期或命令已经应用") diff --git a/offline-backend-20260512/backend/app/report_collaboration/planning/planner_agent.py b/offline-backend-20260512/backend/app/report_collaboration/planning/planner_agent.py new file mode 100644 index 00000000..242a164d --- /dev/null +++ b/offline-backend-20260512/backend/app/report_collaboration/planning/planner_agent.py @@ -0,0 +1,422 @@ +"""规划师智能体（PlannerAgent）——真智能体，不是规则路由 + 模板文案。 + +设计 v2 的需求对话与工作流生成主路径： +- ``converse``：与用户多轮自由对话确认需求。回复文案（含开场）完全由 + 智能体生成并直接落库展示 —— 「收到您的报告需求。为了给出…」这类 + 死模板文案已从该路径移除。 +- ``design_workflow``：带着**完整对话上下文**（规划师问了什么、用户 + 答了什么）生成受词表约束的自定义工作流（``StrategyRoleSet``）， + 与确定性模板走同一条 装配 → 校验 → 绑定 链路。 + +规则路由与模板只作为模型彻底不可用时的最后防线，不出现在用户面前。 +""" + +from __future__ import annotations + +import json +import logging +import re +from typing import Any, Callable + +from pydantic import BaseModel, ConfigDict, Field, ValidationError + +from app.report_collaboration.agentscope_runtime.model_resolver import ROLE_MODEL_FIELD, RoleModelSnapshot, resolve_role_model +from app.report_collaboration.contracts.plans import RoleDemand +from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot +from app.report_collaboration.planning.controller import RoleSlot, StrategyRoleSet +from app.report_collaboration.requirement.intent_router import _merge_snapshot, apply_default_assumptions +from deerflow.config.app_config import AppConfig, get_app_config + +logger = logging.getLogger(__name__) + +# ── 词表（结构化输出的安全边界，LLM 不受约束的自由只在文案层） ───────────── + +MEMBER_ROLES: tuple[str, ...] = tuple(key for key in ROLE_MODEL_FIELD if key != "coordinator") + +ROLE_ARTIFACTS: dict[str, tuple[str, ...]] = { + "researcher": ("EvidenceBundle",), + "verifier": ("VerificationResult",), + "analyst": ("AngleAnalysis",), + "synthesizer": ("SynthesisOutline",), + "writer": ("ReportSectionDraft",), + "reviewer": ("ReviewDecision",), + "reviser": ("ReportSectionDraft",), +} + +QUALITY_GATES: tuple[str, ...] = ("cite_required", "dual_verification", "independent_review", "coverage_matrix") + +# ── 智能体输出契约 ───────────────────────────────────────────────────────── + + +class PlannerTurn(BaseModel): + """规划师一轮对话的完整产出：自由文案 + 结构化状态。""" + + model_config = ConfigDict(extra="ignore") + + action: str = "clarify" + reply: str = "" + questions: list[str] = Field(default_factory=list) + impact: list[str] = Field(default_factory=list) + assumptions: list[str] = Field(default_factory=list) + requirement: dict[str, Any] = Field(default_factory=dict) + + +class StrategyNodeDraft(BaseModel): + model_config = ConfigDict(extra="ignore") + + slot_id: str + role_key: str + display_name: str + angle: str | None = None + mission: str + depends_on: list[str] = Field(default_factory=list) + output_artifact_type: str + suggested_tools: list[str] = Field(default_factory=list) + + +class StrategyDraft(BaseModel): + model_config = ConfigDict(extra="ignore") + + reply: str = "" + strategy: str = "custom" + title: str + summary: str + rationale: str + estimated_duration_seconds: int = 900 + estimated_cost_level: str = "medium" + quality_gate_ids: list[str] = Field(default_factory=lambda: ["cite_required"]) + nodes: list[StrategyNodeDraft] = Field(default_factory=list) + + +LlmFactory = Callable[[], Any] + + +def planner_model_snapshot(app_config: AppConfig | None = None) -> RoleModelSnapshot: + """规划师身份沿用角色解析链：coordinator_model → planner_model → 默认模型。""" + + return resolve_role_model("coordinator", app_config=app_config or get_app_config()) + + +def build_planner_llm(app_config: AppConfig | None = None) -> Any: + from deerflow.models.factory import create_chat_model + + snapshot = planner_model_snapshot(app_config) + return create_chat_model(snapshot.config_name, app_config=app_config or get_app_config()) + + +_THINK_BLOCK_RE = re.compile(r".*?", re.DOTALL | re.IGNORECASE) + + +def _extract_json(text: str) -> dict[str, Any] | None: + if not text: + return None + # 思考模型（MiniMax-M2.7 等）回复可能带 …；JSON 在思考块之后。 + # 先剥思考块提取；剥完没有再回退用全文（极端情况 JSON 混在思考内）。 + for candidate in (_THINK_BLOCK_RE.sub("", text).strip(), text.strip()): + if not candidate: + continue + cleaned = candidate + if cleaned.startswith("```"): + cleaned = cleaned.strip("`") + if cleaned.lower().startswith("json"): + cleaned = cleaned[4:] + cleaned = cleaned.strip() + start = cleaned.find("{") + end = cleaned.rfind("}") + if start < 0 or end <= start: + continue + try: + parsed = json.loads(cleaned[start : end + 1]) + except Exception: + continue + if isinstance(parsed, dict): + return parsed + return None + + +async def _invoke_json(llm: Any, messages: list[Any]) -> tuple[dict[str, Any] | None, str]: + """调用模型并提取 JSON。返回 (解析结果, 原始文本)；解析失败结果为 None。""" + + try: + response = await llm.ainvoke(messages) + raw = getattr(response, "content", response) + if isinstance(raw, list): + raw = "".join(str(getattr(block, "text", block) if not isinstance(block, str) else block) for block in raw) + raw = str(raw) + return _extract_json(raw), raw + except Exception as exc: # noqa: BLE001 — 智能体故障必须显式失败 + logger.warning("rc planner agent llm call failed: %s: %s", exc.__class__.__name__, exc) + return None, "" + + +async def _invoke_json_with_retry(llm: Any, system: str, user_prompt: str) -> dict[str, Any] | None: + """带一次格式重试：模型输出不含 JSON 时，附原始输出要求只出 JSON 再问一次。""" + + parsed, raw = await _invoke_json(llm, [("system", system), ("human", user_prompt)]) + if parsed is None: + logger.warning("rc planner reply without JSON (raw head: %.200s)", raw[:200]) + retry_prompt = ( + user_prompt + + "\n\n【格式提醒】你上一轮的输出没有包含合法 JSON。这次必须只输出一个 JSON 对象，" + + f"不要在 JSON 之外输出任何文字。你上一轮的输出是：{raw[:500]}" + ) + parsed, _ = await _invoke_json(llm, [("system", system), ("human", retry_prompt)]) + return parsed + + +# ── 规划师智能体 ─────────────────────────────────────────────────────────── + + +class PlannerAgent: + """需求对话与工作流设计的唯一主路径（真智能体）。""" + + def __init__(self, llm: Any, *, app_config: AppConfig | None = None) -> None: + self._llm = llm + self._app_config = app_config + + # ── 需求对话 ─────────────────────────────────────────────────────────── + + async def converse( + self, + *, + user_text: str, + history: list[dict[str, str]], + previous_snapshot: dict[str, Any] | None, + previous_revision: int, + clarification_round: int, + max_rounds: int, + ) -> PlannerTurn | None: + """一轮规划对话。返回 None 表示智能体不可用（调用方走最后防线）。""" + + forced = clarification_round >= max_rounds + system = _CONVERSE_SYSTEM.format( + roles=", ".join(MEMBER_ROLES), + forced_note=( + "\n\n【重要】澄清轮次已达上限，本轮你必须判定需求就绪（action=ready）：" + "自然地告诉用户你将按当前理解开始规划，把未能确认的点写进 assumptions。" + if forced + else "" + ), + ) + known = {k: v for k, v in (previous_snapshot or {}).items() if v not in (None, "", [])} + known.pop("revision", None) + history_text = "\n".join( + f"{'用户' if item.get('role') == 'human' else '规划师'}: {str(item.get('content') or '')[:400]}" + for item in history[-10:] + ) + user_prompt = ( + f"当前已确认的需求快照（JSON）：{json.dumps(known, ensure_ascii=False)}\n\n" + f"最近的对话：\n{history_text or '（无）'}\n\n" + f"用户最新消息：{user_text}" + ) + parsed = await _invoke_json_with_retry(self._llm, system, user_prompt) + if parsed is None: + return None + try: + turn = PlannerTurn.model_validate(parsed) + except ValidationError: + logger.warning("rc planner turn invalid: %s", str(parsed)[:200]) + return None + if not (turn.reply or "").strip(): + return None + if forced and turn.action != "ready": + turn.action = "ready" + if turn.action == "ready" and not turn.assumptions: + turn.assumptions = ["按对话中已确认的信息直接规划"] + return turn + + def merge_requirement( + self, + turn: PlannerTurn, + *, + previous_snapshot: dict[str, Any] | None, + previous_revision: int, + ) -> ReportRequirementSnapshot: + """把智能体本轮带回的需求字段合并进既有快照（revision 语义同规则路由）。""" + + known = {k: v for k, v in (previous_snapshot or {}).items() if k != "revision"} + # LLM 契约是「只填能确认的字段」；再过滤一次空值，避免空串覆盖已确认信息。 + patch = { + k: v + for k, v in (turn.requirement or {}).items() + if k != "revision" and v not in (None, "", []) + } + revision = ( + max(1, int(previous_revision) + 1) + if (previous_snapshot or patch) + else max(1, int(previous_revision) or 1) + ) + if not previous_snapshot: + revision = 1 + merged = _merge_snapshot(known, patch, revision=revision) + if turn.action == "ready": + merged = apply_default_assumptions(merged)[0] + return merged + + # ── 工作流设计 ───────────────────────────────────────────────────────── + + async def design_workflow( + self, + *, + requirement: ReportRequirementSnapshot, + history: list[dict[str, str]], + max_team_members: int = 8, + ) -> tuple[StrategyRoleSet, str] | None: + """带着完整对话上下文设计工作流；返回 (策略, 展示文案) 或 None。""" + + fields = {k: v for k, v in requirement.model_dump().items() if v not in (None, "", [])} + fields.pop("revision", None) + history_text = "\n".join( + f"{'用户' if item.get('role') == 'human' else '规划师'}: {str(item.get('content') or '')[:300]}" + for item in history[-12:] + ) + system = _DESIGN_SYSTEM.format( + roles=", ".join(MEMBER_ROLES), + artifacts="\n".join(f" - {role}: {' / '.join(arts)}" for role, arts in ROLE_ARTIFACTS.items()), + gates=", ".join(QUALITY_GATES), + max_nodes=min(max_team_members, 8), + ) + user_prompt = ( + f"需求快照（JSON）：{json.dumps(fields, ensure_ascii=False)}\n\n" + f"需求确认阶段的对话（你与用户）：\n{history_text or '（无）'}\n\n请设计工作流。" + ) + parsed = await _invoke_json_with_retry(self._llm, system, user_prompt) + if parsed is None: + return None + try: + draft = StrategyDraft.model_validate(parsed) + except ValidationError: + logger.warning("rc planner strategy draft invalid: %s", str(parsed)[:200]) + return None + strategy = _validate_strategy_dag(draft) + if strategy is None: + logger.warning("rc planner strategy failed vocabulary/DAG validation: %s", str(parsed)[:200]) + return None + reply = (draft.reply or "").strip() or f"已按需求设计好协作工作流「{draft.title}」。" + return strategy, reply + + def workflow_reply_fallback(self, strategy: StrategyRoleSet) -> str: + return ( + f"我按确认的需求设计了「{strategy.title}」：{strategy.summary}\n" + "可以直接采用开始执行；想调整就告诉我（例如「加一条竞品对比线」）。" + ) + + +# ── 提示词 ───────────────────────────────────────────────────────────────── + +_CONVERSE_SYSTEM = """你是「报告协作工作台」的规划师智能体。用户想让你组织一个多智能体团队协作产出一份报告，你的任务是先通过自然对话把需求弄清楚，然后（由你自己）设计协作工作流。 + +你的对话风格： +- 像一位资深顾问：先回应用户刚说的话，再追问。回复里**不要**出现「收到您的报告需求」这类机械开场——用你自己的话自然表达。 +- 每轮最多追问 2~3 个真正影响方案设计的问题：关注对象与范围、给谁看、必须覆盖的角度、篇幅与深度、时间范围、特殊约束。用户已说清的绝不重复问。 +- 用户回复后先简短确认你的理解（一两句），再决定是否继续问。 +- 信息足够时，自然地总结你对需求的把握并判定就绪（action=ready），告诉用户下一步你会为他们设计工作流。 + +可用的团队角色（供你构思，不必对用户罗列）：{roles}{forced_note} + +每轮严格输出一个 JSON 对象（不要输出 JSON 以外的文字）： +{{ + "reply": "给用户看的完整回复（自然对话，直接展示）", + "action": "clarify" 或 "ready", + "questions": ["追问的问题（与 reply 一致，供系统记录）"], + "impact": ["为什么需要问（与 questions 对应）"], + "assumptions": ["ready 时：你采用的默认假设"], + "requirement": {{"topic": "...", "objective": "...", "audience": "...", "required_angles": ["..."], "time_range": "...", "length_target": "...", "tone": "..."}} +}} + +requirement 只填你从对话中能确认的字段（含此前轮次已确认的），其余省略。reply 用中文。""" + +_DESIGN_SYSTEM = """你是「报告协作工作台」的规划师智能体。需求已经与你确认完毕，现在请设计多智能体协作工作流。 + +词表约束（结构化部分必须严格遵守）： +- role_key 只能取：{roles} +- 各角色的 output_artifact_type 只能取： +{artifacts} +- quality_gate_ids 只能取：{gates} + +结构要求： +- 2~{max_nodes} 个节点；至少 1 个 writer；正式报告建议加 reviewer。 +- 研究线（researcher）按角度并行；research → verify → analyze → write → review 是常见主干，可按需求裁剪。 +- depends_on 引用其他节点的 slot_id；禁止环；至少一个节点 depends_on 为空（起点）。 +- slot_id 用简短英文（如 research-market / write-main）。mission 写清楚该节点做什么、产出什么。 + +严格输出 JSON（不要输出 JSON 以外的文字）： +{{ + "reply": "给用户看的介绍：说明你为什么这样设计（对应需求的哪些要求），一两句即可", + "strategy": "custom", + "title": "方案标题", + "summary": "一句话概述团队构成与链路", + "rationale": "设计理由", + "estimated_duration_seconds": 1500, + "estimated_cost_level": "low|medium|high", + "quality_gate_ids": ["cite_required"], + "nodes": [ + {{"slot_id": "research-market", "role_key": "researcher", "display_name": "研究员", "angle": "市场份额", "mission": "...", "depends_on": [], "output_artifact_type": "EvidenceBundle", "suggested_tools": ["web_search"]}} + ] +}}""" + + +# ── 工作流草稿校验（词表 + DAG） ────────────────────────────────────────── + + +def _validate_strategy_dag(draft: StrategyDraft) -> StrategyRoleSet | None: + nodes = draft.nodes + if not 2 <= len(nodes) <= 8: + return None + slot_ids = {node.slot_id for node in nodes} + if len(slot_ids) != len(nodes): + return None + if not any(node.role_key == "writer" for node in nodes): + return None + for node in nodes: + if node.role_key not in MEMBER_ROLES: + return None + if node.output_artifact_type not in ROLE_ARTIFACTS.get(node.role_key, ()): + return None + for dep in node.depends_on: + if dep not in slot_ids or dep == node.slot_id: + return None + indegree = {node.slot_id: len(node.depends_on) for node in nodes} + children: dict[str, list[str]] = {node.slot_id: [] for node in nodes} + for node in nodes: + for dep in node.depends_on: + children[dep].append(node.slot_id) + queue = [slot for slot, degree in indegree.items() if degree == 0] + seen = 0 + while queue: + current = queue.pop() + seen += 1 + for child in children[current]: + indegree[child] -= 1 + if indegree[child] == 0: + queue.append(child) + if seen != len(nodes): + return None + + gates = [gate for gate in draft.quality_gate_ids if gate in QUALITY_GATES] or ["cite_required"] + return StrategyRoleSet( + strategy="custom" if draft.strategy == "custom" else f"custom_{draft.strategy[:24]}", + title=draft.title[:80], + summary=draft.summary[:200], + rationale=draft.rationale[:300], + estimated_duration_seconds=max(300, min(int(draft.estimated_duration_seconds or 1500), 7200)), + estimated_cost_level=draft.estimated_cost_level if draft.estimated_cost_level in {"low", "medium", "high"} else "medium", + recommended=True, + slots=[ + RoleSlot( + slot_id=node.slot_id, + demand=RoleDemand( + role_key=node.role_key, + display_name=node.display_name[:32] or node.role_key, + angle=(node.angle or None), + mission=node.mission[:400], + depends_on_roles=[], + output_artifact_type=node.output_artifact_type, + suggested_tools=[tool for tool in node.suggested_tools if tool][:4], + ), + depends_on=list(node.depends_on), + ) + for node in nodes + ], + quality_gate_ids=gates, + ) diff --git a/offline-backend-20260512/backend/app/report_collaboration/planning/service.py b/offline-backend-20260512/backend/app/report_collaboration/planning/service.py index 62f82a20..ea50239e 100644 --- a/offline-backend-20260512/backend/app/report_collaboration/planning/service.py +++ b/offline-backend-20260512/backend/app/report_collaboration/planning/service.py @@ -1,17 +1,18 @@ """Fulfill a durable plan_request: catalog → select → assemble → validate → persist. -Does not start a run. Rule-based until the BE-7 model adapter is wired. +Does not start a run. P1.2 起优先由规划师 LLM 生成定制策略，模板为回退。 """ from __future__ import annotations +import logging from typing import Any from uuid import uuid4 from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot from app.report_collaboration.planning.assembler import PlanAssembler from app.report_collaboration.planning.catalog import project_agent_catalog -from app.report_collaboration.planning.controller import PlanController +from app.report_collaboration.planning.planner_agent import LlmFactory from app.report_collaboration.planning.validator import apply_validation, validate_assembled_plan from app.report_collaboration.requirement.defaults import COORDINATOR_NAME, DEFAULT_TITLE from deerflow.config.app_config import get_app_config @@ -22,6 +23,8 @@ from deerflow.persistence.report_collaboration import ( ReportCollaborationValidationError, ) +logger = logging.getLogger(__name__) + _BUSY = frozenset({"running", "awaiting_input", "reviewing", "completed", "failed", "cancelled"}) @@ -51,12 +54,15 @@ class PlanProposalService: *, agent_store: Any | None = None, config: ReportCollaborationConfig | None = None, + llm_factory: LlmFactory | None = None, ) -> None: self._store = store self._agent_store = agent_store self._config = config - self._controller = PlanController() self._assembler = PlanAssembler() + self._llm_factory = llm_factory + self._llm: Any | None = None + self._llm_tried = False def _cfg(self) -> ReportCollaborationConfig: return self._config or get_app_config().report_collaboration @@ -110,12 +116,29 @@ class PlanProposalService: requirement = requirement_from_session(record) cfg = self._cfg() catalog = await project_agent_catalog(user_id=user_id, config=cfg, agent_store=self._agent_store) - strategies = self._controller.propose( - requirement, - max_parallel_tasks=cfg.max_parallel_tasks, + group_id = f"rpg_{uuid4().hex[:12]}" + + # P1-R：规划师智能体是唯一路径 —— 带着完整对话上下文设计工作流。 + # 模型不可用即无法继续（无模板兜底）。 + from app.report_collaboration.planning.planner_agent import PlannerAgent + + llm = self._planner_llm() + history = [ + {"role": "human" if item.get("role") == "human" else "ai", "content": str(item.get("content") or "")} + for item in await self._store.list_messages(session_id) + ] + designed = await PlannerAgent(llm).design_workflow( + requirement=requirement, + history=history, max_team_members=cfg.max_team_members, ) - group_id = f"rpg_{uuid4().hex[:12]}" + if designed is None: + raise ReportCollaborationValidationError( + "PLANNER_UNAVAILABLE", "规划师智能体未能生成有效工作流，请重试或换个说法再试" + ) + strategies: list[Any] = [designed[0]] + planner_reply: str | None = designed[1] + assembled_list = [ self._assembler.assemble( strategy, @@ -140,13 +163,44 @@ class PlanProposalService: for candidate in persisted: saved.append(await self._store.insert_plan(session_id, candidate.model_dump())) await self._store.update_command_status(command_id, status="completed") + content = planner_reply or "工作流已生成，可开始执行。" await self._store.create_message( session_id, role="ai", - content="已根据当前需求生成 3 个候选协作方案。请选择其中一个后再开始执行；选择方案不会自动开跑。", + content=content, idempotency_key=f"{idempotency_key}:plans", name=COORDINATOR_NAME, - metadata={"plan_request_command_id": command_id, "proposal_group_id": group_id}, + metadata={"plan_request_command_id": command_id, "proposal_group_id": group_id, "plan_source": "planner_agent"}, completed=True, ) return saved + + def _planner_llm(self) -> Any: + """规划师模型：注入优先（测试）；否则按 planner_llm_enabled 懒构建。不可用即抛错。""" + + if self._llm_factory is not None: + llm = self._llm_factory() + if llm is None: + raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置") + return llm + try: + if not (self._config or get_app_config().report_collaboration).planner_llm_enabled: + raise ReportCollaborationValidationError( + "PLANNER_UNAVAILABLE", "规划师智能体未启用（report_collaboration.planner_llm_enabled）" + ) + except ReportCollaborationValidationError: + raise + except Exception as exc: + raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", f"规划师智能体配置不可读：{exc}") from None + if not self._llm_tried: + self._llm_tried = True + try: + from app.report_collaboration.planning.planner_agent import build_planner_llm + + self._llm = build_planner_llm() + except Exception as exc: + logger.warning("rc planner agent llm build failed: %s", exc) + self._llm = None + if self._llm is None: + raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置") + return self._llm diff --git a/offline-backend-20260512/backend/app/report_collaboration/requirement/__init__.py b/offline-backend-20260512/backend/app/report_collaboration/requirement/__init__.py index d1e29fb8..4e7b840d 100644 --- a/offline-backend-20260512/backend/app/report_collaboration/requirement/__init__.py +++ b/offline-backend-20260512/backend/app/report_collaboration/requirement/__init__.py @@ -1,6 +1,6 @@ """Pre-plan requirement resolution package.""" -from app.report_collaboration.requirement.intent_router import PrePlanDecision, route_pre_plan_turn +from app.report_collaboration.requirement.intent_router import PrePlanDecision from app.report_collaboration.requirement.resolver import RequirementResolver -__all__ = ["PrePlanDecision", "RequirementResolver", "route_pre_plan_turn"] +__all__ = ["PrePlanDecision", "RequirementResolver"] diff --git a/offline-backend-20260512/backend/app/report_collaboration/requirement/intent_router.py b/offline-backend-20260512/backend/app/report_collaboration/requirement/intent_router.py index d8ff1275..8088f786 100644 --- a/offline-backend-20260512/backend/app/report_collaboration/requirement/intent_router.py +++ b/offline-backend-20260512/backend/app/report_collaboration/requirement/intent_router.py @@ -1,4 +1,9 @@ -"""Pre-plan IntentRouterAgent: understand the user turn, never dispatch work.""" +"""Pre-plan 数据结构与快照合并工具。 + +P1-R 之后需求确认完全由规划师智能体（``planning/planner_agent.py``）驱动； +本模块只保留被复用的数据结构与合并工具。原规则路由 ``route_pre_plan_turn`` +已按设计 v2 移除 —— 模型不可用即报错，不再有规则兜底。 +""" from __future__ import annotations @@ -7,12 +12,11 @@ from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot -from app.report_collaboration.requirement.defaults import DEFAULT_ANGLES, DEFAULT_AUDIENCE, DEFAULT_LANGUAGE, DEFAULT_SOURCE_CONSTRAINTS, DEFAULT_TONE, PLAN_AFFECTING_GAPS -from app.report_collaboration.requirement.extractor import extract_requirement_patch, wants_direct_start +from app.report_collaboration.requirement.defaults import DEFAULT_ANGLES, DEFAULT_AUDIENCE, DEFAULT_LANGUAGE, DEFAULT_SOURCE_CONSTRAINTS, DEFAULT_TONE class PrePlanDecision(BaseModel): - """Structured output of the pre-plan router. It never builds a graph or starts a run.""" + """规划师智能体一轮决策。It never builds a graph or starts a run.""" model_config = ConfigDict(extra="forbid") @@ -24,11 +28,13 @@ class PrePlanDecision(BaseModel): used_defaults: bool = False reason: str confidence: float = Field(ge=0, le=1) + #: 规划师智能体产出的完整对话文案，直接落库展示。 + reply: str | None = None def _merge_snapshot(previous: dict[str, Any] | None, patch: dict[str, Any], *, revision: int) -> ReportRequirementSnapshot: base = dict(previous or {}) - base.update(patch) + base.update(patch or {}) topic = str(base.get("topic") or "").strip() return ReportRequirementSnapshot( topic=topic or "未指定主题", @@ -49,21 +55,6 @@ def _merge_snapshot(previous: dict[str, Any] | None, patch: dict[str, Any], *, r ) -def _gaps(requirement: ReportRequirementSnapshot) -> list[tuple[str, str, str]]: - missing: list[tuple[str, str, str]] = [] - if not requirement.topic or requirement.topic == "未指定主题": - missing.append(("topic", "这份报告要写什么主题？请用一两句话说明对象、时间和目的。", "没有主题无法选择方案")) - for field, question, impact in PLAN_AFFECTING_GAPS: - value = getattr(requirement, field, None) - if field == "required_angles": - empty = not value - else: - empty = not (value or "").strip() - if empty: - missing.append((field, question, impact)) - return missing - - def apply_default_assumptions(requirement: ReportRequirementSnapshot) -> tuple[ReportRequirementSnapshot, list[str]]: assumptions: list[str] = [] data = requirement.model_dump() @@ -82,62 +73,3 @@ def apply_default_assumptions(requirement: ReportRequirementSnapshot) -> tuple[R data["tone"] = DEFAULT_TONE assumptions.append(DEFAULT_TONE) return ReportRequirementSnapshot.model_validate(data), assumptions - - -def route_pre_plan_turn( - text: str, - *, - previous_snapshot: dict[str, Any] | None, - previous_revision: int, - clarification_round: int, - max_clarification_rounds: int, -) -> PrePlanDecision: - """Rule-based pre-plan routing. Does not call tools or create plans.""" - - patch = extract_requirement_patch(text, previous=previous_snapshot) - revision = max(1, int(previous_revision) + 1) if previous_snapshot or patch else max(1, int(previous_revision) or 1) - if not previous_snapshot: - revision = 1 - requirement = _merge_snapshot(previous_snapshot, patch, revision=revision) - skip = wants_direct_start(text) - hit_cap = clarification_round >= max_clarification_rounds - gaps = _gaps(requirement) - - if requirement.topic == "未指定主题" and skip: - return PrePlanDecision( - action="clarify", - requirement=requirement, - questions=["这份报告要写什么主题？请用一两句话说明对象、时间和目的。"], - impact=["没有主题无法选择方案"], - reason="用户要求直接开始，但还没有主题", - confidence=0.9, - ) - - if skip or (hit_cap and gaps): - filled, assumptions = apply_default_assumptions(requirement) - return PrePlanDecision( - action="ready", - requirement=filled, - assumptions=assumptions, - used_defaults=True, - reason="用户要求按默认假设继续" if skip else "已达澄清上限，使用明确假设继续", - confidence=0.86 if skip else 0.7, - ) - - if gaps: - return PrePlanDecision( - action="clarify", - requirement=requirement, - questions=[item[1] for item in gaps], - impact=[item[2] for item in gaps], - reason="缺少真正影响方案选择的信息", - confidence=0.8, - ) - - return PrePlanDecision( - action="ready", - requirement=requirement, - reason="必要信息已齐，可以进入规划", - confidence=0.84, - used_defaults=False, - ) diff --git a/offline-backend-20260512/backend/app/report_collaboration/requirement/resolver.py b/offline-backend-20260512/backend/app/report_collaboration/requirement/resolver.py index fbb89aaf..efca78e1 100644 --- a/offline-backend-20260512/backend/app/report_collaboration/requirement/resolver.py +++ b/offline-backend-20260512/backend/app/report_collaboration/requirement/resolver.py @@ -1,18 +1,39 @@ -"""Validate and persist requirement revisions. Never starts planning or a run.""" +"""Requirement resolver: the planner agent is the only path. + +模型不可用即无法继续 —— 直接抛 ``PLANNER_UNAVAILABLE``，没有任何规则/ +模板兜底（设计 v2：需求确认必须由真智能体完成）。 +""" from __future__ import annotations +import logging from typing import Any from uuid import uuid4 from app.report_collaboration.contracts.clarifications import ClarificationRecord -from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot from app.report_collaboration.requirement.defaults import COORDINATOR_NAME, DEFAULT_TITLE -from app.report_collaboration.requirement.intent_router import PrePlanDecision, route_pre_plan_turn +from app.report_collaboration.requirement.intent_router import PrePlanDecision from deerflow.config.app_config import get_app_config -from deerflow.persistence.report_collaboration import ReportCollaborationStore +from deerflow.persistence.report_collaboration import ( + ReportCollaborationStore, + ReportCollaborationValidationError, +) from deerflow.persistence.report_collaboration.codec import iso_now +logger = logging.getLogger(__name__) + +# 仅供类型注解；运行时惰性导入避免 requirement ↔ planning 循环依赖。 +LlmFactory = Any + + +def _title_from_topic(topic: str, current: str | None) -> str | None: + if current and current != DEFAULT_TITLE: + return None + cleaned = (topic or "").strip() + if not cleaned or cleaned == "未指定主题": + return None + return cleaned[:40] + def _state_from_session(record: dict[str, Any]) -> dict[str, Any]: raw = record.get("requirement_json") @@ -35,58 +56,56 @@ def _state_from_session(record: dict[str, Any]) -> dict[str, Any]: return {"snapshot": None, "clarifications": [], "assumptions": [], "clarification_round": 0, "ready_to_plan": False} -def _dump_state( - *, - snapshot: ReportRequirementSnapshot, - clarifications: list[dict[str, Any]], - assumptions: list[str], - clarification_round: int, - ready_to_plan: bool, -) -> dict[str, Any]: - return { - "snapshot": snapshot.model_dump(), - "clarifications": clarifications, - "assumptions": assumptions, - "clarification_round": clarification_round, - "ready_to_plan": ready_to_plan, - } - - -def format_clarification_message(questions: list[str], *, allow_skip: bool = True) -> str: - numbered = "\n".join(f"{index}. {question}" for index, question in enumerate(questions, start=1)) - suffix = "\n\n也可以直接回复「直接开始」，我会按默认假设记录需求。澄清完成不会自动生成方案。" if allow_skip else "" - return f"收到您的报告需求。为了给出最合适的协作方案，请先确认：\n\n{numbered}{suffix}" - - -def format_ready_message(decision: PrePlanDecision) -> str: - req = decision.requirement - angles = "、".join(req.required_angles) if req.required_angles else "（未指定）" - lines = [ - f"已写入需求版本 v{req.revision}。", - f"主题：{req.topic}", - f"读者：{req.audience or '未指定'}", - f"角度：{angles}", - ] - if decision.assumptions: - lines.append("默认假设：" + "；".join(decision.assumptions) + "。") - lines.append("澄清完成不等于已经生成候选方案，需要显式发起规划后才会出现方案卡。") - return "\n".join(lines) - - -def _title_from_topic(topic: str, current: str | None) -> str | None: - if current and current != DEFAULT_TITLE: - return None - cleaned = (topic or "").strip() - if not cleaned or cleaned == "未指定主题": - return None - return cleaned[:40] - - class RequirementResolver: """Writes requirement revisions and clarification records. Does not enqueue plans.""" - def __init__(self, store: ReportCollaborationStore) -> None: + def __init__( + self, + store: ReportCollaborationStore, + *, + planner_llm_factory: LlmFactory | None = None, + config: Any | None = None, + ) -> None: self._store = store + self._planner_llm_factory = planner_llm_factory + self._config = config + self._default_llm: Any = None + self._default_llm_tried = False + + def _planner_llm(self) -> Any: + """规划师模型：注入优先（测试）；否则按 planner_llm_enabled 懒构建。不可用即抛错。""" + + if self._planner_llm_factory is not None: + llm = self._planner_llm_factory() + if llm is None: + raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置") + return llm + try: + from deerflow.config.report_collaboration_config import ReportCollaborationConfig + + cfg: Any = self._config + if cfg is None: + cfg = get_app_config().report_collaboration + if not ReportCollaborationConfig.model_validate(cfg).planner_llm_enabled: + raise ReportCollaborationValidationError( + "PLANNER_UNAVAILABLE", "规划师智能体未启用（report_collaboration.planner_llm_enabled）" + ) + except ReportCollaborationValidationError: + raise + except Exception as exc: + raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", f"规划师智能体配置不可读：{exc}") from None + if not self._default_llm_tried: + self._default_llm_tried = True + try: + from app.report_collaboration.planning.planner_agent import build_planner_llm + + self._default_llm = build_planner_llm() + except Exception as exc: + logger.warning("rc planner agent llm build failed: %s", exc) + self._default_llm = None + if self._default_llm is None: + raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体不可用，请检查模型配置") + return self._default_llm async def handle_user_message( self, @@ -111,16 +130,67 @@ class RequirementResolver: cfg_rounds = max_clarification_rounds if cfg_rounds is None: cfg_rounds = int(get_app_config().report_collaboration.max_clarification_rounds) - decision = route_pre_plan_turn( - str(user_message.get("content") or ""), - previous_snapshot=state["snapshot"], + + text = str(user_message.get("content") or "") + history = [ + {"role": "human" if item.get("role") == "human" else "ai", "content": str(item.get("content") or "")} + for item in messages + if item.get("message_id") != user_message["message_id"] + ] + + decision = await self._agent_decision( + text, + history=history, + state=state, previous_revision=int(record.get("requirement_revision") or 0), - clarification_round=int(state["clarification_round"] or 0), - max_clarification_rounds=cfg_rounds, + max_rounds=cfg_rounds, ) + if decision is None: + # 模型调用失败/输出不合法 —— 无法继续，直接报错（无兜底）。 + raise ReportCollaborationValidationError("PLANNER_UNAVAILABLE", "规划师智能体本轮未返回有效决策，请重试") await self._apply(session_id, record, state, decision, user_message=user_message, idempotency_key=idempotency_key) return decision + async def _agent_decision( + self, + text: str, + *, + history: list[dict[str, str]], + state: dict[str, Any], + previous_revision: int, + max_rounds: int, + ) -> PrePlanDecision | None: + llm = self._planner_llm() + from app.report_collaboration.planning.planner_agent import PlannerAgent + + agent = PlannerAgent(llm) + turn = await agent.converse( + user_text=text, + history=history, + previous_snapshot=state["snapshot"], + previous_revision=previous_revision, + clarification_round=int(state["clarification_round"] or 0), + max_rounds=max_rounds, + ) + if turn is None: + return None + requirement = agent.merge_requirement( + turn, + previous_snapshot=state["snapshot"], + previous_revision=previous_revision, + ) + return PrePlanDecision( + action="ready" if turn.action == "ready" else "clarify", + requirement=requirement, + questions=[item for item in turn.questions if item and item.strip()][:3], + impact=[item for item in turn.impact if item][:3], + assumptions=[item for item in turn.assumptions if item], + used_defaults=bool(turn.assumptions), + reason="planner_agent", + confidence=0.9, + reply=turn.reply.strip(), + ) + async def _apply( self, session_id: str, @@ -146,7 +216,7 @@ class RequirementResolver: created_at=iso_now(), ) clarifications.append(clarification.model_dump()) - content = format_clarification_message(decision.questions) + content = decision.reply or "" metadata = { "preplan_for": user_message["message_id"], "clarification": clarification.model_dump(), @@ -155,17 +225,22 @@ class RequirementResolver: else: for item in clarifications: item["resolved"] = True - content = format_ready_message(decision) - metadata = {"preplan_for": user_message["message_id"], "ready_to_plan": True} + content = decision.reply or "" + metadata = { + "preplan_for": user_message["message_id"], + "ready_to_plan": True, + # 结构化需求快照：前端据此渲染「需求摘要卡」（计划模式 UI）。 + "requirement": decision.requirement.model_dump(), + } ready = True - payload = _dump_state( - snapshot=decision.requirement, - clarifications=clarifications, - assumptions=list(decision.assumptions), - clarification_round=round_count, - ready_to_plan=ready, - ) + payload = { + "snapshot": decision.requirement.model_dump(), + "clarifications": clarifications, + "assumptions": list(decision.assumptions), + "clarification_round": round_count, + "ready_to_plan": ready, + } await self._store.save_requirement_state( session_id, requirement_json=payload, diff --git a/offline-backend-20260512/backend/config.yaml b/offline-backend-20260512/backend/config.yaml index 6654817b..b3fdab7e 100644 --- a/offline-backend-20260512/backend/config.yaml +++ b/offline-backend-20260512/backend/config.yaml @@ -748,6 +748,7 @@ workflows: report_collaboration: enabled: true worker_enabled: true + planner_llm_enabled: true planner_model: null coordinator_model: null research_model: null diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/report_collaboration_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/report_collaboration_config.py index e0f0af17..a4349211 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/report_collaboration_config.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/report_collaboration_config.py @@ -18,6 +18,10 @@ class ReportCollaborationConfig(BaseModel): default=False, description="When true, a dispatcher/worker may claim run leases. Gateway-only processes keep this false.", ) + planner_llm_enabled: bool = Field( + default=False, + description="P1: LLM planner conversation (clarify + custom strategy). Off keeps the deterministic rule router and template plans.", + ) planner_model: str | None = Field(default=None, description="Model name from AppConfig.models for planning") coordinator_model: str | None = Field(default=None, description="Model name from AppConfig.models for the constrained coordinator") research_model: str | None = Field(default=None, description="Model name from AppConfig.models for research members") diff --git a/offline-backend-20260512/backend/tests/test_report_collaboration_api.py b/offline-backend-20260512/backend/tests/test_report_collaboration_api.py index 439770a1..76de6304 100644 --- a/offline-backend-20260512/backend/tests/test_report_collaboration_api.py +++ b/offline-backend-20260512/backend/tests/test_report_collaboration_api.py @@ -25,6 +25,40 @@ class _DisabledCfg: report_collaboration = ReportCollaborationConfig(enabled=False) +class _PlannerFakeLLM: + """规划师假模型：需求确认回「读者」追问；工作流设计回单份定制方案。""" + + _converse = ( + '{"reply": "好的。先确认：这份报告的主要读者是谁？", "action": "clarify",' + '"questions": ["这份报告的主要读者是谁？"], "impact": ["读者影响研究线划分"],' + '"requirement": {"topic": "新能源汽车趋势报告"}}' + ) + _ready = ( + '{"reply": "明白了，读者为企业战略决策人员。我来设计工作流。", "action": "ready",' + '"assumptions": [], "requirement": {"topic": "新能源汽车趋势报告", "audience": "企业战略决策人员"}}' + ) + _design = ( + '{"reply": "已按需求设计定制工作流。", "strategy": "custom", "title": "定制方案",' + '"summary": "研究后写作", "rationale": "贴合需求", "estimated_duration_seconds": 900,' + '"estimated_cost_level": "low", "quality_gate_ids": ["cite_required"],' + '"nodes": [' + '{"slot_id": "research-1", "role_key": "researcher", "display_name": "研究员", "mission": "检索证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"},' + '{"slot_id": "write-1", "role_key": "writer", "display_name": "撰写员", "mission": "成稿", "depends_on": ["research-1"], "output_artifact_type": "ReportSectionDraft"}]}' + ) + + async def ainvoke(self, messages): + from types import SimpleNamespace + + prompt = str(messages[-1][1]) if messages else "" + if "请设计工作流" in prompt: + payload = self._design + elif "企业战略决策人员" in prompt or "读者" in prompt and "用户最新消息" in prompt: + payload = self._ready + else: + payload = self._converse + return SimpleNamespace(content=payload) + + def _client(monkeypatch: pytest.MonkeyPatch, *, enabled: bool = True, user: str | None = "user_a") -> tuple[TestClient, MemoryReportCollaborationStore]: monkeypatch.setattr(report_collaboration, "get_app_config", lambda: _Cfg() if enabled else _DisabledCfg()) app = FastAPI() @@ -32,6 +66,7 @@ def _client(monkeypatch: pytest.MonkeyPatch, *, enabled: bool = True, user: str inner = MemoryReportCollaborationStore() store = PublishingReportCollaborationStore(inner, hub) app.state.report_collaboration_store = store + app.state.report_collaboration_planner_llm = _PlannerFakeLLM app.state.report_collaboration_live_hub = hub app.state.report_collaboration_sse_settings = SseStreamSettings( deadline_seconds=0.6, @@ -126,14 +161,14 @@ def test_session_message_plan_run_http(monkeypatch: pytest.MonkeyPatch) -> None: proposed_body = proposed.json() assert proposed_body["session"]["status"] == "proposal_ready" plans = proposed_body["plans"] - assert {item["strategy"] for item in plans} == {"balanced_review", "parallel_depth", "focused_fast"} + assert {item["strategy"] for item in plans} == {"custom"} plan = next(item for item in plans if item.get("recommended")) assert plan["validation"]["ok"] is True assert all(node.get("output_artifact_type") and node.get("acceptance_criteria") for node in plan["nodes"]) listed = client.get(f"/api/report-collaboration/sessions/{session_id}/plans") assert listed.status_code == 200 - assert len(listed.json()["plans"]) == 3 + assert len(listed.json()["plans"]) == 1 selected = client.post( f"/api/report-collaboration/sessions/{session_id}/plans/{plan['id']}/select", diff --git a/offline-backend-20260512/backend/tests/test_report_collaboration_models.py b/offline-backend-20260512/backend/tests/test_report_collaboration_models.py index af550fcf..14cbe1c1 100644 --- a/offline-backend-20260512/backend/tests/test_report_collaboration_models.py +++ b/offline-backend-20260512/backend/tests/test_report_collaboration_models.py @@ -7,7 +7,7 @@ from types import SimpleNamespace import pytest from app.report_collaboration.agentscope_runtime.event_adapter import ConversationReplica, MemberStreamContext, SeqAllocator, make_emitter -from app.report_collaboration.agentscope_runtime.model_adapter import DeerFlowChatModelAdapter, ModelAdapterError, record_attempt_usage, redact_secrets +from app.report_collaboration.agentscope_runtime.model_adapter import DeerFlowChatModelAdapter, ModelAdapterError, estimate_tokens_for_text, record_attempt_usage, redact_secrets from app.report_collaboration.agentscope_runtime.model_resolver import freeze_plan_models, resolve_role_model from deerflow.config.app_config import AppConfig from deerflow.config.model_config import ModelConfig @@ -128,6 +128,33 @@ async def test_adapter_streams_text_tool_and_usage() -> None: assert frozen.value.code == "MODEL_FROZEN" +@pytest.mark.asyncio +async def test_adapter_count_tokens_and_context_size_contract() -> None: + """AgentScope compress_context 依赖 model.context_size 与 count_tokens(messages, tools)。""" + + snapshot = resolve_role_model("researcher", app_config=_app()) + adapter = DeerFlowChatModelAdapter(snapshot, app_config=_app()) + assert isinstance(adapter.context_size, int) and adapter.context_size > 0 + + class _Msg: + def __init__(self, text: str) -> None: + self.content = text + + def get_content_blocks(self): + return [{"type": "text", "text": self.content}] + + agentscope_style = [_Msg("市场分析" * 20)] + tokens_msg = await adapter.count_tokens(agentscope_style, None) + dict_style = [{"role": "user", "content": "hello world"}] + tokens_dict = await adapter.count_tokens(dict_style, None) + tools = [{"name": "web_search", "parameters": {"type": "object"}}] + tokens_tools = await adapter.count_tokens(dict_style, tools) + assert tokens_msg > 0 + assert tokens_dict == estimate_tokens_for_text("hello world") + assert tokens_tools > tokens_dict + assert await adapter.count_tokens([], None) == 0 + + @pytest.mark.asyncio async def test_adapter_redacts_provider_errors_and_records_metrics() -> None: snapshot = resolve_role_model("writer", app_config=_app()) diff --git a/offline-backend-20260512/backend/tests/test_report_collaboration_planner_agent.py b/offline-backend-20260512/backend/tests/test_report_collaboration_planner_agent.py new file mode 100644 index 00000000..dc692959 --- /dev/null +++ b/offline-backend-20260512/backend/tests/test_report_collaboration_planner_agent.py @@ -0,0 +1,181 @@ +"""P1-R 规划师智能体：真对话（文案由智能体产出直接落库）+ 工作流设计 + 降级路径。""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot +from app.report_collaboration.planning.planner_agent import PlannerAgent +from app.report_collaboration.requirement.resolver import RequirementResolver +from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore + + +class _FakeLLM: + def __init__(self, payload: str) -> None: + self._payload = payload + + async def ainvoke(self, messages): + return SimpleNamespace(content=self._payload) + + +_TURN_CLARIFY = """```json +{"reply": "好的，智能家居是个不错的方向。为了让团队分工更准，我想先确认两点：这份报告主要给谁看？大概需要多长的篇幅？", + "action": "clarify", + "questions": ["这份报告主要给谁看？", "大概需要多长的篇幅？"], + "impact": ["读者决定研究线划分", "篇幅决定成员数量"], + "requirement": {"topic": "智能家居设备2026年市场分析", "required_angles": ["市场规模", "竞争格局", "技术趋势"]}} +```""" + +_TURN_READY = """{"reply": "明白了：给公司管理层的战略规划参考，5000字左右，覆盖三个角度。我现在就可以为你们设计工作流了。", + "action": "ready", "assumptions": ["地域默认中国大陆"], + "requirement": {"topic": "智能家居设备2026年市场分析", "audience": "公司管理层", "required_angles": ["市场规模", "竞争格局", "技术趋势"], "length_target": "5000字"}}""" + +_DESIGN_JSON = """{"reply": "我按三个角度设计了两条并行研究线汇写再审的链路，正好对齐管理层决策的阅读习惯。", + "strategy": "custom", "title": "三角度并行研究工作流", "summary": "三线并行 + 汇写 + 独立审稿", + "rationale": "需求明确三个角度，管理层需要可信结论", "estimated_duration_seconds": 1800, + "estimated_cost_level": "medium", "quality_gate_ids": ["cite_required", "independent_review"], + "nodes": [ + {"slot_id": "research-scale", "role_key": "researcher", "display_name": "研究员", "angle": "市场规模", + "mission": "检索市场规模证据", "depends_on": [], "output_artifact_type": "EvidenceBundle", "suggested_tools": ["web_search"]}, + {"slot_id": "research-comp", "role_key": "researcher", "display_name": "研究员", "angle": "竞争格局", + "mission": "检索竞争格局证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"}, + {"slot_id": "research-tech", "role_key": "researcher", "display_name": "研究员", "angle": "技术趋势", + "mission": "检索技术趋势证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"}, + {"slot_id": "write-main", "role_key": "writer", "display_name": "撰写员", + "mission": "汇总三线证据成稿", "depends_on": ["research-scale", "research-comp", "research-tech"], "output_artifact_type": "ReportSectionDraft"}, + {"slot_id": "review-1", "role_key": "reviewer", "display_name": "审稿员", + "mission": "独立审稿", "depends_on": ["write-main"], "output_artifact_type": "ReviewDecision"} + ]}""" + + +# ── converse：智能体对话 ─────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_converse_returns_agent_reply_and_questions() -> None: + agent = PlannerAgent(_FakeLLM(_TURN_CLARIFY)) + turn = await agent.converse( + user_text="帮我写一份智能家居设备2026年市场分析报告", + history=[], + previous_snapshot=None, + previous_revision=0, + clarification_round=0, + max_rounds=3, + ) + assert turn is not None and turn.action == "clarify" + assert "为了让团队分工更准" in turn.reply + assert len(turn.questions) == 2 + merged = agent.merge_requirement(turn, previous_snapshot=None, previous_revision=0) + assert merged.topic == "智能家居设备2026年市场分析" + assert merged.revision == 1 + + +@pytest.mark.asyncio +async def test_converse_forced_ready_at_round_cap() -> None: + agent = PlannerAgent(_FakeLLM(_TURN_CLARIFY)) + turn = await agent.converse( + user_text="再说说", + history=[], + previous_snapshot={"topic": "智能家居市场", "revision": 2}, + previous_revision=2, + clarification_round=3, + max_rounds=3, + ) + assert turn is not None and turn.action == "ready" + assert turn.assumptions + + +@pytest.mark.asyncio +async def test_converse_invalid_json_returns_none() -> None: + agent = PlannerAgent(_FakeLLM("我拒绝输出 JSON")) + assert ( + await agent.converse( + user_text="写报告", + history=[], + previous_snapshot=None, + previous_revision=0, + clarification_round=0, + max_rounds=3, + ) + is None + ) + + +@pytest.mark.asyncio +async def test_resolver_persists_agent_reply_without_template_wrapper() -> None: + store = MemoryReportCollaborationStore() + session = await store.create_session(owner_id="u1", title="新会话", idempotency_key="ik1") + resolver = RequirementResolver(store, planner_llm_factory=lambda: _FakeLLM(_TURN_CLARIFY)) + decision = await resolver.handle_user_message( + session["id"], + user_message={"message_id": "m1", "role": "human", "content": "帮我写一份智能家居设备2026年市场分析报告"}, + idempotency_key="ik-m1", + ) + assert decision is not None and decision.reply and "为了让团队分工更准" in decision.reply + messages = await store.list_messages(session["id"]) + ai_messages = [item for item in messages if item["role"] == "ai"] + assert ai_messages, "agent reply message missing" + assert "为了让团队分工更准" in ai_messages[-1]["content"] + assert "收到您的报告需求" not in ai_messages[-1]["content"] # 模板文案不得出现 + + +@pytest.mark.asyncio +async def test_resolver_raises_when_agent_output_invalid() -> None: + """模型输出不合法 → 直接报错（无规则/模板兜底）。""" + + from deerflow.persistence.report_collaboration import ReportCollaborationValidationError + + store = MemoryReportCollaborationStore() + session = await store.create_session(owner_id="u1", title="新会话", idempotency_key="ik1") + resolver = RequirementResolver(store, planner_llm_factory=lambda: _FakeLLM("not json")) + with pytest.raises(ReportCollaborationValidationError) as exc: + await resolver.handle_user_message( + session["id"], + user_message={"message_id": "m1", "role": "human", "content": "帮我写一份智能家居设备2026年市场分析报告"}, + idempotency_key="ik-m1", + ) + assert exc.value.code == "PLANNER_UNAVAILABLE" + + +# ── design_workflow：工作流设计 ─────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_design_workflow_returns_strategy_with_context_reply() -> None: + requirement = ReportRequirementSnapshot( + topic="智能家居设备2026年市场分析", + required_angles=["市场规模", "竞争格局", "技术趋势"], + revision=2, + ) + history = [{"role": "human", "content": "给管理层看，覆盖三个角度"}] + designed = await PlannerAgent(_FakeLLM(_DESIGN_JSON)).design_workflow( + requirement=requirement, history=history + ) + assert designed is not None + strategy, reply = designed + assert strategy.strategy == "custom" + assert {slot.demand.role_key for slot in strategy.slots} == {"researcher", "writer", "reviewer"} + assert "管理层" in reply or "设计" in reply + write = next(slot for slot in strategy.slots if slot.slot_id == "write-main") + assert set(write.depends_on) == {"research-scale", "research-comp", "research-tech"} + + +@pytest.mark.asyncio +async def test_design_workflow_rejects_bad_vocabulary_and_cycles() -> None: + requirement = ReportRequirementSnapshot(topic="测试", revision=1) + agent = PlannerAgent(_FakeLLM(_DESIGN_JSON)) + + bad_role = _DESIGN_JSON.replace('"role_key": "writer"', '"role_key": "project_manager"') + assert await PlannerAgent(_FakeLLM(bad_role)).design_workflow(requirement=requirement, history=[]) is None + + cyclic = _DESIGN_JSON.replace( + '"depends_on": [], "output_artifact_type": "EvidenceBundle", "suggested_tools"', + '"depends_on": ["review-1"], "output_artifact_type": "EvidenceBundle", "suggested_tools"', + ) + assert await PlannerAgent(_FakeLLM(cyclic)).design_workflow(requirement=requirement, history=[]) is None + + bad_artifact = _DESIGN_JSON.replace('"output_artifact_type": "ReportSectionDraft"', '"output_artifact_type": "PDF"') + assert await PlannerAgent(_FakeLLM(bad_artifact)).design_workflow(requirement=requirement, history=[]) is None + _ = agent # 保持引用清晰 diff --git a/offline-backend-20260512/backend/tests/test_report_collaboration_planning.py b/offline-backend-20260512/backend/tests/test_report_collaboration_planning.py index 77d5af9f..381f4b5a 100644 --- a/offline-backend-20260512/backend/tests/test_report_collaboration_planning.py +++ b/offline-backend-20260512/backend/tests/test_report_collaboration_planning.py @@ -12,7 +12,7 @@ from app.report_collaboration.planning.selector import AgentSelector from app.report_collaboration.planning.service import PlanProposalService from app.report_collaboration.planning.validator import validate_assembled_plan from deerflow.config.report_collaboration_config import ReportCollaborationConfig -from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore +from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore, ReportCollaborationValidationError class _AgentStore: @@ -135,6 +135,26 @@ def test_validator_rejects_unknown_agent_id() -> None: asyncio.run(_run()) +class _DesignFakeLLM: + """规划师工作流设计假模型：返回单份合法定制方案。""" + + _payload = ( + '{"reply": "我按需求设计了两条研究线汇写再审的定制工作流。",' + '"strategy": "custom", "title": "定制方案", "summary": "两线并行汇写再审",' + '"rationale": "贴合需求", "estimated_duration_seconds": 1200,' + '"estimated_cost_level": "medium", "quality_gate_ids": ["cite_required"],' + '"nodes": [' + '{"slot_id": "research-a", "role_key": "researcher", "display_name": "研究员", "angle": "市场", "mission": "检索市场证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"},' + '{"slot_id": "write-main", "role_key": "writer", "display_name": "撰写员", "mission": "成稿", "depends_on": ["research-a"], "output_artifact_type": "ReportSectionDraft"}' + ']}' + ) + + async def ainvoke(self, messages): + from types import SimpleNamespace + + return SimpleNamespace(content=self._payload) + + @pytest.mark.asyncio async def test_service_writes_plans_without_starting_a_run() -> None: store = MemoryReportCollaborationStore() @@ -146,15 +166,20 @@ async def test_service_writes_plans_without_starting_a_run() -> None: status="clarifying", ) command = await store.enqueue_plan_request(session["id"], idempotency_key="ik-plan", expected_revision=1) - saved = await PlanProposalService(store).fulfill(session["id"], user_id="u1", command_id=command["command_id"], idempotency_key="ik-plan") - assert len(saved) == 3 + saved = await PlanProposalService(store, llm_factory=lambda: _DesignFakeLLM()).fulfill( + session["id"], user_id="u1", command_id=command["command_id"], idempotency_key="ik-plan" + ) + assert len(saved) == 1 # 智能体生成单份定制方案 snapshot = await store.get_snapshot(session["id"]) assert snapshot["session"]["status"] == "proposal_ready" assert snapshot["session"]["selected_plan_id"] is None assert snapshot["run"] is None - assert {item["strategy"] for item in snapshot["plans"]} == {"balanced_review", "parallel_depth", "focused_fast"} + assert snapshot["plans"][0]["strategy"] == "custom" assert all(item["requirement_revision"] == 1 for item in snapshot["plans"]) assert snapshot["session"]["active_run_id"] is None + ai = [item for item in snapshot["messages"] if item["role"] == "ai"][-1] + assert ai["metadata"]["plan_source"] == "planner_agent" + assert "定制工作流" in ai["content"] selected = await store.select_plan(session["id"], saved[0]["id"], idempotency_key="ik-select", expected_revision=1) after_select = await store.get_snapshot(session["id"]) @@ -163,6 +188,24 @@ async def test_service_writes_plans_without_starting_a_run() -> None: assert after_select["session"]["active_run_id"] is None +@pytest.mark.asyncio +async def test_service_raises_when_planner_unavailable() -> None: + store = MemoryReportCollaborationStore() + session = await store.create_session(owner_id="u1", title="", idempotency_key="s1b") + await store.save_requirement_state( + session["id"], + requirement_json={"snapshot": _requirement().model_dump(), "clarifications": [], "assumptions": [], "clarification_round": 1, "ready_to_plan": True}, + requirement_revision=1, + status="clarifying", + ) + command = await store.enqueue_plan_request(session["id"], idempotency_key="ik-plan2", expected_revision=1) + with pytest.raises(ReportCollaborationValidationError) as exc: + await PlanProposalService(store, llm_factory=lambda: None).fulfill( + session["id"], user_id="u1", command_id=command["command_id"], idempotency_key="ik-plan2" + ) + assert exc.value.code == "PLANNER_UNAVAILABLE" + + def test_cycle_is_rejected() -> None: cfg = ReportCollaborationConfig() candidate_nodes_cycle = AssembledPlan.model_validate( diff --git a/offline-backend-20260512/backend/tests/test_report_collaboration_requirement.py b/offline-backend-20260512/backend/tests/test_report_collaboration_requirement.py index c633a490..0de5dbba 100644 --- a/offline-backend-20260512/backend/tests/test_report_collaboration_requirement.py +++ b/offline-backend-20260512/backend/tests/test_report_collaboration_requirement.py @@ -1,76 +1,35 @@ -"""RC-BE-005: pre-plan intent router and requirement resolver.""" +"""P1-R 需求确认：规划师智能体是唯一路径（无规则/模板兜底）。""" from __future__ import annotations +from types import SimpleNamespace + import pytest from app.report_collaboration.requirement.extractor import extract_topic, wants_direct_start -from app.report_collaboration.requirement.intent_router import route_pre_plan_turn from app.report_collaboration.requirement.resolver import RequirementResolver -from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore +from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore, ReportCollaborationValidationError -def test_topic_only_asks_plan_affecting_gaps() -> None: - decision = route_pre_plan_turn( - "写一份新能源汽车主要品牌市场趋势分析报告", - previous_snapshot=None, - previous_revision=0, - clarification_round=0, - max_clarification_rounds=3, - ) - assert decision.action == "clarify" - assert "新能源" in decision.requirement.topic - assert decision.requirement.revision == 1 - assert len(decision.questions) >= 2 - assert any("读者" in item for item in decision.questions) - assert any("角度" in item for item in decision.questions) +class _ScriptedLLM: + """按顺序吐出预置 JSON 决策（模拟规划师智能体多轮）。""" + + def __init__(self, payloads: list[str]) -> None: + self._payloads = list(payloads) + + async def ainvoke(self, messages): + payload = self._payloads.pop(0) if self._payloads else self._payloads[0] + return SimpleNamespace(content=payload) -def test_direct_start_records_default_assumptions() -> None: - first = route_pre_plan_turn( - "撰写新能源汽车趋势报告，直接开始", - previous_snapshot=None, - previous_revision=0, - clarification_round=0, - max_clarification_rounds=3, - ) - assert first.action == "ready" - assert first.used_defaults is True - assert first.requirement.audience == "企业战略决策人员" - assert "市场份额" in first.requirement.required_angles - assert first.assumptions +_CLARIFY = """{"reply": "新能源汽车趋势是个好题目。先确认两个关键点：这份报告主要给谁看？希望覆盖哪些角度？", + "action": "clarify", "questions": ["这份报告主要给谁看？", "希望覆盖哪些角度？"], + "impact": ["读者影响研究线", "角度影响成员分工"], + "requirement": {"topic": "新能源汽车趋势报告"}}""" - -def test_direct_start_without_topic_still_asks_topic() -> None: - decision = route_pre_plan_turn( - "直接开始", - previous_snapshot=None, - previous_revision=0, - clarification_round=0, - max_clarification_rounds=3, - ) - assert decision.action == "clarify" - assert any("主题" in item for item in decision.questions) - - -def test_clarification_cap_uses_explicit_assumptions() -> None: - first = route_pre_plan_turn( - "写一份行业报告", - previous_snapshot=None, - previous_revision=0, - clarification_round=0, - max_clarification_rounds=1, - ) - assert first.action == "clarify" - capped = route_pre_plan_turn( - "先这样吧", - previous_snapshot=first.requirement.model_dump(), - previous_revision=first.requirement.revision, - clarification_round=1, - max_clarification_rounds=1, - ) - assert capped.action == "ready" - assert capped.used_defaults is True +_READY = """{"reply": "清楚了：给投资机构，覆盖市场份额、价格策略和政策环境。我来为你们设计工作流。", + "action": "ready", "assumptions": ["时间范围默认近两年"], + "requirement": {"topic": "新能源汽车趋势报告", "audience": "投资机构", "required_angles": ["市场份额", "价格策略", "政策环境"]}}""" def test_extract_helpers() -> None: @@ -79,14 +38,15 @@ def test_extract_helpers() -> None: @pytest.mark.asyncio -async def test_resolver_writes_clarification_without_planning() -> None: +async def test_resolver_clarify_round_persists_agent_reply_and_record() -> None: store = MemoryReportCollaborationStore() session = await store.create_session(owner_id="u1", title="", idempotency_key="s1") user = await store.create_message(session["id"], role="human", content="写一份新能源汽车趋势报告", idempotency_key="m1") - resolver = RequirementResolver(store) + llm = _ScriptedLLM([_CLARIFY]) + resolver = RequirementResolver(store, planner_llm_factory=lambda: llm) decision = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m1", max_clarification_rounds=3) - assert decision is not None - assert decision.action == "clarify" + assert decision is not None and decision.action == "clarify" + assert "先确认两个关键点" in (decision.reply or "") snapshot = await store.get_snapshot(session["id"]) assert snapshot["session"]["status"] == "clarifying" @@ -96,10 +56,14 @@ async def test_resolver_writes_clarification_without_planning() -> None: assert len(ai) == 1 assert ai[0]["completed_at"] assert ai[0]["name"] == "协调者" + # 智能体文案直接落库，模板开场不得出现 + assert "先确认两个关键点" in ai[0]["content"] + assert "收到您的报告需求" not in ai[0]["content"] record = await store.get_session_record(session["id"]) assert record["requirement_json"]["ready_to_plan"] is False assert record["requirement_json"]["clarifications"][0]["resolved"] is False + # 幂等：同一消息重放不产生第二条 AI 回复 replay = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m1", max_clarification_rounds=3) assert replay is None snapshot2 = await store.get_snapshot(session["id"]) @@ -107,52 +71,63 @@ async def test_resolver_writes_clarification_without_planning() -> None: @pytest.mark.asyncio -async def test_resolver_ready_is_not_plan_ready() -> None: +async def test_resolver_ready_writes_requirement_metadata_for_summary_card() -> None: store = MemoryReportCollaborationStore() session = await store.create_session(owner_id="u1", title="", idempotency_key="s2") user = await store.create_message( - session["id"], - role="human", - content="写一份新能源汽车趋势报告，面向战略决策层，必须覆盖市场、政策、海外和风险，直接开始", - idempotency_key="m2", + session["id"], role="human", content="写一份新能源汽车趋势报告，给投资机构看", idempotency_key="m2" ) - resolver = RequirementResolver(store) + llm = _ScriptedLLM([_READY]) + resolver = RequirementResolver(store, planner_llm_factory=lambda: llm) decision = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m2") - assert decision is not None - assert decision.action == "ready" + assert decision is not None and decision.action == "ready" snapshot = await store.get_snapshot(session["id"]) assert snapshot["session"]["status"] == "clarifying" - assert snapshot["plans"] == [] + assert snapshot["plans"] == [] # ready ≠ 已生成方案 + ai = [item for item in snapshot["messages"] if item["role"] == "ai"][-1] + assert ai["metadata"]["ready_to_plan"] is True + assert ai["metadata"]["requirement"]["audience"] == "投资机构" + assert "设计工作流" in ai["content"] record = await store.get_session_record(session["id"]) assert record["requirement_json"]["ready_to_plan"] is True - assert all(item["resolved"] for item in record["requirement_json"]["clarifications"]) - assert "不等于已经生成" in snapshot["messages"][-1]["content"] @pytest.mark.asyncio -async def test_answer_merges_and_resolves_clarification() -> None: +async def test_resolver_answer_merges_fields_across_rounds() -> None: store = MemoryReportCollaborationStore() session = await store.create_session(owner_id="u1", title="", idempotency_key="s3") first = await store.create_message(session["id"], role="human", content="写一份新能源汽车趋势报告", idempotency_key="m3a") - resolver = RequirementResolver(store) + llm = _ScriptedLLM([_CLARIFY, _READY]) + resolver = RequirementResolver(store, planner_llm_factory=lambda: llm) await resolver.handle_user_message(session["id"], user_message=first, idempotency_key="m3a") second = await store.create_message( - session["id"], - role="human", - content="读者是投资机构，必须覆盖市场份额、价格策略和政策环境", - idempotency_key="m3b", + session["id"], role="human", content="读者是投资机构，必须覆盖市场份额、价格策略和政策环境", idempotency_key="m3b" ) decision = await resolver.handle_user_message(session["id"], user_message=second, idempotency_key="m3b") - assert decision is not None - assert decision.action == "ready" - assert decision.used_defaults is False - assert decision.requirement.audience == "投资机构" + assert decision is not None and decision.action == "ready" assert decision.requirement.revision == 2 + assert decision.requirement.audience == "投资机构" + assert set(decision.requirement.required_angles) == {"市场份额", "价格策略", "政策环境"} record = await store.get_session_record(session["id"]) - assert record["requirement_revision"] == 2 assert all(item["resolved"] for item in record["requirement_json"]["clarifications"]) - snapshot = await store.get_snapshot(session["id"]) - assert snapshot["plans"] == [] - assert snapshot["session"]["status"] == "clarifying" + + +@pytest.mark.asyncio +async def test_resolver_raises_planner_unavailable_without_fallback() -> None: + store = MemoryReportCollaborationStore() + session = await store.create_session(owner_id="u1", title="", idempotency_key="s4") + user = await store.create_message(session["id"], role="human", content="写报告", idempotency_key="m4") + + # 工厂显式禁用（模型不可用）→ 直接报错，无规则兜底 + disabled = RequirementResolver(store, planner_llm_factory=lambda: None) + with pytest.raises(ReportCollaborationValidationError) as exc: + await disabled.handle_user_message(session["id"], user_message=user, idempotency_key="m4") + assert exc.value.code == "PLANNER_UNAVAILABLE" + + # 模型输出不合法 → 同样报错 + broken = RequirementResolver(store, planner_llm_factory=lambda: _ScriptedLLM(["我说人话不行吗"])) + with pytest.raises(ReportCollaborationValidationError) as exc2: + await broken.handle_user_message(session["id"], user_message=user, idempotency_key="m4") + assert exc2.value.code == "PLANNER_UNAVAILABLE" diff --git a/offline-backend-20260512/backend/tests/test_report_collaboration_runner.py b/offline-backend-20260512/backend/tests/test_report_collaboration_runner.py index 6b1cf3cc..89e64cfc 100644 --- a/offline-backend-20260512/backend/tests/test_report_collaboration_runner.py +++ b/offline-backend-20260512/backend/tests/test_report_collaboration_runner.py @@ -155,6 +155,89 @@ async def test_waves_researcher_before_writer_and_text_does_not_complete() -> No assert phases[f"{run_id}-research-attempt1"] != phases[f"{run_id}-write-attempt1"] +@pytest.mark.asyncio +async def test_command_rework_cascades_downstream_and_reruns_in_order() -> None: + """P0.3 指定重跑级联：重跑 research 后 write 被级联 superseded 并按依赖序重跑。""" + + store = MemoryReportCollaborationStore() + run_id = await _boot(store) + kernel = ScriptedMemberKernel() + kernel.enqueue("research", TaskResult(kind="artifact_draft", artifact_type="EvidenceBundle", artifact_draft=_evidence())) + kernel.enqueue("write", TaskResult(kind="artifact_draft", artifact_type="ReportSectionDraft", artifact_draft=_draft())) + executor = WaveExecutor(store, kernel) + assert await executor.run_until_idle(run_id) == "completed" + + ledger = TaskLedger(store) + changed = await ledger.apply_command_rework( + run_id, + command_id="cmd-cascade-1", + instruction="重新查，重点看价格战", + node_ids=["research"], + cascade_downstream=True, + ) + assert set(changed) == {"research", "write"} + + nodes = await store.list_node_runs(run_id) + by_node: dict[str, list[dict]] = {} + for item in nodes: + by_node.setdefault(item["node_id"], []).append(item) + research_attempts = {item["attempt"]: item["status"] for item in by_node["research"]} + write_attempts = {item["attempt"]: item["status"] for item in by_node["write"]} + assert research_attempts[1] == "superseded" and research_attempts[2] == "planned" + assert write_attempts[1] == "superseded" and write_attempts[2] == "planned" + events = [item for item in await store.list_events(run_id) if item["type"] == "node.status.changed"] + cascade_events = [item for item in events if (item.get("data") or {}).get("reason") == "downstream_cascade"] + assert cascade_events and cascade_events[0]["data"]["node_id"] == "write" + + kernel2 = ScriptedMemberKernel() + kernel2.enqueue("research", TaskResult(kind="artifact_draft", artifact_type="EvidenceBundle", artifact_draft=_evidence("价格战证据"))) + kernel2.enqueue("write", TaskResult(kind="artifact_draft", artifact_type="ReportSectionDraft", artifact_draft=_draft())) + executor2 = WaveExecutor(store, kernel2) + assert await executor2.run_until_idle(run_id) == "completed" + assert executor2.dispatched == ["research", "write"] + + +@pytest.mark.asyncio +async def test_exhausted_interrupted_node_fails_run_instead_of_zombie() -> None: + """P0.2 僵尸 run 守卫：attempt 耗尽的 interrupted 节点应让 run 进入 failed。""" + + store = MemoryReportCollaborationStore() + run_id = await _boot(store, _plan(max_attempts=2)) + # 模拟两次崩溃被 recovery 打断后 attempt 耗尽：research attempt2 interrupted。 + nodes = await store.list_node_runs(run_id) + research = next(item for item in nodes if item["node_id"] == "research") + research["attempt"] = 2 + research["status"] = "interrupted" + research["error"] = "lease_reclaimed" + await store.save_node_run(research) + write = next(item for item in nodes if item["node_id"] == "write") + write["status"] = "planned" + + executor = WaveExecutor(store, ScriptedMemberKernel()) + status = await executor.run_until_idle(run_id) + assert status == "failed" + run = await store.get_run(run_id) + assert run["status"] == "failed" + events = [item for item in await store.list_events(run_id) if item["type"] == "run.status.changed"] + assert events and events[-1]["data"]["reason"] == "node_attempts_exhausted" + assert events[-1]["data"]["node_id"] == "research" + + +@pytest.mark.asyncio +async def test_interrupted_with_budget_left_keeps_run_alive() -> None: + """未耗尽的 interrupted 节点仍留给 recovery 补开 attempt，不能提前 fail。""" + + store = MemoryReportCollaborationStore() + run_id = await _boot(store, _plan(max_attempts=3)) + nodes = await store.list_node_runs(run_id) + research = next(item for item in nodes if item["node_id"] == "research") + research["attempt"] = 1 + research["status"] = "interrupted" + await store.save_node_run(research) + ledger = TaskLedger(store) + assert await ledger.refresh_run_terminal(run_id) != "failed" + + @pytest.mark.asyncio async def test_parallel_researchers_share_phase() -> None: store = MemoryReportCollaborationStore()