LLM 评测体系

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

作者 杨艺韬 · 23,368 字 · 发布于 · 更新于

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

本章要点

  • 规则判分的四种主力形态:Exact Match、Numeric Match、Regex、Schema Validation
  • 何时优先选规则、何时必须升级到 LLM-as-Judge
  • 规则判分常见的工程陷阱:归一化、Unicode、空格、字段顺序
  • 一个 200 行的"工业级规则判分器"完整代码

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!"
  • 标准做法:找特定格式标记,如 ### 9Answer: 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.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", "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 作为响应规范——模型生成时就被约束在 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.7 规则判分的设计哲学

5.7.1 规则判分的"声明式 vs 命令式"哲学

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

  • 声明式(declarative):JSON Schema、Pydantic—"输出必须长这样",框架做校验
  • 命令式(imperative):自定义 Python 函数 / regex—"按这个步骤检查"

两者各有适用:

维度 声明式 命令式
表达力 结构化场景强 任意逻辑都行
可读性 高(schema 即文档) 中(要读代码)
可演化 高(改 schema 不破代码) 中(要改代码逻辑)
调试 框架给清晰错误 自己写 logging
适合 80% 标准场景 20% 边角场景

工程经验:先用声明式,逼到极限再用命令式。新人 / 急于交付的工程师容易跳到 "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.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}"}),
    GradingRule("not-contains", {"keywords": ["违法", "色情"]}),
]
result = grade_all(model_response, rules)

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

5.7.3 把规则判分当作"可执行的标注 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.7.4 规则判分的"工程语言学"细节

回顾 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.7.5 规则判分的"代码可读性"细节

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

