大模型API流式输出怎么实现？SSE流式调用教程

流式输出（Streaming）是指模型一边生成一边把结果分块推送给客户端，而不是等全部生成完再一次性返回。底层基于 SSE（Server-Sent Events）协议，服务端通过长连接持续推送 data: 开头的事件流，客户端逐块接收。

大模型生成速度有限（每秒几十到上百 token），一个长回答可能要等 10-30 秒，非流式下用户只能盯着空白等待，体验很差；流式下用户能逐字看到输出，首 token 延迟通常只有 1-2 秒，体感快得多。所有主流模型（GPT-5.5、Claude、Gemini、DeepSeek）和聚合平台如 EnlyAI 都支持流式，且兼容 OpenAI 的 stream 协议。

用 openai SDK 调用流式 API 非常简单，只需在 create 时加 stream=True，返回的就不再是单个 response 而是一个迭代器，遍历它即可逐块拿到 delta 内容。通过 EnlyAI 调用时完全兼容，把 base_url 换成 https://api.enlyai.com/v1 即可。

# pip install openai
from openai import OpenAI

client = OpenAI(
    api_key="sk-你的EnlyAI密钥",
    base_url="https://api.enlyai.com/v1"
)

stream = client.chat.completions.create(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "写一首关于编程的短诗"}],
    stream=True
)

for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:  # 首个 chunk 可能只有 role，delta 为 None
        print(delta, end="", flush=True)

关键点：每个 chunk 的 choices[0].delta.content 是增量文本，可能为 None；用 print(..., end='', flush=True) 实现逐字打印效果；遍历结束自动关闭连接。生产环境建议加上异常处理，流式中途断连时能恢复或提示用户。

核心区别：非流式一次性返回完整结果，调用简单但用户等待时间长；流式逐块返回，首 token 延迟低、体验好，但需要处理增量拼接。计费上两者相同，流式不会更贵。

使用场景：聊天机器人、AI 写作、代码补全等面向终端用户的应用强烈建议用流式，提升交互体验；后台批处理、数据抽取、分类等无需实时展示的场景用非流式更简单，因为可以直接拿到完整 JSON 便于解析。一个折中方案是流式接收但在服务端缓冲，攒够一段再推给前端，兼顾体验和可控性。EnlyAI 等聚合平台对两种模式都完整支持。

处理流式响应的关键是把每个 chunk 的 delta.content 累加到一个字符串里，最终得到完整回答。同时要处理：

1. 首个 chunk 通常只有 role 没有 content，要做空值判断；
2. 最后一个 chunk 的 finish_reason 会标记结束原因（stop、length、content_filter）；
3. 流式下 usage（token 计数）通常在最后一个 chunk 或需要单独请求获取，部分平台支持 stream_options.include_usage 在流末尾返回用量；
4. 中途网络断开要做容错。

stream = client.chat.completions.create(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "解释递归"}],
    stream=True,
    stream_options={"include_usage": True}  # 流末尾返回 token 用量
)

full_text = ""
usage = None
for chunk in stream:
    if chunk.choices:
        delta = chunk.choices[0].delta.content
        if delta:
            full_text += delta
    if chunk.usage:  # 最后一个 chunk 携带用量
        usage = chunk.usage

print(f"完整回答：{full_text}")
print(f"token 用量：{usage}")

通过 EnlyAI 调用时支持 stream_options，能方便地拿到 token 用量做成本统计。建议封装一个 stream_chat 函数，返回完整文本和 usage，业务层无感知。