用 OpenAI 软件开发工具包做流式输出

流式输出，决定了一个聊天框是干等八秒纹丝不动，还是不到一秒就开始往外 蹦字。无论你 base_url 背后的模型是 Claude、Gemini 还是 GPT，机制都一模一样：设上 stream=True，迭代分片，读取每个 分片上的 delta，到 [DONE] 这个哨兵值就停下来。 正因为 Brievio 讲的是 OpenAI Chat Completions 协议，同一个循环能原封不动 地跑在每一个真正的第一方模型上 —— 你只改 model 字符串，别的 什么都不用动。

本文会讲清楚 Server-Sent Events 流式输出在 OpenAI 兼容接口上到底是怎么 运作的、怎么用 stream_options 在最后一个分片上拿到准确的 token 用量、Python 与 Node 里如出一辙的写法，以及两种会让流看起来一切正常、实则在背地里坑你的静默失效模式：假流式（缓冲式）和缺失用量。

HTTP 上的「流式」到底指什么

非流式调用是一来一回：服务端思考几秒，然后一次性把整段补全交给你。流式 则会让 HTTP 连接保持打开，借助 Server-Sent Events（SSE） 在模型边生成边把答案分片推送给你。在传输层上，每个分片都是一行以 data: 开头、后跟一个 JSON 对象的文本，整个流以一行字面量 data: [DONE] 结束。

这段文本你基本上永远不用自己去解析 —— 软件开发工具包会替你搞定。你在 代码里拿到的是一个由分片组成的可迭代对象。每个分片看起来都像一个 普通的补全对象，只不过内容存在 choices[0].delta 里，而不是 choices[0].message 里，而且只装着自上一个分片以来新生成的那 一小段。把每个 delta.content 按顺序拼起来，你就还原出了完整 消息。这里唯一重要的指标是首个 token 的时间（TTFB）： 第一个非空 delta 多久才冒头。这个数字，正是流式存在的全部理由。

Python 写法

下面就是全部 —— 一个真正的流式循环，顺带把用量也接住。唯一一行 Brievio 专属的代码就是 base_url：

from openai import OpenAI

client = OpenAI(
    api_key="sk-brievio-...",
    base_url="https://api.brievio.com/v1",   # 一个 base_url，背后是真正的第一方模型
)

# stream=True 会把响应切换成 Server-Sent Events 流。
# 你迭代这个对象；每个元素都是一个携带局部 "delta" 的分片。
stream = client.chat.completions.create(
    model="claude-sonnet-4-6",               # 也可以换成 gemini-2.5-flash、gpt-... 等
    messages=[{"role": "user", "content": "Explain Raft consensus in 200 words."}],
    stream=True,
    stream_options={"include_usage": True},  # 要求在最后一个分片上返回用量
)

usage = None
for chunk in stream:
    # [DONE] 之前的最后一个数据事件携带用量，且 choices 列表为空。
    if chunk.usage is not None:
        usage = chunk.usage
        continue
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)     # token 一到就立刻渲染

print()
# usage 之所以有值，全因为设了 include_usage。这些都是真实数量。
print(usage.prompt_tokens, usage.completion_tokens, usage.total_tokens)

有三个细节最容易把人绊倒。第一，某些分片上的内容 delta 可能是 None 或空（开头那个分片往往只是设一下 role），所以打印前先 判一下空。第二，携带 usage 的那个分片在内容生成完之后才到，而且它的 choices 列表是空的 —— 这正是示例里先检查 chunk.usage 再 continue 的原因。第三，你不用自己 去找 [DONE]；软件开发工具包会吃掉那个哨兵值，并替你结束迭代。 要是你用原始的 requests 或 fetch 直接调接口，那 就得自己按换行切分，再在 [DONE] 处手动跳出。

Node 里同样的循环

Node 软件开发工具包把流暴露成一个异步可迭代对象，所以结构是一致的 —— 把 for 换成 for await ... of，把会刷新缓冲的 print 换成 process.stdout.write：

import OpenAI from "openai";

const client = new OpenAI({
  apiKey: "sk-brievio-...",
  baseURL: "https://api.brievio.com/v1",
});

// Node 里也是同一套约定：stream=true 返回一个分片的异步可迭代对象。
const stream = await client.chat.completions.create({
  model: "claude-sonnet-4-6",
  messages: [{ role: "user", content: "Explain Raft consensus in 200 words." }],
  stream: true,
  stream_options: { include_usage: true },
});

