feat(bioflow): 样本配置 LLM 兜底——规则无法归纳时 DeepSeek 生成描述

- inspect_unknown_files:文件特征清单(路径/大小/magic/file 类型/压缩包
  内嵌名),无扩展名随机名也能拿到结构化特征
- llm_sample_fallback:描述文件+实况分析均无结果时,把特征清单交给
  complete_json(复用 DOCGEN_LLM_* 键)生成样本描述;r1/r2 回验存在性
  (清单路径或解压后),幻觉路径丢弃;LLM 失败/未配置静默跳过
- prompt 独立文件(bioflow_service/app/prompts/llm_sample_analysis.txt)
- 5 个新单测(特征×2+回验/幻觉拒绝/不可用×3)+ 340 用例全绿;已部署 176

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

bioflow_service/app/prompts/llm_sample_analysis.txt

@@ -0,0 +1,11 @@
+你是生信数据文件分析助手。用户提供一份数据目录的文件特征清单(含路径、大小、扩展名、magic 类型、压缩包内嵌文件名),请你推断哪些文件是 RNA-seq 测序数据(FASTQ),并按配对样本输出。
+
+只输出 JSON,格式:
+{"samples":[{"name":"样本名","r1":"R1文件路径或内嵌名","r2":"R2文件路径或内嵌名","run":"测序run标识","group_role":"control或compare或留空","condition":"实验条件或留空","genotype":"基因型或留空","replicate":"重复编号或留空"}]}
+
+规则:
+1. r1/r2 必须是清单中真实出现的路径或内嵌文件名,禁止臆造不存在的路径。
+2. 成对判断依据:文件名(或内嵌名)中含 R1/f1/1/forward 与 R2/f2/2/reverse 的配对;若无明确配对线索,单端文件只填 r1。
+3. 无法确定任何样本时,输出 {"samples":[]}。
+4. 不确定的字段留空字符串,不要编造。
+5. 只输出 JSON,不要任何解释或 markdown 围栏。

bioflow_service/app/routers/biojobs.py

