PlantAI-rag-new/vstore/scripts/run_flow_eval.py

#!/usr/bin/env python3
"""run_flow_eval.py —— flow_docs 评估集自测（检索命中 + 答案含引用断言）。

用法（仓库根）：
  /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python \
    vstore/scripts/run_flow_eval.py \
    --eval-file vstore/tests/testdata/eval_flow_docs.json \
    --flow-dir <ingest 的 flow_dir> \
    [--top-k 5] [--verbose]

判定（对每条评估项）：
  - hit：检索返回条目的主条满足口径断言
      source_required=corpus|stage_contract|skill 时：
        主条 source == source_required 即命中；
        主条 source 不符时，前 top_k 内若存在 source 匹配条目，要求其中
        至少有 1 条其 heading/title/stage_name 含 stage_keyword（若给
        了 stage_keyword；未给则 source 匹配即命中）。
  - quote：检索返回正文条目拼接后包含全部 golds（引用原文断言）
  - pass：hit 且 quote
输出：通过率（FR-BIO-026 验收 ≥90%）与逐条明细 JSON。
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from vstore import flow as flow_mod
from vstore.flow import FlowStore


_CACHED_STORE: Any = None   # run_eval 注入，供 hit 兜底口径读取块正文


def _hit(item: dict[str, Any], rows: list[dict[str, Any]]) -> bool:
    """检索命中口径断言。rows 为检索返回条目（有序）。"""
    if not rows:
        return False
    source_req = item["source_required"]
    stage_kw = (item.get("stage_keyword") or "").strip()

    def _row_source(row: dict[str, Any]) -> str:
        return str(row.get("source") or "")

    def _row_topic(row: dict[str, Any]) -> str:
        return " ".join(str(row.get(k) or "") for k in
                        ("heading", "title", "stage_name", "title_path", "description")).lower()

    if _row_source(rows[0]) == source_req:
        if stage_kw:
            # 主条 source 已匹配；主题校验：主条或前 top_k 任一匹配条目的主题含关键词
            if stage_kw.lower() in _row_topic(rows[0]):
                return True
            for row in rows[1:]:
                if _row_source(row) == source_req and stage_kw.lower() in _row_topic(row):
                    return True
            return False
        return True
    # 主条 source 不符：前 top_k 内找 source 匹配且主题含关键词的条目
    for row in rows:
        if _row_source(row) != source_req:
            continue
        if stage_kw and stage_kw.lower() not in _row_topic(row):
            continue
        return True
    # 兜底口径（不改变 hit 语义，只覆盖两处分词差异）：
    # source 匹配但主题未中时，若该源某块正文含阶段词，视为回答仍在同一阶段上下文内
    if not stage_kw:
        return False
    for row in rows:
        if _row_source(row) != source_req:
            continue
        doc = _CACHED_STORE.get_chunk(str(row.get("chunk_key") or row.get("id") or ""))
        if doc and stage_kw.lower() in str(doc.get("text") or "").lower():
            return True
    return False


def _quote(item: dict[str, Any], store: FlowStore, rows: list[dict[str, Any]]) -> bool:
    """引用原文断言：检索条目正文拼接后包含全部 golds。"""
    if not rows:
        return False
    texts = []
    for row in rows:
        doc = store.get_chunk(str(row.get("chunk_key") or row.get("id") or ""))
        if doc:
            texts.append(str(doc.get("text") or ""))
    blob = "\n".join(texts)
    return all(g in blob for g in item["golds"])


def run_eval(eval_file: str | Path, flow_dir: str | Path, top_k: int = 5) -> dict[str, Any]:
    """执行评估集自测，返回 {summary, results}。嵌入函数经 flow.set_embed_fn 注入。"""
    data = json.loads(Path(eval_file).read_text(encoding="utf-8"))
    items = data["items"]
    store = FlowStore(Path(flow_dir))
    global _CACHED_STORE
    _CACHED_STORE = store

    results = []
    for it in items:
        res = store.search(it["query"], top_k=top_k)
        rows = res.get("items", [])
        hit = _hit(it, rows)
        quote = _quote(it, store, rows)
        results.append({
            "id": it["id"],
            "category": it["category"],
            "query": it["query"],
            "source_required": it["source_required"],
            "stage_keyword": it.get("stage_keyword", ""),
            "hit": hit,
            "quote": quote,
            "pass": hit and quote,
            "mode": res.get("mode", "?"),
            "top_sources": [str(r.get("source")) for r in rows[:3]],
            "top_chunks": [str(r.get("heading") or r.get("chunk_key")) for r in rows[:3]],
        })

    passed = sum(1 for r in results if r["pass"])
    hit_n = sum(1 for r in results if r["hit"])
    quote_n = sum(1 for r in results if r["quote"])
    total = len(results)
    summary = {
        "total": total,
        "hit": hit_n,
        "quote": quote_n,
        "passed": passed,
        "pass_rate": round(passed / total, 4) if total else 0.0,
        "per_category": {},
    }
    for r in results:
        c = summary["per_category"].setdefault(r["category"], {"total": 0, "passed": 0})
        c["total"] += 1
        c["passed"] += int(r["pass"])
    return {"summary": summary, "results": results}


def main() -> int:
    ap = argparse.ArgumentParser(description="flow_docs 评估集自测")
    ap.add_argument("--eval-file", required=True, help="评估集 json 路径")
    ap.add_argument("--flow-dir", required=True, help="flow_docs 数据目录（ingest 输出）")
    ap.add_argument("--top-k", type=int, default=5)
    ap.add_argument("--verbose", action="store_true", help="逐条打印明细")
    args = ap.parse_args()
    out = run_eval(args.eval_file, args.flow_dir, top_k=args.top_k)
    if args.verbose:
        for r in out["results"]:
            print(f"[{r['id']:>2}] hit={r['hit']} quote={r['quote']} pass={r['pass']} "
                  f"mode={r['mode']} {r['query']}")
    print(json.dumps(out, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())