第 5 章 规则判分：Exact Match / Regex / JSON Schema

“When you can solve a problem with a regex, you usually shouldn’t reach for an LLM.” —— 一条没人署名但流传很广的评测格言

5.1 为什么先讲规则判分

回想第 4 章的多指标体系：Faithfulness 要 LLM-judge、Trajectory Match 要 LLM-judge、Answer Relevance 也要 LLM-judge。但不是所有判分都要 LLM 来做——很多判分用 100 行 Python 就能解决，速度比 LLM-judge 快 1000 倍、成本低 10000 倍、可重复性 100%。

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。原因是三个：

1. 成本：规则判分是本地运算，无 API 费用
2. 速度：1000 条样例的规则判分 < 1 秒；LLM-judge 要分钟级
3. 可重复：规则判分是确定性函数，同一输入永远同一输出；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 = s.strip()
    s = s.lower()
    s = unicodedata.normalize("NFKC", s)        # 全角半角统一
    s = re.sub(r"[^\w\s]", "", s)                # 去标点
    s = re.sub(r"\s+", " ", s)                  # 空白归一
    return s

def exact_match(prediction: str, expected: str) -> bool:
    return normalize(prediction) == normalize(expected)

这段代码已经能应付 80% 的英文场景。中文场景还要额外处理：

• 全角 vs 半角标点（”。” vs ”.”；”，” vs ”,”）
• 简繁体（“巴黎” vs “巴黎”，简繁是不同 unicode）
• 数字大小写（“100” vs “一百”）

每一类归一化都是工程团队踩过坑后沉淀的——本书末尾附录会提供一份”评测归一化标准库”。

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 的实现就是这套——把答案 prompt 强制要求模型输出 Answer: <number>，然后 regex 抽取。

5.3.2 浮点数与单位

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())

容忍度 tolerance 必须根据题目设置。算术题用 1e-6，物理题（涉及测量误差）可能用 1%，金融题（涉及小数舍入）要按业务规则定。

5.3.3 单位陷阱

"10 km" vs "10000 m" 数值不同但答案相同——这种场景要在 normalize 阶段做单位换算，不能简单 numeric match。详见第 13 章 RAG 评测里的”领域归一化”小节。

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，遇到 aaaaaaaaaaaaaaab 会触发指数级回溯，几秒钟卡死 Python 进程。生产环境的判分器必须用 regex 库（不是 re）+ timeout：

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 False

5.4.2 何时升级到 LLM-judge

Regex 判分会让你写出越来越复杂的模式去捕捉”语义对、字面差”的情况。当你的 regex 超过 100 字符还在演化、覆盖一堆 OR 分支时，这就是该升级到 LLM-judge 的信号。

5.5 Schema Validation：结构化输出的最强判分

如果你的应用要求 LLM 输出结构化数据（JSON / YAML / TOML），Schema Validation 是判分的最强工具。

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 的优势

1. 覆盖度全：能检查类型、范围、枚举、嵌套结构、必填字段、模式
2. 错误消息友好：哪里挂了哪里说，failure case 直接可读
3. 行业标准：JSONSchema / OpenAPI / Pydantic 都是成熟规范，工具生态丰富

5.5.1.5 一个真实的 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", "ranking_score": 0.97}——ranking_score 不在 schema 里，但 schema validation 会通过，让你以为模型行为合规。这是评测里最隐蔽的 bug 之一。

5.5.2 与 OpenAI Structured Outputs / Anthropic Tool Use 的协同

OpenAI 的 Structured Outputs（2024-08 上线）和 Anthropic 的 Tool Use 都直接接受 JSON Schema 作为响应规范——模型生成时就被约束在 schema 内。

这个能力对评测的影响是革命性的：只要模型支持 Structured Output，schema validation 几乎 100% 通过率。所以现代 RAG / Agent 系统的趋势是：

• 应用层强制 Structured Output
• 评测层用 Schema Validation 做”硬约束”判分
• LLM-judge 只用来评 schema 内字段的”语义质量”

第 14 章会详述这种”硬约束 + 软评估”的双层评测模式。

5.6 一个工业级规则判分器：完整代码

把上述所有方法整合成一个 200 行的工业级判分器：

