流式响应

流式响应允许你在模型生成过程中逐步接收结果，而非等待完整响应返回。这在聊天界面和长文本生成场景中特别有用，能够让用户实时看到输出内容。

OpenModel 在所有三种 API 格式中均通过 Server-Sent Events (SSE) 支持流式响应。

启用流式响应

OpenAI 格式

在请求体中设置 "stream": true，响应将以 SSE 流的形式返回，每行 data: 包含一个 JSON 数据块。

curl -N https://api.openmodel.ai/v1/responses \
  -H "Authorization: Bearer $OPENMODEL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-4o",
    "input": "写一首关于编程的俳句。",
    "stream": true
  }'

Anthropic 格式

在请求体中设置 "stream": true。Anthropic 的流式格式使用命名事件类型来组织响应结构。

curl -N https://api.openmodel.ai/v1/messages \
  -H "x-api-key: $OPENMODEL_API_KEY" \
  -H "Content-Type: application/json" \
  -H "anthropic-version: 2023-06-01" \
  -d '{
    "model": "claude-sonnet-4-20250514",
    "max_tokens": 1024,
    "stream": true,
    "messages": [
      {"role": "user", "content": "写一首关于编程的俳句。"}
    ]
  }'

其他 Messages 协议供应商： DeepSeek、DashScope 和 Xiaomi 均支持流式响应，行为和事件格式与 Anthropic 一致。使用 "stream": true 和 Anthropic SDK 的流式方法即可 — 代码完全相同，只需更换模型名称。

Gemini 格式

Gemini 格式需要使用 streamGenerateContent 端点替代 generateContent，并附加 ?alt=sse 参数以获取 SSE 格式的响应。

curl -N "https://api.openmodel.ai/v1beta/models/gemini-2.0-flash:streamGenerateContent?alt=sse&key=$OPENMODEL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "contents": [
      {"parts": [{"text": "写一首关于编程的俳句。"}]}
    ]
  }'

事件格式差异

三种 API 格式的流式事件结构各有不同。

OpenAI 事件

OpenAI 流式响应发送 data: 行，每行包含一个 JSON 对象，流结束时发送 data: [DONE]。

data: {"type":"response.output_item.added",...}
data: {"type":"response.content_part.added",...}
data: {"type":"response.output_text.delta","delta":"你好"}
data: {"type":"response.output_text.delta","delta":"世界"}
data: {"type":"response.output_text.done",...}
data: {"type":"response.completed",...}

Anthropic 事件

Anthropic 使用命名 SSE 事件类型，包含 event: 和 data: 字段：

event: message_start
data: {"type":"message_start","message":{...}}

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"你好"}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"世界"}}

event: content_block_stop
data: {"type":"content_block_stop","index":0}

event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":12}}

event: message_stop
data: {"type":"message_stop"}

Gemini 事件

Gemini SSE 流发送 data: 行，每个数据块包含一个 candidates 数组：

data: {"candidates":[{"content":{"parts":[{"text":"你好"}],"role":"model"}}]}

data: {"candidates":[{"content":{"parts":[{"text":"世界"}],"role":"model"}}]}

data: {"candidates":[{"content":{"parts":[{"text":"！"}],"role":"model"},"finishReason":"STOP"}],"usageMetadata":{...}}

最佳实践

处理不完整的数据块

SSE 数据可能以不完整的 TCP 分段到达。确保你的解析逻辑能够缓冲传入数据，并按完整的 \n\n 分隔符拆分事件，而非假设每次网络读取都包含完整的事件。

设置合理的超时时间

流式请求的耗时比非流式请求更长。为你的 HTTP 客户端配置更长的读取超时（例如 5 分钟以上），以避免长时间生成被中断：

import httpx
from openai import OpenAI

client = OpenAI(
    base_url="https://api.openmodel.ai/v1",
    api_key="your-api-key",
    timeout=httpx.Timeout(300.0, connect=10.0),
)

处理连接中断

网络波动可能导致流在响应中途终止。在应用中实现重连逻辑：

捕获连接错误和超时异常
对于非幂等请求，考虑保存已接收的部分响应并通知用户
尽可能使用 SDK 内置的重试机制

界面渲染的缓冲与刷新

在用户界面中渲染流式文本时，应缓冲传入的增量数据，按固定间隔（例如使用 requestAnimationFrame）刷新到显示层，而非在每个增量事件到达时都重新渲染。这可以避免过多的 DOM 更新，保持界面流畅。

On this page