Agent 的评估、测试与可观测性：上线前后都要看得见

Agent 的评估、测试与可观测性：上线前后都要看得见

Agent 的输出具有不确定性：同一个输入，两次运行的路径可能不同。传统的“写几个单元测试就放心”在这里行不通。本文介绍如何给 Agent 建立 评估（Evaluation） 与 可观测性（Observability） 体系：怎样拆解评估维度、构建评测集、选择打分方式、记录 Trace，并顺带讨论成本与延迟的护栏。文中给出 Python 的回归评测与 Trace 记录代码、Java 的埋点示例，以及常见坑和检查清单。

一、背景：为什么 Agent 特别需要评估和观测

传统软件的行为大多是确定的：相同输入得到相同输出，单元测试通过就有较强的信心。Agent 有三点不同：

1. 非确定性：模型本身有随机性，同一任务可能走不同路径，甚至有时成功有时失败；

2. 多步累积：每一步都可能出错，错误会沿着链路传递，最终结果的失败往往要追溯到前面某一步；

3. 外部依赖多：模型服务、工具接口、检索数据任何一个变化，都可能改变整体表现。

因此“改了一句提示词，感觉变好了”这种判断很不可靠。你需要两样东西：评估——用一批固定样本衡量改动前后的变化；可观测性——线上出问题时，能回放“它当时到底做了什么”。

二、评估什么

不要只看“最终答案对不对”，建议拆成三层：

• 结果层：任务是否完成？答案是否正确、完整？

• 过程层：工具选择是否合理？参数是否正确？有没有多余或重复的调用？

• 效率层：用了多少步、多少 token、多长时间、花了多少钱？

三层分开看的价值在于定位问题。比如最终答案正确，但走了 15 步（本来 3 步就够）——结果层合格，效率层不合格，说明有优化空间；再如最终答案错误，但过程层显示工具选择与参数都正确——问题可能出在检索数据或最后的总结环节。

三、构建评测集

评测集是 Agent 开发中最有价值的资产之一。

1. 从真实场景收集：日志里的真实用户请求，比你自己想象的更有代表性。

2. 覆盖三类样本：典型场景、边界情况、已知的失败案例。

3. 每条样本写明期望：可以是标准答案、必须调用的工具、或必须满足的条件。

4. 小而精起步：30~50 条高质量样本，好过 1000 条随意拼凑的。

5. 版本化管理：每次修改提示词或工具，都用同一批样本回归。

一条样本可以这样表示：

{
  "id": "case-017",
  "input": "帮我查一下明天杭州的天气，并告诉我要不要带伞",
  "expect": {
    "must_call": ["get_weather"],
    "args_contain": {"city": "杭州"},
    "answer_should": ["明确给出是否带伞的建议"]
  }
}

样本的分类建议

四、几种打分方式

建议组合使用：能用规则断言的先用规则，开放式部分再用 LLM 评审，并定期人工抽查评审结果。

使用“LLM 作为评审”时有几个注意点：评审提示词要给出明确的评分标准和示例；让评审模型先写理由再给分；尽量与被评测的模型分开，避免自我偏好；对同一批样本，用人工标注的一小部分结果检验评审模型的一致性。

五、一个可用的回归评测脚本

因为输出不确定，同一条样本最好多次运行并统计通过率，而不是只看一次结果。下面的脚本包含：多次运行、规则断言、步数与 token 统计，以及与基线对比。

import json, statistics, time
from dataclasses import dataclass

@dataclass
class Trace:
    steps: list            # 每步：{"tool": str|None, "args": dict, "output": str, "tokens": int}
    final_answer: str
    total_tokens: int
    elapsed_s: float

