OpenAI Streaming API - GPT-4、Claude 实时流式响应。实现流式聊天，减少首字符延迟，提升 AI 应用用户体验。

流式响应

使用 Server-Sent Events (SSE) 逐 token 流式传输响应，实现实时输出。

启用流式传输

在请求中设置 stream: true：

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "讲一个故事"}],
    stream=True
)

示例请求

from openai import OpenAI

client = OpenAI(
    base_url="https://api.uniterouter.ai/v1",
    api_key="your_api_key",
)

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

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

import OpenAI from "openai";

const client = new OpenAI({
  baseURL: "https://api.uniterouter.ai/v1",
  apiKey: process.env.UNITEROUTER_API_KEY,
});

const stream = await client.chat.completions.create({
  model: "gpt-4o",
  messages: [{ role: "user", content: "写一首关于编程的俳句" }],
  stream: true,
});

for await (const chunk of stream) {
  const content = chunk.choices[0]?.delta?.content;
  if (content) {
    process.stdout.write(content);
  }
}

curl https://api.uniterouter.ai/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $UNITEROUTER_API_KEY" \
  -N \
  -d '{
    "model": "gpt-4o",
    "messages": [{"role": "user", "content": "写一首俳句"}],
    "stream": true
  }'

流式数据块格式

每个数据块都是一个 Server-Sent Event：

data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1699000000,"model":"gpt-4o","choices":[{"index":0,"delta":{"content":"你好"},"finish_reason":null}]}

data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1699000000,"model":"gpt-4o","choices":[{"index":0,"delta":{"content":"世界"},"finish_reason":null}]}

data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1699000000,"model":"gpt-4o","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

收集完整响应

full_response = ""
for chunk in stream:
    if chunk.choices[0].delta.content:
        full_response += chunk.choices[0].delta.content

print(full_response)

带工具的流式传输

函数调用也支持流式传输：

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "东京的天气怎么样？"}],
    tools=tools,
    stream=True
)

tool_calls = []
for chunk in stream:
    delta = chunk.choices[0].delta
    
    if delta.tool_calls:
        for tool_call in delta.tool_calls:
            if tool_call.index >= len(tool_calls):
                tool_calls.append({"id": "", "function": {"name": "", "arguments": ""}})
            
            tc = tool_calls[tool_call.index]
            if tool_call.id:
                tc["id"] = tool_call.id
            if tool_call.function.name:
                tc["function"]["name"] = tool_call.function.name
            if tool_call.function.arguments:
                tc["function"]["arguments"] += tool_call.function.arguments

流式选项

获取使用统计：

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "你好"}],
    stream=True,
    stream_options={"include_usage": True}
)

for chunk in stream:
    if chunk.usage:
        print(f"Tokens: {chunk.usage.total_tokens}")

错误处理

try:
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "你好"}],
        stream=True
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="")
            
except Exception as e:
    print(f"流式错误: {e}")

Web 集成

React 示例

async function streamChat(message: string) {
  const response = await fetch("https://api.uniterouter.ai/v1/chat/completions", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "Authorization": `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      model: "gpt-4o",
      messages: [{ role: "user", content: message }],
      stream: true,
    }),
  });

  const reader = response.body?.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader!.read();
    if (done) break;
    
    const chunk = decoder.decode(value);
    const lines = chunk.split("\n").filter(line => line.startsWith("data: "));
    
    for (const line of lines) {
      const data = line.slice(6);
      if (data === "[DONE]") return;
      
      const json = JSON.parse(data);
      const content = json.choices[0]?.delta?.content;
      if (content) {
        // 更新 UI
        setResponse(prev => prev + content);
      }
    }
  }
}

优势

优势	描述
更低延迟	第一个 token 立即显示
更好的用户体验	用户实时看到进度
内存高效	按到达顺序处理 token
可取消	可以中途停止生成

流式响应

流式响应

启用流式传输

示例请求

流式数据块格式

收集完整响应

带工具的流式传输

流式选项

错误处理

Web 集成

React 示例

优势

相关文档

On this page