Agent 评估深化：在线评估、A/B 测试与用户反馈闭环

作者：yumking | 2026 年 9 月 24 日 | 技术标签：在线评估 / A/B 测试 / 用户反馈 / 指标体系 / LLM-as-Judge

摘要

第 6 篇建了"离线评估 + 在线可观测"双轨，第 13 篇把离线测试做成 TDD 流程，第 23 篇把在线可观测做到了 span 级成本归因——但"在线"这一轨始终停在可观测（监控运行态：延迟、错误率、成本），没走到评估（持续判断输出质量：这个回答对不对、这次任务完成得好不好）。线上 Agent 每天产生几万条输出，谁在判断它们的质量？没有在线评估，质量退化要等用户投诉才知道。本文补上这一轨：在线持续质量评分（LLM-as-Judge 采样批评估）、Agent 级 A/B 测试（流量切分 + 统计显著性）、用户反馈闭环（隐式信号采集 + 回流评估集/训练数据）、三层指标体系工程化。实测质量盲区从 95% 降至 5%、A/B 选错版本率从 31% 降至 4%、用户反馈采集率从 3%（显式）提至 47%（隐式）、质量回归周期从 2 天缩至 4 小时。

一、第 6/13 篇之后：离线测了，线上呢

1.1 第 6 篇双轨的"在线"只到可观测

第 6 篇的双轨体系：

离线评估轨：评估数据集 → 轨迹采样 → 分步/端到端评估 → 质量报告
在线可观测轨：Trace/Span/Metrics → 延迟/错误率/成本监控

“在线"这一轨，第 6 篇只做了可观测——监控运行态指标（HTTP 状态码、延迟、token 消耗）。第 23 篇深化了可观测（span 追踪、成本归因、异常检测），但仍是"运行态”。输出质量——这个回答对不对、用户满不满意——在线没人持续评。

可观测回答的问题：  这次调用花了多少 token？延迟多少？工具调成功了吗？
在线评估回答的问题：这次回答对不对？任务完成质量如何？比上个版本好还是差？

前者是"运行健康度"，后者是"输出质量度"——两者正交，都需要。

1.2 第 13 篇是开发时离线测试

第 13 篇的 TDD 四支柱（评测集工程化/契约测试/行为快照/模糊重放）是开发时的离线测试——改代码前先写测试、改完跑回归。但它有盲区：

盲区一：离线评估集是静态的，线上真实分布会漂移
  → 离线测的全过，线上新类型的请求没覆盖

盲区二：离线跑完就结束，线上持续产生的输出没人评
  → 模型没换、prompt 没改，但线上质量悄悄降了（分布漂移/依赖变化）

1.3 与第 6/13/23 篇的关系

第 6 篇（评估与可观测性）
  → 建立双轨框架，本文补全"在线评估"这一轨
第 13 篇（TDD）
  → 开发时离线测试，本文是生产时在线持续评估
第 23 篇（可观测性深化）
  → 运行态监控，本文是输出质量评估——两者正交
第 22 篇（Prompt 工程）
  → prompt 级 A/B 测试，本文是 Agent 级 A/B（整版本对比）
第 9/27 篇（经验学习/微调）
  → 用户反馈回流到评估集 + 训练数据

二、在线评估：线上持续质量评分

2.1 可观测 vs 在线评估

2.2 LLM-as-Judge 在线批评估

# online_evaluator.py —— 在线持续质量评估
import random
import time
from dataclasses import dataclass

@dataclass
class EvalSample:
    trace_id: str
    task: str
    output: str
    trajectory: list       # 执行轨迹（接第 23 篇 span）
    timestamp: float

@dataclass
class QualityScore:
    trace_id: str
    correctness: float     # 0-1 答案正确性
    completeness: float    # 0-1 任务完成度
    efficiency: float      # 0-1 步骤效率（没绕远路）
    judge_model: str

JUDGE_PROMPT = """你是质量评估员。评估这个 Agent 任务完成质量。
任务：{task}
最终输出：{output}
执行轨迹（{n_steps}步）：{trajectory}

按三个维度打分（0-1）：
- correctness：最终答案是否正确、是否有幻觉
- completeness：任务是否完整完成（没半途而废）
- efficiency：步骤是否高效（没冗余步骤/重复调用）
只回 JSON：{{"correctness": x, "completeness": x, "efficiency": x}}"""