def check(trace: Trace, expect: dict) -> tuple[bool, list[str]]:
    """规则断言：返回 (是否通过, 失败原因列表)"""
    reasons = []
    called = [s["tool"] for s in trace.steps if s.get("tool")]
    for t in expect.get("must_call", []):
        if t not in called:
            reasons.append(f"应调用 {t} 但未调用")
    for t in expect.get("must_not_call", []):
        if t in called:
            reasons.append(f"不应调用 {t}")
    for k, v in expect.get("args_contain", {}).items():
        if not any(str(v) in json.dumps(s["args"], ensure_ascii=False)
                   for s in trace.steps if s.get("tool")):
            reasons.append(f"参数中未包含 {k}={v}")
    for kw in expect.get("answer_contains", []):
        if kw not in trace.final_answer:
            reasons.append(f"答案缺少关键词：{kw}")
    if len(trace.steps) > expect.get("max_steps", 10):
        reasons.append(f"步数 {len(trace.steps)} 超过上限")
    return (not reasons), reasons

def evaluate(agent, cases: list[dict], runs: int = 3) -> list[dict]:
    report = []
    for case in cases:
        results, steps, tokens, times, fail_reasons = [], [], [], [], []
        for _ in range(runs):
            t0 = time.time()
            trace: Trace = agent.run(case["input"])      # 需返回上面的 Trace 结构
            ok, reasons = check(trace, case["expect"])
            results.append(ok)
            steps.append(len(trace.steps))
            tokens.append(trace.total_tokens)
            times.append(time.time() - t0)
            fail_reasons += reasons
        report.append({
            "id": case["id"],
            "pass_rate": sum(results) / runs,
            "avg_steps": statistics.mean(steps),
            "avg_tokens": statistics.mean(tokens),
            "p_max_latency_s": max(times),
            "top_fail_reasons": sorted(set(fail_reasons))[:3],
        })
    return report

def compare(baseline: list[dict], current: list[dict], drop_tolerance: float = 0.1):
    """与上一版本对比，找出通过率明显下降的样本"""
    base = {r["id"]: r for r in baseline}
    regress = [r["id"] for r in current
               if r["id"] in base and base[r["id"]]["pass_rate"] - r["pass_rate"] > drop_tolerance]
    return regress      # 若非空，阻止发布并人工查看

注意 runs=3 与 drop_tolerance=0.1 只是示例值。样本少、运行次数少时，通过率的波动本身就很大，不要把一两次的差异当成“显著变化”。

六、可观测性：记录 Trace

线上出了问题，你需要能回放“它当时到底做了什么”。建议至少记录：

• 每次任务的唯一 trace id；

• 每一步的 模型输入/输出、工具名、参数、返回值、耗时；

• token 用量与费用；

• 错误、重试、超时、触发的人工确认；

• 用户反馈（点赞/点踩）关联到对应 trace。

注意对日志中的敏感信息脱敏，并设置保存期限。

Trace 的结构：一条任务、多个 Span

Trace（一次任务，trace_id = a1b2c3）
 ├─ Span 1  llm_call      模型决策           耗时 1.8s  tokens 820
 ├─ Span 2  tool_call     get_weather        耗时 0.4s  参数 {"city":"杭州"}
 ├─ Span 3  llm_call      模型总结           耗时 1.2s  tokens 640
 └─ 结果：success，总耗时 3.4s，总 token 1460

Python：一个轻量的 Trace 记录器

import json, time, uuid, contextlib
from contextvars import ContextVar

_current_trace: ContextVar[dict | None] = ContextVar("trace", default=None)

SENSITIVE_KEYS = {"api_key", "password", "token", "authorization"}

def redact(obj):
    """对常见敏感字段脱敏，避免凭证进入日志"""
    if isinstance(obj, dict):
        return {k: ("***" if k.lower() in SENSITIVE_KEYS else redact(v)) for k, v in obj.items()}
    if isinstance(obj, list):
        return [redact(x) for x in obj]
    return obj

@contextlib.contextmanager
def trace_task(user_id: str):
    trace = {"trace_id": uuid.uuid4().hex, "user_id": user_id,
             "start": time.time(), "spans": [], "status": "running"}
    token = _current_trace.set(trace)
    try:
        yield trace
        trace["status"] = "success"
    except Exception as e:
        trace["status"] = "error"
        trace["error"] = f"{type(e).__name__}: {e}"
        raise
    finally:
        trace["elapsed_s"] = round(time.time() - trace["start"], 3)
        print(json.dumps(redact(trace), ensure_ascii=False))   # 生产中写入日志系统/存储
        _current_trace.reset(token)