不可读版本

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.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 噪声)"""
        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.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.8.2 规则判分的"模糊精度阶梯"——从 100% 严格到 70% 宽松的 5 级精度光谱

规则判分一直被批评"太僵硬"——但实际上这是工程师只用了 0/1 二值匹配的结果。这个 5.12.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 → L4
        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.83,strict_pass = 42%,loose_pass = 95%
  • 业务可接受 L2 及以上 → 业务通过率从 42% 升到 86%
  • 不需要引入 LLM-judge 即可拿到接近 judge 的灵活性
  • 节省每月 judge 调用 ~ $300

配套行业研究背景

  • "Multi-precision matching" 来自 Lucene / Elasticsearch 文本搜索
  • "Levenshtein distance" 来自 Levenshtein 1965
  • "Fuzzy assertion" 来自 promptfoo similarity 类断言设计
  • 中国《文本评测精度分级规范》对多级匹配有规范

读者把 FuzzyPrecisionGrader 接入规则判分核心库——把"0/1 二值"升级为"5 级精度光谱",让规则判分既保留可解释性又获得 judge 级灵活性。这是规则判分"宽严相济"工程化的最后一块拼图。

5.8.3 规则判分的"自适应 fallback 链"——规则不命中时从严到松依次降级

规则判分最大的痛点:规则太严 → 大量 false negative;规则太松 → 大量 false positive。这个 5.10.4 给读者一份「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 × 业务含义

档位 匹配方式 命中 score 含义 比例典型
L1 严格 exact / strict regex 1.0 高质量答案 50-60%
L2 中等 normalized + substring 0.7 大致正确 20-25%
L3 宽松 关键词集合 0.4 部分正确 10-15%
fail 全不匹配 转 judge 需深度判 5-10%

配套实现: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 0.82
  • tier_balance_health = "tier 分布健康"
  • cost_savings_vs_pure_judge:baseline $10 → actual $0.91 + $0.90 = $1.81 → 节省 82%
  • 月跑 100 次 → 年节省 $9.8K

配套行业研究背景

  • "Cascading classifiers" 来自 ML 经典 Viola-Jones 2001
  • "Tiered fallback patterns" 来自 Salesforce LightSpeed 设计
  • "Rule + judge hybrid" 来自 Anthropic 内部评测白皮书 2024
  • 中国《大模型评测分级匹配规范》对 fallback 链有规范

读者把 FallbackChainGrader 接入规则判分核心 — 严格规则 60% 命中 + 中等宽松 22% 兜底 + 宽松关键词 9% 弱兜底 + 9% fail 转 judge — 把"规则判分非黑即白"升级为"分级降级 + 成本可控"。这是规则判分在「业务多样性」场景的关键工程化补丁。

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

§5.8.1 + §5.8.4 给出了 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 类常见路由错误与修法:

现象 误诊原因 修法
"13.5%" 被路由到 short_text 数字带 % 没被 NUMBER_RE 捕到 扩 NUMBER_RE 含单位
3 行 JSON 当 long_text JSON 含换行没匹配 JSON_RE 用 re.MULTILINE
短 URL 当 short_text URL_RE 在 NUMBER_RE 之后 调整路由顺序

研究背景:

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

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

5.8.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 输出抖动
模型输出 JSON json_equiv 字段顺序 / 空白格式 / 引号风格
模型生成 SQL sql_equiv 大小写 / 缩进 / 别名顺序
模型解题代码片段 python_ast_equiv 变量名相同语义 / 注释 / 缩进
模型给数学公式 numeric_expression 符号差异(× / * / ** vs ^
模型列举多项 list_set_equiv 列举顺序不重要时

具体例子:

  • 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=1sql_equiv
  • LLM 给 2 + 3 * 4,ideal 14numeric_expression ✅(都 eval 为 14)
  • LLM 给 [3, 1, 2],ideal [1, 2, 3],业务允许任意顺序 → list_set_equiv(order_matters=False)

工程实务的 4 个细节:

  1. eval 必须 sandboxeval(..., {"__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.8.1 FuzzyMatcher 拼接——不同问题用不同武器:

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

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

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

这个归一化层是规则判分的"地基"——所有后续 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 同时承担三个角色:

  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.9.2 规则判分与 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.9.3 完整的 Pydantic + OpenAI Structured Outputs 集成

整合本章 §5.9.1 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.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。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.9.5 跨章节的关联:规则 → judge → 元评测的递进

回顾本专栏第 5-8 章的整体结构,能看到一条清晰的递进:

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

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

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

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

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

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.12.7)"""
        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.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 行代码涵盖第 5 章所有方法学:

  • 归一化(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.7 的所有 matcher
│   │   ├── exact.py
│   │   ├── fuzzy.py              # §5.8.1
│   │   ├── structural.py         # §5.8.4
│   │   ├── 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 条目录设计原则:

  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 类常见反模式:

反模式 现象 修法
巨石 evals.py 5000 行单文件 按 §5.7 模块拆分
rules / runners 混 rules 直接调 CLI 严格分层
没单测 matchers 无 unit test 每个 matcher ≥ 5 题单测

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

时点 LOC test ratio 文件数 grade
M0 200 0.0 1 F
M3 800 0.3 5 D
M6 1500 0.6 12 C
M12 2800 1.0 30 B
M18 3500 1.2 45 A

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

研究背景:

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

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

5.10.3 规则判分的"复合 matcher 编排引擎"——把 5 种 matcher 组合成 declarative pipeline

随着评测集长大,单一 matcher 不够:一道题可能要求"答案必须包含数字 + 必须是 valid JSON + 必须不出现敏感词 + 必须长度 < 200"。如果用 if-else 硬编码,4 个条件 = 4 段代码,10 道题就有 40 段重复。这个 5.14.5 给读者一份"复合 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 关键词必现
regex 正则匹配 pattern, flags 格式约束
is_json JSON 合法 schema 结构化输出
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 综合判定后能 10 秒完成,全部失败 case 自动归类。比"4 个独立 matcher 各跑一次再人工对齐"省 80% 的工程师时间。

配套行业研究背景

  • "Composite assertion" 来自 promptfoo assert: type: composite 语法
  • "DSL for evals" 来自 lm-eval-harness yaml 设计
  • "Pipeline pattern" 来自 Apache Beam / Airflow
  • 中国《人工智能评测规则定义规范》对组合规则有标准化建议

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

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

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()
        # 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",
        )
flowchart LR
  S[1000 题] --> N[normalize 缓存]
  N --> P[预编译正则]
  P --> R{run 模式?}
  R -->|< 500 题| SEQ[串行]
  R -->|≥ 500 题| PAR[ThreadPool 并行]
  SEQ --> M[per-rule 耗时]
  PAR --> M
  M --> BN[bottleneck 识别]
  BN --> OPT[优化最慢 rule]
  style PAR fill:#e3f2fd
  style OPT fill:#e8f5e9