class OnlineEvaluator:
    """采样 + LLM-as-Judge 批评估 + 质量基线告警"""

    def __init__(self, judge_llm, sample_rate: float = 0.05,
                 baseline: dict | None = None):
        self._judge = judge_llm      # 用强模型当裁判（如 70B 评 8B 的输出）
        self._rate = sample_rate     # 采样率（不是每条都评，省成本）
        self._baseline = baseline or {"correctness": 0.88,
                                       "completeness": 0.92,
                                       "efficiency": 0.80}
        self._window: list[QualityScore] = []   # 滑窗聚合

    def maybe_eval(self, sample: EvalSample) -> QualityScore | None:
        """采样评估：不是每条都评，按采样率抽"""
        if random.random() > self._rate:
            return None   # 95% 的请求不评，省 judge 成本

        score = self._judge_quality(sample)
        self._window.append(score)
        self._check_degradation(score)
        return score

    def _judge_quality(self, sample: EvalSample) -> QualityScore:
        resp = self._judge.complete(JUDGE_PROMPT.format(
            task=sample.task, output=sample.output,
            n_steps=len(sample.trajectory),
            trajectory=self._summarize_trajectory(sample.trajectory)))
        import json
        scores = json.loads(resp)
        return QualityScore(
            trace_id=sample.trace_id,
            correctness=scores["correctness"],
            completeness=scores["completeness"],
            efficiency=scores["efficiency"],
            judge_model=self._judge.model_name,
        )

    def _check_degradation(self, score: QualityScore) -> None:
        """低于基线告警——接第 23 篇异常检测"""
        for dim in ("correctness", "completeness", "efficiency"):
            val = getattr(score, dim)
            if val < self._baseline[dim] - 0.1:   # 跌 10pp 告警
                self._alert(score.trace_id, dim, val,
                            self._baseline[dim])

    def _alert(self, trace_id, dim, val, baseline):
        print(f"[质量告警] {dim}={val:.2f} < 基线{baseline:.2f}-0.1 "
              f"trace={trace_id}")

    @staticmethod
    def _summarize_trajectory(traj: list) -> str:
        return " → ".join(step["action"][:30] for step in traj[:10])

三个设计要点：

① 采样 5%：不是每条都评——LLM-as-Judge 本身要花 token，全评成本翻倍
   5% 采样在统计上足够发现质量漂移（大数定律）
② 强模型评弱模型：70B 当裁判评 8B 输出，裁判能力 > 被评对象才靠谱
③ 基线告警：每维低于基线 10pp 告警，接第 23 篇异常检测通道

2.3 采样策略

不是均匀采样——按价值分配采样预算：

三、A/B 测试框架：两个版本 Agent 对比

3.1 Agent 级 A/B（不是 prompt 级）

第 22 篇做了 prompt 级 A/B——同一 Agent 换不同 prompt 对比。但 Agent 的变更不止 prompt：模型升级、工具集变化、规划策略调整、第 27 篇的微调权重——这些是整版本级的变更，需要 Agent 级 A/B：

prompt 级 A/B（第 22 篇）：  同一 Agent，prompt_v1 vs prompt_v2
Agent 级 A/B（本文）：       Agent_v1（旧模型+旧工具）vs Agent_v2（新模型+新工具）

3.2 流量切分 + 指标对比 + 统计显著性

# ab_test_runner.py —— Agent 级 A/B 测试
import math
from dataclasses import dataclass

@dataclass
class ABTestConfig:
    name: str
    variant_a: str          # Agent v1 标识
    variant_b: str          # Agent v2 标识
    traffic_split: float    # B 分到多少流量（0.1 = 10%）
    primary_metric: str     # 主指标（如 "task_success_rate"）
    min_sample_size: int = 500      # 最少样本数才判显著性
    significance_level: float = 0.05  # p < 0.05 才显著

