From e58da3fbdf12922731991f865ef9d8fdc2c04ea6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Thu, 3 Sep 2026 20:06:54 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BB=A3=E7=A0=81=E6=8F=90=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../workspace/messages/message-list.tsx | 7 +- .../src/core/threads/use-multi-round-rag.ts | 200 +++++++++++++++--- frontend-web/src/pages/ScheduledTasksPage.tsx | 49 +++-- .../deerflow/runtime/scheduler/service.py | 41 ++-- 4 files changed, 240 insertions(+), 57 deletions(-) diff --git a/frontend-web/src/components/workspace/messages/message-list.tsx b/frontend-web/src/components/workspace/messages/message-list.tsx index aee530cf..577d844a 100644 --- a/frontend-web/src/components/workspace/messages/message-list.tsx +++ b/frontend-web/src/components/workspace/messages/message-list.tsx @@ -563,7 +563,12 @@ export function MessageList({ questionMarkdown={ msg.id ? questionMarkdownByAssistantId.get(msg.id) : undefined } - isLoading={messageIsLoading} + isLoading={ + isMultiRoundRagMessage(msg) + ? thread.isLoading && + msg.id === messages[messages.length - 1]?.id + : messageIsLoading + } threadId={threadId} tokenUsageEnabled={tokenUsageEnabled} displaySources={displaySources} diff --git a/frontend-web/src/core/threads/use-multi-round-rag.ts b/frontend-web/src/core/threads/use-multi-round-rag.ts index 09b11942..90627604 100644 --- a/frontend-web/src/core/threads/use-multi-round-rag.ts +++ b/frontend-web/src/core/threads/use-multi-round-rag.ts @@ -181,17 +181,110 @@ function lastRound(trace: MultiRoundRagTrace): MultiRoundRagRound { return next; } -function parseEventLine(line: string): MultiRoundEvent | null { +function eventsFromLine(line: string): MultiRoundEvent[] { const trimmed = line.trim(); - if (!trimmed.startsWith("data:")) return null; - const payload = trimmed.replace(/^data:\s?/, "").trim(); - if (!payload || payload === "[DONE]") return { event: "done" }; - try { - const parsed = parseJson(payload); - return asRecord(parsed) as MultiRoundEvent | null; - } catch { + if (!trimmed || trimmed.startsWith(":")) return []; + let payload = trimmed; + if (/^data:\s?/i.test(trimmed)) { + payload = trimmed.replace(/^data:\s?/i, "").trim(); + } else if (/^(id|event|retry):/i.test(trimmed)) { + return []; + } + if (!payload) return []; + if (payload === "[DONE]") return [{ event: "done" }]; + if (payload.startsWith("{") || payload.startsWith("[")) { + try { + return expandStreamEvents(JSON.parse(payload)); + } catch { + return []; + } + } + return expandStreamEvents(payload); +} + +function asStreamEvent(value: unknown): MultiRoundEvent | null { + const parsed = parseJson(value); + if (Array.isArray(parsed)) { return null; } + const record = asRecord(parsed); + if (!record) { + if (typeof parsed === "string" && parsed.trim() && !/^[{[]/.test(parsed.trim())) { + return { event: "message", data: parsed }; + } + return null; + } + const eventName = asText(record.event ?? record.type ?? record.event_type); + if (eventName) { + return { + event: eventName, + data: record.data !== undefined ? record.data : (record.payload ?? record), + }; + } + if (record.tool_name || record.name) { + return { event: "tool_call", data: record }; + } + return record; +} + +function expandStreamEvents(value: unknown): MultiRoundEvent[] { + const parsed = parseJson(value); + if (Array.isArray(parsed)) { + return parsed.flatMap((item) => expandStreamEvents(item)); + } + const event = asStreamEvent(parsed); + return event ? [event] : []; +} + +function extractJsonObjects(raw: string): { objects: string[]; rest: string } { + const objects: string[] = []; + let depth = 0; + let stringQuote: '"' | "'" | null = null; + let escaped = false; + let start = -1; + for (let index = 0; index < raw.length; index += 1) { + const char = raw[index]!; + if (stringQuote) { + if (escaped) escaped = false; + else if (char === "\\") escaped = true; + else if (char === stringQuote) stringQuote = null; + continue; + } + if (char === '"' || char === "'") { + stringQuote = char; + continue; + } + if (char === "{") { + if (depth === 0) start = index; + depth += 1; + } else if (char === "}") { + depth -= 1; + if (depth === 0 && start !== -1) { + objects.push(raw.slice(start, index + 1)); + start = -1; + } + } + } + return depth > 0 && start !== -1 + ? { objects, rest: raw.slice(start) } + : { objects, rest: "" }; +} + +function pullStreamEvents(buffer: string): { events: MultiRoundEvent[]; rest: string } { + const events: MultiRoundEvent[] = []; + const lines = buffer.split(/\r?\n/); + const restLine = lines.pop() ?? ""; + for (const line of lines) { + events.push(...eventsFromLine(line)); + } + if (events.length > 0) { + return { events, rest: restLine }; + } + const extracted = extractJsonObjects(buffer); + for (const jsonText of extracted.objects) { + events.push(...expandStreamEvents(parseJson(jsonText))); + } + return { events, rest: extracted.rest || restLine }; } function mapReferenceToSource(item: unknown, index: number): MultiRoundRagSource { @@ -230,6 +323,22 @@ function extractInfoText(result: unknown): string { ); } +function mergeThink(existing: string, incoming: string) { + if (!incoming) return existing; + if (!existing) return incoming; + if (incoming.startsWith(existing)) return incoming; + if (existing.endsWith(incoming)) return existing; + return `${existing}${incoming}`; +} + +function extractThinkContent(chatText: string) { + const closed = chatText.match(/([\s\S]*?)<\/think>/i); + if (closed?.[1] != null) return closed[1].trim(); + const open = chatText.match(/([\s\S]*)$/i); + if (open?.[1] != null) return open[1].trim(); + return chatText.replace(/<\/?think>/gi, "").trim(); +} + /** * GeneralQA `handleToolCall` 的独立实现：按 tool_name 把 SSE 工具事件填进最近一轮 trace。 * `show_text` / `show-text` 负责思考与最终回答；检索类工具先写检索词，再把 result 渲染成卡片。 @@ -245,9 +354,21 @@ export function applyMultiRoundToolCall(trace: MultiRoundRagTrace, toolValue: un switch (toolName) { case "show_text": { const chatText = asText(input.text ?? input.content); + if (!chatText) return; if (chatText.includes("")) { - const thinkContent = chatText.replace("", "").replace("", "").trim(); - if (trace.rounds[0]?.title !== "正在准备检索") { + const thinkContent = extractThinkContent(chatText); + const firstRound = trace.rounds[0]; + const stillOnFirstThink = + firstRound?.title === "正在准备检索" || + ((!round.query && round.retrievals.length === 0) && + (round.title === "思考过程" || round.title === "思考中" || round.title === "正在准备检索")); + if (stillOnFirstThink && firstRound) { + firstRound.title = "思考过程"; + firstRound.thoughtMarkdown = mergeThink(firstRound.thoughtMarkdown, thinkContent); + } else if (!round.query && round.retrievals.length === 0) { + round.title = round.title === "正在准备检索" ? "思考过程" : round.title; + round.thoughtMarkdown = mergeThink(round.thoughtMarkdown, thinkContent); + } else { round.queryLoading = false; round.queryShow = round.retrievals.length > 0; trace.rounds.push({ @@ -259,9 +380,6 @@ export function applyMultiRoundToolCall(trace: MultiRoundRagTrace, toolValue: un queryShow: true, retrievals: [], }); - } else { - trace.rounds[0]!.title = "思考过程"; - trace.rounds[0]!.thoughtMarkdown = thinkContent; } } else { round.queryLoading = false; @@ -515,8 +633,20 @@ export function applyMultiRoundToolCall(trace: MultiRoundRagTrace, toolValue: un function consumeToolCallPayload(trace: MultiRoundRagTrace, payload: unknown) { const parsed = parseJson(payload); - const tools = Array.isArray(parsed) ? parsed : [parsed]; - for (const tool of tools) applyMultiRoundToolCall(trace, tool); + if (Array.isArray(parsed)) { + for (const tool of parsed) applyMultiRoundToolCall(trace, tool); + return; + } + const record = asRecord(parsed); + if (record?.tool_name || record?.name) { + applyMultiRoundToolCall(trace, record); + return; + } + if (record?.data != null) { + consumeToolCallPayload(trace, record.data); + return; + } + applyMultiRoundToolCall(trace, parsed); } function textFromEventData(value: unknown): string { @@ -532,7 +662,7 @@ function applyStreamEvent( onSources?: (sources: MultiRoundRagSource[]) => void, onError?: (message: string) => void, ) { - const eventType = asText(event.event); + const eventType = normalizeToolName(asText(event.event)); const data = event.data; if (eventType === "tool_call") { consumeToolCallPayload(trace, data); @@ -560,6 +690,8 @@ function applyStreamEvent( const finalText = textFromEventData(data); if (finalText && !trace.finalAnswerMarkdown) trace.finalAnswerMarkdown += finalText; finishTrace(trace); + } else if (asRecord(data)?.tool_name || asRecord(parseJson(data))?.tool_name) { + consumeToolCallPayload(trace, data); } } @@ -579,7 +711,10 @@ async function* readMultiRoundSse( ): AsyncGenerator { const response = await fetch(`${baseUrl.replace(/\/$/, "")}/generate`, { method: "POST", - headers: { "Content-Type": "application/json" }, + headers: { + "Content-Type": "application/json", + Accept: "text/event-stream, application/json, text/plain", + }, signal, body: JSON.stringify({ prompt: question }), }); @@ -596,17 +731,25 @@ async function* readMultiRoundSse( const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split(/\r?\n/); - buffer = lines.pop() ?? ""; - for (const line of lines) { - const event = parseEventLine(line); - if (event) yield event; + const pulled = pullStreamEvents(buffer); + buffer = pulled.rest; + for (const event of pulled.events) { + yield event; + if (normalizeToolName(asText(event.event)) === "done") return; } } buffer += decoder.decode(); - for (const line of buffer.split(/\r?\n/)) { - const event = parseEventLine(line); - if (event) yield event; + const pulled = pullStreamEvents(`${buffer}\n`); + for (const event of pulled.events) { + yield event; + } + if (pulled.rest.trim()) { + for (const event of [ + ...eventsFromLine(pulled.rest), + ...expandStreamEvents(parseJson(pulled.rest)), + ]) { + yield event; + } } } @@ -622,8 +765,10 @@ export function useMultiRoundRagMessages(threadId: string) { const [isLoading, setIsLoading] = useState(false); const [error, setError] = useState(null); const controllerRef = useRef(null); + const inFlightThreadIdRef = useRef(null); useEffect(() => { + if (inFlightThreadIdRef.current === threadId) return; const stored = readStoredMessages(threadId); messagesRef.current = stored; setMessages(stored); @@ -683,6 +828,7 @@ export function useMultiRoundRagMessages(threadId: string) { }; updateMessages(targetThreadId, (current) => [...current, userMessage, assistantMessage]); + inFlightThreadIdRef.current = targetThreadId; setError(null); setIsLoading(true); @@ -717,6 +863,7 @@ export function useMultiRoundRagMessages(threadId: string) { setError, ); refreshAssistant(); + if (normalizeToolName(asText(event.event)) === "done") break; } finishTrace(trace); refreshAssistant(); @@ -729,6 +876,9 @@ export function useMultiRoundRagMessages(threadId: string) { } refreshAssistant(); } finally { + if (inFlightThreadIdRef.current === targetThreadId) { + inFlightThreadIdRef.current = null; + } if (controllerRef.current === controller) controllerRef.current = null; setIsLoading(false); } diff --git a/frontend-web/src/pages/ScheduledTasksPage.tsx b/frontend-web/src/pages/ScheduledTasksPage.tsx index ca50aefb..59b091b6 100644 --- a/frontend-web/src/pages/ScheduledTasksPage.tsx +++ b/frontend-web/src/pages/ScheduledTasksPage.tsx @@ -138,8 +138,9 @@ const emptyForm = { html_negative_prompt: "", html_reference_favorite_id: "", html_model_name: "", - // 参考选了内置模板(tpl:)时的「数据抽取」开关：关闭=用模板默认数据，开启=检索抽取。 - html_template_extract: false, + // 参考选了内置模板(tpl:)时的「数据抽取」开关：默认开启(检索抽取+抽不到回退 + // 模板原数据)，仅可转换模板生效；显式关闭=直接用模板默认数据。 + html_template_extract: true, }; /** 覆盖 Textarea 默认 field-sizing-content，避免输入过多时撑破弹窗 */ @@ -331,7 +332,8 @@ export default function ScheduledTasksPage({ html_negative_prompt: html.negative_prompt ?? "", html_reference_favorite_id: html.reference_favorite_id ?? "", html_model_name: html.model_name ?? "", - html_template_extract: !!html.template_extract, + // 与后端缺省一致：未存该键(老任务)视为开启，仅显式 false 才关闭。 + html_template_extract: html.template_extract !== false, }); // Prefer the canonical cron_expr over the user-typed schedule_text: the // latter may be natural language ("每天9点") that parseCron can't read, @@ -358,6 +360,10 @@ export default function ScheduledTasksPage({ isTemplateAgent || form.require_html, ); const selectedTemplate = htmlTemplates.find((t) => t.id === form.template_id); + // 「参考收藏页面」选中的内置模板(require_html 模式,值形如 tpl:)。 + const selectedTplRef = htmlTemplates.find( + (t) => `tpl:${t.id}` === form.html_reference_favorite_id, + ); // 选中的模板是「只展示」型(convertible===false)时,不需要源 JSON,隐藏上传/输入。 const templateDisplayOnly = isTemplateAgent && selectedTemplate?.convertible === false; @@ -1396,23 +1402,28 @@ export default function ScheduledTasksPage({ )} - {/* 参考选了内置模板(tpl:)时：数据抽取开关。关闭=用模板默认数据。 */} - {form.html_reference_favorite_id.startsWith("tpl:") && ( -
-
-

执行数据抽取

-

- 关闭则直接用所选模板的默认数据生成（含全部子页面）；开启才检索并按模板结构抽取数据填充。 -

+ {/* 参考选了内置模板(tpl:)时：数据抽取开关(仅可转换模板;只展示模板给提示)。 */} + {form.html_reference_favorite_id.startsWith("tpl:") && + (selectedTplRef?.convertible === false ? ( +

+ 该模板暂为「直接展示已有数据」，执行时将原样输出其现有页面，不做数据抽取。 +

+ ) : ( +
+
+

执行数据抽取

+

+ 开启后执行时由所选智能体检索生成内容，并按模板结构抽取数据填充；抽取不到的子页面自动回退模板原有数据。关闭则直接用所选模板的默认数据生成（含全部子页面）。 +

+
+ + setForm((value) => ({ ...value, html_template_extract: v })) + } + />
- - setForm((value) => ({ ...value, html_template_extract: v })) - } - /> -
- )} + ))}

生成模型