流式输出实战：SSE 解析、chunk 拼装与断流重试

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

流式输出实战：SSE 解析、chunk 拼装与断流重试

结论先放这儿

流式是体验工程的第一优先级，首 token 延迟比总时长更影响体感。这篇把流式接入的工程细节补齐，与语音篇的流水线、扛量篇的限流配着看。

一、SSE 长什么样

流式接口返回的是一条持续打开的 HTTP 连接，服务端逐块推数据，每块都是一小段 JSON。OpenAI 兼容接口的每一块长这样：

data: {"choices":[{"delta":{"content":"你"}}]}

data: {"choices":[{"delta":{"content":"好"}}]}

data: [DONE]

三件事要注意。每块以 data: 开头，解析时按行切、去掉前缀再 json.loads。结尾的 [DONE] 是文本不是 JSON，要单独判断。还有一类以冒号开头的行（: keep-alive），是 SSE 规范里的心跳注释，直接跳过。

二、chunk 拼装与 usage

每块 delta 里只有增量文本，要自己拼。拼接的同时统计 token 数：

import json, httpx

def stream_chat(messages):
content = []
usage = None
with httpx.stream('POST', BASE_URL + '/chat/completions',
headers=HEADERS, json={
'model': MODEL, 'messages': messages,
'stream': True,
'stream_options': {'include_usage': True},
}, timeout=120) as r:
for line in r.iter_lines():
if not line.startswith('data:'):
continue
payload = line[5:].strip()
if payload == '[DONE]':
break
chunk = json.loads(payload)
if chunk.get('usage'):
usage = chunk['usage'] # 最后一块才有
for choice in chunk.get('choices', []):
delta = choice.get('delta', {}).get('content')
if delta:
content.append(delta)
yield delta
print('usage:', usage)

两个细节容易踩。usage 挂在最后一个 chunk 里，且要显式传 include_usage 才有，不传就永远拿不到对账数据。choices 可能为空列表（只有 usage 的那块），遍历前判一下。

三、断流重试的重试边界

流式的重试比普通请求危险。连接还没建立、首字节还没到的时候重试是安全的，什么都没发生。输出到一半断了，重试等于从头再生成一次，用户看到内容闪重，账单计两次。

所以重试逻辑按阶段分。首字节超时，放心重试。输出中途断，有两种处理，要完整答案就带着已拼内容重新请求（让模型从已有内容续写），要体感就只提示用户重发。对账以 usage 为准，重试导致的双份计费要在日志里能对出来。

def stream_with_retry(messages, retries=2):
for i in range(retries + 1):
try:
first = None
gen = stream_chat(messages)
for delta in gen:
if first is None:
first = delta # 首字节到了，之后断流不再整体重试
yield delta
return
except httpx.ReadTimeout:
if first is not None:
raise # 输出中途断，交给上层决定
if i == retries:
raise

四、前端渲染别每个 token 都刷

拿到 delta 直接往 DOM 里塞，Markdown 渲染会闪，半个代码块会错位。两个惯用做法。按时间节流，50-100ms 批量刷一次。按句渲染，攒到标点再重绘，顺带给 TTS 之类的下游切句用。

写在最后

流式接入三件事，SSE 逐行解析（心跳注释和 [DONE] 单独处理），增量拼接加 usage 对账，重试按阶段分层（首字节前安全，输出中途要小心）。配上节流渲染，体感和账单都稳。

本文方法与代码按自己的项目实践整理，与语音篇的流水线、扛量篇的对账配着用。