UniterRouter
Guides

流式响应

处理 LLM API 的实时流式响应。

流式响应

流式传输允许你在响应生成时逐 token 接收，为聊天应用提供更好的用户体验。

工作原理

启用流式传输时，API 返回一系列服务器发送事件 (SSE) 而不是等待完整响应。每个事件包含响应的一个块。

启用流式传输

在请求中添加 stream: true：

{
  "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" \
  -d '{
    "model": "gpt-4o",
    "messages": [{"role": "user", "content": "给我讲个故事"}],
    "stream": true
  }'
package main

import (
    "context"
    "fmt"
    "io"
    openai "github.com/sashabaranov/go-openai"
)

func main() {
    config := openai.DefaultConfig("your_api_key")
    config.BaseURL = "https://api.uniterouter.ai/v1"
    client := openai.NewClientWithConfig(config)

    stream, _ := client.CreateChatCompletionStream(
        context.Background(),
        openai.ChatCompletionRequest{
            Model: "gpt-4o",
            Messages: []openai.ChatCompletionMessage{
                {Role: "user", Content: "给我讲个故事"},
            },
            Stream: true,
        },
    )
    defer stream.Close()

    for {
        response, err := stream.Recv()
        if err == io.EOF {
            break
        }
        fmt.Print(response.Choices[0].Delta.Content)
    }
}

流事件格式

每个 SSE 事件包含一个 JSON 块：

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

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

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

data: [DONE]

Vercel AI SDK 集成

对于 React/Next.js 应用，使用 Vercel AI SDK：

// app/api/chat/route.ts
import { createOpenAI } from "@ai-sdk/openai";
import { streamText } from "ai";

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

export async function POST(req: Request) {
  const { messages } = await req.json();

  const result = streamText({
    model: uniterouter("gpt-4o"),
    messages,
  });

  return result.toDataStreamResponse();
}
// components/Chat.tsx
"use client";
import { useChat } from "ai/react";

export function Chat() {
  const { messages, input, handleInputChange, handleSubmit } = useChat();

  return (
    <div>
      {messages.map((m) => (
        <div key={m.id}>
          <strong>{m.role}:</strong> {m.content}
        </div>
      ))}
      <form onSubmit={handleSubmit}>
        <input value={input} onChange={handleInputChange} />
        <button type="submit">发送</button>
      </form>
    </div>
  );
}

处理流选项

stream_options

获取流式传输的使用信息：

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"使用的 token: {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}")

最佳实践

性能提示

  1. 对长响应使用流式传输 — 更好的故事、代码、解释的用户体验
  2. 处理连接中断 — 实现重试逻辑
  3. 适当缓冲输出 — 不要在每个 token 时更新 UI
  4. 考虑超时设置 — 流式连接可能持续更长时间

非流式 vs 流式

方面非流式流式
首个 token 时间慢（等待完整响应）快（立即）
总响应时间相同相同
用户体验加载状态渐进显示
实现更简单稍复杂
使用场景后台处理交互式聊天

带函数调用的流式传输

使用工具/函数与流式传输：

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

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

下一步

How is this guide?

SDK Integration

Use UniterRouter with popular AI SDKs and libraries.

AWS Bedrock Integration

Connect AWS Bedrock to UniterRouter for access to foundation models

On this page

流式响应
工作原理
启用流式传输
实现示例
流事件格式
Vercel AI SDK 集成
处理流选项
stream_options
错误处理
最佳实践
非流式 vs 流式
带函数调用的流式传输
下一步