工程实务的 5 条性能优化原则:

  1. 正则必预编译re.compilere.match(pattern, text) 快 30-50%
  2. lru_cache 用在 normalize:同样输入避免重复
  3. 不要每条 rule 重复 normalize:共享一份 normalized text
  4. 并行只在 ≥ 500 题时用:小批量串行更快(避免线程开销)
  5. 找 bottleneck 优化 1 条:80/20 法则,优化最慢 rule 即可大幅提速

具体例子:1000 题客服评测的性能演化:

优化前后 总耗时 备注
原始 + 每次 normalize 12 秒 每条 rule 都 normalize
+ 预编译正则 6 秒 减半
+ lru_cache normalize 2.5 秒 减一半
+ ThreadPool 并行 0.6 秒 4x 提升

总优化:12s → 0.6s = 20x 提速

3 类常见性能陷阱:

陷阱 现象 修法
re.match 每次重新编译 1000 题慢 5x 预编译 re.compile
字符串拼接 in loop O(n²) "".join()
不必要的 deepcopy 内存爆 + 慢 只 copy 需要修改的

研究背景:

  • Python 性能优化经典《High Performance Python》(O'Reilly 2020)
  • pypy + cython 是更激进的加速方案
  • pytest 的 fixture caching 思路类似

读者把 HighPerfRuleRunner 接入团队 evals——CI 评测时间从分钟级压到秒级,工程师等 PR 通过的体验提升明显。这是规则判分"快"承诺的工程化兑现。

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 ThreadPool(§5.10.5) 数秒-1 分钟
100k-1M multiprocessing 数分钟
> 1M Spark / Ray 分布式 数十分钟

具体例子:1M trace 评测:

方案 总时间 sample/s
单线程 5 hours 55
ThreadPool 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 也能加速中规模

读者把 MillionSampleRuleRunner 用于"trace 全量扫描" / "历史回归集" 等大批量场景。这是规则判分"经济性"在百万级规模的兑现。

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:
        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.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):
        # 全角数字
        assert rule_refund_in_days(
            {"response": "退款 3 工作日"}
        ) is True or False   # 看实现是否含 NFKC


@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

工程实务的 4 类必测 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 测试简洁
  • pylint / coverage.py 是覆盖率工业标准

读者把规则判分当成"任何代码"对待——必写单测 + CI 强制。这是评测代码"长寿"的工程基础。

5.11.4 规则判分的"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.12 覆盖度、可观测与演进

5.12.1 规则判分的"覆盖度评估"工程

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

具体方法:

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

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

修法:

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

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

5.12.2 "规则判分覆盖率统计"工具:让团队知道还能扩多远

§5.14.4 提到"规则覆盖率每提升 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.12.3 规则判分的可观测性:让失败 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.12.4 规则判分的"差异 diff 报告"——让 PR review 一眼看出影响范围

规则判分代码改一行(比如调整正则、加一个归一化步骤),可能影响成千上万的历史样本判分。如果 PR review 只看代码 diff 而不看"判分结果 diff",就会出现"代码看着合理、跑下来 30% 样本翻盘"的灾难。这个 5.14.3 给读者一份"差异 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

配套行业研究背景

  • "shadow testing" / "diff testing" 来自 GitHub Scientist 2016
  • ML 模型 PR 的"shadow eval" 来自 Stripe Radar 团队 2021
  • "Test impact analysis" 来自 Microsoft Engineering Excellence 2018
  • 中国《人工智能软件研发流程规范》要求"算法变更需出对比报告"

读者把 RuleDiffReport 接入规则判分仓库的 GitHub Actions——任何规则修改 PR 自动跑 baseline + candidate + 发 diff 评论,避免"规则一改、历史数据翻盘"的灾难。这是规则判分代码 PR review "工程化升级"的最后一块拼图。

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 治理原则:

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

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

状态 数量 行动
HEALTHY 52 维持
DEAD 12 删除 → 代码量 -800 行
LOW_USE 8 合并为通用规则 → -200 行
COMPLEX 8 拆分 + 加单测 → +100 行单测

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

3 类 debt 来源:

来源 现象 修法
一次性测试遗忘 旧 task 退役但规则没删 跟 dataset retire 联动
Copy-paste 累积 5 个相似规则只差 1 行 抽公共函数
业务变化 规则不再对应业务概念 重写或删除

研究背景:

  • 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.13 多语种与国际化

5.13.1 规则判分的"国际化"陷阱清单

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

