diff --git a/frontend-web/src/components/workspace/chats/chat-box.tsx b/frontend-web/src/components/workspace/chats/chat-box.tsx index 6985e5d5..2cdc2fba 100644 --- a/frontend-web/src/components/workspace/chats/chat-box.tsx +++ b/frontend-web/src/components/workspace/chats/chat-box.tsx @@ -19,6 +19,7 @@ import { mergeArtifactPaths, } from "@/core/artifacts/listing"; import { latestInFlightWriteFilePreview } from "@/core/artifacts/write-file-in-flight"; +import { shouldAutoOpenWritingArtifact } from "@/core/threads/writing-mode"; import { getBackendBaseURL } from "@/core/config"; import { isThreadRagModeLocked, useLocalSettings } from "@/core/settings"; import { isSingleRoundRagMessage } from "@/core/threads/use-single-round-rag"; @@ -390,9 +391,12 @@ const ChatBox: React.FC<{ ((artifactDirectoryModeEnabled && threadChanged) || ((hasNewArtifact || !hadArtifacts) && !selectedArtifact)) ) { - selectArtifact(nextArtifacts[nextArtifacts.length - 1]!); - setArtifactsOpen(true); - setSideTab("document"); + const newest = nextArtifacts[nextArtifacts.length - 1]!; + if (shouldAutoOpenWritingArtifact(newest, thread.messages)) { + selectArtifact(newest); + setArtifactsOpen(true); + setSideTab("document"); + } } // DO NOT automatically deselect the artifact when switching threads, because the artifacts auto discovering is not work now. @@ -424,6 +428,7 @@ const ChatBox: React.FC<{ setArtifacts, availableArtifacts, disableArtifacts, + thread.messages, ]); // Live sandbox: open as soon as any in-flight write_file has a path or diff --git a/frontend-web/src/components/workspace/messages/message-group.tsx b/frontend-web/src/components/workspace/messages/message-group.tsx index 996f9675..3505ef92 100644 --- a/frontend-web/src/components/workspace/messages/message-group.tsx +++ b/frontend-web/src/components/workspace/messages/message-group.tsx @@ -69,6 +69,7 @@ import { type RuntimeDisplaySource, } from "@/core/skill-display/runtime"; import { normalizeDisplaySources } from "@/core/skill-display/display-adapter"; +import { isInterceptedWritingReportWrite } from "@/core/threads/writing-mode"; import { extractTitleFromMarkdown } from "@/core/utils/markdown"; import { env } from "@/env"; import { cn } from "@/lib/utils"; @@ -866,6 +867,22 @@ function convertToSteps(messages: Message[]): CoTStep[] { if (tool_call.name === "task") { continue; } + const writePath = + typeof (tool_call.args as { path?: unknown } | undefined)?.path === "string" + ? (tool_call.args as { path: string }).path + : ""; + if ( + ( + tool_call.name === "write_file" || + tool_call.name === "str_replace" || + tool_call.name === "read_file" || + tool_call.name === "read_multiple_files" || + tool_call.name === "present_files" + ) && + isInterceptedWritingReportWrite(writePath, message.id, messages) + ) { + continue; + } const step: CoTToolCallStep = { id: tool_call.id, messageId: message.id, diff --git a/frontend-web/src/core/artifacts/write-file-in-flight.test.ts b/frontend-web/src/core/artifacts/write-file-in-flight.test.ts index 3638d77f..21e99b2d 100644 --- a/frontend-web/src/core/artifacts/write-file-in-flight.test.ts +++ b/frontend-web/src/core/artifacts/write-file-in-flight.test.ts @@ -61,6 +61,73 @@ test("latestInFlightWriteFilePreview keeps markdown when a later json write is i assert.match(preview?.artifactPath ?? "", /report\.md/); }); +test("latestInFlightWriteFilePreview skips intercepted custom outputs markdown", () => { + const preview = latestInFlightWriteFilePreview([ + { + id: "m1", + type: "ai", + tool_calls: [ + { + id: "w1", + name: "write_file", + args: { + path: "/mnt/user-data/outputs/ai-agent-architecture-paper.md", + content: "# 草稿\n", + }, + }, + ], + }, + ]); + assert.equal(preview, null); +}); + +test("latestInFlightWriteFilePreview skips report.md while the first-turn pipeline is active", () => { + const preview = latestInFlightWriteFilePreview([ + { type: "human", content: "写一篇论文" }, + { + id: "m-pipe", + type: "ai", + tool_calls: [{ id: "p1", name: "deep_research_report", args: { topic: "课题" } }], + }, + { + id: "m-write", + type: "ai", + tool_calls: [ + { + id: "w1", + name: "write_file", + args: { + path: "/mnt/user-data/outputs/report.md", + content: "# 我自己写\n", + }, + }, + ], + }, + ]); + assert.equal(preview, null); +}); + +test("latestInFlightWriteFilePreview still opens the injected research stub", () => { + const preview = latestInFlightWriteFilePreview([ + { + id: "research-report-j1", + type: "ai", + tool_calls: [ + { + id: "research-report-j1-write", + name: "write_file", + args: { + path: "/mnt/user-data/outputs/report.md", + content: "# 报告\n", + }, + }, + ], + }, + ]); + assert.ok(preview); + assert.match(preview?.artifactPath ?? "", /report\.md/); +}); + test("latestInFlightWriteFilePreview does not auto-open a json-only write", () => { const preview = latestInFlightWriteFilePreview([ { diff --git a/frontend-web/src/core/artifacts/write-file-in-flight.ts b/frontend-web/src/core/artifacts/write-file-in-flight.ts index 3d1fefef..f3487bf6 100644 --- a/frontend-web/src/core/artifacts/write-file-in-flight.ts +++ b/frontend-web/src/core/artifacts/write-file-in-flight.ts @@ -1,3 +1,4 @@ +import { isInterceptedWritingReportWrite } from "../threads/writing-mode"; import { isImageFile } from "../utils/files"; import { @@ -123,6 +124,11 @@ export function latestInFlightWriteFilePreview( if (!sourcePath) { continue; } + // Writing mode rewrites custom outputs/*.md to deep_research_report. + // Opening that filename would 404-poll until report.md lands. + if (isInterceptedWritingReportWrite(sourcePath, message.id, messages)) { + continue; + } // JSON (e.g. agentfx task-reports.json) is too large to live-preview: // switching the sandbox onto it freezes the main thread, then the rest // of the turn appears to dump at once. Keep showing markdown instead. diff --git a/frontend-web/src/core/messages/utils-group.test.ts b/frontend-web/src/core/messages/utils-group.test.ts index e984a338..8d2e212d 100644 --- a/frontend-web/src/core/messages/utils-group.test.ts +++ b/frontend-web/src/core/messages/utils-group.test.ts @@ -56,6 +56,61 @@ test("write_file tool after 用户协助 opens a processing group instead of err ]); }); +test("writing-mode progress merges into the deep_research_report bar so the summary sits below", () => { + const types = groupTypes([ + { type: "human", id: "h1", content: "写一篇论文" } as Message, + { + type: "ai", + id: "a1", + content: "", + tool_calls: [{ id: "c1", name: "deep_research_report", args: { query: "多智能体" } }], + } as Message, + { + type: "tool", + id: "t1", + name: "deep_research_report", + content: "ok", + tool_call_id: "c1", + } as Message, + { + type: "ai", + id: "progress", + content: "", + tool_calls: [ + { id: "p1", name: "deep_research_progress", args: { label: "撰写研究报告", status: "complete" } }, + ], + } as Message, + { + type: "ai", + id: "write", + content: "", + tool_calls: [{ id: "w1", name: "write_file", args: { path: "/mnt/user-data/outputs/report.md" } }], + } as Message, + { + type: "ai", + id: "summary", + content: "已生成多智能体论文，详见 report.md。", + } as Message, + ]); + + assert.deepEqual(types, ["human", "assistant:processing", "assistant"]); +}); + +test("a lone deep-research progress bar still opens its own group", () => { + const types = groupTypes([ + { + type: "ai", + id: "progress", + content: "", + tool_calls: [ + { id: "p1", name: "deep_research_progress", args: { label: "撰写研究报告", status: "active" } }, + ], + } as Message, + ]); + + assert.deepEqual(types, ["assistant:processing"]); +}); + test("tool messages still attach to an open processing group", () => { const types = groupTypes([ { diff --git a/frontend-web/src/core/messages/utils.ts b/frontend-web/src/core/messages/utils.ts index 04970806..5e3369b8 100644 --- a/frontend-web/src/core/messages/utils.ts +++ b/frontend-web/src/core/messages/utils.ts @@ -166,14 +166,19 @@ export function groupMessages( messages: [message], }); } else if (hasOnlyDeepResearchProgress(message)) { - // Each deep-research writing bar is its own group. Merging consecutive - // processing messages would pin a new report's in-progress steps onto - // the parent bar, which sits above the original file card. - groups.push({ - id: message.id, - type: "assistant:processing", - messages: [message], - }); + // 写作模式把进度条插在同一次 `deep_research_report` 之后，必须并进 + // 当前 processing 组，总结才能排在整条步骤条下面。新一轮写作前面 + // 会有 human/assistant 隔开，不会误并到上一轮的条上。 + const lastGroup = groups[groups.length - 1]; + if (lastGroup?.type === "assistant:processing" && groupHasDeepResearchReport(lastGroup)) { + lastGroup.messages.push(message); + } else { + groups.push({ + id: message.id, + type: "assistant:processing", + messages: [message], + }); + } } else if (hasSubagent(message)) { groups.push({ id: message.id, @@ -426,6 +431,14 @@ export function hasOnlyDeepResearchProgress(message: Message) { ); } +function groupHasDeepResearchReport(group: MessageGroup): boolean { + return group.messages.some( + (message) => + message.type === "ai" && + Boolean(message.tool_calls?.some((toolCall) => toolCall.name === "deep_research_report")), + ); +} + export function isClarificationToolMessage(message: Message) { return message.type === "tool" && message.name === "ask_clarification"; } diff --git a/frontend-web/src/core/threads/use-research-report-stream.test.ts b/frontend-web/src/core/threads/use-research-report-stream.test.ts index 197cee59..0ad77669 100644 --- a/frontend-web/src/core/threads/use-research-report-stream.test.ts +++ b/frontend-web/src/core/threads/use-research-report-stream.test.ts @@ -6,6 +6,7 @@ import { buildResearchFeedMessages, buildResearchProgressToolCalls, buildResearchReportVirtualUrl, + composeResearchDisplayMessages, clearResearchReportStream, finalizeResearchReportStream, ingestResearchReportEvent, @@ -148,6 +149,22 @@ test("progress tool calls reveal stages stepwise and mark interruption", () => { assert.equal(buildResearchProgressToolCalls(writing).length, 3); assert.equal(buildResearchProgressToolCalls(writing)[2].args.status, "active"); + const completedByPhase = streamOf({ jobId: "j7b", phase: "done", reportText: "# x" }); + const completedCalls = buildResearchProgressToolCalls(completedByPhase); + assert.equal(completedCalls.length, 3); + assert.ok(completedCalls.every((call) => call.args.status === "complete")); + assert.match(completedCalls[2].args.detail, /report\.md/); + + const completedByStatus = streamOf({ + jobId: "j7c", + phase: "writing", + status: "success", + reportText: "# x", + }); + assert.ok( + buildResearchProgressToolCalls(completedByStatus).every((call) => call.args.status === "complete"), + ); + const failed = streamOf({ jobId: "j8", phase: "writing", status: "error", error: "模型超时" }); const failedCalls = buildResearchProgressToolCalls(failed); assert.match(failedCalls.at(-1)?.args.label ?? "", /已中断/); @@ -157,15 +174,73 @@ test("progress tool calls reveal stages stepwise and mark interruption", () => { test("feed messages attach the write_file stub only once report text exists", () => { const noText = streamOf({ jobId: "j9", phase: "collecting" }); assert.equal(buildResearchFeedMessages(noText).length, 1); - assert.equal(buildResearchFeedMessages(noText)[0].tool_calls[0].name, "deep_research_progress"); + assert.equal(buildResearchFeedMessages(noText)[0].tool_calls?.[0]?.name, "deep_research_progress"); + + const writingEmpty = streamOf({ jobId: "j9b", phase: "writing" }); + assert.equal(buildResearchFeedMessages(writingEmpty).length, 1); const withText = streamOf({ jobId: "j10", phase: "writing", reportText: "# 报告正文" }); const feed = buildResearchFeedMessages(withText); assert.equal(feed.length, 2); - const stub = feed[1].tool_calls[0]; - assert.equal(stub.name, "write_file"); - assert.equal(stub.args.path, "/mnt/user-data/outputs/report.md"); - assert.equal(stub.args.content, "# 报告正文"); + const stub = feed[1].tool_calls?.[0]; + assert.equal(stub?.name, "write_file"); + assert.equal(stub?.args.path, "/mnt/user-data/outputs/report.md"); + assert.equal(stub?.args.content, "# 报告正文"); + + const done = streamOf({ jobId: "j10b", phase: "done", reportText: "# 报告正文" }); + const doneFeed = buildResearchFeedMessages(done); + assert.equal(doneFeed.length, 3); + assert.equal(doneFeed[2].type, "tool"); + assert.equal(doneFeed[2].tool_call_id, "research-report-j10b-write"); +}); + +test("report_completed flips phase to done and write step completes before tool_finished", () => { + const tid = nextThread(); + ingestAll(tid, [ + { type: "report_delta", jobId: "jdone", payload: { delta: "正文", offset: 0 } }, + { type: "report_completed", jobId: "jdone", payload: { sourceCount: 2 } }, + ]); + const state = peekResearchReportStream(tid); + assert.equal(state?.phase, "done"); + assert.equal(state?.status, "active"); + const calls = buildResearchProgressToolCalls(state!); + assert.ok(calls.every((call) => call.args.status === "complete")); +}); + +test("composeResearchDisplayMessages inserts the bar before the summary and drops leaked prose", () => { + const feed = buildResearchFeedMessages(streamOf({ + jobId: "jsplice", + phase: "done", + reportText: "# 报告", + })); + const composed = composeResearchDisplayMessages( + [ + { id: "h", type: "human", content: "写论文" }, + { + id: "ai-tool", + type: "ai", + content: "", + tool_calls: [{ id: "c1", name: "deep_research_report", args: {} }], + }, + { id: "leak", type: "ai", content: "# 报告\n\n泄漏到消息列表的正文" }, + { id: "tool", type: "tool", tool_call_id: "c1", content: "ok" }, + { id: "summary", type: "ai", content: "已生成论文，详见 report.md。" }, + ], + feed, + ); + + assert.deepEqual( + composed.map((message) => message.id), + [ + "h", + "ai-tool", + "tool", + "research-progress-jsplice", + "research-report-jsplice", + "research-report-jsplice-write-result", + "summary", + ], + ); }); test("virtual URL matches the message-list ArtifactWriteToolCall URL (one suppression key)", () => { diff --git a/frontend-web/src/core/threads/use-research-report-stream.ts b/frontend-web/src/core/threads/use-research-report-stream.ts index 487ffff5..0e26f9c2 100644 --- a/frontend-web/src/core/threads/use-research-report-stream.ts +++ b/frontend-web/src/core/threads/use-research-report-stream.ts @@ -84,8 +84,8 @@ function getStore(threadId: string): StoreEntry { * 通知若逐帧发出，会让聊天页整棵树（长消息列表 + 沙箱 Markdown 预览）按 * token 频率重渲染，页面直接卡死。工作台用 requestAnimationFrame 批处理 * （DeepResearchWorkbench），聊天这里等效地按固定间隔合并——每秒最多 - * 10 次通知，流式观感不受影响。终态帧（tool_finished）与 - * finalize/clear 立即 flush，结束态不滞后。 + * 10 次通知，流式观感不受影响。终态帧（tool_finished / + * report_completed）与 finalize/clear 立即 flush，结束态不滞后。 */ const EMIT_COALESCE_MS = 100; @@ -193,7 +193,8 @@ export function ingestResearchReportEvent(threadId: string, frame: ResearchRepor case "report_completed": next = { ...base, - phase: "writing", + // 文件已落盘：步骤条应立刻标完成，不必等随后的 tool_finished。 + phase: "done", sourceCount: typeof payload.sourceCount === "number" && payload.sourceCount > 0 ? payload.sourceCount : base.sourceCount, @@ -223,7 +224,7 @@ export function ingestResearchReportEvent(threadId: string, frame: ResearchRepor if (next !== previous) { store.state = next; // 状态同步更新（peek/快照立即可见）；通知按 token 帧频率合并，终态立即。 - if (frame.type === "tool_finished") { + if (frame.type === "tool_finished" || frame.type === "report_completed") { emitNow(store); } else { emitCoalesced(store); @@ -280,7 +281,11 @@ export type ResearchProgressToolCall = { export function buildResearchProgressToolCalls(state: ResearchReportStreamState): ResearchProgressToolCall[] { const rank = phaseRank(state.phase); const finished = state.status !== "active"; - const effectiveRank = finished && state.status === "success" ? PHASE_ORDER.length - 1 : rank; + // success 时必须用 length（越过 write 的 boundary），length-1 永远完不成最后一步。 + // report_completed 会先把 phase 置 done，此时工具可能还没返回，也视为写完。 + const succeeded = + state.status === "success" || (state.status === "active" && state.phase === "done"); + const effectiveRank = succeeded ? PHASE_ORDER.length : rank; const stages = [ { id: "plan", @@ -304,9 +309,11 @@ export function buildResearchProgressToolCalls(state: ResearchReportStreamState) id: "write", label: "撰写研究报告", boundary: PHASE_ORDER.length, - detail: state.reportText - ? "正在按「通用分析结构」逐字撰写，右侧沙箱同步预览" - : "正在整理资料并开始撰写报告", + detail: succeeded + ? "已按「通用分析结构」写完并保存为 report.md" + : state.reportText + ? "正在按「通用分析结构」逐字撰写，右侧沙箱同步预览" + : "正在整理资料并开始撰写报告", }, ]; const activeIndex = stages.findIndex((stage) => effectiveRank < stage.boundary); @@ -330,19 +337,25 @@ export function buildResearchProgressToolCalls(state: ResearchReportStreamState) export type ResearchFeedMessage = { id: string; - type: "ai"; + type: "ai" | "tool"; content: string; - tool_calls: { + name?: string; + tool_call_id?: string; + tool_calls?: { id: string; name: string; args: Record; }[]; }; +function isResearchWriteFinished(state: ResearchReportStreamState): boolean { + return state.status === "success" || state.phase === "done"; +} + /** - * 组装聊天页要注入的两条虚拟 ai 消息：进度步骤条 + report.md 的 write_file - * stub（沙箱 write-file 虚拟 URL 预览的数据源）。仅在报告尚无正文或仍在 - * 流式时输出 stub —— 完成后由真实 artifacts 接管。 + * 组装聊天页要注入的虚拟消息：进度步骤条 + report.md 的 write_file stub + * （沙箱 write-file 虚拟 URL 预览的数据源）。写完后补一条 tool result， + * 让沙箱的「正在生成」立刻收起，不必等真实 artifacts 交接。 */ export function buildResearchFeedMessages(state: ResearchReportStreamState): ResearchFeedMessage[] { const messages: ResearchFeedMessage[] = [ @@ -353,23 +366,127 @@ export function buildResearchFeedMessages(state: ResearchReportStreamState): Res tool_calls: buildResearchProgressToolCalls(state), }, ]; - if (state.reportText.trim()) { + // 必须等正文首 token：思考阶段挂 stub 会提前显示「写入文件 report.md」。 + if (!state.reportText.trim()) { + return messages; + } + const writeCallId = `research-report-${state.jobId}-write`; + messages.push({ + id: `research-report-${state.jobId}`, + type: "ai", + content: "", + tool_calls: [ + { + id: writeCallId, + name: "write_file", + args: { path: RESEARCH_REPORT_OUTPUT_PATH, content: state.reportText }, + }, + ], + }); + if (isResearchWriteFinished(state)) { messages.push({ - id: `research-report-${state.jobId}`, - type: "ai", - content: "", - tool_calls: [ - { - id: `research-report-${state.jobId}-write`, - name: "write_file", - args: { path: RESEARCH_REPORT_OUTPUT_PATH, content: state.reportText }, - }, - ], + id: `research-report-${state.jobId}-write-result`, + type: "tool", + name: "write_file", + tool_call_id: writeCallId, + content: "ok", }); } return messages; } +function isDeepResearchReportAi(message: { type?: string; tool_calls?: Array<{ name?: string; id?: string }> }): boolean { + return ( + message.type === "ai" && + Boolean(message.tool_calls?.some((toolCall) => toolCall.name === "deep_research_report")) + ); +} + +function deepResearchReportCallIds(message: { tool_calls?: Array<{ name?: string; id?: string }> }): Set { + return new Set( + (message.tool_calls ?? []) + .filter((toolCall) => toolCall.name === "deep_research_report" && toolCall.id) + .map((toolCall) => toolCall.id as string), + ); +} + +function messageHasVisibleText(message: { type?: string; content?: unknown; tool_calls?: unknown[] }): boolean { + if (message.type !== "ai") return false; + if (Array.isArray(message.tool_calls) && message.tool_calls.length > 0) return false; + return typeof message.content === "string" && message.content.trim().length > 0; +} + +/** + * 丢掉管线内部 LLM 泄漏到父图 messages 流的正文：它们夹在 + * `deep_research_report` 调用与其 tool result 之间，会让左侧气泡和沙箱一起写。 + */ +export function omitLeakedReportTokens; tool_call_id?: string; content?: unknown }>( + messages: T[], +): T[] { + const hidden = new Set(); + let pendingIds: Set | null = null; + for (let i = 0; i < messages.length; i++) { + const message = messages[i]; + if (isDeepResearchReportAi(message)) { + pendingIds = deepResearchReportCallIds(message); + continue; + } + if (!pendingIds) continue; + if ( + message.type === "tool" && + typeof message.tool_call_id === "string" && + pendingIds.has(message.tool_call_id) + ) { + pendingIds = null; + continue; + } + if (messageHasVisibleText(message)) { + hidden.add(i); + } + } + return hidden.size === 0 ? messages : messages.filter((_, index) => !hidden.has(index)); +} + +/** + * 把进度条 / write_file stub 插到 `deep_research_report` 及其结果之后、 + * 模型总结之前，这样总结气泡永远在步骤条下面。 + */ +export function insertResearchFeedMessages(messages: T[], feed: T[]): T[] { + if (feed.length === 0) return messages; + let insertAt = -1; + let callIds = new Set(); + for (let i = 0; i < messages.length; i++) { + const message = messages[i] as { type?: string; tool_calls?: Array<{ name?: string; id?: string }> }; + if (isDeepResearchReportAi(message)) { + insertAt = i + 1; + callIds = deepResearchReportCallIds(message); + } + } + if (insertAt === -1) { + return [...messages, ...feed]; + } + while (insertAt < messages.length) { + const message = messages[insertAt] as { type?: string; tool_call_id?: string }; + if ( + message.type === "tool" && + typeof message.tool_call_id === "string" && + callIds.has(message.tool_call_id) + ) { + insertAt += 1; + continue; + } + break; + } + return [...messages.slice(0, insertAt), ...feed, ...messages.slice(insertAt)]; +} + +export function composeResearchDisplayMessages; tool_call_id?: string; content?: unknown }>( + messages: T[], + feed: T[], +): T[] { + return insertResearchFeedMessages(omitLeakedReportTokens(messages), feed); +} + /** write-file 虚拟 URL（ArtifactFileDetail 用它从 stub 消息读流式内容）。 */ export function buildResearchReportVirtualUrl(state: ResearchReportStreamState): string { return new URL( diff --git a/frontend-web/src/core/threads/use-research-writing-feed.ts b/frontend-web/src/core/threads/use-research-writing-feed.ts index 4c50bbf4..06441e41 100644 --- a/frontend-web/src/core/threads/use-research-writing-feed.ts +++ b/frontend-web/src/core/threads/use-research-writing-feed.ts @@ -3,8 +3,8 @@ /** * 聊天页「写作模式」深度研究进度注入。 * - * 把 `useResearchReportStream` 的流式状态转成 displayThread 尾部追加的 - * 虚拟消息（进度步骤条 + report.md 的 write_file stub）： + * 把 `useResearchReportStream` 的流式状态插到 `deep_research_report` + * 工具结果之后、模型总结之前（进度步骤条 + report.md 的 write_file stub）： * * - 进行中 / 中断：始终注入（中断态展示「已中断」，下一轮 jobId 变化自然淘汰）； * - 成功：注入到真实 report.md 产物落地为止（artifacts 出现该路径即撤出）， @@ -26,13 +26,16 @@ import { artifactPreviewKey, useArtifactsOptional, } from "@/components/workspace/artifacts/context"; +import { resolveWriteFilePath } from "@/core/artifacts/write-file-in-flight"; import { buildResearchFeedMessages, buildResearchReportVirtualUrl, + composeResearchDisplayMessages, RESEARCH_REPORT_OUTPUT_PATH, useResearchReportStream, } from "./use-research-report-stream"; +import { isInterceptedWritingReportWrite } from "./writing-mode"; type FeedThreadLike = { messages: Message[]; @@ -55,10 +58,15 @@ export function useResearchWritingFeed( [thread.values?.artifacts], ); - // 报告首帧正文到达且仍在写入 → 主动展开右侧沙箱并选中 write-file 虚拟 - // 预览，报告在沙箱里边写边流式渲染。 + // 正文首 token 到达且仍在写入 → 主动展开右侧沙箱。思考阶段不打开， + // 避免「写入文件 report.md」出现在模型还在想的时候。 useEffect(() => { - if (!research || research.status !== "active" || !research.reportText.trim()) { + if ( + !research || + research.status !== "active" || + research.phase !== "writing" || + !research.reportText.trim() + ) { return; } if (autoOpenedJobRef.current === research.jobId) return; @@ -77,7 +85,14 @@ export function useResearchWritingFeed( if (!artifacts) return; handedOverJobRef.current = research.jobId; const current = artifacts.selectedArtifact; - if (current !== null && current !== buildResearchReportVirtualUrl(research)) { + const virtualUrl = buildResearchReportVirtualUrl(research); + const currentPath = current ? resolveWriteFilePath(current) : ""; + const keepUserSelection = + current !== null && + current !== virtualUrl && + current !== RESEARCH_REPORT_OUTPUT_PATH && + !isInterceptedWritingReportWrite(currentPath); + if (keepUserSelection) { return; } artifacts.select(RESEARCH_REPORT_OUTPUT_PATH); @@ -91,11 +106,11 @@ export function useResearchWritingFeed( }, [hasRealArtifact, research]); const displayThread = useMemo(() => { - if (feedMessages.length === 0) return thread; - const messages = [ - ...thread.messages, - ...(feedMessages as unknown as Message[]), - ]; + const messages = composeResearchDisplayMessages( + thread.messages, + feedMessages as unknown as Message[], + ); + if (messages === thread.messages) return thread; return { ...thread, values: { ...thread.values, messages }, diff --git a/frontend-web/src/core/threads/writing-mode.test.ts b/frontend-web/src/core/threads/writing-mode.test.ts new file mode 100644 index 00000000..a7f2e86e --- /dev/null +++ b/frontend-web/src/core/threads/writing-mode.test.ts @@ -0,0 +1,101 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + isInterceptedWritingReportWrite, + isWritingFirstReportTurn, + shouldAutoOpenWritingArtifact, +} from "./writing-mode"; + +test("custom outputs markdown is always intercepted except the research stub", () => { + assert.equal( + isInterceptedWritingReportWrite("/mnt/user-data/outputs/paper.md"), + true, + ); + assert.equal( + isInterceptedWritingReportWrite( + "/mnt/user-data/outputs/report.md", + "research-report-j1", + ), + false, + ); +}); + +test("report.md is intercepted for the whole first-report turn", () => { + const beforePipeline = [{ type: "human", content: "写一篇论文" }]; + assert.equal(isWritingFirstReportTurn(beforePipeline), true); + assert.equal( + isInterceptedWritingReportWrite( + "/mnt/user-data/outputs/report.md", + "m1", + beforePipeline, + ), + true, + ); + + const inFlight = [ + { type: "human", content: "写一篇论文" }, + { + type: "ai", + tool_calls: [{ name: "deep_research_report" }], + }, + ]; + assert.equal(isWritingFirstReportTurn(inFlight), true); + assert.equal( + isInterceptedWritingReportWrite("/mnt/user-data/outputs/report.md", "m2", inFlight), + true, + ); + + const succeeded = [ + ...inFlight, + { + type: "tool", + name: "deep_research_report", + content: "研究报告已完成：已写入 report.md。", + }, + ]; + assert.equal(isWritingFirstReportTurn(succeeded), true); + assert.equal( + isInterceptedWritingReportWrite("/mnt/user-data/outputs/report.md", "m3", succeeded), + true, + ); +}); + +test("sandbox auto-open waits for report.md on the first-report turn", () => { + const firstTurn = [ + { type: "human", content: "写一篇论文" }, + { + type: "ai", + tool_calls: [{ name: "deep_research_report" }], + }, + ]; + assert.equal( + shouldAutoOpenWritingArtifact("/mnt/user-data/outputs/research-notes.md", firstTurn), + false, + ); + assert.equal( + shouldAutoOpenWritingArtifact("/mnt/user-data/outputs/report.md", firstTurn), + true, + ); +}); + +test("a later user turn releases report.md live edits", () => { + const followUp = [ + { type: "human", content: "写一篇论文" }, + { + type: "ai", + tool_calls: [{ name: "deep_research_report" }], + }, + { + type: "tool", + name: "deep_research_report", + content: "研究报告已完成：已写入 report.md。", + }, + { type: "human", content: "把第二节改短一点" }, + ]; + assert.equal(isWritingFirstReportTurn(followUp), false); + assert.equal( + isInterceptedWritingReportWrite("/mnt/user-data/outputs/report.md", "m4", followUp), + false, + ); +}); diff --git a/frontend-web/src/core/threads/writing-mode.ts b/frontend-web/src/core/threads/writing-mode.ts new file mode 100644 index 00000000..79a6e5c3 --- /dev/null +++ b/frontend-web/src/core/threads/writing-mode.ts @@ -0,0 +1,112 @@ +/** Only the writing-mode deliverable counts as a follow-up edit target. */ + +const OUTPUTS_PREFIX = "/mnt/user-data/outputs"; + +export function normalizeWritingPath(path: string | null | undefined): string { + if (!path) return ""; + return path.replace(/\\/g, "/").split("?")[0]?.split("#")[0]?.replace(/\/+$/, "") ?? ""; +} + +export function isWritingModeReportPath( + path: string | null | undefined, +): boolean { + const normalized = normalizeWritingPath(path); + return /(?:^|\/)report\.md$/i.test(normalized); +} + +/** Markdown under the sandbox outputs dir — writing mode intercepts these as the report. */ +export function isOutputsMarkdownPath(path: string | null | undefined): boolean { + const normalized = normalizeWritingPath(path); + if (!normalized || !/\.(?:md|markdown)$/i.test(normalized)) return false; + return ( + normalized === OUTPUTS_PREFIX || + normalized.startsWith(`${OUTPUTS_PREFIX}/`) || + normalized === "outputs" || + normalized.startsWith("outputs/") + ); +} + +type WritingPipelineMessage = { + type?: string; + name?: string; + content?: unknown; + additional_kwargs?: Record; + tool_calls?: Array<{ name?: string }>; +}; + +/** + * First-report turn: no successful pipeline yet, or the latest visible user + * message is still the one that triggered that pipeline. Follow-up edits + * (a later visible human after success) keep live `report.md` writes visible. + */ +export function isWritingFirstReportTurn( + messages: readonly WritingPipelineMessage[] | null | undefined, +): boolean { + if (!messages?.length) return false; + let lastVisibleHuman = -1; + let lastSuccess = -1; + for (let index = 0; index < messages.length; index += 1) { + const message = messages[index]; + if ( + message?.type === "human" && + message.additional_kwargs?.hide_from_ui !== true + ) { + lastVisibleHuman = index; + } + if (message?.type === "tool" && message.name === "deep_research_report") { + const content = typeof message.content === "string" ? message.content : ""; + if (content && !content.toLowerCase().startsWith("error")) { + lastSuccess = index; + } + } + } + return lastVisibleHuman >= 0 && (lastSuccess < 0 || lastVisibleHuman <= lastSuccess); +} + +/** + * Hand-written outputs/*.md writes the backend rewrites or drops: + * custom filenames always, and `report.md` while the first-turn pipeline is + * in-flight or just succeeded. The injected research stub stays visible. + */ +export function isInterceptedWritingReportWrite( + path: string | null | undefined, + messageId?: string | null, + messages?: readonly WritingPipelineMessage[] | null, +): boolean { + if (typeof messageId === "string" && messageId.startsWith("research-report-")) { + return false; + } + if (!isOutputsMarkdownPath(path)) { + return false; + } + if (!isWritingModeReportPath(path)) { + return true; + } + return isWritingFirstReportTurn(messages); +} + +/** Auto-open the sandbox only for the real report during the first-report turn. */ +export function shouldAutoOpenWritingArtifact( + path: string | null | undefined, + messages?: readonly WritingPipelineMessage[] | null, +): boolean { + if (!path) { + return false; + } + if (isWritingFirstReportTurn(messages)) { + return isWritingModeReportPath(path); + } + return !isInterceptedWritingReportWrite(path, undefined, messages); +} + +export function resolveWritingModeReportPath( + artifacts: readonly string[] | null | undefined, + selectedArtifact?: string | null, +): string | undefined { + if (isWritingModeReportPath(selectedArtifact)) { + return selectedArtifact ?? undefined; + } + return [...(artifacts ?? [])] + .reverse() + .find((path) => isWritingModeReportPath(path)); +} diff --git a/frontend-web/src/pages/AgentChatPage.tsx b/frontend-web/src/pages/AgentChatPage.tsx index 20774b4f..1cf53d5a 100644 --- a/frontend-web/src/pages/AgentChatPage.tsx +++ b/frontend-web/src/pages/AgentChatPage.tsx @@ -45,6 +45,7 @@ import { stopActiveThreadBeforeClarificationReply, stopThreadRun, useThreadStrea import { useResearchWritingFeed } from "@/core/threads/use-research-writing-feed"; import { clearResearchReportStream } from "@/core/threads/use-research-report-stream"; import { textOfMessage } from "@/core/threads/utils"; +import { resolveWritingModeReportPath } from "@/core/threads/writing-mode"; import { OptType, writeYLog } from "@/core/log/y-log"; import { Select, @@ -612,16 +613,13 @@ export default function AgentChatPage({ } as typeof displayThread; }, [displayThread, showReportImportCard, taskId, threadId]); - // 写作模式续写轮：已有 md 产物（report.md 已生成）时回传 writing_artifact_path， - // 后端据此走「直接编辑画布」分支而不是再次挂载深度研究写报告管线（与 - // ChatPage 的回传逻辑对齐）。 - const writingArtifactPath = useMemo(() => { - const isMarkdown = (path: string | null | undefined) => - !!path && /\.(md|markdown)(?:$|[?#])/i.test(path); - return [...(thread.values.artifacts ?? [])] - .reverse() - .find((path) => isMarkdown(path)); - }, [thread.values.artifacts]); + // 写作模式续写轮：仅当已有写作模式产物 report.md 时回传 writing_artifact_path， + // 后端据此走「直接编辑画布」分支而不是再次挂载深度研究写报告管线。其它 leftover + // .md 不得跳过管线（与 ChatPage 对齐）。 + const writingArtifactPath = useMemo( + () => resolveWritingModeReportPath(thread.values.artifacts), + [thread.values.artifacts], + ); // ── 「模板数据转换器」对话页专属调试入口（仅此 agent）────────────────────────── // 选一个页面模板，问答跑完后点「渲染预览」：取最近一条 AI 回复里的 JSON 填进模板， diff --git a/frontend-web/src/pages/ChatPage.tsx b/frontend-web/src/pages/ChatPage.tsx index c52559cf..9873ff47 100644 --- a/frontend-web/src/pages/ChatPage.tsx +++ b/frontend-web/src/pages/ChatPage.tsx @@ -63,6 +63,7 @@ import { import { useThreadStream, stopActiveThreadBeforeClarificationReply } from "@/core/threads/hooks"; import { useResearchWritingFeed } from "@/core/threads/use-research-writing-feed"; import { clearResearchReportStream } from "@/core/threads/use-research-report-stream"; +import { resolveWritingModeReportPath } from "@/core/threads/writing-mode"; import { queueLlmWikiConversationDeposit } from "@/core/llmwiki/conversation-deposit"; import { useSingleRoundRagMessages } from "@/core/threads/use-single-round-rag"; import { useMultiRoundRagMessages } from "@/core/threads/use-multi-round-rag"; @@ -354,16 +355,11 @@ export default function ChatPage() { researchDisplayThread, ]); - const writingArtifactPath = useMemo(() => { - const isMarkdown = (path: string | null | undefined) => - !!path && /\.(md|markdown)(?:$|[?#])/i.test(path); - if (isMarkdown(selectedArtifact)) { - return selectedArtifact ?? undefined; - } - return [...(thread.values.artifacts ?? [])] - .reverse() - .find((path) => isMarkdown(path)); - }, [selectedArtifact, thread.values.artifacts]); + const writingArtifactPath = useMemo( + () => + resolveWritingModeReportPath(thread.values.artifacts, selectedArtifact), + [selectedArtifact, thread.values.artifacts], + ); const handleSubmit = useCallback( (message: PromptInputMessage) => { diff --git a/frontend-web/src/pages/IframeChatPage.tsx b/frontend-web/src/pages/IframeChatPage.tsx index cc6a010c..cd1f29b7 100644 --- a/frontend-web/src/pages/IframeChatPage.tsx +++ b/frontend-web/src/pages/IframeChatPage.tsx @@ -42,6 +42,7 @@ import { import { usePluginManagementSettings } from "@/core/system-settings/hooks"; import { stopActiveThreadBeforeClarificationReply, useThreadStream } from "@/core/threads/hooks"; import { textOfMessage } from "@/core/threads/utils"; +import { resolveWritingModeReportPath } from "@/core/threads/writing-mode"; import { env } from "@/env"; import { cn } from "@/lib/utils"; import { @@ -285,16 +286,11 @@ export default function IframeChatPage() { threadMetadataAgentId, ]); - const writingArtifactPath = useMemo(() => { - const isMarkdown = (path: string | null | undefined) => - !!path && /\.(md|markdown)(?:$|[?#])/i.test(path); - if (isMarkdown(selectedArtifact)) { - return selectedArtifact ?? undefined; - } - return [...(thread.values.artifacts ?? [])] - .reverse() - .find((path) => isMarkdown(path)); - }, [selectedArtifact, thread.values.artifacts]); + const writingArtifactPath = useMemo( + () => + resolveWritingModeReportPath(thread.values.artifacts, selectedArtifact), + [selectedArtifact, thread.values.artifacts], + ); const handleSubmit = useCallback( ( diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/adapters/llm.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/adapters/llm.py index 8f27f0ba..0da7c2ca 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/adapters/llm.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/adapters/llm.py @@ -43,6 +43,41 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +def _parent_isolated_run_config(run_name: str) -> dict[str, Any]: + """Keep LangGraph runtime, drop only inherited LLM streaming callbacks. + + ``StreamWriter`` reads ``get_config()[configurable]`` at emit time. Replacing + the child config with a blank dict makes ``report_delta`` / ``report_thinking`` + raise and get swallowed, so the sandbox only appears after the file lands. + """ + parent: dict[str, Any] = {} + try: + from langchain_core.runnables.config import var_child_runnable_config + + current = var_child_runnable_config.get() + if isinstance(current, dict): + parent = dict(current) + except Exception: # noqa: BLE001 — isolation must never fail the completion + parent = {} + tags = [tag for tag in (parent.get("tags") or []) if tag != "nostream"] + tags.append("nostream") + parent["callbacks"] = [] + parent["tags"] = tags + parent["run_name"] = run_name + return parent + + +def _clear_inherited_runnable_config(isolated: dict[str, Any]) -> Callable[[], None]: + """Install the isolated config for the duration of ``astream``.""" + try: + from langchain_core.runnables.config import var_child_runnable_config + except ImportError: + return lambda: None + + token = var_child_runnable_config.set(isolated) + return lambda: var_child_runnable_config.reset(token) + + # ── message conversion ────────────────────────────────────────────────────── @@ -347,26 +382,39 @@ class DeerFlowCompletionBackend: text_parts: list[str] = [] usage: dict[str, Any] = {} inline_think_filter = InlineThinkTagFilter() - async for chunk in model.astream(to_langchain_messages(messages), **kwargs): - if self._cancel is not None: - self._cancel.raise_if_cancelled() - reasoning = _reasoning_delta(chunk) - if reasoning and on_reasoning is not None: - await on_reasoning(reasoning) - # A few OpenAI-compatible models put their reasoning in the - # ordinary content channel instead of ``reasoning_content``. - # Split it from Markdown even when tags span tokens: callers may - # display a live thought trace, but it never joins ``text``. - delta, inline_reasoning = inline_think_filter.push_parts(_content_to_text(chunk)) - if inline_reasoning and on_reasoning is not None: - await on_reasoning(inline_reasoning) - if delta: - text_parts.append(delta) - if on_delta is not None: - await on_delta(delta) - chunk_usage = _extract_usage(chunk) - if any(chunk_usage.values()): - usage = chunk_usage + # Isolate from the parent LangGraph run: inherited callbacks would + # replay every report token into the chat ``messages`` stream, so the + # left-hand bubble and the sandbox would write the same prose at once. + # Custom ``report_delta`` events remain the only live preview channel. + isolated_config = _parent_isolated_run_config(f"deep_research:{operation}") + reset_parent_config = _clear_inherited_runnable_config(isolated_config) + try: + async for chunk in model.astream( + to_langchain_messages(messages), + config=isolated_config, + **kwargs, + ): + if self._cancel is not None: + self._cancel.raise_if_cancelled() + reasoning = _reasoning_delta(chunk) + if reasoning and on_reasoning is not None: + await on_reasoning(reasoning) + # A few OpenAI-compatible models put their reasoning in the + # ordinary content channel instead of ``reasoning_content``. + # Split it from Markdown even when tags span tokens: callers may + # display a live thought trace, but it never joins ``text``. + delta, inline_reasoning = inline_think_filter.push_parts(_content_to_text(chunk)) + if inline_reasoning and on_reasoning is not None: + await on_reasoning(inline_reasoning) + if delta: + text_parts.append(delta) + if on_delta is not None: + await on_delta(delta) + chunk_usage = _extract_usage(chunk) + if any(chunk_usage.values()): + usage = chunk_usage + finally: + reset_parent_config() tail, inline_reasoning_tail = inline_think_filter.finish_parts() if inline_reasoning_tail and on_reasoning is not None: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/runners/basic.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/runners/basic.py index fea239f7..751d7b47 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/runners/basic.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/runners/basic.py @@ -198,35 +198,63 @@ class BasicRunner: cancellation.raise_if_cancelled() if not all_materials: + knowledge_fill = _allows_knowledge_fill(request.config.report_instruction) await events.emit( "warning", phase="curating", - payload={"code": "no_materials", "message": "未收集到任何研究材料", "recoverable": False}, + payload={ + "code": "no_materials", + "message": ( + "未收集到外部研究材料，改用领域知识撰写" + if knowledge_fill + else "未收集到任何研究材料" + ), + "recoverable": knowledge_fill, + }, ) - report_md = f"# {request.query}\n\n未能收集到与研究课题「{request.query}」相关的材料，无法生成报告。" - await adapters.artifacts.write_text("report.md", report_md) - await events.emit("report_completed", phase="writing", payload={"reportVersion": 1, "sourceCount": 0}) - return DeepResearchResult(report_markdown=report_md, source_ids=[], plan=plan) + if not knowledge_fill: + report_md = ( + f"# {request.query}\n\n" + f"未能收集到与研究课题「{request.query}」相关的材料，无法生成报告。" + ) + await adapters.artifacts.write_text("report.md", report_md) + await events.emit( + "report_completed", + phase="writing", + payload={"reportVersion": 1, "sourceCount": 0}, + ) + return DeepResearchResult(report_markdown=report_md, source_ids=[], plan=plan) + deduped: list[ResearchMaterial] = [] + selected_ids: list[str] = [] + context_list: list[dict[str, Any]] = [] + context_text = _KNOWLEDGE_FILL_EMPTY_CONTEXT + else: + from deerflow.agents.deep_research.adapters.material import dedupe_materials - from deerflow.agents.deep_research.adapters.material import dedupe_materials + deduped = dedupe_materials(all_materials) + context_comp.set_materials(deduped) - deduped = dedupe_materials(all_materials) - context_comp.set_materials(deduped) + selected_ids = [m.id for m in deduped] + await events.emit( + "sources_curated", + phase="curating", + payload={"selectedIds": selected_ids, "removedIds": [], "reason": "deduped"}, + ) - selected_ids = [m.id for m in deduped] - await events.emit( - "sources_curated", - phase="curating", - payload={"selectedIds": selected_ids, "removedIds": [], "reason": "deduped"}, - ) + # ── 4. compress context ──────────────────────────────────────── + await events.emit( + "phase_changed", + phase="compressing", + payload={"from": "curating", "to": "compressing"}, + ) + if cancellation: + cancellation.raise_if_cancelled() - # ── 4. compress context ──────────────────────────────────────── - await events.emit("phase_changed", phase="compressing", payload={"from": "curating", "to": "compressing"}) - if cancellation: - cancellation.raise_if_cancelled() - - context_list = await context_comp.async_get_context(request.query, max_results=min(len(deduped), config.max_search_results_per_query * 2)) - context_text = _format_context(context_list, deduped) + context_list = await context_comp.async_get_context( + request.query, + max_results=min(len(deduped), config.max_search_results_per_query * 2), + ) + context_text = _format_context(context_list, deduped) # ── 5. writing: generate report ──────────────────────────────── await events.emit("phase_changed", phase="writing", payload={"from": "compressing", "to": "writing"}) @@ -403,6 +431,13 @@ class BasicRunner: ) outline_block = _report_outline_instruction(outline) requirements_block = _report_requirements_instruction(request.config.report_instruction) + if _allows_knowledge_fill(request.config.report_instruction): + system_extra = ( + "用户已要求在材料不足时用模型自身的公开领域知识写成完整报告。" + "禁止把「材料不足/未提供文献」写成章节主线；参考文献可标注为基于公开领域知识。" + ) + else: + system_extra = "" if request.config.structure_mode == "fixed": # Recency bias: keep the skeleton immediately before "请撰写报告". context_block = ( @@ -427,6 +462,8 @@ class BasicRunner: if filled_preamble: await emitter.add(filled_preamble + "\n\n") system = _SYSTEM + if system_extra: + system += system_extra if request.config.structure_mode == "fixed" and (outline or "").strip(): system += "撰写时必须原样沿用用户大纲中的章节标题与层级，不得改写成通用报告套式。" result = await llm.stream_complete( @@ -624,12 +661,31 @@ def ensure_markdown_standalone_lines(text: str) -> str: return "\n".join(output) +_KNOWLEDGE_FILL_EMPTY_CONTEXT = ( + "（本次未收集到外部研究材料。请基于模型自身的公开领域知识撰写完整、有实质内容的研究报告，" + "含架构说明；用户要求流程图时输出 Mermaid。" + "禁止只写「无法生成报告」或把「材料不足」写成全文。）" +) +_KNOWLEDGE_FILL_MARKERS = ("必须基于模型自身的领域知识", "基于模型自身的领域知识") + + +def _allows_knowledge_fill(report_instruction: str | None) -> bool: + text = report_instruction or "" + return any(marker in text for marker in _KNOWLEDGE_FILL_MARKERS) + + def _report_requirements_instruction(report_instruction: str | None) -> str: """Render confirmed generation requirements as bounded, untrusted content.""" if not report_instruction: return "User generation requirements: none provided." + evidence_note = ( + "these requirements override the default 'do not go beyond materials' rule " + "when they ask to write from model knowledge" + if _allows_knowledge_fill(report_instruction) + else "content only; system and evidence rules still win" + ) return ( - "User generation requirements (content only; system and evidence rules still win):\n" + f"User generation requirements ({evidence_note}):\n" f"\n{report_instruction}\n" ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py index b11e2eeb..53a338e4 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py @@ -1,5 +1,5 @@ -import hashlib -import logging +import hashlib +import logging from langchain.agents import create_agent from langchain.agents.middleware import AgentMiddleware @@ -38,45 +38,45 @@ logger = logging.getLogger(__name__) # 写作配置 Q&A 现在「先生成大纲并与用户多轮确认 → 再提交配置」，需要更多来回，故上限放宽。 # 仍是兜底安全阀：到顶才用 tool_choice 强制提交，正常流程由用户确认大纲后自然提交。 WRITING_SETUP_MAX_ROUNDS = 8 -REPORT_STRUCTURE_SETUP_MAX_ROUNDS = 8 - -_ES_QUERY_ROUTING_OPEN_TAG = '' - - -def _log_es_query_routing_registration( - *, - system_prompt: str, - app_config: AppConfig, - agent_id: str | None, - available_skills: set[str] | None, -) -> None: - """Log whether the exact configured ES routing block reached this Q&A prompt.""" - routing = app_config.skills.es_query_routing - configured_prompt = (routing.prompt or "").strip() - expected_block = ( - f"{_ES_QUERY_ROUTING_OPEN_TAG}\n{configured_prompt}\n" - if configured_prompt - else "" - ) - eligible = available_skills is None or "es_query" in available_skills - registered = bool(expected_block) and expected_block in system_prompt - fingerprint = ( - hashlib.sha256(configured_prompt.encode("utf-8")).hexdigest()[:12] - if configured_prompt - else "-" - ) - - log = logger.info if registered or not routing.enabled or not eligible else logger.warning - log( - "ES query routing prompt registration: agent_id=%s enabled=%s eligible=%s " - "registered=%s prompt_chars=%d prompt_sha256=%s", - agent_id or "default", - routing.enabled, - eligible, - registered, - len(configured_prompt), - fingerprint, - ) +REPORT_STRUCTURE_SETUP_MAX_ROUNDS = 8 + +_ES_QUERY_ROUTING_OPEN_TAG = '' + + +def _log_es_query_routing_registration( + *, + system_prompt: str, + app_config: AppConfig, + agent_id: str | None, + available_skills: set[str] | None, +) -> None: + """Log whether the exact configured ES routing block reached this Q&A prompt.""" + routing = app_config.skills.es_query_routing + configured_prompt = (routing.prompt or "").strip() + expected_block = ( + f"{_ES_QUERY_ROUTING_OPEN_TAG}\n{configured_prompt}\n" + if configured_prompt + else "" + ) + eligible = available_skills is None or "es_query" in available_skills + registered = bool(expected_block) and expected_block in system_prompt + fingerprint = ( + hashlib.sha256(configured_prompt.encode("utf-8")).hexdigest()[:12] + if configured_prompt + else "-" + ) + + log = logger.info if registered or not routing.enabled or not eligible else logger.warning + log( + "ES query routing prompt registration: agent_id=%s enabled=%s eligible=%s " + "registered=%s prompt_chars=%d prompt_sha256=%s", + agent_id or "default", + routing.enabled, + eligible, + registered, + len(configured_prompt), + fingerprint, + ) def _get_runtime_config(config: RunnableConfig) -> dict: @@ -426,6 +426,18 @@ def _build_middlewares( if agent_name == FORCED_RESEARCH_AGENT_ID: middlewares.append(ForcedResearchMiddleware()) + # 写作模式首轮：允许先聊/检索；只有模型开始写 md 时才改走 deep_research_report。 + # 不设 tool_choice——thinking 模式的兼容接口会 400。 + writing_mode = bool(cfg.get("writing_mode", False)) + writing_artifact_path = cfg.get("writing_artifact_path") if writing_mode else None + if writing_mode: + from deerflow.tools.builtins.deep_research_report_tool import is_writing_mode_report_path + + if not is_writing_mode_report_path(writing_artifact_path if isinstance(writing_artifact_path, str) else None): + from deerflow.agents.middlewares.writing_mode_report_middleware import WritingModeReportMiddleware + + middlewares.append(WritingModeReportMiddleware()) + # LoopDetectionMiddleware — detect and break repetitive tool call loops middlewares.append(LoopDetectionMiddleware()) @@ -517,6 +529,12 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): writing_artifact_path = cfg.get("writing_artifact_path") if writing_mode else None if not isinstance(writing_artifact_path, str) or not writing_artifact_path.strip(): writing_artifact_path = None + # 只有写作模式产物 report.md 才走续写；其它 leftover .md 不得跳过管线。 + if writing_artifact_path is not None: + from deerflow.tools.builtins.deep_research_report_tool import is_writing_mode_report_path + + if not is_writing_mode_report_path(writing_artifact_path): + writing_artifact_path = None agent_id = validate_agent_id(cfg.get("agent_id") or cfg.get("agent_name")) requested_excluded_tools = cfg.get("excluded_tools") or None @@ -691,10 +709,10 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): from deerflow.tools.builtins.notebook_search_tool import notebook_search_tool default_tools = [notebook_search_tool] + default_tools - # 写作模式首轮（尚无 md 产物）：挂深度研究报告工具，报告由管线生成而非 - # 模型手工 write_file。续写轮（已有 writing_artifact_path）不挂，保持 - # 原有「直接编辑该文件」行为。不受 tool_groups/excluded_tools 过滤—— - # 它是写作模式合约的核心交付手段。 + # 写作模式首轮（尚无 report.md）：挂深度研究报告工具，报告由管线生成而非 + # 模型手工 write_file。续写轮（已有 report.md）不挂，保持原有「直接编辑 + # 该文件」行为。不受 tool_groups/excluded_tools 过滤——它是写作模式合约 + # 的核心交付手段。 if writing_mode and not writing_artifact_path: from deerflow.tools.builtins.deep_research_report_tool import deep_research_report_tool @@ -726,8 +744,8 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): writing_mode=writing_mode, writing_artifact_path=writing_artifact_path, ) - else: - system_prompt = apply_prompt_template( + else: + system_prompt = apply_prompt_template( subagent_enabled=subagent_enabled, max_concurrent_subagents=max_concurrent_subagents, agent_id=agent_id, @@ -737,20 +755,20 @@ def _make_lead_agent(config: RunnableConfig, *, app_config: AppConfig): available_skills=prompt_available_skills, app_config=resolved_app_config, memory_injection_enabled=memory_injection_enabled, - rag_mode_enabled=rag_mode_enabled, - is_notebook_mode=is_notebook_mode, - notebook_source_ids=notebook_source_ids, - is_scheduled_run=bool(cfg.get("is_scheduled_run", False)), - ) - - _log_es_query_routing_registration( - system_prompt=system_prompt, - app_config=resolved_app_config, - agent_id=agent_id, - available_skills=prompt_available_skills, - ) - - return create_agent( + rag_mode_enabled=rag_mode_enabled, + is_notebook_mode=is_notebook_mode, + notebook_source_ids=notebook_source_ids, + is_scheduled_run=bool(cfg.get("is_scheduled_run", False)), + ) + + _log_es_query_routing_registration( + system_prompt=system_prompt, + app_config=resolved_app_config, + agent_id=agent_id, + available_skills=prompt_available_skills, + ) + + return create_agent( model=create_chat_model(name=model_name, thinking_enabled=thinking_enabled, reasoning_effort=reasoning_effort, force_disable_thinking=thinking_force_disabled, app_config=resolved_app_config), tools=default_tools, middleware=_build_middlewares(config, model_name=model_name, agent_name=agent_id, app_config=resolved_app_config), diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py index cb556124..623ca245 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py @@ -486,19 +486,19 @@ task(description="Oracle Cloud analysis", prompt="...", subagent_type="general-p """ -_ORDINARY_QA_MARKDOWN_FORMAT_SECTION = """ - -当普通问答需要输出带章节结构的较长内容时，必须采用以下 Markdown 规范，以便下载为 Word 后得到正确的自动编号和公文标题格式： -- 文档主标题使用 `# 主标题`；主标题只写标题文字，不添加章节编号。 -- 一级章节使用 `## 一、标题`，二级章节使用 `### (一) 标题`，三级章节使用 `#### 1. 标题`。二级编号必须使用半角括号 `(一)`。 -- 不要在标题中使用 `2.1`、`2.1.1` 等多级阿拉伯数字编号；章节编号直接按上述三种形式书写。 -- 默认禁止使用以 `-`、`*`、`+` 或 `•` 开头的无序圆点列表；并列内容优先改写为完整自然段，确需逐项表达时使用有序列表。只有用户明确要求无序列表时才使用。 -- 简短问答不必为了套用格式而强行添加标题；本规范只约束实际使用的标题和列表。 - -""" - - -SYSTEM_PROMPT_TEMPLATE = """ +_ORDINARY_QA_MARKDOWN_FORMAT_SECTION = """ + +当普通问答需要输出带章节结构的较长内容时，必须采用以下 Markdown 规范，以便下载为 Word 后得到正确的自动编号和公文标题格式： +- 文档主标题使用 `# 主标题`；主标题只写标题文字，不添加章节编号。 +- 一级章节使用 `## 一、标题`，二级章节使用 `### (一) 标题`，三级章节使用 `#### 1. 标题`。二级编号必须使用半角括号 `(一)`。 +- 不要在标题中使用 `2.1`、`2.1.1` 等多级阿拉伯数字编号；章节编号直接按上述三种形式书写。 +- 默认禁止使用以 `-`、`*`、`+` 或 `•` 开头的无序圆点列表；并列内容优先改写为完整自然段，确需逐项表达时使用有序列表。只有用户明确要求无序列表时才使用。 +- 简短问答不必为了套用格式而强行添加标题；本规范只约束实际使用的标题和列表。 + +""" + + +SYSTEM_PROMPT_TEMPLATE = """ You are {agent_name}, an internal super agent. @@ -624,15 +624,15 @@ You: "Deploying to staging..." [proceed] {acp_section} - -- Clear and Concise: Avoid over-formatting unless requested -- Natural Tone: Use paragraphs and prose, not bullet points by default -- Action-Oriented: Focus on delivering results, not explaining processes - - -{ordinary_qa_format_section} - - + +- Clear and Concise: Avoid over-formatting unless requested +- Natural Tone: Use paragraphs and prose, not bullet points by default +- Action-Oriented: Focus on delivering results, not explaining processes + + +{ordinary_qa_format_section} + + **CRITICAL: Always include citations when using web search results** - **When to Use**: MANDATORY after web_search, web_fetch, or any external information source @@ -792,8 +792,7 @@ def apply_agent_only_prompt_template( else "" ) if writing_mode and not writing_artifact_path: - # 与通用模板一致：对话行为照旧，交付时刻的「写 md」动作交给 - # deep_research_report 管线；资料不足时先经 ask_clarification 问用户。 + # 对话行为照旧；只有开始写 md 时才走 deep_research_report。资料不足也直接开写。 writing_mode_section = """ Writing mode is enabled for this run (first report turn — no existing artifact). - HARD RULE: the report file is produced ONLY by the `deep_research_report` tool — never write it yourself with `write_file`, never `present_files` it. @@ -801,11 +800,11 @@ Writing mode is enabled for this run (first report turn — no existing artifact `topic` = the user's subject copied faithfully, `focus` = the user's extra requirements (if any). The tool reuses the conversation's collected materials and writes the full report (general-analysis structure with references) to `/mnt/user-data/outputs/report.md`, presented automatically. -- If the tool reports insufficient materials, call `ask_clarification` to ask whether to search for more; after the user answers, - re-call the tool with the same `topic`/`focus` plus `retrieve=true` (search more) or `retrieve=false` (existing materials only). - NEVER pass `retrieve=true` before the user has answered — the tool rejects it and bounces you back to the question. +- If harvested materials are thin, still call `deep_research_report` once and let it write immediately. + Do NOT ask the user whether to search for more, and do NOT call `ask_clarification` for insufficient materials. +- Search with your normal conversation tools. `deep_research_report` only writes the report file from materials already in this conversation; it does not search. - Do NOT write the report body yourself with `write_file`, and do NOT call `present_files` for it. -- After the tool returns, reply briefly (2-3 sentences) summarizing the core conclusions; do not duplicate the report inline. +- After the tool returns successfully: do NOT call `read_file`, `write_file`, `str_replace`, or `present_files` on the report. Reply briefly (2-3 sentences) summarizing the core conclusions. Do not rewrite even if you think materials were thin. - Only if the tool fails, tell the user and offer to retry. """ @@ -1407,7 +1406,7 @@ def build_report_structure_setup_prompt( """ -def apply_prompt_template( +def apply_prompt_template( subagent_enabled: bool = False, max_concurrent_subagents: int = 3, *, @@ -1419,11 +1418,11 @@ def apply_prompt_template( is_bootstrap: bool = False, app_config: AppConfig | None = None, memory_injection_enabled: bool = True, - rag_mode_enabled: bool = False, - is_notebook_mode: bool = False, - notebook_source_ids: list[int] | None = None, - is_scheduled_run: bool = False, -) -> str: + rag_mode_enabled: bool = False, + is_notebook_mode: bool = False, + notebook_source_ids: list[int] | None = None, + is_scheduled_run: bool = False, +) -> str: storage_agent_id = agent_id if agent_id is not None else agent_name # Get memory context @@ -1473,9 +1472,8 @@ def apply_prompt_template( else "" ) if writing_mode and not writing_artifact_path: - # 首轮：对话行为一切照旧，仅在交付时刻把「写 md」动作交给 - # deep_research_report 管线（收割对话资料 + 通用分析结构）。资料不足时 - # 工具会返回指令，让模型经 ask_clarification 征询用户是否补充检索。 + # 对话行为一切照旧；只有开始写 md 时才把交付交给 deep_research_report。 + # 资料不足也直接开写，不再经 ask_clarification 征询用户是否补充检索。 writing_mode_section = """ Writing mode is enabled for this run (first report turn — no existing artifact). @@ -1487,14 +1485,13 @@ Writing mode is enabled for this run (first report turn — no existing artifact - When the report file is due — the user asks to start writing, or you have finished gathering what the conversation needs — call the `deep_research_report` tool: - `topic`: the user's subject/thesis, copied faithfully. - `focus`: the user's extra requirements (length, tone, must-cover points), if any. -- The tool reuses the materials already collected in this conversation (and searches more only if the user agrees), then writes the full +- The tool reuses the materials already collected in this conversation, then writes the full report (general-analysis structure with a numbered reference section) to `/mnt/user-data/outputs/report.md`, presented automatically. -- If the tool replies that the harvested materials are insufficient, it returns instructions: immediately call `ask_clarification` asking - whether to search for more materials, and after the user answers, re-call `deep_research_report` with the same `topic`/`focus` plus - `retrieve=true` (user agreed) or `retrieve=false` (user declined — write with what exists). - NEVER pass `retrieve=true` before the user has answered — the tool rejects it and bounces you back to the question. +- If harvested materials are thin, still call `deep_research_report` once and let it start the writing pipeline immediately. + Do NOT ask the user whether to search for more materials, and do NOT call `ask_clarification` for that reason. +- Search with your normal conversation tools. `deep_research_report` only writes the report file from materials already in this conversation; it does not search. - Do NOT write the report body yourself with `write_file`, and do NOT call `present_files` for the report — the tool already handles both. -- While the tool runs, stay silent; after it returns, reply briefly (2-3 sentences) summarizing the report's core conclusions and point the user to the Markdown artifact. Do not duplicate the report inline. +- While the tool runs, stay silent. After it returns successfully: do NOT call `read_file`, `write_file`, `str_replace`, or `present_files` on the report. Reply briefly (2-3 sentences) summarizing the report's core conclusions and point the user to the Markdown artifact. Do not duplicate the report inline, and do not rewrite even if you think materials were thin. - Only if the tool fails, tell the user what went wrong and offer to retry; do not silently fall back to writing the report by hand unless the user explicitly asks. """ @@ -1528,7 +1525,7 @@ Writing mode is enabled for this run. # Build the offline-network policy block (empty unless DEER_FLOW_OFFLINE_MODE is set). network_policy_section = _build_network_policy_section() - rag_mode_section = ( + rag_mode_section = ( """ RAG/reference display mode is enabled. @@ -1541,15 +1538,15 @@ When a search, knowledge, skill, or MCP tool returns a result list: """ if rag_mode_enabled - else "" - ) - ordinary_qa_format_section = ( - _ORDINARY_QA_MARKDOWN_FORMAT_SECTION - if not is_bootstrap and not writing_mode and not is_notebook_mode and not is_scheduled_run - else "" - ) - - # Format the prompt with dynamic skills and memory + else "" + ) + ordinary_qa_format_section = ( + _ORDINARY_QA_MARKDOWN_FORMAT_SECTION + if not is_bootstrap and not writing_mode and not is_notebook_mode and not is_scheduled_run + else "" + ) + + # Format the prompt with dynamic skills and memory prompt = SYSTEM_PROMPT_TEMPLATE.format( agent_name=agent_name or storage_agent_id or "CM 助手", soul=get_agent_soul(storage_agent_id), @@ -1562,10 +1559,10 @@ When a search, knowledge, skill, or MCP tool returns a result list: network_policy_section=network_policy_section, subagent_section=subagent_section, subagent_reminder=subagent_reminder, - subagent_thinking=subagent_thinking, - acp_section=acp_and_mounts_section, - ordinary_qa_format_section=ordinary_qa_format_section, - ) + subagent_thinking=subagent_thinking, + acp_section=acp_and_mounts_section, + ordinary_qa_format_section=ordinary_qa_format_section, + ) prompt = prompt + rag_mode_section + _current_time_context() diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/writing_mode_report_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/writing_mode_report_middleware.py new file mode 100644 index 00000000..a1ef9887 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/writing_mode_report_middleware.py @@ -0,0 +1,468 @@ +"""Redirect writing-mode Markdown deliverables onto ``deep_research_report``. + +Writing mode still allows normal chat, search, and clarification. The pipeline +starts only when the model begins writing the report file (``write_file`` / +``str_replace`` of ``report.md`` or any Markdown under +``/mnt/user-data/outputs/``): those calls are rewritten to +``deep_research_report``. Sibling ``present_files`` of the same report are +dropped because the pipeline presents the file itself. + +After the pipeline has succeeded once in this thread, further +``deep_research_report`` / hand-written report writes / ``read_file`` of the +report are dropped. The first time that happens, a hidden summary nudge is +injected and the graph jumps back to the model so it can write the 2–3 +sentence wrap-up. If it still tries to loop after that nudge, the turn ends. +Ending immediately on the first rewrite attempt left the model's +"I will rewrite this" text as the visible summary. + +This middleware never sets ``tool_choice``. Thinking-mode providers (Aliyun +compatible-mode) reject ``required`` / object ``tool_choice`` with HTTP 400. +""" + +from __future__ import annotations + +import json +import logging +from collections.abc import Awaitable, Callable +from copy import deepcopy +from typing import Any, override + +from langchain.agents import AgentState +from langchain.agents.middleware import AgentMiddleware, hook_config +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage +from langgraph.prebuilt.tool_node import ToolCallRequest +from langgraph.runtime import Runtime +from langgraph.types import Command + +from deerflow.tools.builtins.deep_research_report_tool import is_writing_mode_report_path + +logger = logging.getLogger(__name__) + +TOOL_NAME = "deep_research_report" +_WRITE_TOOLS = frozenset({"write_file", "str_replace"}) +_READ_TOOLS = frozenset({"read_file", "read_multiple_files"}) +_PRESENT_TOOL = "present_files" +_OUTPUTS_PREFIX = "/mnt/user-data/outputs" +_MAX_TOPIC_CHARS = 500 +_NUDGE_KWARG = "writing_mode_summary_nudge" +_REJECT_HANDWRITTEN = ( + "写作模式的报告文件必须由 `deep_research_report` 生成，禁止用 write_file / " + "str_replace 手写报告。请立即调用 `deep_research_report`（topic=用户课题原文）。" +) +_REJECT_AFTER_SUCCESS = ( + "写作管线已经完成，报告已写入 /mnt/user-data/outputs/report.md 并展示给用户。" + "禁止再次调用 deep_research_report，也禁止用 write_file / str_replace / " + "read_file 读写该报告。请只用 2-3 句话总结核心结论后结束本轮，不要再调用任何文件或写作工具。" +) +_SUMMARY_NUDGE = ( + "\n" + "写作管线已经完成，报告已写入 /mnt/user-data/outputs/report.md 并展示给用户。\n" + "请只用 2-3 句话总结核心结论，不要再调用任何工具，不要阅读或重写该文件。\n" + "不要评价材料是否不足，不要计划重写。\n" + "" +) + + +def _is_visible_human(message: Any) -> bool: + if not isinstance(message, HumanMessage): + return False + kwargs = getattr(message, "additional_kwargs", None) or {} + return not kwargs.get("hide_from_ui") + + +def _normalize_args(raw: object) -> dict[str, Any]: + if isinstance(raw, dict): + return raw + if isinstance(raw, str): + try: + parsed = json.loads(raw) + except (TypeError, ValueError, json.JSONDecodeError): + return {} + return parsed if isinstance(parsed, dict) else {} + return {} + + +def _tool_field(tool_call: object, key: str, default: Any = None) -> Any: + if isinstance(tool_call, dict): + return tool_call.get(key, default) + getter = getattr(tool_call, "get", None) + if callable(getter): + return getter(key, default) + return getattr(tool_call, key, default) + + +def _tool_name(tool_call: object) -> str: + return str(_tool_field(tool_call, "name", "") or "") + + +def _tool_path(tool_call: object) -> str: + args = _normalize_args(_tool_field(tool_call, "args", {})) + path = args.get("path") + if path is None: + files = args.get("files") or args.get("filepaths") or args.get("paths") + if isinstance(files, list) and files: + path = files[0] + return str(path or "").strip().replace("\\", "/") + + +def _is_outputs_markdown_path(path: str) -> bool: + if is_writing_mode_report_path(path): + return True + if not path: + return False + lowered = path.lower() + if not (lowered.endswith(".md") or lowered.endswith(".markdown")): + return False + if path == _OUTPUTS_PREFIX or path.startswith(f"{_OUTPUTS_PREFIX}/"): + return True + relative = path.lstrip("./") + return relative == "outputs" or relative.startswith("outputs/") + + +def is_handwritten_report_call(tool_call: object) -> bool: + """True when the model is about to hand-write the writing-mode Markdown file.""" + if _tool_name(tool_call) not in _WRITE_TOOLS: + return False + return _is_outputs_markdown_path(_tool_path(tool_call)) + + +def is_report_present_call(tool_call: object) -> bool: + if _tool_name(tool_call) != _PRESENT_TOOL: + return False + path = _tool_path(tool_call) + if path: + return _is_outputs_markdown_path(path) + args = _normalize_args(_tool_field(tool_call, "args", {})) + files = args.get("files") or args.get("filepaths") or args.get("paths") or [] + if isinstance(files, list): + return any(_is_outputs_markdown_path(str(item or "").strip().replace("\\", "/")) for item in files) + return False + + +def is_pipeline_call(tool_call: object) -> bool: + return _tool_name(tool_call) == TOOL_NAME + + +def is_report_reread_call(tool_call: object) -> bool: + """True when the model is about to re-read the writing-mode Markdown file.""" + if _tool_name(tool_call) not in _READ_TOOLS: + return False + path = _tool_path(tool_call) + if path: + return _is_outputs_markdown_path(path) + args = _normalize_args(_tool_field(tool_call, "args", {})) + files = args.get("files") or args.get("filepaths") or args.get("paths") or [] + if isinstance(files, list): + return any(_is_outputs_markdown_path(str(item or "").strip().replace("\\", "/")) for item in files) + return False + + +def is_deliverable_loop_call(tool_call: object) -> bool: + """Pipeline, hand-written report, or presenting the report — all loop fuel.""" + return is_pipeline_call(tool_call) or is_handwritten_report_call(tool_call) or is_report_present_call(tool_call) + + +def is_post_success_loop_call(tool_call: object) -> bool: + """After a successful pipeline, re-reading the report is also a loop.""" + return is_deliverable_loop_call(tool_call) or is_report_reread_call(tool_call) + + +def _already_nudged_summary(messages: list[Any]) -> bool: + for message in messages: + if not isinstance(message, HumanMessage): + continue + kwargs = getattr(message, "additional_kwargs", None) or {} + if kwargs.get(_NUDGE_KWARG): + return True + return False + + +def _summary_nudge_message() -> HumanMessage: + return HumanMessage( + content=_SUMMARY_NUDGE, + additional_kwargs={"hide_from_ui": True, _NUDGE_KWARG: True}, + ) + + +def is_post_success_loop_call(tool_call: object) -> bool: + """After a successful pipeline, re-reading the report is also a loop.""" + return is_deliverable_loop_call(tool_call) or is_report_reread_call(tool_call) + + +def _already_nudged_summary(messages: list[Any]) -> bool: + for message in messages: + if not isinstance(message, HumanMessage): + continue + kwargs = getattr(message, "additional_kwargs", None) or {} + if kwargs.get(_NUDGE_KWARG): + return True + return False + + +def _summary_nudge_message() -> HumanMessage: + return HumanMessage( + content=_SUMMARY_NUDGE, + additional_kwargs={"hide_from_ui": True, _NUDGE_KWARG: True}, + ) + + +def _report_tool_succeeded(message: ToolMessage) -> bool: + if str(getattr(message, "name", "") or "") != TOOL_NAME: + return False + status = str(getattr(message, "status", "") or "").lower() + if status == "error": + return False + content = str(getattr(message, "content", "") or "").strip() + if not content: + return False + return not content.lower().startswith("error") + + +def pipeline_already_succeeded(messages: list[Any]) -> bool: + """True after any successful ``deep_research_report`` in this thread.""" + return any(isinstance(message, ToolMessage) and _report_tool_succeeded(message) for message in messages) + + +def _message_text(message: Any) -> str: + content = getattr(message, "content", "") + if isinstance(content, list): + parts: list[str] = [] + for part in content: + if isinstance(part, str): + parts.append(part) + elif isinstance(part, dict): + text = part.get("text") + if isinstance(text, str): + parts.append(text) + return " ".join(parts).strip() + return str(content or "").strip() + + +def extract_report_topic(messages: list[Any]) -> str: + """Use the latest visible user text as ``deep_research_report.topic``.""" + for message in reversed(messages): + if _is_visible_human(message): + text = " ".join(_message_text(message).split()) + if text: + return text[:_MAX_TOPIC_CHARS] + return "当前课题" + + +def _as_report_call(tool_call: object, topic: str) -> dict[str, Any]: + call_id = str(_tool_field(tool_call, "id") or "") + return { + "name": TOOL_NAME, + "args": {"topic": topic}, + "id": call_id, + "type": "tool_call", + } + + +def _kept_call_ids(calls: list[Any]) -> set[str]: + return {str(_tool_field(call, "id") or "") for call in calls if _tool_field(call, "id")} + + +def _sync_raw_tool_calls(last: AIMessage, kept: list[Any]) -> dict[str, Any]: + update: dict[str, Any] = {"tool_calls": kept} + additional_kwargs = dict(getattr(last, "additional_kwargs", {}) or {}) + raw_calls = additional_kwargs.get("tool_calls") + kept_ids = _kept_call_ids(kept) + rewritten_by_id = { + str(_tool_field(call, "id") or ""): call + for call in kept + if _tool_field(call, "id") + } + if isinstance(raw_calls, list) and raw_calls: + synced: list[Any] = [] + for item in raw_calls: + if not isinstance(item, dict): + continue + item_id = str(item.get("id") or "") + if item_id not in kept_ids: + continue + replacement = rewritten_by_id.get(item_id) + if replacement is not None and _tool_name(replacement) == TOOL_NAME: + function = dict(item.get("function") or {}) + function["name"] = TOOL_NAME + function["arguments"] = json.dumps( + _normalize_args(_tool_field(replacement, "args", {})), + ensure_ascii=False, + ) + synced.append({**item, "function": function}) + else: + synced.append(item) + if synced: + additional_kwargs["tool_calls"] = synced + else: + additional_kwargs.pop("tool_calls", None) + additional_kwargs.pop("function_call", None) + elif not kept: + additional_kwargs.pop("tool_calls", None) + additional_kwargs.pop("function_call", None) + update["additional_kwargs"] = additional_kwargs + if not kept: + response_metadata = deepcopy(getattr(last, "response_metadata", {}) or {}) + if response_metadata.get("finish_reason") == "tool_calls": + response_metadata["finish_reason"] = "stop" + update["response_metadata"] = response_metadata + return update + + +def rewrite_handwritten_report_calls( + last: AIMessage, + *, + topic: str, + pipeline_already_done: bool = False, +) -> AIMessage | None: + """Rewrite or drop report deliverable tool calls. + + Before the pipeline has succeeded: rewrite the first hand-written report + write to ``deep_research_report`` and drop sibling presents. + + After it has succeeded: drop every pipeline / hand-written / present / + report-reread call so the model cannot start another writing loop. + """ + tool_calls = list(getattr(last, "tool_calls", None) or []) + if not tool_calls: + return None + + if pipeline_already_done: + kept = [call for call in tool_calls if not is_post_success_loop_call(call)] + if len(kept) == len(tool_calls): + return None + update = _sync_raw_tool_calls(last, kept) + if not kept: + update["content"] = "" + return last.model_copy(update=update) + + has_pipeline = any(is_pipeline_call(call) for call in tool_calls) + rewritten: list[Any] = [] + changed = False + redirected = has_pipeline + for call in tool_calls: + if is_pipeline_call(call): + rewritten.append(call) + continue + if is_handwritten_report_call(call): + changed = True + if redirected: + continue + rewritten.append(_as_report_call(call, topic)) + redirected = True + continue + if is_report_present_call(call) and ( + has_pipeline or redirected or any(is_handwritten_report_call(item) for item in tool_calls) + ): + changed = True + continue + rewritten.append(call) + + if not changed: + return None + return last.model_copy(update=_sync_raw_tool_calls(last, rewritten)) + + +def _messages_from_tool_request(request: ToolCallRequest) -> list[Any]: + state = getattr(request, "state", None) or {} + if isinstance(state, dict): + return list(state.get("messages") or []) + messages = getattr(state, "messages", None) + return list(messages or []) + + +class WritingModeReportMiddleware(AgentMiddleware[AgentState]): + """Start the writing pipeline only when the model begins writing the Markdown file.""" + + state_schema = AgentState + + def _redirect_or_stop(self, state: AgentState) -> dict[str, Any] | None: + messages = list(state.get("messages") or []) + if not messages: + return None + last = messages[-1] + if not isinstance(last, AIMessage): + return None + already_done = pipeline_already_succeeded(messages) + patched = rewrite_handwritten_report_calls( + last, + topic=extract_report_topic(messages), + pipeline_already_done=already_done, + ) + if patched is None: + return None + result: dict[str, Any] = {"messages": [patched]} + if already_done and not (patched.tool_calls or []): + if _already_nudged_summary(messages): + logger.info( + "Writing-mode pipeline already succeeded; summary nudge ignored, ending turn" + ) + result["jump_to"] = "end" + return result + logger.info( + "Writing-mode pipeline already succeeded; dropping deliverable tools and nudging summary" + ) + result["messages"] = [patched, _summary_nudge_message()] + result["jump_to"] = "model" + return result + + def _reject_loop_tool(self, request: ToolCallRequest) -> ToolMessage | None: + tool_call = getattr(request, "tool_call", None) or {} + already_done = pipeline_already_succeeded(_messages_from_tool_request(request)) + if already_done and is_post_success_loop_call(tool_call): + return ToolMessage( + content=f"Error: {_REJECT_AFTER_SUCCESS}", + tool_call_id=str(_tool_field(tool_call, "id") or ""), + name=str(_tool_name(tool_call) or TOOL_NAME), + ) + if not already_done and is_handwritten_report_call(tool_call): + return ToolMessage( + content=f"Error: {_REJECT_HANDWRITTEN}", + tool_call_id=str(_tool_field(tool_call, "id") or ""), + name=str(_tool_name(tool_call) or "write_file"), + ) + return None + + @hook_config(can_jump_to=["end", "model"]) + @override + def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: + return self._redirect_or_stop(state) + + @hook_config(can_jump_to=["end", "model"]) + @override + async def aafter_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: + return self._redirect_or_stop(state) + + @override + def wrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], ToolMessage | Command], + ) -> ToolMessage | Command: + rejected = self._reject_loop_tool(request) + if rejected is not None: + return rejected + return handler(request) + + @override + async def awrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]], + ) -> ToolMessage | Command: + rejected = self._reject_loop_tool(request) + if rejected is not None: + return rejected + return await handler(request) + + +__all__ = [ + "TOOL_NAME", + "WritingModeReportMiddleware", + "extract_report_topic", + "is_handwritten_report_call", + "is_post_success_loop_call", + "is_report_present_call", + "is_report_reread_call", + "pipeline_already_succeeded", + "rewrite_handwritten_report_calls", +] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/deep_research_report_tool.py b/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/deep_research_report_tool.py index 596b57ac..58d81783 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/deep_research_report_tool.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/deep_research_report_tool.py @@ -7,19 +7,12 @@ ``outputs/`` 目录——前端画布 / 编辑 / 下载 / 续写全部沿用既有 artifacts 通道。 除「写 md」这一动作换成管线外，agent 的对话 / 检索 / 澄清行为一切照旧。 -资料来源三分支（收割池先按**当前主题**做词面相关度强过滤——旧课题的检索 -结果既不计入门槛，也不进报告）： +资料来源（收割池先按**当前主题**做词面相关度强过滤——旧课题的检索 +结果不进报告）： -- 主题相关池足够（≥ :data:`_MIN_HARVESTED_MATERIALS`）→ 直接用对话资料写； -- 不足且用户已答复 → ``retrieve=True`` 自检索（对话资料合并去重）、 - ``retrieve=False`` 只用现有资料（哪怕为空——basic runner 有优雅兜底）； -- 不足且未询问 → 返回指令 ToolMessage，让模型用 ``ask_clarification`` - 询问用户是否补充检索，用户答复后带 ``retrieve`` 重调本工具。 - -``retrieve=True`` 不只是提示词约束：工具会核验消息史里存在 -「本工具的资料不足答复 → ask_clarification → 用户回复」完整链 -（:func:`_retrieval_confirmed_by_user`），模型擅自传 ``retrieve=True`` -会被拒绝并打回询问流程。 +- 一律用对话里已经产生的资料写（哪怕为空——basic runner 有领域知识兜底）， + **不再**因条数不足打断用户，也**不再**在管线内部补检索。 + 需要检索时由 lead agent 用对话工具先搜，再调用本工具写文件。 报告管线继承聊天里选择的模型与思考开关（``context.model_name`` / ``thinking_enabled`` / ``thinking_force_disabled``，经 run config 白名单透 @@ -60,12 +53,11 @@ from deerflow.agents.deep_research.events import DeepResearchEvent, EventType, R from deerflow.agents.deep_research.types import AdapterBundle, DeepResearchRequest, ResearchMaterial from deerflow.agents.thread_state import ThreadState from deerflow.persistence.report_structures.defaults import GENERAL_OUTLINE -from deerflow.runtime.references import _message_name, _message_type from deerflow.runtime.thread_paths import resolve_path_user_id logger = logging.getLogger(__name__) -# 收割池低于该条数视为「资料不足」，需要先征询用户是否补充检索。 +# 收割池低于该条数视为「资料偏少」，仅记日志，不再打断用户。 _MIN_HARVESTED_MATERIALS = 3 # 引擎事件中值得转发到聊天的子集（model_usage / heartbeat / job_* 只对持久 @@ -96,6 +88,17 @@ _FORWARDABLE_LIVE_EVENTS: frozenset[str] = frozenset({"report_delta", "report_th REPORT_VIRTUAL_PATH = "/mnt/user-data/outputs/report.md" +def is_writing_mode_report_path(path: str | None) -> bool: + """Only ``report.md`` counts as the writing-mode deliverable for follow-up edits. + + Leftover notes / skill Markdown must not skip the first-report pipeline. + """ + if not path or not isinstance(path, str): + return False + normalized = path.replace("\\", "/").split("?", 1)[0].split("#", 1)[0].rstrip("/") + return normalized.lower().endswith("/report.md") or normalized.lower() == "report.md" + + class StreamWriterEventSink: """把引擎事件经 LangGraph custom 通道转发到聊天前端。 @@ -211,6 +214,15 @@ class HarvestPlusSearchMaterialProvider: return dedupe_materials(combined) +# 写作模式常在「不用搜索 / 对话资料很少」时开写。工作台默认的「材料不足就 +# 写明资料不足」会把报告写成元分析；这里强制用领域知识补全正文。 +_KNOWLEDGE_WRITE_INSTRUCTION = ( + "对话材料很少或用户要求直接开写时：必须基于模型自身的领域知识写成完整、有实质内容的研究报告" + "（含架构说明；用户要求流程图时用 Mermaid）。禁止把「材料不足/未提供文献/现有材料高度浓缩」" + "写成报告主线或用其填充多个章节。参考文献可标为基于公开领域知识，不要用「材料未提供」占整节。" +) + + def build_writing_research_config( *, focus: str | None = None, @@ -224,7 +236,10 @@ def build_writing_research_config( 覆盖 fast/smart/strategic 三个角色，使报告管线与对话用同一个模型和思考 开关（缺省回落系统默认模型）。 """ - instruction = focus.strip() if focus and focus.strip() else None + parts: list[str] = [] + if focus and focus.strip(): + parts.append(focus.strip()) + parts.append(_KNOWLEDGE_WRITE_INSTRUCTION) return DeepResearchConfig( mode="basic", collection_mode="legacy", @@ -235,7 +250,7 @@ def build_writing_research_config( fast_model=model_name, smart_model=model_name, strategic_model=model_name, - **({"report_instruction": instruction} if instruction else {}), + report_instruction="\n".join(parts), ).clamp() @@ -325,33 +340,6 @@ def _filter_relevant_materials( return relevant, len(materials) - len(relevant) -# ── retrieve=True 的用户确认核验（P1：不靠提示词，消息史里找证据链） ────────── - - -def _retrieval_confirmed_by_user(messages: list[Any]) -> bool: - """核验消息史存在「本工具资料不足答复 → ask_clarification → 用户回复」。 - - 模型无法伪造 ToolMessage / HumanMessage（分别由工具执行与真实用户输入 - 产生），因此该链成立 ⇔ 用户确实被问过并答过。跨 run 依然有效——消息 - 历史就是 LangGraph 的持久状态。 - """ - last_report_reply = -1 - for index, message in enumerate(messages): - if _message_type(message) == "tool" and str(_message_name(message) or "") == "deep_research_report": - last_report_reply = index - if last_report_reply < 0: - return False - asked = False - for message in messages[last_report_reply + 1 :]: - if _message_type(message) != "tool": - if asked and _message_type(message) == "human": - return True - continue - if str(_message_name(message) or "") == "ask_clarification": - asked = True - return False - - def _safe_stream_writer() -> Any: """获取当前 run 的 custom 事件写入器；非流式上下文返回 None。""" try: @@ -362,45 +350,6 @@ def _safe_stream_writer() -> Any: return None -def _insufficient_materials_message( - relevant_count: int, - minimum: int, - *, - topic: str, - total_count: int, - dropped_count: int, -) -> str: - detail = ( - f"对话中共整理出 {total_count} 条资料，其中与当前主题「{topic}」相关的仅 {relevant_count} 条" - if dropped_count > 0 - else f"对话中仅整理出 {relevant_count} 条与主题「{topic}」相关的资料" - ) - return ( - f"[需要用户确认] {detail}（低于 {minimum} 条的最低参考线）。" - "在生成报告前，请立即调用 `ask_clarification` 询问用户：\n" - 'question: "对话中已有的资料较少，是否自动检索更多资料来写报告？"\n' - 'clarification_type: "missing_info"\n' - f'context: "当前对话仅整理出 {relevant_count} 条与主题相关的资料；补充检索可让报告更充实，但会多花一些时间。"\n' - 'options: ["检索更多资料，再写报告", "不用检索，直接用现有资料写"]\n' - "用户选择检索 → 携带 retrieve=true 重新调用本工具（参数 topic/focus 保持不变）；" - "用户选择不检索 → 携带 retrieve=false 重新调用本工具。" - "在得到用户答复前，不要自行检索，也不要用 write_file 手工撰写报告。" - "注意：未经上述询问并获得用户答复，本工具会拒绝 retrieve=true。" - ) - - -def _retrieve_without_confirmation_message() -> str: - return ( - "[参数被拒绝] retrieve=true 只允许在用户通过 ask_clarification 明确同意补充检索之后使用；" - "本次调用未检测到「询问用户 → 用户答复」的记录，已按资料不足处理。" - "请立即调用 `ask_clarification` 询问用户是否检索更多资料" - '（question: "对话中已有的资料较少，是否自动检索更多资料来写报告？"，' - 'clarification_type: "missing_info"，options: ["检索更多资料，再写报告", "不用检索，直接用现有资料写"]），' - "用户答复后再按其选择重新调用本工具（retrieve=true / false）。" - "不要不经询问直接传 retrieve=true，也不要用 write_file 手工撰写报告。" - ) - - @tool("deep_research_report", parse_docstring=True) async def deep_research_report_tool( runtime: ToolRuntime[ContextT, ThreadState], @@ -423,20 +372,18 @@ async def deep_research_report_tool( automatically; do NOT write the report body yourself with write_file. If the harvested pool is thin (or little of it matches the current topic), the - tool returns instructions to ask the user whether to search for more materials - (via ask_clarification); after the user answers, re-call this tool with - `retrieve` set accordingly (same topic/focus). `retrieve=true` is REJECTED - unless the conversation shows that ask_clarification exchange — never pass it - speculatively. + tool still starts the writing pipeline immediately with whatever relevant + materials exist — do NOT ask the user whether to search for more, and do NOT + call ask_clarification for that reason. Search in the conversation first with + your normal tools; this tool only writes the file. Args: topic: The research topic / thesis to investigate. Copy the user's subject faithfully; do not paraphrase it into something narrower unless the user asked for that scope. focus: Optional extra write-time requirements from the user (length, tone, must-cover points, language preferences). Omit when the user gave none. - retrieve: Only set on a re-call after the user answered the insufficient-materials - question. True = search more materials (combined with the conversation pool); - False = write directly with existing materials only. Omit on the first call. + retrieve: Ignored. Kept for older callers; the writing-mode pipeline never + searches. False / omitted / True all write from conversation materials only. tool_call_id: Injected tool call id. """ # 与 present_file_tool 相同的解析顺序：runtime.context → runtime.config。 @@ -473,63 +420,22 @@ async def deep_research_report_tool( dropped_count, ) - if retrieve is True and not _retrieval_confirmed_by_user(conversation_messages): - # P1：retrieve=true 只认「已询问且用户已答复」，提示词约束之外在工具层强制。 - logger.warning( - "[deep_research_report] thread=%s rejected retrieve=true without user confirmation", - thread_id, - ) - return Command( - update={ - "messages": [ - ToolMessage( - content=_retrieve_without_confirmation_message(), - tool_call_id=tool_call_id, - name="deep_research_report", - ) - ] - } - ) - - if harvested_count < _MIN_HARVESTED_MATERIALS and retrieve is None: - # 资料不足且未征询过用户：不静默自检索，交给模型发起用户协助。 + if harvested_count < _MIN_HARVESTED_MATERIALS: logger.info( - "[deep_research_report] thread=%s insufficient materials (%d < %d); asking user", + "[deep_research_report] thread=%s thin materials (%d < %d); writing immediately", thread_id, harvested_count, _MIN_HARVESTED_MATERIALS, ) - return Command( - update={ - "messages": [ - ToolMessage( - content=_insufficient_materials_message( - harvested_count, - _MIN_HARVESTED_MATERIALS, - topic=query, - total_count=len(harvested), - dropped_count=dropped_count, - ), - tool_call_id=tool_call_id, - name="deep_research_report", - ) - ] - } - ) - if harvested_count >= _MIN_HARVESTED_MATERIALS or retrieve is False: - materials_provider: Any = StaticHarvestMaterialProvider(materials) - material_source = "对话资料" - else: # retrieve is True 且已核验用户同意补充检索 - materials_provider = HarvestPlusSearchMaterialProvider( - materials, - DeerFlowMaterialProvider( - user_id=resolve_path_user_id(str(thread_id)), - session_id=f"chat-{thread_id}", - runtime_thread_id=str(thread_id), - ), + if retrieve is True: + logger.info( + "[deep_research_report] thread=%s ignoring retrieve=true; " + "writing-mode pipeline writes from conversation materials only", + thread_id, ) - material_source = "对话资料+补充检索" + materials_provider: Any = StaticHarvestMaterialProvider(materials) + material_source = "对话资料" user_id = resolve_path_user_id(str(thread_id)) session_id = f"chat-{thread_id}" @@ -635,7 +541,9 @@ async def deep_research_report_tool( __all__ = [ "HarvestPlusSearchMaterialProvider", + "REPORT_VIRTUAL_PATH", "StreamWriterEventSink", "build_writing_research_config", "deep_research_report_tool", + "is_writing_mode_report_path", ]