端到端实战:为客服 Agent 搭一套完整评估

系统简介

假设我们有一个上线中的客服 Agent,架构如下:

目标是为这套系统建立可持续运营的评估体系,能回答 3 个问题:

1. 新 prompt / 新模型是不是真的变好了?
2. 线上是不是稳定在合格水位?
3. 具体哪类问题做得差、为什么?

Step 1:从线上工单采样

我们有 30 天、18 万条真实工单。不能 all-in,先按意图/难度/反馈分层采样:

import pandas as pd
from sklearn.cluster import KMeans
from sklearn.feature_extraction.text import TfidfVectorizer

df = pd.read_parquet("tickets_last_30d.parquet")

# 1) 先按意图分桶
buckets = {
    "order_status":   df[df.intent == "order_status"],
    "refund":         df[df.intent == "refund"],
    "product_q":      df[df.intent == "product_q"],
    "complaint":      df[df.intent == "complaint"],
    "other":          df[df.intent == "other"],
}

# 2) 每桶内按难度/反馈再采 40 条:
#    - 20 条典型(多数 query)
#    - 10 条长尾(KMeans 聚类的小簇)
#    - 10 条历史翻车(👎 / regenerate / human takeover)
def stratified_sample(bucket, n_typical=20, n_tail=10, n_bad=10):
    vec = TfidfVectorizer(max_features=512).fit_transform(bucket.user_input)
    clusters = KMeans(n_clusters=20, random_state=42).fit_predict(vec)
    bucket = bucket.assign(cluster=clusters)

    typical = bucket[bucket.cluster.isin(bucket.cluster.value_counts().head(3).index)].sample(n_typical)
    tail    = bucket[bucket.cluster.isin(bucket.cluster.value_counts().tail(5).index)].sample(n_tail)
    bad     = bucket[bucket.had_issue == True].sample(n_bad)
    return pd.concat([typical, tail, bad])

samples = pd.concat(stratified_sample(b) for b in buckets.values())
# 5 桶 * 40 = 200 条 golden 候选

Step 2:人工标注,产出 golden set

200 条候选由 2 位客服专家独立标注:

• expected_answer:标准回复(语气、信息完整度)
• required_tools:应该调用哪些工具(允许多条合理路径,标最直接的一条)
• must_contain / must_not_contain:硬约束关键词
• difficulty:easy / medium / hard
• label_notes:标注说明(这条为什么判断这样)

完成后跑 Cohen's κ:

from sklearn.metrics import cohen_kappa_score
κ = cohen_kappa_score(expert_A_labels, expert_B_labels, weights="quadratic")
print(f"Inter-annotator κ = {κ:.2f}")
# κ = 0.74  ✅ good agreement
# 低于 0.6 的样本组织讨论,达成共识再入库

Step 3:定义 6 个维度指标

一个客服 Agent 只看"是否正确"远远不够。6 维评估:

Scorer 实现

from braintrust import Score
from autoevals import Factuality

def score_task_success(output, expected):
    # LLM Judge:pointwise 1-5
    result = llm_judge(JUDGE_TASK_SUCCESS.format(
        question=output.question, answer=output.reply, expected=expected,
    ))
    return Score(name="TaskSuccess", score=result["score"] / 5, metadata=result)

def score_tool_correctness(output, expected_tools):
    actual = [(t.name, t.args) for t in output.trajectory]
    # 按工具名 + 关键参数匹配,允许顺序不同
    name_match = set(t[0] for t in actual) >= set(t[0] for t in expected_tools)
    arg_match = all(args_match(a, e) for a, e in zip(actual, expected_tools))
    score = float(name_match and arg_match)
    return Score(name="ToolCorrectness", score=score)

def score_faithfulness(output):
    return Score(
        name="Faithfulness",
        score=ragas_faithfulness(output.question, output.contexts, output.reply),
    )

def score_tone(output):
    result = llm_judge(JUDGE_TONE.format(answer=output.reply))
    return Score(name="Tone", score=result["score"] / 5)