@contextlib.contextmanager
def span(kind: str, name: str, **attrs):
    t0 = time.time()
    record = {"kind": kind, "name": name, "attrs": attrs}
    try:
        yield record
    except Exception as e:
        record["error"] = f"{type(e).__name__}: {e}"
        raise
    finally:
        record["elapsed_s"] = round(time.time() - t0, 3)
        trace = _current_trace.get()
        if trace is not None:
            trace["spans"].append(record)

# 用法示例
with trace_task(user_id="tmpu1"):
    with span("tool_call", "get_weather", args={"city": "杭州"}) as s:
        s["output"] = get_weather(city="杭州")

Java：用 Micrometer 记录指标

在 Spring 项目里，可以用 Micrometer 记录调用次数、耗时和 token 用量，接入已有的监控系统：

import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.springframework.stereotype.Component;

@Component
public class AgentMetrics {
    private final MeterRegistry registry;

    public AgentMetrics(MeterRegistry registry) {
        this.registry = registry;
    }

    /** 记录一次工具调用：按工具名和结果打标签，便于按维度聚合 */
    public <T> T timeTool(String toolName, java.util.concurrent.Callable<T> call) throws Exception {
        Timer.Sample sample = Timer.start(registry);
        String outcome = "success";
        try {
            return call.call();
        } catch (Exception e) {
            outcome = "error";
            throw e;
        } finally {
            sample.stop(Timer.builder("agent.tool.duration")
                    .tag("tool", toolName)
                    .tag("outcome", outcome)
                    .register(registry));
        }
    }

    /** 记录 token 用量（区分输入与输出） */
    public void recordTokens(String model, long inputTokens, long outputTokens) {
        registry.counter("agent.llm.tokens", "model", model, "type", "input").increment(inputTokens);
        registry.counter("agent.llm.tokens", "model", model, "type", "output").increment(outputTokens);
    }
}

需要提醒：指标的标签（tag）取值要有限，不要把 user id、trace id 这类高基数值作为标签，否则会让监控系统的存储膨胀。这类信息应放在日志或 Trace 系统中。

七、成本与延迟控制

• 设置 最大步数、最大 token、最大耗时，超限时优雅终止。

• 简单任务用小模型，复杂推理才用大模型（路由策略）。

• 缓存重复的检索和工具结果。

• 压缩上下文：摘要历史、裁剪工具输出。

• 对相互独立的工具调用做并行。

• 按用户/租户设置额度和限流。

一个简单的预算护栏：

class Budget:
    def __init__(self, max_steps=10, max_tokens=30_000, max_seconds=60):
        self.max_steps, self.max_tokens, self.max_seconds = max_steps, max_tokens, max_seconds
        self.steps = self.tokens = 0
        self.start = time.time()

    def charge(self, tokens: int):
        self.steps += 1
        self.tokens += tokens

    def exceeded(self) -> str | None:
        if self.steps >= self.max_steps:
            return "步数已达上限"
        if self.tokens >= self.max_tokens:
            return "token 已达上限"
        if time.time() - self.start >= self.max_seconds:
            return "耗时已达上限"
        return None    # 未超限

# 在 Agent 循环中：每步结束后 charge，开始下一步前检查 exceeded()，
# 超限则终止并向用户说明“已完成到哪一步、为什么停止”，而不是悄悄返回残缺结果。

八、上线后的闭环

1. 线上监控关键指标：成功率、平均步数、费用、延迟、错误率。

2. 定期抽查失败与低分 trace。

3. 把新发现的失败案例补进评测集。

4. 改动后先跑回归，再灰度发布。

没有评测集，所有的“优化”都只是感觉；没有 trace，所有的“排查”都只是猜测。

一个排查故事：从一条 Trace 找到根因

假设用户反馈：“问天气，Agent 总说查不到。”你拿到对应的 trace：

1. Span 1：模型决定调用 get_weather，参数 {"city": "杭州市"}；

2. Span 2：工具返回“城市不存在”；

3. Span 3：模型没有重试，直接回答“抱歉，暂时查不到”。

