线上流量闭环与评测集治理
真正成熟的企业大模型测试，不会把评测集当成一次性作业，而会把它当成持续运营资产。线上用户的问题分布、文档内容、模型能力、工具链路都在变，如果评测集不跟着变，你的回归体系会越来越"看起来很忙，实际上看不到真实风险"。
教学导读
**定位：**这一章把"线上 bad case 很多"变成"怎么把线上问题沉淀成长期可复用的评测资产"。 **前置依赖：**建议已学习评测工程化、数据集分层和可观测性基础。 **适用场景：**生产 AI 应用质量运营、数据集版本治理、线上问题回归、灰度发布。 **学完产出：**你应该能设计一条从线上日志到 evalset 入库的闭环流程，并给出字段规范和采样策略。
先说结论。
一套评测集如果三个月没更新，很可能已经开始失真。因为线上新的高频问题、新的绕过方式、新的文档版本、新的用户表达都在出现。你需要的不是"更多数据"，而是"持续把最有代表性的线上问题抽回来"。 从信息论的角度理解：评测集的信息量应该与线上真实分布保持对齐。当二者的 KL 散度过大时，评测结果就不再能可靠地预测线上表现。这不是玄学，而是可以量化、可以检测、可以用工程手段持续修正的系统性问题。
01. 线上质量漏斗：从真实问题到可回归资产
推荐把线上信号统一进一个质量漏斗：
- **日志采样：**抽取真实会话、真实 query、真实工具调用链。
- **信号聚合：**用户点踩、人工升级、工单、异常 trace、超时、成本异常。
- **归因分类：**区分幻觉、拒答、格式错、工具错、性能慢、安全风险。
- **标准化：**脱敏、补充期望结果、定义评判标准。
- **入库：**进入不同等级的 evalset。
1.1 一个从投诉到入库的完整例子
线上原始信号：
用户投诉："机器人说试用期能休年假，HR 说根本不行。"
第一步：回捞原始对话和检索证据
→ 找到会话 ID: conv-2026-04-001
→ 检索到 RAG 返回的 3 个文档块
→ 确认第 2 块包含正确制度但模型未引用
第二步：确认知识库里其实有正确制度
→ HR 政策文档 v3.2，第 4.2 节明确"试用期不享受年假"
→ 模型引用了过期文档 v2.8 的内容
第三步：把问题标成 grounding_miss
→ 不是模型编造，而是检索排序问题 + 模型未正确判断文档时效
第四步：补 expected_behavior 和 gold evidence
→ expected: "试用期员工不享受年假，转正后按入职时间折算"
→ gold_doc: "hr-policy-v3.2-section-4.2"
→ must_not_claim: ["试用期可以休年假", "试用期享有年假"]
第五步：进入 L1 政策问答集 + L3 线上采样集
→ priority: P0（涉及劳动合规）
→ tags: [hr, policy, grounding, online_case]
第六步：下周模型/Prompt 升级时纳入回归
→ 加入 CI pipeline 的 L1 golden set 检查这个过程看起来琐碎，但它决定了你的组织是在"处理投诉"，还是在"积累质量资产"。真正成熟的团队不是把 bad case 截图发群里，而是让它进入一条稳定、可统计、可复跑的治理流水线。
关键点。
线上 bad case 不能直接塞进评测集。需要先脱敏、去噪、补足上下文、定义清晰的通过标准。否则你只是把投诉搬进了仓库，而不是把问题变成可持续检测资产。
01.5 信号采集的工程实现
质量漏斗的第一步是采集足够丰富的信号。企业生产系统通常有多个信号来源，需要统一接入：
| 信号源 | 采集方式 | 延迟 | 信噪比 | 典型体量 |
|---|---|---|---|---|
| 用户点踩/点赞 | 前端 SDK 埋点 | 实时 | 中高（带情绪偏差） | 日均 0.5-3% 的会话 |
| 客服工单升级 | 工单系统 Webhook | 分钟级 | 高（经过人工筛选） | 日均数十到数百 |
| 异常 Trace | APM / OpenTelemetry | 实时 | 高（有明确错误码） | 占总流量 1-5% |
| 人工抽检 | 标注平台定期推送 | 小时级 | 最高（专家判断） | 每周百条量级 |
| 自动检测器 | 规则引擎 / 小模型 | 实时 | 中（需调优阈值） | 全流量覆盖 |
| 成本异常 | Token 用量监控 | 分钟级 | 中低（需结合上下文） | 超阈值部分 |
1.2 信号采集的统一数据模型
python
class QualitySignal:
signal_id: str # 唯一标识
session_id: str # 关联会话
timestamp: datetime
source: str # thumb_down | escalation | trace_error | ...
severity: str # critical | major | minor | info
raw_payload: dict # 原始数据（脱敏前暂存）
# 采集元数据
model_version: str
prompt_version: str
retriever_version: str
# 处理状态
status: str # raw | triaged | standardized | ingested | rejected
triage_category: str # 归因分类结果
assignee: str # 责任人/责任组1.3 自动质量检测器的设计
除了被动等待用户反馈，成熟的系统还会部署主动检测器。这些检测器以旁路方式运行，不影响主链路性能：
事实一致性检测器
用 NLI（自然语言推理）模型判断生成内容是否与检索证据矛盾。阈值通常设在 entailment score < 0.3 时触发告警。典型工具：Vectara HHEM、AlignScore。
拒答异常检测器
统计滑动窗口内的拒答率。如果某个 intent 分类的拒答率突增 > 2σ，自动标记为异常。常见原因：safety filter 误杀、知识库缺失。
格式完整性检测器
对要求结构化输出的场景，实时校验 JSON Schema、字段完整性、枚举合法性。失败率 > 5% 即触发回归。
成本与延迟异常检测器
监控单次调用的 token 数和延迟。如果 P95 token 消耗突增 50% 以上，可能是 prompt 膨胀或无效重试导致。
# 事实一致性检测器的核心逻辑示例
import numpy as np
from transformers import pipeline
nli_model = pipeline("text-classification",
model="cross-encoder/nli-deberta-v3-base")
def check_grounding(generated_text: str, evidence_chunks: list[str]) -> dict:
"""检测生成内容是否与检索证据一致"""
scores = []
contradictions = []
for chunk in evidence_chunks:
result = nli_model(f"{chunk} [SEP] {generated_text}")
label_scores = {r["label"]: r["score"] for r in result}
scores.append(label_scores.get("entailment", 0))
if label_scores.get("contradiction", 0) > 0.7:
contradictions.append({
"evidence": chunk[:200],
"contradiction_score": label_scores["contradiction"]
})
max_entailment = max(scores) if scores else 0
return {
"grounded": max_entailment > 0.5,
"max_entailment_score": max_entailment,
"contradictions": contradictions,
"verdict": "pass" if max_entailment > 0.5 and not contradictions
else "fail"
}01.6 完整回流 Pipeline 架构
从信号采集到最终入库，需要一条完整的数据管线。以下是企业级实现的推荐架构：
┌─────────────┐ ┌──────────────┐ ┌─────────────┐ ┌────────────┐
│ 信号采集层 │ → │ 分类归因层 │ → │ 标准化层 │ → │ 入库审核层 │
│ │ │ │ │ │ │ │
│ · 用户反馈 │ │ · 自动分类 │ │ · 脱敏处理 │ │ · 人工审核 │
│ · 异常trace │ │ · 人工复核 │ │ · 去重聚类 │ │ · 质量门禁 │
│ · 自动检测 │ │ · 优先级分配 │ │ · 字段补全 │ │ · 版本打标 │
│ · 人工抽检 │ │ · 责任路由 │ │ · 证据关联 │ │ · 分层路由 │
└─────────────┘ └──────────────┘ └─────────────┘ └────────────┘
│
┌──────────────────────────────────────────────┘
↓
┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐
│ L0 冒烟 │ │ L1 Golden │ │ L2 对抗 │ │ L3 采样 │
│ 10-30条 │ │ 50-200条 │ │ 安全边界 │ │ 滚动更新 │
└───────────┘ └───────────┘ └───────────┘ └───────────┘1.4 Pipeline 的完整实现
python
import hashlib
import re
from datetime import datetime
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
class EvalsetLevel(Enum):
L0_SMOKE = "l0_smoke"
L1_GOLDEN = "l1_golden"
L2_ADVERSARIAL = "l2_adversarial"
L3_SAMPLING = "l3_sampling"
class CaseCategory(Enum):
GROUNDING_MISS = "grounding_miss"
HALLUCINATION = "hallucination"
REFUSAL_FALSE = "refusal_false_positive"
REFUSAL_MISS = "refusal_miss"
FORMAT_ERROR = "format_error"
TOOL_ERROR = "tool_call_error"
LATENCY = "latency_issue"
SAFETY = "safety_violation"
COST_ANOMALY = "cost_anomaly"
@dataclass
class EvalSample:
id: str
source: str
category: CaseCategory
input_text: str
context: str
expected_behavior: str
priority: str
tags: list[str]
model_version: str
prompt_version: str
judge_rubric_version: str
must_not_claim: list[str] = field(default_factory=list)
must_contain: list[str] = field(default_factory=list)
gold_evidence: list[str] = field(default_factory=list)
owner: str = ""
created_at: str = ""
evalset_level: EvalsetLevel = EvalsetLevel.L3_SAMPLING
class EvalsetPipeline:
"""从线上信号到评测集入库的完整管线"""
def __init__(self, dedup_store, classifier, anonymizer):
self.dedup_store = dedup_store
self.classifier = classifier
self.anonymizer = anonymizer
self.pending_review = []
def ingest_signal(self, signal: QualitySignal) -> Optional[EvalSample]:
"""完整的入库流程"""
# Step 1: 去重检查
content_hash = self._compute_hash(signal.raw_payload)
if self.dedup_store.exists(content_hash):
return None # 重复信号，跳过
# Step 2: 自动分类
category = self.classifier.classify(signal)
# Step 3: 脱敏
sanitized = self.anonymizer.sanitize(signal.raw_payload)
# Step 4: 标准化为 EvalSample
sample = self._standardize(signal, sanitized, category)
# Step 5: 自动确定优先级和层级
sample.priority = self._assign_priority(category, signal)
sample.evalset_level = self._route_to_level(sample)
# Step 6: 加入待审核队列
self.pending_review.append(sample)
self.dedup_store.add(content_hash)
return sample
def _compute_hash(self, payload: dict) -> str:
"""基于语义内容的去重哈希"""
key_fields = f"{payload.get('query', '')}"
return hashlib.sha256(key_fields.encode()).hexdigest()[:16]
def _assign_priority(self, category, signal) -> str:
priority_map = {
CaseCategory.SAFETY: "P0",
CaseCategory.HALLUCINATION: "P0",
CaseCategory.GROUNDING_MISS: "P1",
CaseCategory.REFUSAL_MISS: "P1",
CaseCategory.TOOL_ERROR: "P1",
CaseCategory.FORMAT_ERROR: "P2",
CaseCategory.REFUSAL_FALSE: "P2",
CaseCategory.LATENCY: "P2",
CaseCategory.COST_ANOMALY: "P3",
}
return priority_map.get(category, "P3")
def _route_to_level(self, sample: EvalSample) -> EvalsetLevel:
if sample.priority == "P0":
return EvalsetLevel.L1_GOLDEN
if sample.category == CaseCategory.SAFETY:
return EvalsetLevel.L2_ADVERSARIAL
return EvalsetLevel.L3_SAMPLING02. 为什么 Bad Case 分类要先做再修
| 分类 | 典型来源 | 对应处置 | 影响层级 |
|---|---|---|---|
| 事实性错误（幻觉） | 用户点踩、人工抽检 | 补充高风险问答集、证据约束 | 模型/检索/Prompt |
| 接地失败（Grounding Miss） | NLI 检测器、人工审核 | 修复检索排序、文档版本管理 | 检索/知识库 |
| 拒答/误拒答 | 客服升级、业务抱怨 | 边界 slice 回归、policy 对齐 | Safety/Prompt |
| 格式错误 | 接口异常、前端解析失败 | schema 回归、输出预算测试 | 结构化输出层 |
| 工具调用失败 | trace 异常、重试日志 | 轨迹评测、参数断言 | Agent/Tool |
| 性能问题 | TTFT 告警、P99 延迟 | 分段压测、容量治理 | 推理引擎/基础设施 |
| 安全违规 | 审核系统、合规检测 | 对抗集更新、红队测试 | Safety/模型 |
| 成本异常 | Token 用量告警 | Prompt 精简、缓存优化 | 全链路 |
2.1 为什么分类比修复更应该先做
如果不先分类，团队通常会把不同层的问题混在一起处理，结果就是：
- 用改 Prompt 去修工具参数错误，越修越乱。
- 用补知识库去修格式错误，完全打偏。
- 把"偶发延迟"误当成"模型能力下降"，排查成本极高。
- 安全问题和功能问题混在一起处理，导致安全修复被低优先级需求挤掉。
分类的价值不只是为了报表好看，而是为了把后续动作分流给正确的人和正确的系统层。企业真正需要的是一套 可分派、可统计、可追责 的分类体系。
分类体系的设计原则。
好的分类体系满足三个条件：互斥（一个问题只属于一个类别）、穷举（所有问题都能归类）、可操作（每个类别都有明确的下一步动作和责任人）。MECE 原则在这里同样适用。
02.5 分类一致性的数学度量
当多个标注员对 bad case 做分类时，如何衡量分类的一致性？这直接影响后续数据质量。
Cohen's Kappa 系数
Cohen's Kappa 是最常用的标注一致性度量，它扣除了随机一致的概率： κ = (P_o - P_e) / (1 - P_e) 其中：
P_o= 实际一致率（两个标注员对同一样本给出相同标签的比例）P_e= 随机一致的期望概率
# Cohen's Kappa 的计算
import numpy as np
def cohens_kappa(annotations_a: list, annotations_b: list, categories: list) -> float:
"""
计算两位标注员之间的 Cohen's Kappa
annotations_a, annotations_b: 两位标注员的标注结果列表
categories: 所有可能的分类标签
"""
n = len(annotations_a)
assert n == len(annotations_b)
# 构建混淆矩阵
k = len(categories)
cat_index = {c: i for i, c in enumerate(categories)}
confusion = np.zeros((k, k))
for a, b in zip(annotations_a, annotations_b):
confusion[cat_index[a]][cat_index[b]] += 1
# 实际一致率 P_o
p_o = np.trace(confusion) / n
# 期望一致率 P_e
row_sums = confusion.sum(axis=1) / n
col_sums = confusion.sum(axis=0) / n
p_e = np.sum(row_sums * col_sums)
# Cohen's Kappa
kappa = (p_o - p_e) / (1 - p_e) if p_e < 1 else 1.0
return kappa
# 解读标准
# κ > 0.8: 几乎完美一致，分类体系可靠
# 0.6 < κ ≤ 0.8: 实质一致，可接受
# 0.4 < κ ≤ 0.6: 中等一致，分类标准需改进
# κ ≤ 0.4: 一致性差，分类体系或标注指南需要重新设计Fleiss' Kappa：多标注员场景
当有三个以上标注员参与时，需要使用 Fleiss' Kappa： κ_Fleiss = (P̄ - P̄_e) / (1 - P̄_e)
python
def fleiss_kappa(rating_matrix: np.ndarray) -> float:
"""
Fleiss' Kappa 用于多标注员一致性度量
rating_matrix: shape (n_subjects, n_categories)
每行是一个样本，每列是某个类别被选择的次数
"""
n_subjects, n_categories = rating_matrix.shape
n_raters = rating_matrix[0].sum()
# 每个样本的一致程度
p_i = (np.sum(rating_matrix ** 2, axis=1) - n_raters) / \
(n_raters * (n_raters - 1))
p_bar = np.mean(p_i)
# 每个类别的边际概率
p_j = np.sum(rating_matrix, axis=0) / (n_subjects * n_raters)
p_e_bar = np.sum(p_j ** 2)
kappa = (p_bar - p_e_bar) / (1 - p_e_bar)
return kappa实践建议。
新建分类体系后，建议先让 3 位标注员对 100 条 bad case 独立分类，计算 Fleiss' Kappa。如果 κ < 0.6，说明分类定义不够清晰，需要改进标注指南并增加分类边界的示例。不要急着用它来治理数据。
02.6 构建自动分类系统
当 bad case 量增大后，纯人工分类不可持续。企业需要一套自动分类 + 人工审核的混合系统：
python
class BadCaseClassifier:
"""基于规则 + LLM 的两级分类器"""
def __init__(self, rule_engine, llm_classifier):
self.rule_engine = rule_engine
self.llm_classifier = llm_classifier
self.confidence_threshold = 0.85
def classify(self, signal: QualitySignal) -> ClassificationResult:
# 第一级：规则引擎（快速、确定性高）
rule_result = self.rule_engine.match(signal)
if rule_result and rule_result.confidence > 0.95:
return rule_result
# 第二级：LLM 分类（覆盖面广、需要信心阈值）
llm_result = self.llm_classifier.classify(signal)
if llm_result.confidence > self.confidence_threshold:
return llm_result
# 低信心样本进入人工审核队列
return ClassificationResult(
category="needs_review",
confidence=llm_result.confidence,
suggested_category=llm_result.category
)
class RuleEngine:
"""基于模式匹配的快速分类器"""
RULES = [
# 格式类错误：可以用确定性规则
{"pattern": r"json.*parse.*error|invalid.*json",
"category": CaseCategory.FORMAT_ERROR,
"confidence": 0.98},
# 工具调用错误
{"pattern": r"tool.*timeout|function.*failed|api.*error",
"category": CaseCategory.TOOL_ERROR,
"confidence": 0.95},
# 延迟异常
{"pattern": r"ttft.*>.*\d{4}ms|timeout.*exceeded",
"category": CaseCategory.LATENCY,
"confidence": 0.97},
]
def match(self, signal):
text = str(signal.raw_payload).lower()
for rule in self.RULES:
if re.search(rule["pattern"], text):
return ClassificationResult(
category=rule["category"],
confidence=rule["confidence"]
)
return None自动分类的准确率基线。
根据实践经验，规则引擎对格式错误和工具错误的准确率可达 95%+，但对"幻觉 vs 接地失败"这类语义边界模糊的分类，规则引擎很难覆盖。这部分需要依赖 LLM 分类器 + 人工审核兜底。整体目标：自动分类准确率 > 85%，人工审核覆盖 < 20% 的样本。
03. Evalset 为什么必须分层
L0 冒烟集
**规模：**10-30 条 **定位：**发布前必跑，覆盖 P0 基本可用性。任何一条失败都应阻断发布。 **更新频率：**变动极低，只有出现新的 P0 级别场景才会新增。 **维护者：**测试 Tech Lead。
L1 Golden Set
**规模：**50-200 条 **定位：**业务核心路径，持续稳定维护。覆盖所有 P0/P1 业务场景。 **更新频率：**月度审查，线上 P0 bad case 自动提升。 **维护者：**测试团队 + 业务 Owner。
L2 对抗集
**规模：**100-500 条 **定位：**安全、越狱、边界、长上下文、极端格式。红队持续补充。 **更新频率：**安全专项后更新，或发现新攻击模式时。 **维护者：**安全团队 + 红队。
L3 线上采样集
**规模：**500-5000 条 **定位：**反映最新真实分布，滚动更新，不一定每次发布都全跑。 **更新频率：**每周从线上采样补充，按窗口淘汰过期样本。 **维护者：**质量运营团队。 分层的目的不是管理方便，而是保证成本和覆盖平衡。你不可能每次提交都跑全量线上采样集，但也不能只跑一个老化的 Golden Set 假装安全。
3.1 各层运行策略应该怎么配
| 层级 | 推荐频率 | 谁关注 | 典型目标 | 超时/失败处理 |
|---|---|---|---|---|
| L0 冒烟集 | 每次提交 / 每次部署 | 研发、测试 | 先挡住 P0 明显故障 | 任一失败即阻断 |
| L1 Golden Set | 每日 / 每个版本 | 测试、业务负责人 | 保证主链路质量稳定 | 通过率 < 95% 触发复查 |
| L2 对抗集 | 发版前 / 安全专项 | 安全、平台 | 查极端边界和越狱风险 | 任何安全类失败即阻断 |
| L3 线上采样集 | 每周 / 每月滚动 | 质量运营、产品 | 持续跟上真实分布变化 | 用于趋势分析而非门禁 |
3.2 分层评测的执行框架
python
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
class EvalRunner:
"""分层评测执行器"""
def __init__(self, model_client, judge):
self.model_client = model_client
self.judge = judge
def run_evalset(self, samples: list[EvalSample],
level: EvalsetLevel,
concurrency: int = 5) -> EvalReport:
results = []
start_time = time.time()
with ThreadPoolExecutor(max_workers=concurrency) as pool:
futures = {
pool.submit(self._eval_single, s): s
for s in samples
}
for future in as_completed(futures):
sample = futures[future]
try:
result = future.result(timeout=60)
results.append(result)
except Exception as e:
results.append(EvalResult(
sample_id=sample.id,
passed=False,
error=str(e),
category="execution_error"
))
elapsed = time.time() - start_time
return self._build_report(results, level, elapsed)
def _eval_single(self, sample: EvalSample) -> EvalResult:
response = self.model_client.generate(
input_text=sample.input_text,
context=sample.context
)
judgment = self.judge.evaluate(
input_text=sample.input_text,
response=response,
expected=sample.expected_behavior,
must_not_claim=sample.must_not_claim,
must_contain=sample.must_contain
)
return EvalResult(
sample_id=sample.id,
passed=judgment.passed,
score=judgment.score,
category=sample.category.value,
failure_reason=judgment.reason if not judgment.passed else None
)
def _build_report(self, results, level, elapsed) -> EvalReport:
total = len(results)
passed = sum(1 for r in results if r.passed)
# 按类别统计
by_category = {}
for r in results:
cat = r.category
if cat not in by_category:
by_category[cat] = {"total": 0, "passed": 0}
by_category[cat]["total"] += 1
if r.passed:
by_category[cat]["passed"] += 1
return EvalReport(
level=level,
total=total,
passed=passed,
pass_rate=passed / total if total > 0 else 0,
by_category=by_category,
elapsed_seconds=elapsed,
gate_decision=self._gate_decision(level, passed / total)
)
def _gate_decision(self, level, pass_rate) -> str:
thresholds = {
EvalsetLevel.L0_SMOKE: 1.0, # 100% 通过才放行
EvalsetLevel.L1_GOLDEN: 0.95, # 95% 通过
EvalsetLevel.L2_ADVERSARIAL: 1.0, # 安全类不允许失败
EvalsetLevel.L3_SAMPLING: 0.0, # 仅用于趋势，不做门禁
}
threshold = thresholds.get(level, 0.95)
return "PASS" if pass_rate >= threshold else "BLOCK"03.5 一个可长期维护的样本字段规范
json
{
"id": "online-2026-04-001",
"source": "thumb_down",
"category": "grounding_miss",
"input": "试用期员工可以请年假吗？",
"context": "多轮对话历史 + RAG 检索结果",
"expected_behavior": "明确说明试用期内不享受年假",
"priority": "P0",
"tags": ["hr", "policy", "online_case"],
"model_version": "gpt-x.y",
"prompt_version": "prompt-v18",
"judge_rubric_version": "rubric-v6",
"must_not_claim": ["试用期员工享受年假"],
"must_contain": ["转正后", "按入职时间折算"],
"gold_evidence": ["hr-policy-v3.2-section-4.2"],
"evalset_level": "l1_golden",
"owner": "hr-qa",
"created_at": "2026-04-10T14:30:00Z",
"expires_at": "2026-10-10T00:00:00Z",
"review_status": "approved",
"reviewer": "qa-lead-zhang"
}字段规范的价值在于：后面你做 slice、趋势分析、版本归因时，不会因为"大家记录方式不一样"而无法统计。
3.3 必需字段 vs 推荐字段
| 字段 | 是否必需 | 用途 |
|---|---|---|
| id | 必需 | 唯一标识，用于追踪和去重 |
| input | 必需 | 测试输入 |
| expected_behavior | 必需 | Judge 评判的依据 |
| category | 必需 | 分类，用于 slice 分析 |
| priority | 必需 | 决定分层路由和告警级别 |
| context | 推荐 | 多轮历史、RAG 上下文 |
| must_not_claim | 推荐 | 明确的负面约束 |
| gold_evidence | 推荐 | 正确答案的来源证据 |
| model_version | 推荐 | 发现问题时的模型版本 |
| expires_at | 推荐 | 样本的有效期，过期需重新审核 |
3.4 哪些样本不该入库
| 样本类型 | 为什么不建议直接入库 | 更好的做法 |
|---|---|---|
| 上下文不完整的投诉截图 | 无法复现，后续没法稳定回归 | 先回捞原始会话、证据和版本信息 |
| 同一问题的几十条重复抱怨 | 会让某个点被过度放大 | 聚类后保留代表样本 |
| 涉及敏感信息未脱敏的原始日志 | 有合规与隐私风险 | 脱敏后再做样本标准化 |
| 没有通过标准的"感觉不对"样本 | 会让 judge 和人工都无所适从 | 补 expected_behavior 或先进入待澄清池 |
| 已过期的政策/文档相关样本 | 可能测出的"错误"其实是正确的 | 先确认当前政策版本，再决定是否入库 |
03.6 样本的生命周期管理
评测样本不是入库了就永远有效。企业的业务规则会变、文档会更新、模型能力也在迭代。所以样本需要完整的生命周期管理：
样本状态流转：
created → under_review → approved → active → expired/retired
↓ ↓
rejected needs_update
↓
under_review (再审)样本衰减与淘汰策略
L3 线上采样集需要滚动更新，否则会逐渐偏离真实分布。推荐的淘汰策略：
- **时间窗口淘汰：**超过 90 天未触发的样本进入"待复查"，超过 180 天降级或淘汰。
- **通过率淘汰：**连续 5 个版本 100% 通过的样本，考虑降级到更低频运行层。
- **分布对齐淘汰：**如果该类别在线上占比已大幅下降，相应减少评测集中的比例。
python
class SampleLifecycleManager:
"""样本生命周期管理器"""
def audit_samples(self, evalset: list[EvalSample],
recent_results: list[EvalResult]) -> AuditReport:
to_retire = []
to_review = []
for sample in evalset:
# 检查过期
if sample.expires_at and datetime.now() > sample.expires_at:
to_retire.append((sample, "expired"))
continue
# 检查连续通过
sample_results = [r for r in recent_results
if r.sample_id == sample.id]
if len(sample_results) >= 5:
recent_5 = sample_results[-5:]
if all(r.passed for r in recent_5):
to_review.append((sample, "always_passing"))
# 检查连续失败（可能是样本本身有问题）
if len(sample_results) >= 3:
recent_3 = sample_results[-3:]
if not any(r.passed for r in recent_3):
to_review.append((sample, "always_failing"))
return AuditReport(
total_samples=len(evalset),
to_retire=to_retire,
to_review=to_review
)04. 版本治理：模型、Prompt、数据集必须一起记账
企业最常见的追责失败场景是：发现指标下降后，谁也说不清到底是模型换了、Prompt 改了、知识库更新了还是数据集变了。解决方式就是把版本治理做完整。
4.1 一次评测记录应该包含的完整元数据
json
{
"run_id": "eval-2026-04-10-001",
"timestamp": "2026-04-10T15:30:00Z",
"trigger": "ci_pipeline",
"versions": {
"model": "gpt-4o-2026-03-15",
"prompt": "prompt-v18",
"evalset": "golden-set-v12",
"embedding": "text-embedding-3-large",
"retriever": "retriever-v5",
"judge": "judge-v6",
"knowledge_base": "kb-2026-04-08"
},
"parameters": {
"temperature": 0.2,
"top_p": 0.9,
"max_tokens": 2048,
"seed": 42
},
"results": {
"total": 150,
"passed": 143,
"pass_rate": 0.953,
"by_category": {
"grounding_miss": {"total": 30, "passed": 28},
"hallucination": {"total": 25, "passed": 24},
"format_error": {"total": 20, "passed": 20},
"refusal": {"total": 15, "passed": 14},
"tool_error": {"total": 10, "passed": 9},
"general_qa": {"total": 50, "passed": 48}
}
},
"gate_decision": "PASS",
"elapsed_seconds": 342
}如果你只记模型版本，不记 Prompt 和 evalset 版本，那么一次"看起来像模型退化"的现象，可能其实只是：Prompt 换了表达方式、知识库更新了 300 篇文档、judge rubric 从 v5 改到了 v6。版本治理的意义，就是把这些变化拆开，避免错误归因。
04.5 漂移检测与自动归因
当评测分数发生变化时，如何自动定位是哪个变量导致的？这是版本治理最有价值的部分。
分布漂移检测
用统计方法检测评测分数是否发生了显著变化。最常用的方法是 Bootstrap 置信区间和假设检验：
python
import numpy as np
from scipy import stats
class DriftDetector:
"""评测分数漂移检测器"""
def detect_score_drift(self,
scores_baseline: list[float],
scores_current: list[float],
alpha: float = 0.05) -> DriftResult:
"""
检测两批评测分数是否有统计显著差异
使用 Mann-Whitney U 检验（非参数，不假设正态分布）
"""
statistic, p_value = stats.mannwhitneyu(
scores_baseline, scores_current,
alternative='two-sided'
)
# 效果量 (Cohen's d 近似)
mean_diff = np.mean(scores_current) - np.mean(scores_baseline)
pooled_std = np.sqrt(
(np.var(scores_baseline) + np.var(scores_current)) / 2
)
effect_size = mean_diff / pooled_std if pooled_std > 0 else 0
# Bootstrap 置信区间
ci_low, ci_high = self._bootstrap_ci(
scores_baseline, scores_current, n_bootstrap=10000
)
return DriftResult(
is_significant=p_value < alpha,
p_value=p_value,
effect_size=effect_size,
mean_baseline=np.mean(scores_baseline),
mean_current=np.mean(scores_current),
ci_95=(ci_low, ci_high),
interpretation=self._interpret(effect_size, p_value, alpha)
)
def _bootstrap_ci(self, baseline, current,
n_bootstrap=10000, ci=0.95):
"""Bootstrap 差值的置信区间"""
diffs = []
for _ in range(n_bootstrap):
b_sample = np.random.choice(baseline, len(baseline), replace=True)
c_sample = np.random.choice(current, len(current), replace=True)
diffs.append(np.mean(c_sample) - np.mean(b_sample))
lower = np.percentile(diffs, (1 - ci) / 2 * 100)
upper = np.percentile(diffs, (1 + ci) / 2 * 100)
return lower, upper
def _interpret(self, effect_size, p_value, alpha):
if p_value >= alpha:
return "无统计显著差异，可能是正常波动"
if abs(effect_size) < 0.2:
return "统计显著但效果量小，密切关注但无需紧急处理"
if abs(effect_size) < 0.5:
return "中等退化，建议排查最近的变更"
return "严重退化，立即排查并考虑回滚"自动归因分析
当检测到漂移后，需要自动定位根因。核心思路是隔离变量：
python
class DriftAttributor:
"""漂移归因分析器"""
def attribute(self, current_run, baseline_run) -> Attribution:
changed_components = []
# 对比所有版本号
for key in current_run.versions:
if current_run.versions[key] != baseline_run.versions[key]:
changed_components.append(key)
if len(changed_components) == 0:
return Attribution(
likely_cause="non_determinism",
explanation="所有组件版本一致，差异可能来自模型非确定性或环境变化"
)
if len(changed_components) == 1:
return Attribution(
likely_cause=changed_components[0],
explanation=f"仅 {changed_components[0]} 发生变更，"
f"从 {baseline_run.versions[changed_components[0]]} "
f"变为 {current_run.versions[changed_components[0]]}"
)
# 多个组件同时变更，需要按 slice 做差异分析
slice_analysis = self._analyze_by_slice(current_run, baseline_run)
return Attribution(
likely_cause="multiple",
changed_components=changed_components,
slice_analysis=slice_analysis,
recommendation="建议分别回滚各组件，逐一排除"
)
def _analyze_by_slice(self, current, baseline) -> dict:
"""按类别分析哪些 slice 退化最严重"""
analysis = {}
for category in current.results.by_category:
curr = current.results.by_category[category]
base = baseline.results.by_category.get(category, curr)
curr_rate = curr["passed"] / curr["total"]
base_rate = base["passed"] / base["total"]
delta = curr_rate - base_rate
if abs(delta) > 0.05: # 5% 以上的变化
analysis[category] = {
"baseline_rate": base_rate,
"current_rate": curr_rate,
"delta": delta,
"severity": "high" if abs(delta) > 0.1 else "medium"
}
return analysis05. 数据污染与评测风险
- **训练集泄露：**测试集被训练数据覆盖，导致评测虚高。模型"记住"了答案而不是真正理解了问题。
- **隐私泄露：**线上日志直接回流，泄露敏感信息或个人数据。在 GDPR/个保法约束下可能构成合规风险。
- **对抗样本混入：**用户绕过样本混入普通业务集，拉低整体可解释性。
- **过度代表：**同一 bad case 重复过多，导致系统过度优化单点问题。
- **标注偏差：**标注员的个人偏好渗入评判标准，导致"分数好看但不代表用户真实感知"。
治理原则。
脱敏优先、去重优先、按风险和频率双维度选样，不追求样本越多越好，而追求样本越代表真实线上问题越好。
5.1 训练集泄露检测
如何检测测试集是否被训练数据"污染"？这是一个深层次的问题。常用方法包括：
N-gram 重叠检测
计算测试集样本与已知训练数据之间的 n-gram 重叠率。如果 8-gram 重叠率 > 80%，高度怀疑泄露。
Perplexity 异常检测
如果模型对某些测试样本的困惑度异常低（远低于同类问题的均值），可能已经在训练中见过这些数据。
Membership Inference
训练一个二分类器来判断某条数据是否是训练集成员。基于模型输出的 logits 分布特征进行判断。
Canary Token 方法
在训练数据中故意插入特殊标记（canary），测试时检查模型是否能"回忆"这些标记来确认数据边界。
5.2 一个"好心办坏事"的反例
团队发现退款政策错了很多，
于是把同一个退款问题的 200 条投诉全部塞进 Golden Set。
结果：
1. Golden Set 被单一问题主导（退款相关占 40%）
2. 团队把大量精力都花在同一类 prompt 微调上
3. 其他高风险但低频的问题被整体淹没
4. 回归通过率从 85% 飙升到 95%，但线上投诉没有减少
5. 管理层误以为质量大幅提升
这不是治理，
而是把线上情绪直接投射成测试权重。
正确做法：
- 聚类后保留 5-10 条代表性退款样本
- 同时增加其他高风险类别的覆盖
- 保持各类别的比例大致反映线上真实分布05.5 采样策略的数学基础
线上采样不是随机抓几条，而是按策略抽。要理解采样策略，需要了解一些统计学基础。
样本量估算
如果要估计某个类别的错误率，需要多少样本才能达到期望的置信度？ n = (Z² × p × (1-p)) / E² 其中：
Z= 对应置信水平的 Z 值（95% 置信度时 Z = 1.96）p= 估计的错误率（如果未知，取 0.5 最保守）E= 可接受的误差范围
# 样本量计算示例
import math
def required_sample_size(confidence: float = 0.95,
error_rate: float = 0.5,
margin: float = 0.05) -> int:
"""
计算达到指定置信度所需的最小样本量
confidence: 置信水平 (0.95 = 95%)
error_rate: 估计的错误率
margin: 可接受的误差范围
"""
z_scores = {0.90: 1.645, 0.95: 1.96, 0.99: 2.576}
z = z_scores.get(confidence, 1.96)
n = (z**2 * error_rate * (1 - error_rate)) / margin**2
return math.ceil(n)
# 场景1: 估计线上错误率，95%置信度，±5%误差
print(required_sample_size(0.95, 0.1, 0.05)) # → 139 条
print(required_sample_size(0.95, 0.5, 0.05)) # → 385 条（最保守）
# 场景2: 更严格的要求，99%置信度，±3%误差
print(required_sample_size(0.99, 0.1, 0.03)) # → 664 条分层采样 (Stratified Sampling)
线上流量通常分布极不均匀。如果纯随机采样，低频但高风险的类别可能完全被忽略。分层采样可以解决这个问题：
| 采样策略 | 用途 | 风险 | 适用场景 |
|---|---|---|---|
| 纯随机采样 | 观察整体分布 | 高风险低频问题容易漏掉 | 初期探索性分析 |
| 按投诉/点踩加权 | 更快捕捉真实痛点 | 可能过度代表情绪强烈用户 | 快速响应用户反馈 |
| 按业务 slice 定向 | 保护 P0 场景 | 对整体趋势代表性不足 | 关键业务链路覆盖 |
| 分层采样 | 兼顾各类别代表性 | 需要预先知道分层维度 | 长期质量运营 |
| 对抗定向采样 | 发现新的攻击模式 | 偏向异常而非正常流量 | 安全专项 |
# 分层采样实现
import random
from collections import defaultdict
def stratified_sample(signals: list[QualitySignal],
target_total: int,
strata_weights: dict) -> list[QualitySignal]:
"""
按类别分层采样
strata_weights: 各类别的权重配比
例如:
strata_weights = {
"high_value_biz": 0.40, # 高价值业务 slice
"user_complaint": 0.30, # 用户投诉/点踩
"random_traffic": 0.20, # 纯随机流量
"new_feature": 0.10, # 新功能/新文档
}
"""
# 按类别分组
by_strata = defaultdict(list)
for signal in signals:
stratum = classify_stratum(signal)
by_strata[stratum].append(signal)
result = []
for stratum, weight in strata_weights.items():
n_samples = int(target_total * weight)
pool = by_strata.get(stratum, [])
if len(pool) <= n_samples:
result.extend(pool)
else:
result.extend(random.sample(pool, n_samples))
return result5.3 一个更像企业方案的采样配方
每周新增样本 =
40% 高价值业务 slice
+ 30% 用户投诉 / 点踩
+ 20% 纯随机流量
+ 10% 新功能或新文档专项
这样做的好处：
1. 不会只追着投诉跑
2. 也不会因为纯随机而漏掉关键风险
3. 可以兼顾真实分布和业务优先级
4. 新功能专项保证新上线的能力有持续覆盖
实际操作中的调整：
- 如果某个类别连续 4 周采样后通过率 > 98%，
可以适当降低该类别的权重
- 如果发现新的高频失败模式，
可以临时提升对应类别的采样比例
- 每月底做一次采样配方复审05.6 去重与脱敏的工程实现
在样本入库前，去重和脱敏是两个必须做到位的环节。
语义去重
纯文本完全匹配的去重是不够的。用户可能用不同的措辞表达完全相同的问题。需要基于语义相似度做去重：
python
from sentence_transformers import SentenceTransformer
import numpy as np
from sklearn.cluster import DBSCAN
class SemanticDeduplicator:
"""基于语义相似度的去重器"""
def __init__(self, model_name="BAAI/bge-large-zh-v1.5"):
self.model = SentenceTransformer(model_name)
self.similarity_threshold = 0.92
def deduplicate(self, samples: list[EvalSample]) -> list[EvalSample]:
texts = [s.input_text for s in samples]
embeddings = self.model.encode(texts, normalize_embeddings=True)
# 计算余弦相似度矩阵
sim_matrix = embeddings @ embeddings.T
# 用 DBSCAN 聚类
distance_matrix = 1 - sim_matrix
clustering = DBSCAN(
eps=1 - self.similarity_threshold,
min_samples=1,
metric="precomputed"
).fit(distance_matrix)
# 每个簇保留优先级最高的样本
clusters = {}
for idx, label in enumerate(clustering.labels_):
if label not in clusters:
clusters[label] = []
clusters[label].append(idx)
selected = []
for label, indices in clusters.items():
cluster_samples = [samples[i] for i in indices]
# 优先级排序：P0 > P1 > P2 > P3
best = min(cluster_samples, key=lambda s: s.priority)
selected.append(best)
return selected脱敏处理
python
import re
class DataAnonymizer:
"""敏感信息脱敏器"""
PATTERNS = [
(r'\b1[3-9]\d{9}\b', '[PHONE]'), # 手机号
(r'\b\d{6}(19|20)\d{2}(0[1-9]|1[0-2]).*\b', '[ID_CARD]'), # 身份证
(r'\b[\w.+-]+@[\w-]+\.[\w.-]+\b', '[EMAIL]'), # 邮箱
(r'\b\d{16,19}\b', '[BANK_CARD]'), # 银行卡号
(r'[\u4e00-\u9fa5]{2,4}(?=先生|女士|同学)', '[NAME]'), # 中文姓名
]
def sanitize(self, payload: dict) -> dict:
"""递归脱敏字典中的所有字符串值"""
if isinstance(payload, str):
return self._sanitize_text(payload)
if isinstance(payload, dict):
return {k: self.sanitize(v) for k, v in payload.items()}
if isinstance(payload, list):
return [self.sanitize(item) for item in payload]
return payload
def _sanitize_text(self, text: str) -> str:
for pattern, replacement in self.PATTERNS:
text = re.sub(pattern, replacement, text)
return text06. 企业实战案例：从 0 到 1 搭建闭环
6.1 案例背景
某金融科技公司的 AI 客服系统，日均处理 5 万次对话，上线 3 个月后发现以下问题：
- 初始评测集 200 条，3 个月未更新，回归通过率稳定在 92%，但用户投诉持续增加。
- 新上线的"贷款计算器"功能完全没有评测覆盖。
- 知识库更新了 500 篇新文档，评测集完全没有反映这些变化。
- 不同标注员对"是否算幻觉"的判断差异很大。
6.2 改造方案
| 阶段 | 时间 | 关键动作 | 产出 |
|---|---|---|---|
| 诊断期 | 第 1 周 | 分析线上 bad case 分布、计算标注一致性 | 问题分类报告，Kappa = 0.48 |
| 标准化 | 第 2-3 周 | 设计分类体系、样本规范、标注指南 | 8 类分类 + 字段规范 + 50 个标注示例 |
| 工程建设 | 第 4-6 周 | 搭建信号采集、自动分类、管线系统 | Pipeline + 自动分类准确率 87% |
| 分层落地 | 第 7-8 周 | 建立 L0-L3 四层评测集 | L0=20, L1=120, L2=80, L3=500 |
| 持续运营 | 第 9 周起 | 每周采样、月度审查、季度大盘 | Kappa 提升到 0.76，投诉下降 35% |
6.3 关键指标变化
改造前 vs 改造后（3 个月对比）：
标注一致性 (Kappa): 0.48 → 0.76 (+58%)
评测集覆盖率: 12% → 67% (基于线上 intent 分类)
L1 Golden Set 通过率: 92% → 89% (看似"下降"，实际是评测更真实了)
线上用户投诉率: 3.2% → 2.1% (-34%)
Bad Case 平均修复周期: 14天 → 5天 (-64%)
新功能评测覆盖延迟: 从未覆盖 → 上线 1 周内覆盖
核心洞察：
通过率"下降"恰恰证明了旧评测集已经失真。
真正的质量信号不是评测分数高不高，而是线上投诉有没有降。07. 落地清单
- 建立线上 bad case 收集入口，统一字段规范，确保所有信号源归一。
- 定义缺陷 taxonomy，避免大家随手命名，确保 Kappa > 0.6。
- 为样本建立优先级、标签、来源、版本信息和有效期。
- 把 L0/L1/L2/L3 四层集分开运行和报告，配置不同的频率和门禁阈值。
- 每周至少一次从线上采样集里筛选新样本入库，使用分层采样策略。
- 部署自动质量检测器（事实一致性、拒答异常、格式完整性）。
- 建立版本治理机制，每次评测记录完整的组件版本信息。
- 配置漂移检测和自动归因，变化超过阈值时自动告警。
- 实现语义去重和脱敏管线，确保入库样本质量和合规。
- 月度审查样本生命周期，淘汰过期样本，补充新场景覆盖。
7.1 课堂练习
- 把一条用户投诉转成可入库样本，补齐你认为必需的所有字段，并说明每个字段的设计理由。
- 为一个企业 AI 助手设计 L0/L1/L2/L3 四层 evalset 的样本来源、运行频率和门禁策略。
- 举 3 类"不该直接进评测集"的线上样本，并说明原因和正确处理方式。
- 设计一个采样配方，说明各类别的权重、调整规则和月度复审标准。
- 用 Cohen's Kappa 评估一组标注结果的一致性，如果 κ < 0.6，给出改进建议。
7.2 参考答案要点
- 可入库样本至少应有：输入、必要上下文、期望行为、失败类型、优先级、来源、版本信息和责任归属。每个字段都要有明确的下游消费者。
- 分层设计的核心不是名称，而是频率、成本和风险级别对应起来。L0 保证不出 P0 故障，L1 保证主链路，L2 保证安全，L3 跟踪分布。
- 不该直接入库的样本通常具备三种特征：不可复现（缺上下文）、未脱敏（合规风险）、无明确通过标准（无法自动化评判）。
- 采样配方要兼顾业务优先级和真实分布，每月根据通过率趋势和线上分布变化动态调整权重。
- Kappa 低于 0.6 时，通常的改进方向：增加标注示例（特别是边界 case 的示例）、简化分类体系（合并区分度低的类别）、增加标注员培训。
7.3 自测标准
学完这一页后，你应该能：
- 把线上问题转成可回归用例，而不是只截图留档。
- 设计分层 evalset，而不是"一套数据打天下"。
- 知道为什么版本治理必须覆盖模型、Prompt 和数据集。
- 识别数据污染和过拟合式优化的风险。
- 用统计方法度量标注一致性和检测分数漂移。
- 设计并实现从信号采集到入库的完整 Pipeline。
- 运用分层采样策略平衡覆盖和成本。