在线评估与 A/B:生产环境的真话

离线 vs 在线:为什么差距永远存在

"离线测试过"不等于"线上能跑"。常见差异来源:

三种在线评估方法

① Shadow Mode(影子模式)

用户感知不到,后台新版本和旧版本同时跑,只发老版本的输出,记录两者 diff。风险最低,最适合灰度前验证。

import asyncio, random, structlog

log = structlog.get_logger()

async def handle_request(user_id, query, sample_rate=0.1):
    # 主调用(给用户的)
    old_task = asyncio.create_task(old_system(query))
    # 影子调用(只采样 10%,不返回给用户)
    new_task = None
    if random.random() < sample_rate:
        new_task = asyncio.create_task(new_system(query))

    result = await old_task

    if new_task:
        try:
            new_result = await asyncio.wait_for(new_task, timeout=30)
            log.info("shadow_diff", user=user_id, old=result, new=new_result, query=query)
        except Exception as e:
            log.warn("shadow_fail", err=str(e))

    return result

② Canary / 逐步放量

按比例分流:1% → 5% → 25% → 100%,每一档观察关键指标,有异常立即回滚。

ROLLOUT = {"v7": 0.05}  # v7 拿 5% 流量,其余走 v6

def pick_variant(user_id):
    import hashlib
    h = int(hashlib.md5(user_id.encode()).hexdigest(), 16) / 2**128
    return "v7" if h < ROLLOUT["v7"] else "v6"

③ A/B 实验

严格对比两个/多个版本在同期流量上的表现。要做到统计显著性。

核心:用户反馈信号

常用隐式信号

• Regenerate 率 — 用户点"重新生成"说明不满意
• Edit 率 — 生成代码后用户立即改的,说明输出偏差
• Copy 率 — 用户复制输出,说明满意(最可靠的正信号)
• Session 追问 — "不对,我要的是..." 说明第一次没答对
• 停留时长 / 滚动深度 — 读完 vs 直接关
• Abandonment — 输出中途跳走

# 隐式信号的"正/负"聚合示例
def implicit_signal_score(events):
    positive = (
        events.get("copy", 0) * 1.0 +
        events.get("like", 0) * 1.5 +
        events.get("follow_up_positive", 0) * 0.8
    )
    negative = (
        events.get("regenerate", 0) * 1.0 +
        events.get("dislike", 0) * 1.5 +
        events.get("correction", 0) * 1.2 +
        events.get("abandon", 0) * 0.5
    )
    return positive - negative

A/B 实验设计

关键决策点

1. 主指标(primary metric):必须事前决定一个,一个实验一个主指标。例如 CSAT、成功率、copy 率
2. 护栏指标(guardrail):不能退的指标。例:延迟 p95、成本、refusal 率。主指标涨但护栏破 → 不上
3. 样本量估算:能不能检出你期望的效应?太小会"看起来没效应"
4. 最小运行时长:至少覆盖一个完整使用周期(通常 1 周,消除 weekly seasonality)
5. 事前注册:把实验假设、主/护栏、停止规则写文档,防止事后捡好数字

样本量计算

from statsmodels.stats.power import NormalIndPower
from statsmodels.stats.proportion import proportion_effectsize

baseline, target = 0.70, 0.73   # 期望 3pp 提升
effect = proportion_effectsize(baseline, target)
n_per_arm = NormalIndPower().solve_power(
    effect=effect, alpha=0.05, power=0.8, alternative="two-sided"
)
print(round(n_per_arm))  # 约 3400 per arm ≈ 两组共 ~7000

显著性检验

from statsmodels.stats.proportion import proportions_ztest

# v6 组: 3000 次请求, 2100 次成功
# v7 组: 3000 次请求, 2200 次成功
successes = [2100, 2200]
nobs      = [3000, 3000]
z, p = proportions_ztest(successes, nobs)
print(f"z={z:.2f}, p={p:.4f}")
# 若 p < 0.05 且满足护栏 → v7 胜出

Novelty Effect 与 Primacy Effect

新功能上线,用户可能因"新鲜感"多用(Novelty),或因"不熟悉"少用(Primacy)。短期指标可能完全不反映长期稳态。

• 至少运行 1 周,最好 2 周,观察趋势是否稳定
• 分 new user / existing user 看,两群的反应可能很不同
• 关注留存指标(7 日回访、次周使用),比当日点击更长期可靠

在线 LLM Judge:采样评估

生产流量太大,不可能每条跑 Judge。典型方案:按 session 采样 1-5%,离线跑 Judge,看分布。

import random

def maybe_sample_for_eval(trace, rate=0.02):
    if random.random() < rate:
        eval_queue.put({
            "trace_id": trace.id,
            "input": trace.input,
            "output": trace.output,
            "variant": trace.variant,
            "ts": trace.ts,
        })

# 异步 Judge worker
async def judge_worker():
    while True:
        item = await eval_queue.get()
        scores = await run_judge(item["input"], item["output"])
        write_to_db(item["trace_id"], scores)

Guardrail 检查:实时守门

除了定期对比,还要有实时异常告警:

• p95 latency > 阈值 → page oncall
• 错误率 > 2% → 回滚
• refusal 率突增 → 审查 prompt 注入或系统变更
• 日成本 > 预算 → 自动限流
• 毒性/PII 检测率上升 → 触发安全流程

实战:一个完整在线评估系统骨架

本章小结