adminandClaude Fable 5
adb97f49a8
语料 106 文件/724 块(全流程 79 章 118 + 契约 60 份 429 + 技能 45 份 177)；QA 树与 vstore 既有集合零改动 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
278 lines
9.5 KiB
Python
278 lines
9.5 KiB
Python
"""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}"
|