"""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 {}
    s = s.strip()
    if opts.get("case_insensitive", True):
        s = s.lower()
    if opts.get("nfkc", True):
        s = unicodedata.normalize("NFKC", s)
    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

# ============= 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-12 章 OpenAI evals / promptfoo 内置 grader 的核心抽象——不同框架在外壳上千差万别，但这套核心是相通的。

5.6.5 实战：用 200 行规则判分覆盖 70% 客服评测

很多团队第一反应是”客服场景这么开放，规则判分不可能覆盖到 70%“。事实并非如此——把客服场景按答案形态拆解，规则判分能解决的远比想象多：

把这张表合并：前三类 60% 完全用规则；第六类 5% 也能；第四、七类各 5-10% 部分用规则——加起来轻松 70% 以上。

剩下 30% 的”投诉、商品对比、复杂闲聊”才真正需要 LLM-judge。这种”规则覆盖大头、LLM-judge 攻克难题”的分层策略，是工业评测的最普遍范式。

5.6.7 一个隐藏成本：Regex 的”软件熵增”

规则判分有一个常被忽视的长期问题：regex 模式会随时间不断膨胀。

典型生命周期：

v1: r"\d{4}-\d{2}-\d{2}"                                # 8 字符, 简洁
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*..."  # 加空格容忍

每一次膨胀都是合理的——线上发现了一类”答对但被判错”的样例。但半年下来，那条 8 字符的 regex 变成了 300 字符的”巨兽”，没人完全看懂、改一处就可能引入新 bug。

这种现象在传统软件工程里叫 “test oracle erosion”——测试断言的复杂度逐渐超过被测代码本身的复杂度。当你的 grader regex 比 LLM prompt 还长，应该升级到 LLM-judge 或 schema validation——后者用更结构化的语言表达同样的约束，长期可维护性高得多。

工程上的判断阈值：

• regex 短于 50 字符 → 继续 regex
• 50-150 字符 → 重构为多个独立规则的组合（更可读）

• 150 字符或包含 5+ 个 OR 分支 → 升级到 schema 或 LLM-judge

5.6.8 OpenAI evals 仓库的真实 grader 形态

第 9 章会逐行剖析 OpenAI evals。这里先看一眼它的 grader 抽象，便于读者建立感性认识：

# evals/registry/evals/match_*.yaml 风格
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。所有规则判分都收敛到这一个抽象，不同 grader 通过子类化实现。这是工业级评测框架的典型设计——把”判分逻辑”和”数据集”解耦，用注册机制管理。

第 9 章会展示这个注册系统的实现细节，以及 OpenAI 内部是如何通过 90+ 种内置 grader 类覆盖各种规则判分场景。

5.6.9 promptfoo 的 assertion 库：业界最完整的规则判分清单

第 12 章会逐行剖析 promptfoo 源码。这里先看它的 assertion 关键字清单——它是业界最完整的”规则判分目录”，几乎所有团队的规则判分需求都能在这套关键字里找到映射：

这张表呈现两个工程洞察：

1. 规则判分远比想象中丰富——不只是 equals / contains，包含成本、延迟、安全、结构等一整套断言体系
2. 规则与 LLM-judge 在同一个抽象层——g-eval / factuality / llm-rubric 这些 LLM-judge 关键字与 equals / regex 共存于同一份 YAML，体现了”判分方法是可组合的”这一核心设计思想

工程团队第一次搭评测时，照着这张表逐一过一遍自己的需求，能避免重复造轮子。第 12 章会展示这套关键字背后的内部抽象（Assertion 基类 + 子类策略模式）。

5.6.10 把规则判分当作”可执行的标注 guideline”

最后一个视角，是规则判分的社会工程意义——它把”什么算合格回答”从口头共识、PPT 文档变成了可执行代码。

这一点对团队协作非常重要。考虑一个场景：PM 说”客服回答必须是友好的”，工程师问”具体指什么？” PM 说”就是别太冷淡”——这种描述无法落地。

但如果把它转成一组规则判分：

- description: 友好的客服回答
  assertions:
    - type: not-contains
      value: ["我不能", "无法", "不行"]  # 不直接拒绝
    - type: regex
      value: "^(您好|你好|好的|没问题)"   # 礼貌开头
    - type: latency
      threshold: 3000                     # 不让用户等
    - type: javascript
      value: "output.length >= 30"        # 不能太敷衍

这就是把”友好”翻译成了 4 条可量化、可校验、可在 CI 里跑的规则。PM 读得懂、工程师能执行、QA 可以追溯。这种”用规则表达共识”的能力，本身就是评测体系给团队带来的最大组织红利之一——比指标曲线还重要。

5.6.11 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 同时承担三个角色：

1. 应用层契约：OpenAI Structured Outputs / Anthropic Tool Use 用它强制约束 LLM 输出
2. 评测层断言：JSON Schema 校验 + Pydantic 校验天然作为规则判分
3. 类型层文档：上下游服务读这个 schema 就知道接口

这种”schema 驱动一切”的范式，能把评测从 30% 的工程时间压缩到 5%——因为每个新增字段的校验都是免费来的。Anthropic、OpenAI 在 2024 年都明确推动了这个方向（OpenAI Structured Outputs、Anthropic Tool Use 都是同一思路的产品落地）。

5.6.12 一个隐藏陷阱：判分代码自身的测试

最后讨论一个反讽的话题——你的 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.6.13 一个常被忽略的前置：输出归一化

判分前必须做的一步——归一化模型输出。这一步决定了后续所有规则判分的可靠性。

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

这个归一化层是规则判分的”地基”——所有后续 Exact Match / Regex / Numeric 都基于归一化后的字符串。工业团队的做法是把它做成共享库（如 evals_normalize），所有 grader 必经过它。这避免了”30 个 grader 各自实现归一化、各自有 bug”的工程债。

5.6.14 规则判分的可观测性：让失败 case 自带诊断信息

规则判分相比 LLM-judge 的另一个优势——失败 case 可以自带诊断信息。

LLM-judge 失败时只能告诉你”答案不对”，规则判分能精确告诉你哪里不对：

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 评论，开发者能在 30 秒内定位问题。

这种”可观测的失败”是规则判分相比 LLM-judge 的最大优势之一——LLM-judge 给出的”reason” 字段虽然有解释，但解释本身的质量不可控。规则判分的诊断信息是确定性的：它说哪里错就是哪里错。

5.6.15 规则判分与 Schema-Driven Generation 的协同

OpenAI Structured Outputs（2024-08）和 Anthropic Tool Use 都让模型输出强制符合 JSON Schema。这种”上游约束 + 下游校验”的模式对规则判分是革命性的影响。

传统流程：

模型输出自由文本 → 后处理解析 JSON → 失败时复杂错误处理 → schema validation

Structured 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 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

三层分工让”规则判分覆盖 70-80%“在 2025 年后成为现实——这是第 5 章的方法学在新工具加持下的延伸。

5.6.16 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

三层各自的成本和能力：

成本差距 100x → 10000x。Hybrid 让 80% 的样例在 Layer 1 就被解决（成本极低），10-15% 进 Layer 2（成本中等），剩下 5% 进 Layer 3（最贵但稀少）—— 整体平均成本接近 Layer 1。

工程实务上的优化：

• 早期失败 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.6.17 规则判分的”声明式 vs 命令式”哲学

回顾本章所有规则判分形态——Exact Match / Numeric Match / Regex / Schema Validation / Contains / Levenshtein——它们能被分成两类哲学：

• 声明式（declarative）：JSON Schema、Pydantic—“输出必须长这样”，框架做校验
• 命令式（imperative）：自定义 Python 函数 / regex—“按这个步骤检查”

两者各有适用：

工程经验：先用声明式，逼到极限再用命令式。新人 / 急于交付的工程师容易跳到 “javascript: () => …” 这种命令式断言——一开始快，但半年后维护成本指数级上升。

声明式的极限：当业务规则需要”基于多个字段的联合判断”（如”如果 intent 是 cancel 则必须有 reason 字段”），单一 JSON Schema 不够，但可以用 Pydantic + custom 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.6.18 规则判分的”组合 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}"}),
    GradingRule("not-contains", {"keywords": ["违法", "色情"]}),
]
result = grade_all(model_response, rules)

这种 Unix 哲学风格的规则组合（“做一件事做好、再组合”）让规则判分系统在 6-12 个月演化中仍然清晰。

5.6.19 一个规则判分常被忽略的工程层：grader 的端到端测试

评测代码自身要测（§5.6.12 单元测试），但还有一个更高层的工程动作——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:
        actual = my_grader(case["output"], expected=case["input"])
        assert actual == case["expected_grade"], f"Failed: {case['name']}"

这种”grader 自身的回归测试”通常会暴露出令人惊讶的 bug——某条 normalize 函数对全角空格不处理、某条 regex 在边缘情况下卡死、某条 schema 校验对 null 字段处理不当。

工业团队的实操：每个 grader 配 30-50 条端到端测试，每次改 grader 代码先跑这套测试。pre-commit hook 自动跑、确保任何改动不破坏 grader 行为。这种”双层测试”——单元测试 + 端到端测试——是评测体系的工程质量保证。

5.6.20 规则判分的”覆盖度评估”工程

写完规则判分后还需要回答一个问题——这套规则覆盖了多少业务场景？这个评估常被忽视。

具体方法：

1. 取 100 条真实生产 trace
2. 用现有规则判分跑一遍
3. 统计：
   • 规则判分能覆盖（pass 或 fail 都明确）的样例 %
   • 规则判分给出”未覆盖”或”无法判断”的样例 %

理想覆盖度 ≥ 80%。如果只有 50%，说明规则覆盖面太窄、需要扩充或回退到 LLM-judge。

修法：

• 覆盖度低：分析未覆盖样例的共性，加 5-10 条新规则
• 覆盖度高但失败率高：规则太宽松，加更严格断言
• 覆盖度高且失败率低：规则体系成熟，可日常使用

这种”覆盖度评估”是规则判分体系的”自我体检”。每月跑一次，能发现规则体系是否随业务演化保持适用。规则不更新会逐渐失去对生产的覆盖能力——这是评测体系工程化的隐藏维护任务。

5.6.21 规则判分的”3 层缓存”工程优化

最后讨论一个工程性能话题——规则判分的缓存优化。

规则判分本身极快（毫秒级），但在大规模评测中（10000+ 样例）仍可优化：

Layer 1: 进程内缓存
  - LRU cache: 同一 (input, expected) 跑过即缓存
  - 命中率: 80-90% (重复样例多)
Layer 2: 持久化缓存（Redis / SQLite）
  - 跨进程 / 跨运行复用
  - 命中率: 60-70%
Layer 3: hash-based 短路
  - 输出 hash 不变 → 不重跑判分
  - 命中率: 40-50%

3 层缓存组合让 10000 样例的规则判分总耗时从 10 秒压到 < 1 秒。对 CI 加速明显——开发者每改 prompt 后跑评测只等 1 秒、不是 10 秒。

工程意义：规则判分本身够快、但配合好的缓存能再快一个量级。这种”工程性能优化”虽然不显眼，但累积下来对开发者体验影响巨大——评测越快、用得越频繁，评测体系真正发挥作用。

5.6.22 规则判分的”长尾扩展”工程

规则判分体系成熟后，会进入一个长期工程节奏——长尾扩展。每周从生产 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.6.23 规则判分的”渐进式严格化”工程节奏

工业团队的规则判分体系不是一次性建立的，而是渐进式严格化。具体节奏：

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.6.24 一个跨章节的关联：规则 → judge → 元评测的递进

回顾本书第 5-8 章的整体结构，能看到一条清晰的递进：

• 规则判分（第 5 章）：能解决就解决（70% 场景）
• LLM-judge（第 6 章）：规则解决不了的（25% 场景）
• 人工（第 7 章）：LLM-judge 解决不了的（5% 场景）
• 元评测（第 8 章）：以上三者自身的可靠性

这条递进让读者理解评测方法学的”分工”——每种方法各有适用场景，组合起来覆盖完整问题空间。

工业实务：搭评测体系时按这个顺序投入：

1. 先 100% 投入规则判分（成本最低、效率最高）
2. 规则判分到天花板后投入 LLM-judge
3. LLM-judge 不够可靠时投入人工
4. 三者都跑起来后投入元评测验证

这种”按效率倒序”的投入路径，让团队的工程时间花在”刀刃上”——而不是一上来就上最复杂的方法。

5.6.25 规则判分的”代码可读性”细节

规则判分的代码看似简单，但长期可读性差异巨大。看两份等价代码：

不可读版本：

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

不可读版本工程师 6 个月后看自己的代码会问”这是什么意思”。可读版本任何人 3 秒能看懂。

可读性的工程红利：

• 维护成本：可读代码改 bug 时间是 1/5
• 新人 onboarding：可读代码不需要写文档
• 跨团队协作：可读代码避免误解
• review 效率：PR review 时可读代码 5 分钟搞定，不可读 30 分钟

工业实务：规则判分代码的可读性优先于行数。哪怕多 5 行代码、长期收益远超短期”省字符”。这是软件工程的老智慧在评测领域的具体应用。

5.6.26 一份完整的规则判分实战 cheatsheet

整合本章方法学，给一份”规则判分用什么方法”的速查 cheatsheet：

对照这份表，团队的规则判分需求 90% 能找到匹配方法。剩下 10% 需要混合 / 自定义——但本章方法学已经给了所有原料。

5.6.27 规则判分的”教学示范”价值

最后讨论规则判分的”教学示范”价值——它是新人入门评测体系的最佳起点。

为什么？

• 概念简单：5 分钟能理解”判分是什么”
• 代码可读：50 行 Python 能跑起来
• 结果直观：通过 / 失败一目了然
• 依赖少：不需要 LLM API / 复杂工具
• 错误友好：bug 容易定位

新人 onboarding 的标准路径：先用规则判分跑通完整评测流程 → 再升级到 LLM-judge → 最后接触元评测。这种”由简入深”的学习曲线让评测从”好像很难”变成”原来如此”。

工程团队的实务：把”用规则判分跑一次评测”作为新人 onboarding 的第一周任务。完成这一步后，新人对评测的认知会从抽象变具体——再学第 6-8 章的高级方法学就有了具象基础。

5.6.28 规则判分的”property-based testing”启示

借鉴软件工程的 property-based testing（如 QuickCheck / Hypothesis），规则判分也可以做”基于属性的测试”：

# 传统单元测试
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)

@given(st.text())
def test_grade_handles_whitespace(text):
    """加空格不影响 normalize 结果"""
    assert normalize(text) == normalize(f"  {text}  ")

property-based testing 用 Hypothesis 等工具自动生成测试输入，能发现单元测试想不到的边缘 case。对规则判分代码（特别是 normalize / regex）的鲁棒性验证特别有用。

工程实务：在 grader 仓库引入 property-based testing 框架（Python Hypothesis / JS fast-check）。每次 grader 改动除了跑常规单测、还跑 100-1000 次随机生成的输入。这种”严苛测试”让 grader 的 bug 在生产前就被暴露。

5.6.29 规则判分的”工程审美”

回顾本章方法学，规则判分体现的”工程审美”：

• 简洁优于复杂：能用 2 行 if 解决就不写 20 行
• 声明优于命令：能用 schema 就不用 if-else 链
• 单一职责优于全能：每个 grader 只做一件事
• 可读优于性能：性能足够时优先可读
• 测试优于希望：grader 必须配单测和 property test

这种审美超出规则判分领域，是所有”工具型代码”的共同标准。读完本章希望读者带走的不只是规则判分技术，更是这种”工程审美”。

审美比技术更难传授——它需要长期实践 + 不断 review + 偶尔的灵感闪现。但只要意识到这是评判工程质量的一个维度，工程师在编码时就会多一份对长期可维护性的敬畏。

5.6.30 规则判分给”工程基本功”的训练价值

规则判分看似简单，但它训练的工程基本功超出评测领域：

• 字符串处理：归一化 / regex / Unicode 处理
• 测试编写：单元测试 + property-based testing
• 错误处理：fail-fast vs graceful degradation
• 代码可读性：如何让 50 行代码长期可维护
• 性能优化：缓存 / 分层 / 提前 short-circuit

每条都是工程师职业发展的核心技能。规则判分代码相对简单，正好是练习这些基本功的好场景——比”调 LLM API”更能锻炼字符串处理 + 测试纪律。

工业实务：把”写规则判分”作为新人入门评测体系的第一周任务。完成后新人不只掌握了评测、还系统训练了几条工程基本功。这种”一举多得”的训练 ROI 极高。

读完本章希望读者带走的最深认知：规则判分是入门评测的最佳起点，也是练工程基本功的最佳场景。这种”教学价值”是评测体系建设的隐藏红利。

5.6.31 规则判分的”读完总结”

读完整章规则判分方法学，给读者一份”知识地图”：

规则判分的核心：
  ├── 4 大方法（Exact / Numeric / Regex / Schema）
  ├── 3 层缓存优化
  ├── 2 种测试纪律（unit + property-based）
  └── 1 个核心原则：能用规则就不用 LLM-judge

工程实践：
  ├── 归一化是基础
  ├── 可读性优于性能
  ├── 单一职责优于全能
  └── 长尾扩展知道何时停

天花板：
  └── 70-80% 场景规则能解决，剩下交给 LLM-judge / 人工

这份”知识地图”让读者整章方法学有个清晰的脉络。任何细节困惑时回到这张图能快速定位。

读完本章希望读者带走的最朴素心态：规则判分不是”低级方法”，是评测体系的高效起点。从规则判分起步、按需升级到 LLM-judge / 人工——这是工业评测的最务实路径。

5.6.32 规则判分的”工程语言学”细节

回顾 OpenAI evals 的 record_and_check_match 函数（参见 §9.2.3），它的 5 个参数命名背后有”工程语言学”考究：

def record_and_check_match(
    prompt: Any,           # 原始输入
    sampled: str,          # 模型实际输出
    expected: ...,         # 期望答案
    separator: ...,        # 分隔符判定函数
    options: ...,          # 候选选项
):

每个名字都精挑细选：

• sampled 而非 output / generated：强调”这是从模型分布中采样得到的”，隐含统计性
• expected 而非 correct / truth：避免”绝对真理”暗示，承认评测的相对性
• separator 而非 delimiter / boundary：明确语义是”分隔字段”而非”切分字符”
• options 而非 choices / candidates：与”多选题”语义贴合
• picked（返回值）而非 selected / chosen：动词时态精准（用过去式表示已选）

这种命名精度是工程师”语言洁癖”的体现——名字精确 = 语义清晰 = 长期可读。学会这种”工程语言学”是工程师从合格到优秀的关键。

读完本章希望读者带走的最深认知：规则判分的代码同样讲究”语言学”——名字、注释、错误信息——每一处都体现工程素养。

5.6.33 一份生产级规则判分类的完整实现

整合本章方法学，给一份”工业级规则判分类”的完整 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.6.13）"""
        opts = opts or {}
        text = text.strip()
        if opts.get("case_insensitive", True):
            text = text.lower()
        if opts.get("nfkc", True):
            text = unicodedata.normalize("NFKC", text)
        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

    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.6.20）"""
        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 行代码涵盖第 5 章所有方法学：

• 归一化（NFKC / case / 标点 / 空白）
• 多种 rule 类型（exact / contains / regex / schema）
• 缓存（避免重复判分）
• 诊断信息（每条规则失败时具体原因）
• 覆盖度报告

工业实务：把这份代码作为团队规则判分库的”基础类”。后续业务专属判分逻辑都继承 / 组合此类。这是规则判分体系工程化的标准基础设施。

5.6.34 一份完整的 Pydantic + OpenAI Structured Outputs 集成

整合本章 §5.6.11 Schema-First Design 方法学，给一份”Pydantic + OpenAI Structured Outputs”的完整工程示例：

# schema_first_eval.py
from pydantic import BaseModel, Field
from typing import Literal, Optional
from openai import OpenAI

class IntentResponse(BaseModel):
    """客服意图识别响应 schema"""
    intent: Literal["query", "book", "cancel", "complaint", "chitchat"]
    confidence: float = Field(ge=0, le=1)
    entities: dict[str, str] = Field(default_factory=dict)
    fallback_to_human: bool = False

    class Config:
        extra = "forbid"  # additionalProperties: false


client = OpenAI()

def classify_intent(user_query: str) -> IntentResponse:
    """OpenAI Structured Outputs - schema 在生成时强制"""
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        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}")
    for k, v in expected.get("entities", {}).items():
        if prediction.entities.get(k) != v:
            failures.append(f"entity {k}: {prediction.entities.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 先行 + 应用层 + 评测层共享”的范式作为团队的工程标准。半年下来减少 30-50% 的”格式校验失败”问题——因为它们在生成阶段就被拦在源头。

5.7 规则判分的天花板

诚实告诉读者规则判分的边界。规则判分绝对解决不了这些场景：

• 语义等价的不同表达：Schema 也不能区分 “巴黎位于法国” 和 “Paris is in France”
• 风格 / 礼貌程度：你想测”语气是否合适”，没有规则能描述
• 创造性输出：写诗、写代码注释、想标题——必须 LLM-judge 或人工
• 多步推理过程评估：Trajectory 是否合理、是否走了弯路

把这条边界画清楚很重要。规则判分是评测体系的”第一道防线”——快、便宜、可解释——但它不是终点。第 6 章接着讲 LLM-as-Judge 怎么补上规则判分覆盖不到的部分。

5.7.1 规则判分的”成本对照”——为什么必须先用规则

下表是基于 2026 年初公开 API 价格的工程成本对照（以 1000 条评测样本为基准）：

工业实务的”3:1 漏斗”经验：

• 70% 的样本应被规则筛掉（pass / fail 一目了然）
• 25% 的样本进 LLM-judge（语义判断）
• 5% 的样本进人工（争议 / 高风险）

这个比例下评测系统的总成本能降到”全 LLM-judge”方案的 1/4，总耗时降到 1/3。这就是为什么”规则判分是第一道防线”不是哲学口号，而是经济学硬约束。

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

记住这条经济学：在评测体系设计中，规则覆盖率每提升 10%，整体评测成本下降 8-12%。所以扩规则永远值得——直到达到边际成本递增的拐点（通常在 70-75% 覆盖率附近）。

5.7.2 规则判分的”国际化”陷阱清单

中国团队最常踩的规则判分坑是国际化适配——同一份”contains_check”的英文规则放到中文 / 日文 / 阿拉伯文 LLM 输出上往往失效。下面是一份基于多语言文本规范化的踩坑清单：

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:
        pattern = r"(?<![\p{L}\p{N}])" + regex.escape(needle) + r"(?![\p{L}\p{N}])"
        return regex.search(pattern, text, regex.UNICODE) is not None

flowchart 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 条原则：

1. 永远先 NFKC 归一化——把全角 / 半角 / 兼容字符都拍平
2. 正则用 regex 库而非 re——支持 \p{Nd}、\p{L}、Unicode property
3. 判长度按 token 不按 char——1000 字中文 ≈ 1500 字英文 ≈ 2500 字阿拉伯文
4. 对极端右到左语言（阿拉伯 / 希伯来）测一次反向 string——某些规则在 RTL 文本上会语义反转

工程实务：把 I18nNormalizingMatcher 作为规则判分库的”基础层”——所有上层规则都通过它做预处理。这是中国团队跨境业务、海外公司支持中文场景的”基本功”。

5.7.3 一份”规则判分覆盖率统计”工具：让团队知道还能扩多远

§5.7.1 提到”规则覆盖率每提升 10%，整体成本下降 8-12%“——但怎么知道当前覆盖率？工程团队往往凭体感”应该 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_candidates top 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

数字不大但稳——把这种”小钱”持续累积，半年后评测体系会”无感”地省下年化 $2-5k。

5.7.4 规则判分的”模糊匹配 + 阈值”工程实践

精确匹配（== / 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 噪声）"""
        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 条策略选择经验：