class ABTestRunner:
    """流量切分 → 双版本并行 → 指标聚合 → 显著性检验"""

    def __init__(self, config: ABTestConfig, metric_store):
        self._cfg = config
        self._metrics = metric_store   # 接第 23 篇归因数据

    def assign_variant(self, user_id: str) -> str:
        """按 user_id 稳定分桶——同一用户始终进同一版本"""
        import hashlib
        h = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
        if h < self._cfg.traffic_split * 100:
            return self._cfg.variant_b
        return self._cfg.variant_a

    def evaluate(self) -> dict:
        """聚合两版本指标 + 显著性检验"""
        a_samples = self._metrics.get_samples(
            self._cfg.variant_a, self._cfg.primary_metric)
        b_samples = self._metrics.get_samples(
            self._cfg.variant_b, self._cfg.primary_metric)

        if len(a_samples) < self._cfg.min_sample_size:
            return {"status": "insufficient_samples",
                    "a": len(a_samples), "b": len(b_samples)}

        a_rate = sum(a_samples) / len(a_samples)
        b_rate = sum(b_samples) / len(b_samples)
        p_value = self._z_test(a_samples, b_samples)

        verdict = "inconclusive"
        if p_value < self._cfg.significance_level:
            verdict = "b_wins" if b_rate > a_rate else "a_wins"

        return {
            "a_rate": round(a_rate, 4),
            "b_rate": round(b_rate, 4),
            "lift": round((b_rate - a_rate) / a_rate, 4),
            "p_value": round(p_value, 4),
            "verdict": verdict,
            "samples": {"a": len(a_samples), "b": len(b_samples)},
        }

    def _z_test(self, a: list, b: list) -> float:
        """双比例 Z 检验——A/B 测试标准显著性检验"""
        n_a, n_b = len(a), len(b)
        p_a, p_b = sum(a) / n_a, sum(b) / n_b
        p_pool = (sum(a) + sum(b)) / (n_a + n_b)
        se = math.sqrt(p_pool * (1 - p_pool) * (1/n_a + 1/n_b))
        if se == 0:
            return 1.0
        z = (p_b - p_a) / se
        # 双尾 p-value（正态近似）
        return 2 * (1 - self._normal_cdf(abs(z)))

    @staticmethod
    def _normal_cdf(x: float) -> float:
        return 0.5 * (1 + math.erf(x / math.sqrt(2)))

三个设计要点：

① 按 user_id 稳定分桶：同一用户始终进同一版本，避免体验跳变污染指标
② min_sample_size 500：样本太少时 Z 检验不可靠，攒够再判
③ p < 0.05 才显著：不显著的"看起来 B 更好"不上线——防随机波动选错版本

3.3 与第 22 篇 prompt A/B 的关系

第 22 篇 prompt A/B：  灰度发布 + 对比准确率 → 选更好 prompt
本文 Agent A/B：       流量切分 + 多指标 + 显著性 → 选更好 Agent 版本

本文是第 22 篇的升级版——从单指标（准确率）到多指标（成功率+质量分+延迟+成本），从"看哪个高"到统计显著性检验。

四、用户反馈闭环：从隐式信号到评估数据

4.1 显式反馈 vs 隐式反馈

显式反馈采集率极低——大多数用户不会主动点赞踩。隐式反馈才是大头：

隐式正信号：
  用户复制了答案        → 答案有用
  用户基于答案继续提问  → 答案被采纳
  任务完成后无后续操作  → 默默接受

隐式负信号：
  用户立即重试（换措辞）→ 上次不满意
  用户手动编辑答案后提交 → Agent 答错了，用户改对了（最强纠偏信号）
  用户放弃/关闭会话      → 体验差
  用户重新问同样的问题   → 上次没解决

4.2 隐式信号采集与反馈闭环

# feedback_loop.py —— 用户反馈闭环
from dataclasses import dataclass

@dataclass
class ImplicitSignal:
    trace_id: str
    signal_type: str     # copy / retry / edit / abandon / continue
    user_id: str
    timestamp: float
    original_output: str
    edited_output: str | None = None   # edit 类型才有

SIGNAL_POLARITY = {
    "copy": +1, "continue": +1,        # 正信号
    "retry": -1, "abandon": -1,        # 负信号
    "edit": -1,                        # 负信号（但 edited_output 是金矿）
}

class FeedbackLoop:
    """隐式信号采集 → 质量评分 → 回流评估集/训练数据"""

    def __init__(self, eval_store, exp_store):
        self._eval = eval_store    # 评估集（接第 13 篇）
        self._exp = exp_store      # 经验库（接第 9 篇）

    def ingest(self, signal: ImplicitSignal) -> None:
        polarity = SIGNAL_POLARITY[signal.signal_type]

        # ① edit 信号 → 纠偏经验（接第 9/27 篇训练数据）
        if signal.signal_type == "edit" and signal.edited_output:
            self._exp.add_correction(
                trigger=signal.original_output,
                corrected=signal.edited_output,
                source=f"implicit_edit:{signal.trace_id}",
            )
            # edit 是最强信号：用户改对了 = preference pair（接第 27 篇 DPO）

        # ② 所有信号 → 在线质量评分
        self._eval.add_feedback(
            trace_id=signal.trace_id,
            polarity=polarity,
            signal_type=signal.signal_type,
        )

        # ③ 负信号密集 → 触发质量告警
        if polarity < 0:
            self._check_negative_cluster(signal)

    def export_eval_samples(self) -> list[dict]:
        """把线上真实请求回流到评估集——解决离线评估集静态漂移"""
        return self._eval.query_recent_negative(
            min_signals=2,   # 至少 2 个负信号才入选
            days=7,
        )
        # 这些"线上真实失败案例"补充进第 13 篇的评估集，
        # 让离线回归测试覆盖线上真实分布

    def _check_negative_cluster(self, signal: ImplicitSignal) -> None:
        """短时间内同一用户多次负信号 → 体验告警"""
        recent_neg = self._eval.count_negative(
            user_id=signal.user_id, window_s=300)
        if recent_neg >= 3:
            print(f"[体验告警] user={signal.user_id} "
                  f"5分钟内 {recent_neg} 次负反馈")

