在生产环境接大模型接口，多数团队会选流式（`stream=True`），因为首 token 延迟直接决定用户体感。
但流式也把一部分复杂度从服务端推到了客户端：响应不再是完整的 JSON，而是一串可能在任何字节位置被切断的事件帧。分帧、解码、重试、中断，这四件事都需要客户端自己处理对。
## 一、流式把哪些问题推给了客户端
流式接口用的是 SSE（Server-Sent Events）。响应的 `Content-Type` 是 `text/event-stream`，body 是一串以空行分隔的帧：
```
data: {"id":"chatcmpl-1","choices":[{"delta":{"role":"assistant"}}]}
data: {"id":"chatcmpl-1","choices":[{"delta":{"content":"你"}}]}
data: {"id":"chatcmpl-1","choices":[{"delta":{"content":"好"}}]}
data: [DONE]
```
每帧以 `data:` 开头，帧之间用一个空行（`\n\n`）分隔。内容是增量的，放在 `choices[0].delta` 里而不是 `message`。最后会来一帧 `data: [DONE]`，收到就该主动关闭连接。
`delta` 里除了 `content`，还可能出现 `role`（第一帧）、`tool_calls`（函数调用）、以及部分厂商的 `reasoning_content`（思维链）。**解析时不要假设 `delta.content` 一定存在**，否则遇到工具调用那一帧就会 `KeyError`。
## 二、三个边界：分帧、解码、重试
### 2.1 帧被切开
最容易写错的版本是这样的——注意它把 `chunk` 当成了一行：
```python
for chunk in resp.iter_bytes():
for line in chunk.decode("utf-8").split("\n"): # ← 问题在这一行
if line.startswith("data:"):
obj = json.loads(line[5:]) # 偶发 JSONDecodeError
```
这段代码在本地跑往往没问题，上线后却偶尔抛 `JSONDecodeError`。原因是：**一个 chunk 不等于一行**。
TCP 是字节流，一次 `read` 返回多少字节，由网络、缓冲区和对端的写入节奏共同决定。一个完整的帧被拆进两次 `read`，完全正常：
```
第一次 read: data: {"id":"chat","choices":[{"delta":{"con
第二次 read: tent":"你好"}}]}
```
两次 `split("\n")` 拿到的都是残缺的 JSON，`json.loads` 必然失败。
**正确做法是维护一个缓冲区，先拼、再按空行切帧。**
补充一句：如果你用的是 `iter_lines()`，它内部**已经做了行缓冲**，会把跨 chunk 的半行拼回来，所以不会出现上面的问题——这也是为什么很多人写了很久都没遇到这个问题。
但还有两件事要处理：
- 有些反向代理会把换行改写成 `\r\n`，分隔符要同时兼容 `\n\n` 和 `\r\n\r\n`
- SSE 规范允许一帧里出现多行 `data:`，按规范它们应当用 `\n` 连接后再解析，不能只取第一行
所以自己按空行聚合成帧，仍然是最稳妥的写法。
### 2.2 字符被切开
这个问题更隐蔽，表现为**输出里偶尔冒出 `�`**。
一个中文字在 UTF-8 里占 3 个字节，emoji 占 4 个。如果一次 `read` 正好切在字符中间：
```
第一次 read 的字节: ... E4 BD ← 「你」的前 2 个字节
第二次 read 的字节: A0 ... ← 「你」的最后 1 个字节
```
每次拿到 chunk 都直接 `.decode("utf-8")` 的话，第一次会因为字节不完整而抛异常，或者用了 `errors="replace"` 就变成一个 `�`。
**正确做法是用增量解码器**，让它把不完整的字节留在内部缓冲里，等下一个 chunk 到了再拼：
```python
import codecs
decoder = codecs.getincrementaldecoder("utf-8")()
text = decoder.decode(raw_bytes, final=False)
```
好消息是：如果用的是 `httpx` 这类现代 HTTP 客户端，它的 `iter_text()` **内部已经用了增量解码器**，不会有这个问题。但如果你的代码是直接读 socket、或者用某些老库逐 chunk 拿 bytes，就必须自己处理。
### 2.3 重试的边界条件
这是最容易被忽略、后果也最严重的一个。
假设你在客户端加了「失败自动重试」。当连接在吐出 20 个字之后断开，重试会发生什么？
```
第一次请求的输出：今天天气
连接断开 → 自动重试
第二次请求的输出：今天天气真不错
最终用户看到的： 今天天气今天天气真不错
```
**内容重复了，而且用户已经开始阅读了。**
所以重试必须加一个前置条件：
> **只有「一个 token 都还没吐出来」时才能重试。**
一旦已经产生过输出，唯一正确的做法是**把错误抛给上层**，由业务决定是提示用户重试、还是把已输出的内容保留下来。
另外还有两个配套细节：
- **只重试可重试的错误**：网络层面的传输错误、`429`（限流）、`5xx`（服务端错误）可以重试；`401`（密钥错）、`400`（参数错）、`404`（模型不存在）重试多少次都一样，属于浪费
- **`429` 要读 `Retry-After` 响应头**，服务端已经告诉你该等多久了，自己瞎猜的退避时间可能太短
## 三、生产可用实现
下面这段用 `httpx`，把上面三个坑都处理了。依赖：`pip install httpx`。
```python
# -*- coding: utf-8 -*-
"""流式调用：SSE 分帧、增量解码、重试边界。"""
import json
import random
import time
import httpx
def stream_once(url, key, payload, timeout=120.0):
"""单次流式请求，逐段 yield 文本增量。"""
headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
buf = ""
with httpx.Client(timeout=timeout) as client:
with client.stream("POST", url, json=payload, headers=headers) as resp:
if resp.status_code >= 400:
resp.read() # 错误体是普通 JSON，先读出来再抛
resp.raise_for_status()
# iter_text() 内部用增量解码器，跨 chunk 的多字节字符不会被切坏
for chunk in resp.iter_text():
# 统一换行，兼容网关把 \n 改写成 \r\n 的情况
buf += chunk.replace("\r\n", "\n")
# 关键：先聚合，再按空行切帧；不要直接按行 json.loads
while "\n\n" in buf:
frame, buf = buf.split("\n\n", 1)
for line in frame.split("\n"):
if not line.startswith("data:"):
continue
data = line[5:].strip()
if data == "[DONE]":
return
try:
obj = json.loads(data)
except json.JSONDecodeError:
continue # 畸形帧跳过，不要因此中断整个流
choices = obj.get("choices") or []
if not choices:
continue
delta = choices[0].get("delta") or {}
text = delta.get("content")
if text:
yield text
def stream_chat(url, key, payload, max_retries=2, base_delay=0.5):
"""带重试的流式调用：只在尚未产出任何内容时才重试。"""
for attempt in range(max_retries + 1):
emitted = False
try:
for piece in stream_once(url, key, payload):
emitted = True
yield piece
return
except httpx.HTTPStatusError as err:
status = err.response.status_code
retryable = status == 429 or status >= 500
if emitted or not retryable or attempt == max_retries:
raise
# 429 优先听服务端的 Retry-After
wait = None
if status == 429:
raw = err.response.headers.get("Retry-After")
if raw and raw.isdigit():
wait = float(raw)
if wait is None:
wait = base_delay * (2 ** attempt) + random.uniform(0, 0.2)
time.sleep(wait)
except httpx.TransportError:
if emitted or attempt == max_retries:
raise
time.sleep(base_delay * (2 ** attempt) + random.uniform(0, 0.2))
```
调用方式和量首 token 延迟：
```python
import os
payload = {
"model": "your-model",
"messages": [{"role": "user", "content": "用三句话介绍什么是 SSE"}],
"stream": True,
}
# 接口地址和密钥从环境变量读，不要写死在代码里
base_url = os.environ["LLM_BASE_URL"] # 形如「协议://主机/v1」
api_key = os.environ["LLM_API_KEY"]
t0 = time.perf_counter()
ttft = None
pieces = []
gen = stream_chat(base_url + "/chat/completions", api_key, payload)
try:
for piece in gen:
if ttft is None:
ttft = time.perf_counter() - t0 # 首 token 延迟
pieces.append(piece)
finally:
gen.close() # 关键：中断时主动关闭，否则上游会继续生成、继续计费
total = time.perf_counter() - t0
print(f"首 token 延迟 {ttft:.2f}s，总耗时 {total:.2f}s")
print("".join(pieces))
```
## 四、体感指标：TTFT
很多人用「总耗时」来评估接口快慢，这在流式场景下会得出错误结论。
一个输出 800 字的回答，总耗时 20 秒，体感却可能很好——因为它第 0.8 秒就开始往外吐字了。反过来，一个总耗时 3 秒但**首 token 要等 2.8 秒**的接口，用户会觉得它「卡住了」。
这两个指标衡量的是完全不同的东西：
| 指标 | 含义 | 用户感知 |
| --- | --- | --- |
| 首 token 延迟（TTFT） | 从发起请求到收到第一个字 | 「快不快」 |
| 总耗时 | 全部生成完 | 「什么时候能复制走」 |
优化方向也不一样。TTFT 高通常是排队、路由或模型预热的问题；总耗时高则多半是输出太长或解码速度不够。
所以在做接口选型或压测时，建议把 TTFT 的 p50/p95 和总耗时分开记录。只看平均值会掩盖长尾——而长尾才是用户抱怨的来源。
## 五、中断与成本
还有一个成本问题容易被忽略。
用户点了「停止生成」，如果客户端只是停止读取、不关闭连接，**上游可能仍在继续生成 token**，而这些都是要计费的。
用生成器实现时，正确的做法是在 `finally` 里显式关闭：
```python
gen = stream_chat(url, key, payload)
try:
for piece in gen:
print(piece, end="", flush=True)
if user_pressed_stop():
break
finally:
gen.close()
```
`httpx` 的 `client.stream()` 在上下文退出时会关闭连接，而生成器被 `close()` 时会触发 `GeneratorExit`，把上下文一起退掉。如果漏了这一步，那个「已经没人看的回答」会一直生成到最大 token 数。
如果是 asyncio 场景，还要注意 `asyncio.CancelledError` 会在 `await` 点抛出，`finally` 里的清理代码要能容忍它。
## 六、检查项
| 问题 | 根因 | 处理方式 |
| --- | --- | --- |
| JSON 解析偶发失败 | 网络把一帧切成了两半 | 维护缓冲区，按空行聚合后再解析 |
| 输出出现 `�` | 多字节字符被 chunk 边界截断 | 用增量解码器（`httpx` 的 `iter_text` 已内置） |
| 重试后内容重复 | 流已产出内容仍在重试 | 只有「零输出」时才允许重试 |
| 白烧 token | 中断时没关闭连接 | 在 `finally` 里 `gen.close()` |
| 觉得接口慢但说不清 | 用总耗时代替了 TTFT | 两个指标分开统计 p50/p95 |
这份实现可以直接用，也可以按需裁剪。如果要落地到生产，建议额外补两件事：
一是把 TTFT 的 p50/p95 和总耗时**分开上报**，只看平均值会掩盖长尾；
二是给「已输出后失败」定义明确的业务处理方式——是提示用户重试，还是保留已生成的部分。这两个决定比代码本身更影响用户体验。
