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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台生成 API Key 即可。拿到 Key 之后先确认两件事一是模型列表里有哪些可用版本A/B 实验的对照组和实验组要选不同版本二是接口的 base_url 和超时设置。TaoToken 的 API 地址是 https://taotoken.net/api 不带 UTM 参数直接填到环境变量里。export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api验证一下连通性import os from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL] ) resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 用一句话解释什么是A/B测试}], temperature0.0 ) print(resp.choices[0].message.content)如果返回正常说明接入没问题。接下来装实验依赖pip install mlflow prometheus-client statsmodels scipy openaiMLflow 用来追踪每次实验的参数、指标和 artifactprometheus-client 在模型服务层暴露 metricsstatsmodels 和 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_namefabtest-{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_idllm-v2.1.3-abtest-20250301, control_modelgpt-4o-mini, treatment_modelgpt-4o, traffic_percent5, duration_hours168 )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( modelmodel_version, messages[{role: user, content: user_input}] ) elapsed time.time() - start LLM_LATENCY.labels(groupgroup, model_versionmodel_version).observe(elapsed) return resp def record_feedback(group, model_version, feedback_type): LLM_FEEDBACK.labels( groupgroup, model_versionmodel_version, typefeedback_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_buckets100): 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.05power0.8import statsmodels.stats.proportion as smp import numpy as np def min_sample_size(mde0.02, base_rate0.5, alpha0.05, power0.8): nobs smp.samplesize_proportions_2indep_onetail( diffmde, prop2base_rate, powerpower, alphaalpha, ratio1.0 ) return int(np.ceil(nobs)) print(f每组需要: {min_sample_size(mde0.02)}) # 约 9604 print(fMDE5% 时每组需要: {min_sample_size(mde0.05)}) # 约 1500MDE2% 时每组要近 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},typelike}}[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, size100000) samples_b beta.rvs(b_alpha, b_beta, size100000) 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_n4800, treatment_n5000)4.2 Prometheus 告警规则把核心告警写成 Prometheus rule 文件groups: - name: llm-abtest-alerts rules: - alert: WinRateDropCritical expr: | (llm_win_rate_ratio{groupcontrol} - llm_win_rate_ratio{grouptreatment}) 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{grouptreatment}[5m])) / histogram_quantile(0.95, rate(llm_response_seconds_bucket{groupcontrol}[5m])) 1.2 for: 10m labels: severity: P1 annotations: summary: 实验组P95延迟增加超过20% description: 需人工确认是否牺牲延迟换质量 - alert: InsufficientSamples expr: | sum(rate(llm_user_feedback_total{grouptreatment}[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( modelgpt-4o, messages[{role: user, content: prompt}], temperature0.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 分流和贝叶斯决策。别一上来就追求全自动手动跑通一次完整流程知道每个环节的数据长什么样再逐步自动化。统计严谨性不是学术洁癖是避免上线即翻车的最低成本手段。