Critic 双层防幻觉
昨天(Day 06)鸟瞰了 toolkit 全貌;今天钻进第一台核心设备——Critic 防幻觉。大模型会一本正经地编造证据、过度自信。Critic 是专门给结论"挑刺"的评审员——用"先 0 成本代码规则、再便宜 LLM"两层把幻觉拦下来。
什么是"幻觉",为什么必须防
幻觉(hallucination)指大模型生成了"看起来很合理、其实是编造"的内容。在本框架的场景里,典型幻觉有:
- 引用了一个根本不存在的证据 ID(比如凭空捏造一条 trace_id);
- 结论过度自信(明明证据不足却说"确定就是它");
- 只引 1 个证据就下大结论、忽略反例;
- 对高风险动作(重启、回滚)不加"先验证"的提示。
对一个要给 SRE 做故障根因的 Agent 来说,幻觉是致命的——你不能让它把运维人员带到错误方向。所以框架专门设了 Critic 节点:在综合出结论后、返回给用户前,先自我审查一遍。代码在 packages/ai-trust-toolkit/src/ai_trust_toolkit/critic/。
trace_id 来"圆场"。你没法靠调 prompt 让它 100% 不编,只能在它输出之后用另一套机制去核对。这就是 Critic 存在的根本理由:把"生成"和"审查"拆成两件事、由不同逻辑负责。conclusion 大致是:{"confidence": "高", "hypotheses": [{"risk_level": "high", "actions": [{"name": "restart", "verify_first": true}]}], "evidence_ids": ["ev-1", "ev-2"]}Critic 要回答的问题就是:这里的
confidence:"高" 合法吗?evidence_ids:["ev-1"] 是真的还是编的?risk_level:"high" 的动作带没带 verify_first?——全是可以用代码逐条核对的。
双层拦截总图:先便宜后贵
Critic 的核心巧思是分两层,按"成本递增"顺序拦截——能用 0 成本代码解决的,绝不浪费 LLM 调用:
代码层:纯 Python 规则检查
机械性错误(编造 ID、置信度非法、专家数不够)用确定性代码就能 100% 拦下。先跑这层,不过直接返回,根本不调 LLM。
毫秒级
100% 准确
LLM 层:Haiku 语义审查
L1 过了,才用便宜的 Haiku 做"语义级"审查:结论和证据逻辑一致吗?有没有过度外推?教学性够不够?
Haiku
两层都过 → 结论可信,放行
否则打回重做(见 L05 三态路由)。
L1 的数据结构与三个常量
先看 L1 用到的数据结构和"标准答案表"——不理解它们,后面 4 项检查会看不懂。代码在 critic/layer1.py 开头(layer1.py:24-41):
# layer1.py:24 —— 哪些 state 字段装着专家产出的 finding(业务可覆盖)
DEFAULT_SP_FIELDS: tuple[str, ...] = (
"trace_finding", "metric_finding", "deploy_finding", "log_finding",
)
# layer1.py:31 —— 置信度 / 风险等级的"合法取值表"
VALID_CONFIDENCE_VALUES: frozenset[str] = frozenset({"高", "中", "低"})
VALID_RISK_LEVELS: frozenset[str] = frozenset({"high", "medium", "low"})
@dataclass # layer1.py:35
class Layer1CheckResult:
passed: bool
violations: list[str] = field(default_factory=list)
detail: dict[str, Any] = field(default_factory=dict)
DEFAULT_SP_FIELDS4 位专家把结果分别写在 state 的这 4 个字段里。它是参数默认值而非写死——因为不同 agent 的专家字段名不同(Day 05 的 sre-rca 用这四个,别的 agent 可以传自己的)。VALID_CONFIDENCE_VALUES置信度只允许中文 "高"/"中"/"低"三个值。模型要是输出 "very high" 或 "0.9" 就算 schema 违规。VALID_RISK_LEVELS风险等级只允许英文 "high"/"medium"/"low"。注意和置信度一个中文一个英文——这是真实代码的约定,别记混。Layer1CheckResult4 项检查跑完的汇总结果:passed=整体过没过;violations=人话违规列表;detail=诊断数据(合法 ID 数、实际专家数等)。| 字段 | 类型 | 作用 |
|---|---|---|
passed | bool | 只要 violations 为空就是 True(见 L06) |
violations | list[str] | 每条是一句人话,如 "① 编造 evidence_id: ['M-9']" |
detail | dict | 如 {"valid_ids_count":3,"actual_specialists":2},给可观测用 |
frozenset 而不是普通 set / list?
三点。① 不可变:这是"标准答案表",任何代码都不该往里加值,frozenset 从语言层面禁止 .add(),防止被意外污染;② 可当模块级常量安全共享:可变 set 做全局常量有"被某处偷偷改掉"的风险,冻结后天然线程安全;③ 成员判断 O(1):confidence not in VALID_CONFIDENCE_VALUES 是哈希查找,比 list 的逐个比对快。用 list 也能跑,但语义上"这是一张不该变的表"就丢了。sp_fields 做成参数、而不是写死在函数里?
因为 toolkit 要被多个业务 agent 复用。sre-rca 的专家字段叫 trace_finding 等,但 alert-triage 可能叫别的。做成"默认值 + 可覆盖参数",既让 90% 的 agent 开箱即用(用默认),又给剩下 10% 留了口子(传自己的)。这就是"约定优于配置,但配置永远可覆盖"。先收集"合法证据 ID 集合"
4 项检查里最硬核的一项——抓编造 ID——依赖一个前置步骤:先把 4 位专家真实产出的证据 ID 全收集起来,组成"现场真有的材料清单"。这就是 collect_valid_evidence_ids()(layer1.py:44):
def collect_valid_evidence_ids(state, sp_fields=DEFAULT_SP_FIELDS, id_extractor=None):
if id_extractor is not None: # 业务可自定义抽取逻辑
return id_extractor(state)
valid: set[str] = set()
for field_name in sp_fields: # 遍历 4 个专家字段
finding = state.get(field_name)
if not finding or not isinstance(finding, dict):
continue # ← 边界:空 finding / 非 dict 直接跳过
# 形态 1:finding["evidence_ids"] = ["ev-1", "ev-2"]
for eid in finding.get("evidence_ids", []) or []:
if isinstance(eid, str):
valid.add(eid)
# 形态 2:finding["evidences"] = [{"id": "ev-1"}, ...]
for evidence in finding.get("evidences", []) or []:
if isinstance(evidence, dict):
eid = evidence.get("id")
if isinstance(eid, str):
valid.add(eid)
return valid
id_extractor 分支又是"可覆盖"套路:默认按下面两种形态抽,业务想自定义就传个函数进来。not finding or not isinstance(...,dict)边界处理:某专家没出结果(None)、或出的不是字典(脏数据)——直接 continue 跳过,绝不报错。宁可少收几个 ID,也不能让收集过程本身崩掉。两种 finding 形态真实项目里专家写法不统一:有的把 ID 平铺成 evidence_ids 列表,有的包成 evidences:[{id:...}] 对象。函数两种都认,兼容历史写法。isinstance(eid, str)层层 isinstance 校验——只收字符串 ID。防止专家误塞进 None/数字污染集合。or []小技巧:finding.get("evidence_ids") 可能返回 None,None or [] 兜成空列表,for 循环就不会炸。{"evidence_ids":["ev-1","ev-2"]};metric 专家:{"evidences":[{"id":"ev-3"}]};deploy 专家:None(挂了);log 专家:{"evidence_ids":["ev-2"]}(重复 ev-2)。收集结果 =
{"ev-1","ev-2","ev-3"}——两种形态都抽到了、None 被跳过、重复自动去重(因为是 set)。这个集合就是下一讲"抓编造 ID"的标准答案。
set 而不是 list?因为下一步要做的核心运算是集合差集 引用的ID - 合法的ID(找出编造的)。差集是 set 的原生操作、O(1) 成员判断;而且天然去重(多个专家引同一 ID 不会重复)。用 list 就得手写双重循环,又慢又啰嗦。L1 四项检查逐个走读
有了合法 ID 集合,现在逐个拆开 4 项检查。每项都是一个独立小函数,返回 (是否通过, 附加信息)——纯代码、0 成本、100% 确定。
① check_evidence_ids —— 抓编造证据 ID(layer1.py:75)
def check_evidence_ids(conclusion, valid_ids) -> tuple[bool, list[str]]:
cited_ids: set[str] = set()
for eid in conclusion.get("evidence_ids", []) or []:
if isinstance(eid, str):
cited_ids.add(eid)
fabricated = cited_ids - valid_ids # ← 核心:集合差集
return (not fabricated, sorted(fabricated))
一句话:把结论引用的 ID 减去合法 ID,剩下的就是凭空编的。fabricated 非空 → 不通过。这就是"用代码 100% 抓幻觉"最漂亮的一行。
{"ev-1","ev-2","ev-3"},结论引用 ["ev-1","ev-9"] → fabricated={"ev-9"} → 返回 (False, ["ev-9"]),当场判违规。② check_min_specialists —— 至少 2 位专家印证(layer1.py:88)
def check_min_specialists(conclusion, state, min_count=2, sp_fields=DEFAULT_SP_FIELDS):
actual = sum(
1 for f in sp_fields
if state.get(f) and isinstance(state[f], dict)
and not state[f].get("fallback", False) # ← 排除降级的假 finding
)
return (actual >= min_count, actual)
一句话:数一数几个专家真的出了结果。关键是 not ...get("fallback")——Day 08 兜底闸给的降级 finding 带 fallback:True,这里要排除掉,否则"4 个专家全挂、全是兜底占位"也会被误判成"证据充分"。默认要求 ≥2,防止单一证据下大结论。
③ check_schema —— 格式合法性(layer1.py:101)
def check_schema(conclusion) -> tuple[bool, list[str]]:
violations = []
confidence = conclusion.get("confidence")
if confidence not in VALID_CONFIDENCE_VALUES: # 用上 L03 的常量表
violations.append(f"confidence={confidence!r} 不在 {VALID_CONFIDENCE_VALUES}")
hypotheses = conclusion.get("hypotheses") or []
if not isinstance(hypotheses, list):
violations.append("hypotheses 必须是 list")
else:
for i, h in enumerate(hypotheses):
if not isinstance(h, dict): continue
risk = h.get("risk_level")
if risk is not None and risk not in VALID_RISK_LEVELS:
violations.append(f"hypotheses[{i}].risk_level={risk!r} 非法")
return (not violations, violations)
一句话:置信度必须 ∈ {高,中,低},每条假设的风险等级(若填了)必须 ∈ {high,medium,low}。注意 risk is not None——风险等级允许不填,但填了就必须合法(区分"没提供"和"提供了非法值")。
④ check_high_risk_verify_first —— 高危动作必须先验证(layer1.py:121)
def check_high_risk_verify_first(conclusion) -> tuple[bool, list[str]]:
violations = []
for h in conclusion.get("hypotheses", []) or []:
if not isinstance(h, dict): continue
if h.get("risk_level") == "high": # 只管高危假设
for j, a in enumerate(h.get("actions") or []):
if isinstance(a, dict) and not a.get("verify_first"):
violations.append(f"...actions[{j}] high-risk 缺少 verify_first")
return (not violations, violations)
一句话:凡是标了 risk_level:"high" 的假设,它下面每个动作都必须带 verify_first:true。这是一条安全红线——不能让 Agent 直接叫人"重启/回滚"却不提示先核实。
risk_level:"high" 的动作是 {"name":"restart","verify_first":true} → 带了 verify_first → ④ 通过。若模型漏写成
{"name":"restart"} → ④ 立刻违规:"high-risk 缺少 verify_first"。
fallback:True,那 ② 就会把"占位假数据"当成真证据,导致"证据充分"误判。所以 Day 08 的兜底闸和这里是强约定关系:兜底闸必须打标记,L1 才敢信任这个标记去数数。两处代码看似无关,实则靠一个字段咬合。汇总入口 run_layer1_checks + 控制流
4 项检查由总入口 run_layer1_checks()(layer1.py:135)串起来,顺序跑、攒违规、打包成 Layer1CheckResult:
def run_layer1_checks(state, valid_ids=None, *, min_specialists=2,
sp_fields=DEFAULT_SP_FIELDS, conclusion_field="conclusion"):
conclusion = state.get(conclusion_field) or {}
if valid_ids is None: # 没传就现场收集
valid_ids = collect_valid_evidence_ids(state, sp_fields)
violations = []
detail = {"valid_ids_count": len(valid_ids)}
ok, fab = check_evidence_ids(conclusion, valid_ids) # ①
if not ok: violations.append(f"① 编造 evidence_id: {fab}")
detail["fabricated_ids"] = fab
ok, n = check_min_specialists(conclusion, state, min_specialists, sp_fields) # ②
if not ok: violations.append(f"② 仅 {n} 位 Specialist 印证(需 ≥ {min_specialists})")
detail["actual_specialists"] = n
ok, schema_v = check_schema(conclusion) # ③
if not ok: violations.append(f"③ schema 违规: {schema_v}")
ok, risk_v = check_high_risk_verify_first(conclusion) # ④
if not ok: violations.append(f"④ 高风险动作缺 verify_first: {risk_v}")
return Layer1CheckResult(passed=not violations, violations=violations, detail=detail)
valid_ids=None 时现场收集调用方可以先算好 valid_ids 传进来(省一次重算),也可以不传让它自己收集。灵活。不短路、跑完 4 项注意它不是"第一项挂就 return",而是 4 项全跑、把所有违规攒进 violations。这样一次就能告诉模型"你哪几处都错了",重试时能一次改全,减少来回。passed = not violations违规列表空 = 通过。极简判据。detail顺手记下"合法 ID 数、编造的 ID、实际专家数"——这些数据后面进日志/指标,用来分析"我们的幻觉主要是哪类"。L2 LLM 层与"短路"设计
L1 全过了,才轮到 L2。代码在 critic/layer2.py。build_critic_node(config)(layer2.py:51)返回一个 async 节点,节点内部先跑 L1、L1 不过直接返回、根本不烧 LLM(layer2.py:59-71):
async def critic_node(state): # layer2.py:59
# ① Layer 1:代码层硬检查
valid_ids = (config.valid_ids_extractor(state) if config.valid_ids_extractor
else collect_valid_evidence_ids(state))
l1 = run_layer1_checks(state, valid_ids)
if not l1.passed:
return {"critique_passed": False,
"critique_feedback": "Layer 1 failed: " + " | ".join(l1.violations),
"critique_layer": 1, # ← 标记:被代码层拦下
"retry_count": 1} # 不调 LLM,直接返回!
# ② Layer 2:只有 L1 过了才调 Haiku
conclusion = state.get(config.conclusion_field) or {}
try:
llm = config.llm_factory() # 通常 get_llm("haiku")
response = await llm.ainvoke([{"role":"system","content":config.system_prompt},
{"role":"user","content": user_msg}])
parsed = _parse_critic_response(getattr(response, "content", str(response)))
return {"critique_passed": parsed["critique_passed"],
"critique_layer": 2, "retry_count": 1, ...}
except Exception as e:
if config.soft_fail: # ← 边界:LLM 挂了怎么办
return {"critique_passed": False, "critique_layer": 2,
"critique_feedback": f"Layer 2 error: {e}", "retry_count": 1}
raise
if not l1.passed: return这就是"短路"——L1 挂了立刻返回,llm_factory() 那行根本不会执行,省下这 ¥0.04。critique_layer=1 / 2记录"被哪层拦的":1=代码层、2=LLM 层。事后能统计"幻觉大多是机械错(1)还是语义错(2)",针对性优化。retry_count: 1每次审查都 +1。它配的是 Day 05 讲的累加 Reducer Annotated[int, add],所以多次审查会累加,L08 的路由靠它判断是否触顶。soft_fail 分支边界:Haiku 调用本身失败(超时/限流)。默认 soft_fail=True:不抛异常,而是当成"审查不通过"返回,让流程走重试而不是整图崩。CriticConfig(layer2.py:30)业务要提供:system_prompt(6 项铁律 prompt)、llm_factory、conclusion_field、valid_ids_extractor、soft_fail。
Haiku 返回的可能不是干净 JSON,所以有个三级兜底解析器 _parse_critic_response()(layer2.py:112):
def _parse_critic_response(content):
try:
return json.loads(content) # ① 先当纯 JSON
except (json.JSONDecodeError, TypeError):
pass
m = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", content, re.DOTALL) # ② markdown 代码块
if m:
try: return json.loads(m.group(1))
except json.JSONDecodeError: pass
passed = "critique_passed: true" in content.lower() or "PASS" in content[:200] # ③ 关键字兜底
return {"critique_passed": passed, "critique_feedback": content[:500]}
```json 包起来、有时前面还啰嗦两句。如果只 json.loads() 一次,稍不规范就抛异常、整个审查报废。三级兜底把"解析失败"的概率压到极低,最后还有关键字兜底不至于崩。这是"与不确定的 LLM 打交道"的通用防御姿势。| 输入结论 | L1 代码检查 | 调 Haiku? | 结果 |
|---|---|---|---|
引用编造 ID ev-9 | ❌ 不过 | 不调(省 ¥0.04) | False / layer=1 |
| ID 都真,但只 1 个专家出结果 | ❌ 不过 | 不调 | False / layer=1 |
| ID 真、专家够、格式对 | ✓ 过 | ✅ 调,做语义审查 | 看 Haiku / layer=2 |
👶 小白:Haiku 更聪明,为什么不直接全交给它,省得写 L1 那堆规则?
👨🏫 老师:三个原因。① 钱:每条都调 Haiku 要 ¥0.04,机械错误占很大比例,代码 0 成本就拦了;② 快:L1 毫秒级,LLM 要几百毫秒;③ 准:编造 ID 是确定性错误,代码 100% 抓得准,LLM 反而有概率看走眼。所以"能用尺子量的绝不请老师傅"——贵的判断力只留给语义问题。
三态路由 + 业务规则 L1
三态路由:把审查结果变成"下一步走哪"
Critic 审完,critic_router(retry_router.py:35,工厂 build_critic_router 在 retry_router.py:12)把结果翻译成三种走向:
✓ 通过
critique_passed=True
→ 去 writeback 沉淀经验
↻ 不过 & 没超次数
回 synthesizer 带着意见重做
■ 不过 & 超 2 次
→ END
诚实说"信息不足"
def critic_router(state) -> str: # retry_router.py:35
if state.get("critique_passed"):
return pass_dest # "writeback":通过
if state.get("retry_count", 0) >= max_retries:
return end_dest # END:触顶,诚实收尾
return retry_dest # "synthesizer":打回重做
retry_count。(这道路由也算 Day 08 的失败闸③。)
业务规则 L1:连 Haiku 都不调的轻量 Critic
有些高频 agent(如 alert-triage,P99 < 5 秒)连 Haiku 都嫌慢。于是 v0.3 抽出纯规则版 business_layer1.py——用 Python 规则校验,0 成本、<1ms、不调任何 LLM。核心是 Rule 数据类(business_layer1.py:34)+ build_business_critic_l1(business_layer1.py:51):
@dataclass
class Rule: # business_layer1.py:34
name: str # 出现在违规报告里,便于追踪
predicate: Callable[[dict], bool] # state -> 通过与否
error_msg: str | Callable[[dict], str] # 违规描述(可静态可动态)
severity: str = "error" # error 计入违规 / warn 只记日志
async def critic_l1_node(state): # business_layer1.py:81
violations = []
for rule in rules:
try:
ok = rule.predicate(state)
except Exception as e: # ← 边界:规则本身崩了
violations.append(f"{rule.name}: 规则计算异常 {type(e).__name__}: {e}")
continue # 视为违规,绝不放过
if not ok and rule.severity == "error":
msg = rule.error_msg(state) if callable(rule.error_msg) else rule.error_msg
violations.append(f"{rule.name}: {msg}")
return {"layer1_passed": len(violations) == 0, "layer1_violations": violations}
predicate一个 state -> bool 的函数,True 表示通过。规则的"判断逻辑"就装在这。except → 视为违规边界与安全默认:规则函数自己抛异常(比如访问了不存在的字段),不当作通过,而是记一条违规。安全系统的默认永远是"拿不准就拦",不是"拿不准就放"。severity: warn非 error 的规则只记不阻断——给"建议类"检查留了空间。业务用 5 个 helper 声明规则即可(都在 business_layer1.py):rule_in_allowed_set(:107 值在白名单)、rule_field_not_empty(:119)、rule_conditional_field(:128 条件必填)、rule_field_in_range(:152)、rule_list_min_length(:180):
critic_l1 = build_business_critic_l1([
rule_in_allowed_set("routing_decision", ALLOWED_DECISIONS),
rule_conditional_field( # 决定"立即呼叫"就必须填业务影响
when=lambda s: s["routing_decision"] == "page_immediately",
require_field="business_impact"),
])
builder.add_node("critic", critic_l1)
三档强度 + 今日小结 + 动手
框架并不强制每个 agent 都上"双层完整 Critic"。它按场景提供三档强度,本身就是很好的"权衡"教学案例:
| 档位 | 代表 agent | 配置 | 为什么 |
|---|---|---|---|
| 满配 | sre-rca | build_critic_node:L1 + L2 + 三态重试 | 根因分析责任重,值得反复审 |
| 仅 L1 | alert-triage | build_business_critic_l1:纯规则不调 LLM | 高频在线,P99<5s,延迟敏感 |
| 无 Critic | risk-reviewer | 不设 Critic 节点 | 刻意验证"toolkit 无 Critic 也可复用" |
🧠 今天你应该能回答
- 幻觉有哪些典型形式?(编造证据 ID、过度自信、证据不足下大结论)
- 为什么先 L1 后 L2?(L1 用代码 0 成本拦机械错误,通过了才花钱让 LLM 审语义)
- L1 怎么 100% 抓出编造的证据 ID?(
collect_valid_evidence_ids收集合法集合,cited - valid差集就是编造的) - 为什么常量用
frozenset、sp_fields做成参数?(不可变防污染 + O(1) 查找;跨 agent 复用可覆盖) - L2 为什么要写三级 JSON 解析、soft_fail 又是干嘛的?(LLM 输出不稳定要兜底;LLM 挂了当"不通过"而非崩图)
- 三态路由是哪三态?触顶为什么诚实收尾?(通过/重试/触顶;宁可说不知道也不误导)
- 三档 Critic 强度对应什么场景?(满配 build_critic_node / 仅 L1 build_business_critic_l1 / 无 Critic)
✋ 动手
# 1. 看 critic 模块能力清单
sed -n '1,61p' packages/ai-trust-toolkit/src/ai_trust_toolkit/critic/__init__.py
# 2. 读 L1 的 4 项检查
sed -n '44,135p' packages/ai-trust-toolkit/src/ai_trust_toolkit/critic/layer1.py
# 3. 读 L2 的"先L1后LLM"短路逻辑
sed -n '51,110p' packages/ai-trust-toolkit/src/ai_trust_toolkit/critic/layer2.py
# 4. 读 sre-rca 的 critic-v1.md(6 项检查 + 三态 + few-shot)
sed -n '1,90p' apps/sre-rca-agent/prompts/critic-v1.md
# 5. 跑相关单测感受
uv run pytest packages/ai-trust-toolkit/tests/test_critic_layer1.py -q