"""test_flow.py —— flow_docs 集合单元测试（tmpdir + 假 embed，不依赖生产 vstore/ollama）。 覆盖： - FlowStore 增删查（幂等 upsert、delete_source、stats、search 过滤） - ingest_flow_docs 对三种语料的切块（fixture 语料 + monkeypatch 目录常量） - 评估集自测（run_flow_eval + fake embed：检索命中/引用断言通过率记录） 运行（仓库根，venv python，临时目录自动清理）： pytest vstore/tests/test_flow.py -v """ from __future__ import annotations import json from pathlib import Path from unittest import mock import numpy as np import pytest import vstore.flow as flow import vstore.vault as vault from vstore.flow import ( DOC_KIND_CONTRACT, DOC_KIND_CORPUS, FlowStore, SOURCE_CORPUS, SOURCE_SKILL, SOURCE_STAGE_CONTRACT, ) # ---- 假嵌入：文本含中文关键词时打同一方向向量（同主题可召回） ---- _TOPIC_GROUPS = { "identify": ["identify", "初筛", "HMMER"], "motif": ["motif", "MEME"], "phylogeny": ["phylogeny", "系统发育", "MAFFT"], "rnaseq": ["rnaseq", "RNA-seq", "DESeq2"], "wgs": ["wgs", "WGS", "BWA", "gvcf"], "gwas": ["gwas", "GWAS", "EMMAX", "kinship"], "pop": ["pop", "群体", "PCA"], } def _topic_of(text: str) -> str: for group, keys in _TOPIC_GROUPS.items(): if any(k in text for k in keys): return group return "general" def fake_embed_topic(texts: list[str]) -> list[list[float]]: """假嵌入：按主题组给方向向量，同主题文本余弦高（检索可召回同类块）。""" dim = 1024 base = np.zeros(dim, dtype=np.float32) out: list[list[float]] = [] slot = 0 for text in texts: if not text or not text.strip(): out.append([0.0] * dim) continue topic = _topic_of(text) vec = base.copy() vec[hash(topic) % dim] = 1.0 # 无主题的查询给一个能命中所有主题的温和向量 if topic == "general": vec[:10] = 0.1 slot += 1 out.append((vec / float(np.linalg.norm(vec))).tolist()) return out @pytest.fixture() def fake_embed(monkeypatch): """注入假 embed：flow（检索/入库脚本）与 vault（脚本内 get_embed_fn 来源）同源。""" monkeypatch.setattr(flow, "_EMBED_FN", fake_embed_topic) monkeypatch.setattr(vault, "_EMBED_FN", fake_embed_topic) return fake_embed_topic # ---- fixture 语料 ---- CORPUS_MD = """# 测试全流程文档 > 引言段。 ## 1. 文档阅读说明 六个问题：目的/前置条件/输入/主要处理/输出/质控。 # 第一部分：基因家族分析 ## 2. 基因家族完整流程总览 基因家族从蛋白组出发。 ## 4. `identify`：家族成员初筛与结构域确认 | 项目 | 说明 | |---|---| | 目的 | 从目标物种全蛋白组中找出可能属于指定基因家族的成员。 | | 输入 | 目标物种全蛋白 FASTA；Pfam HMM。 | | 输出 | `final/{prefix}.final_members.id`。 | # 第二部分：RNA-seq 分析 ## 25. `rnaseq_qc`：原始数据质控与清洗 | 项目 | 说明 | |---|---| | 主要处理 | 默认使用 fastp。 | | 输出 | clean R1/R2。 | """ CONTRACT_MD = """# Ortholog Stage Contract v1.0 ## 阶段信息 - **阶段名称**: ortholog - **阶段描述**: 使用 BLASTP 在目标物种家族蛋白与参考物种家族蛋白之间建立同源/映射关系。 ## 输入契约 | 字段 | 类型 | 描述 | |---|---|---| | scope_file | json | target/reference/comparative scope JSON | """ SKILL_MD = """# Gene Identify Skill 基因家族鉴定技能 - 基于HMMER和固定方向BLAST的自动化基因家族成员鉴定 ## 参数 | 参数 | 类型 | 默认值 | 说明 | |------|------|--------|------| | evalue | float | 1e-5 | HMMER e-value阈值 | | coverage | float | 0.3 | 最小覆盖度 | """ @pytest.fixture() def corpus_file(tmp_path): p = tmp_path / "corpus.md" p.write_text(CORPUS_MD, encoding="utf-8") return p @pytest.fixture() def stages_dir(tmp_path): d = tmp_path / "stages" d.mkdir() (d / "ortholog_contract_v1.md").write_text(CONTRACT_MD, encoding="utf-8") return d @pytest.fixture() def skills_dir(tmp_path): d = tmp_path / "skills" (d / "gene-identify").mkdir(parents=True) (d / "gene-identify" / "SKILL.md").write_text(SKILL_MD, encoding="utf-8") return d # ---- FlowStore 单元 ---- def test_add_chunk_idempotent(tmp_path, fake_embed): store = FlowStore(tmp_path / "flow") chunk = { "source": SOURCE_CORPUS, "source_id": SOURCE_CORPUS, "doc_kind": DOC_KIND_CORPUS, "doc_kind_label": "全流程文档", "title": "T", "title_path": "A | B", "heading": "B", "path": "docs/x.md", "cite": {"anchor": "B"}, "line_start": 1, "line_end": 5, "chunk_key": "corpus__abc", "created_at": "2026-09-05T00:00:00Z", } vec = fake_embed_topic(["identify"])[0] store.add_chunk(chunk, "正文 identify 内容", vec) store.add_chunk(chunk, "正文 identify 内容 v2", vec) # 重跑覆盖 stats = store.stats() assert stats["total_chunks"] == 1 # 幂等：不重复 doc = store.get_chunk("corpus__abc") assert "v2" in doc["text"] def test_search_filter_and_delete_source(tmp_path, fake_embed): store = FlowStore(tmp_path / "flow") def _add(key, source, kind, heading, text, topic): store.add_chunk({ "source": source, "source_id": source, "doc_kind": kind, "doc_kind_label": kind, "title": "T", "title_path": f"X | {heading}", "heading": heading, "path": "p.md", "cite": {"anchor": heading}, "line_start": 1, "line_end": 1, "chunk_key": key, "created_at": "2026-09-05T00:00:00Z", }, text, fake_embed_topic([topic])[0]) _add("c1", SOURCE_CORPUS, DOC_KIND_CORPUS, "identify 章", "Pfam HMM 输入", "identify") _add("c2", SOURCE_CORPUS, DOC_KIND_CORPUS, "rnaseq_qc 章", "fastp 清洗", "rnaseq") _add("s1", SOURCE_STAGE_CONTRACT, DOC_KIND_CONTRACT, "ortholog 契约", "BLASTP 映射", "general") res = store.search("identify 输入", top_k=5) assert res["items"], "identify 查询应有命中" res_corpus = store.search("identify 输入", top_k=5, source=SOURCE_CORPUS) assert all(r["source"] == SOURCE_CORPUS for r in res_corpus["items"]) removed = store.delete_source(SOURCE_CORPUS) assert removed == 2 stats = store.stats() assert stats["total_chunks"] == 1 assert stats["per_source"].get(SOURCE_STAGE_CONTRACT) == 1 # ---- ingest 切块 ---- def test_ingest_corpus_chunking(tmp_path, corpus_file, fake_embed): from vstore.scripts.ingest_flow_docs import ingest_corpus store = FlowStore(tmp_path / "flow") n = ingest_corpus(store, corpus_file, "docs/bioflow_corpus/corpus.md") stats = store.stats() assert n == stats["total_chunks"] assert stats["per_source"].get(SOURCE_CORPUS, 0) == n assert n >= 4 # 文档头 + 前导章 + 组块 + 至少一个二级章 rows = store.index.rows() headings = [str(r.get("heading") or "") for r in rows] assert any("identify" in h for h in headings) # 引用定位字段 ident = next(r for r in rows if "identify" in str(r.get("heading") or "")) assert ident["cite"]["anchor"] == "## 4. `identify`：家族成员初筛与结构域确认" assert ident["line_start"] > 0 def test_ingest_contracts_and_skills(tmp_path, stages_dir, skills_dir, fake_embed): from vstore.scripts.ingest_flow_docs import ingest_contracts, ingest_skills store = FlowStore(tmp_path / "flow") n1 = ingest_contracts(store, stages_dir, "gfa_core/schemas/stages") n2 = ingest_skills(store, skills_dir, "skills") stats = store.stats() assert n1 >= 2 # ortholog 契约按 ## 切成 2+ 节 assert n2 >= 1 assert stats["per_source"].get(SOURCE_STAGE_CONTRACT, 0) == n1 assert stats["per_source"].get(SOURCE_SKILL, 0) == n2 rows = store.index.rows() skill_rows = [r for r in rows if r["source"] == SOURCE_SKILL] assert skill_rows and skill_rows[0]["stage_name"] == "gene-identify" # ---- 评估集自测 ---- @pytest.fixture() def eval_file(tmp_path): items = [ {"id": 1, "category": "gf_stage_io", "type": "hit", "query": "identify 阶段需要什么输入？", "stage_keyword": "identify", "source_required": "corpus", "golds": ["Pfam HMM", "全蛋白"]}, {"id": 2, "category": "rnaseq_stage_io", "type": "hit", "query": "rnaseq_qc 用什么工具？", "stage_keyword": "rnaseq_qc", "source_required": "corpus", "golds": ["fastp", "clean R1/R2"]}, {"id": 3, "category": "software", "type": "hit", "query": "ortholog 用什么比对？", "stage_keyword": "ortholog", "source_required": "stage_contract", "golds": ["BLASTP", "scope_file"]}, ] p = tmp_path / "eval.json" p.write_text(json.dumps({"items": items}, ensure_ascii=False), encoding="utf-8") return p def test_run_eval_pass_rate(tmp_path, corpus_file, stages_dir, eval_file, fake_embed): from vstore.scripts.ingest_flow_docs import ingest_corpus, ingest_contracts from vstore.scripts.run_flow_eval import run_eval store = FlowStore(tmp_path / "flow") ingest_corpus(store, corpus_file, "docs/bioflow_corpus/corpus.md") ingest_contracts(store, stages_dir, "gfa_core/schemas/stages") out = run_eval(eval_file, tmp_path / "flow", top_k=5) assert out["summary"]["total"] == 3 assert out["summary"]["pass_rate"] >= 2 / 3, out["summary"] for r in out["results"]: assert r["quote"], f"item {r['id']} 引用断言应通过：{r}"