具体例子：

• 用户问”今日北京天气”，LLM 答 “13.5℃“，ideal “13.50摄氏度” → numeric_close ✅
• LLM 答”我不太清楚”，ideal “我不清楚” → edit_distance ratio 0.91 ✅
• LLM 答”AB123CD45”（OCR 错位），ideal “AB123-CD45” → composite OR 命中 ✅

工程实务陷阱：不要把所有评测题都用 fuzzy —— 严格场景（精算 / 法律 / 安全）必须 exact match，fuzzy 反而掩盖错误。判断标准：业务方说”用户能接受这种近似吗？“——若答案是”必须精确”，就别 fuzzy。

研究背景：FuzzyWuzzy（已并入 RapidFuzz）是 fuzzy 匹配工业标准库，本节 edit_distance 思路与之一致。Levenshtein 距离公式（Levenshtein 1965）是这类匹配的方法学起点。

把这套 matcher 纳入团队规则判分基础库，能解决 LLM 评测中”过度严格 → 假 negative 多”的常见问题。它是 §5.4 关键字匹配 + §5.5 schema 校验之外的”第三种规则武器”。

5.7.5 规则判分的”语义结构”层——超越字符匹配的工程模式

字符级匹配（exact / fuzzy）解决”形似”，但很多 LLM 输出需要语义结构正确——例如 SQL 查询、JSON 数据、数学表达式。下面给出几种高频”语义结构”判分模式：