def score_constraints(output, must_contain, must_not_contain):
    ok = all(kw in output.reply for kw in must_contain)
    ok &= all(kw not in output.reply for kw in must_not_contain)
    return Score(name="Constraints", score=float(ok))

def score_efficiency(output):
    n_steps = len(output.trajectory)
    penalty = max(0, (n_steps - 3) / 10)   # 3 步以内是理想
    return Score(name="Efficiency", score=max(0, 1 - penalty),
                 metadata={"n_steps": n_steps, "tokens": output.total_tokens})

Step 4:Judge Prompt 与校准

JUDGE_TASK_SUCCESS = """你是客服质量审核专家。给下面这条回复打分(1-5)。

维度:是否真正解决了用户问题。忽略语气和格式,只看核心问题解决度。

5 — 完整解决,信息准确完整
4 — 基本解决,有小瑕疵但不影响
3 — 部分解决,缺关键信息
2 — 严重偏题或错误
1 — 完全答非所问或有害

## 例子
(标注员共识的 3 个 few-shot 例子)

## 判断
用户问题: {question}
参考回复: {expected}
实际回复: {answer}

先 2 句话分析,再返回 JSON: {{"reasoning": "...", "score": 1-5}}"""

校准流程(第 4 章讲过的闭环):

# 挑 50 条已人工打分的 case,对比 Judge
human = []   # 人工 1-5
judge = []   # Judge 1-5

for case in calibration_set:
    h = case.human_score
    j = judge_task_success(case.output, case.expected)
    human.append(h); judge.append(j)

from sklearn.metrics import cohen_kappa_score
κ = cohen_kappa_score(human, judge, weights="quadratic")
print(f"Judge-Human κ = {κ:.2f}")
# 目标 ≥ 0.6。本例首版 0.48,加了 few-shot 后 0.66 ✅

Step 5:接入 Braintrust 跑评估

from braintrust import Eval, init_dataset

dataset = init_dataset(project="customer-support", name="golden-v1")

def agent_task(case):
    # 跑我们的 agent,返回带 trajectory 的结构化输出
    return run_agent(case["user_input"])

Eval(
    "customer-support",
    data=dataset,
    task=agent_task,
    scores=[
        score_task_success, score_tool_correctness, score_faithfulness,
        score_tone, score_constraints, score_efficiency,
    ],
    experiment_name="v7-sonnet-4-6",
    metadata={
        "model": "claude-sonnet-4-6",
        "prompt_version": "v7",
        "kb_snapshot": "2026-04-28",
    },
    max_concurrency=10,
)

Braintrust UI 里看到:

• 六个指标的整体均值 + 标准差
• 与上次实验的 diff,哪些 case 变好/变差,按分差排序
• 按 intent 分层:complaint 只有 0.61,order_status 达 0.91
• 按 difficulty 分层:hard 只有 0.53 → 定位优化重点

Step 6:CI 集成,当作"质量门禁"

每个 PR 如果改动了 prompt、tool、agent 代码,自动跑回归集 + Judge,不过线挂红:

# .github/workflows/evals.yml
name: Evals Gate
on:
  pull_request:
    paths:
      - "prompts/**"
      - "agent/**"
      - "tools/**"

jobs:
  evals:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -r requirements.txt
      - name: Run golden set
        env:
          OPENAI_API_KEY:     ${{ secrets.OPENAI_API_KEY }}
          ANTHROPIC_API_KEY:  ${{ secrets.ANTHROPIC_API_KEY }}
          BRAINTRUST_API_KEY: ${{ secrets.BRAINTRUST_API_KEY }}
        run: python -m evals.run --dataset golden-v1 --experiment ci-${{ github.sha }}
      - name: Check thresholds
        run: python -m evals.gate --min-task-success 0.80 --min-tool 0.90

# evals/gate.py
import sys, argparse
from braintrust import summarize_experiment

p = argparse.ArgumentParser()
p.add_argument("--min-task-success", type=float, default=0.80)
p.add_argument("--min-tool", type=float, default=0.90)
args = p.parse_args()

