diff --git a/bioflow_service/app/prompts/llm_sample_analysis.txt b/bioflow_service/app/prompts/llm_sample_analysis.txt new file mode 100644 index 0000000..3b6799f --- /dev/null +++ b/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 围栏。 diff --git a/bioflow_service/app/routers/biojobs.py b/bioflow_service/app/routers/biojobs.py index 2cd8251..0bb8f6c 100644 --- a/bioflow_service/app/routers/biojobs.py +++ b/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( diff --git a/bioflow_service/app/services/bioflow/runner.py b/bioflow_service/app/services/bioflow/runner.py index eaddb4e..bcfdaf4 100644 --- a/bioflow_service/app/services/bioflow/runner.py +++ b/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]]]