143 lines
6.0 KiB
Python
143 lines
6.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""iAOP-Core · LLM 网关 —— 幻觉/事实性校验中间件（EPIC #6 主体，Issue #47 雏形）。
|
||
|
||
对应 PRD 5.4「④ LLM 网关 + RAG」：
|
||
- **事实性校验**：RAG 答案强制**引用溯源**（返回命中文档片段+来源）；
|
||
对高利害输出（如处置建议）设置信度阈值，低于阈值触发"人工确认"；
|
||
定期用评测集检验事实一致性。
|
||
|
||
本模块实现 `HallucinationGuard`：
|
||
- **引用溯源校验**：模型输出中声称引用的片段（`[来源: <doc>]`）必须能在
|
||
RAG 检索命中的文档片段中找到对应来源，找不到即判定 `unsupported`
|
||
（无源引用 = 幻觉嫌疑）；
|
||
- **信度阈值**：对高利害输出（处置建议 / 报警解释）要求信度 ≥ 阈值，
|
||
低于阈值返回 `human_review`（转人工确认，PRD 5.4 异常时转人工）；
|
||
- **评测集检验**：`evaluate()` 对 (prompt, answer, expected_sources) 样本
|
||
批量评估事实一致性（供"定期评测"脚本调用）。
|
||
|
||
设计说明（供子任务 #47 继续细化）：
|
||
- 本版实现校验核心（溯源 + 信度阈值 + 评测入口）；
|
||
- 子任务 #47 将在此基础上补齐与 Prompt 版本库的联动与评测报告脚本。
|
||
|
||
测试：`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)
|
||
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,
|
||
"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) -> GuardVerdict:
|
||
"""校验一条模型输出。返回结论（不修改输出，由调用方决定如何处置）。"""
|
||
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,
|
||
)
|
||
self._audit.append(verdict.to_dict())
|
||
return verdict
|
||
|
||
# -- 评测集检验（定期事实一致性评测入口） ------------------------------
|
||
|
||
def evaluate(self, samples: List[Dict[str, object]]) -> Dict[str, object]:
|
||
"""批量评估事实一致性。
|
||
|
||
`samples`：`[{"answer", "sources", "confidence", "high_stakes"}, ...]`。
|
||
返回支持率 / 人工复核率 / 未支持率。子任务 #47 将扩展为评测报告。
|
||
"""
|
||
total = len(samples)
|
||
if total == 0:
|
||
return {"supported_rate": 0.0, "human_review_rate": 0.0, "total": 0}
|
||
supported = 0
|
||
human = 0
|
||
for s in samples:
|
||
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)),
|
||
)
|
||
if v.supported:
|
||
supported += 1
|
||
if v.action == "human_review":
|
||
human += 1
|
||
return {
|
||
"supported_rate": round(supported / total, 4),
|
||
"human_review_rate": round(human / total, 4),
|
||
"unsupported_rate": round((total - supported) / total, 4),
|
||
"total": total,
|
||
}
|
||
|
||
# -- 审计 --------------------------------------------------------------
|
||
|
||
def drain_audit(self) -> List[Dict[str, object]]:
|
||
out, self._audit = self._audit, []
|
||
return out
|
||
|
||
def __repr__(self) -> str: # pragma: no cover - 调试辅助
|
||
return f"<HallucinationGuard threshold={self.default_threshold}>"
|