import json
import ast
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}")

    def numeric_expression(self, expected: str,
                            actual: str) -> StructuralMatchResult:
        """数学表达式 eval 后数值等价"""
        try:
            e_val = eval(expected, {"__builtins__": {}}, {})
            a_val = eval(actual, {"__builtins__": {}}, {})
            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[eval 比较 with 容差]
  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 给 {"name": "Alice", "age": 30}，ideal {"age":30,"name":"Alice"} → json_equiv ✅
• LLM 给 SELECT name FROM users WHERE id = 1，ideal select name from users where id=1 → sql_equiv ✅
• LLM 给 2 + 3 * 4，ideal 14 → numeric_expression ✅（都 eval 为 14）
• LLM 给 [3, 1, 2]，ideal [1, 2, 3]，业务允许任意顺序 → list_set_equiv(order_matters=False) ✅

工程实务的 4 个细节：

1. eval 必须 sandbox：eval(..., {"__builtins__": {}}, {}) 防止注入
2. AST 比较忽略变量名时：用 ast.dump 默认含变量名，需要时可用 astor 库做”变量重命名归一化”
3. SQL 多表 join 顺序差异：复杂 SQL 需要更专业的解析器（如 sqlglot），sqlparse 在简单 case 够用
4. list 等价 + 元素 fuzzy：复合场景（list 内含字符串）要 list 长度等 + 逐元素 fuzzy