summary = summarize_experiment("customer-support", experiment="ci-latest")
ts  = summary.scores["TaskSuccess"].mean
tc  = summary.scores["ToolCorrectness"].mean

if ts < args.min_task_success:
    print(f"❌ TaskSuccess {ts:.2f} < {args.min_task_success}"); sys.exit(1)
if tc < args.min_tool:
    print(f"❌ ToolCorrectness {tc:.2f} < {args.min_tool}"); sys.exit(1)
print(f"✅ TaskSuccess={ts:.2f}  ToolCorrectness={tc:.2f}")

Step 7:灰度发布

回归过了只是第一关。接下来走第 8 章的流程:

1. Shadow 1 天:v7 不发给用户,离线跑 Judge 对比 v6,看 Faithfulness / Tone 有无突变
2. Canary 1%:观察 1 天的真实用户 thumbs / regenerate 率
3. Canary 5% / 25% / 100%:每档 1-2 天,主指标显著胜出且护栏不破
4. 任何档发现退步 → 立即回滚

# canary_gate.py — 每次扩量前跑
def can_expand(variant="v7", baseline="v6"):
    m = query_metrics(hours=24)
    checks = {
        "thumbs_down_rate":  m[variant].down_rate <= m[baseline].down_rate * 1.1,
        "regen_rate":        m[variant].regen <= m[baseline].regen * 1.1,
        "p95_latency_ms":    m[variant].p95 <= m[baseline].p95 * 1.2,
        "cost_per_session":  m[variant].cost <= m[baseline].cost * 1.3,
        "judge_sample_score": m[variant].judge >= m[baseline].judge - 0.02,
    }
    failed = [k for k, ok in checks.items() if not ok]
    return (not failed), failed

Step 8:持续运营的日常

上线只是起点。每天/每周/每季度的固定动作:

最终分数卡示例

Experiment: v7-sonnet-4-6 (2026-04-30)
Dataset: golden-v1 (200 cases)
Model: claude-sonnet-4-6
Prompt: v7 (tone-tweak)

┌─────────────────────┬────────┬─────────┬──────────┐
│ Metric              │  Mean  │ vs v6   │ Baseline │
├─────────────────────┼────────┼─────────┼──────────┤
│ TaskSuccess         │  0.86  │ +0.04 ↑ │  0.82    │
│ ToolCorrectness     │  0.93  │ +0.01   │  0.92    │
│ Faithfulness        │  0.91  │ +0.03 ↑ │  0.88    │
│ Tone                │  0.88  │ +0.11 ↑ │  0.77    │
│ Constraints         │  1.00  │   —     │  1.00    │
│ Efficiency          │  0.79  │ -0.02   │  0.81    │
└─────────────────────┴────────┴─────────┴──────────┘

By intent:
  order_status:  0.91   refund: 0.84   product_q: 0.87
  complaint:     0.78   other:  0.72

By difficulty:
  easy:   0.94   medium: 0.86   hard: 0.68

Cost per session (avg):  $0.011
p95 latency:              2.4s

Gate decision: ✅ PASS (TaskSuccess 0.86 ≥ 0.80, ToolCorrectness 0.93 ≥ 0.90)
Ready for canary rollout.

常见踩坑

验收清单:你的评估体系达标了吗?

• ✅ Golden set ≥ 100 条,覆盖主要 intent、难度、历史翻车 case
• ✅ 2+ 位专家标注,Cohen's κ ≥ 0.6
• ✅ 至少 3 维度指标:正确性 / 过程 / 硬约束
• ✅ Judge 与人工 κ ≥ 0.6,且 Judge 异源
• ✅ 接入 Braintrust / LangSmith / Promptfoo 之一,有历史对比
• ✅ CI 有质量门禁,回归集必跑
• ✅ 线上有 OTel trace,成本/延迟/token 上报
• ✅ 灰度流程:shadow → canary → full,每档有退出条件
• ✅ 线上反馈(thumbs / regenerate)关联到 trace_id
• ✅ 每周有线上 case 进入 regression set

全书回顾