业务 LLM 评测设计
业务 Agent 的 LLM 评测要解决的不是“模型能不能答对某道题”,而是“我们能不能持续、可信地知道系统有没有变更好”。这包含下面的考虑的地方:
- 文本 vs 产物:最终回复是模型“说”出来的,可以被润色、改写、重排;真正流进下游系统的是结构化产物。
- LLM Judge 的便捷与不可靠:用另一个 LLM 打分很方便,但它会有偏差与漂移。
- 单点正确 vs 实际系统表现:一个 case 过了不等于整个系统变好了;两个均值的简单相减也常常掩盖真实的回归。
下面基于我个人的开发经验和一些相关的博客文章,整理了 Agent evaluation 中的一些设计。
1. 可信赖评测的通用架构
一个可信赖的业务 Agent 评测系统,可以抽象成六层:
| 层级 | 核心问题 | 原则 |
|---|---|---|
| 评测对象 | 评什么? | 结构化 Artifact,不是聊天文本 |
| 判定权 | 谁说了算? | 确定性规则持有 authoritative,LLM Judge 只做诊断 |
| 评分激励 | 分数会诱导什么行为? | 奖励 abstain,惩罚 confident wrong |
| 失败定位 | 分数低了怎么办? | 每个失败带根因码,指向可修复模块 |
| 真值工程 | Ground truth 怎么管? | Gold 数据集是版本化的生产系统 |
| 对比实验 | 怎么证明改动更好? | 先冻结指纹,再跑配对差值和统计置信 |
评测对象是 Artifact,不是聊天文本
如果评测去解析 Markdown 或聊天文本,会立刻遇到四个问题:
- 文本重述会污染评分:同一份清单可以被 LLM 用多种方式“润色”,解析器对措辞变体极其脆弱。
- 错误无法定位到阶段:端到端文本里一个数字错了,无法分清是检索、分配、计算还是生成哪个具体阶段的问题。
- 评测无法异步重跑:聊天记录和产物版本如果没有绑定,之后会无法复现之前的评分。
- 分数无法回溯产物:业务人员看到“Record Recall 70%”时,必须能点回当时的产物,否则分数只是数字。
Hamel Husain 在 evals FAQ 里反复强调 error analysis 优先于指标:如果连“错在哪”都定位不了,指标只会掩盖真实的问题。把评测对象锚定在结构化产物上,是 error analysis 能被工程化的前提。
我们把这个产物叫做 Artifact——一份有明确 schema 的业务文档,例如:
{
"schema_id": "artifact.schema.v1",
"artifact_id": "art_01J4X...",
"produced_by": "workflow.finalise",
"checksum": "sha256:3f9a...",
"records": [
{
"id": "REC_A",
"stage": "stage_a",
"sequence": 1,
"value": "120.00",
"unit": "unit_x"
},
{
"id": "REC_B",
"stage": "stage_b",
"sequence": 2,
"value": "8.50",
"unit": "unit_y"
}
]
}
checksum 不只是校验字段,也是评测任务的幂等键。以 (task_id, artifact_checksum) 作为唯一约束,可以防止工作流重复提交或人工重跑时产生重复评分;同时它把分数和“被评产物的内容”绑定,而不是和可能变化的 artifact_id 绑定,这是分数可追溯的基础。
也不一定要是 JSON 文档,只要是一段可持久化、重放的结构化输出即可。例如在
nonoka-agent中的产物是工作区快照,记录created/modified/deleted、before_digest/after_digest和policy_violations:
appendRunEvidence({
schema_version: 1,
kind: "workspace_effect",
source: "nonoka-opencode-provider",
tool_call_id: toolCallId,
tool_name: toolName,
changed: workspaceChanged,
created,
modified,
deleted,
policy_violations: policyViolations,
before_digest: before.digest,
after_digest: after.digest,
});
因此“Artifact 优先”不只是“评测 JSON”,而是把评测对象定义为系统真实交付的东西,而不是模型说的话。
确定性规则持有 authoritative
业务 Artifact 里有大量可以被精确比较的东西:schema 是否合法、字段是否存在、ID 是否匹配、数值是否在容差内、单位是否归一化。这些判断应该由确定性规则完成,而不是交给 LLM Judge,原因是 LLM Judge 有下面的问题:
- 随机性:同一 prompt 两次运行可能给出不同分数,历史分数不可比。
- 版本漂移:如果换了一个本地小模型,之前的历史基准全部作废。
- 易被格式欺骗:本地模型对 JSON 格式敏感,格式对了但内容错了可能拿到高分。
- 无法进回归门禁:分数每次不一样,CI 无法稳定通过或失败。
Eugene Yan 在 LLM-Evaluators 里系统总结了 Judge 的偏差:self-enhancement、verbosity bias、position bias。在企业场景下,这些问题会被进一步放大。
因此原则是:凡是能用规则判断的,就不让 LLM 判断;LLM Judge 只能做需要理解上下文和证据链的诊断工作。
LLM Judge 与确定性评测隔离
即使把 LLM Judge 限制在诊断任务上,如果它和确定性评测共用数据流、生命周期或元数据,一次 Judge 超时、一次模型升级、一次 prompt 格式变化,都可能让原本稳定的指标出现“phantom regression”。因此需要在四个层面做隔离。
数据流隔离
Judge 不应该重新读取原始 Artifact 并自行计算指标。它的输入应该是确定性 evaluator 已经算好的 diff 和报告,prompt 里明确禁止它重算或篡改精确比较的结果。这样 Judge 的职责就被收窄为“基于已有事实做语义诊断”,而不是“再判一次对错”。
确定性 diff 是客观事实证据,不是唯一总分;
不得重算或篡改 finding_id、severity、component 的精确比较。
生命周期隔离
确定性报告必须先写入数据库,成为不可变记录;Judge 失败、超时或被重跑时,只能更新 judge_status 这个附属状态,不能回改已经落库的 metrics。这意味着即使 Judge 服务完全不可用,评测门禁仍然可以基于确定性结果运行,业务系统也不会被阻塞。具体的异步管线实现见下文。
元数据隔离
观测平台上的两类分数必须分开标记。确定性分数来自规则比较,固定 authoritative: true;Judge 分数来自模型判断,固定 authoritative: false。业务门禁、趋势图、告警规则只应消费 authoritative 分数,Judge 分数只用于辅助诊断和人工 review。
{
"metric_name": "record_f1",
"value": 0.72,
"authoritative": true,
"source": "deterministic_evaluator"
}
{
"metric_name": "evidence_safety",
"value": 0.85,
"authoritative": false,
"source": "llm_judge"
}
故障隔离以及具体的 Eval 管线设计
LLM Judge 运行超时或崩溃,不应该影响主报告和门禁。在实现时可以把 Judge 放到独立线程里执行,并包一层 wall-clock timeout;任何异常都被捕获为 judge_status = failed,而主 metrics 保持为 completed。
LLM Judge 在整个流程里的作用是解释为什么某个 finding 缺失、给“证据安全”打分、标记 prompt 泄漏风险或 unsupported claim——而不是决定“这个报告能不能过”。
可以用异步队列 + Worker 的设计来实现上面的这些 Eval 流程:EvalJob 描述任务状态,EvalStore 提供幂等认领和崩溃恢复,EvalWorker 先跑确定性评测、再异步跑 Judge:
from dataclasses import dataclass
from datetime import datetime
from typing import Literal
@dataclass
class EvalJob:
evaluation_id: str
trace_id: str
artifact_id: str
schema_id: str
status: Literal["pending", "running", "completed", "failed"]
attempt: int = 0
idempotency_key: str = "" # trace_id + evaluator + schema_version
judge_status: Literal["pending", "running", "completed", "failed", "disabled"] = "pending"
claimed_at: datetime | None = None
max_attempts: int = 3
class EvalWorker:
def __init__(self, store, evaluator, judge=None):
self.store = store
self.evaluator = evaluator
self.judge = judge
def start(self):
# 启动时回收崩溃前卡住的 running 任务
self.store.recover_running(stale_seconds=300)
self.store.recover_judge(stale_seconds=300)
while True:
job = self.store.claim_pending()
if job is None:
time.sleep(1)
continue
self._process(job)
def _process(self, job: EvalJob):
try:
artifact = fetch_artifact(job.artifact_id)
metrics, report = self.evaluator.evaluate(artifact)
# 1) 确定性指标先落库,这是 authoritative 的
self.store.complete_deterministic(
job.evaluation_id,
metrics=metrics,
report=report,
)
# 2) LLM Judge 异步补充,失败不影响主报告
if self.judge and judge_enabled_for(job.schema_id):
self._run_judge(job, artifact, report)
except RetryableError as exc:
# 网络抖动、锁冲突、artifact store 暂时不可用:保留 pending,指数退避
if job.attempt + 1 < job.max_attempts:
self.store.retry(
job.evaluation_id,
reason=str(exc),
backoff_seconds=2**job.attempt,
)
else:
self.store.fail(job.evaluation_id, reason=str(exc))
except PermanentError as exc:
# schema 不匹配、artifact 不存在:直接失败,不再重试
self.store.fail(job.evaluation_id, reason=str(exc))
def _run_judge(self, job: EvalJob, artifact, report):
self.store.mark_judge_running(job.evaluation_id)
try:
judge_result = self.judge.diagnose(artifact, report, timeout=120)
self.store.complete_judge(job.evaluation_id, judge_result)
except JudgeTimeout:
self.store.fail_judge(job.evaluation_id, reason="timeout")
except Exception as exc:
self.store.fail_judge(job.evaluation_id, reason=str(exc))
EvalStore 提供下面的原子操作来保证崩溃恢复和幂等:
| 操作 | 职责 |
|---|---|
claim_pending() | CAS 认领:只把 pending 任务标为 running,并记录 claimed_at |
recover_running(stale_seconds) | 把超时的 running 任务重置为 pending |
recover_judge(stale_seconds) | 把 judge_status='running' 但卡死的任务重置 |
complete_deterministic(...) | 确定性指标落库,这是 authoritative 的 |
retry(...) / fail(...) | 区分可重试错误与永久错误 |
这个管线同时实现了生命周期隔离和故障隔离:确定性指标先入库,Judge 在独立路径运行,任何一方的失败都不会污染 authoritative metrics。
评分激励设计
生成 Artifact 的系统最怕的不是答不出来,而是模型找不到证据后自己编造。OpenAI 在 Why Language Models Hallucinate 里指出:主流 benchmark 把 abstention 当错误惩罚,会逼模型用自信的编造来优化分数。因为“猜错”和“不回答”都得 0 分,而“猜对”得 1 分,模型从期望收益上一定会选择编造。
所以评分要设计成不对称的:
- 当某个槽位缺证据时,将其留空而不计入预测集合,因此不减 Precision;
- Gold 侧如实记录 miss,让 Recall 反映漏报的真实代价;
unresolved_assignments单独落库、单独追踪,不混入主指标;- 对 abstain 给予中性或正向激励,对 confident wrong 从严惩罚。
TruthRL 的 ternary reward 也是同一个思想:正确给正奖励、幻觉给负奖励、abstention 给中性奖励。应用到业务 Artifact 上,就是不要让“留空”和“填错”落到同一个分桶里。
失败能定位到根因码
评测结果不能只告诉我们“结果如何”,还要告诉我们“问题出在哪里”。每个失败应该带一个可定位的根因码,指向具体系统层而不是笼统地写“response not helpful”。一个简单的实现如下:
root_causes = []
for active, code, label in (
(has_unresolved_records, "unresolved_record", "存在未解析条目"),
(bool(missing_records), "missing_record", "缺少条目"),
(bool(extra_records), "extra_record", "多余条目"),
(bool(wrong_stage), "wrong_stage", "条目归属错误"),
(bool(unit_mismatch), "unit_mismatch", "单位错误"),
(bool(value_mismatches), "value_mismatch", "数值超出容差"),
(not stage_sequence_exact, "stage_sequence_mismatch", "阶段路线或顺序错误"),
):
if active:
root_causes.append({"code": code, "label": label})
这些码不是随便写的。Hamel Husain 的 error analysis 流程 建议:先对真实 trace 做 open coding,再聚类成 failure taxonomy,最后把每个分类变成 narrow、二元的评测标准。
Arize 对 Hamel 的访谈 也强调,好的分类标签要能直接对应到一个工程模块或产品规则,例如“called the write tool before receiving approval”,而不是“response not helpful”。
Gold 数据集是版本化的生产系统
Gold Dataset 不是把一份 CSV 放进 gold/ 目录就结束了。它是评测系统的另一半生产系统:要有来源、规范化、审核、版本和泄漏检查,才能让一次失败在下个月仍然代表同一个业务事实。
下面是整个 Gold Dataset 构建流程的简单代码示例。首先是简单的规范化:
import re
import unicodedata
def normalize_description(value: str) -> str:
text = unicodedata.normalize("NFKC", str(value or ""))
text = text.translate(str.maketrans({",": ",", "。": ".", "!": "!"}))
text = re.sub(r"\s+", " ", text).strip()
text = re.sub(r"[.!;]+$", "", text).rstrip()
return text.casefold()
归一化只处理全角标点、空格、句末标点和大小写,不做同义词扩展。句尾多一个标点可以视为同一描述,但两个业务含义不同的请求不能因为文字相似被合并。归一化时还要保留 historical_variant_count 和 ambiguity 标记,把一对多冲突交给人工审核,而不是静默选一条看起来最新的记录。
审核通过后的 case 需要被 manifest 锁定。这里可以抽象成一个通用的 Evaluation Manifest 模式:把所有会影响评测结果的因素放进固定 schema,做 deterministic 序列化,再求哈希得到 manifest_sha256。这个 hash 就是这次评测的“版本号”;hash 一变,旧报告不被覆盖,而是产生新的 dataset version。
通用 manifest 至少应包含四类字段:
| 类别 | 字段示例 |
|---|---|
| 数据集 | source_id, sha256, case_count, case_ids |
| 真值/Oracle | oracle_snapshot_id, oracle_sha256 |
| 评测代码 | evaluator_version, schema_version, validator_version |
| 策略/环境 | model, prompt_version, temperature, worktree_sha256, lockfile_sha256 |
import hashlib
import json
def build_manifest(manifest):
canonical = json.dumps(manifest, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
return {**manifest, "manifest_sha256": hashlib.sha256(canonical.encode()).hexdigest()}
指纹冻结
单点评测回答“这次产出对不对”,对比实验回答“这次改动有没有让系统更好”。
“指纹先行”不是另一种指标,而是实验开始前的可比性检查。这里的 fingerprint 就是 节 Evaluation Manifest 中 manifest_sha256 所锁定的内容——用来唯一标识“这次评测到底用了什么”。Baseline 和 Candidate 必须共享同一组指纹,这样才能进行之后的比较。下面是简单的在 CI 前进行对比实验的代码:
def paired_comparison(baseline, variant, bootstrap_samples=5000):
if baseline["dataset_sha256"] != variant["dataset_sha256"]:
raise ValueError("Replay reports must use the same dataset fingerprint")
if baseline["oracle_snapshot_id"] != variant["oracle_snapshot_id"]:
raise ValueError("Replay reports must use the same Oracle snapshot")
# 按 case 计算差值
deltas = {metric: [v - b for b, v in zip(baseline_values, variant_values)]
for metric, (baseline_values, variant_values) in ...}
# 固定种子 bootstrap
rng = random.Random(20260729)
ci = {metric: percentile([mean(rng.choices(deltas[metric], k=len(deltas[metric])))
for _ in range(bootstrap_samples)], [0.025, 0.975])
for metric in deltas}
2. 指标设计的工程细节
先写 Metric Contract,再写公式
一份结构化 Artifact 不是无序的字符串列表。每个指标都应先写清楚输入、匹配键和失败含义:
| 指标 | 匹配键或比较方式 | 能回答的问题 |
|---|---|---|
record_f1 | record_id 的 multiset | 需要的条目是否完整,是否多报? |
assignment_f1 | (record_id, stage, occurrence) | 条目是否归属到正确阶段? |
stage_f1 | 阶段集合 + 顺序 | 流程节点是否完整、顺序是否正确? |
value_accuracy | Assignment 对齐后的 Decimal 容差 | 数值字段是否在业务允许误差内? |
unit_accuracy | Assignment 对齐后的规范化字符串 | 单位或量纲是否一致? |
举个例子。假设一份通用诊断报告的 Gold 和预测分别是:
// Gold
[
{ "id": "F001", "stage": "discovery", "value": "120.00", "unit": "unit_x" },
{ "id": "F001", "stage": "validation", "value": "120.00", "unit": "unit_x" },
{ "id": "F002", "stage": "discovery", "value": "8.50", "unit": "unit_y" }
]
// Prediction
[
{ "id": "F001", "stage": "discovery", "value": "120.00", "unit": "unit_x" },
{ "id": "F002", "stage": "discovery", "value": "8.50", "unit": "unit_y" }
]
按 multiset 计算:TP=2、FP=0、FN=1,所以 Precision=1.0、Recall≈0.667、record_f1≈0.8。如果改用 set,Gold 和预测都会只剩 {F001, F002},record_f1 会被错误地算成 1.0,漏掉一次 F001 出现的事实就被掩盖了。
再假设预测把 F002 放进了 validation 阶段:
// Prediction
[
{ "id": "F001", "stage": "discovery", "value": "120.00", "unit": "unit_x" },
{ "id": "F002", "stage": "validation", "value": "8.50", "unit": "unit_y" }
]
此时 record_f1 仍然是 1.0,但 assignment_f1 会下降,因为它比较的是 (record_id, stage, occurrence)。这样就能立刻定位:检索和解析没问题,是归属阶段把条目放错了位置。
这张表比一个总分更重要。比如 record_f1 上升而 assignment_f1 下降,说明检索或解析可能变好了,但后续归属阶段把条目放错了位置;如果只看总 F1,这个回归会被平均掉。
multiset 配对与 wrong_stage 重分类
同一个条目可能在不同阶段或不同对象中出现多次,不能先转成 set。例如 Gold 是 [REC_A, REC_A, REC_B],预测是 [REC_A, REC_B, REC_C]。按 multiset 配对后,TP=2、FP=1、FN=1,所以 Precision = Recall = F1 = 0.667。若使用 set,Gold 和预测都会只剩一个 REC_A,漏掉一条记录也会被错误地判成正确。
阶段还要单独比较顺序。例如预测包含了所有阶段,但顺序从 [stage_a, stage_b, stage_c] 变成 [stage_a, stage_c, stage_b],集合 F1 可能仍为 1,sequence_exact 却必须为 0。
指标输出必须带 diff 和根因码
每个指标都返回数值、匹配总数和可定位 diff,而不是只返回一个浮点数。一个 case 可以有多个根因,但同一条事实只应在所属层计一次,避免总分被重复惩罚。例如下面的例子:
{
"evaluation_id": "eval_01J4X",
"artifact_id": "art_01J4X",
"summary": {
"record_f1": 0.8,
"assignment_f1": 0.67,
"stage_f1": 1.0,
"value_accuracy": 1.0,
"unit_accuracy": 1.0
},
"diff": {
"missing_records": [
{
"id": "F001",
"stage": "validation",
"value": "120.00",
"unit": "unit_x"
}
],
"wrong_stage": [
{
"gold": {
"id": "F002",
"stage": "discovery",
"value": "8.50",
"unit": "unit_y"
},
"predicted": {
"id": "F002",
"stage": "validation",
"value": "8.50",
"unit": "unit_y"
}
}
],
"value_mismatches": [],
"unit_mismatches": []
},
"root_causes": [
{ "code": "missing_record", "label": "缺少条目" },
{ "code": "wrong_stage", "label": "条目归属错误" }
]
}
当前在 Agent evaluation 相关设计就做了这些,一些项目中的 evaluation 实现主要也是以 adapter 的形式实现、没有进行架构上的设计,先暂时记录这些吧。