Kimi K3 流式输出中途停住，先别去改 prompt。比较稳的判断路径是两条线并行：一条看响应本身有没有给出结束原因，另一条看客户端计时。结束原因字段明确写着长度类标识、且已生成 token 数贴着 max_tokens 配置，基本可以判定是输出长度到顶；如果整段流里根本没有结束原因，最后一个 chunk 的时间减去请求开始时间又接近某个配置好的超时值，那么截断更可能发生在调用端或网关的超时与连接层。这两种情况在界面上看起来一样，都是「只剩半句话」，但处理动作完全相反：前者要调大输出上限或改成续写，后者要调超时、关重试、查连接复用。

先读结束原因字段，再算已生成 token 数与 max_tokens 的差距；两项都对不上，就把首个 token 到最后一个 token 的时间差拿去和客户端读超时、网关空闲超时逐项比对。最后关掉重试与连接复用跑一次直连对照。长度截断会随上限上移，超时截断的停住时间点基本不动，这是最省事的区分方法。字段名随 SDK 不同，需要结合所用版本的实际返回结构确认。

先把结束原因字段和已生成 token 数打出来

流式响应里，结束原因通常藏在最后一个事件里，此时增量内容为空，只带一个原因标识；部分服务还会在同一事件或末尾单独给 usage 统计。字段命名不统一，可能是 finish_reason、stop_reason、finishReason，用量可能是 usage.completion_tokens 或 output_tokens，所以先做一个宽松的取值函数，把每个事件都过一遍，比猜字段名快。

def pick(d, *keys):
    for k in keys:
        if k in d and d[k] is not None:
            return d[k]
    return None

last_reason = None
last_usage = None
for event in stream:
    payload = getattr(event, 'model_dump', lambda: event)()
    reason = pick(payload, 'finish_reason', 'stop_reason', 'finishReason')
    usage = pick(payload, 'usage', 'token_usage')
    if reason:
        last_reason = reason
    if usage:
        last_usage = usage

print('end_reason=', last_reason)
print('usage=', last_usage)
print('max_tokens_cfg=', MAX_TOKENS)

把这一行日志和本次请求的 max_tokens 配置写进同一行记录里，形成「结束原因 / 输出 token 数 / 上限 / 耗时」的对照记录。常见结论有三种：结束原因是长度类标识，说明被上限截断；结束原因是正常停止类标识，说明模型自己收尾了，界面上的半句话是前端渲染或拼接问题；结束原因是空，说明流没有走完就断了，往下看超时和连接。

固定 prompt，把最大输出长度调大一档复测

验证长度上限假设时，唯一要变的变量是最大输出长度。prompt、系统提示、温度、top_p、是否流式都保持不变，用同一段输入跑两档：一档是当前配置值，另一档调到明显更大的一档，比如从几百调到两千上下，具体数值结合模型支持的区间定。每档至少留一条完整记录。

run,A档 max_tokens,B档 max_tokens,结束原因,输出 token,最后一句是否完整,总耗时
1,512,2048,,,,
2,512,2048,,,,

判读只看一件事：截断位置有没有随上限上移。如果 B 档输出的内容明显更长、结束原因从长度类变成正常停止，长度上限的嫌疑成立，处理方向是调大上限、把长回答拆成多段续写，或在提示里约束篇幅。如果 B 档仍然停在同一句话附近、时间差也几乎一样，那就不是长度问题，把注意力转到超时与连接层。两档都记录最后一句的原文和字符位置，靠肉眼比对比只看 token 数可靠。

记录首个 token 与最后一个 token 的时间差

超时截断的特征是「时间到了」，所以计时粒度要细到首包和末包。首 token 间隔反映排队与首字延迟，末 token 时间反映整段生成持续了多久，chunk 之间的最大间隔则能暴露网关空闲超时——这类超时看的不是总时长，而是两段数据之间的空档。

import time
t0 = time.monotonic()
first = last = None
max_gap = 0.0
for event in stream:
    now = time.monotonic()
    if first is None:
        first = now
    if last is not None:
        max_gap = max(max_gap, now - last)
    last = now

print('ttft_ms=', round((first - t0) * 1000))
print('last_ms=', round((last - t0) * 1000))
print('max_gap_ms=', round(max_gap * 1000))

拿到三个数后逐项比对配置：客户端 SDK 的连接超时与读超时、httpx 或 requests 里 timeout 元组的第二个值、网关或反向代理的空闲超时、服务端单次请求的最长处理时间。末包时间贴着客户端读超时，说明是客户端放弃等待；最大间隔贴着网关空闲超时，说明是链路中间先掐断；两者都远小于配置值却仍然中断，就去看结束原因和连接报错，不要继续在超时上打转。

关掉重试与连接复用再跑一次

重试和连接复用会制造「假截断」。开启了自动重试的客户端，在半句中报错后可能悄悄再发一次请求，前端看到的是半句话后重新开头，或者干脆卡住；复用已久的连接被中间设备回收，则可能直接抛出连接重置或流提前结束，日志里看不到任何结束原因。

import requests

s = requests.Session()
s.trust_env = False                      # 直连对照，不继承环境里的转发配置
adapter = requests.adapters.HTTPAdapter(max_retries=0)   # 关掉重试
s.mount('https://', adapter)

resp = s.post(url, headers=headers, json=body,
              stream=True, timeout=(10, 300))
for line in resp.iter_lines():
    if line:
        handle(line)
resp.close()
s.close()

SDK 侧同理，把 max_retries 或自动重连开关设为 0，禁用连接池复用（每次新建会话）跑一轮，再恢复默认跑一轮，对比两次的差异记录：中断位置是否相同、有没有出现重复开头、异常类型是读超时还是连接重置。直连单连接那一轮正常、恢复复用就断，问题在连接层；两轮都断在同一位置，回到长度上限和计时那两节。需要说明的是，关重试只是定位手段，不是线上长期方案。

把判别顺序整理成一张对照表

现场排查时按现象直接查表，能少走弯路。原则是先看响应自带的字段，再看自己的配置，最后才怀疑网络。

按这张表走完一轮仍定位不到，就把单次请求的原始事件序列完整落盘，连同客户端配置一起留档；没有结束原因、时间差又不贴任何超时值的极端情况，需要结合具体环境和服务端日志再确认，不要凭单次现象下结论。