feat: 完成 issue #47 Prompt 版本管理 + 幻觉校验中间件(版本库联动 + 评测报告脚本)

This commit is contained in:
2026-08-04 23:15:39 +08:00
parent 34c15f7ff8
commit 8dc379c479
7 changed files with 509 additions and 36 deletions
+57 -7
View File
@@ -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)。
+5 -4
View File
@@ -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,
+156
View File
@@ -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:]))
+157 -21
View File
@@ -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)
# -- 审计 --------------------------------------------------------------
+5 -4
View File
@@ -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 目录下执行)。
"""
@@ -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()
@@ -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()