1. 为什么离线评测跑满分,上线还是被用户骂
很多团队做 LLM 评测的流程是这样的:准备一份 MMLU 或者自建测试集,跑一遍 BLEU、ROUGE,再让 GPT-4 打个分,分数过了 0.85 就上线。结果上线第二天,客服群里全是截图——模型答非所问、上下文断裂、延迟飙到 8 秒。问题出在哪?离线测试集永远覆盖不了线上长尾分布:用户真实提问的措辞、上下文长度、领域混合度,跟固定测试集完全是两个世界。更麻烦的是指标本身和用户体验弱相关,一个回答 BLEU 0.85 但带事实性错误,用户直接点踩;另一个转述略有不同但正确友好,用户点赞。
所以在线评测不是"锦上添花",而是 LLM 上线后的必选项。它直接用真实用户流量,通过 A/B 实验比较两个模型版本在回答质量、延迟、对话连贯性上的差异。离线评测当 Gate(通过率低于 90% 不上线),在线 A/B 做最终验证。这篇文章我会把整套闭环拆开:MLflow 追踪实验版本、Prometheus 采集线上指标、A/B 分流、回归自动化告警,每一步都给可复制的配置和脚本。适合正在做 LLM 应用迭代、被"上线即翻车"困扰的工程团队。
2. 前置准备:TaoToken 接入与实验环境搭建
在搭评测体系之前,得先有一个稳定的模型调用入口。我这边用 TaoToken 做统一接入,它兼容 OpenAI 的接口格式,MLflow 里记录实验、Prometheus 里打标签都不需要额外适配。官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,注册后在控制台生成 API Key 即可。
拿到 Key 之后,先确认两件事:一是模型列表里有哪些可用版本(A/B 实验的对照组和实验组要选不同版本),二是接口的 base_url 和超时设置。TaoToken 的 API 地址是 https://taotoken.net/api ,不带 UTM 参数,直接填到环境变量里。
export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"验证一下连通性:
import os from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"] ) resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "用一句话解释什么是A/B测试"}], temperature=0.0 ) print(resp.choices[0].message.content)如果返回正常,说明接入没问题。接下来装实验依赖:
pip install mlflow prometheus-client statsmodels scipy openaiMLflow 用来追踪每次实验的参数、指标和 artifact;prometheus-client 在模型服务层暴露 metrics;statsmodels 和 scipy 做显著性检验。这套组合的好处是全部开源、可自托管,不依赖任何商业实验平台。
注意:API Key 不要硬编码在脚本里,用环境变量或者密钥管理服务。MLflow 的 tracking server 如果多人共用,记得配好认证。
3. 可复制配置:MLflow 实验追踪 + Prometheus 指标暴露
3.1 MLflow 实验配置
MLflow 的核心作用是记录每次模型迭代的"实验快照"——用了哪个模型版本、流量比例多少、跑了多久、最终胜率和延迟是多少。这样回溯的时候不用翻聊天记录。
先启动 tracking server(本地测试用 sqlite 就够):
mlflow server \ --backend-store-uri sqlite:///mlflow.db \ --default-artifact-root ./mlruns \ --host 0.0.0.0 \ --port 5000然后写一个实验记录脚本,每次创建 A/B 实验时调用:
import mlflow import json from datetime import datetime mlflow.set_tracking_uri("http://localhost:5000") mlflow.set_experiment("llm-online-abtest") def log_ab_experiment(exp_id, control_model, treatment_model, traffic_percent, duration_hours): with mlflow.start_run(run_name=f"abtest-{exp_id}"): mlflow.log_params({ "experiment_id": exp_id, "control_model": control_model, "treatment_model": treatment_model, "traffic_percent": traffic_percent, "duration_hours": duration_hours, "start_time": datetime.utcnow().isoformat() }) # 占位,后续由回归脚本回填指标 mlflow.log_metric("win_rate_control", 0.0) mlflow.log_metric("win_rate_treatment", 0.0) mlflow.log_metric("p95_latency_control", 0.0) mlflow.log_metric("p95_latency_treatment", 0.0) mlflow.log_artifact("ab_config.json") return mlflow.active_run().info.run_id log_ab_experiment( exp_id="llm-v2.1.3-abtest-20250301", control_model="gpt-4o-mini", treatment_model="gpt-4o", traffic_percent=5, duration_hours=168 )MLflow 会把参数、指标、artifact 都存下来,后面做基线对比直接查 run 就行。
3.2 Prometheus 指标暴露
模型服务层需要暴露带group和model_version标签的 metrics。核心指标有三个:响应时间 histogram、用户反馈 counter、实时胜率。
from prometheus_client import Counter, Histogram, Gauge, start_http_server import time # 响应时间分布 LLM_LATENCY = Histogram( "llm_response_seconds", "LLM response latency in seconds", ["group", "model_version"], buckets=[0.1, 0.3, 0.5, 1.0, 2.0, 5.0, 10.0] ) # 用户反馈计数 LLM_FEEDBACK = Counter( "llm_user_feedback_total", "User feedback count", ["group", "model_version", "type"] # type: like / dislike ) # 实时胜率(由回归脚本更新) LLM_WIN_RATE = Gauge( "llm_win_rate_ratio", "Real-time win rate", ["group", "model_version"] ) start_http_server(8000) # Prometheus 从这里抓 def handle_request(group, model_version, user_input): start = time.time() # 调用模型 resp = client.chat.completions.create( model=model_version, messages=[{"role": "user", "content": user_input}] ) elapsed = time.time() - start LLM_LATENCY.labels(group=group, model_version=model_version).observe(elapsed) return resp def record_feedback(group, model_version, feedback_type): LLM_FEEDBACK.labels( group=group, model_version=model_version, type=feedback_type ).inc()Prometheus 的 scrape 配置:
scrape_configs: - job_name: 'llm-service' scrape_interval: 15s static_configs: - targets: ['llm-service:8000']3.3 流量分层与分桶
A/B 实验的流量分层用两层结构:第一层 Pre-split 把用户按user_id哈希切成 100 个桶,第二层从这些桶里随机选一部分参与实验。同一个用户的多次请求始终落在同一组,保证一致性。
import hashlib def assign_bucket(user_id, num_buckets=100): h = hashlib.md5(str(user_id).encode()).hexdigest() return int(h, 16) % num_buckets def assign_group(user_id, experiment_id, traffic_percent): bucket = assign_bucket(user_id) # 用 experiment_id 做盐,不同实验的桶分配独立 exp_hash = int(hashlib.md5(experiment_id.encode()).hexdigest(), 16) threshold = int(traffic_percent / 100 * 100) if (bucket + exp_hash) % 100 < threshold: return "treatment" return "control"样本量计算别拍脑袋。假设要检测胜率提升 2%(从 50% 到 52%),α=0.05,power=0.8:
import statsmodels.stats.proportion as smp import numpy as np def min_sample_size(mde=0.02, base_rate=0.5, alpha=0.05, power=0.8): nobs = smp.samplesize_proportions_2indep_onetail( diff=mde, prop2=base_rate, power=power, alpha=alpha, ratio=1.0 ) return int(np.ceil(nobs)) print(f"每组需要: {min_sample_size(mde=0.02)}") # 约 9604 print(f"MDE=5% 时每组需要: {min_sample_size(mde=0.05)}") # 约 1500MDE=2% 时每组要近 1 万个样本,总共 2 万条有效反馈。如果 MDE 放宽到 5%,样本量降到 1500。所以小流量产品别追求 2% 的精度,先定 5% 跑起来。
4. 验证请求:从流量切分到指标对比的完整动作
4.1 回归检查脚本
每 15 分钟从 Prometheus 拉一次指标,计算实验组和对照组的胜率差值及置信区间。如果置信区间完全落在 [-1%, +1%] 之外,触发告警;胜率下降超过 3%,自动回滚。
import requests import numpy as np from scipy.stats import beta PROM_URL = "http://localhost:9090/api/v1/query" def query_prometheus(expr): resp = requests.get(PROM_URL, params={"query": expr}) return resp.json()["data"]["result"] def get_win_rate(exp_id, group): expr = f''' rate(llm_user_feedback_total{{exp="{exp_id}",group="{group}",type="like"}}[1h]) / rate(llm_user_feedback_total{{exp="{exp_id}",group="{group}"}}[1h]) ''' result = query_prometheus(expr) if not result: return None return float(result[0]["value"][1]) def bayesian_probability(n_success_a, n_total_a, n_success_b, n_total_b): """P(实验组胜率 > 对照组胜率)""" a_alpha = 1 + n_success_a a_beta = 1 + n_total_a - n_success_a b_alpha = 1 + n_success_b b_beta = 1 + n_total_b - n_success_b samples_a = beta.rvs(a_alpha, a_beta, size=100000) samples_b = beta.rvs(b_alpha, b_beta, size=100000) return np.mean(samples_a > samples_b) def check_ab_experiment(exp_id, control_n, treatment_n): ctrl_rate = get_win_rate(exp_id, "control") treat_rate = get_win_rate(exp_id, "treatment") if ctrl_rate is None or treat_rate is None: print("指标未就绪,实验继续运行") return ctrl_success = int(ctrl_rate * control_n) treat_success = int(treat_rate * treatment_n) prob = bayesian_probability( treat_success, treatment_n, ctrl_success, control_n ) diff = treat_rate - ctrl_rate print(f"对照组胜率: {ctrl_rate:.4f}, 实验组: {treat_rate:.4f}, " f"差值: {diff:.4f}, P(实验>对照): {prob:.3f}") if prob > 0.95 and diff > 0.01: print("实验组显著优于对照组,建议全量上线") elif prob < 0.05 and diff < -0.03: print("实验组显著差于对照组,触发自动回滚") # auto_rollback(exp_id) else: print("差异不显著,实验继续") check_ab_experiment("llm-v2.1.3-abtest-20250301", control_n=4800, treatment_n=5000)4.2 Prometheus 告警规则
把核心告警写成 Prometheus rule 文件:
groups: - name: llm-abtest-alerts rules: - alert: WinRateDropCritical expr: | (llm_win_rate_ratio{group="control"} - llm_win_rate_ratio{group="treatment"}) > 0.03 for: 15m labels: severity: P0 annotations: summary: "实验组胜率下降超过3%,需立即回滚" description: "对照组 {{ $labels.model_version }} 胜率高于实验组" - alert: LatencyIncreaseWarning expr: | histogram_quantile(0.95, rate(llm_response_seconds_bucket{group="treatment"}[5m])) / histogram_quantile(0.95, rate(llm_response_seconds_bucket{group="control"}[5m])) > 1.2 for: 10m labels: severity: P1 annotations: summary: "实验组P95延迟增加超过20%" description: "需人工确认是否牺牲延迟换质量" - alert: InsufficientSamples expr: | sum(rate(llm_user_feedback_total{group="treatment"}[1h])) < 10 for: 1h labels: severity: P3 annotations: summary: "实验组样本量不足,实验继续运行"4.3 上下文连贯性自动评估
胜率只能反映显式反馈,大多数用户不会主动点赞。上下文连贯性用 LLM-as-Judge 做抽样子集评估,每天 1000 对就够。
import random def judge_coherence(context, response_a, response_b): # 随机交换顺序,消除位置偏差 if random.random() > 0.5: response_a, response_b = response_b, response_a swapped = True else: swapped = False prompt = f"""以下是一个对话历史(只列出最后三轮)。 有两个回答A和B,请判断哪个回答更符合上下文逻辑、不引入矛盾、且延续之前的话题。 输出格式: "A" 或 "B" 或 "tie" 对话上下文: {context} 回答A: {response_a} 回答B: {response_b}""" resp = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], temperature=0.0 ) verdict = resp.choices[0].message.content.strip() if swapped: verdict = {"A": "B", "B": "A"}.get(verdict, verdict) return verdict顺序偏差必须控制,评判模型本身也有偏见(偏好更长、更华丽的回答),所以最好用多个评判模型取多数。成本方面,一个判断约 300-600 tokens,每天 1000 对的话成本可控。
5. 本篇常见错排查
5.1 MLflow 指标回填失败
现象:实验跑完了,MLflow 里win_rate_treatment还是 0。原因通常是回归脚本没有正确关联 run_id。排查步骤:先确认mlflow.active_run()在记录时是否有效,如果脚本是独立进程,需要用mlflow.start_run(run_id=...)显式指定。另外 artifact 路径如果是相对路径,tracking server 和 client 不在同一台机器时会找不到文件,建议用绝对路径或者 S3 兼容存储。
5.2 Prometheus 抓不到指标
现象:llm_response_seconds_bucket在 Prometheus 里查不到。先检查start_http_server(8000)是否在模型服务启动时执行了,再确认 Prometheus 的 targets 页面里 job 状态是不是 UP。如果模型服务跑在容器里,localhost:8000抓不到,要改成容器名或者宿主机 IP。还有一个坑:Histogram 的 buckets 设置不合理,比如最大 bucket 只有 5 秒,但实际延迟经常 8 秒,那 P95 会失真,建议 buckets 覆盖到 10 秒以上。
5.3 胜率计算出现除零
现象:rate(llm_user_feedback_total[1h])返回空,脚本报 ZeroDivisionError。原因是新实验刚创建,还没有用户反馈。处理方式是在get_win_rate里加空值判断,返回 None 让实验继续跑。另外 Prometheus 的rate函数在时间窗口内样本少于两个时会返回空,可以改用increase或者把窗口拉长到 6 小时。
5.4 贝叶斯概率波动大
现象:P(实验>对照)在 0.4 到 0.7 之间反复横跳。这是样本量不足的典型表现,不是代码问题。检查control_n和treatment_n是否达到了最小样本量要求。如果流量太小,要么放宽 MDE,要么延长实验周期。别在样本不足时手动"盯 p 值"停车,假阳性会飙升。
5.5 自动回滚误触发
现象:实验组胜率只降了 1%,但告警触发了。检查告警规则的for时长,15 分钟可能太短,建议改成 30 分钟或者 1 小时。另外胜率下降 3% 的阈值要结合业务定,如果基线胜率本身只有 2%,降 3% 是不可能的,阈值要按相对比例算。回滚动作最好加人工确认环节,P0 告警自动回滚,P1 告警只通知。
6. 把评测闭环跑起来:从工具到习惯
整套体系搭完之后,模型迭代的节奏会明显不一样。以前是"离线跑个分,感觉差不多就上",现在是"离线 Gate 通过 → 灰度 5% 流量 → 每 15 分钟看贝叶斯概率 → 显著优于基线就全量,显著差就回滚"。迭代周期从周级缩到天级,而且每次上线都有统计可信的用户价值提升。
工具链上,MLflow 管实验和模型版本,Prometheus + Grafana 管监控看板,Alertmanager 管告警路由。如果团队还没有实验平台,GrowthBook 或者自建 Redis + Prometheus 都能起步。模型调用入口用 TaoToken 统一管理,API Key 在控制台生成,接入文档在 https://taotoken.net/api-keys 和 https://taotoken.net/doc 可以查到详细的参数说明。需要长期跑编码类 Agent 实验的话,Coding Plan 的配额模式比按量计费更适合高频迭代场景。
最后给一个实操建议:先把离线 Gate 和 Prometheus 指标暴露跑通,再上 A/B 分流和贝叶斯决策。别一上来就追求全自动,手动跑通一次完整流程,知道每个环节的数据长什么样,再逐步自动化。统计严谨性不是学术洁癖,是避免"上线即翻车"的最低成本手段。