同一句提问,上午答得干净利落,下午答得含糊其辞;同一个接口,压测时正常,线上流量一上来就变差。这类“时强时弱”往往不是模型突然变笨,而是推理链路里某个环节在不同条件下走了不同的路径。做完这套体检,你能拿到四样东西:一份可复现的评测基线、一张批不变性对照表、一张 MoE 路由命中分布图、一张采样与量化参数扫描表,并且这些检查会变成每次发版都自动跑的回归测试。
前置条件清单
动手前先确认这些东西到位,缺一项后面都会卡住。
- 一套能稳定复现的推理服务:本地部署或远程都行,关键是你能控制它,能改 batch、能改量化、能重启。
- 一份固定评测集:200~500 条,覆盖你业务里真实出现的任务类型(问答、抽取、代码、数学、多轮)。每条带标准答案或参考答案。
- 能读到单条请求的原始输出与(如果服务端支持)token 概率。
- 一个能跑 Python 脚本和版本管理工具的环境。评测集要进版本库。
- 模型权重和推理框架的版本记录方式。具体用哪个版本的框架、哪个量化方案,以官方文档当前版本为准,本文只讲方法。
体检的核心原则只有一条:一次只动一个变量,并且把所有变量都记下来。
步骤一:把“波动”变成数字
肉眼感觉“今天不太好”没有意义。先建固定评测集和固定解码参数,把主观感受压缩成一个分数。
评测集用 JSONL 存,一行一条:
```json
{"id": "extract-001", "task": "extract", "prompt": "从下面这段文字里抽出所有日期:……", "reference": "2024-03-01;2024-05-17"}
{"id": "math-002", "task": "math", "prompt": "一个水池……", "reference": "3.5"}
```
然后写一个跑评测的脚本,每次运行都记录“运行指纹”:
```python
import json, hashlib, time, os
from openai import OpenAI
client = OpenAI(base_url=os.environ["BASE_URL"], api_key=os.environ["API_KEY"])
def fingerprint(cfg: dict) -> str:
blob = json.dumps(cfg, sort_keys=True)
return hashlib.sha256(blob.encode()).hexdigest()[:12]
def run_one(item: dict, cfg: dict) -> dict:
t0 = time.time()
resp = client.chat.completions.create(
model=cfg["model"],
messages=[{"role": "user", "content": item["prompt"]}],
temperature=cfg["temperature"],
top_p=cfg["top_p"],
max_tokens=cfg["max_tokens"],
seed=cfg["seed"],
服务端支持时才返回;不支持就去掉这两个参数
logprobs=True,
top_logprobs=1,
)
choice = resp.choices[0]
return {
"id": item["id"],
"task": item["task"],
"text": choice.message.content,
"finish_reason": choice.finish_reason,
"latency_ms": int((time.time() - t0) * 1000),
"fingerprint": fingerprint(cfg),
}
```
把 temperature 设为 0、seed 固定、max_tokens 固定,然后把同一份评测集连跑 5 遍,比较这 5 遍的结果:
```python
from collections import defaultdict, Counter
def stability_report(all_runs: list[list[dict]]) -> dict:
by_id = defaultdict(list)
for run in all_runs:
for row in run:
by_id[row["id"]].append(row["text"])
stable, unstable = 0, []
for item_id, texts in by_id.items():
if len(set(texts)) == 1:
stable += 1
else:
unstable.append((item_id, Counter(texts)))
total = len(by_id)
return {
"total": total,
"stable": stable,
"consistency_rate": round(stable / total, 4),
"unstable_items": unstable[:20], # 先看前 20 条
}
```
如果 temperature=0 下一致率不是 1.0,恭喜,你已经抓到第一类根因了:推理本身就不是确定性的。常见来源是不同 batch 下的归约顺序、不同的融合算子、以及浮点累加顺序。这类差异通常在少数 token 上体现,长回答会被逐步放大。
步骤二:批不变性对照
同一句话,单独发和混在 32 条里一起发,结果应该一样。不一样,就是批不变性被破坏了。
```python
def batch_invariance_check(prompts: list[str], cfg: dict, batch_sizes=(1, 8, 32)):
"""同一批 prompt,分别以不同 batch size 发出去,比较每条的输出。"""
import concurrent.futures as cf
results = {}
for bs in batch_sizes:
if bs == 1:
outs = [run_single(p, cfg) for p in prompts]
else:
with cf.ThreadPoolExecutor(max_workers=bs) as pool:
outs = list(pool.map(lambda p: run_single(p, cfg), prompts))
results[bs] = [o["text"] for o in outs]
base = results[batch_sizes[0]]
diff = []
for i, p in enumerate(prompts):
for bs in batch_sizes[1:]:
if results[bs][i] != base[i]:
diff.append({"index": i, "batch_size": bs,
"prompt": p[:60], "base": base[i][:80],
"other": results[bs][i][:80]})
return diff
```
比较时有个技巧:别只比字符串,也比 token 概率。有时答案文字一样,但模型对它的置信度已经漂了,这种漂移在更长、更难的问题上迟早会翻车。如果服务端不返回 logprobs,就退而求其次,把温度稍微调高、跑多条采样,比较分布形状。
对照表长这样:
| batch size | 输出一致率 | 平均首个分歧 token 位置 | 备注 |
|---|---|---|---|
| 1 vs 8 | 0.97 | 第 120 个 token | 长回答分歧更多 |
| 1 vs 32 | 0.91 | 第 64 个 token | 明显劣化 |
如果 batch 越大差异越明显,重点检查:推理框架里的AI 词典:连续批处理">连续批处理(continuous batching)实现、attention 的变长 kernel、以及 padding 处理。这些地方最容易引入 batch 相关的路径分叉。
步骤三:MoE 路由命中统计
如果模型是 MoE 结构,“时强时弱”还有一类很隐蔽的根因:同一个 token 在不同条件下被路由到了不同的专家。专家能力并不均等,路由一变,输出质量就变了。
原理上,每层的 router 会为每个 token 算出各专家的分数,取 top-k。你要做的是把这个选择记录下来。不同框架的 router 输出结构不一样,通常是 logits 或 topk 索引,按实际返回取:
```python
import torch
routing_log = [] # 全局收集,生产环境要换成有界队列
def make_router_hook(layer_idx: int):
def hook(module, inputs, output):
output 可能是 (topk_weights, topk_ids) 或 logits,按实现取
ids = None
if isinstance(output, (tuple, list)):
for part in output:
if torch.is_tensor(part) and part.dtype in (torch.int32, torch.int64):
ids = part
elif torch.is_tensor(output):
ids = output
if ids is not None:
flat = ids.detach().to("cpu").reshape(-1)
routing_log.append({"layer": layer_idx,
"experts": flat.tolist()})
return hook
def attach_hooks(model):
handles = []
for i, mod in enumerate(model.modules()):
if hasattr(mod, "gate") or "router" in type(mod).__name__.lower():
handles.append(mod.register_forward_hook(make_router_hook(i)))
return handles
```
统计时看这几个量:
```python
from collections import Counter
import math
def routing_stats(records: list[dict]) -> dict:
per_layer = {}
for r in records:
per_layer.setdefault(r["layer"], Counter()).update(r["experts"])
summary = {}
for layer, counter in per_layer.items():
total = sum(counter.values())
probs = [c / total for c in counter.values()]
entropy = -sum(p * math.log(p + 1e-12) for p in probs)
summary[layer] = {
"distinct_experts": len(counter),
"entropy": round(entropy, 3),
"share": counter.most_common(5),
}
return summary
```
关键对比:同一批 prompt 在 batch size 1 和 batch size 32 下,路由命中是否一致。如果一致率低于 1.0,说明路由本身受批处理影响,那么前面看到的输出波动就有相当一部分来自这里,而不是采样参数。
另一个要看的指标是专家负载均衡。如果少数专家吃掉了大部分 token,说明路由塌缩,模型在长尾问题上的表现会明显偏弱。熵值随时间下降也值得警惕。
注意:挂 hook 本身有开销,会改变时序。做严谨对比时,先测一次“挂 hook”和“不挂 hook”的基线差异,把这项开销单独记下来。
步骤四:采样与量化参数扫描
前面两步排查的是系统层面的差异,这一步排查的是配置层面的差异。
采样参数用网格扫描,别一次只调一个值然后凭感觉判断:
```python
import itertools, json
grid = {
"temperature": [0.0, 0.3, 0.7],
"top_p": [0.8, 0.95, 1.0],
"repetition_penalty": [1.0, 1.05],
}
rows = []
for combo in itertools.product(*grid.values()):
cfg = dict(zip(grid.keys(), combo))
cfg.update({"model": "your-model", "max_tokens": 512, "seed": 42})
scores = evaluate(eval_set, cfg) # 你的打分函数
rows.append({cfg, scores})
rows.sort(key=lambda r: r["score"], reverse=True)
print(json.dumps(rows, ensure_ascii=False, indent=2))
```
量化这块要拆开看,混在一起会得出错误结论:
- 权重量化:影响全局,不同 bit 宽度的差异会体现在长尾任务上。
- KV cache 量化:影响长上下文,短问答几乎看不出,长文档摘要会露馅。
- 激活量化:对 batch size 更敏感,容易出现“小 batch 正常、大 batch 崩”。
做量化对比时,保持推理框架、参数、评测集完全一致,只换量化配置。如果换了量化顺带换了后端或 kernel,那这次对比就没有意义了。具体支持哪些量化方案、怎么配,以推理框架官方文档当前版本为准。
扫描结果建议记录成一张表:配置、总分、各任务分项分、平均延迟、失败样本 ID 列表。最后一项经常被忽略,但它比总分更有信息量——两个配置总分接近,失败样本集合却完全不同,说明它们坏的方面不一样。
步骤五:固化成回归测试
体检做完,如果不变成自动检查,下次发版一切照旧。把上面四步写成脚本,纳入发版流程。
```python
tests/test_inference_stability.py
import json
import pytest
CFG = json.load(open("config/eval_config.json"))
BASELINE = json.load(open("baselines/stability_baseline.json"))
def test_deterministic_consistency():
report = stability_report(run_n_times(CFG, n=3))
assert report["consistency_rate"] >= BASELINE["consistency_rate"] - 0.01, \
f"确定性一致率下降:{report['consistency_rate']}"
def test_batch_invariance():
diff = batch_invariance_check(SMOKE_PROMPTS, CFG, batch_sizes=(1, 16))
rate = 1 - len(diff) / len(SMOKE_PROMPTS)
assert rate >= BASELINE["batch_invariance_rate"] - 0.02, \
f"批不变性劣化:{rate},前几条分歧:{diff[:3]}"
def test_expert_load_balance():
stats = routing_stats(collect_routing(SMOKE_PROMPTS, CFG))
for layer, s in stats.items():
assert s["distinct_experts"] >= BASELINE["min_experts_per_layer"], \
f"第 {layer} 层疑似路由塌缩"
```
三条建议:阈值不要拍脑袋定,用稳定版本的实测值减去合理余量;评测集分成“冒烟集”(跑得快,每次提交都跑)和“全量集”(发版前跑);每次失败都把当时的配置和原始输出存成 artifact,方便回溯。
常见坑与排错
temperature=0 就以为完全确定。 不是。浮点运算的累加顺序、不同的 kernel 实现都会带来微小差异,长输出会放大。正确做法是承认它并量化它,而不是假设它不存在。
只看字符串,不看概率。 答案文字相同、概率分布已经漂移的情况很常见。有 logprobs 就用,没有就用多次采样的分布对比代替。
评测集太小。 50 条评测集上的 2% 差异基本是噪声。做根因定位时,样本量不够会得出完全相反的结论。
反复请求同一进程,被前缀缓存骗了。 很多推理服务会缓存相同前缀的 KV,第二次请求走的是捷径。做一致性测试时要么换 prompt,要么显式关掉缓存,要么在不同进程/实例上跑。
只比平均分,不比失败集合。 两个配置总分一样,坏的题目可能完全不同。永远把失败样本 ID 列表一起存下来。
把网络和排队算成模型波动。 超时、限流、网关重试都会让结果看起来变差。记录每次请求的延迟、状态码和重试次数,把这些过滤掉再分析。
对比量化时换了多个变量。 一次只动一个。换了量化又换了 batch 策略,得出的结论没法归因。
MoE 路由统计开销影响结论。 挂 hook 会拖慢推理、改变批处理节奏。先测一次无 hook 的基线。
把波动当成 bug 去修。 有些波动是预期的(采样、长尾任务)。体检的目的是把波动分成“可解释的”和“不可解释的”,只对后者动手。
下一步建议
体检跑通一次之后,把它变成常态:
1. 挂进发布流程。 每次换模型、换量化、升级推理框架,自动跑冒烟集,任何一项指标越界就拦住发版。
2. 建立影子流量对比。 把线上真实请求复制一份发给新配置,比较两边的输出分歧率和延迟分布,比离线评测更接近真实。
3. 监控线上漂移。 统计每天的请求长度分布、任务类型分布、拒绝率和重试率,分布变形往往早于指标下滑。
4. 给评测集做版本管理。 每发现一个新失败案例,就补一条进评测集。评测集会随着体检次数增长,这才是长期收益。
5. 把路由统计接到监控。 MoE 模型的专家负载熵如果持续下降,是路由塌缩的早期信号,早发现比事后查便宜得多。
体检的价值不在于找到某个“罪魁祸首”,而在于建立起一套能反复使用的判断流程。第一次可能要花一天,之后每次发版只多花几分钟,而这几分钟能省下大量“到底是哪里不对”的排查时间。
