# -*- coding: utf-8 -*- """iAOP-Core · LLM 网关 —— 幻觉/事实性校验中间件（Issue #47 完成交付）。 对应 PRD 5.4「④ LLM 网关 + RAG」： - **事实性校验**：RAG 答案强制**引用溯源**（返回命中文档片段+来源）； 对高利害输出（如处置建议）设置信度阈值，低于阈值触发"人工确认"； 定期用评测集检验事实一致性。 本模块实现 `HallucinationGuard`： - **引用溯源校验**：模型输出中声称引用的片段（`[来源: ]`）必须能在 RAG 检索命中的文档片段中找到对应来源，找不到即判定 `unsupported` （无源引用 = 幻觉嫌疑）； - **信度阈值**：对高利害输出（处置建议 / 报警解释）要求信度 ≥ 阈值， 低于阈值返回 `human_review`（转人工确认，PRD 5.4 异常时转人工）； - **与 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 目录下执行）。 """ from __future__ import annotations import re import uuid from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Dict, List, Optional, Sequence # 输出中引用声明的格式：`[来源: 文档标题]` 或 `[src: doc_id]` _SOURCE_REF_RE = re.compile(r"\[来源[:：]\s*([^\]]+)\]", re.IGNORECASE) @dataclass(frozen=True) class GuardVerdict: """一次事实性校验的结论。""" answer: str supported: bool # 所有引用声明均有真实来源 confidence: float # 调用方给出的信度（0~1） 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()) def to_dict(self) -> Dict[str, object]: return { "verdict_id": self.verdict_id, "created_at": self.created_at, "supported": self.supported, "confidence": self.confidence, "threshold": self.threshold, "action": self.action, "missing_sources": self.missing_sources, "prompt_name": self.prompt_name, "prompt_version": self.prompt_version, "answer": self.answer, } class HallucinationGuard: """幻觉/事实性校验中间件。 `check(answer, sources, confidence, high_stakes=False)`： - `sources`：本次 RAG 检索实际命中的文档标题列表； - `high_stakes=True`：启用信度阈值（处置建议 / 报警解释等）， 低于阈值 → `human_review`； - 输出中所有 `[来源: X]` 声明必须出现在 `sources` 中， 否则 → `unsupported`（缺失引用列表随结论返回）。 """ def __init__(self, default_threshold: float = 0.8) -> None: self.default_threshold = default_threshold self._audit: List[Dict[str, object]] = [] def check(self, answer: str, sources: Sequence[str], confidence: float = 1.0, high_stakes: bool = False, 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) available = set(sources) missing = [s.strip() for s in declared if s.strip() not in available] supported = not missing # 2) 高利害 → 信度阈值 if high_stakes and confidence < th: action = "human_review" elif not supported: action = "unsupported" else: action = "pass" 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]], registry: Optional[object] = None, threshold: Optional[float] = None) -> Dict[str, object]: """批量评估事实一致性，产出结构化评测报告。 `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: 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=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 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), "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) # -- 审计 -------------------------------------------------------------- def drain_audit(self) -> List[Dict[str, object]]: out, self._audit = self._audit, [] return out def __repr__(self) -> str: # pragma: no cover - 调试辅助 return f""