研究背景：

• HumanEval（Chen et al. arXiv:2107.03374）评测 Python 代码用 ast 等价 + 单元测试双判
• Spider / WikiSQL benchmark 评测 SQL 用 token-level 与 execution-equivalence 两种
• MATH 数据集评测数学解答用”符号表达式”等价（SymPy 库）

读者把 StructuralRuleMatcher 与 §5.7.4 FuzzyMatcher 拼接——不同问题用不同武器：

• 文本类 → fuzzy
• 结构化 → structural
• 完全严格 → exact

这是 LLM 评测中”判分到最后一公里”的工程艺术。

5.7.6 规则判分的”调度器”——根据 expected 自动选 matcher

§5.7.4 + §5.7.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 == "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 条调度器价值：

1. 判分写规则不再纠结：auto_match(expected, actual) 一行搞定 90% 场景
2. 特征分类可扩展：业务特定（如电话号码 / 中国身份证）添加专属分类
3. 保留可读性：matcher_used 字段告诉读者”为什么这条用了 fuzzy 不用 exact”
4. 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)。半年内规则判分代码量从 800 行减到 200 行。

3 类常见路由错误与修法：

研究背景：

• pytest 的 fixture 自动选择是这种”按签名 / 类型自动路由”的灵感源
• ragas 0.2.x 的 Metric.adapt 也用类似 type-aware 路由
• LangChain 的 OutputParser 自动选择策略（pydantic / json / regex）思路一致

