feat(bioflow): skill 编排续作 WIP checkpoint（多轮链续接/plan 回填/job_id 提取/历史 blocks 文本提取）

2026-10-14 09:41:44 +08:00
parent 8ab4279b04
commit 3c00e1675e
7 changed files with 160 additions and 26 deletions
@@ -46,4 +46,4 @@ paper_writer_service/paper.log
 # 部署方案文档（含主机凭据，不入库）
 docs/DEPLOY_PAPER_176.md
 # Python 覆盖率运行时产物(coverage.py --cov 生成)
-.coverage
+.coveragevstore/vstore.log
@@ -48,6 +48,9 @@ _CHAIN_TO_CAPABILITY = {
    "rnaseq": "bioflow-rnaseq",
 }

+# 能力名 → chain 名（反向映射，供规划阶段从命中能力回填 chain，续接多轮对话）
+_CAPABILITY_TO_CHAIN = {v: k for k, v in _CHAIN_TO_CAPABILITY.items()}
+

 def _chain_to_capability(chain: str) -> Optional[str]:
    """把 chain 名（gene_family/rnaseq）映射为能力名；未知返回 None。"""
@@ -323,6 +326,7 @@ def _make_payload(
    chain: Optional[str] = None,
    readiness: Optional[Dict[str, Any]] = None,
    versioning: Optional[Dict[str, Any]] = None,
+    plan: Optional[Dict[str, Any]] = None,
 ) -> Dict[str, Any]:
    payload: Dict[str, Any] = {
        "success": True,
@@ -346,6 +350,8 @@ def _make_payload(
        payload["bioflow"]["readiness"] = readiness
    if versioning:
        payload["bioflow"]["versioning"] = versioning
+    if plan:
+        payload["bioflow"]["plan"] = plan
    return payload


@@ -637,16 +643,24 @@ def _try_bioflow_agent_impl(
    if isinstance(card_result, dict) and str(card_result.get("card") or "").strip() in CARD_TYPES:
        return _route_card_result(card_result, llm_client, capabilities, bioflow, progress_callback)

-    # NL 路径：活跃任务续接优先（有活跃任务时跳过意图预筛，直接按任务 chain 续接）
+    # NL 路径：活跃任务/生信链续接优先，其次意图预筛
    active_job_id = (bioflow or {}).get("active_job_id")
+    chain = str((bioflow or {}).get("chain") or "").strip()
    if active_job_id:
+        # 有活跃任务：跳过意图预筛，按任务 chain 续接（任务链不可变）
        manifests = _registered_capabilities(capabilities)
-        cap_name = _chain_to_capability((bioflow or {}).get("chain") or "")
+        cap_name = _chain_to_capability(chain)
        if cap_name:
            manifests = [m for m in manifests if m.name == cap_name]
    else:
-        # 无活跃任务：门 2 + 门 3 意图预筛（新任务入口）
+        # 无活跃任务：先门 2 + 门 3 意图预筛（新任务入口）
        manifests = _candidate_capabilities(question, capabilities)
+        # 意图未命中但 session 留有生信链（上轮规划）→ 续接该链（如路径选择「1」）
+        if not manifests and chain:
+            cap_name = _chain_to_capability(chain)
+            if cap_name:
+                registered = _registered_capabilities(capabilities)
+                manifests = [m for m in registered if m.name == cap_name]
    if not manifests:
        return None

@@ -678,24 +692,40 @@ def _try_bioflow_agent_impl(
    tool_results = (audit or {}).get("tool_results") or {}
    readiness = None
    versioning = None
+    new_job_id = None
+    plan = None
    for name, result_text in tool_results.items():
+        if name in ("create_gene_family_job", "create_rnaseq_job"):
+            jid = _extract_job_id(result_text)
+            if jid:
+                new_job_id = jid
        data = _try_parse_json(result_text)
        if not isinstance(data, dict):
            continue
+        if name.startswith("plan_") and data.get("ok"):
+            plan = {
+                "candidates": data.get("candidates") or [],
+                "auto_filled": data.get("auto_filled") or [],
+                "reuse_archive": data.get("reuse_archive"),
+                "reuse_name": data.get("reuse_name"),
+                "fallback": data.get("fallback"),
+            }
        if name == "check_readiness":
            readiness = _map_readiness(data)
        elif name == "propose_change":
            versioning = _map_versioning(data)

+    effective_chain = chain or (_CAPABILITY_TO_CHAIN.get(manifests[0].name) if manifests else "")
    return _make_payload(
        question,
        answer,
        audit=audit,
        capabilities=[m.name for m in manifests],
-        active_job_id=(bioflow or {}).get("active_job_id"),
-        chain=(bioflow or {}).get("chain"),
+        active_job_id=(bioflow or {}).get("active_job_id") or new_job_id,
+        chain=effective_chain or None,
        readiness=readiness,
        versioning=versioning,
+        plan=plan,
    )


@@ -96,6 +96,7 @@ class ToolRunner:
                break
            assistant_message: Dict[str, Any] = {"role": "assistant", "content": content or ""}
            tool_call_blocks: List[Dict[str, Any]] = []
+            messages.append(assistant_message)
            for call in tool_calls:
                name = str(call.get("name") or call.get("function", {}).get("name") or "")
                raw_args = call.get("arguments") or call.get("function", {}).get("arguments") or {}
@@ -129,7 +130,6 @@ class ToolRunner:
                audit["tool_results"][name] = result_text
                messages.append({"role": "tool", "tool_call_id": call_id, "content": result_text})
            assistant_message["tool_calls"] = tool_call_blocks
-            messages.append(assistant_message)
            continue
        # 最后一轮：拿最终文本（无工具或轮次用尽后由调用方用 messages 再 complete 一次）
        return self._final_text(system_text, messages, tools), audit
@@ -7,9 +7,26 @@ assemble_history(db, session_id, rounds=3, budget):
 """
 from __future__ import annotations

+import html as html_mod
+import re
 from typing import Any, List

 from ..config import get_settings
+from ..db import loads_json
+
+
+def _blocks_text(blocks_json) -> str:
+    """从落库 blocks 提取 assistant 答案纯文本（text 块 html 去标签）。"""
+    blocks = loads_json(blocks_json, default=[])
+    parts = []
+    for b in blocks:
+        if not isinstance(b, dict) or b.get("type") != "text":
+            continue
+        h = b.get("html") or b.get("text") or ""
+        t = html_mod.unescape(re.sub(r"<[^>]+>", "", str(h))).strip()
+        if t:
+            parts.append(t)
+    return "\n".join(parts)


 async def assemble_history(
@@ -40,7 +57,7 @@ async def assemble_history(
        return []
    try:
        rows = await db.fetchall(
-            "SELECT role, status, text FROM messages "
+            "SELECT role, status, text, blocks FROM messages "
            "WHERE session_id = ? AND role IN ('user','assistant') "
            "AND status != 'streaming' "
            "ORDER BY created_at ASC, rowid ASC",
@@ -55,6 +72,8 @@ async def assemble_history(
    for row in reversed(rows):
        role = row["role"]
        text = (row["text"] or "").strip()
+        if not text and role == "assistant":
+            text = _blocks_text(row["blocks"])
        if not text:
            continue
        if role == "assistant":
@@ -66,6 +66,7 @@ function handleFiles(fileList) {
    chat.uploadBioflowFileToJob(jobId.value, file, {
      onProgress: (pct) => {
        const rec = props.block.files.find((x) => x.name === file.name && x.status === 'uploading');
+        console.log('[bioflow-upload] onProgress pct=', pct, 'recFound=', !!rec, 'files=', (props.block.files || []).length);
        if (rec) rec.progress = pct;
      },
    })
@@ -61,6 +61,7 @@ function backToSelect() {
 /** 复用名 → 表单默认值来源标注（与后端 params 预填配合；来源标注在表单卡展示） */
 async function confirmCreate() {
  const cand = selectedCandidate.value;
+  console.log('[bioflow-card] confirmCreate selected=', selected.value, 'cand=', !!cand, 'jobId=', props.block.jobId);
  if (!cand || confirming.value) return;
  if (props.block.jobId) return; // 已创建过（防重复创建；FR-BIO 任务幂等语义）
  confirming.value = true;
@@ -19,6 +19,7 @@ import { useSessionsStore } from './sessions.js';
 import { PHASE_LABEL, PHASE_DESC, T } from '../i18n/labels.js';
 import { MODES } from '../config/modes.js';
 import { renderMarkdown } from '../components/chat/markdown.js';
+console.log('[bioflow-version] chat.js raw.bioflow-fix v2 loaded');

 // ---- P3 文档生成：独立分支（send/_runStream 零改动） ----
 import { DOCGEN_STEPS, CAP_BADGES, DG as DG_LABELS } from '../i18n/docgen.js';
@@ -2316,18 +2317,41 @@ export const useChatStore = defineStore('chat', {
      try {
        const data = await uploadBioflowFile(jobId, form, {
          onUploadProgress: (e) => {
-            item.progress = e.pct;
+            console.log('[bioflow-upload] onUploadProgress pct=', e.pct, 'loaded=', e.loaded, 'total=', e.total);
+            // item 是原始对象，直接改会绕过 Vue 响应式——经响应式数组找代理更新
+            const rec = found && Array.isArray(found.blk.files)
+              ? found.blk.files.find((x) => x.name === item.name && x.status === 'uploading')
+              : null;
+            if (rec) rec.progress = e.pct;
+            else item.progress = e.pct;
            if (typeof onProgress === 'function') onProgress(e.pct, item);
          },
        });
-        item.status = 'done';
-        item.progress = 1;
-        item.fileId = (data && data.data && data.data.file_id) || null;
+        const doneRec = found && Array.isArray(found.blk.files)
+          ? found.blk.files.find((x) => x.name === item.name)
+          : null;
+        if (doneRec) {
+          doneRec.status = 'done';
+          doneRec.progress = 1;
+          doneRec.fileId = (data && data.data && data.data.file_id) || null;
+        } else {
+          item.status = 'done';
+          item.progress = 1;
+          item.fileId = (data && data.data && data.data.file_id) || null;
+        }
        if (found) this.scrollSeq++;
        return (data && data.data) || data || {};
      } catch (err) {
-        item.status = 'error';
-        item.error = err?.message || BF.fileUploadError;
+        const errRec = found && Array.isArray(found.blk.files)
+          ? found.blk.files.find((x) => x.name === item.name)
+          : null;
+        if (errRec) {
+          errRec.status = 'error';
+          errRec.error = err?.message || BF.fileUploadError;
+        } else {
+          item.status = 'error';
+          item.error = err?.message || BF.fileUploadError;
+        }
        if (found) this.scrollSeq++;
        throw err;
      }
@@ -2458,6 +2482,7 @@ export const useChatStore = defineStore('chat', {
     *  经 BioflowAgent 工具执行 + 自然语言复述；卡片与自然语言两条通道语义一致。
     *  card ∈ path_selection|form_confirm|file_upload|archive_choice|extension_action。 */
    async sendCardResult(card, payload, { text = '', userMsg = null } = {}) {
+      console.log('[bioflow-card] sendCardResult', card, 'busy=', this.busy);
      if (this.busy) return;
      const sessions = useSessionsStore();
      if (!this.sessionId) {
@@ -2550,10 +2575,12 @@ export const useChatStore = defineStore('chat', {
              const p = ev.payload || {};
              const answer = p.answer || '';
              updateLocal(() => {
-                const blocks = [
-                  { type: 'bioflow_path', status: 'done', candidates: [] },
-                  { type: 'text', html: answer },
-                ];
+                const blocks = [];
+                // 仅路径选择卡回填 path 块（候选由上一轮路径卡承载）；其余卡操作只出自然语言复述
+                if (card === 'path_selection') {
+                  blocks.push({ type: 'bioflow_path', status: 'done', candidates: [] });
+                }
+                blocks.push({ type: 'text', html: answer });
                const bfPayload = p.bioflow || {};
                if (bfPayload.readiness) {
                  blocks.push({ type: 'bioflow_readiness', ...bfPayload.readiness, error: null });
@@ -2689,7 +2716,12 @@ export const useChatStore = defineStore('chat', {
                // S2：bioflow 帧（生信对话编排）不推 pipeline 相位——生信消息走
                // bioflow 块，由 sendCardResult / bioflow 分支消费，此处仅记录。
                const isBioflow = typeof p.stage === 'string' && p.stage.startsWith('bioflow_');
-                if (isBioflow) return;
+                if (isBioflow) {
+                  // 生信消息：一旦收到 bioflow 帧，立即移除检索 pipeline 骨架（避免闪烁残留）
+                  const pi = (aMsg.blocks || []).findIndex((b) => b && b.type === 'pipeline');
+                  if (pi >= 0) aMsg.blocks.splice(pi, 1);
+                  return;
+                }
                if (!ctx.completedOnce && phase !== 'done') this._advancePipeline(aMsg, phase, label);
                if (phase === 'done') {
                  // 上游 completed 事件：当前相位定格 done（此时 complete 尚未到达）
@@ -2728,18 +2760,68 @@ export const useChatStore = defineStore('chat', {
              // complete 帧之后流即关闭：同步收口，避免「流自然结束」误判为连接中断
              endTask();
              const payload = ev.payload || {};
+              // S2 多轮续接：生信规划返回的 chain/active_job_id 写回 session，
+              // 下轮（如路径选择「1」）据此跳过意图预筛续接生信 agent。
+              const bf = payload.bioflow || (payload.raw && payload.raw.bioflow) || {};
+              if (bf.chain || bf.active_job_id) {
+                updateSessionQuiet(sessionId, {
+                  meta: { bioflow: { active_job_id: bf.active_job_id || "", chain: bf.chain || "" } },
+                }).catch(() => {});
+                // 任务事件订阅：自然语言路径建任务后补 connectJobEvents，
+                // 把 stage_await_files / stage_await_confirm / stage_failed / job_done
+                // 桥接进消息块（文件卡/表单卡/诊断卡/交付卡），与卡片路径一致。
+                if (bf.active_job_id) {
+                  try {
+                    const bioflow = useBioflowStore();
+                    const jobId = bf.active_job_id;
+                    bioflow.connectJobEvents(jobId, sessionId, {
+                      onEvent: (type, data, seq) => {
+                        const frame = patchBioflowEvent(aMsg, { type, data });
+                        if (frame) this.scrollSeq++;
+                        if (type === 'stage_await_confirm' && data) {
+                          patchBioflowGuidance(aMsg, data.guidance);
+                          patchBioflowForm(aMsg, data);
+                          this.scrollSeq++;
+                        } else if (type === 'stage_await_files' && data) {
+                          patchBioflowGuidance(aMsg, data.guidance || data.message);
+                          patchBioflowFile(aMsg, data);
+                          this.scrollSeq++;
+                        } else if (type === 'stage_failed' && data) {
+                          patchBioflowGuidance(aMsg, data.guidance);
+                          patchBioflowDiagnosis(aMsg, data);
+                          this.scrollSeq++;
+                        } else if (type === 'job_done' && data) {
+                          patchBioflowGuidance(aMsg, data.guidance);
+                          if (data.delivery) patchBioflowDelivery(aMsg, data.delivery);
+                          const blk = patchBioflowArchiveSuggest(aMsg, data.job_id || jobId);
+                          if (blk) blk.jobId = data.job_id || jobId;
+                          this.scrollSeq++;
+                        }
+                      },
+                    });
+                  } catch { /* EventSource 不可用：任务事件仍由快照兜底 */ }
+                }
+              }
              const blocks = qaToBlocks(payload);
+              const isBioflow = !!(bf && bf.handled);
              updateLocal(() => {
                // text 块先不放：由揭示动画以 stream 块渐进显示，终了替换为最终 text
-                aMsg.blocks = [
-                  aMsg.blocks[0],
-                  ...blocks.filter((b) => b.type !== 'text'),
-                ];
+                // 生信消息不显示检索 pipeline 骨架——直接挂生信卡片 + 自然语言。
+                aMsg.blocks = isBioflow
+                  ? [...blocks.filter((b) => b.type !== 'text')]
+                  : [aMsg.blocks[0], ...blocks.filter((b) => b.type !== 'text')];
+                // 生信卡片：路径卡/就绪卡/版本卡（与自然语言 answer 同时出现）
+                if (bf.plan) patchBioflowPlan(aMsg, bf.plan);
+                if (bf.readiness) patchBioflowReadiness(aMsg, bf.readiness);
+                if (bf.versioning) patchBioflowVersioning(aMsg, bf.versioning);
                aMsg.raw = payload.raw ?? payload;
-                const pipe = aMsg.blocks[0];
-                pipe.status = 'done';
-                pipe.steps.forEach((s) => { s.state = 'done'; });
+                if (!isBioflow) {
+                  const pipe = aMsg.blocks[0];
+                  pipe.status = 'done';
+                  pipe.steps.forEach((s) => { s.state = 'done'; });
+                }
              });
+              console.log('[bioflow-debug] onComplete blocks=', (aMsg.blocks || []).map((b) => b && b.type), 'bf.handled=', !!(bf && bf.handled), 'bf.plan=', !!(bf && bf.plan));
              ctx.completedOnce = true;
              // 揭示动画（FR-CHAT-010）：stream 块瞬态、不落库
              if (payload.answer) {
@@ -2747,6 +2829,7 @@ export const useChatStore = defineStore('chat', {
              } else {
                updateLocal(() => { aMsg.status = 'done'; });
              }
+              console.log('[bioflow-debug] after reveal blocks=', (aMsg.blocks || []).map((b) => b && b.type));
              const dur = ((Date.now() - startAt) / 1000).toFixed(1) + 's';
              // FR-PRIV-005：onComplete 整体替换 blocks 后 flush 私有 pending（防丢块）
              this._flushPrivate(aMsg);