代码提交

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}

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(/<think>([\s\S]*?)<\/think>/i);
+  if (closed?.[1] != null) return closed[1].trim();
+  const open = chatText.match(/<think>([\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("<think>")) {
-        const thinkContent = chatText.replace("<think>", "").replace("</think>", "").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<MultiRoundEvent> {
   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<string | null>(null);
   const controllerRef = useRef<AbortController | null>(null);
+  const inFlightThreadIdRef = useRef<string | null>(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);
       }

frontend-web/src/pages/ScheduledTasksPage.tsx

@@ -138,8 +138,9 @@ const emptyForm = {
   html_negative_prompt: "",
   html_reference_favorite_id: "",
   html_model_name: "",
-  // 参考选了内置模板(tpl:<id>)时的「数据抽取」开关：关闭=用模板默认数据，开启=检索抽取。
-  html_template_extract: false,
+  // 参考选了内置模板(tpl:<id>)时的「数据抽取」开关：默认开启(检索抽取+抽不到回退
+  // 模板原数据)，仅可转换模板生效；显式关闭=直接用模板默认数据。
+  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:<id>)。
+  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({
                     )}
                   </div>
                 </div>
-                {/* 参考选了内置模板(tpl:<id>)时：数据抽取开关。关闭=用模板默认数据。 */}
-                {form.html_reference_favorite_id.startsWith("tpl:") && (
-                  <div className="flex items-center justify-between rounded-md border p-3">
-                    <div className="min-w-0 pr-3">
-                      <p className="text-sm font-medium">执行数据抽取</p>
-                      <p className="text-muted-foreground text-xs">
-                        关闭则直接用所选模板的默认数据生成（含全部子页面）；开启才检索并按模板结构抽取数据填充。
-                      </p>
+                {/* 参考选了内置模板(tpl:<id>)时：数据抽取开关(仅可转换模板;只展示模板给提示)。 */}
+                {form.html_reference_favorite_id.startsWith("tpl:") &&
+                  (selectedTplRef?.convertible === false ? (
+                    <p className="text-muted-foreground rounded-md border p-3 text-xs">
+                      该模板暂为「直接展示已有数据」，执行时将原样输出其现有页面，不做数据抽取。
+                    </p>
+                  ) : (
+                    <div className="flex items-center justify-between rounded-md border p-3">
+                      <div className="min-w-0 pr-3">
+                        <p className="text-sm font-medium">执行数据抽取</p>
+                        <p className="text-muted-foreground text-xs">
+                          开启后执行时由所选智能体检索生成内容，并按模板结构抽取数据填充；抽取不到的子页面自动回退模板原有数据。关闭则直接用所选模板的默认数据生成（含全部子页面）。
+                        </p>
+                      </div>
+                      <Switch
+                        checked={form.html_template_extract}
+                        onCheckedChange={(v) =>
+                          setForm((value) => ({ ...value, html_template_extract: v }))
+                        }
+                      />
                     </div>
-                    <Switch
-                      checked={form.html_template_extract}
-                      onCheckedChange={(v) =>
-                        setForm((value) => ({ ...value, html_template_extract: v }))
-                      }
-                    />
-                  </div>
-                )}
+                  ))}
                 <div className="flex flex-col gap-1">
                   <p className="text-muted-foreground text-xs">生成模型</p>
                   <Select

offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py

@@ -1445,6 +1445,27 @@ class ScheduledTaskService:
             return "\n".join(p for p in parts if p)
         return str(content)
 
+    async def _agent_soul_and_model(
+        self, task: dict[str, Any], model_name: str | None
+    ) -> tuple[str | None, str | None]:
+        """Resolve the task agent's SOUL; fall back to its model when none chosen.
+
+        Used by both template pipelines so the **task's selected agent** shapes
+        content conversion: the SOUL carries the agent's domain instructions,
+        and when the user didn't pick an explicit model the agent's configured
+        model is preferred over the system default. Returns ``(soul, model)``.
+        """
+        agent_ref = task.get("execution_agent_name")
+        agent_id = await self._resolve_agent_id(agent_ref, task["user_id"]) if agent_ref else None
+        soul = load_agent_soul(agent_id) if agent_id else None
+        if not model_name and agent_id:
+            try:
+                agent_cfg = load_agent_config(agent_id)
+                model_name = (agent_cfg.model if agent_cfg else None) or None
+            except Exception:
+                model_name = None
+        return soul, model_name
+
     async def _convert_to_template_data(
         self,
         template_id: str,
@@ -1574,8 +1595,9 @@ class ScheduledTaskService:
         cfg = self._html_config(task)
         model_name = str(cfg.get("model_name") or "").strip() or None
         # 「执行数据抽取」开关**只**控制 HTML 用「抽取数据」还是「模板默认数据」,
-        # 不影响 Markdown 生成等其它正常操作。
-        extract = bool(cfg.get("template_extract"))
+        # 不影响 Markdown 生成等其它正常操作。缺省(未存该键的老任务)视为开启:
+        # 模板有真实数据来源后,默认走「智能体检索 → 转换填充」,显式 False 才关闭。
+        extract = cfg.get("template_extract") is not False
         also_markdown = _task_requires_markdown(task)
         prompt = str(task.get("prompt") or "").strip()
         description = str(task.get("description") or "").strip()
@@ -1626,9 +1648,12 @@ class ScheduledTaskService:
         source = (data_text or "").strip()
         used_default = True
         if do_extract and source:
+            # 转换阶段同样带上任务所选智能体:SOUL 提供领域口径,未显式选模型时
+            # 兜底用该智能体配置的模型(与模板填充模式一致)。
+            soul, conv_model = await self._agent_soul_and_model(task, model_name)
             try:
                 converted = await self._convert_to_template_data(
-                    tpl_id, source, model_name=model_name, soul=None, run_id=run_id
+                    tpl_id, source, model_name=conv_model, soul=soul, run_id=run_id
                 )
                 # 子页面没抽到有效内容的,回退模板原有数据,避免子页面空白。
                 data_obj = template_fill.merge_with_reference(tpl_id, converted)
@@ -1712,16 +1737,8 @@ class ScheduledTaskService:
             tcfg = tcfg if isinstance(tcfg, dict) else {}
 
             # 选中的内置 agent 的 SOUL + 模型(template_config.model_name 优先,其次 agent.model)。
-            agent_ref = task.get("execution_agent_name")
-            agent_id = await self._resolve_agent_id(agent_ref, task["user_id"]) if agent_ref else None
-            soul = load_agent_soul(agent_id) if agent_id else None
             model_name = str(tcfg.get("model_name") or "").strip() or None
-            if not model_name and agent_id:
-                try:
-                    agent_cfg = load_agent_config(agent_id)
-                    model_name = (agent_cfg.model if agent_cfg else None) or None
-                except Exception:
-                    model_name = None
+            soul, model_name = await self._agent_soul_and_model(task, model_name)
 
             converted = await self._convert_to_template_data(
                 template_id, source, model_name=model_name, soul=soul, run_id=run_id