把 AutoMatcherRouter 接入团队 evals 框架——评测代码量大降，新人 onboarding 速度快 2x。这是规则判分工程化的”压轴武器”——前面 §5.6-5.7.5 都在造武器，本节给出武器调度。

5.7.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 治理原则：

1. 每月跑 audit：cron 触发，结果发到 evals owner Slack
2. DEAD 规则直接删除：保留备份在 git，但当前代码移除
3. COMPLEX 规则按需拆：不是所有复杂规则都重构——只重构会被人改的那部分
4. debt_score > 0.7 触发 review：不必非要做什么，但必须有人看

具体例子：某团队 80 个规则文件 6 个月债务报告：

净效果：规则代码量从 3200 行降到 2300 行（-28%），单测覆盖率从 35% 升到 65%——半年下来代码可维护性大幅提升。

3 类 debt 来源：

研究背景：

• Sculley et al. 2015 “Hidden Technical Debt in ML Systems” 是 ML debt 概念的源头
• Google 内部”code health”团队季度跑 dead code analysis
• ratio of test-to-code 在工业实践通常 ≥ 1:1

读者把 RuleDebtAnalyzer 加入团队 monthly cron——规则代码自我减肥，让评测体系长期可维护。这是评测体系”长寿”的工程纪律。

5.7.8 一份”规则判分库”的最佳布局——所有评测代码该怎么组织

