LLM 评测体系
第 5 章 规则判分：Exact Match / Regex / JSON Schema
规则判分不是 LLM 判分的低配版。Exact Match、数值匹配、正则、Schema 校验各有明确的适用边界——能用规则说清的判定，交给规则更快、更便宜，也更可复现。
本章要点
- 规则判分的四种主力形态：Exact Match、Numeric Match、Regex、Schema Validation
- 何时优先选规则、何时必须升级到 LLM-as-Judge
- 规则判分常见的工程陷阱：归一化、Unicode、空格、字段顺序
- 一个百行以内的"工业级规则判分器"完整代码
5.1 为什么先讲规则判分
回想第 4 章的多指标体系：Faithfulness 要 LLM-judge，Trajectory Match 要 LLM-judge，Answer Relevance 也要 LLM-judge。但不是所有判分都要 LLM 来做——很多判分用 100 行 Python 就能解决，速度和成本都比 LLM-judge 低几个数量级，而且结果完全可复现。
graph LR
A[判分方法选择] --> B{答案是否<br/>结构化?}
B -->|是<br/>有 ground truth| C[规则判分]
B -->|否<br/>需要语义理解| D[LLM-judge]
C --> C1[Exact Match]
C --> C2[Numeric Match]
C --> C3[Regex]
C --> C4[Schema Validation]
D --> D1[Pointwise]
D --> D2[Pairwise]
D --> D3[Reference-based]
style C fill:#dcfce7
style D fill:#fef3c7
工程上的原则：能用规则就不用 LLM-judge。原因是三个：
- 成本：规则判分是本地运算，无 API 费用
- 速度：千条量级样例的规则判分通常在秒级以内；LLM-judge 要分钟级
- 可重复：规则判分是确定性函数，同一输入永远同一输出；LLM-judge 自身有 noise
只在规则真的覆盖不到的时候才升级到 LLM-judge。本章拆解规则判分的四种主力形态，以及它们在实战中的具体写法。
5.2 Exact Match：看似简单，陷阱最多
5.2.1 朴素实现
def exact_match(prediction: str, expected: str) -> bool:
return prediction == expected这一行代码几乎从不能直接用。原因是：
exact_match("Paris", " Paris") # False (前导空格)
exact_match("paris", "Paris") # False (大小写)
exact_match("Paris.", "Paris") # False (标点)
exact_match("巴黎", "Paris") # False (语种)
exact_match("Paris,France", "Paris, France") # False (空格位置)每一个 False 都是真实业务里的"假阴性"——模型答对了但被判成错。如果你的评测集 100 题里有 20 条因为这种原因被误判，整个 accuracy 会偏低 20pp，团队会做错决策。
5.2.2 工业级 Exact Match：归一化是核心
import re, unicodedata
def normalize(s: str) -> str:
s = unicodedata.normalize("NFKC", s) # 全角半角统一（先做，见 §5.11.4）
s = s.lower()
s = re.sub(r"[^\w\s]", "", s) # 去标点
s = re.sub(r"\s+", " ", s) # 空白归一
return s.strip()
def exact_match(prediction: str, expected: str) -> bool:
return normalize(prediction) == normalize(expected)这段代码已经能应付大多数英文场景。中文场景还要额外处理：
- 全角 vs 半角标点（"。" vs "."；"，" vs ","）
- 简繁体（"发票" vs "發票"，简繁是不同 Unicode 码位，NFKC 不会合并）
- 数字大小写（"100" vs "一百"）
每一类归一化都是工程团队踩过坑后沉淀的，§5.8.6 与 §5.13 会继续补充。
5.2.3 何时不该用 Exact Match
Exact Match 只在以下场景合适：
- 答案是单值的（一个数字、一个实体、一个分类标签）
- 答案有正则化的标准形式
- 不需要"语义等价"的容忍度
如果你需要判定 "巴黎是法国首都" 和 "France's capital is Paris" 等价，Exact Match 一定失败——这种场景必须升级到 §4.3 的语义指标。
5.3 Numeric Match：数学题的标尺
GSM8K（小学数学题 benchmark）、MATH（竞赛数学）、SVAMP 等数据集的判分核心都是 Numeric Match。
5.3.1 难点：从自然语言里抽数字
模型回答经常长这样：
"Let's solve step by step. First, we have 3 apples + 4 apples = 7 apples.
Then we add 2 oranges. The total is 7 + 2 = 9 fruits."Reference 是 9。怎么从这段话里精确抽出 9？
- 不能用
r"\d+"找第一个数字——会匹配到 "3" - 不能用最后一个数字——可能模型最后一句是 "Hope this helps!"
- 标准做法：找特定格式标记，如
### 9或Answer: 9或\boxed{9}
openai/evals 里最接近的是 evals/elsuite/basic/match.py，但它并不抽数字：Match.eval_sample 把模型输出交给 evals/api.py:55 的 record_and_check_match，后者只判 sampled.startswith(option)（可选再用 separator 检查紧跟其后的那个字符是不是分隔符）。也就是说 evals 的口径是要求模型把答案放在输出开头，而不是从一大段推理里抽答案。旁边的 fuzzy_match.py 走的是另一条路——evals/elsuite/utils.py:46 的 fuzzy_match 归一化后做双向子串包含（s1 in s2 or s2 in s1），同样不做数字抽取。要在长回答里定位数字，得自己写上面这种 regex。
5.3.2 浮点数与单位
浮点数不能直接比字面量，得先把数字抽出来再带容差比。下面是示意实现，非 evals 源码原文——numeric_match / extract_number 都是这里现取的名字，仓库里并没有：
def numeric_match(prediction: str, expected: float, tolerance: float = 1e-6) -> bool:
pred_num = extract_number(prediction)
if pred_num is None:
return False
return abs(pred_num - expected) < tolerance
def extract_number(s: str) -> float | None:
m = re.search(r"-?\d+\.?\d*", s.replace(",", ""))
if not m:
return None
return float(m.group())注意 extract_number 取的是第一个数字，只适合先按上文约定的 Answer: / \boxed{} 标记截出答案片段，再交给它的场景；直接喂整段推理，就会犯上面说的"抽到 3"的错。使用前还要 import re。
容忍度 tolerance 必须根据题目设置。算术题用 1e-6，物理题（涉及测量误差）可能用 1%，金融题（涉及小数舍入）要按业务规则定。
5.3.3 单位陷阱
"10 km" vs "10000 m" 数值不同但答案相同——这种场景要在 normalize 阶段做单位换算，不能简单 numeric match。
5.4 Regex：灵活但危险
Regex 判分适合"答案符合某种模式但不固定"的场景：
def regex_match(prediction: str, pattern: str) -> bool:
return bool(re.search(pattern, prediction, re.IGNORECASE))例如评测"模型回答里必须包含日期格式"：
pattern = r"\d{4}-\d{2}-\d{2}"但 regex 判分有两个工程隐患：
flowchart TD A[Regex 判分隐患] --> B[过度宽松<br/>误判通过] A --> C[过度严格<br/>正确答案被毙] A --> D[ReDoS 攻击<br/>恶意输入卡死] B --> B1["匹配任何数字 → 答错也通过"] C --> C1["只匹配大写字母 → 错过中文回答"] D --> D1["嵌套量词 → 指数级回溯"] style D fill:#fee2e2
5.4.1 ReDoS 实例
(a+)+$ 这类有"嵌套量词"的 regex，遇到"一串 a 后跟一个 b"会触发指数级回溯：用标准库 re 匹配，每多一个 a 耗时约翻一倍，二十几个 a 就要以秒计（在 Python 3.10 上实测 26 个 a 约 6 秒），足以卡死判分进程。标准库 re 不支持超时，所以生产环境的判分器必须用第三方 regex 库 + timeout。regex 库对 (a+)+$ 这类模式本身有优化，但对 (a|aa)+$ 之类仍会回溯爆炸，timeout 不能省；超时抛出的是内置 TimeoutError：
import regex
def regex_match_safe(prediction: str, pattern: str, timeout: float = 1.0) -> bool:
try:
return bool(regex.search(pattern, prediction, regex.IGNORECASE, timeout=timeout))
except TimeoutError:
return False5.4.2 何时升级到 LLM-judge
Regex 判分会让你写出越来越复杂的模式去捕捉"语义对、字面差"的情况。当你的 regex 超过 100 字符还在演化，覆盖一堆 OR 分支时，这就是该升级到 LLM-judge 的信号。
5.5 Schema Validation：结构化输出的最强判分
如果你的应用要求 LLM 输出结构化数据（JSON / YAML / TOML），Schema Validation 是判分的最强工具。
import json
from jsonschema import validate, ValidationError
SCHEMA = {
"type": "object",
"required": ["intent", "entities"],
"properties": {
"intent": {"type": "string", "enum": ["query", "book", "cancel"]},
"entities": {
"type": "object",
"properties": {
"date": {"type": "string", "pattern": r"\d{4}-\d{2}-\d{2}"},
"city": {"type": "string"},
},
},
},
}
def schema_match(prediction: str) -> tuple[bool, str]:
try:
obj = json.loads(prediction)
validate(instance=obj, schema=SCHEMA)
return True, ""
except json.JSONDecodeError as e:
return False, f"invalid JSON: {e}"
except ValidationError as e:
return False, f"schema violation: {e.message}"5.5.1 Schema Validation 的优势
- 覆盖度全：能检查类型、范围、枚举、嵌套结构、必填字段、模式
- 错误消息友好：哪里挂了哪里说，failure case 直接可读
- 行业标准：JSONSchema / OpenAPI / Pydantic 都是成熟规范，工具生态丰富
5.5.2 真实的 Schema 设计案例：客服意图识别
意图识别是客服 chatbot 第一步路由——把用户输入分到 query / book / cancel / complaint / chitchat 几类。
设计 schema 时常见的错误是把 enum 列得太宽（"complaint, problem, issue, trouble"），导致模型选哪个都 OK，评测分高但实际质量没区分。正确做法是把语义重叠的合并、把高敏感的独立：
INTENT_SCHEMA = {
"type": "object",
"required": ["intent", "confidence", "entities"],
"properties": {
"intent": {
"type": "string",
"enum": ["query", "book", "cancel", "complaint", "chitchat"],
},
"confidence": {"type": "number", "minimum": 0, "maximum": 1},
"entities": {
"type": "object",
"properties": {
"order_id": {"type": "string", "pattern": r"^[A-Z0-9]{8,12}$"},
"date": {"type": "string", "pattern": r"^\d{4}-\d{2}-\d{2}$"},
},
"additionalProperties": False,
},
},
"additionalProperties": False,
}additionalProperties: False 这一句很关键——它强制模型不能编造 schema 没定义的字段。没有这一句，模型可能输出 {"intent": "query", "confidence": 0.9, "entities": {}, "ranking_score": 0.97}——ranking_score 不在 schema 里，但 schema validation 会通过，让你以为模型行为合规。这是评测里最隐蔽的 bug 之一。
5.5.3 与 OpenAI Structured Outputs / Anthropic Tool Use 的协同
OpenAI 的 Structured Outputs（2024-08 上线）和 Anthropic 的 Tool Use 都直接接受 JSON Schema。开启严格模式后（OpenAI 的 strict: true；Anthropic 在 2025-11 推出结构化输出，工具定义上的 strict: true 与响应上的 output_config.format），模型生成时就被约束在 schema 内。
这个能力对评测的影响很大：在严格模式下，schema validation 的通过率接近满分，它不再能区分模型好坏，只剩兜底作用。所以现代 RAG / Agent 系统的趋势是：
- 应用层强制 Structured Output
- 评测层用 Schema Validation 做"硬约束"判分
- LLM-judge 只用来评 schema 内字段的"语义质量"
第 14 章在 Agent 工具调用的失败模式里还会用到这种组合（参数格式错 → strict mode + schema validation）。
5.6 一个工业级规则判分器：完整代码
把上述所有方法整合成一个百行以内的工业级判分器：
"""rule_grader.py — 工业级规则判分器"""
import json, re, regex, unicodedata
from dataclasses import dataclass
from typing import Any
from jsonschema import validate, ValidationError
# ============= 归一化 =============
def normalize(s: str, opts: dict | None = None) -> str:
opts = opts or {}
if opts.get("nfkc", True): # NFKC 必须在 lower 之前，见 §5.11.4
s = unicodedata.normalize("NFKC", s)
if opts.get("case_insensitive", True):
s = s.lower()
if opts.get("strip_punct", False):
s = re.sub(r"[^\w\s]", "", s)
if opts.get("collapse_ws", True):
s = re.sub(r"\s+", " ", s)
return s.strip()
# ============= Exact Match =============
def grade_exact(pred: str, expected: str, opts: dict | None = None) -> dict:
ok = normalize(pred, opts) == normalize(expected, opts)
return {"ok": ok, "method": "exact_match"}
# ============= Numeric Match =============
def extract_first_number(s: str) -> float | None:
s = s.replace(",", "")
m = re.search(r"-?\d+\.?\d*", s)
return float(m.group()) if m else None
def grade_numeric(pred: str, expected: float, tol: float = 1e-6) -> dict:
n = extract_first_number(pred)
if n is None:
return {"ok": False, "method": "numeric", "reason": "no number found"}
return {"ok": abs(n - expected) < tol, "method": "numeric"}
# ============= Regex Match =============
def grade_regex(pred: str, pattern: str, timeout: float = 1.0) -> dict:
try:
ok = bool(regex.search(pattern, pred, regex.IGNORECASE, timeout=timeout))
return {"ok": ok, "method": "regex"}
except TimeoutError:
return {"ok": False, "method": "regex", "reason": "timeout (ReDoS?)"}
# ============= Contains / Keyword =============
def grade_contains(pred: str, keywords: list[str], require: str = "all") -> dict:
pred_norm = normalize(pred)
hits = [k for k in keywords if normalize(k) in pred_norm]
if require == "all":
ok = len(hits) == len(keywords)
elif require == "any":
ok = len(hits) > 0
else:
raise ValueError(f"unknown require: {require}")
return {"ok": ok, "method": "contains", "hits": hits}
# ============= Schema Validation =============
def grade_schema(pred: str, schema: dict) -> dict:
try:
obj = json.loads(pred)
except json.JSONDecodeError as e:
return {"ok": False, "method": "schema", "reason": f"invalid JSON: {e}"}
try:
validate(instance=obj, schema=schema)
return {"ok": True, "method": "schema"}
except ValidationError as e:
return {"ok": False, "method": "schema", "reason": e.message}
# ============= 多规则组合 =============
@dataclass
class GradingRule:
method: str
config: dict
def apply(self, pred: str) -> dict:
if self.method == "exact":
return grade_exact(pred, self.config["expected"], self.config.get("opts"))
elif self.method == "numeric":
return grade_numeric(pred, self.config["expected"], self.config.get("tol", 1e-6))
elif self.method == "regex":
return grade_regex(pred, self.config["pattern"])
elif self.method == "contains":
return grade_contains(pred, self.config["keywords"], self.config.get("require", "all"))
elif self.method == "schema":
return grade_schema(pred, self.config["schema"])
else:
raise ValueError(f"unknown method: {self.method}")
def grade_all(pred: str, rules: list[GradingRule]) -> dict:
results = [r.apply(pred) for r in rules]
return {"ok": all(r["ok"] for r in results), "details": results}使用示例：
rules = [
GradingRule("schema", {"schema": INTENT_SCHEMA}),
GradingRule("contains", {"keywords": ["beijing", "shanghai"], "require": "any"}),
GradingRule("regex", {"pattern": r"\d{4}-\d{2}-\d{2}"}),
]
result = grade_all(model_response, rules)
print(result["ok"], result["details"])这一份不到 100 行（去掉空白和注释）的代码，覆盖了上面五种规则判分的全部方法，并支持任意组合。第 9 章的 OpenAI evals 与第 12 章的 promptfoo 内置 grader 也是这几类判分的组合——不同框架在外壳上千差万别，但这套核心是相通的。
5.7 规则判分的设计哲学
5.7.1 规则判分的"声明式 vs 命令式"哲学
回顾本章的规则判分形态——Exact Match / Numeric Match / Regex / Schema Validation / Contains，以及 §5.8 要讲的 Levenshtein——它们能被分成两类哲学：
- 声明式（declarative）：JSON Schema、Pydantic—"输出必须长这样"，框架做校验
- 命令式（imperative）：自定义 Python 函数 / regex—"按这个步骤检查"
两者各有适用：
| 维度 | 声明式 | 命令式 |
|---|---|---|
| 表达力 | 结构化场景强 | 任意逻辑都行 |
| 可读性 | 高（schema 即文档） | 中（要读代码） |
| 可演化 | 高（改 schema 不破代码） | 中（要改代码逻辑） |
| 调试 | 框架给清晰错误 | 自己写 logging |
| 适合 | 多数标准场景 | 少数边角场景 |
工程经验：先用声明式，逼到极限再用命令式。新人 / 急于交付的工程师容易跳到 "javascript: () => ..." 这种命令式断言——一开始快，但时间一长维护成本越来越高。
声明式的极限：当业务规则需要"基于多个字段的联合判断"（如"如果 intent 是 cancel 则必须有 reason 字段"），单一 JSON Schema 不够，但可以用 Pydantic + custom validator：
from typing import Literal, Optional
from pydantic import BaseModel, model_validator
class IntentResponse(BaseModel):
intent: Literal["query", "book", "cancel"]
reason: Optional[str] = None
@model_validator(mode="after")
def cancel_must_have_reason(self):
if self.intent == "cancel" and not self.reason:
raise ValueError("cancel intent requires reason")
return self这种"声明式 + 局部命令式增强"是最优解——保持声明式的可读性，遇到极端场景才用命令式 escape。
5.7.2 规则判分的"组合 vs 单一"工程取舍
团队常面临一个工程决策——写一个大而全的判分函数，还是写多个小判分函数组合？
| 方案 | 优点 | 缺点 |
|---|---|---|
| 单一大函数 | 一处覆盖所有规则，调用简单 | 改一处影响全部、难测试 |
| 多个小函数 | 单一职责、易测试、易复用 | 调用方需要组合 |
工程经验：永远选多个小函数 + 组合层。每个小函数：
- 只测一件事（contains 关键词 / 验证 schema / 检查长度）
- 自带单元测试（5-10 行 unittest）
- 输入 / 输出明确（参考 §5.6 的
GradingRule抽象）
组合层把多个小函数串起来：
rules = [
GradingRule("schema", {"schema": INTENT_SCHEMA}),
GradingRule("contains", {"keywords": ["北京", "上海"], "require": "any"}),
GradingRule("regex", {"pattern": r"\d{4}-\d{2}-\d{2}"}),
# "不得出现违禁词"这类规则，需要先给 §5.6 的 GradingRule.apply 加一个 not-contains 分支
]
result = grade_all(model_response, rules)这种 Unix 哲学风格的规则组合（"做一件事做好、再组合"）让规则判分系统在 6-12 个月演化中仍然清晰。
5.7.3 把规则判分当作"可执行的标注 guideline"
最后一个视角，是规则判分的社会工程意义——它把"什么算合格回答"从口头共识、PPT 文档变成了可执行代码。
这一点对团队协作非常重要。考虑一个场景：PM 说"客服回答必须是友好的"，工程师问"具体指什么？" PM 说"就是别太冷淡"——这种描述无法落地。
但如果把它转成一组规则判分：
- description: 友好的客服回答
assert:
- type: not-contains-any
value: ["我不能", "无法", "不行"] # 不直接拒绝
- type: regex
value: "^(您好|你好|好的|没问题)" # 礼貌开头
- type: latency
threshold: 3000 # 不让用户等
- type: javascript
value: "output.length >= 30" # 不能太敷衍这就是把"友好"翻译成了 4 条可量化、可校验、可在 CI 里跑的规则。PM 读得懂，工程师能执行，QA 可以追溯。这种"用规则表达共识"的能力，本身就是评测体系给团队带来的最大组织红利之一——比指标曲线还重要。
5.7.4 规则判分的"工程语言学"细节
回顾 OpenAI evals 的 record_and_check_match 函数（evals/api.py:55），它的 5 个参数命名值得细读（下面是节选，省略了默认值与函数体）：
def record_and_check_match(
prompt: Any, # 原始输入
sampled: str, # 模型实际输出
expected: ..., # 期望答案
separator: ..., # 分隔符判定函数
options: ..., # 候选选项
):
...这些名字可以这样读（这是本专栏的解读，仓库里并没有说明命名动机）：
sampled而非output/generated：强调"这是从模型分布中采样得到的"，隐含统计性expected而非correct/truth：避免"绝对真理"暗示，承认评测的相对性separator：docstring 写的是"检查某个字符是否为分隔符的函数"，实现里只拿它判断紧跟在选项后面的那一个字符，并不用来切分字段options而非choices/candidates：与"多选题"语义贴合picked（返回值）而非selected/chosen：动词时态精准（用过去式表示已选）
5.7.5 规则判分的"代码可读性"细节
规则判分的代码看似简单，但长期可读性差异巨大。看两份意图相同的代码（可读版顺手给关键词补上了 strip 与大小写归一，严格说行为并不完全等价）：
不可读版本：
def g(p,e):
return p.lower().strip()==e.lower().strip() or any(k in p.lower() for k in e.split(','))可读版本：
def grade(prediction: str, expected: str) -> bool:
"""规则判分: exact match 或包含任一关键词"""
pred_normalized = prediction.lower().strip()
expected_normalized = expected.lower().strip()
keywords = [k.strip() for k in expected.split(',')]
is_exact_match = pred_normalized == expected_normalized
contains_any_keyword = any(
kw.lower() in pred_normalized for kw in keywords
)
return is_exact_match or contains_any_keyword不可读版本，工程师几个月后看自己的代码都会问"这是什么意思"；可读版本一眼就能看懂。
可读性的工程红利：
- 维护成本：改 bug 时不必先逆向理解代码
- 新人 onboarding：可读代码能少写很多文档
- 跨团队协作：可读代码避免误解
- review 效率：PR review 时可读代码审得更快
工业实务：规则判分代码的可读性优先于行数。哪怕多 5 行代码，长期收益也远超短期"省字符"。这是软件工程的老智慧在评测领域的具体应用。
5.8 匹配策略：从严格到宽松
5.8.1 规则判分的"模糊匹配 + 阈值"工程实践
精确匹配（== / in）在很多业务场景过于严苛——LLM 输出 "13.5%" 与 ideal "13.50%" 视为不同就太苛刻。下面给一份"模糊匹配"工具集，平衡严格与灵活：
import re
import unicodedata
from dataclasses import dataclass
from typing import Callable
from difflib import SequenceMatcher
@dataclass
class FuzzyMatchResult:
matched: bool
score: float
method: str
detail: str
class FuzzyRuleMatcher:
"""4 种模糊匹配策略 + threshold"""
NUMBER_RE = re.compile(r"-?\d+\.?\d*")
def numeric_close(self, expected: str, actual: str,
tolerance: float = 0.01) -> FuzzyMatchResult:
"""数值匹配：13.5% ≈ 13.50% 视为相同（容差 1%）"""
e_nums = [float(m.group()) for m in self.NUMBER_RE.finditer(expected)]
a_nums = [float(m.group()) for m in self.NUMBER_RE.finditer(actual)]
if not e_nums or not a_nums:
return FuzzyMatchResult(False, 0.0, "numeric_close",
"no numbers found")
diffs = [abs(e - a) / max(abs(e), 1e-9)
for e, a in zip(e_nums, a_nums)]
max_diff = max(diffs)
ok = max_diff <= tolerance
return FuzzyMatchResult(ok, 1 - max_diff, "numeric_close",
f"max_rel_diff={max_diff:.4f}")
def normalized_string_eq(self, expected: str,
actual: str) -> FuzzyMatchResult:
"""归一化字符串等价：去全角 / 大小写 / 空白"""
def norm(s):
s = unicodedata.normalize("NFKC", s)
s = re.sub(r"\s+", " ", s)
return s.casefold().strip()
ok = norm(expected) == norm(actual)
return FuzzyMatchResult(ok, 1.0 if ok else 0.0,
"normalized_eq", "")
def edit_distance_ratio(self, expected: str, actual: str,
threshold: float = 0.85) -> FuzzyMatchResult:
"""相似度比 ≥ threshold 视为相同（鲁棒于错别字 / OCR 噪声）。
注意 difflib 用的是 Ratcliff-Obershelp 算法，与 Levenshtein 相近但不等同"""
ratio = SequenceMatcher(None, expected, actual).ratio()
return FuzzyMatchResult(ratio >= threshold, ratio,
"edit_distance",
f"ratio={ratio:.3f}")
def keyword_set_overlap(self, expected_kws: set[str], actual: str,
min_overlap: float = 0.7) -> FuzzyMatchResult:
"""关键词集合交集占比 ≥ threshold"""
actual_words = set(re.findall(r"\w+", actual.lower()))
hit = expected_kws & actual_words
ratio = len(hit) / max(len(expected_kws), 1)
return FuzzyMatchResult(ratio >= min_overlap, ratio,
"keyword_overlap",
f"hit={len(hit)}/{len(expected_kws)}")
def composite_match(self, expected: str, actual: str,
strategies: list[str] = None) -> FuzzyMatchResult:
"""多策略 OR 组合"""
strategies = strategies or ["normalized_eq", "numeric_close",
"edit_distance"]
results = []
if "normalized_eq" in strategies:
results.append(self.normalized_string_eq(expected, actual))
if "numeric_close" in strategies:
results.append(self.numeric_close(expected, actual))
if "edit_distance" in strategies:
results.append(self.edit_distance_ratio(expected, actual))
best = max(results, key=lambda r: r.score)
return FuzzyMatchResult(any(r.matched for r in results),
best.score, "composite",
f"best={best.method}")flowchart LR E[expected] --> M[FuzzyMatcher] A[actual] --> M M --> S1[normalized_string_eq] M --> S2[numeric_close] M --> S3[edit_distance] M --> S4[keyword_overlap] S1 -. "1.0/0.0" .-> COMP[composite OR] S2 -. "1-rel_diff" .-> COMP S3 -. "ratio" .-> COMP S4 -. "overlap" .-> COMP COMP --> R[FuzzyMatchResult] style COMP fill:#e3f2fd
工程实务的 4 条策略选择经验：
| 业务场景 | 推荐策略 | 阈值 |
|---|---|---|
| 数值答案（金额、温度、百分比） | numeric_close | tolerance 0.01-0.05 |
| 文本答案（含偏差容忍） | normalized + edit_distance | ratio ≥ 0.85 |
| 答案是 keyword 列表 | keyword_set_overlap | overlap ≥ 0.7 |
| 严格相等（合同条款 / 法律） | 不要用 fuzzy，用精确 | -- |
具体例子（示意数据，需用你自己的数据复算）：
- 用户问"今日北京天气"，LLM 答 "13.5℃"，ideal "13.50摄氏度" → numeric_close ✅
- LLM 答"我不太清楚"，ideal "我不清楚" → edit_distance ratio 0.89 ✅
- LLM 答"AB123CD45"（OCR 错位），ideal "AB123-CD45" → composite OR 命中 ✅
工程实务陷阱：不要把所有评测题都用 fuzzy —— 严格场景（精算 / 法律 / 安全）必须 exact match，fuzzy 反而掩盖错误。判断标准：业务方说"用户能接受这种近似吗？"——若答案是"必须精确"，就别 fuzzy。
研究背景：FuzzyWuzzy（现更名 TheFuzz，底层改用 RapidFuzz）是常用的 fuzzy 匹配库，本节 edit_distance 思路与之相近。Levenshtein 距离公式（Levenshtein 1965）是这类匹配的方法学起点。
把这套 matcher 纳入团队规则判分基础库，能解决 LLM 评测中"过度严格 → 假 negative 多"的常见问题。它是 §5.4 正则匹配 + §5.5 schema 校验之外的"第三种规则武器"。
5.8.2 规则判分的"模糊精度阶梯"——从 100% 严格到 60% 宽松的 5 级精度光谱
规则判分一直被批评"太僵硬"——但实际上这是工程师只用了 0/1 二值匹配的结果。本节给读者一份"5 级精度阶梯"——把规则判分从严格 exact 到宽松"语义近似"分 5 档，让规则判分在保留可解释性的同时拥有 LLM-judge 的灵活性。
graph LR
A[候选答案 vs 期望] --> B[5 级精度判定]
B --> C[L0 exact 100%]
B --> D[L1 normalized 95%]
B --> E[L2 substring 85%]
B --> F[L3 fuzzy edit 70%]
B --> G[L4 keyword set 60%]
C --> H[严格匹配通过 → 1.0]
D --> I[NFKC 后等于 → 0.95]
E --> J[包含期望子串 → 0.85]
F --> K[编辑距离 ≤ 阈值 → 0.70]
G --> L[关键词都在 → 0.60]
H & I & J & K & L --> M[多档分数<br/>不再 0/1 二值]
M --> N[选最高匹配级]
5 级精度 × 适用场景：
| 级别 | 匹配方式 | 分数 | 适用场景 | 示例 |
|---|---|---|---|---|
| L0 exact | 字符级完全等 | 1.00 | 数学答案 / ID / hash | "100" vs "100 元" → fail |
| L1 normalized | NFKC + lower + trim | 0.95 | 一般文本 | "Hello" vs "hello" → 0.95 |
| L2 substring | 期望是输出的子串 | 0.85 | "包含某关键事实" | "退款是 100 元" 含期望"100" |
| L3 fuzzy edit | Levenshtein ≤ N | 0.70 | 容错少量错字 / 标点 | "苹果" vs "苹菓" |
| L4 keyword set | 关键词集合 ⊆ 输出 | 0.60 | 多关键词 / 顺序无关 | "[退款, 100]" 都在输出 |
配套实现：5 级精度阶梯判分器：
import unicodedata
from dataclasses import dataclass
from typing import Literal
PrecisionLevel = Literal["L0_exact", "L1_normalized", "L2_substring",
"L3_fuzzy_edit", "L4_keyword_set"]
@dataclass
class FuzzyPrecisionGrader:
edit_distance_threshold: int = 2
case_sensitive: bool = False
def normalize(self, s: str) -> str:
n = unicodedata.normalize("NFKC", s)
if not self.case_sensitive: n = n.lower()
return n.strip()
def levenshtein(self, a: str, b: str) -> int:
if len(a) < len(b): a, b = b, a
if len(b) == 0: return len(a)
prev = list(range(len(b) + 1))
for i, ca in enumerate(a, start=1):
curr = [i] + [0] * len(b)
for j, cb in enumerate(b, start=1):
ins, dele, sub = curr[j-1] + 1, prev[j] + 1, prev[j-1] + (ca != cb)
curr[j] = min(ins, dele, sub)
prev = curr
return prev[-1]
def grade(self, candidate: str, expected: str | list[str]) -> dict:
# L4 关键词集合（expected 为 list 时优先）
if isinstance(expected, list):
cand_norm = self.normalize(candidate)
hits = [k for k in expected if self.normalize(k) in cand_norm]
return {
"matched_level": "L4_keyword_set",
"score": 0.60 if len(hits) == len(expected) else 0.0,
"hits": hits,
}
# 单字符串：依次试 L0 → L3
if candidate == expected:
return {"matched_level": "L0_exact", "score": 1.00}
cand_n, exp_n = self.normalize(candidate), self.normalize(expected)
if cand_n == exp_n:
return {"matched_level": "L1_normalized", "score": 0.95}
if exp_n in cand_n:
return {"matched_level": "L2_substring", "score": 0.85}
dist = self.levenshtein(cand_n, exp_n)
if dist <= self.edit_distance_threshold:
return {"matched_level": "L3_fuzzy_edit", "score": 0.70,
"edit_distance": dist}
# 都不命中
return {"matched_level": None, "score": 0.0,
"best_edit_distance": dist}
def stats_over_dataset(self, results: list[dict]) -> dict:
levels = {"L0_exact": 0, "L1_normalized": 0, "L2_substring": 0,
"L3_fuzzy_edit": 0, "L4_keyword_set": 0, "fail": 0}
for r in results:
lv = r.get("matched_level") or "fail"
levels[lv] = levels.get(lv, 0) + 1
n = max(len(results), 1)
avg_score = sum(r.get("score", 0.0) for r in results) / n
return {
"total": len(results),
"by_level": levels,
"avg_score": round(avg_score, 3),
"strict_pass_pct": (levels["L0_exact"] + levels["L1_normalized"]) / n * 100,
"loose_pass_pct": sum(v for k, v in levels.items() if k != "fail") / n * 100,
}举例（示意数据，需用你自己的数据复算）：某客服 100 题：
- 之前 0/1 严格匹配：通过 42 / 失败 58 → 通过率 42%
- 改用 5 级阶梯：L0 30 / L1 12 / L2 18 / L3 25 / L4 10 / fail 5
- avg_score = 0.80，strict_pass = 42%，loose_pass = 95%
- 业务可接受 L2 及以上 → 业务通过率从 42% 升到 60%（接受 L3 及以上则为 85%）
- 不需要引入 LLM-judge 就能拿到一部分 judge 式的灵活性
- 省下的是这部分样例原本要花的 judge 调用费
配套行业研究背景：
- 分档打分的思路可对照 Lucene / Elasticsearch 的模糊查询与相关性打分
- "Levenshtein distance" 来自 Levenshtein 1965
- promptfoo 0.106.0 的
levenshtein（编辑距离阈值）与similar（embedding 相似度）断言是同类能力的现成实现
5.8.3 规则判分的"自适应 fallback 链"——规则不命中时从严到松依次降级
规则判分最大的痛点：规则太严 → 大量 false negative；规则太松 → 大量 false positive。本节给读者一份「fallback 降级链」工程方案——按"严 → 中 → 宽"3 档规则依次匹配，命中即停，让规则判分既保住严格性又提供宽松后备。
graph LR
A[候选答案] --> B[L1: 严格匹配]
B -->|命中| C[score 1.0]
B -->|不命中| D[L2: 中等宽松]
D -->|命中| E[score 0.7]
D -->|不命中| F[L3: 关键词匹配]
F -->|命中| G[score 0.4]
F -->|不命中| H[彻底 fail<br/>转 LLM-judge]
C & E & G & H --> I[fallback 链报告]
I --> J[L1 占比统计]
I --> K[L2 占比统计]
I --> L[L3 占比统计]
I --> M[fail 转 judge 比例]
3 档 fallback × 命中 score × 业务含义（各档的实际占比强依赖任务形态，必须自己统计——这个分布本身就是最有价值的诊断信号：L1 占比持续下降说明模型输出格式在漂）：
| 档位 | 匹配方式 | 命中 score | 含义 | 比例典型 |
|---|---|---|---|---|
| L1 严格 | exact / strict regex | 1.0 | 高质量答案 | 取决于任务 |
| L2 中等 | normalized + substring | 0.7 | 大致正确 | 取决于任务 |
| L3 宽松 | 关键词集合 | 0.4 | 部分正确 | 取决于任务 |
| fail | 全不匹配 | 转 judge | 需深度判 | 取决于任务 |
配套实现：fallback 链评分器：
import re
import unicodedata
from dataclasses import dataclass, field
from typing import Callable
@dataclass
class FallbackChainGrader:
strict_regex: re.Pattern | None = None
expected_keywords: list[str] = field(default_factory=list)
expected_substring: str | None = None
def normalize(self, s: str) -> str:
return unicodedata.normalize("NFKC", s).lower().strip()
def grade(self, candidate: str) -> dict:
# L1 严格
if self.strict_regex and self.strict_regex.search(candidate):
return {"level": "L1_strict", "score": 1.0,
"matched_pattern": self.strict_regex.pattern}
cand_n = self.normalize(candidate)
# L2 中等
if self.expected_substring and self.normalize(self.expected_substring) in cand_n:
return {"level": "L2_normalized_substring", "score": 0.7,
"matched": self.expected_substring}
# L3 宽松
kw_n = [self.normalize(k) for k in self.expected_keywords]
hits = [k for k in kw_n if k in cand_n]
if len(hits) == len(kw_n) and kw_n:
return {"level": "L3_keyword_set", "score": 0.4,
"hits": hits}
# fail
return {"level": "fail", "score": None,
"needs_judge": True, "partial_keyword_hits": hits}
@dataclass
class FallbackChainAggregator:
results: list[dict] = field(default_factory=list)
def record(self, r: dict):
self.results.append(r)
def summary(self) -> dict:
from collections import Counter
levels = Counter(r["level"] for r in self.results)
n = max(len(self.results), 1)
scores = [r["score"] for r in self.results if r["score"] is not None]
avg_score = sum(scores) / len(scores) if scores else 0.0
return {
"total": n,
"level_distribution": {k: round(v / n * 100, 2) for k, v in levels.items()},
"rule_pass_pct": (n - levels.get("fail", 0)) / n * 100,
"avg_score_when_passed": round(avg_score, 3),
"fail_to_judge_pct": levels.get("fail", 0) / n * 100,
"tier_balance_health": self._health_check(levels, n),
}
def _health_check(self, levels: dict, n: int) -> str:
l1 = levels.get("L1_strict", 0) / n
fail = levels.get("fail", 0) / n
if l1 > 0.8: return "L1 占比过高 — 可能规则太严"
if l1 < 0.30: return "L1 占比过低 — 严格规则可能没覆盖好"
if fail > 0.20: return "fail 占比过高 — 转 judge 太多，成本飙升"
return "tier 分布健康"
def cost_savings_vs_pure_judge(self, judge_cost_per_call: float = 0.01,
rule_cost_per_call: float = 0.0001) -> dict:
n = len(self.results)
from collections import Counter
levels = Counter(r["level"] for r in self.results)
rule_handled = n - levels.get("fail", 0)
judge_handled = levels.get("fail", 0)
baseline_cost = n * judge_cost_per_call # 全 judge
actual_cost = rule_handled * rule_cost_per_call + judge_handled * judge_cost_per_call
return {
"baseline_pure_judge_usd": round(baseline_cost, 4),
"actual_fallback_chain_usd": round(actual_cost, 4),
"savings_usd": round(baseline_cost - actual_cost, 4),
"savings_pct": round((baseline_cost - actual_cost) / baseline_cost * 100, 1),
}举例（示意数据，需用你自己的数据复算）：某团队 1000 题客服评测：
- L1 strict 600 (60%) / L2 normalized 220 / L3 keyword 90 / fail 90 (9%)
- avg_score_when_passed 0.87
- tier_balance_health = "tier 分布健康"
- cost_savings_vs_pure_judge：baseline $10 → actual $0.091 + $0.90 ≈ $0.99 → 节省约 90%
- 月跑 100 次 → 年节省约 $10.8K
配套行业研究背景：
- 级联判定可对照 ML 经典的级联分类器（Viola-Jones 2001）
- "Tiered fallback patterns" 是客服系统的通用做法（分级兜底：规则 → 模板 → 人工），不是某一家的专有设计
FallbackChainGrader 适合业务形态多样的场景。
5.8.4 规则判分的"调度器"——根据 expected 自动选 matcher
§5.8.1 与 §5.8.5 给出了 fuzzy / structural / exact 三种 matcher 的实现，但工程实务中最痛苦的是"每条 rule 都得手动选 matcher"。下面是一份调度器，根据 expected 字段的特征自动路由到合适的 matcher：
import json
import re
from dataclasses import dataclass
from typing import Any, Callable
@dataclass
class MatcherRoutingDecision:
matcher_used: str
matched: bool
score: float
reason: str
class AutoMatcherRouter:
"""根据 expected 字段的特征自动选择最合适的 matcher"""
NUMBER_RE = re.compile(r"^-?\d+\.?\d*\s*(%|℃|°|元|美元|\$)?$")
JSON_RE = re.compile(r"^\s*[\{\[]")
SQL_RE = re.compile(r"^\s*(SELECT|INSERT|UPDATE|DELETE)",
re.IGNORECASE)
URL_RE = re.compile(r"^https?://")
def __init__(self, fuzzy_matcher, structural_matcher,
exact_match_threshold: int = 30):
self.fuzzy = fuzzy_matcher
self.structural = structural_matcher
self.exact_threshold = exact_match_threshold
def _classify_expected(self, expected: str) -> str:
"""根据特征分类"""
e = expected.strip()
if self.JSON_RE.match(e):
return "json"
if self.SQL_RE.match(e):
return "sql"
if self.NUMBER_RE.match(e):
return "numeric"
if self.URL_RE.match(e):
return "url"
if "," in e and len(e.split(",")) > 2 and len(e) < 100:
return "list"
if len(e) <= self.exact_threshold:
return "short_text"
return "long_text"
def route(self, expected: str, actual: str) -> MatcherRoutingDecision:
category = self._classify_expected(expected)
if category == "json":
r = self.structural.json_equiv(expected, actual)
return MatcherRoutingDecision(
matcher_used=f"structural.{r.method}",
matched=r.matched,
score=1.0 if r.matched else 0.0,
reason=r.detail,
)
if category == "sql":
r = self.structural.sql_equiv(expected, actual)
return MatcherRoutingDecision(
matcher_used=f"structural.{r.method}",
matched=r.matched,
score=1.0 if r.matched else 0.0,
reason=r.detail,
)
if category == "numeric":
r = self.fuzzy.numeric_close(expected, actual)
return MatcherRoutingDecision(
matcher_used=f"fuzzy.{r.method}",
matched=r.matched,
score=r.score,
reason=r.detail,
)
if category == "url":
return MatcherRoutingDecision(
matcher_used="exact_url",
matched=expected.strip() == actual.strip(),
score=1.0 if expected.strip() == actual.strip() else 0.0,
reason="URL must match exactly",
)
if category == "list":
r = self.structural.list_set_equiv(
[x.strip() for x in expected.split(",")],
[x.strip() for x in actual.split(",")],
)
return MatcherRoutingDecision(
matcher_used=f"structural.{r.method}",
matched=r.matched,
score=1.0 if r.matched else 0.0,
reason=r.detail,
)
if category == "short_text":
r = self.fuzzy.normalized_string_eq(expected, actual)
return MatcherRoutingDecision(
matcher_used=f"fuzzy.{r.method}",
matched=r.matched,
score=r.score,
reason=r.detail,
)
# long_text → composite fuzzy
r = self.fuzzy.composite_match(expected, actual)
return MatcherRoutingDecision(
matcher_used=f"fuzzy.{r.method}",
matched=r.matched,
score=r.score,
reason=r.detail,
)flowchart TB
E[expected 字段] --> CL[特征分类]
CL --> Q1{"以 [ 或 { 开头?"}
Q1 -->|是| JSON[json_equiv]
CL --> Q2{以 SELECT 开头?}
Q2 -->|是| SQL[sql_equiv]
CL --> Q3{纯数字?}
Q3 -->|是| NUM[numeric_close]
CL --> Q4{以 http 开头?}
Q4 -->|是| URL[exact_url]
CL --> Q5{含 \\, 且短?}
Q5 -->|是| LIST[list_set_equiv]
CL --> Q6{≤ 30 字符?}
Q6 -->|是| SHORT[normalized_string_eq]
Q6 -->|否| LONG[composite fuzzy]
JSON --> R[MatcherRoutingDecision]
SQL --> R
NUM --> R
URL --> R
LIST --> R
SHORT --> R
LONG --> R
style R fill:#e8f5e9
工程实务的 4 条调度器价值：
- 判分写规则不再纠结：
AutoMatcherRouter(...).route(expected, actual)一行搞定多数场景 - 特征分类可扩展：业务特定（如电话号码 / 中国身份证）添加专属分类
- 保留可读性：
matcher_used字段告诉读者"为什么这条用了 fuzzy 不用 exact" - fallback 健全：未知格式默认 composite fuzzy（不会 crash）
具体例子（示意数据，需用你自己的数据复算）：评测集 1000 题分类分布：
- json：120 题 → 自动用 json_equiv
- sql：85 题 → 自动用 sql_equiv
- numeric：210 题 → fuzzy numeric_close
- url：35 题 → 严格 exact
- short_text：380 题 → normalized fuzzy
- long_text：170 题 → composite fuzzy
工程师不再为"该用哪个 matcher"思考——直接 router.route(expected, actual)。
3 类常见路由错误与修法：
| 现象 | 误诊原因 | 修法 |
|---|---|---|
| "13.5%" 被路由到 short_text | 数字带 % 没被 NUMBER_RE 捕到 | 扩 NUMBER_RE 含单位 |
| JSON 答案被当 long_text | 模型把 JSON 包在 Markdown 代码围栏里，开头不是 { |
分类前先剥掉代码围栏 |
大写 HTTPS:// 被当 short_text |
URL_RE 区分大小写 | URL_RE 加 re.IGNORECASE |
研究背景：
- 按输入特征分派处理函数，是软件工程里常见的分派表 / 策略模式
- 第 12 章会看到 promptfoo 0.106.0 用一张 type → handler 的派发表路由断言（
src/assertions/index.ts），区别在于那里由用户显式写 type，这里由 expected 的形态自动推断
把 AutoMatcherRouter 接入团队 evals 框架，能省掉大量逐条选 matcher 的重复代码。前面 §5.6-§5.8.3 都在造武器，本节给出武器调度。
5.8.5 规则判分的"语义结构"层——超越字符匹配的工程模式
字符级匹配（exact / fuzzy）解决"形似"，但很多 LLM 输出需要语义结构正确——例如 SQL 查询、JSON 数据、数学表达式。下面给出几种高频"语义结构"判分模式：
import json
import ast
import operator
import sqlparse
from dataclasses import dataclass
from typing import Any
@dataclass
class StructuralMatchResult:
method: str
matched: bool
detail: str
class StructuralRuleMatcher:
"""语义结构层判分：解析后比较，而非比字符串"""
def json_equiv(self, expected: str, actual: str) -> StructuralMatchResult:
"""JSON 等价：忽略 key 顺序、空白、引号风格"""
try:
e = json.loads(expected)
a = json.loads(actual)
return StructuralMatchResult(
"json_equiv", e == a,
f"e={type(e).__name__} a={type(a).__name__}",
)
except json.JSONDecodeError as e:
return StructuralMatchResult("json_equiv", False,
f"parse_error: {e}")
def sql_equiv(self, expected: str, actual: str) -> StructuralMatchResult:
"""SQL 等价：normalize 后比较 token 流"""
try:
e_tokens = self._sql_normalize(expected)
a_tokens = self._sql_normalize(actual)
return StructuralMatchResult(
"sql_equiv", e_tokens == a_tokens,
f"len_e={len(e_tokens)} len_a={len(a_tokens)}",
)
except Exception as exc:
return StructuralMatchResult("sql_equiv", False, str(exc))
def _sql_normalize(self, sql: str) -> list[str]:
parsed = sqlparse.parse(sql)
if not parsed:
return []
tokens = []
for tok in parsed[0].flatten():
if tok.ttype not in (sqlparse.tokens.Whitespace,
sqlparse.tokens.Newline):
tokens.append(tok.value.lower().strip())
return [t for t in tokens if t]
def python_ast_equiv(self, expected: str,
actual: str) -> StructuralMatchResult:
"""Python 表达式 AST 等价"""
try:
e_ast = ast.dump(ast.parse(expected, mode="eval"))
a_ast = ast.dump(ast.parse(actual, mode="eval"))
return StructuralMatchResult(
"python_ast_equiv", e_ast == a_ast,
f"ast_len_e={len(e_ast)} ast_len_a={len(a_ast)}",
)
except SyntaxError as e:
return StructuralMatchResult("python_ast_equiv", False,
f"syntax_error: {e}")
_BIN_OPS = {ast.Add: operator.add, ast.Sub: operator.sub,
ast.Mult: operator.mul, ast.Div: operator.truediv,
ast.Pow: operator.pow}
_UNARY_OPS = {ast.UAdd: operator.pos, ast.USub: operator.neg}
def _safe_eval(self, expr: str) -> float:
"""白名单求值：只放行数字字面量与 + - * / ** 运算，不调用 eval"""
def ev(node):
if isinstance(node, ast.Expression):
return ev(node.body)
if isinstance(node, ast.Constant) and type(node.value) in (int, float):
return node.value
if isinstance(node, ast.UnaryOp) and type(node.op) in self._UNARY_OPS:
return self._UNARY_OPS[type(node.op)](ev(node.operand))
if isinstance(node, ast.BinOp) and type(node.op) in self._BIN_OPS:
left, right = ev(node.left), ev(node.right)
if isinstance(node.op, ast.Pow) and abs(right) > 100:
raise ValueError("exponent too large") # 防 9**9**9 耗尽 CPU
return self._BIN_OPS[type(node.op)](left, right)
raise ValueError(f"disallowed syntax: {type(node).__name__}")
return ev(ast.parse(expr, mode="eval"))
def numeric_expression(self, expected: str,
actual: str) -> StructuralMatchResult:
"""数学表达式求值后数值等价"""
try:
e_val = self._safe_eval(expected)
a_val = self._safe_eval(actual)
return StructuralMatchResult(
"numeric_expr", abs(e_val - a_val) < 1e-6,
f"e={e_val} a={a_val}",
)
except Exception as e:
return StructuralMatchResult("numeric_expr", False, str(e))
def list_set_equiv(self, expected: list, actual: list,
order_matters: bool = False) -> StructuralMatchResult:
"""列表 / 集合等价（可选是否在意顺序）"""
if order_matters:
ok = expected == actual
else:
ok = sorted(expected) == sorted(actual)
return StructuralMatchResult(
"list_set_equiv", ok,
f"order_matters={order_matters}",
)flowchart LR
E[expected] --> M[StructuralMatcher]
A[actual] --> M
M --> D{output_type?}
D -->|JSON| J[parse → dict 比较]
D -->|SQL| S[sqlparse → token 流比较]
D -->|Python expr| P[ast.dump 比较]
D -->|数值表达式| N[白名单求值 + 容差比较]
D -->|list / set| L[sort + 比较]
J --> R[StructuralMatchResult]
S --> R
P --> R
N --> R
L --> R
style M fill:#e3f2fd
style R fill:#e8f5e9
工程实务的 5 类高频应用：
| 场景 | 推荐方法 | 解决的 LLM 输出抖动 |
|---|---|---|
| 模型输出 JSON | json_equiv |
字段顺序 / 空白格式 / 引号风格 |
| 模型生成 SQL | sql_equiv |
大小写 / 缩进 / 别名顺序 |
| 模型解题代码片段 | python_ast_equiv |
空白 / 冗余括号（变量名不同仍判不等） |
| 模型给数学公式 | numeric_expression |
等价写法（2 + 3 * 4 vs 14）；×、^ 需先替换成 *、** |
| 模型列举多项 | list_set_equiv |
列举顺序不重要时 |
具体例子（示意数据，需用你自己的数据复算）：
- LLM 给
{"name": "Alice", "age": 30}，ideal{"age":30,"name":"Alice"}→json_equiv✅ - LLM 给
SELECT name FROM users WHERE id = 1，idealselect name from users where id=1→sql_equiv✅ - LLM 给
2 + 3 * 4，ideal14→numeric_expression✅（都 eval 为 14） - LLM 给
[3, 1, 2]，ideal[1, 2, 3]，业务允许任意顺序 →list_set_equiv(order_matters=False)✅
工程实务的 4 个细节：
- 不要用 eval 判模型输出：
eval(..., {"__builtins__": {}}, {})不是沙箱，清空内置函数挡不住已知的逃逸手法，所以上面改用 AST 白名单求值。要执行模型生成的完整代码（如跑单元测试），必须放进独立的隔离进程或容器：无网络、只读文件系统、非 root 用户、CPU / 内存 / 墙钟时间配额，判分进程只收它的退出码和输出 - AST 比较忽略变量名时：
ast.dump默认含变量名，需要时自己写一个ast.NodeTransformer按出现顺序把变量名重命名成v0、v1……再比较 - SQL 多表 join 顺序差异：复杂 SQL 需要更专业的解析器（如 sqlglot），sqlparse 在简单 case 够用
- list 等价 + 元素 fuzzy：复合场景（list 内含字符串）要 list 长度等 + 逐元素 fuzzy
研究背景：
- HumanEval（Chen et al. arXiv:2107.03374）评测 Python 代码靠执行单元测试判功能正确性（pass@k），不做 AST 比较；论文同样强调要在沙箱里执行模型生成的代码
- Spider 用组件级 / exact set match 与执行结果比对（execution accuracy），WikiSQL 用逻辑形式匹配与执行结果比对
- MATH 原论文对
\boxed{}里的答案做归一化后精确匹配；后来的 Minerva 等工作改用 SymPy 判符号等价，lm-eval v0.4.5 的lm_eval/tasks/minerva_math/utils.py就用了 SymPy
StructuralRuleMatcher 可与 §5.8.1 FuzzyMatcher 拼接，按问题类型选 matcher：
- 文本类 → fuzzy
- 结构化 → structural
- 完全严格 → exact
5.8.6 常被忽略的前置：输出归一化
判分前必须做的一步——归一化模型输出。这一步决定了后续所有规则判分的可靠性。
LLM 输出常见的非确定性变体：
预期: "巴黎"
模型输出 1: "巴黎"
模型输出 2: "巴黎。"
模型输出 3: " 巴黎"
模型输出 4: "答: 巴黎"
模型输出 5: "**巴黎**"
模型输出 6: "巴黎(France)"不归一化的话，6 种输出在 Exact Match 下 5 种都判错。归一化层（normalize 函数）的标准操作：
def normalize_for_grading(text: str) -> str:
text = text.strip() # 去前后空白
text = re.sub(r'\*+', '', text) # 去 markdown 强调
text = re.sub(r'^(答[:：]?\s*|Answer[:：]?\s*)', '', text) # 去前缀
text = re.sub(r'[。\.\s]+$', '', text) # 去末尾标点
text = re.sub(r'\s*\([^)]*\)\s*$', '', text) # 去末尾括注
return text这个归一化层（用前记得 import re）是规则判分的"地基"——所有后续 Exact Match / Regex / Numeric 都基于归一化后的字符串。工业团队的做法是把它做成共享库（如 evals_normalize），所有 grader 必经过它。这避免了"30 个 grader 各自实现归一化、各自有 bug"的工程债。
5.9 与生成侧协同：Schema-First
5.9.1 Schema-First 应用设计：让评测从代码诞生时开始
进阶团队的趋势是把评测前置到应用设计阶段——Schema-First Design。
传统流程：
PM 定需求 → 工程师写 prompt → 评测员事后写测试集Schema-First：
PM 定 schema → schema 同时定义应用契约 + 评测断言 → prompt 在 schema 约束下设计举个具体例子。客服意图识别的 schema：
class IntentResponse(BaseModel):
intent: Literal["query", "book", "cancel", "complaint"]
confidence: float = Field(ge=0.0, le=1.0)
entities: dict[str, str]
fallback_to_human: bool = False这个 schema 同时承担三个角色：
- 应用层契约：OpenAI Structured Outputs / Anthropic Tool Use 用它强制约束 LLM 输出
- 评测层断言：JSON Schema 校验 + Pydantic 校验天然作为规则判分
- 类型层文档：上下游服务读这个 schema 就知道接口
这种"schema 驱动一切"的范式能显著压缩判分器的工程投入——因为每个新增字段的校验都是免费来的。OpenAI Structured Outputs（2024-08）与 Anthropic 结构化输出（2025-11）都是同一思路的产品落地。
5.9.2 规则判分与 Schema-Driven Generation 的协同
OpenAI Structured Outputs（2024-08）和 Anthropic 结构化输出（2025-11，含 strict: true 工具定义）都能让模型输出强制符合 JSON Schema。这种"上游约束 + 下游校验"的模式深刻改变了规则判分的分工。
传统流程：
模型输出自由文本 → 后处理解析 JSON → 失败时复杂错误处理 → schema validationStructured Generation 流程：
定义 Pydantic 类 → 模型直接输出符合 class 的 JSON → 直接 isinstance 校验代码对比：
# 传统
def grade_traditional(output: str, schema: dict) -> bool:
try:
obj = json.loads(output)
jsonschema.validate(obj, schema)
return True
except Exception:
return False
# Structured Output
from typing import Literal
from pydantic import BaseModel
class IntentResponse(BaseModel):
intent: Literal["query", "book", "cancel"]
confidence: float
def grade_structured(response: IntentResponse) -> bool:
return True # 已经类型安全, 不需要 validation
# 业务逻辑 grade 直接读 response.intent / response.confidence工程意义：Structured Output 把 schema validation 从"运行时检查"提前到"生成时约束"——大部分 schema 不一致问题不再发生。规则判分从"防御性"转向"业务性"——只判断业务逻辑（如"intent 是否符合用户意图"），不判断格式。
这种范式转变让规则判分的工程负担显著下降。工业团队的实操：
- 第一线：所有结构化输出都用 Structured Generation
- 第二线：规则判分只做业务断言（contains 关键词、numeric range 等）
- 第三线：复杂语义用 LLM-judge
三层分工让规则判分能覆盖的比例明显提高——这是本章方法学在新工具加持下的延伸。
5.9.3 完整的 Pydantic + OpenAI Structured Outputs 集成
整合本章 §5.9.1 Schema-First Design 方法学，给一份"Pydantic + OpenAI Structured Outputs"的完整工程示例：
# schema_first_eval.py
from pydantic import BaseModel, ConfigDict, Field
from typing import Literal
from openai import OpenAI
class Entity(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str
value: str
class IntentResponse(BaseModel):
"""客服意图识别响应 schema"""
model_config = ConfigDict(extra="forbid") # additionalProperties: false
intent: Literal["query", "book", "cancel", "complaint", "chitchat"]
confidence: float = Field(ge=0, le=1)
# 严格模式要求每个对象都是 additionalProperties: false，
# dict[str, str] 这种任意键映射过不了，所以改成键值对列表
entities: list[Entity]
fallback_to_human: bool
client = OpenAI()
# 填一个支持 Structured Outputs 的现行型号（以 OpenAI 官方模型页与弃用页为准）
OPENAI_MODEL = "<your-openai-model-id>"
def classify_intent(user_query: str) -> IntentResponse:
"""OpenAI Structured Outputs - schema 在生成时强制"""
completion = client.chat.completions.parse( # 旧版 SDK 在 client.beta.chat.completions.parse
model=OPENAI_MODEL,
messages=[
{"role": "system", "content": "Classify customer support intent."},
{"role": "user", "content": user_query},
],
response_format=IntentResponse,
)
return completion.choices[0].message.parsed
def grade_intent(prediction: IntentResponse, expected: dict) -> dict:
"""评测：schema 已被强制，只判业务逻辑"""
failures = []
if prediction.intent != expected["intent"]:
failures.append(f"intent mismatch: {prediction.intent} vs {expected['intent']}")
if prediction.confidence < 0.7:
failures.append(f"low confidence: {prediction.confidence}")
got = {e.name: e.value for e in prediction.entities}
for k, v in expected.get("entities", {}).items():
if got.get(k) != v:
failures.append(f"entity {k}: {got.get(k)} vs {v}")
return {"passed": len(failures) == 0, "failures": failures}
# 使用：评测代码不再需要写 schema validation
result = classify_intent("我要退货")
# IntentResponse(intent='query', confidence=0.95, entities=[...], fallback_to_human=False)
grade = grade_intent(result, expected={"intent": "query", "entities": {"action": "refund"}})约 40 行代码展示了 Schema-First 范式的完整工程价值：
- Pydantic class 定义"应用契约 + 评测断言"
extra="forbid"等价于additionalProperties: false（防模型编造字段）- OpenAI Structured Outputs 强制生成符合 schema
- 评测代码只判业务逻辑（intent / confidence / entities），不需要再做 schema validation
工业实务：把这种"schema 先行 + 应用层 + 评测层共享"的范式作为团队的工程标准，"格式校验失败"类问题会明显减少——因为它们在生成阶段就被拦在源头。
5.9.4 Hybrid Grading：规则与 LLM-judge 的协同流水线
实务中很少"纯规则"或"纯 LLM-judge"，Hybrid Grading才是工业标配。一个典型的 hybrid grading pipeline：
flowchart LR Output[模型输出] --> Layer1[Layer 1: 规则判分<br/>schema / regex / contains] Layer1 -->|失败| F1[标记失败<br/>不进入 LLM-judge] Layer1 -->|通过| Layer2[Layer 2: LLM-judge<br/>语义评估] Layer2 -->|高置信| Final[最终分数] Layer2 -->|低置信| Layer3[Layer 3: 人工抽查] Layer3 --> Final style Layer1 fill:#dbeafe style Layer2 fill:#dcfce7 style Layer3 fill:#fef3c7
三层各自的成本和能力（单价是示意量级，以自己的实际账单为准）：
| 层 | 单条成本 | 速度 | 能力 |
|---|---|---|---|
| 规则判分 | 0.0001 元 | 毫秒 | 格式 / 关键词 / 范围 |
| LLM-judge | 0.01 元 | 秒级 | 语义 / 事实 / 风格 |
| 人工 | 5-50 元 | 分钟级 | 极度细致 / 创意 / 合规 |
成本差距 100x → 10000x。示意地说，如果大部分样例在 Layer 1 就被解决，一小部分进 Layer 2，只有百分之几进 Layer 3，整体成本会远低于全部走 LLM-judge 或全部人工；但要注意，平均成本并不接近 Layer 1——哪怕只有 5% 进人工，按上表单价算，人工这一层就占了账单大头（§5.14.4 会再算一遍）。
工程实务上的优化：
- 早期失败 fail-fast：Layer 1 失败的 case 不再调 Layer 2 浪费 token
- Layer 2 高置信 short-circuit：LLM-judge 给 0.95+ 或 0.05- 时不需 Layer 3 介入
- Layer 3 抽样：不是所有 Layer 2 都送 Layer 3，按 1-5% 抽样保证成本可控
这种"分层判分"是评测体系工程成熟的标志——不是哪个 grader 更好，而是组合用得更聪明。
5.9.5 跨章节的关联：规则 → judge → 元评测的递进
回顾本专栏第 5-8 章的整体结构，能看到一条清晰的递进：
- 规则判分（第 5 章）：能解决就解决（通常是大头）
- LLM-judge（第 6 章）：规则解决不了的
- 人工（第 7 章）：LLM-judge 解决不了的（应压到少数）
- 元评测（第 8 章）：以上三者自身的可靠性
这条递进让读者理解评测方法学的"分工"——每种方法各有适用场景，组合起来覆盖完整问题空间。
工业实务：搭评测体系时按这个顺序投入：
- 先 100% 投入规则判分（成本最低、效率最高）
- 规则判分到天花板后投入 LLM-judge
- LLM-judge 不够可靠时投入人工
- 三者都跑起来后投入元评测验证
这种"按效率倒序"的投入路径，让团队的工程时间花在"刀刃上"——而不是一上来就上最复杂的方法。
5.10 判分器的工程化
5.10.1 生产级规则判分类的完整实现
整合本章方法学，给一份"工业级规则判分类"的完整 Python 实现：
# rule_grader.py
import re, json, unicodedata
from dataclasses import dataclass
from typing import Any, Callable
from jsonschema import validate, ValidationError
class RuleGrader:
"""生产级规则判分器 - 包含归一化、缓存、诊断、覆盖度评估"""
def __init__(self):
self._cache = {}
self._stats = {"hits": 0, "misses": 0, "uncovered": 0}
def normalize(self, text: str, opts: dict = None) -> str:
"""归一化（参见 §5.2.2、§5.8.6）"""
opts = opts or {}
if opts.get("nfkc", True):
text = unicodedata.normalize("NFKC", text)
if opts.get("case_insensitive", True):
text = text.lower()
if opts.get("strip_punct", False):
text = re.sub(r"[^\w\s]", "", text)
if opts.get("collapse_ws", True):
text = re.sub(r"\s+", " ", text)
return text.strip()
def grade(self, pred: str, rules: list[dict]) -> dict:
"""运行多个 rule 并聚合结果"""
cache_key = (pred, json.dumps(rules, sort_keys=True))
if cache_key in self._cache:
self._stats["hits"] += 1
return self._cache[cache_key]
self._stats["misses"] += 1
diagnostics = []
passed = True
for r in rules:
ok, reason = self._apply_rule(pred, r)
if not ok:
passed = False
diagnostics.append(reason)
result = {
"passed": passed,
"diagnostics": diagnostics,
"applicable_rules": len(rules),
}
self._cache[cache_key] = result
return result
def _apply_rule(self, pred: str, rule: dict) -> tuple[bool, str]:
rule_type = rule["type"]
pred_norm = self.normalize(pred)
if rule_type == "exact":
ok = pred_norm == self.normalize(rule["expected"])
return ok, "" if ok else f"exact_match failed"
elif rule_type == "contains":
keywords = rule["keywords"]
require = rule.get("require", "all")
hits = [k for k in keywords if self.normalize(k) in pred_norm]
if require == "all":
ok = len(hits) == len(keywords)
return ok, "" if ok else f"missing: {set(keywords) - set(hits)}"
else:
ok = len(hits) > 0
return ok, "" if ok else "no keyword hit"
elif rule_type == "regex":
ok = bool(re.search(rule["pattern"], pred, re.IGNORECASE))
return ok, "" if ok else f"regex not matched: {rule['pattern']}"
elif rule_type == "schema":
try:
obj = json.loads(pred)
validate(obj, rule["schema"])
return True, ""
except json.JSONDecodeError as e:
return False, f"invalid JSON: {e}"
except ValidationError as e:
return False, f"schema violation: {e.message}"
else:
self._stats["uncovered"] += 1
return False, f"unknown rule_type: {rule_type}"
def coverage_report(self) -> dict:
"""覆盖度报告（参见 §5.12.1）"""
total = self._stats["hits"] + self._stats["misses"]
return {
"cache_hit_rate": self._stats["hits"] / total if total else 0,
"uncovered_rate": self._stats["uncovered"] / total if total else 0,
**self._stats,
}约 80 行代码涵盖本章的主要方法：
- 归一化（NFKC / case / 标点 / 空白）
- 多种 rule 类型（exact / contains / regex / schema）
- 缓存（避免重复判分）
- 诊断信息（每条规则失败时具体原因）
- 覆盖度报告
工业实务：把这份代码作为团队规则判分库的"基础类"。后续业务专属判分逻辑都继承 / 组合此类。这是规则判分体系工程化的标准基础设施。
5.10.2 "规则判分库"的最佳布局——所有评测代码该怎么组织
读完 §5.6 + §5.7 的所有规则判分工具，最后一个工程问题是"这些代码该怎么放？"——下面是工业团队最普遍的目录结构：
evals/
├── pyproject.toml # 包定义
├── README.md # quickstart
│
├── core/ # 核心抽象
│ ├── __init__.py
│ ├── matchers/ # §5.8 的所有 matcher
│ │ ├── exact.py
│ │ ├── fuzzy.py # §5.8.1
│ │ ├── structural.py # §5.8.5
│ │ ├── i18n.py # §5.13.1
│ │ └── router.py # §5.8.4 自动调度
│ ├── rules/ # 业务规则集
│ │ ├── __init__.py
│ │ ├── customer_service/
│ │ ├── medical/
│ │ └── ...
│ ├── coverage_analyzer.py # §5.12.2
│ └── debt_analyzer.py # §5.12.7
│
├── datasets/ # 评测集（§3）
│ ├── golden/
│ ├── adversarial/
│ └── regression/
│
├── runners/ # 跑评测的入口
│ ├── cli.py
│ ├── ci.py
│ └── distributed.py
│
├── scripts/ # 维护工具
│ ├── monthly_audit.py
│ ├── flaky_detection.py
│ └── coverage_report.py
│
├── tests/ # 评测代码自身的单测
│ ├── test_matchers/
│ ├── test_rules/
│ └── test_runners/
│
└── docs/
├── architecture.md
├── onboarding.md
└── runbooks/from pathlib import Path
from dataclasses import dataclass
@dataclass
class CodebaseHealthReport:
total_lines: int
test_to_code_ratio: float
public_apis_count: int
breaking_changes_30d: int
cyclic_dependencies: list[str]
health_grade: str
class EvalsCodebaseHealthChecker:
"""评测代码库的整体健康度检查"""
def assess(self, root: Path) -> CodebaseHealthReport:
code_files = list(root.glob("core/**/*.py")) + \
list(root.glob("runners/**/*.py"))
test_files = list(root.glob("tests/**/*.py"))
total_lines = sum(len(f.read_text().splitlines())
for f in code_files)
test_lines = sum(len(f.read_text().splitlines())
for f in test_files)
ratio = test_lines / max(total_lines, 1)
# 数 public API（exported 在 __init__.py）
public_apis = 0
for init_file in root.glob("core/**/__init__.py"):
text = init_file.read_text()
public_apis += text.count("from .") + text.count("import ")
# 简化：breaking changes 数据来自 git tag
breaking_30d = 0 # 实际从 CHANGELOG 读
# 简化：cyclic dep 检测
cyclic = [] # 实际用 importlab / pylint 检测
if ratio >= 1.0 and not cyclic:
grade = "A"
elif ratio >= 0.7:
grade = "B"
elif ratio >= 0.4:
grade = "C"
else:
grade = "D"
return CodebaseHealthReport(
total_lines=total_lines,
test_to_code_ratio=round(ratio, 2),
public_apis_count=public_apis,
breaking_changes_30d=breaking_30d,
cyclic_dependencies=cyclic,
health_grade=grade,
)flowchart LR R[evals/ root] --> C[core/] R --> D[datasets/] R --> RU[runners/] R --> S[scripts/] R --> T[tests/] R --> DOC[docs/] C --> M[matchers] C --> RL[rules] C --> ANL[analyzers] M --> RT[router 调度] RT --> RU T -. "test_to_code ≥ 1.0" .-> C S -. "monthly cron" .-> ANL style C fill:#e3f2fd style T fill:#e8f5e9
工程实务的 5 条目录设计原则：
- core/ 与 runners/ 分开：matcher 不依赖 CLI，让 core 可被 import 到任何环境
- rules/ 按业务域分：customer_service / medical / legal 各自一目录
- scripts/ 是 cron 入口：让运维一眼看到"该跑什么"
- test_to_code_ratio ≥ 1.0：评测代码的单测比例必须高于普通代码
- docs/runbooks/ 写关键操作：如何回滚、如何 retire 旧 case 等
3 类常见反模式：
| 反模式 | 现象 | 修法 |
|---|---|---|
| 巨石 evals.py | 5000 行单文件 | 按上面的目录拆分 |
| rules / runners 混 | rules 直接调 CLI | 严格分层 |
| 没单测 | matchers 无 unit test | 每个 matcher ≥ 5 题单测 |
下面这组数字是示意性的演化轨迹（不是某个真实仓库的统计），要看的是三条曲线的相对走向——代码量增长趋缓、测试比例持续上升、文件数随模块拆分增加：
| 时点 | LOC | test ratio | 文件数 | grade |
|---|---|---|---|---|
| M0 | 200 | 0.0 | 1 | D |
| M3 | 800 | 0.3 | 5 | D |
| M6 | 1500 | 0.6 | 12 | C |
| M12 | 2800 | 0.9 | 30 | B |
| M18 | 3500 | 1.2 | 45 | A |
洞察：真正的判据不是 LOC 或文件数，而是 test ratio 有没有跟上代码量——判分器自己没有测试，就是"用没测过的尺子去量别人"。
研究背景：
- 这套布局是 Python 项目常见分层习惯的应用，并非出自某个特定项目
- 英国 AISI 开源的 Inspect（https://inspect.aisi.org.uk/）是可对照的开源评测框架实现
- Python Packaging User Guide 对 src layout 与 flat layout 的讨论，可作为包结构取舍的参考
5.10.3 规则判分的"复合 matcher 编排引擎"——把 5 种 matcher 组合成 declarative pipeline
随着评测集长大，单一 matcher 不够：一道题可能要求"答案必须包含数字 + 必须是 valid JSON + 必须不出现敏感词 + 必须长度 < 200"。如果用 if-else 硬编码，4 个条件 = 4 段代码，10 道题就有 40 段重复。本节给读者一份"复合 matcher 编排引擎"——用 declarative DSL 表达组合规则，让规则判分从"代码堆叠"升级为"配置驱动"。
graph LR
A[一道题] --> B[规则定义 YAML]
B --> C{匹配引擎}
C --> D[matcher 1: contains_number]
C --> E[matcher 2: is_json]
C --> F[matcher 3: blocklist_safe]
C --> G[matcher 4: length_under]
D & E & F & G --> H{聚合策略}
H --> I[all_must_pass<br/>AND]
H --> J[any_must_pass<br/>OR]
H --> K[score_weighted<br/>加权]
I & J & K --> L[最终判分]
L --> M[score + 失败 matcher 列表]
5 类基础 matcher + 4 种聚合策略 = 20 种组合能力：
| 基础 matcher | 含义 | 配置项 | 典型用途 |
|---|---|---|---|
contains |
子串匹配 | needles[], case_sensitive | 关键词必现 |
regex |
正则匹配 | pattern, flags | 格式约束 |
is_json |
JSON 合法 | required_keys（可选） | 结构化输出 |
length |
长度区间 | min, max | 控制冗余 |
blocklist |
黑名单 | terms[] | 安全 / 合规 |
| 聚合策略 | 语义 | 适用场景 |
|---|---|---|
| all | 所有 matcher 必过（AND） | 严格质量门 |
| any | 任一 matcher 过即可（OR） | 多种合理答案 |
| weighted | 各 matcher 加权求和 | 业务侧重不同 |
| at_least_n | 至少 N 个过 | 容错场景 |
配套实现：复合 matcher 编排引擎：
import json
import re
from dataclasses import dataclass, field
from typing import Literal, Callable, Any
MatcherKind = Literal["contains", "regex", "is_json", "length", "blocklist"]
AggregateStrategy = Literal["all", "any", "weighted", "at_least_n"]
@dataclass
class MatcherSpec:
kind: MatcherKind
config: dict
weight: float = 1.0
def evaluate(self, output: str) -> dict:
if self.kind == "contains":
needles = self.config["needles"]
case = self.config.get("case_sensitive", False)
haystack = output if case else output.lower()
needles_lower = needles if case else [n.lower() for n in needles]
hits = [n for n in needles_lower if n in haystack]
passed = len(hits) == len(needles)
return {"passed": passed, "hit_count": len(hits)}
if self.kind == "regex":
pattern = re.compile(self.config["pattern"], self.config.get("flags", 0))
m = pattern.search(output)
return {"passed": bool(m), "match": m.group(0) if m else None}
if self.kind == "is_json":
try:
obj = json.loads(output)
# 可选 schema 校验
if "required_keys" in self.config:
missing = set(self.config["required_keys"]) - set(obj.keys())
return {"passed": not missing, "missing_keys": list(missing)}
return {"passed": True}
except json.JSONDecodeError as e:
return {"passed": False, "error": str(e)}
if self.kind == "length":
length = len(output)
min_l = self.config.get("min", 0)
max_l = self.config.get("max", 1_000_000)
return {"passed": min_l <= length <= max_l, "length": length}
if self.kind == "blocklist":
terms = [t.lower() for t in self.config["terms"]]
output_lower = output.lower()
hits = [t for t in terms if t in output_lower]
return {"passed": len(hits) == 0, "blocklist_hits": hits}
@dataclass
class CompositeMatcherPipeline:
matchers: list[MatcherSpec]
strategy: AggregateStrategy = "all"
at_least_n: int = 1
def evaluate(self, output: str) -> dict:
results = [{"matcher": m.kind, **m.evaluate(output), "weight": m.weight}
for m in self.matchers]
passed_count = sum(r["passed"] for r in results)
total = len(results)
if self.strategy == "all":
final = passed_count == total
elif self.strategy == "any":
final = passed_count >= 1
elif self.strategy == "at_least_n":
final = passed_count >= self.at_least_n
elif self.strategy == "weighted":
total_weight = sum(r["weight"] for r in results)
passed_weight = sum(r["weight"] for r in results if r["passed"])
final = (passed_weight / max(total_weight, 0.01)) >= 0.6
return {
"final_pass": final,
"passed_count": passed_count,
"total_matchers": total,
"details": results,
"failed_matchers": [r["matcher"] for r in results if not r["passed"]],
}
@classmethod
def from_yaml(cls, yaml_dict: dict) -> "CompositeMatcherPipeline":
matchers = [MatcherSpec(kind=m["kind"], config=m.get("config", {}),
weight=m.get("weight", 1.0))
for m in yaml_dict["matchers"]]
return cls(matchers=matchers,
strategy=yaml_dict.get("strategy", "all"),
at_least_n=yaml_dict.get("at_least_n", 1))举例（示意数据，需用你自己的数据复算）：电商客服题目 yaml：
matchers:
- kind: contains
config: {needles: ["订单号", "退款"]}
weight: 2
- kind: is_json
config: {required_keys: ["order_id", "amount"]}
weight: 3
- kind: blocklist
config: {terms: ["保证", "100% 退款", "立即退"]}
weight: 5
- kind: length
config: {min: 50, max: 500}
weight: 1
strategy: weighted跑 200 题评测：blocklist 失败 5 道题（业务必修）/ length 失败 30 题（输出过长）/ is_json 失败 12 题。final_pass 综合判定一次跑完，全部失败 case 按 matcher 自动归类，省掉"4 个独立 matcher 各跑一次再人工对齐"的工夫。
配套行业研究背景：
- promptfoo 0.106.0 的
assert-set（src/types/index.ts里的AssertionSetSchema）把一组断言打包、可设weight与threshold，是同类的组合写法 - 用 YAML 声明评测配置，可对照 lm-eval-harness 的 task yaml（第 10 章）
CompositeMatcherPipeline 把规则写成 yaml 配置，PM / QA 也能参与维护。
5.10.4 规则判分的"3 层缓存"工程优化
最后讨论一个工程性能话题——规则判分的缓存优化。
规则判分本身极快（毫秒级），但在大规模评测中（10000+ 样例）仍可优化：
Layer 1: 进程内缓存
- LRU cache: 同一 (input, expected) 跑过即缓存
- 命中率: 80-90% (重复样例多)
Layer 2: 持久化缓存（Redis / SQLite）
- 跨进程 / 跨运行复用
- 命中率: 60-70%
Layer 3: hash-based 短路
- 输出 hash 不变 → 不重跑判分
- 命中率: 40-50%按上面三层的命中率（示意数据，需用你自己的数据复算）估算，3 层缓存组合能把万级样例规则判分的总耗时压掉一个量级。对 CI 加速明显——开发者每改 prompt 后跑评测不必再等一整轮全量判分。
工程意义：规则判分本身够快，但配合好的缓存能再快一个量级。这种"工程性能优化"虽然不显眼，但累积下来对开发者体验影响巨大——评测越快、用得越频繁，评测体系真正发挥作用。
5.10.5 "规则判分性能优化"工程实战——让 1000 题评测在 1 秒内跑完
规则判分的核心承诺是"快"——但代码写不好仍可能 1000 题跑 30 秒。下面给出工程化的性能优化武器：
import re
import functools
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Iterable, Callable
@dataclass
class PerfReport:
rule_count: int
sample_count: int
total_time_ms: float
avg_time_per_sample_us: float
p99_time_per_sample_us: float
bottleneck_rule: str
class HighPerfRuleRunner:
"""批量规则判分的工程化加速"""
def __init__(self, rules: list[Callable]):
self.rules = rules
# 预编译正则
self._compile_patterns()
# 线程池只对 IO bound 部分有效：纯 Python 的规则判分是 CPU 密集，
# 受 GIL 限制基本不提速，CPU 密集要换多进程（见 §5.10.6）
self.executor = ThreadPoolExecutor(max_workers=8)
def _compile_patterns(self):
"""预编译所有正则——省掉每次调用时查缓存 / 重编译的开销"""
for rule in self.rules:
if hasattr(rule, "_pattern_str"):
rule._compiled = re.compile(rule._pattern_str)
@functools.lru_cache(maxsize=10000)
def _normalize_text(self, text: str) -> str:
"""常用文本归一化做缓存"""
import unicodedata
return unicodedata.normalize("NFKC", text).casefold().strip()
def run_batch(self, samples: list[dict]) -> PerfReport:
import time
start = time.perf_counter()
per_sample_times = []
rule_total_times = {r.__name__: 0 for r in self.rules}
for sample in samples:
t0 = time.perf_counter()
normalized = self._normalize_text(sample["input"])
for rule in self.rules:
rt0 = time.perf_counter()
rule(sample, normalized)
rule_total_times[rule.__name__] += time.perf_counter() - rt0
per_sample_times.append((time.perf_counter() - t0) * 1_000_000)
total = (time.perf_counter() - start) * 1000
bottleneck = max(rule_total_times.items(), key=lambda x: x[1])[0]
return PerfReport(
rule_count=len(self.rules),
sample_count=len(samples),
total_time_ms=round(total, 2),
avg_time_per_sample_us=round(sum(per_sample_times) /
max(len(per_sample_times), 1), 1),
p99_time_per_sample_us=round(
sorted(per_sample_times)[int(0.99 * len(per_sample_times))],
1),
bottleneck_rule=bottleneck,
)
def parallel_run(self, samples: list[dict],
chunk_size: int = 100) -> PerfReport:
"""大批量用线程池并行"""
import time
start = time.perf_counter()
chunks = [samples[i:i+chunk_size]
for i in range(0, len(samples), chunk_size)]
list(self.executor.map(self.run_batch, chunks))
total = (time.perf_counter() - start) * 1000
return PerfReport(
rule_count=len(self.rules),
sample_count=len(samples),
total_time_ms=round(total, 2),
avg_time_per_sample_us=round(total * 1000 / max(len(samples), 1), 1),
p99_time_per_sample_us=0, # 简化
bottleneck_rule="parallel-mode",
)flowchart LR
S[1000 题] --> N[normalize 缓存]
N --> P[预编译正则]
P --> R{run 模式?}
R -->|< 500 题| SEQ[串行]
R -->|≥ 500 题| PAR[并行<br/>CPU 密集用多进程]
SEQ --> M[per-rule 耗时]
PAR --> M
M --> BN[bottleneck 识别]
BN --> OPT[优化最慢 rule]
style PAR fill:#e3f2fd
style OPT fill:#e8f5e9
工程实务的 5 条性能优化原则：
- 正则必预编译：
re.compile后复用，省掉re.match(pattern, text)每次查内部缓存的开销（缓存被挤掉时还要重编译） - lru_cache 用在 normalize：同样输入避免重复
- 不要每条 rule 重复 normalize：共享一份 normalized text
- 并行只在 ≥ 500 题时用：小批量串行更快（避免调度开销）；CPU 密集的规则要用多进程，线程池受 GIL 限制
- 找 bottleneck 优化 1 条：80/20 法则，优化最慢 rule 即可大幅提速
下面是一组示意数字（不是基准测试结果，绝对值取决于机器、规则条数与文本长度），要看的是四步优化各自的性质：
| 优化前后 | 总耗时 | 备注 |
|---|---|---|
| 原始 + 每次 normalize | 12 秒 | 每条 rule 都 normalize |
| + 预编译正则 | 6 秒 | 减半 |
| + lru_cache normalize | 2.5 秒 | 减一半 |
| + 多进程并行 | 0.6 秒 | 4x 提升 |
四步里前三步（预编译、缓存归一化、避免重复工作）省的是常数因子，最后一步并行省的是墙钟时间但不省 CPU——而且只有多进程才省得下来，纯 Python 的 CPU 密集代码用线程池基本不提速。顺序不能反：先并行再优化单条规则，等于把浪费也并行了。真实收益请用 time.perf_counter 在自己的规则集上测——规则判分本来就在毫秒量级，多数团队根本不需要走到第四步。
3 类常见性能陷阱：
| 陷阱 | 现象 | 修法 |
|---|---|---|
| re.match 每次重新编译 | 大批量下明显变慢（Python 有内置正则缓存，但缓存被挤掉后会退化） | 预编译 re.compile 并复用 |
| 字符串拼接 in loop | O(n²) | 用 "".join() |
| 不必要的 deepcopy | 内存爆 + 慢 | 只 copy 需要修改的 |
研究背景：
- Python 性能优化经典《High Performance Python》(O'Reilly 2020)
- pypy + cython 是更激进的加速方案
- pytest 的 fixture caching 思路类似
HighPerfRuleRunner 的 per-rule 耗时可以用来找出最慢的那条规则。
5.10.6 规则判分的"可扩展性"工程模式——百万 sample 怎么跑
§5.10.5 把 1000 题压到 1 秒——但工业团队有时需要扫一周生产 trace（约 1M sample）。下面给出更激进的可扩展性方案：
import asyncio
import multiprocessing as mp
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable, Callable
@dataclass
class ScaleEvalReport:
sample_count: int
total_time_seconds: float
samples_per_second: float
cpu_count_used: int
memory_peak_mb: float
class MillionSampleRuleRunner:
"""跑百万级 sample 评测"""
def __init__(self, rules: list[Callable],
worker_count: int = None):
self.rules = rules
self.workers = worker_count or mp.cpu_count()
@staticmethod
def _process_chunk(args):
chunk, rules_list = args
results = []
for sample in chunk:
sample_results = {}
for rule in rules_list:
sample_results[rule.__name__] = rule(sample)
results.append(sample_results)
return results
def run_million(self, samples: Iterable[dict],
chunk_size: int = 10000) -> ScaleEvalReport:
import time, resource, sys
start = time.perf_counter()
# 1. 分 chunk
chunks = []
current = []
for s in samples:
current.append(s)
if len(current) >= chunk_size:
chunks.append((current, self.rules))
current = []
if current:
chunks.append((current, self.rules))
# 2. 多进程并行
with mp.Pool(self.workers) as pool:
chunk_results = pool.map(self._process_chunk, chunks)
# 3. 汇总
total = sum(len(r) for r in chunk_results)
elapsed = time.perf_counter() - start
# 4. 内存峰值
mem_kb = resource.getrusage(
resource.RUSAGE_SELF).ru_maxrss
# mac 是 bytes / linux 是 kb
mem_mb = (mem_kb / 1024 if sys.platform != "darwin"
else mem_kb / 1024 / 1024)
return ScaleEvalReport(
sample_count=total,
total_time_seconds=round(elapsed, 1),
samples_per_second=round(total / max(elapsed, 0.001), 0),
cpu_count_used=self.workers,
memory_peak_mb=round(mem_mb, 0),
)flowchart LR S[1M samples] --> C[chunk 10k each] C --> P[multiprocessing Pool<br/>worker = CPU count] P --> W1[worker 1] P --> W2[worker 2] P --> W3[worker N] W1 --> M[merge results] W2 --> M W3 --> M M --> R[ScaleEvalReport] style P fill:#e3f2fd style M fill:#e8f5e9
工程实务的 4 类规模化经验：
| 规模 | 推荐方案 | 时间 |
|---|---|---|
| < 1k | 单线程 | 1 秒 |
| 1k-100k | 单进程 + §5.10.5 的常数优化 | 数秒-1 分钟 |
| 100k-1M | multiprocessing | 数分钟 |
| > 1M | Spark / Ray 分布式 | 数十分钟 |
具体例子（示意数据，需用你自己的数据复算）：1M trace 评测：
| 方案 | 总时间 | sample/s |
|---|---|---|
| 单线程 | 5 hours | 55 |
| 单机多进程 8 worker | 1.5 hours | 185 |
| multiprocessing 32 worker | 25 min | 660 |
| Ray 分布式 100 worker | 8 min | 2080 |
3 类大规模规模化坑：
| 坑 | 现象 | 修法 |
|---|---|---|
| 内存爆 | 1M sample 一次 load 到内存 | streaming + chunk |
| pickling 慢 | multiprocessing 序列化慢 | 用 dill 或共享内存 |
| 不监控 | OOM 时不知道 | 必含 memory_peak |
研究背景：
- Apache Spark 是 100M+ 样本评测的工业标杆
- Ray 是 ML 友好的分布式框架
- pandas + numba 也能加速中规模
5.11 给判分器写测试
5.11.1 隐藏陷阱：判分代码自身的测试
最后讨论一个反讽的话题——你的 grader 代码本身要不要测？
听起来像递归问题，但答案是肯定的。grader 是软件工程的一部分，bug 同样会污染评测结论。最常见的 grader bug：
- 归一化函数漏处理某种 unicode → 一类样例全部被错判
- regex 写错 → 大批样例假阳性 / 假阴性
- LLM-judge prompt 模板里有 typo → 给所有样例打分都偏 1-2 分
- Bootstrap 抽样函数有 off-by-one → CI 计算偏差
修法：grader 模块要写单元测试——传统软件工程的做法直接复用：
def test_normalize_handles_chinese_punct():
assert normalize("巴黎，法国") == normalize("巴黎,法国")
def test_grade_numeric_extracts_correct_number():
assert grade_numeric("Answer: 42 fruits", 42)["ok"]
assert not grade_numeric("Answer: 42 fruits", 43)["ok"]
def test_grade_schema_rejects_extra_fields():
schema = {"type": "object", "additionalProperties": False, "required": ["a"]}
assert not grade_schema('{"a": 1, "b": 2}', schema)["ok"]这一层测试是评测体系工程化的最后一公里。它把"评测"和"评测代码本身"分开管理——后者用传统单测保证正确，前者用元评测保证可靠。这种"递归不递归化"的设计，是工业级评测系统的标志。
5.11.2 规则判分常被忽略的工程层：grader 的端到端测试
评测代码自身要测（§5.11.1 单元测试），但还有一个更高层的工程动作——grader 的端到端测试。
具体做法：构造一组"已知应该通过 / 已知应该失败"的样例，让 grader 跑一遍，验证它的判定是否符合预期：
# grader 端到端测试样例
TEST_CASES = [
{
"name": "完全正确的回答",
"expected_grade": True,
"input": "中国首都?",
"output": "北京",
},
{
"name": "多了空格但语义对",
"expected_grade": True, # 期望 grader 能容忍
"input": "中国首都?",
"output": "北 京",
},
{
"name": "完全错误",
"expected_grade": False,
"input": "中国首都?",
"output": "上海",
},
{
"name": "答非所问",
"expected_grade": False,
"input": "中国首都?",
"output": "中国是亚洲国家",
},
# ... 20-50 条覆盖各种情况
]
def test_grader():
for case in TEST_CASES:
# my_grader 是被测的判分函数：输入模型输出与参考答案，返回 bool
actual = my_grader(case["output"], reference="北京")
assert actual == case["expected_grade"], f"Failed: {case['name']}"这种"grader 自身的回归测试"通常会暴露出令人惊讶的 bug——某条 normalize 函数对全角空格不处理，某条 regex 在边缘情况下卡死，某条 schema 校验对 null 字段处理不当。
工业团队的实操：每个 grader 配 30-50 条端到端测试，每次改 grader 代码先跑这套测试。pre-commit hook 自动跑，确保任何改动不破坏 grader 行为。这种"双层测试"——单元测试 + 端到端测试——是评测体系的工程质量保证。
5.11.3 规则判分的"测试驱动开发"——给规则写单测的工程化
规则判分代码不写单测 = 早晚出 bug。下面给出 TDD 风格的规则开发模板：
import pytest
from dataclasses import dataclass
# 规则定义
def rule_refund_in_days(sample: dict) -> bool:
"""检测回答是否提到退款时效（3-5 工作日）"""
text = sample.get("response", "").lower()
if "退款" not in text:
return False
has_days = any(t in text for t in ["3", "5", "三", "五", "工作日"])
return has_days
# 单测（pytest 风格）
class TestRefundRule:
@pytest.mark.parametrize("response,expected", [
("退款将在 3-5 个工作日到账", True),
("退款 3 天内到", True),
("您的退款将到账，请耐心等待", False), # 没说几天
("退款 1 个月", False), # 时效错
("您的订单已发货", False), # 完全无关
("Your refund will arrive in 5 business days", False), # 英文（设计为只查中文）
])
def test_basic_cases(self, response, expected):
assert rule_refund_in_days({"response": response}) == expected
def test_empty_response(self):
assert rule_refund_in_days({"response": ""}) is False
def test_missing_response_field(self):
assert rule_refund_in_days({}) is False
def test_unicode_normalization(self):
# 全角数字：本实现没做 NFKC，"３" 匹配不到 "3"——用测试把这个行为钉住，
# 哪天决定支持全角，就先把这里改成 True 再改实现
assert rule_refund_in_days(
{"response": "退款 ３ 天"}
) is False
@dataclass
class RuleTestCoverage:
rule_name: str
test_count: int
pass_count: int
edge_cases_covered: int
coverage_grade: str
class RuleTDDChecker:
"""检查每条规则的单测覆盖度"""
REQUIRED_EDGE_CASES = [
"empty_input",
"missing_field",
"unicode_edge",
"whitespace_only",
"multilingual",
]
def assess(self, rule_name: str, test_results: dict) -> RuleTestCoverage:
n_tests = test_results.get("total", 0)
n_pass = test_results.get("passed", 0)
# 检查 edge case 覆盖
edge_covered = sum(
1 for ec in self.REQUIRED_EDGE_CASES
if ec in str(test_results.get("test_names", []))
)
if n_tests >= 10 and edge_covered >= 4:
grade = "A"
elif n_tests >= 5 and edge_covered >= 2:
grade = "B"
elif n_tests >= 3:
grade = "C"
else:
grade = "F"
return RuleTestCoverage(
rule_name=rule_name,
test_count=n_tests,
pass_count=n_pass,
edge_cases_covered=edge_covered,
coverage_grade=grade,
)flowchart LR R[新规则需求] --> T1[Step 1: 写单测<br/>5+ basic + 3+ edge] T1 --> T2[Step 2: 跑测试 → 全失败] T2 --> I[Step 3: 实现规则] I --> T3[Step 4: 跑测试 → 全 pass] T3 --> CI[Step 5: PR + CI gate] CI --> MERGE[merge] T3 -. "失败" .-> BACK[修规则] BACK --> T3 style T1 fill:#e3f2fd style MERGE fill:#e8f5e9
工程实务的 5 类必测 edge case：
| edge case | 测试理由 |
|---|---|
| empty input | 防 KeyError / IndexError |
| missing field | 数据 schema 不一致时 |
| unicode edge | 全角 / emoji / RTL 文字 |
| whitespace only | 空格回答 |
| multilingual | 中英混排 |
具体例子（示意数据，需用你自己的数据复算）：某团队 80 条规则单测覆盖度：
| grade | 数量 | 行动 |
|---|---|---|
| A | 25 | 维持 |
| B | 35 | 加 edge case |
| C | 15 | 加单测到 ≥ 5 个 |
| F | 5 | 必修 - 几乎无单测 |
3 类常见 TDD 反模式：
| 反模式 | 现象 | 修法 |
|---|---|---|
| 写完代码再补单测 | 测出来全 pass 假象 | 先单测 → 失败 → 实现 |
| 单测和实现同人写 | bias 严重 | reviewer 必查 edge case |
| 不测错误路径 | 异常输入崩 | 必测 empty / missing / 异常 |
研究背景：
- Test-Driven Development (Kent Beck 2003) 是 TDD 经典
- pytest fixtures + parametrize 让 LLM eval 测试简洁
- coverage.py（配合 pytest-cov）是 Python 覆盖率统计的常用工具
5.11.4 规则判分的"property-based testing"启示
借鉴软件工程的 property-based testing（如 QuickCheck / Hypothesis），规则判分也可以做"基于属性的测试"：
from hypothesis import given, strategies as st
# 传统单元测试
def test_grade_specific():
assert grade("北京", expected="北京")
# Property-based 测试
@given(st.text())
def test_grade_normalize_idempotent(text):
"""归一化是幂等的"""
assert normalize(normalize(text)) == normalize(text)
@given(st.text(min_size=1))
def test_grade_self_match(text):
"""任何文本与自己 exact match"""
assert grade_exact(text, text)["ok"] # grade_exact 返回 dict，要取 ok
@given(st.text())
def test_grade_handles_whitespace(text):
"""加空格不影响 normalize 结果"""
assert normalize(text) == normalize(f" {text} ")property-based testing 用 Hypothesis 等工具自动生成测试输入，能发现单元测试想不到的边缘 case。一个真实的例子：如果 §5.6 的 normalize 先 lower() 再做 NFKC，第一条幂等测试会立刻失败——ℌ 经 NFKC 变成大写 H，第二次归一化又被转成 h；本章代码因此把 NFKC 放在最前、strip() 放在最后。对规则判分代码（特别是 normalize / regex）的鲁棒性验证特别有用。
工程实务：在 grader 仓库引入 property-based testing 框架（Python Hypothesis / JS fast-check）。每次 grader 改动除了跑常规单测，还跑 100-1000 次随机生成的输入。这种"严苛测试"让 grader 的 bug 在生产前就被暴露。
5.12 覆盖度、可观测与演进
5.12.1 规则判分的"覆盖度评估"工程
写完规则判分后还需要回答一个问题——这套规则覆盖了多少业务场景？这个评估常被忽视。
具体方法：
- 取 100 条真实生产 trace
- 用现有规则判分跑一遍
- 统计：
- 规则判分能覆盖（pass 或 fail 都明确）的样例 %
- 规则判分给出"未覆盖"或"无法判断"的样例 %
覆盖度的目标值没有统一标准，可以先定一个经验线（例如八成，示意阈值）；如果只有一半左右，说明规则覆盖面太窄，需要扩充或回退到 LLM-judge。
修法：
- 覆盖度低：分析未覆盖样例的共性，加 5-10 条新规则
- 覆盖度高但失败率高：规则太宽松，加更严格断言
- 覆盖度高且失败率低：规则体系成熟，可日常使用
这种"覆盖度评估"是规则判分体系的"自我体检"。每月跑一次，能发现规则体系是否随业务演化保持适用。规则不更新会逐渐失去对生产的覆盖能力——这是评测体系工程化的隐藏维护任务。
5.12.2 "规则判分覆盖率统计"工具：让团队知道还能扩多远
规则覆盖率越高，需要送进 LLM-judge 的样本越少，整体判分成本越低——但怎么知道当前覆盖率？工程团队往往凭体感"应该 50% 吧"——错了 20pp 都不奇怪。下面是一份覆盖率统计工具，每周给团队"还能再扩多少"的客观数字：
import json
from dataclasses import dataclass, field
from collections import defaultdict
from typing import Iterable
@dataclass
class CoverageReport:
total_samples: int
rule_handled: int
llm_judge_handled: int
human_handled: int
rule_coverage_pct: float
llm_judge_coverage_pct: float
human_coverage_pct: float
extension_candidates: list[dict]
estimated_savings_per_10pp_extension_usd: float
class RuleCoverageAnalyzer:
"""统计当前评测集中规则可处理的占比，并标注"还能扩"的候选"""
def __init__(self, rule_funcs: list,
llm_judge_cost_per_call: float = 0.015,
human_cost_per_call: float = 1.0):
self.rule_funcs = rule_funcs
self.llm_cost = llm_judge_cost_per_call
self.human_cost = human_cost_per_call
def _can_be_ruled(self, sample: dict) -> bool:
for f in self.rule_funcs:
try:
if f(sample):
return True
except Exception:
continue
return False
def _candidate_features(self, sample: dict) -> list[str]:
"""提取"看起来能写规则"的特征"""
features = []
ans = sample.get("expected", "")
if ans.strip().isdigit() or any(c in ans for c in "0123456789"):
features.append("contains_number")
if ans.startswith("{") and ans.endswith("}"):
features.append("looks_like_json")
if len(ans) < 30:
features.append("very_short")
if any(kw in ans.lower() for kw in ["yes", "no", "是", "否", "true", "false"]):
features.append("binary_answer")
if any(c in ans for c in "ABCD") and len(ans) <= 5:
features.append("multiple_choice")
return features
def analyze(self, samples: list[dict]) -> CoverageReport:
rule_count, llm_count, human_count = 0, 0, 0
un_ruled_features = defaultdict(list)
for sample in samples:
if self._can_be_ruled(sample):
rule_count += 1
continue
grading_hint = sample.get("grading_method", "llm_judge")
if grading_hint == "human":
human_count += 1
else:
llm_count += 1
for feat in self._candidate_features(sample):
un_ruled_features[feat].append(sample.get("id", "?"))
n = max(len(samples), 1)
candidates = [
{"feature": feat, "count": len(ids),
"potential_pct_lift": round(len(ids) / n * 100, 1),
"sample_ids": ids[:5]}
for feat, ids in sorted(un_ruled_features.items(),
key=lambda x: -len(x[1]))[:5]
]
savings = (n * 0.10 *
(self.llm_cost * llm_count / max(llm_count + rule_count, 1) +
self.human_cost * human_count / max(human_count + rule_count, 1)))
return CoverageReport(
total_samples=n,
rule_handled=rule_count,
llm_judge_handled=llm_count,
human_handled=human_count,
rule_coverage_pct=round(rule_count / n * 100, 1),
llm_judge_coverage_pct=round(llm_count / n * 100, 1),
human_coverage_pct=round(human_count / n * 100, 1),
extension_candidates=candidates,
estimated_savings_per_10pp_extension_usd=round(savings, 2),
)flowchart LR
S[评测集 N 题] --> A[Coverage Analyzer]
A --> R{逐题}
R -->|可规则化| RC[rule_handled]
R -->|不可| LJ{有 grading_method=human?}
LJ -->|是| HC[human_handled]
LJ -->|否| LC[llm_judge_handled]
LC --> FE[特征提取]
FE --> EX[extension_candidates<br/>top 5]
RC --> RPT[CoverageReport]
HC --> RPT
EX --> RPT
RPT --> AGG[reach_potential + savings_$]
style EX fill:#fff3e0
style RC fill:#e8f5e9
工程实务的 4 条使用模式：
- 每周一报：cron 跑一次，结果发到 Slack #evals-weekly
- 覆盖率 < 50% 红线：必须立即补规则
extension_candidatestop 5 直接转 ticket：每条带 sample_ids，工程师 1-2 小时能写新规则- 预估 savings 是说服管理层的关键数字：把"扩规则"从 nice-to-have 变成 "省 $X/year"
具体例子（示意数据，需用你自己的数据复算）：1000 题评测集，当前规则覆盖 35%、llm-judge 60%、human 5%。analyzer 报告：
- top extension 候选：
binary_answer占 12%（120 题）→ 1 个 yes/no 规则可全覆盖 - top 候选 2：
looks_like_json占 8%（80 题）→ 1 个 schema 验证规则可全覆盖 - 提升 20pp 覆盖率 → llm 调用降 200 题 × $0.015 = $3/run → 周度跑 = $156/year
数字不大但稳；评测集越大、跑得越频繁，这笔账越可观。
5.12.3 规则判分的可观测性：让失败 case 自带诊断信息
规则判分相比 LLM-judge 的另一个优势——失败 case 可以自带诊断信息。
LLM-judge 失败时只能告诉你"答案不对"，规则判分能精确告诉你哪里不对：
import json, re
import jsonschema
def grade_with_diagnostics(pred: str, expected: dict) -> dict:
diag = []
if expected.get("must_contain"):
for kw in expected["must_contain"]:
if kw not in pred:
diag.append(f"missing keyword: {kw}")
if expected.get("must_not_contain"):
for kw in expected["must_not_contain"]:
if kw in pred:
diag.append(f"forbidden keyword present: {kw}")
if expected.get("regex"):
if not re.search(expected["regex"], pred):
diag.append(f"regex not matched: {expected['regex']}")
if expected.get("schema"):
try:
obj = json.loads(pred)
jsonschema.validate(obj, expected["schema"])
except (json.JSONDecodeError, jsonschema.ValidationError) as e:
diag.append(f"schema violation: {e}")
return {"ok": len(diag) == 0, "diagnostics": diag}带诊断信息的判分输出，让开发者直接看到失败原因。在 CI Quality Gate 里把这些诊断信息直接 echo 到 PR 评论，开发者能很快定位问题。
这种"可观测的失败"是规则判分相比 LLM-judge 的最大优势之一——LLM-judge 给出的"reason" 字段虽然有解释，但解释本身的质量不可控。规则判分的诊断信息是确定性的：它说哪里错就是哪里错。
5.12.4 规则判分的"差异 diff 报告"——让 PR review 一眼看出影响范围
规则判分代码改一行（比如调整正则、加一个归一化步骤），可能影响成千上万的历史样本判分。如果 PR review 只看代码 diff 而不看"判分结果 diff"，就会出现"代码看着合理、跑下来 30% 样本翻盘"的灾难。本节给读者一份"差异 diff 报告"工程方案——任何规则修改自动生成"前后对照报告"附在 PR 上。
graph LR
A[工程师改规则代码] --> B[git push 触发 CI]
B --> C[CI 跑 baseline<br/>用 main 分支规则]
B --> D[CI 跑 candidate<br/>用 PR 分支规则]
C & D --> E[diff 引擎比对每个样本]
E --> F[新过 / 新挂 / 维持<br/>三类统计]
F --> G[生成报告]
G --> H[PR comment 自动发布]
H --> I[reviewer 看到红绿对照]
I --> J{影响范围可接受?}
J -->|是| K[merge]
J -->|否| L[block + 调查根因]
diff 报告的 4 类样本变化分类：
| 分类 | 含义 | reviewer 关注度 | 处理建议 |
|---|---|---|---|
| 新通过（baseline 挂、candidate 过） | 规则放宽或修复 bug | 高 | 抽样 5 个手动 review，确认 expected 真的应该过 |
| 新失败（baseline 过、candidate 挂） | 规则收紧或引入 bug | 极高 | 必抽样全部 + 找 root cause；> 5% 必须 block PR |
| 维持通过 | 无变化 | 低 | 跳过 |
| 维持失败 | 无变化 | 低 | 跳过 |
配套实现：规则 diff 报告生成器：
from dataclasses import dataclass
from typing import Callable, Literal
ChangeKind = Literal["newly_pass", "newly_fail", "still_pass", "still_fail"]
@dataclass
class SampleDiff:
sample_id: str
baseline_pass: bool
candidate_pass: bool
def kind(self) -> ChangeKind:
if self.baseline_pass and self.candidate_pass: return "still_pass"
if not self.baseline_pass and not self.candidate_pass: return "still_fail"
return "newly_pass" if self.candidate_pass else "newly_fail"
@dataclass
class RuleDiffReport:
sample_diffs: list[SampleDiff]
block_threshold_newly_fail_pct: float = 5.0 # newly_fail 占总数 >= 5% 时阻止 merge
def by_kind(self) -> dict[ChangeKind, list[str]]:
groups: dict[ChangeKind, list[str]] = {
"newly_pass": [], "newly_fail": [],
"still_pass": [], "still_fail": [],
}
for d in self.sample_diffs:
groups[d.kind()].append(d.sample_id)
return groups
def summary(self) -> dict:
groups = self.by_kind()
total = len(self.sample_diffs)
return {
"total": total,
"newly_pass": len(groups["newly_pass"]),
"newly_fail": len(groups["newly_fail"]),
"still_pass": len(groups["still_pass"]),
"still_fail": len(groups["still_fail"]),
"newly_fail_pct": len(groups["newly_fail"]) * 100 / max(total, 1),
"net_delta": len(groups["newly_pass"]) - len(groups["newly_fail"]),
}
def merge_decision(self) -> Literal["allow", "warn", "block"]:
s = self.summary()
if s["newly_fail_pct"] >= self.block_threshold_newly_fail_pct:
return "block"
if s["newly_fail"] > 0:
return "warn"
return "allow"
def render_pr_comment(self) -> str:
s = self.summary()
decision = self.merge_decision()
emoji = {"allow": "OK", "warn": "WARN", "block": "BLOCK"}[decision]
return (
f"## 规则判分 diff 报告 [{emoji}]\n\n"
f"- 总样本：{s['total']}\n"
f"- 新通过：{s['newly_pass']}\n"
f"- 新失败：{s['newly_fail']}（{s['newly_fail_pct']:.1f}%）\n"
f"- 净变化：{s['net_delta']:+d}\n"
f"- 决策：{decision}\n\n"
f"newly_fail 样本（前 10）：{self.by_kind()['newly_fail'][:10]}"
)举例（示意数据，需用你自己的数据复算）：某 PR 调整了一行 regex（把 \d+ 改成 \d+(\.\d+)? 以支持小数），跑全 1500 样本：
- newly_pass = 87（小数答案的样本现在能过）
- newly_fail = 12（某些原本 match 的样本因贪婪匹配错位失败）
- newly_fail_pct = 0.8% → 决策 = warn
- reviewer 抽样 12 个 newly_fail 后发现都是 corner case，加 2 个 unit test 后批准 merge
配套行业研究背景：
- 新旧实现并跑、比对结果差异的做法，代表实现是 GitHub 2016 年开源的 Scientist
- 只跑受改动影响的测试（Test Impact Analysis），Visual Studio / Azure Pipelines 都有同类功能
5.12.5 规则判分的"长尾扩展"工程
规则判分体系成熟后，会进入一个长期工程节奏——长尾扩展。每周从生产 trace 中发现 1-2 个新失败模式 → 加 1-2 条新规则。
具体节奏（示意数据）：
Week 1: 规则数 50, 覆盖 70%
Week 4: 规则数 60, 覆盖 75%
Week 12: 规则数 80, 覆盖 80%
Week 26: 规则数 110, 覆盖 85%
Week 52: 规则数 150, 覆盖 88%注意：规则增长是次线性的——规则数翻倍，覆盖度只多 5-10pp。这是因为长尾失败模式越来越罕见，新规则解决的问题越来越窄。
工程上的判断信号：
- 当增加 10 条规则只能提升 0.5pp 覆盖度 → 长尾区间到了，停下来
- 当某条规则只解决 < 1% 样例 → 不值得加，让 LLM-judge 兜底
这种"知道何时停"的判断比"无脑加规则"重要。长尾不必追求完美，留 5-15% 给 LLM-judge / 人工是务实的工程姿态。
5.12.6 规则判分的"渐进式严格化"工程节奏
工业团队的规则判分体系不是一次性建立的，而是渐进式严格化。具体节奏：
Month 1: 5 条核心规则, 严格度低 (尽量不误伤)
- exact match, contains keywords
- 阈值宽松 (e.g., 75%+ 即通过)
Month 2: 10 条规则, 严格度中
- 加 schema validation, regex
- 阈值收紧到 80%
Month 3-6: 20-30 条, 严格度中-高
- 加专项规则 (合规 / 业务约束)
- 阈值 85%
Month 6-12: 50+ 条, 严格度高
- 全覆盖 + 长尾扩展
- 阈值 90%+这种"先松后紧"的节奏让团队不会一上来就被严格度卡住——前期容忍较多 false positive，后期才追求高精度。这是评测体系建设的"心智 ramp up" — 比"上来就追求 100% 严格"务实。
工程实务：把"严格度演化时间表"作为评测体系建设的 OKR——每月明确"这个月要从 X% 严格度提升到 Y%"。这种"可度量的演化路径"让评测体系建设有了具体节奏，避免"做了几个月还是不知道好不好"的迷茫状态。
5.12.7 规则判分的"工程债务"识别——什么时候该删 / 重写
规则判分的代码会随着评测集成长而累积——5 年下来 3000 行规则代码不奇怪。下面给出"债务识别"工具，定期清理：
import re
from dataclasses import dataclass
from collections import defaultdict
from pathlib import Path
from typing import Iterable
@dataclass
class RuleDebtIndicator:
rule_file: str
line_count: int
last_triggered_days: int
trigger_count_30d: int
coverage: float # 触发频率 / 评测集大小
code_complexity: int # cyclomatic complexity
debt_score: float
recommendation: str
class RuleDebtAnalyzer:
"""识别规则判分代码中的"工程债务"——可删 / 可重写"""
DEAD_THRESHOLD_DAYS = 90
LOW_COVERAGE_THRESHOLD = 0.001 # 0.1% 评测集触发
HIGH_COMPLEXITY_THRESHOLD = 15
def _measure_complexity(self, file_path: Path) -> int:
"""简化版圈复杂度——数 if/elif/while/for"""
text = file_path.read_text()
return (text.count("if ") + text.count("elif ") +
text.count("while ") + text.count("for ") +
text.count("\nexcept "))
def _line_count(self, file_path: Path) -> int:
return len(file_path.read_text().splitlines())
def analyze(self, file_path: Path,
trigger_log: list[dict]) -> RuleDebtIndicator:
recent = [r for r in trigger_log
if r.get("rule_file") == str(file_path)]
triggers_30d = len([r for r in recent
if r.get("days_ago", 999) <= 30])
if recent:
last_triggered = min(r.get("days_ago", 999) for r in recent)
else:
last_triggered = 999
complexity = self._measure_complexity(file_path)
lines = self._line_count(file_path)
coverage = triggers_30d / 1000 # 假设评测集 1000 题
debt = (
(last_triggered / 365) * 0.4 +
(1 - min(coverage / 0.1, 1)) * 0.3 +
(complexity / 50) * 0.3
)
debt = min(debt, 1.0)
if last_triggered > self.DEAD_THRESHOLD_DAYS and triggers_30d == 0:
rec = "DEAD: 删除（90+ 天未触发）"
elif coverage < self.LOW_COVERAGE_THRESHOLD:
rec = "LOW_USE: 评估是否仍需要——可能合并到通用规则"
elif complexity > self.HIGH_COMPLEXITY_THRESHOLD:
rec = "COMPLEX: 重构——拆 + 加单测"
else:
rec = "HEALTHY: 维持"
return RuleDebtIndicator(
rule_file=str(file_path),
line_count=lines,
last_triggered_days=last_triggered,
trigger_count_30d=triggers_30d,
coverage=round(coverage, 4),
code_complexity=complexity,
debt_score=round(debt, 3),
recommendation=rec,
)flowchart TB
R[规则文件] --> A[Debt Analyzer]
A --> CC[圈复杂度]
A --> LT[last_triggered]
A --> CO[coverage]
CC --> S{score 算法}
LT --> S
CO --> S
S --> Q1{> 90 天<br/>未触发?}
S --> Q2{coverage<br/>< 0.1%?}
S --> Q3{复杂度<br/>> 15?}
Q1 -->|是| DEAD[DEAD: 删除]
Q2 -->|是| LOW[LOW_USE: 合并 / 评估]
Q3 -->|是| CMP[COMPLEX: 重构]
Q1 -->|否| OK
Q2 -->|否| OK
Q3 -->|否| OK[HEALTHY 维持]
style DEAD fill:#ffebee
style OK fill:#e8f5e9
工程实务的 4 条 debt 治理原则：
- 每月跑 audit：cron 触发，结果发到 evals owner Slack
- DEAD 规则直接删除：保留备份在 git，但当前代码移除
- COMPLEX 规则按需拆：不是所有复杂规则都重构——只重构会被人改的那部分
- debt_score > 0.7 触发 review：不必非要做什么，但必须有人看
具体例子（示意数据，需用你自己的数据复算）：某团队 80 个规则文件 6 个月债务报告：
| 状态 | 数量 | 行动 |
|---|---|---|
| HEALTHY | 52 | 维持 |
| DEAD | 12 | 删除 → 代码量 -800 行 |
| LOW_USE | 8 | 合并为通用规则 → -200 行 |
| COMPLEX | 8 | 拆分 + 加单测 → +100 行单测 |
净效果：规则代码量从 3200 行降到约 2200 行（约 -31%），单测覆盖率从 35% 升到 65%——半年下来代码可维护性大幅提升。
3 类 debt 来源：
| 来源 | 现象 | 修法 |
|---|---|---|
| 一次性测试遗忘 | 旧 task 退役但规则没删 | 跟 dataset retire 联动 |
| Copy-paste 累积 | 5 个相似规则只差 1 行 | 抽公共函数 |
| 业务变化 | 规则不再对应业务概念 | 重写或删除 |
研究背景：
- Sculley et al. 2015 "Hidden Technical Debt in Machine Learning Systems"（NeurIPS 2015）是 ML 技术债的经典论述
- 死代码检测可借助 vulture 这类静态分析工具，再结合上面的触发日志交叉确认
5.13 多语种与国际化
5.13.1 规则判分的"国际化"陷阱清单
中国团队最常踩的规则判分坑是国际化适配——同一份"contains_check"的英文规则放到中文 / 日文 / 阿拉伯文 LLM 输出上往往失效。下面是一份基于多语言文本规范化的踩坑清单：
| 类型 | 英文规则示例 | 多语言失效形态 | 工程修法 |
|---|---|---|---|
| 标点 | text.endswith(".") |
中文用句号 。、日文 。、阿拉伯文 ۔ |
用 Unicode 类别 unicodedata.category(c) == 'Po' |
| 数字 | re.search(r"\d+") |
Python 3 的 \d 默认匹配所有 Unicode Nd（含阿拉伯-印度数字 ٠١٢），范围比预期宽；中文数字 一二三 不属于 Nd，NFKC 也不转换 |
只要 ASCII 数字就用 [0-9] 或 re.ASCII；中文数字需专门转换 |
| 大小写 | text.lower() == "yes" |
德语 ß 与 ss 不等；土耳其语 İ/ı 需要区域化映射 | text.casefold() 能处理 ß→ss 等，但它与区域无关，土耳其语仍需单独处理（如 PyICU） |
| 空白 | text.strip() |
strip() 能去全角空格 U+3000，但零宽 U+200B 不算空白，\s 与 NFKC 也都去不掉 |
显式删除零宽字符（U+200B / U+200C / U+200D / U+FEFF） |
| 引号 | '"' in text |
中文 “…” / 法文 «…» / 德文 „…“ |
把所有引号归一化到 ASCII 后再判 |
| 长度 | len(text) > 100 |
中英文单个字符的信息量差别很大，同一阈值不可比 | 按 token 数判而非 char 数 |
| 片段比对 | "hello" in text |
"hello" 会命中 helloworld |
英文词用 (?<![A-Za-z])hello(?![A-Za-z])；别用 \b，汉字也算 \w，说hello吧 里 \bhello\b 匹配不到 |
| 编号格式 | r"^\d+\. " |
中文 1、2、3、 用顿号 |
多写一条 r"^\d+[\.、]" |
import unicodedata
import regex
from typing import Callable
class I18nNormalizingMatcher:
"""跨语言鲁棒的规则判分预处理"""
QUOTES_TO_ASCII = {
'“': '"', '”': '"', '‘': "'", '’': "'",
'「': '"', '」': '"', '『': '"', '』': '"',
'«': '"', '»': '"', '„': '"',
}
def normalize(self, text: str) -> str:
text = unicodedata.normalize("NFKC", text)
text = "".join(self.QUOTES_TO_ASCII.get(c, c) for c in text)
text = regex.sub(r"\s+", " ", text)
return text.casefold().strip()
def contains(self, text: str, needle: str) -> bool:
return self.normalize(needle) in self.normalize(text)
def word_boundary_match(self, text: str, needle: str) -> bool:
# 注意：汉字同属 \p{L}，中文 needle 或中英黏连时会漏判，中文请用 contains()
pattern = r"(?<![\p{L}\p{N}])" + regex.escape(needle) + r"(?![\p{L}\p{N}])"
return regex.search(pattern, text, regex.UNICODE) is not Noneflowchart LR IN[原始 LLM 输出] --> NFKC[NFKC 归一化] NFKC --> Q[引号统一 ASCII] Q --> WS[空白 collapse] WS --> CF[casefold] CF --> NORM[标准化文本] NORM --> M1[contains 检查] NORM --> M2[正则 \\b 匹配] NORM --> M3[长度按 token] style NORM fill:#e3f2fd
工程实务的 4 条原则：
- 永远先 NFKC 归一化——把全角 / 半角 / 兼容字符都拍平
- 正则用
regex库而非re——支持\p{Nd}、\p{L}、Unicode property - 判长度按 token 不按 char——不同文字单个字符的信息量差别很大
- 对右到左语言（阿拉伯 / 希伯来）测一次双向控制符——RTL 文本在内存里仍按逻辑顺序存储，规则不需要反转字符串；要测的是 U+200F、U+202E 这类控制符混进输出后规则是否仍然成立
工程实务：把 I18nNormalizingMatcher 作为规则判分库的"基础层"——所有上层规则都通过它做预处理。这是中国团队跨境业务、海外公司支持中文场景的"基本功"。
5.13.2 规则判分的"语言无关"评测——多语种产品如何复用同一套规则
国际化的 LLM 应用面对中 / 英 / 日 / 西 / 阿等多语种。如果每种语言写一套规则 = 5 倍代码 + 5 倍维护成本。本节给读者一份"语言无关规则"工程模式——把规则解耦成 (语义 + 语言适配器) 两层，让一套核心规则覆盖多语种，把规则代码维护成本压到 1.x 倍而非 5x。
graph LR
A[国际化产品 LLM 输出] --> B{语言检测}
B --> C[zh-CN]
B --> D[en]
B --> E[ja]
B --> F[es]
B --> G[ar]
C & D & E & F & G --> H[语言适配器层]
H --> I[标准化输出<br/>统一 token / 数字 / 日期]
I --> J[语义规则核心]
J --> K[contains_intent]
J --> L[is_polite]
J --> M[has_disclaimer]
K & L & M --> N[最终判分]
5 类语言 × 适配器关键差异：
| 语言 | 数字格式 | 日期格式 | 标点 | 适配器关键步骤 | 规则核心是否复用 |
|---|---|---|---|---|---|
| zh-CN | 1,234.56 / 全角 | 2026-04-28 / 4月28日 | 全角双引号、句号 | NFKC + 中文意图同义词表 | ✓ |
| en | 1,234.56 | 2026-04-28 / Apr 28 | 半角 | 标准化 stop words | ✓ |
| ja | 1,234.56 / 全角 | 2026年4月28日 | 全角、波形号 | NFKC + 日文敬语正常化 | ✓ |
| es | 1.234,56 | 28/04/2026 | 倒置感叹号 | 数字逗号互换 + 倒置标点 | ✓ |
| ar | ١٬٢٣٤٫٥٦ | RTL 显示 | RTL bidi | RTL 处理 + 阿拉伯数字 → ASCII | ✓ |
配套实现：语言无关规则引擎：
import re
import unicodedata
from dataclasses import dataclass, field
from typing import Literal, Callable
LangCode = Literal["zh-CN", "en", "ja", "es", "ar"]
@dataclass
class LanguageAdapter:
code: LangCode
def normalize(self, text: str) -> str:
normalized = unicodedata.normalize("NFKC", text)
if self.code == "es":
normalized = self._es_number_normalize(normalized)
if self.code == "ar":
normalized = self._ar_number_to_ascii(normalized)
return normalized
@staticmethod
def _es_number_normalize(s: str) -> str:
# "1.234,56" → "1234.56"
return re.sub(r"(\d)\.(\d{3})", r"\1\2", s).replace(",", ".")
@staticmethod
def _ar_number_to_ascii(s: str) -> str:
ar_digits = "٠١٢٣٤٥٦٧٨٩"
for i, d in enumerate(ar_digits):
s = s.replace(d, str(i))
return s
def intent_synonyms(self, intent: str) -> list[str]:
table = {
"refund": {
"zh-CN": ["退款", "退还", "退钱"],
"en": ["refund", "return money", "reimburse"],
"ja": ["返金", "払い戻し"],
"es": ["reembolso", "devolución"],
"ar": ["استرداد", "إعادة المال"],
},
"polite": {
"zh-CN": ["请", "您", "麻烦", "谢谢"],
"en": ["please", "thank you", "kindly"],
"ja": ["ありがとう", "お願い", "敬具"],
"es": ["por favor", "gracias", "cordial"],
"ar": ["من فضلك", "شكرا"],
},
}
return table.get(intent, {}).get(self.code, [])
@dataclass
class LanguageAgnosticRule:
name: str
intent: str
must_appear: bool = True
def evaluate(self, text: str, adapter: LanguageAdapter) -> dict:
normalized = adapter.normalize(text)
synonyms = adapter.intent_synonyms(self.intent)
hits = [s for s in synonyms if s.lower() in normalized.lower()]
passed = (len(hits) > 0) == self.must_appear
return {
"rule": self.name,
"intent": self.intent,
"lang": adapter.code,
"passed": passed,
"hits": hits,
}
@dataclass
class MultilingualRuleEngine:
rules: list[LanguageAgnosticRule]
def detect_language(self, text: str) -> LangCode:
if re.search(r"[ぁ-んァ-ン]", text): return "ja" # 先查假名：日文也含汉字
if re.search(r"[一-鿿]", text): return "zh-CN"
if re.search(r"[-ۿ]", text): return "ar"
if re.search(r"[áéíóúñ¿¡]", text, re.IGNORECASE): return "es"
return "en"
def evaluate(self, text: str) -> dict:
lang = self.detect_language(text)
adapter = LanguageAdapter(code=lang)
results = [r.evaluate(text, adapter) for r in self.rules]
passed = all(r["passed"] for r in results)
return {"lang": lang, "all_pass": passed,
"details": results,
"code_reuse_pct": 100, # 单一规则覆盖所有语言
}举例（示意数据，需用你自己的数据复算）：某 5 语种客服产品规则集：
- 4 条核心规则：refund 必须包含 / 礼貌词必须 ≥ 1 / 不能出现公司隐私词 / 长度合理
- 5 个语言适配器（zh/en/ja/es/ar）共享这 4 条规则
- 跑 1000 题（每语种 200）→ 一套规则代码 250 行，覆盖 5 语种 1000 题
- 旧方案：5 套规则 × 250 行 = 1250 行 + 5 倍维护 → 现在压到 250 行 + 适配器 80 行 = 330 行
- 维护面从 5 套规则缩到 1 套规则 + 5 个适配器
配套行业研究背景：
- 多语种评测基准可对照 Microsoft 的 XGLUE（Liang et al. 2020, arXiv:2004.01401）
- 各语言的数字、日期格式以 Unicode CLDR 的区域数据为准
- 意图同义词表是对话系统 NLU 的常见做法
5.14 生态对照与实战
5.14.1 OpenAI evals 仓库的真实 grader 形态
第 9 章会逐行剖析 OpenAI evals。这里先看一眼它的 grader 抽象，便于读者建立感性认识：
# 示意：仿 evals/registry/evals/*.yaml 的写法，match_capitals 是虚构的 eval 名
match_capitals:
id: match_capitals.s1.simple-v0
description: Match country capitals
metrics: [accuracy]
match_capitals.s1.simple-v0:
class: evals.elsuite.basic.match:Match
args:
samples_jsonl: capitals/samples.jsonl它的核心抽象是把 grader 注册成一个类（这里是 evals.elsuite.basic.match.Match）+ 数据集 jsonl。不同规则判分是 evals.Eval 基类的不同子类，evals/elsuite/basic/ 下就有 Match、Includes、FuzzyMatch、JsonMatch、JsonValidator 等。这是工业级评测框架的典型设计——把"判分逻辑"和"数据集"解耦，用注册机制管理。
第 9 章会展示这个注册系统的实现细节。
5.14.2 promptfoo 的 assertion 库：业界最完整的规则判分清单
第 12 章会逐行剖析 promptfoo 源码。这里先看它的 assertion 关键字清单——它是业界最完整的"规则判分目录"，几乎所有团队的规则判分需求都能在这套关键字里找到映射：
| 类别 | 关键字 | 含义 |
|---|---|---|
| 字符串匹配 | equals |
精确匹配 |
| 字符串匹配 | contains |
包含子串 |
| 字符串匹配 | contains-all |
包含所有子串 |
| 字符串匹配 | contains-any |
包含任一子串 |
| 字符串匹配 | not-contains |
不包含 |
| 字符串匹配 | starts-with |
前缀 |
| 字符串匹配 | regex |
正则 |
| 字符串匹配 | not-regex |
反向正则 |
| 结构化 | is-json |
是合法 JSON |
| 结构化 | is-valid-openai-tools-call |
符合 OpenAI tool schema |
| 结构化 | contains-json |
包含一个 JSON |
| 结构化 | javascript |
自定义 JS 断言函数 |
| 结构化 | python |
自定义 Python 断言函数 |
| 数值 | cost |
调用成本上限 |
| 数值 | latency |
延迟上限 |
| 数值 | perplexity |
困惑度阈值 |
| 数值 | levenshtein |
编辑距离阈值 |
| 语义 | similar |
embedding 相似度 |
| 语义 | factuality |
事实一致性 |
| 语义 | model-graded-closedqa |
闭式 QA 模型判分 |
| 语义 | g-eval |
G-Eval 模板 |
| 语义 | llm-rubric |
自定义 rubric |
| 语义 | answer-relevance |
答案相关性 |
| 安全 | moderation |
内容审核 |
| 安全 | guardrails |
读取 provider 返回的安全护栏标记 |
| 工具 | webhook |
调外部 webhook 判分 |
这张表呈现两个工程洞察：
- 规则判分远比想象中丰富——不只是
equals/contains，包含成本、延迟、安全、结构等一整套断言体系 - 规则与 LLM-judge 在同一个抽象层——
g-eval/factuality/llm-rubric这些 LLM-judge 关键字与equals/regex共存于同一份 YAML，体现了"判分方法是可组合的"这一核心设计思想
工程团队第一次搭评测时，照着这张表逐一过一遍自己的需求，能避免重复造轮子。第 12 章会展示这套关键字背后的内部抽象（Zod enum 定义的断言类型 + type 到 handler 函数的派发表）。
5.14.3 实战：用 200 行规则判分覆盖 70% 客服评测
很多团队第一反应是"客服场景这么开放，规则判分不可能覆盖到 70%"。事实并非如此——把客服场景按答案形态拆解，规则判分能解决的远比想象多。下表的占比是一份示例分布（每家的 intent 分布都不一样，必须用自家 trace 统计替换），要看的是拆解方法而不是这些百分比：
| 客服 query 类别 | 占比（示例） | 规则判分能不能 | 用什么规则 |
|---|---|---|---|
| 物流查询（"我的快递到哪了"） | 25% | 能 | Schema：必须包含状态字段，不能编造单号 |
| 政策咨询（"你们退货政策是什么"） | 20% | 能 | Contains：必须命中政策关键词 |
| 订单状态（"订单 #123 怎么了"） | 15% | 能 | Regex：必须包含订单号回引 |
| 商品咨询（"X 和 Y 哪个好"） | 10% | 部分 | LLM-judge 兜底 |
| 投诉处理（"我对服务很不满意"） | 10% | 否 | 必须 LLM-judge + 人审 |
| 闲聊（"你叫什么名字"） | 5% | 能 | Schema：必须有人格设定字段 |
| 其他 | 15% | 部分 | LLM-judge |
把这张表合并：前三类 60% 完全用规则；第六类 5% 也能；第四、七类各 5-10% 部分用规则——加起来轻松 70% 以上。
剩下 30% 的"投诉、商品对比、复杂闲聊"才真正需要 LLM-judge。这种"规则覆盖大头、LLM-judge 攻克难题"的分层策略，是工业评测的最普遍范式。
5.14.4 规则判分的"成本对照"——为什么必须先用规则
下表是各类 grader 的量级对照（以 1000 条评测样本为基准）。延迟与单价随模型和时间变动，必须以调用当天的官方价目表为准；这里要读的是"零成本 / 分级成本"之间的数量级差距，不是具体数值：
| Grader 类型 | 单条延迟 | 单条成本量级 | 1k 条总成本量级 | 1k 条总耗时 | 同输入重跑是否稳定 |
|---|---|---|---|---|---|
| 正则 / equality | < 0.1ms | 0 | 0 | 1 秒 | 完全确定 |
| Schema 校验（Pydantic） | < 1ms | 0 | 0 | 1 秒 | 完全确定 |
| 关键字 + 排除词 | < 1ms | 0 | 0 | 1 秒 | 完全确定 |
| 数值精度断言 | < 0.5ms | 0 | 0 | 1 秒 | 完全确定 |
| LLM-judge（小模型档） | 亚秒级 | 约千分之一美元量级 | ~$1 量级 | 10 分钟（并发 10） | 不确定，需实测 |
| LLM-judge（旗舰档） | 秒级 | 千分之几到百分之几美元 | ~$5-20 量级 | 20-30 分钟（并发 10） | 不确定，需实测 |
| 人工评测 | ~30s | 众包到专业标注差一个量级 | $100-1000 量级 | 8 小时 | 不确定，需用 §7 的 κ 度量 |
最后一列必须自己测：任何声称"某某 judge 的可重复性是 95%"的表格都要追问它是怎么测的——同一条样本重跑 K 次看结论翻转率（§8.2.1 的 self-consistency），还是与人工标注的一致率（§8.2.2 的 κ）？这是两个完全不同的量，也是本专栏第 8 章要单独讲一章的原因。这里唯一能确定的一列是规则判分：同样输入必然同样输出，重跑一万次不变。
工业实务的漏斗经验（下面的比例是示意值，按自己的数据统计）：
- 70% 的样本应被规则筛掉（pass / fail 一目了然）
- 25% 的样本进 LLM-judge（语义判断）
- 5% 的样本进人工（争议 / 高风险）
算一下这个漏斗省在哪：1000 条全走 LLM-judge，判分调用是 1000 次；走漏斗只剩 300 次，judge 这一项的开销降到约 3/10。但注意漏斗底部那 5%（50 条）走人工——按人工单条成本比 judge 高两三个数量级算，人工那一层往往才是账单大头，这也正是它必须被压到个位数百分比的原因。所以"规则判分是第一道防线"不是哲学口号：它同时压掉了 judge 调用量和进入人工队列的样本量。
flowchart TB IN[1000 条样本] --> R[规则判分层] R -->|700 条 pass/fail 确定| OUT1[直接出分] R -->|300 条不确定| L[LLM-judge 层] L -->|250 条 LLM 决断| OUT2[出分 + 记录] L -->|50 条争议| H[人工评测层] H --> OUT3[最终判定 + 反哺 guideline] style R fill:#e8f5e9 style L fill:#fff3e0 style H fill:#ffebee
这里有个容易算错的地方：如果规则判分的边际成本近似为 0，总成本正比于剩下要走 judge / 人工的比例 1 − r。所以覆盖率 r 从 0.6 提到 0.7 时，成本降的是 0.1 / (1 − 0.6) = 25%，而不是 10%——覆盖率越高，再提 10 个百分点省下的相对比例越大（从 0.8 提到 0.9 直接砍掉一半）。
真正让扩规则停下来的不是收益递减，而是规则本身的维护成本递增：覆盖率爬到七成以上时，新增规则往往要处理越来越刁钻的边角情形，误判（把对的判成错的）风险随之上升。判断拐点的标准不是覆盖率数字，而是"新加的规则有没有开始制造假阴性"——这正是下一节要讲的问题。
5.14.5 隐藏成本：Regex 的"软件熵增"
规则判分有一个常被忽视的长期问题：regex 模式会随时间不断膨胀。
典型生命周期：
v1: r"\d{4}-\d{2}-\d{2}" # 17 字符, 简洁
v2: r"\d{4}[-/]\d{2}[-/]\d{2}" # 加斜杠分隔符
v3: r"\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?" # 加中文
v4: r"(?:\d{4}|\d{2})[-/年]\d{1,2}[-/月]\d{1,2}日?" # 加两位年份
v5: r"(?:\d{4}|\d{2})[-/年.]\s*\d{1,2}\s*[-/月.]\s*..." # 加空格容忍每一次膨胀都是合理的——线上发现了一类"答对但被判错"的样例。但半年下来，那条 17 字符的 regex 可能膨胀成几百字符的"巨兽"，没人完全看懂，改一处就可能引入新 bug。
这可以叫作判分断言的"腐化"——断言的复杂度逐渐超过被测对象本身。当你的 grader regex 比 LLM prompt 还长，应该升级到 LLM-judge 或 schema validation——后者用更结构化的语言表达同样的约束，长期可维护性高得多。
工程上的判断阈值：
- regex 短于 50 字符 → 继续 regex
- 50-150 字符 → 重构为多个独立规则的组合（更可读）
-
150 字符或包含 5+ 个 OR 分支 → 升级到 schema 或 LLM-judge
5.14.6 完整的规则判分实战 cheatsheet
整合本章方法学，给一份"规则判分用什么方法"的速查 cheatsheet：
| 业务场景 | 推荐方法 | 备注 |
|---|---|---|
| 抽取式 QA | Exact Match (归一化后) | 简单可靠 |
| 数值答案 | Numeric Match | 提前约定输出格式 |
| 包含关键词 | Contains | 配 normalization |
| 格式校验 | JSON Schema / Pydantic | 强类型 |
| 部分容忍模糊 | Levenshtein | 阈值 ≤ 3 |
| Tool calling | OpenAI tools schema | 结构化 |
| 引用合规 | Regex + Whitelist | 严格 |
| 事实陈述 | LLM-as-Judge | 升级 |
对照这份表，团队的大部分规则判分需求都能找到匹配方法；剩下的需要混合 / 自定义，本章方法学已经给了原料。
5.15 跨专栏关联
- 《MCP 协议设计与实现》专栏第 5 章定义的 tool calling JSON Schema，正是本章 §5.5 的 schema validation 直接对象
- 《LangGraph 设计与实现》专栏第 4 章 Channel 与 Reducer 产生的状态更新，本质上是带 schema 的结构化输出，可用本章方法判分
- 本专栏第 9 章会拆解 OpenAI evals 仓库
evals/elsuite/basic/match.py，对照本章方法 - 本专栏第 12 章会展示 promptfoo YAML 里 assertion 关键字（
equals、contains、is-json、javascript）如何映射到本章方法
5.16 本章小结
- 规则判分是评测体系第一道防线，能用规则就不用 LLM-judge——成本和耗时低几个数量级，结果完全可复现
- Exact Match 必须先做归一化（NFKC / 大小写 / 标点 / 空白 / trim），否则会有大量假阴性
- Numeric Match 要从自然语言里精确抽数字，需要约定输出格式（
Answer: <num>） - Regex 灵活但危险，必须 timeout 防 ReDoS；越写越长、分支越来越多的 regex 就该升级 schema 或 LLM-judge
- Schema Validation 是结构化输出的最强武器，与 Structured Outputs / Tool Use 完美协同
- 一份 100 行内的判分器可以覆盖大部分可规则化的评测需求；剩下的是 LLM-judge 的领地
下一章我们进入 LLM-as-Judge——评测体系最强但也最危险的工具。