Kev 自部署后同一输入忽好忽坏，通常先查推理参数和并发，再回头查训练数据。原因很直接：采样参数、批处理串扰属于配置层，改一次、跑一轮就能看到变化；训练数据分布属于数据层，动它周期长、影响面大。建议的判断顺序是：固定输入重复调用确认不稳定是否真实存在，固定 temperature、top_p、seed 等随机性字段看输出是否收敛，再对比单并发与多并发，最后才把线上输入分布和自训练数据分布做覆盖对比。这几步走完，多数情况下能判断出问题落在哪一侧。

同一输入输出不一致时，优先排查推理参数与并发，因为这两类问题通常可以通过改配置、加固定参数在本地快速验证。操作上先做重复调用基线，再固定 temperature、top_p、seed，之后用单并发和多并发对照。只有当参数和并发都排除后，才需要回头统计自训练数据的长度、关键词与类别分布。推理侧改动能解决的问题属于配置问题，覆盖不了的才可能是数据边界问题。

固定输入重复调用，记录输出差异

先确认“不一致”是不是真实存在。把同一段输入、同一套参数连续发给本地推理接口，每次把原始输出落盘，用哈希或原文比对。判断方式很简单：一批调用里出现了多少个不同的输出。如果 temperature 本身非零，这一步只是建立基线，不做对错判断；基线的作用是给后面的参数对照和并发对照提供参照。

import json, hashlib, time, requests

URL = 'http://127.0.0.1:8000/v1/chat/completions'   # 替换成 Kev 服务实际地址
BODY = {
    'model': 'kev',
    'messages': [{'role': 'user', 'content': '把这句话改写得更正式：明天下午三点开会'}],
    'temperature': 0.7,   # 第一步按现状跑，先不要改
    'max_tokens': 256,
}

def call_once():
    resp = requests.post(URL, json=BODY, timeout=60)
    resp.raise_for_status()
    return resp.json()['choices'][0]['message']['content']

rows = []
for i in range(20):
    out = call_once()
    rows.append({'i': i, 'md5': hashlib.md5(out.encode()).hexdigest(), 'output': out})
    time.sleep(0.5)

with open('kev_repeat.jsonl', 'w', encoding='utf-8') as f:
    for row in rows:
        f.write(json.dumps(row, ensure_ascii=False) + '\n')

uniq = {r['md5'] for r in rows}
print('total', len(rows), 'unique', len(uniq))

这段脚本放在能访问 Kev 服务的机器上执行，只需替换 URL、model 名和输入文本。跑完看 kev_repeat.jsonl：md5 去重后数量接近 1，说明相同参数下输出基本稳定，可以转向并发或数据分布；数量明显多于 1，就先把推理参数固定下来再看。需要留意服务端是否有结果缓存，重复请求可能命中缓存，可以在请求里加一个变化的 nonce 字段，或先确认缓存策略再采集。

检查推理参数中影响随机性的字段

把随机性参数作为第一排查对象，是因为它最容易被改动，也最容易验证。常见字段包括 temperature、top_p、top_k、seed、do_sample，以及重复惩罚类参数。它们中任何一个没固定，输出都可能飘。建议先做一次“全部固定”的对照：把 temperature 设为 0、top_p 设为 1、给出固定 seed、关闭采样开关（服务支持时），其余参数保持不变。

固定参数的请求体可以直接替换到上一节的 BODY 里：

{
  "model": "kev",
  "messages": [{"role": "user", "content": "同一段固定输入"}],
  "temperature": 0,
  "top_p": 1,
  "top_k": 1,
  "seed": 42,
  "do_sample": false,
  "max_tokens": 256
}

验证方式还是重复调用：固定参数后如果输出收敛到同一个哈希，问题基本落在采样配置上；如果换了几组参数仍然发散，就要怀疑服务端批处理或并发串扰。需要注意的是，部分推理框架对 temperature=0 的处理不同，有些走贪心解码，有些仍保留采样路径；seed 也未必在所有后端都生效，这需要用同参数重复调用自行确认，不能只看配置文件是否写了。

排查并发和批处理是否导致请求串扰

如果参数固定后仍不稳定，下一步排除并发和批处理。连续批处理、动态合批、KV 缓存复用都可能让不同请求互相影响，尤其是在共享前缀或输入长度接近的场景下。做法是拿同一段输入，分别在单并发和多并发下各跑若干次，比较输出集合。

import concurrent.futures, hashlib, requests

URL = 'http://127.0.0.1:8000/v1/chat/completions'