读完 §5.6 + §5.7 的所有规则判分工具，最后一个工程问题是”这些代码该怎么放？“——下面是工业团队最普遍的目录结构：

evals/
├── pyproject.toml                # 包定义
├── README.md                     # quickstart
│
├── core/                         # 核心抽象
│   ├── __init__.py
│   ├── matchers/                 # §5.7 的所有 matcher
│   │   ├── exact.py
│   │   ├── fuzzy.py              # §5.7.4
│   │   ├── structural.py         # §5.7.5
│   │   ├── i18n.py               # §5.7.2
│   │   └── router.py             # §5.7.6 自动调度
│   ├── rules/                    # 业务规则集
│   │   ├── __init__.py
│   │   ├── customer_service/
│   │   ├── medical/
│   │   └── ...
│   ├── coverage_analyzer.py      # §5.7.3
│   └── debt_analyzer.py          # §5.7.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 条目录设计原则：

1. core/ 与 runners/ 分开：matcher 不依赖 CLI，让 core 可被 import 到任何环境
2. rules/ 按业务域分：customer_service / medical / legal 各自一目录
3. scripts/ 是 cron 入口：让运维一眼看到”该跑什么”
4. test_to_code_ratio ≥ 1.0：评测代码的单测比例必须高于普通代码
5. docs/runbooks/ 写关键操作：如何回滚、如何 retire 旧 case 等

3 类常见反模式：

具体例子：某团队 18 个月演化的 evals 仓库：

洞察：18 个月稳定演化能从 F 到 A——这是工程纪律的回报。

研究背景：

• pytest 项目本身的目录结构是这套布局的源头
• Anthropic 的 evals 内部仓库（公开 Inspect 是其精简版）也用类似分层
• Python Packaging Authority (PyPA) 的 packaging guide 推荐相同模式

读者把这份目录结构作为”评测代码库 day 1 决策”——好的开局让 18 个月后不必大改。这是评测体系的”软件工程基本功”——再好的方法学没好的代码组织也撑不久。

5.7.9 一份”规则判分性能优化”工程实战——让 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()
        # ThreadPool 用于 IO bound 部分
        self.executor = ThreadPoolExecutor(max_workers=8)

    def _compile_patterns(self):
        """编译时预编译所有正则——节省 30-50% 时间"""
        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",
        )