用 SSE 流式接收对话补全，费用在最后一个分块里
2 分钟阅读作者 genmux API Team
流式输出让对话界面显得即时，但它也改变了计费的方式：模型停止生成之前用量都还不知道，所以用量必须走在流里，而不是响应头里。本文讲如何手工解析 genmux 的 SSE 流（如果你不用 OpenAI SDK，或者想弄明白 SDK 替你做了什么，这很有用），以及提前切断连接会发生什么。
线上格式
POST /v1/chat/completions 带 "stream": true 会返回 text/event-stream：一串 data: {...}\n\n 行，每一行都是一个 JSON chat.completion.chunk，最后以字面量 data: [DONE]\n\n 结束。除最后一个外，每个分块都带一个 delta，装着内容的下一段：
data: {"id":"gnx_9f2c...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"The"},"finish_reason":null}]}
data: {"id":"gnx_9f2c...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":" Volga"},"finish_reason":null}]}
data: {"id":"gnx_9f2c...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":12,"completion_tokens":47,"total_tokens":59,"cost":0.071,"cost_source":"estimated"}}
data: [DONE][DONE] 之前的最后一个分块，其 choices[].delta 为空，而顶层的 usage 有值——整个流里就这一行告诉你这次调用花了多少。流式调用的 cost_source 总是 "estimated"：上游厂商不为流式输出上报精确用量，所以 genmux 用自己能看到的 token 数算出它。
用 Python 解析
import json
import requests
def stream_chat(model: str, messages: list[dict], **kwargs):
with requests.post(
"https://api.genmux.tech/v1/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={"model": model, "messages": messages, "stream": True, **kwargs},
stream=True,
timeout=60,
) as resp:
resp.raise_for_status()
usage = None
for line in resp.iter_lines(decode_unicode=True):
if not line or not line.startswith("data: "):
continue
payload = line[len("data: "):]
if payload == "[DONE]":
break
chunk = json.loads(payload)
delta = chunk["choices"][0]["delta"] if chunk["choices"] else {}
if "content" in delta:
print(delta["content"], end="", flush=True)
if chunk.get("usage"):
usage = chunk["usage"]
print()
return usage
usage = stream_chat("deepseek-v4-flash", [{"role": "user", "content": "Describe the Volga in one sentence."}])
print(usage) # {'prompt_tokens': 12, 'completion_tokens': 47, 'total_tokens': 59, 'cost': 0.071, 'cost_source': 'estimated'}有两个细节值得保留：在对 chunk["choices"] 取下标之前先检查它（最后一个分块的数组可能是空的），以及对每个分块都检查 chunk.get("usage")，而不要假定它只出现在最后一个上——这样写的代码，在将来分块格式变化时仍然能正常工作。
用 Node 解析
async function streamChat(model, messages, extra = {}) {
const resp = await fetch("https://api.genmux.tech/v1/chat/completions", {
method: "POST",
headers: {
Authorization: `Bearer ${process.env.GENMUX_API_KEY}`,
"Content-Type": "application/json",
},
body: JSON.stringify({ model, messages, stream: true, ...extra }),
});
const reader = resp.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let usage = null;
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n\n");
buffer = lines.pop(); // keep the incomplete tail for the next read
for (const line of lines) {
if (!line.startsWith("data: ")) continue;
const payload = line.slice(6);
if (payload === "[DONE]") return usage;
const chunk = JSON.parse(payload);
const delta = chunk.choices[0]?.delta ?? {};
if (delta.content) process.stdout.write(delta.content);
if (chunk.usage) usage = chunk.usage;
}
}
return usage;
}如果你已经在用 openai 的 Node 或 Python SDK，上面这些解析都不用你操心——传 stream: true（Python：stream=True），然后在带 usage 的那个分块上把它读出来即可；拆行和 JSON 解析 SDK 都替你做了。
流式中途中断
如果你的客户端在 [DONE] 之前关闭了连接，或者用户离开了页面，genmux 只按已经生成的 token 计费——不会多收，也不会有一笔固定的「已取消」费用。这里有一个值得围绕着设计的重要不对称：
实际意义是：对话界面上的「停止」按钮可以放心接上，计费侧不需要任何特殊处理：用户想什么时候切断就切断，账单反映的正是他们在屏幕上看到的内容。
事后对账
如果你的进程在流式中途崩溃、丢掉了那个 usage 分块，这次调用并没有丢——每次对话调用都有一个 genmux 任务 id（每个分块上的 id 字段），任务结算后 GET /v1/generations/{id} 会返回包含 usage 和 cost 的完整记录。做周期性对账时，GET /v1/generations?capability=text.chat 会列出该账户的所有对话调用，它比你的客户端当场抓到的任何东西都更可靠。
开始使用
流式输出在平台上每个文本模型上的行为完全一致——DeepSeek、Qwen3、豆包，以及后续上线的更多模型——线上格式没有逐模型的怪癖。完整接口说明见文档；如果用的是推理模型，请配合思维链是怎么计费的一起看，因为推理 token 会在最终答案之前，从同一个 delta.content 字段流出来。