let usage = null;
for await (const chunk of stream) {
  // 最后一个事件：choices 为空，usage 出现。
  if (chunk.usage) {
    usage = chunk.usage;
    continue;
  }
  const delta = chunk.choices[0]?.delta?.content;
  if (delta) process.stdout.write(delta); // 每个 token 都即时写到终端
}

console.log();
console.log(usage?.prompt_tokens, usage?.completion_tokens, usage?.total_tokens);

留意那个可选链（chunk.choices[0]?.delta?.content）。在携带 用量的最后一个分片上，choices 是空的，所以不加保护就去取 [0] 会恰好在临门一脚时抛错。这是 Node 流式处理器在整段响应 都跑得好好的、却偏偏在最后一个事件上崩溃的头号原因。

在最后一个分片上拿到真实用量

默认情况下，流式响应不包含 token 数 —— 任何一个分片上都没有 usage 对象。这是 OpenAI 协议有意为之的一部分，也坑过那些在 生产环境里用流式、事后却对不上账单的团队。解法只有一个参数：

• 设上 stream_options={"include_usage": True} （Python）或 stream_options: { include_usage: true } （Node）。
• 服务端随后会在 [DONE] 之前多发一个分片，它的 choices 为空，usage 里则装着 prompt_tokens、completion_tokens 和 total_tokens。
• 在 Brievio 上，这些都是模型上报的真实数量 —— 与你做 一次非流式调用拿到的数字别无二致，按官方费率约打八五折计费。这里没有 灌过水的用量对象，也没有注入进来、把输入一侧撑大的系统提示。

如果你跳过 include_usage，又仍然需要一个 token 估算，那唯一 的办法就是用模型的分词器在本地数 —— 既只是近似，又是一份维护负担。直接 把这个开关设上就好。

静默的失效：假流式与缺失用量

有两种失效模式能骗过随便扫一眼的目测，只有细究起来才会露馅。在你把真实 流量托付给一个网关之前，这两项都值得花 20 秒查一查。

• 缓冲式的「假」流式。有些网关接受了 stream=True，却要等整段上游补全都生成完，再在最后 把它当成一阵分片爆发回放给你。你的循环照跑，delta 照到，一切看着都像 流式 —— 但 TTFB 和非流式调用一模一样，因为在模型生成完之前什么都没发 出去。识别的窍门很简单：测一下从发出请求到第一个非空 delta 之间的间隔。 真正的流式里，这远远在一秒以内就落地；而缓冲式回放里，它等于整段生成 时间。要是首个 token 的延迟跟着总延迟走，那你看的就不是流式，而是一段 录像回放。
• 缺失或伪造的用量。一个不响应 include_usage 的网关，会让你在流式调用上拿不到任何 token 数 —— 于是你只能对着空气去 核账单。更糟的是，一个不诚实的网关可以挂上一个数字灌过水的 usage 对象，因为在流上客户端很少会去重新数。用最朴素的办法 去验证：把同一段提示分别用流式和非流式各跑一次，确认流式那次 最后分片的用量与非流式的 usage 相符。它们本该一模一样。
• 看着像干净收尾、实则中途出错。如果上游模型在半路出错， 一个正确的网关会把它当成异常在你的循环里抛出来，而不是悄无声息地截断。 在把文本当成已完成来处理之前，永远先确认你收到了一个结束原因（或那个 用量分片）—— 一个就这么停住的流，跟一个真正生成完的流，不是一回事。

一句话总结

在 OpenAI 兼容接口上做流式，就是四个可动的零件：stream=True、 迭代分片、读取每个 delta，再把 [DONE] 交给软件 开发工具包去处理。再加上 stream_options={"include_usage": True}，你 还能在最后一个分片上拿到诚实的 token 数。同样这十五行代码，能在一个 base_url 背后的 Claude Sonnet 4.6、Gemini 2.5 Flash 和 GPT 系列上原封不动地跑 —— 换掉 model 字符串，循环照旧。

在上线之前，量一量首个 token 的时间，再把流式与非流式的用量对比一下。 真正的流式给你的是亚秒级的 TTFB 和对得上的数量；缓冲式回放则会在延迟上 露出马脚。在 Brievio 上，失败的 4xx/5xx 调用不 计费，所以这些检查你都可以免费跑。完整参数列表见 Chat Completions 参考，在同一个流上使用 工具与视觉的其余内容见 API 文档，非流式的基础用法见 用 OpenAI 软件开发工具包调用 Claude 的指南，而你能让这个循环指向的每一个标识符，都列在 模型目录里。