Dark Dwarf Blog background

业务 LLM 评测设计

业务 LLM 评测设计

业务 Agent 的 LLM 评测要解决的不是“模型能不能答对某道题”,而是“我们能不能持续、可信地知道系统有没有变更好”。这包含下面的考虑的地方:

  1. 文本 vs 产物:最终回复是模型“说”出来的,可以被润色、改写、重排;真正流进下游系统的是结构化产物。
  2. LLM Judge 的便捷与不可靠:用另一个 LLM 打分很方便,但它会有偏差与漂移。
  3. 单点正确 vs 实际系统表现:一个 case 过了不等于整个系统变好了;两个均值的简单相减也常常掩盖真实的回归。

下面基于我个人的开发经验和一些相关的博客文章,整理了 Agent evaluation 中的一些设计。

1. 可信赖评测的通用架构

一个可信赖的业务 Agent 评测系统,可以抽象成六层:

层级核心问题原则
评测对象评什么?结构化 Artifact,不是聊天文本
判定权谁说了算?确定性规则持有 authoritative,LLM Judge 只做诊断
评分激励分数会诱导什么行为?奖励 abstain,惩罚 confident wrong
失败定位分数低了怎么办?每个失败带根因码,指向可修复模块
真值工程Ground truth 怎么管?Gold 数据集是版本化的生产系统
对比实验怎么证明改动更好?先冻结指纹,再跑配对差值和统计置信

a.a. 评测对象是 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”,而是把评测对象定义为系统真实交付的东西,而不是模型说的话。

b.b. 确定性规则持有 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 只能做需要理解上下文和证据链的诊断工作。

c.c. LLM Judge 与确定性评测隔离

即使把 LLM Judge 限制在诊断任务上,如果它和确定性评测共用数据流、生命周期或元数据,一次 Judge 超时、一次模型升级、一次 prompt 格式变化,都可能让原本稳定的指标出现“phantom regression”。因此需要在四个层面做隔离。

i.i. 数据流隔离

Judge 不应该重新读取原始 Artifact 并自行计算指标。它的输入应该是确定性 evaluator 已经算好的 diff 和报告,prompt 里明确禁止它重算或篡改精确比较的结果。这样 Judge 的职责就被收窄为“基于已有事实做语义诊断”,而不是“再判一次对错”。

确定性 diff 是客观事实证据,不是唯一总分;
不得重算或篡改 finding_id、severity、component 的精确比较。

ii.ii. 生命周期隔离

确定性报告必须先写入数据库,成为不可变记录;Judge 失败、超时或被重跑时,只能更新 judge_status 这个附属状态,不能回改已经落库的 metrics。这意味着即使 Judge 服务完全不可用,评测门禁仍然可以基于确定性结果运行,业务系统也不会被阻塞。具体的异步管线实现见下文。

iii.iii. 元数据隔离

观测平台上的两类分数必须分开标记。确定性分数来自规则比较,固定 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"
}

iv.iv. 故障隔离以及具体的 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。

d.d. 评分激励设计

生成 Artifact 的系统最怕的不是答不出来,而是模型找不到证据后自己编造。OpenAI 在 Why Language Models Hallucinate 里指出:主流 benchmark 把 abstention 当错误惩罚,会逼模型用自信的编造来优化分数。因为“猜错”和“不回答”都得 0 分,而“猜对”得 1 分,模型从期望收益上一定会选择编造。

所以评分要设计成不对称的:

  • 当某个槽位缺证据时,将其留空而不计入预测集合,因此不减 Precision;
  • Gold 侧如实记录 miss,让 Recall 反映漏报的真实代价;
  • unresolved_assignments 单独落库、单独追踪,不混入主指标;
  • 对 abstain 给予中性或正向激励,对 confident wrong 从严惩罚。

TruthRL 的 ternary reward 也是同一个思想:正确给正奖励、幻觉给负奖励、abstention 给中性奖励。应用到业务 Artifact 上,就是不要让“留空”和“填错”落到同一个分桶里。

e.e. 失败能定位到根因码

评测结果不能只告诉我们“结果如何”,还要告诉我们“问题出在哪里”。每个失败应该带一个可定位的根因码,指向具体系统层而不是笼统地写“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”。

f.f. 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
真值/Oracleoracle_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()}

g.g. 指纹冻结

单点评测回答“这次产出对不对”,对比实验回答“这次改动有没有让系统更好”。

“指纹先行”不是另一种指标,而是实验开始前的可比性检查。这里的 fingerprint 就是 ff 节 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. 指标设计的工程细节

a.a. 先写 Metric Contract,再写公式

一份结构化 Artifact 不是无序的字符串列表。每个指标都应先写清楚输入、匹配键和失败含义:

指标匹配键或比较方式能回答的问题
record_f1record_id 的 multiset需要的条目是否完整,是否多报?
assignment_f1(record_id, stage, occurrence)条目是否归属到正确阶段?
stage_f1阶段集合 + 顺序流程节点是否完整、顺序是否正确?
value_accuracyAssignment 对齐后的 Decimal 容差数值字段是否在业务允许误差内?
unit_accuracyAssignment 对齐后的规范化字符串单位或量纲是否一致?

举个例子。假设一份通用诊断报告的 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,这个回归会被平均掉。

b.b. 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。

c.c. 指标输出必须带 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 的形式实现、没有进行架构上的设计,先暂时记录这些吧。