定位很清晰：工具描述里写的是“城市中文名，不要带‘市’字”，但模型传入了“杭州市”，而错误信息不够可行动，没有提示“请去掉‘市’”。修复方案是：（1）在工具内部兼容带“市”的输入；（2）让错误信息包含建议；（3）把这条样本加入评测集，防止回归。没有 trace，这类问题只能靠猜。

九、把评估接入发布流程：从“手动跑”到“自动挡”

评测脚本只有被纳入日常流程才会发挥价值，否则很快会被遗忘。一个务实的落地方式是分三个阶段推进：

1. 本地快速回归：开发者改动提示词或工具后，在本地跑一小批核心样本（例如 10~20 条），几分钟内得到反馈，用来做“冒烟测试”；

2. 提交前完整回归：在持续集成（CI）中跑全量评测集，与基线对比。若有样本的通过率明显下降，或平均成本、步数显著上升，就阻止合并并要求说明原因；

3. 灰度发布与线上对比：新版本先对一小部分流量开放，对比新旧版本的成功率、延迟、费用和用户反馈，确认没有退化再逐步放量。

这里有几个需要事先约定的细节：

• 基线是什么：通常是当前线上版本在同一评测集上的结果，而不是某次“最好”的结果；

• 什么算显著变化：样本少时波动大，建议设置“容忍区间”，并对落在区间边缘的变化做人工复核；

• 成本与延迟也是发布门槛：一次改动让通过率提升了几个百分点，却让单次成本翻倍，是否值得要由业务来权衡，而不是默认接受；

• 评测集本身也要维护：当产品能力或工具变化后，过时的期望会产生“假失败”，需要定期清理和更新。

还要提醒的是，离线评测分数高，并不保证线上表现好：线上用户的输入分布会变化，评测集总是滞后于现实。所以线上的 trace 与反馈回流才是评测集保持“新鲜”的关键。

十、常见坑与解决办法

十一、评估与可观测性检查清单

• ☐ 有 30~50 条以上的评测样本，覆盖典型、边界、失败、不应调用四类。

• ☐ 每条样本有可程序判断的期望，开放性问题有评分标准。

• ☐ 每条样本多次运行，报告通过率而不是单次结果。

• ☐ 每次修改提示词、工具、模型版本都跑回归并与基线比较。

• ☐ 每个任务有 trace id，记录每步输入输出、耗时、token、错误。

• ☐ 日志已脱敏，有保存期限和访问控制。

• ☐ 有步数、token、耗时、租户额度的上限。

• ☐ 线上失败案例会定期回流到评测集。

十二、常见问题（FAQ）

Q1：评测集需要多大才够？没有固定数字。起步阶段 30~50 条覆盖面好的样本就能发现大量问题。随着产品成熟，再按失败案例和新功能持续扩充，重点是“覆盖面”和“贴近真实”，而不是数量本身。

Q2：非确定性这么强，怎么判断一次改动是否真的更好？多次运行取通过率，并尽量用同一批样本对比。如果差异很小且样本量不大，就不要下结论；需要时扩大样本或增加运行次数。也可以把关键指标的变化结合人工抽查一起判断。

Q3：可以把用户的完整对话都记进日志吗？要谨慎。完整对话可能含个人隐私或商业机密。建议明确记录范围，对敏感字段脱敏，设置保存期限与访问权限，并遵守所在地区的合规要求。必要时允许用户选择不被记录。

Q4：应该先做评估还是先做可观测性？两者互补，但如果只能先做一个：上线前先做最小的评测集和回归脚本；只要有真实用户使用，就尽快补上 trace。最低成本的做法是先把每步的输入输出、耗时与 token 写成结构化日志。

十三、小结与下一步

把 Agent 当作一个需要持续运营的系统：有评测集做回归，有 trace 做排查，有预算做护栏。这三件事做好，Agent 才能从 Demo 走向生产。

建议的下一步：

1. 从线上日志或自己的使用中挑出 30 条任务，写成带期望的评测样本；

2. 用上面的脚本跑一遍基线，记录通过率、平均步数与 token；

3. 给 Agent 循环加上 trace 记录与预算护栏；

4. 每次修改后对比基线，把新发现的失败案例补进评测集，形成持续改进的闭环。