跳到主内容
快讯直播
AI智模界
教程

给自家模型做一次“时强时弱”体检:定位推理波动根因

同一句提问,上午答得干净利落,下午答得含糊其辞;同一个接口,压测时正常,线上流量一上来就变差。这类“时强时弱”往往不是模型突然变笨,而是推理链路里某个环节在不同条件下走了不同的路径。做完这套体检,你能拿到四样东西:一份可复现的评测基线、一张批不变性对照表、一张 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 80.97第 120 个 token长回答分歧更多
1 vs 320.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 模型的专家负载熵如果持续下降,是路由塌缩的早期信号,早发现比事后查便宜得多。

体检的价值不在于找到某个“罪魁祸首”,而在于建立起一套能反复使用的判断流程。第一次可能要花一天,之后每次发版只多花几分钟,而这几分钟能省下大量“到底是哪里不对”的排查时间。

AI 生成本文由 AI 基于公开信息自动生成,仅供参考。