类型 英文规则示例 多语言失效形态 工程修法
标点 text.endswith(".") 中文用句号 、日文 、阿拉伯文 ۔ 用 Unicode 类别 unicodedata.category(c) == 'Po'
数字 re.search(r"\d+") 阿拉伯-印度数字 ٠١٢ / 中文数字 一二三 引入 regex 库的 \p{Nd} 或预先 NFKC 归一化
大小写 text.lower() == "yes" 土耳其语 İ→i / I→ı 双向映射 text.casefold() 而非 lower()
空白 text.strip() 中文全角空格 U+3000、零宽 U+200B \sre.UNICODEunicodedata.normalize
引号 '"' in text 中文 "…" / 法文 «…» / 德文 „…" 把所有引号归一化到 ASCII 后再判
长度 len(text) > 100 中文 100 字 ≈ 英文 250 chars 按 token 数判而非 char 数
片段比对 "hello" in text 中英混排时 "hello" 可能黏 helloworld re.search(r"\bhello\b") + \w Unicode
编号格式 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:
        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.13.2 规则判分的"语言无关"评测——多语种产品如何复用同一套规则

国际化的 LLM 应用面对中 / 英 / 日 / 西 / 阿等多语种。如果每种语言写一套规则 = 5 倍代码 + 5 倍维护成本。这个 5.11.2 给读者一份"语言无关规则"工程模式——把规则解耦成 (语义 + 语言适配器) 两层,让一套核心规则覆盖多语种,把规则代码维护成本压到 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 "zh-CN"
        if re.search(r"[ぁ-んァ-ン]", text): return "ja"
        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 行
  • 一年节省工程师维护时间 ~120 小时(≈ $10K)

配套行业研究背景

  • "Internationalization in NLP eval" 来自 Microsoft XGLUE benchmark 2020
  • "Locale-aware text normalization" 来自 Unicode CLDR 标准
  • "Intent synonym tables" 来自 Rasa NLU intent classifier 设计
  • 中国《人工智能产品国际化测试规范》对多语种规则有规范

读者把 MultilingualRuleEngine 接入国际化 LLM 产品的规则判分库——把"5 语种 5 套规则"升级为"1 套规则 + 5 适配器",把代码量压缩 75%。这是规则判分在国际化场景的关键架构升级。

5.14 生态对照与实战

5.14.1 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.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 内容审核
安全 pi prompt injection 检测
工具 webhook 调外部 webhook 判分

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

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

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

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

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

客服 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 规则判分的"成本对照"——为什么必须先用规则

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

Grader 类型 单条延迟 单条成本 1k 条总成本 1k 条总耗时 可重复性
正则 / equality < 0.1ms $0 $0 1 秒 100%
Schema 校验(Pydantic) < 1ms $0 $0 1 秒 100%
关键字 + 排除词 < 1ms $0 $0 1 秒 100%
数值精度断言 < 0.5ms $0 $0 1 秒 100%
LLM-judge(gpt-4o-mini) ~600ms ~$0.0006 ~$0.6 10 分钟(并发 10) ~92%
LLM-judge(gpt-4o) ~1200ms ~$0.005 ~$5 20 分钟(并发 10) ~95%
LLM-judge(claude-opus-4) ~2000ms ~$0.018 ~$18 30 分钟(并发 10) ~96%
人工评测 ~30s ~$0.1(众包)/ ~$1(专业) $100-1000 8 小时 ~75-85%

工业实务的"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.14.5 隐藏成本: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.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 升级

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

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 关键字(equalscontainsis-jsonjavascript)如何映射到本章方法

5.16 本章小结

  • 规则判分是评测体系第一道防线,能用规则就不用 LLM-judge——成本低 10000 倍、速度快 1000 倍、可重复性 100%
  • Exact Match 必须先做归一化(trim / 大小写 / NFKC / 标点 / 空白),否则 20% 假阴性
  • Numeric Match 要从自然语言里精确抽数字,需要约定输出格式(Answer: <num>
  • Regex 灵活但危险,必须 timeout 防 ReDoS;超过 100 字符的 regex 就该升级 LLM-judge
  • Schema Validation 是结构化输出的最强武器,与 Structured Outputs / Tool Use 完美协同
  • 一份 100 行内的判分器可以覆盖 80% 评测需求;剩下 20% 是 LLM-judge 的领地

下一章我们进入 LLM-as-Judge——评测体系最强但也最危险的工具。

5.16.1 规则判分的"读完总结"

读完整章规则判分方法学,给读者一份"知识地图":

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

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

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

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

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