diff --git a/core/llm-gateway/README.md b/core/llm-gateway/README.md index a75e5ab..11b7d3e 100644 --- a/core/llm-gateway/README.md +++ b/core/llm-gateway/README.md @@ -11,8 +11,9 @@ core/llm-gateway/ ├── __init__.py 包入口(导出各模块 API) ├── dlp.py DLP 敏感数据拦截引擎(Issue #48) ├── router.py 敏感度路由规则引擎(Issue #43 雏形) -├── prompts.py Prompt 版本管理(Issue #47 雏形) -├── hallucination.py 幻觉/事实性校验中间件(Issue #47 雏形) +├── prompts.py Prompt 版本管理(Issue #47) +├── hallucination.py 幻觉/事实性校验中间件(Issue #47) +├── evaluate_hallucination.py 幻觉/事实性评测报告脚本(Issue #47) ├── gateway.py 混合网关主编排(EPIC #6 主体交付) ├── config/ │ ├── dlp.template.yaml 模板 DLP 规则资产(ti-cl4 示例) @@ -107,16 +108,65 @@ if result.needs_human: - **敏感度路由**(router.py,Issue #43 雏形):模板配置驱动,DLP 拦截 fail-closed 强制 block,未知内容保守走本地(数据不出厂); -- **Prompt 版本管理**(prompts.py,Issue #47 雏形):semver 版本库、 +- **Prompt 版本管理**(prompts.py,Issue #47):semver 版本库、 运行时绑定(可复现)、一键回滚、变更审计; -- **幻觉/事实性校验**(hallucination.py,Issue #47 雏形):`[来源: X]` - 引用溯源强制校验 + 高利害信度阈值 → 人工确认; +- **幻觉/事实性校验**(hallucination.py,Issue #47):`[来源: X]` + 引用溯源强制校验 + 高利害信度阈值 → 人工确认,并与 Prompt 版本库联动 + (评测按 `name@version` 分解,Prompt 变更后可对比各版本事实一致性); - **推理后端抽象**(gateway.py 内 `InferenceBackend`):业务代码只依赖 接口,本地 70B / 云端 API 具体接入由子任务 #44 / #45 实现。 +## Prompt 版本管理 + 幻觉/事实性校验(Issue #47) + +对应 PRD 5.4「Prompt 版本管理 + 幻觉/事实性校验」。 + +### Prompt 版本管理(prompts.py) + +所有提示词模板纳入版本库(semver),变更须评审并记录(同版本号覆盖报错), +支持一键回滚;生产流程运行时按 `(name, version)` 绑定,可复现。 + +```python +from llm_gateway.prompts import PromptRegistry + +reg = PromptRegistry.from_template_config("config/prompts.template.yaml") +pv = reg.get("qa", version="1.0.0") # 显式绑定旧版本:行为不受后续变更影响 +rendered = pv.render(query="炉温偏高怎么处理") +reg.promote("qa", "1.0.1") +reg.rollback("qa") # 一键回滚 +``` + +### 幻觉/事实性校验(hallucination.py) + +RAG 答案强制引用溯源:输出中所有 `[来源: X]` 声明必须命中本次 RAG 检索的 +文档片段,否则判 `unsupported`(幻觉嫌疑);高利害输出(处置建议 / 报警解释) +要求信度 ≥ 阈值,否则 `human_review` 转人工确认。 + +```python +from llm_gateway.hallucination import HallucinationGuard + +guard = HallucinationGuard(default_threshold=0.8) +v = guard.check( + answer="建议降温。[来源: 沸腾氯化炉异常处置SOP]", + sources=["沸腾氯化炉异常处置SOP"], + confidence=0.95, high_stakes=True, + prompt_name="alarm_explain", prompt_version="1.0.0", # 与版本库联动 +) +print(v.action) # pass / human_review / unsupported +``` + +### 定期评测报告脚本(evaluate_hallucination.py) + +PRD 5.4「定期用评测集检验事实一致性」:评测集 JSON + 版本库配置 → +Markdown 评测报告(按 `name@version` 分解,Prompt 变更后可对比各版本幻觉率)。 + +```bash +cd core/llm-gateway +python evaluate_hallucination.py --demo --output reports/hallucination.md +python evaluate_hallucination.py --samples eval_set.json --output report.md +``` + ## 后续子任务(EPIC #6 拆分,待扩展) - 敏感度路由调优与路由准确率评估脚本(Issue #49,本版已提供评估入口); - 本地 70B 模型接入与推理封装(Issue #44); -- 云端 API(Qwen/DeepSeek)接入与安全网关(Issue #45); -- Prompt 版本管理 + 幻觉校验中间件完善(Issue #47,本版已提供核心)。 +- 云端 API(Qwen/DeepSeek)接入与安全网关(Issue #45)。 diff --git a/core/llm-gateway/__init__.py b/core/llm-gateway/__init__.py index 09da2cc..a04f83e 100644 --- a/core/llm-gateway/__init__.py +++ b/core/llm-gateway/__init__.py @@ -10,16 +10,17 @@ 全量审计。 - router 敏感度路由规则引擎(Issue #43 雏形):敏感度分级路由(local/cloud/ block),模板配置驱动,DLP 拦截即 fail-closed 转 block。 -- prompts Prompt 版本管理(Issue #47 雏形):semver 版本库、运行时绑定、 +- prompts Prompt 版本管理(Issue #47 完成交付):semver 版本库、运行时绑定、 一键回滚、变更审计。 -- hallucination 幻觉/事实性校验中间件(Issue #47 雏形):引用溯源 + - 高利害信度阈值 → 人工确认。 +- hallucination 幻觉/事实性校验中间件(Issue #47 完成交付):引用溯源 + + 高利害信度阈值 → 人工确认,与 Prompt 版本库联动(评测按 name@version + 分解,配套评测报告脚本 evaluate_hallucination.py)。 - gateway 混合网关主编排(EPIC #6 主体):路由 → 生成 → 溯源校验 → DLP 出站防线,端到端闭环。 测试:`python -m unittest discover -s tests -v`(在 core/llm-gateway 目录下执行)。 """ -__version__ = "0.2.0" +__version__ = "0.3.0" from .dlp import ( DLP_DEFAULT_RULES, diff --git a/core/llm-gateway/evaluate_hallucination.py b/core/llm-gateway/evaluate_hallucination.py new file mode 100644 index 0000000..3a0ec89 --- /dev/null +++ b/core/llm-gateway/evaluate_hallucination.py @@ -0,0 +1,156 @@ +# -*- 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:])) diff --git a/core/llm-gateway/hallucination.py b/core/llm-gateway/hallucination.py index ff04aab..50b2676 100644 --- a/core/llm-gateway/hallucination.py +++ b/core/llm-gateway/hallucination.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -"""iAOP-Core · LLM 网关 —— 幻觉/事实性校验中间件(EPIC #6 主体,Issue #47 雏形)。 +"""iAOP-Core · LLM 网关 —— 幻觉/事实性校验中间件(Issue #47 完成交付)。 对应 PRD 5.4「④ LLM 网关 + RAG」: - **事实性校验**:RAG 答案强制**引用溯源**(返回命中文档片段+来源); @@ -12,12 +12,13 @@ (无源引用 = 幻觉嫌疑); - **信度阈值**:对高利害输出(处置建议 / 报警解释)要求信度 ≥ 阈值, 低于阈值返回 `human_review`(转人工确认,PRD 5.4 异常时转人工); -- **评测集检验**:`evaluate()` 对 (prompt, answer, expected_sources) 样本 - 批量评估事实一致性(供"定期评测"脚本调用)。 - -设计说明(供子任务 #47 继续细化): -- 本版实现校验核心(溯源 + 信度阈值 + 评测入口); -- 子任务 #47 将在此基础上补齐与 Prompt 版本库的联动与评测报告脚本。 +- **与 Prompt 版本库联动**:`check()` / `evaluate()` 可携带 + `prompt_name` / `prompt_version`(运行时绑定,可复现),评测报告按 + `name@version` 分解——Prompt 变更后可用同一评测集对比各版本事实一致性, + 定位"版本升级导致幻觉率上升"(PRD 5.4 定期评测); +- **评测报告**:`evaluate()` 产出结构化报告,`render_evaluation_report()` + 渲染 Markdown 报告,配套 CLI 脚本 `evaluate_hallucination.py` + (评测集 JSON + 版本库配置 → 报告文件)。 测试:`python -m unittest discover -s tests -v`(在 core/llm-gateway 目录下执行)。 """ @@ -43,6 +44,8 @@ class GuardVerdict: threshold: float # 本次校验使用的信度阈值 action: str # pass / human_review / unsupported missing_sources: List[str] = field(default_factory=list) + prompt_name: Optional[str] = None # 生成本次答案的 Prompt 模板名(版本库联动) + prompt_version: Optional[str] = None # 运行时绑定的模板版本(可复现) verdict_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12]) created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) @@ -55,6 +58,8 @@ class GuardVerdict: "threshold": self.threshold, "action": self.action, "missing_sources": self.missing_sources, + "prompt_name": self.prompt_name, + "prompt_version": self.prompt_version, "answer": self.answer, } @@ -77,8 +82,14 @@ class HallucinationGuard: def check(self, answer: str, sources: Sequence[str], confidence: float = 1.0, high_stakes: bool = False, - threshold: Optional[float] = None) -> GuardVerdict: - """校验一条模型输出。返回结论(不修改输出,由调用方决定如何处置)。""" + threshold: Optional[float] = None, + prompt_name: Optional[str] = None, + prompt_version: Optional[str] = None) -> GuardVerdict: + """校验一条模型输出。返回结论(不修改输出,由调用方决定如何处置)。 + + `prompt_name` / `prompt_version`:生成本次答案的 Prompt 模板与其 + 运行时绑定版本(与 PromptRegistry 联动,可复现,随审计落库)。 + """ th = threshold if threshold is not None else self.default_threshold # 1) 引用溯源:输出中声明的来源必须真实存在 declared = _SOURCE_REF_RE.findall(answer) @@ -97,40 +108,165 @@ class HallucinationGuard: verdict = GuardVerdict( answer=answer, supported=supported, confidence=confidence, threshold=th, action=action, missing_sources=missing, + prompt_name=prompt_name, prompt_version=prompt_version, ) self._audit.append(verdict.to_dict()) return verdict # -- 评测集检验(定期事实一致性评测入口) ------------------------------ - def evaluate(self, samples: List[Dict[str, object]]) -> Dict[str, object]: - """批量评估事实一致性。 + def evaluate(self, samples: List[Dict[str, object]], + registry: Optional[object] = None, + threshold: Optional[float] = None) -> Dict[str, object]: + """批量评估事实一致性,产出结构化评测报告。 - `samples`:`[{"answer", "sources", "confidence", "high_stakes"}, ...]`。 - 返回支持率 / 人工复核率 / 未支持率。子任务 #47 将扩展为评测报告。 + `samples`:`[{"answer", "sources", "confidence", "high_stakes", + "prompt_name", "prompt_version"}, ...]`; + `registry`:`PromptRegistry`(可选)——与 Prompt 版本库联动: + - 样本带 `prompt_name` 时,按 `prompt_version`(缺省取当前默认版本) + 解析为 `name@version`,报告按此分解,便于对比各版本事实一致性; + - 版本库中不存在该模板时,样本记录 `prompt_error` 而非中断评测。 + + 返回:`total` / `supported_rate` / `human_review_rate` / + `unsupported_rate`,以及 `prompts`(按 `name@version` 分解的明细, + 无 prompt 样本时为 `{}`)与 `samples`(逐样本记录)。 """ total = len(samples) + base: Dict[str, object] = {"total": total} if total == 0: - return {"supported_rate": 0.0, "human_review_rate": 0.0, "total": 0} + base.update({"supported_rate": 0.0, "human_review_rate": 0.0, + "unsupported_rate": 0.0, "prompts": {}, + "samples": []}) + return base + + per_prompt: Dict[str, Dict[str, object]] = {} + sample_records: List[Dict[str, object]] = [] supported = 0 human = 0 for s in samples: + answer = str(s.get("answer", "")) + sources = [str(x) for x in s.get("sources", [])] + confidence = float(s.get("confidence", 1.0)) + high_stakes = bool(s.get("high_stakes", False)) + name = s.get("prompt_name") + version = s.get("prompt_version") + prompt_error: Optional[str] = None + label: Optional[str] = None + + # 与 Prompt 版本库联动:解析运行时绑定版本(可复现) + if name and registry is not None: + try: + if version is None: + version = registry.current(str(name)).version + else: + registry.get(str(name), str(version)) # 校验存在性 + label = f"{name}@{version}" + except (KeyError, ValueError) as exc: + prompt_error = f"{exc}" + label = f"{name}@unknown" + v = self.check( - answer=str(s.get("answer", "")), - sources=[str(x) for x in s.get("sources", [])], - confidence=float(s.get("confidence", 1.0)), - high_stakes=bool(s.get("high_stakes", False)), + answer=answer, sources=sources, confidence=confidence, + high_stakes=high_stakes, threshold=threshold, + prompt_name=str(name) if name else None, + prompt_version=str(version) if version else None, ) if v.supported: supported += 1 if v.action == "human_review": human += 1 - return { + + if label is not None: + group = per_prompt.setdefault(label, { + "total": 0, "supported": 0, "human_review": 0, + "missing_sources": [], + }) + group["total"] += 1 + if v.supported: + group["supported"] += 1 + if v.action == "human_review": + group["human_review"] += 1 + if v.missing_sources: + group["missing_sources"].extend(v.missing_sources) + + record = { + "answer": answer, "action": v.action, "supported": v.supported, + "confidence": confidence, "prompt": label, + "missing_sources": v.missing_sources, + } + if prompt_error: + record["prompt_error"] = prompt_error + sample_records.append(record) + + # 补齐每组未支持数(total - supported 中可能含 human_review) + for g in per_prompt.values(): + g["unsupported"] = g["total"] - g["supported"] + g["supported_rate"] = round(g["supported"] / g["total"], 4) + g["human_review_rate"] = round(g["human_review"] / g["total"], 4) + g["unsupported_rate"] = round(g["unsupported"] / g["total"], 4) + + base.update({ "supported_rate": round(supported / total, 4), "human_review_rate": round(human / total, 4), "unsupported_rate": round((total - supported) / total, 4), - "total": total, - } + "prompts": per_prompt, + "samples": sample_records, + }) + return base + + # -- 评测报告渲染 ------------------------------------------------------ + + def render_evaluation_report(self, report: Dict[str, object]) -> str: + """把 `evaluate()` 的结构化报告渲染为 Markdown 文本(评测报告脚本用)。""" + total = int(report.get("total", 0)) + lines: List[str] = [ + "# LLM 网关 · 幻觉/事实性校验评测报告", + "", + f"- 生成时间:{datetime.now(timezone.utc).isoformat(timespec='seconds')}", + f"- 样本总数:{total}", + ] + if total: + lines.append( + "- 支持率 {:.1%}|人工复核率 {:.1%}|未支持率 {:.1%}".format( + float(report.get("supported_rate", 0.0)), + float(report.get("human_review_rate", 0.0)), + float(report.get("unsupported_rate", 0.0)), + ) + ) + lines.append("") + + prompts = report.get("prompts") or {} + if prompts: + lines += [ + "## 按 Prompt 版本分解", + "", + "| Prompt@版本 | 样本数 | 支持数 | 支持率 | 人工复核 | 未支持 |", + "|---|---|---|---|---|---|", + ] + for label in sorted(prompts): + g = prompts[label] + lines.append( + "| {} | {} | {} | {:.1%} | {} | {} |".format( + label, g["total"], g["supported"], + float(g["supported_rate"]), g["human_review"], + g["unsupported"], + ) + ) + lines.append("") + + bad = [r for r in report.get("samples", []) + if r.get("action") in ("unsupported", "human_review")] + if bad: + lines += ["## 未通过样本明细", ""] + for i, r in enumerate(bad, 1): + prompt = r.get("prompt") or "(未关联 Prompt 版本)" + lines.append( + f"{i}. **[{r['action']}]** {prompt}\n" + f" - 答案:{r['answer']}\n" + f" - 缺失来源:{r.get('missing_sources') or '—'}" + ) + lines.append("") + return "\n".join(lines) # -- 审计 -------------------------------------------------------------- diff --git a/core/llm-gateway/prompts.py b/core/llm-gateway/prompts.py index 11ab605..2b20e24 100644 --- a/core/llm-gateway/prompts.py +++ b/core/llm-gateway/prompts.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -"""iAOP-Core · LLM 网关 —— Prompt 版本管理(EPIC #6 主体,Issue #47 雏形)。 +"""iAOP-Core · LLM 网关 —— Prompt 版本管理(Issue #47 完成交付)。 对应 PRD 5.4「④ LLM 网关 + RAG」: - **Prompt 版本管理**:所有提示词模板纳入版本库(semver),变更须评审并记录, @@ -14,9 +14,10 @@ `rollback(name)` 回滚到上一版本(一键回滚); - **变更审计**:`update()` / `promote()` / `rollback()` 均落结构化变更记录。 -设计说明(供子任务 #47 继续细化): -- 本版实现版本库核心(绑定 / 回滚 / 审计); -- 子任务 #47 将在此基础上补齐幻觉/事实性校验中间件(见 hallucination.py)。 +与幻觉/事实性校验中间件(hallucination.py)联动:评测样本携带 +`prompt_name` / `prompt_version`,`HallucinationGuard.evaluate(registry=...)` +按 `name@version` 分解事实一致性,Prompt 变更后可用同一评测集对比 +各版本幻觉率(配套 CLI:`evaluate_hallucination.py`)。 测试:`python -m unittest discover -s tests -v`(在 core/llm-gateway 目录下执行)。 """ diff --git a/core/llm-gateway/tests/test_evaluate_script.py b/core/llm-gateway/tests/test_evaluate_script.py new file mode 100644 index 0000000..2b976b7 --- /dev/null +++ b/core/llm-gateway/tests/test_evaluate_script.py @@ -0,0 +1,55 @@ +# -*- coding: utf-8 -*- +"""评测报告脚本(evaluate_hallucination.py)端到端测试:--demo / --samples / --output。""" +import json +import os +import subprocess +import sys +import tempfile +import unittest + +LLM_GW_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +SCRIPT = os.path.join(LLM_GW_DIR, "evaluate_hallucination.py") + + +def _run(args, cwd): + env = dict(os.environ) + env["PYTHONIOENCODING"] = "utf-8" # 避免 Windows 控制台 GBK 编码问题 + return subprocess.run( + [sys.executable, SCRIPT] + args, + cwd=cwd, capture_output=True, text=True, encoding="utf-8", env=env, + timeout=60, + ) + + +class EvaluateScriptTest(unittest.TestCase): + def test_demo_to_stdout(self): + proc = _run(["--demo"], cwd=LLM_GW_DIR) + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertIn("LLM 网关 · 幻觉/事实性校验评测报告", proc.stdout) + self.assertIn("样本总数", proc.stdout) + self.assertIn("支持率", proc.stdout) + self.assertIn("按 Prompt 版本分解", proc.stdout) + + def test_samples_json_to_output_file(self): + samples = [ + {"answer": "好[来源: X]", "sources": ["X"], "prompt_name": "qa"}, + {"answer": "坏[来源: 不存在]", "sources": ["Y"], "prompt_name": "qa"}, + ] + with tempfile.TemporaryDirectory() as tmp: + samples_path = os.path.join(tmp, "samples.json") + report_path = os.path.join(tmp, "report.md") + with open(samples_path, "w", encoding="utf-8") as fh: + json.dump(samples, fh, ensure_ascii=False) + + proc = _run(["--samples", samples_path, "--output", report_path], + cwd=LLM_GW_DIR) + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertTrue(os.path.exists(report_path)) + with open(report_path, "r", encoding="utf-8") as fh: + text = fh.read() + self.assertIn("样本总数:2", text) + self.assertIn("qa@", text) # 与版本库联动:name@version 分解 + + +if __name__ == "__main__": + unittest.main() diff --git a/core/llm-gateway/tests/test_hallucination.py b/core/llm-gateway/tests/test_hallucination.py index e8b70c7..19dd934 100644 --- a/core/llm-gateway/tests/test_hallucination.py +++ b/core/llm-gateway/tests/test_hallucination.py @@ -87,10 +87,13 @@ class EvaluateTest(unittest.TestCase): # 支持 2 条(a/c),human_review 1 条(c) self.assertEqual(report["supported_rate"], round(2 / 3, 4)) self.assertEqual(report["human_review_rate"], round(1 / 3, 4)) + self.assertEqual(len(report["samples"]), 3) + self.assertEqual(report["prompts"], {}) def test_empty_samples(self): report = self.guard.evaluate([]) self.assertEqual(report["total"], 0) + self.assertEqual(report["samples"], []) def test_audit_records(self): self.guard.check("x[来源: A]", sources=["A"]) @@ -99,5 +102,76 @@ class EvaluateTest(unittest.TestCase): self.assertEqual(records[0]["action"], "pass") +class PromptLinkageTest(unittest.TestCase): + """Issue #47:与 Prompt 版本库的联动(check / evaluate 按 name@version 分解)。""" + + def setUp(self): + from llm_gateway.prompts import PromptRegistry + + self.guard = HallucinationGuard() + self.reg = PromptRegistry() + self.reg.update("qa", "问题:{query}", "1.0.0") + self.reg.update("qa", "问题:{query} 请引用SOP", "1.0.1") + self.reg.promote("qa", "1.0.1") + + def test_check_records_prompt_version(self): + verdict = self.guard.check( + "按SOP处理。[来源: SOP]", sources=["SOP"], + prompt_name="qa", prompt_version="1.0.0", + ) + self.assertEqual(verdict.prompt_name, "qa") + self.assertEqual(verdict.prompt_version, "1.0.0") + records = self.guard.drain_audit() + self.assertEqual(records[0]["prompt_name"], "qa") + self.assertEqual(records[0]["prompt_version"], "1.0.0") + + def test_evaluate_breakdown_by_prompt_version(self): + samples = [ + {"answer": "好[来源: X]", "sources": ["X"], "prompt_name": "qa", "prompt_version": "1.0.0"}, + {"answer": "坏[来源: Y]", "sources": ["Z"], "prompt_name": "qa", "prompt_version": "1.0.0"}, + {"answer": "好[来源: X]", "sources": ["X"], "prompt_name": "qa"}, # 缺省取当前默认 1.0.1 + ] + report = self.guard.evaluate(samples, registry=self.reg) + prompts = report["prompts"] + self.assertIn("qa@1.0.0", prompts) + self.assertIn("qa@1.0.1", prompts) + self.assertEqual(prompts["qa@1.0.0"]["total"], 2) + self.assertEqual(prompts["qa@1.0.0"]["supported"], 1) + self.assertEqual(prompts["qa@1.0.1"]["total"], 1) + # 逐样本记录携带 prompt 标签 + self.assertEqual(report["samples"][0]["prompt"], "qa@1.0.0") + + def test_evaluate_explicit_version_validated(self): + samples = [ + {"answer": "x[来源: A]", "sources": ["A"], "prompt_name": "qa", "prompt_version": "1.0.1"}, + ] + report = self.guard.evaluate(samples, registry=self.reg) + self.assertIn("qa@1.0.1", report["prompts"]) + self.assertNotIn("prompt_error", report["samples"][0]) + + def test_evaluate_unknown_prompt_does_not_crash(self): + samples = [ + {"answer": "x[来源: A]", "sources": ["A"], "prompt_name": "not_exist"}, + ] + report = self.guard.evaluate(samples, registry=self.reg) + self.assertEqual(report["total"], 1) + self.assertEqual(report["samples"][0]["prompt"], "not_exist@unknown") + self.assertIn("prompt_error", report["samples"][0]) + + def test_render_report_contains_rates_and_prompt_section(self): + samples = [ + {"answer": "好[来源: X]", "sources": ["X"], "prompt_name": "qa", "prompt_version": "1.0.0"}, + {"answer": "坏[来源: 不存在]", "sources": ["Z"], "prompt_name": "qa", "prompt_version": "1.0.0"}, + ] + report = self.guard.evaluate(samples, registry=self.reg) + text = self.guard.render_evaluation_report(report) + self.assertIn("样本总数:2", text) + self.assertIn("支持率", text) + self.assertIn("## 按 Prompt 版本分解", text) + self.assertIn("qa@1.0.0", text) + self.assertIn("## 未通过样本明细", text) + self.assertIn("不存在", text) + + if __name__ == "__main__": unittest.main()