流式响应

流式响应会在模型生成内容的同时逐段返回结果，适合聊天界面、代码生成和较长回答。用户可以更早看到首段内容，但流式传输不会减少模型生成的 Token，也不代表总耗时一定更短。

本页示例使用 Python 的 requests 库：

python -m pip install requests

流式响应的基本形式

OpenAI、Anthropic 和 Gemini 兼容接口通常通过 Server-Sent Events（SSE）连续发送事件。响应不再是一个可以直接调用 response.json() 读取的完整 JSON，而是多行事件数据：

data: {"type":"...","...":"..."}

data: {"type":"...","...":"..."}

处理流式响应时需要：

在请求中启用流式输出，或调用协议专用的流式端点。
使用 stream=True 避免 requests 等待完整响应。
逐行读取以 data: 开头的内容。
按当前协议解析事件并拼接文本。
收到结束事件后结束读取。

不要直接调用 response.json()

流式响应由多个事件组成，不是单个 JSON 文档。应逐行读取并分别解析每个 data: 事件。

OpenAI Chat Completions

Chat Completions 在请求体中设置 "stream": true。文本增量位于 choices[0].delta.content，流通常以 data: [DONE] 结束。

import json

import requests

response = requests.post(
    "https://www.bita-api.com/v1/chat/completions",
    headers={"Authorization": "Bearer YOUR_BITA_API_KEY"},
    json={
        "model": "YOUR_MODEL_ID",
        "messages": [
            {
                "role": "user",
                "content": "请介绍流式响应的作用。",
            }
        ],
        "stream": True,
    },
    stream=True,
    timeout=(10, 300),
)
response.raise_for_status()

for raw_line in response.iter_lines():
    if not raw_line:
        continue

    line = raw_line.decode("utf-8")
    if not line.startswith("data:"):
        continue

    data = line.removeprefix("data:").strip()
    if data == "[DONE]":
        break

    event = json.loads(data)
    choices = event.get("choices", [])
    if not choices:
        continue

    text = choices[0].get("delta", {}).get("content")
    if text:
        print(text, end="", flush=True)

print()

部分事件只包含角色、停止原因或用量，不一定带有 content。代码应允许没有文本的事件正常通过。

OpenAI Responses

Responses API 同样在请求体中设置 "stream": true，但它使用语义化事件类型。文本增量事件通常为 response.output_text.delta，完成事件为 response.completed。

import json

import requests

response = requests.post(
    "https://www.bita-api.com/v1/responses",
    headers={"Authorization": "Bearer YOUR_BITA_API_KEY"},
    json={
        "model": "YOUR_MODEL_ID",
        "input": "请介绍流式响应的作用。",
        "stream": True,
    },
    stream=True,
    timeout=(10, 300),
)
response.raise_for_status()

for raw_line in response.iter_lines():
    if not raw_line:
        continue

    line = raw_line.decode("utf-8")
    if not line.startswith("data:"):
        continue

    event = json.loads(line.removeprefix("data:").strip())
    event_type = event.get("type")

    if event_type == "response.output_text.delta":
        print(event.get("delta", ""), end="", flush=True)
    elif event_type == "response.completed":
        break
    elif event_type in {"response.failed", "error"}:
        raise RuntimeError(event)

print()

Responses 流还可能包含输出项创建、内容块完成、工具调用等事件。只需要文本时，可以只处理 response.output_text.delta。

Anthropic Messages

Anthropic Messages 在请求体中设置 "stream": true。文本通常位于 content_block_delta 事件的 delta.text，结束事件为 message_stop。

import json

import requests

response = requests.post(
    "https://www.bita-api.com/v1/messages",
    headers={
        "x-api-key": "YOUR_BITA_API_KEY",
        "anthropic-version": "2023-06-01",
    },
    json={
        "model": "YOUR_MODEL_ID",
        "max_tokens": 1024,
        "messages": [
            {
                "role": "user",
                "content": "请介绍流式响应的作用。",
            }
        ],
        "stream": True,
    },
    stream=True,
    timeout=(10, 300),
)
response.raise_for_status()

for raw_line in response.iter_lines():
    if not raw_line:
        continue

    line = raw_line.decode("utf-8")
    if not line.startswith("data:"):
        continue

    event = json.loads(line.removeprefix("data:").strip())
    event_type = event.get("type")

    if event_type == "content_block_delta":
        delta = event.get("delta", {})
        if delta.get("type") == "text_delta":
            print(delta.get("text", ""), end="", flush=True)
    elif event_type == "message_stop":
        break
    elif event_type == "error":
        raise RuntimeError(event)

print()

message_start、content_block_start、message_delta 等事件包含消息元数据、停止原因或用量信息，不应当作为文本直接显示。

Gemini 原生接口

Gemini 使用 streamGenerateContent 流式端点，并通过 alt=sse 请求 SSE 格式。每个事件仍采用 Gemini 的 candidates[].content.parts[] 结构。

import json

import requests

response = requests.post(
    "https://www.bita-api.com/v1beta/models/YOUR_MODEL_ID:streamGenerateContent",
    params={"alt": "sse"},
    headers={"x-goog-api-key": "YOUR_BITA_API_KEY"},
    json={
        "contents": [
            {
                "role": "user",
                "parts": [
                    {
                        "text": "请介绍流式响应的作用。",
                    }
                ],
            }
        ]
    },
    stream=True,
    timeout=(10, 300),
)
response.raise_for_status()

for raw_line in response.iter_lines():
    if not raw_line:
        continue

    line = raw_line.decode("utf-8")
    if not line.startswith("data:"):
        continue

    event = json.loads(line.removeprefix("data:").strip())
    for candidate in event.get("candidates", []):
        for part in candidate.get("content", {}).get("parts", []):
            if "text" in part:
                print(part["text"], end="", flush=True)

print()

Gemini 流通常在服务器关闭响应后结束。最后一个事件可能同时包含 finishReason 和 usageMetadata。

超时和中断处理

示例使用：

timeout=(10, 300)

其中 10 秒是连接超时，300 秒是等待下一段响应数据的读取超时，并不是整个生成任务的总时限。

生产环境还应考虑：

用户主动停止生成时关闭 HTTP 连接。
网络中断时保留已经接收的文本，并明确提示输出可能不完整。
不要盲目自动重试已经开始输出的请求，否则可能产生重复内容和额外费用。
代理、网关或 Web 服务器需要关闭响应缓冲，并允许足够长的空闲超时。
页面离开或客户端断开时，及时释放连接和相关资源。

常见问题

现象	建议检查
等到最后才一次性显示	是否同时设置了接口的流式参数和 `requests` 的 `stream=True`；中间代理是否缓冲响应
JSON 解析失败	是否只解析 `data:` 后的内容，并跳过空行、`event:` 行和 `[DONE]`
中文显示乱码	是否按 UTF-8 解码事件数据
中途断开	检查读取超时、代理空闲超时和网络状态
没有用量字段	用量通常只出现在最后几个事件中，也可以到 Bita API 使用日志核对
接口返回普通 JSON	模型、端点或上游渠道可能不支持当前流式格式

不同模型可能附加推理内容、工具调用或其他增量字段。生产代码应保留未识别事件的日志，以便根据实际模型响应补充处理逻辑。

下一步：工具调用与结构化输出。

流式响应的基本形式​

OpenAI Chat Completions​

OpenAI Responses​

Anthropic Messages​

Gemini 原生接口​

超时和中断处理​

常见问题​