配置 ChatGPT API 流式响应需在请求体中设置 stream: true 参数，适用于需要实时展示生成内容的对话场景。主要风险边界在于网络波动可能导致流中断，客户端需具备重连或错误处理机制。

命令速用版

使用 curl 命令可快速验证接口是否支持流式输出，无需编写完整代码即可测试连通性与响应格式。

curl https://api.openai.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "model": "gpt-3.5-turbo",
    "messages": [{"role": "user", "content": "Hello"}],
    "stream": true
  }'

为什么会这样

流式响应本质是将完整响应拆分为多个数据块逐步发送，而非等待生成完成后一次性返回。

这种机制基于服务器发送事件（SSE）协议，服务端每生成一个 token 就推送一次数据，客户端接收后立即渲染。相比非流式模式，用户感知的首字延迟显著降低，但总耗时通常相近。

分步处理

以下以 Python requests 库为例，展示如何在代码中启用并处理流式响应。

步骤 1：设置请求参数

在发送 POST 请求的 JSON body 中，必须包含 "stream": true 字段，否则服务端将返回完整结果。

payload = {
    "model": "gpt-3.5-turbo",
    "messages": [{"role": "user", "content": "Hello"}],
    "stream": True  # 关键参数
}

步骤 2：启用流式接收

调用 requests.post 时设置 stream=True 参数，防止客户端一次性缓冲所有数据。

response = requests.post(url, headers=headers, json=payload, stream=True)

步骤 3：逐行解析数据

遍历 response.iter_lines()，过滤掉空行，识别以 data: 开头的行，直到遇到 [DONE] 信号。

for line in response.iter_lines():
    if line:
        decoded_line = line.decode('utf-8')
        if decoded_line == 'data: [DONE]':
            break
        if decoded_line.startswith('data: '):
            content = json.loads(decoded_line[6:])['choices'][0]['delta'].get('content', '')
            print(content, end='', flush=True)

怎么验证是否生效

通过检查 HTTP 响应头和数据结构可确认流式模式已激活。

检查响应头：Content-Type 字段应包含 text/event-stream。

检查数据包：响应体由多行组成，每行以 data: 开头，而非单个完整 JSON 对象。

检查前端表现：页面文本应呈现逐字增加的打字机效果，而非等待数秒后一次性显示全部。

常见坑

1. 错误处理困难：流式模式下，HTTP 状态码可能为 200，但业务错误包含在数据块中，需解析每个块的 error 字段。

2. 连接超时：生成时间长内容时，需调整客户端 timeout 设置，避免中途断开。

3. Token 计算：流式与非流式的 Token 消耗计费标准一致，但流式片段中可能不包含完整 usage 信息，需在最后一个包或单独接口查询。

常见问题

流式模式会比普通模式更贵吗？

不会。OpenAI 官方计费按实际生成的 Token 数量计算，与是否开启流式传输无关。

如何中途停止生成？

客户端主动关闭 HTTP 连接即可，服务端检测到连接断开后会停止生成以节省资源。

为什么收到的数据包含空内容？

部分数据块的 delta 字段可能为空或仅包含工具调用信息，代码需判断 content 是否存在后再渲染。

参考来源

OpenAI API Reference - Create Chat Completion
URL: https://platform.openai.com/docs/api-reference/chat/create