def call(text):
    body = {
        'model': 'kev',
        'messages': [{'role': 'user', 'content': text}],
        'temperature': 0, 'top_p': 1, 'seed': 42, 'max_tokens': 256,
    }
    r = requests.post(URL, json=body, timeout=60)
    r.raise_for_status()
    out = r.json()['choices'][0]['message']['content']
    return hashlib.md5(out.encode()).hexdigest()

SAME = '同一段固定输入'

def serial(n=10):
    return [call(SAME) for _ in range(n)]

def parallel(n=10, workers=4):
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
        return list(ex.map(lambda _: call(SAME), range(n)))

a = serial()
b = parallel()
print('serial uniq', len(set(a)), 'parallel uniq', len(set(b)))
print('overlap', len(set(a) & set(b)))

验证方式是观察同一输入在不同并发下的输出是否一致：单并发只有一种哈希、多并发出现多种，基本可以判断存在请求串扰；两者都只有一种且相同，并发这一侧可以基本排除。接下来可以在服务配置里把批处理关掉、或把最大批大小调小，再跑一次同一脚本做对照。这类调整适合作为定位手段，是否长期保留需要结合吞吐需求评估。

回看自训练数据是否覆盖当前输入分布

参数和并发都排除后，再看自训练数据的覆盖情况。判断逻辑是：如果线上出现的输入在长度、关键词或类别上明显偏离训练集，模型在这些边界样本上输出漂移就更可能是数据覆盖不足造成的，而不是解码随机性。先做统计，不急着下结论。

import json, re
from collections import Counter

def load_text(obj):
    if 'input' in obj:
        return obj['input']
    msgs = obj.get('messages', [])
    return msgs[-1].get('content', '') if msgs else ''

def stats(path):
    lens, words, labels = [], Counter(), Counter()
    for line in open(path, encoding='utf-8'):
        line = line.strip()
        if not line:
            continue
        obj = json.loads(line)
        text = load_text(obj)
        lens.append(len(text))
        for w in re.findall(r'[\u4e00-\u9fa5]{2,}', text):
            words[w] += 1
        labels[obj.get('label', 'unknown')] += 1
    lens.sort()
    n = len(lens)
    return {
        'count': n,
        'len_min': lens[0] if n else 0,
        'len_p50': lens[n // 2] if n else 0,
        'len_p90': lens[int(n * 0.9)] if n else 0,
        'top_words': words.most_common(20),
        'labels': labels,
    }

print('online', stats('online_requests.jsonl'))
print('train', stats('train_set.jsonl'))

把线上请求的统计结果和训练集统计结果并排放，重点看三处：长度分位是否有明显外移、高频词和领域词是否在训练集里缺失、类别占比是否与线上请求结构差异过大。如果线上输入大量落在训练集长度分位之外，或出现训练集中少见的关键词，说明当前输入分布超出了训练覆盖范围，输出不稳定可能来自这里。统计结果只作为线索，具体是否影响输出，还需要用下一节的样本集来验证。

把稳定和不稳定的样本分别建集，定位边界

不管问题最终落在哪一侧，都需要一个可复现的样本集。把稳定输出和不稳定输出的输入分开存，每条记录保留输入、参数、多次调用结果和期望结果，后续修复后直接重跑这个集合比对。模板字段建议如下：

{
  "id": "s001",
  "input": "同一段固定输入",
  "params": {"temperature": 0, "top_p": 1, "seed": 42, "concurrency": 1},
  "outputs": ["第一次输出", "第二次输出"],
  "expected": "人工确认的目标输出，可为空",
  "verdict": "unstable",
  "note": "出现在多并发场景，单并发时未复现"
}

建集时先粗分两类：重复调用输出一致的进 stable 集，出现多种结果的进 unstable 集；如果某个输入在单并发下稳定、多并发下不稳定，在 note 里标注并发条件，单独成组。重跑脚本可以沿用前面的 call 函数：

cases = [json.loads(l) for l in open('kev_cases.jsonl', encoding='utf-8')]

for c in cases:
    outs = [call(c['input']) for _ in range(5)]
    c['rerun_uniq'] = len(set(outs))
    c['verdict_after'] = 'stable' if c['rerun_uniq'] == 1 else 'unstable'

with open('kev_cases_after.jsonl', 'w', encoding='utf-8') as f:
    for c in cases:
        f.write(json.dumps(c, ensure_ascii=False) + '\n')

验证方式是比较修复前后的 verdict 变化：unstable 集合中转为稳定的条目、以及仍然不稳定的条目，都值得逐条看一遍它们的参数和输入特征。如果参数固定后这些条目仍然不稳定，而它们在训练数据分布检查里又是明显的边界样本，那么后续处理方向就应该从推理配置转向数据补充或微调。样本集保持原样保存，不要因为一次修复就删除，它是后面每次改配置、改模型后的回归依据。