@@ -289,6 +289,14 @@ async def _register_path(record: Any, server_path: str, settings: Any) -> dict[s
         else:
             source_roots.append(path.parent)
         extra_mounts = ensure_samples_yaml(input_dir, source_roots=source_roots)
+        if not extra_mounts:
+            # 规则无法归纳(无描述文件、实况配对 0 样本) → LLM 兜底
+            try:
+                from ..services.bioflow.runner import llm_sample_fallback
+                if await llm_sample_fallback(input_dir, source_roots):
+                    logger.info("samples.yaml LLM 兜底生成: %s", input_dir)
+            except Exception as exc:  # noqa: BLE001
+                logger.warning("samples.yaml LLM 兜底失败（不阻断登记）: %s", exc)
         if extra_mounts:
             logger.info("samples.yaml 分析完成(挂载父目录 %d 个): %s", len(extra_mounts), input_dir)
             record.registered_server_paths = list(

bioflow_service/app/services/bioflow/runner.py

@@ -352,6 +352,172 @@ def ensure_samples_yaml(job_input_dir: Path, source_roots: list[Path] | None = N
     return sorted(mounts)
 
 
+def inspect_unknown_files(paths: list[Path], max_files: int = 200) -> list[dict]:
+    """可观测特征函数:对无法按规则归类的文件,收集结构化特征供 LLM 兜底。
+
+    每项:{path, size, ext, magic(首 64 字节 hex + file 类型), parent,
+          archive_kind(tar|zip|gz|None), archive_members(≤20 内嵌名)}。
+    纯 stdlib + 子进程 file/tar/zcat/unzip,失败静默跳过该文件。
+    """
+    import subprocess
+    features: list[dict] = []
+    for p in sorted(paths, key=lambda q: (q.stat().st_size if q.is_file() else 0), reverse=True)[:max_files]:
+        if not p.is_file():
+            continue
+        try:
+            size = p.stat().st_size
+            head = p.read_bytes()[:64]
+            magic = head.hex()
+            ext = p.suffix.lower() or ""
+            parent = str(p.parent)
+            archive_kind = None
+            members: list[str] = []
+            file_type = ""
+            try:
+                file_type = subprocess.run(
+                    ["file", "-b", str(p)], capture_output=True, text=True, timeout=10
+                ).stdout.strip()
+            except Exception:  # noqa: BLE001
+                pass
+            if file_type.startswith("gzip") or magic.startswith("1f8b"):
+                archive_kind = "gz"
+                try:
+                    out = subprocess.run(
+                        ["gzip", "-l", str(p)], capture_output=True, text=True, timeout=10
+                    ).stdout
+                    members = out.strip().splitlines()[1:]
+                except Exception:  # noqa: BLE001
+                    pass
+            elif file_type.startswith("POSIX tar") or magic.startswith(("7573746172",)):
+                archive_kind = "tar"
+                try:
+                    out = subprocess.run(
+                        ["tar", "-tzf", str(p)], capture_output=True, text=True, timeout=15
+                    ).stdout
+                    members = [m for m in out.splitlines() if m.strip()][:20]
+                except Exception:  # noqa: BLE001
+                    pass
+            elif file_type.startswith("Zip archive") or magic.startswith("504b"):
+                archive_kind = "zip"
+                try:
+                    out = subprocess.run(
+                        ["unzip", "-l", str(p)], capture_output=True, text=True, timeout=15
+                    ).stdout
+                    members = [m.strip().split()[-1] for m in out.splitlines()[3:-2] if m.strip()][:20]
+                except Exception:  # noqa: BLE001
+                    pass
+            features.append({
+                "path": str(p), "size": size, "ext": ext, "magic": magic[:64],
+                "file_type": file_type, "parent": parent,
+                "archive_kind": archive_kind, "archive_members": members,
+            })
+        except Exception:  # noqa: BLE001
+            continue
+    return features
+
+
+async def llm_sample_fallback(job_input_dir: Path, source_roots: list[Path]) -> bool:
+    """LLM 兜底:规则无法归纳时,把文件特征清单交给 DeepSeek 生成样本描述。
+
+    触发条件:描述文件解析 0 条且实况配对 0 样本(调用方判断)。
+    返回是否成功生成 samples.yaml。LLM 失败/未配置/输出无法回验 → False,
+    不影响登记主链。r1/r2 必须回验存在(清单路径或解压后可解析),否则丢弃。
+    """
+    import yaml as _yaml
+    from .llm import LLMError, complete_json
+
+    # 收集未知文件(排除已识别的序列/描述文件)
+    known_exts = (".tsv", ".csv", ".xlsx", ".xls", ".yaml", ".yml", ".md",
+                  ".fq.gz", ".fastq.gz", ".fq", ".fastq")
+    unknown: list[Path] = []
+    for root in [job_input_dir, *[Path(r) for r in source_roots if Path(r).exists()]]:
+        for p in root.rglob("*"):
+            if not p.is_file() or p.name == "samples.yaml":
+                continue
+            if p.name.lower().endswith(known_exts):
+                continue
+            unknown.append(p)
+    if not unknown:
+        return False
+    features = inspect_unknown_files(unknown)
+    if not features:
+        return False
+
+    # system prompt(独立文件,可迭代)
+    prompt_path = Path(__file__).resolve().parent.parent / "prompts" / "llm_sample_analysis.txt"
+    try:
+        system = prompt_path.read_text(encoding="utf-8")
+    except OSError:
+        system = "你是生信数据文件分析助手,只输出 JSON。"
+    import json as _json
+    user = (
+        "以下是一个数据目录的文件特征清单,请推断其中的 RNA-seq FASTQ 样本并输出 JSON。\n"
+        f"{_json.dumps(features, ensure_ascii=False, default=str)[:24000]}"
+    )
+    try:
+        result = await complete_json(system, user, temperature=0.2, max_tokens=4000, timeout=30.0)
+    except (LLMError, Exception):  # noqa: BLE001
+        return False
+    samples = result.get("samples") if isinstance(result, dict) else None
+    if not isinstance(samples, list) or not samples:
+        return False
+
+    # 回验 r1/r2:必须是清单路径(存在)或压缩包内嵌名(解压后可解析)
+    valid_paths = {f["path"] for f in features if Path(f["path"]).exists()}
+    archive_members: dict[str, str] = {}
+    for f in features:
+        for m in (f.get("archive_members") or []):
+            archive_members.setdefault(m, f["path"])
+    yaml_samples: dict[str, dict] = {}
+    mounts: set[str] = set()
+    for s in samples:
+        if not isinstance(s, dict):
+            continue
+        r1 = str(s.get("r1") or "").strip()
+        r2 = str(s.get("r2") or "").strip()
+        entry: dict = {}
+        for key, val in (("r1", r1), ("r2", r2)):
+            if not val:
+                continue
+            if val in valid_paths:
+                entry[key] = val
+                mounts.add(str(Path(val).parent))
+            elif val in archive_members:
+                # 解压该压缩包(幂等),用内嵌名构造解出路径
+                arch = Path(archive_members[val])
+                target_dir = job_input_dir / "archives" / (arch.name.rsplit(".", 1)[0])
+                try:
+                    import tarfile
+                    import zipfile
+                    if not target_dir.exists():
+                        target_dir.mkdir(parents=True, exist_ok=True)
+                        if arch.name.endswith((".tar.gz", ".tar", ".tgz")):
+                            with tarfile.open(arch, "r:*") as tf:
+                                tf.extractall(target_dir)
+                        else:
+                            with zipfile.ZipFile(arch) as zf:
+                                zf.extractall(target_dir)
+                    resolved = target_dir / val
+                    if resolved.exists():
+                        entry[key] = str(resolved)
+                        mounts.add(str(resolved.parent))
+                except Exception:  # noqa: BLE001
+                    pass
+        if not entry:
+            continue
+        for k in ("run", "group_role", "condition", "genotype", "replicate"):
+            v = str(s.get(k) or "").strip()
+            if v:
+                entry[k] = v
+        yaml_samples[str(s.get("name") or Path(entry.get("r1") or "sample").stem)] = entry
+    if not yaml_samples:
+        return False
+    (job_input_dir / "samples.yaml").write_text(
+        _yaml.safe_dump({"samples": yaml_samples}, allow_unicode=True), encoding="utf-8"
+    )
+    return True
+
+
 # ---- runner ----
 
 ExecutorFn = Callable[..., Awaitable[dict[str, Any]]]