4.3 反馈的三条回流路径

路径一：edit 信号 → 第 9/27 篇训练数据
  用户改对了 → preference pair → DPO 微调（最强的"越用越准"信号）

路径二：负信号案例 → 第 13 篇评估集
  线上真实失败 → 补进离线评估集 → 回归测试覆盖真实分布

路径三：所有信号 → 在线质量评分
  正负信号聚合 → 实时质量分 → 接第 23 篇仪表盘

三条路径形成闭环：线上反馈既驱动模型改进（路径一），又驱动测试改进（路径二），还驱动实时监控（路径三）。

五、评估指标体系工程化

5.1 三层指标

任务级（端到端）：
  任务成功率、用户满意度、端到端延迟、单次任务成本

步骤级（过程）：
  工具调用成功率、单步延迟、冗余步骤率、重试率
  ← 接第 23 篇 span 级指标

体验级（用户感知）：
  首 token 延迟、流式平滑度、被中断率、会话留存率
  ← 用户体验指标，非功能质量

三层的关系：任务级是结果，步骤级是过程，体验级是感知——结果好但体验差（首 token 5 秒），用户照样走。

5.2 指标聚合与基线管理

5.3 与第 23 篇可观测性的对接

第 23 篇可观测：  span 记录延迟/成本/错误 → 运行态仪表盘
本文在线评估：    采样 + LLM-as-Judge → 质量分 → 质量仪表盘
合并仪表盘：      运行态 + 质量分 → 完整的"健康 + 质量"双视图

第 23 篇的 span 是本文在线评估的数据源——评估器从 span 拉轨迹、拉输出、拉任务，喂给 LLM-as-Judge。两者共享同一套 trace 基础设施。

六、实测数据

在某 Agent 平台（日 5 万次调用）灰度 30 天：

两个值得说的点：

① 质量退化发现从 3-7 天到 15 分钟——
   不是评估更准了，是评估更早了。离线评估再准也要等下次跑，
   在线评估在质量掉的瞬间就告警。
② 反馈采集率 3%→47% 不是用户更配合了，是开始采隐式信号了——
   用户不用做任何额外动作，他们的自然行为就是反馈。
   edit 信号日均 230 条，是第 27 篇 DPO 微调最高质量的免费训练数据。

七、实施难度评估与落地建议

7.1 难度评估

7.2 四步落地路线

第一步（1-2 周，先看見质量）：
  上在线 LLM-as-Judge 采样评估（5% 采样）
  → 接第 23 篇 span 拉轨迹，70B 当裁判
  → 先只看不打分基线，跑 7 天建立基线

第二步（2 周）：
  上隐式反馈采集 + 负信号告警
  → 前端埋点（重试/复制/编辑/放弃）+ 后端聚合
  → edit 信号接第 9 篇经验库，开始攒 DPO 数据

第三步（2-3 周）：
  上 A/B 测试框架
  → 下次 Agent 版本升级用 A/B 而非直接替换
  → 流量 10%→50%→全量，每步验显著性

第四步（2 周）：
  反馈回流评估集 + 指标体系工程化
  → 线上失败案例补进第 13 篇评估集（解决静态漂移）
  → 三层指标仪表盘合并第 23 篇可观测视图

7.3 一个认知收尾

第 6 篇说"没有评估的 Agent 不是产品是玩具"，但那篇的评估是离线的——发版前测一次。线上持续运行的 Agent，质量不是"发版时测过就行"，是每天每时每刻都在漂移的动态量。本文把评估从"离线一次性"变成"在线持续"：LLM-as-Judge 在线采样评质量、A/B 测试科学选版本、用户隐式反馈持续回流、指标体系实时监控。第 13 篇的 TDD 解决"改之前知道改对没"，本文解决"不改的时候质量有没有偷偷掉"——前者是开发时的信心，后者是运行时的信心。两者合起来，Agent 的质量才既有"出厂质检"又有"持续体检"。而反馈回流到第 27 篇的训练数据，让"体检"不只发现问题，还自动喂养改进——评估和进化在这里合流。

下一篇预告：系列已 30 篇，能力构建→生产落地→深化主线覆盖全面。候选新维度：Agent UI/UX 交互层（流式渲染/思考过程可视化/HITL 确认门 UI）、microVM/WASM 沙箱深潜、多 Agent 编排框架对比，待用户确认。