iAOP/core/llm-gateway/evaluate_hallucination.py

# -*- coding: utf-8 -*-
"""iAOP-Core · LLM 网关 —— 幻觉/事实性校验评测报告脚本（Issue #47）。

对应 PRD 5.4「定期用评测集检验事实一致性」：把评测集 JSON + Prompt 版本库
配置跑一遍 `HallucinationGuard.evaluate()`，产出 Markdown 评测报告
（按 `name@version` 分解，Prompt 变更后可用于对比各版本事实一致性）。

用法示例：
    # 使用内置演示评测集，输出报告到文件（utf-8）
    python evaluate_hallucination.py --demo --output reports/hallucination.md

    # 使用自定义评测集 JSON（见下方 SAMPLE 字段说明）
    python evaluate_hallucination.py --samples eval_set.json --output report.md

    # 不指定 --output：报告打印到 stdout
    python evaluate_hallucination.py --demo

评测集 JSON 格式（顶层为数组）：
    [
      {
        "answer": "建议降温。[来源: 沸腾氯化炉异常处置SOP]",
        "sources": ["沸腾氯化炉异常处置SOP"],
        "confidence": 0.9,
        "high_stakes": true,
        "prompt_name": "alarm_explain",     // 可选，与版本库联动
        "prompt_version": "1.0.0"           // 可选，缺省取当前默认版本
      }
    ]

测试：`python -m unittest discover -s tests -v`（在 core/llm-gateway 目录下执行）。
"""
from __future__ import annotations

import argparse
import json
import os
import sys
import types
from typing import List

# ---------------------------------------------------------------------------
# 包挂载（目录名含连字符，无法直接以包名 import；与 tests/_bootstrap.py 同款）
# ---------------------------------------------------------------------------
_LLM_GW_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _LLM_GW_DIR)
if "llm_gateway" not in sys.modules:
    _pkg = types.ModuleType("llm_gateway")
    _pkg.__path__ = [_LLM_GW_DIR]
    sys.modules["llm_gateway"] = _pkg

from llm_gateway.hallucination import HallucinationGuard  # noqa: E402
from llm_gateway.prompts import PromptRegistry  # noqa: E402

# 内置演示评测集（无 --samples 时使用，覆盖 pass / unsupported / human_review）
DEMO_SAMPLES: List[dict] = [
    {
        "answer": "建议降低氯化炉温度并观察。[来源: 沸腾氯化炉异常处置SOP]",
        "sources": ["沸腾氯化炉异常处置SOP"],
        "confidence": 0.95,
        "high_stakes": True,
        "prompt_name": "alarm_explain",
    },
    {
        "answer": "应立即停机检修。[来源: 不存在的工艺文档]",
        "sources": ["沸腾氯化炉异常处置SOP"],
        "confidence": 0.9,
        "high_stakes": True,
        "prompt_name": "alarm_explain",
    },
    {
        "answer": "班次生产概况：炉温稳定。[来源: 交接班规范]",
        "sources": ["交接班规范"],
        "confidence": 0.6,
        "high_stakes": False,
        "prompt_name": "shift_handover",
        "prompt_version": "1.0.1",
    },
]


def load_samples(path: str) -> List[dict]:
    """从 JSON 文件加载评测集（顶层为样本数组）。"""
    with open(path, "r", encoding="utf-8") as fh:
        data = json.load(fh)
    if not isinstance(data, list):
        raise ValueError(f"评测集文件 {path} 顶层必须是样本数组")
    return data


def main(argv: List[str]) -> int:
    parser = argparse.ArgumentParser(
        prog="evaluate_hallucination",
        description="幻觉/事实性校验评测报告（PRD 5.4 定期评测）",
    )
    parser.add_argument(
        "--prompts-config", default="config/prompts.template.yaml",
        help="Prompt 版本库模板资产路径（默认 config/prompts.template.yaml）",
    )
    parser.add_argument(
        "--samples", default=None,
        help="评测集 JSON 文件路径（与 --demo 二选一）",
    )
    parser.add_argument(
        "--demo", action="store_true",
        help="使用内置演示评测集（未指定 --samples 时默认开启）",
    )
    parser.add_argument(
        "--threshold", type=float, default=0.8,
        help="高利害信度阈值（默认 0.8）",
    )
    parser.add_argument(
        "--output", default=None,
        help="报告输出文件路径（utf-8）；缺省打印到 stdout",
    )
    args = parser.parse_args(argv)

    # 评测集：--samples 优先，否则内置演示集
    if args.samples:
        samples = load_samples(args.samples)
    else:
        samples = DEMO_SAMPLES
        print("[evaluate_hallucination] 未指定 --samples，使用内置演示评测集",
              file=sys.stderr)

    # Prompt 版本库（与评测联动：按 name@version 分解）
    registry = PromptRegistry.from_template_config(args.prompts_config)

    guard = HallucinationGuard(default_threshold=args.threshold)
    report = guard.evaluate(samples, registry=registry)
    text = guard.render_evaluation_report(report)

    if args.output:
        os.makedirs(os.path.dirname(os.path.abspath(args.output)) or ".", exist_ok=True)
        with open(args.output, "w", encoding="utf-8") as fh:
            fh.write(text)
        print(f"[evaluate_hallucination] 评测报告已写入：{args.output}", file=sys.stderr)
    else:
        # Windows 控制台可能为 GBK：显式用 utf-8 输出避免编码错误
        try:
            sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        except (AttributeError, ValueError):  # pragma: no cover - 旧版解释器
            pass
        print(text)

    # 简报（stderr，便于定时任务抓取结论）
    print(
        "[evaluate_hallucination] 样本 {total}｜支持率 {sr:.1%}｜人工复核率 {hr:.1%}"
        .format(total=report["total"], sr=float(report["supported_rate"]),
                hr=float(report["human_review_rate"])),
        file=sys.stderr,
    )
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))