流式输出中断、429 重试与费用核算

更新于 2026-09-28；软件与模型验证范围见正文。

本文示例对应 Chat Completions SSE；Responses 的事件类型和 output 字段不同，不能复用 choices 解析。2026-09-28，本文可下载示例真实运行成功，gpt-4o 的 SSE 文本、[DONE] 和 usage 均已读取（输入 15、输出 28、总计 43 Token）。分片、429 和中断另通过本地测试，线上失败与恢复尚未复现。

第 1 步：准备依赖与模型

python3 -m pip install requests
export tokenba_API_KEY="YOUR_tokenba_API_KEY"

示例选择 gpt-4o 标准聊天，stream=true。若换成其他模型，核对 token 上限与推理参数，不直接复用同一请求字段。

第 2 步：解析 SSE 并限制 429 重试

下载完整 Python 示例 stream-retry.py；这是可运行代码，令牌通过环境变量读取，不是含凭据的配置模板。

import email.utils
import json
import os
import random
import time
from datetime import datetime, timezone
from pathlib import Path

import requests


def retry_delay(value, attempt):
    if value:
        try:
            seconds = float(value)
        except ValueError:
            try:
                date = email.utils.parsedate_to_datetime(value)
                if date.tzinfo is None:
                    date = date.replace(tzinfo=timezone.utc)
                seconds = (date - datetime.now(timezone.utc)).total_seconds()
            except (TypeError, ValueError):
                seconds = None
        if seconds is not None:
            if seconds > 60:
                raise RuntimeError("Retry-After exceeds 60s; retry manually later")
            return max(0, seconds)
    return min(30, 2 ** attempt) + random.uniform(0, 1)


def open_stream(body, headers):
    for attempt in range(3):
        response = requests.post(
            "https://api.apizt.com/v1/chat/completions",
            json=body,
            headers=headers,
            stream=True,
            timeout=(10, 60),
        )
        if response.status_code == 429 and attempt < 2:
            retry_after = response.headers.get("Retry-After")
            response.close()
            time.sleep(retry_delay(retry_after, attempt))
            continue
        try:
            response.raise_for_status()
            if "text/event-stream" not in response.headers.get("Content-Type", "").lower():
                raise ValueError("Expected text/event-stream")
            return response
        except Exception:
            response.close()
            raise
    raise RuntimeError("Retry limit reached")


def sse_data(lines):
    parts = []
    for line in lines:
        if isinstance(line, bytes):
            line = line.decode("utf-8")
        if not line:
            if parts:
                yield "\n".join(parts)
                parts = []
        elif line.startswith("data:"):
            value = line[5:]
            parts.append(value[1:] if value.startswith(" ") else value)
    if parts:
        yield "\n".join(parts)


def consume(response, output):
    usage = None
    complete = False
    for data in sse_data(response.iter_lines()):
        if data == "[DONE]":
            complete = True
            break
        event = json.loads(data)
        if event.get("error"):
            raise RuntimeError(str(event["error"]))
        if event.get("usage"):
            usage = event["usage"]
        for choice in event.get("choices", []):
            text = choice.get("delta", {}).get("content")
            if text:
                output.write(text)
                output.flush()
                print(text, end="", flush=True)
    if not complete:
        raise RuntimeError("Stream ended without [DONE]; partial output kept, request not replayed")
    return usage


def main():
    headers = {"Authorization": "Bearer " + os.environ["tokenba_API_KEY"]}
    body = {
        "model": "gpt-4o",
        "messages": [{"role": "user", "content": "Explain API streaming in one short sentence."}],
        "max_tokens": 64,
        "stream": True,
        "stream_options": {"include_usage": True},
    }
    try:
        with open_stream(body, headers) as response, Path("partial-output.txt").open("w", encoding="utf-8") as output:
            usage = consume(response, output)
        print("\nusage:", usage)
    except (requests.RequestException, RuntimeError, ValueError) as error:
        print("\nStopped:", str(error))
        raise


if __name__ == "__main__":
    main()

第 3 步：中断后保留结果，不自动重放

收到首段后中断，partial-output.txt 会保留已收到文本。程序不会自动重新 POST 整个请求；用户可检查调用日志后决定新建请求、续写或缩短任务。缺少 [DONE]、错误事件和异常 JSON 都应明确失败，不能当完整答案。

第 4 步：核对 Retry-After 与实际费用

仅在读取输出前遇到 HTTP 429 时重试，最多 3 次。Retry-After 支持秒数和 HTTP 日期；超过 60 秒时停止自动重试，稍后手动再试。其他错误进入对应排错，不让所有 400/401/超时都无限重试。

stream_options.include_usage 的支持取决于模型与分组。收到 usage 时保留原字段；没有最终 usage 时到控制台核对实际消耗。不要按显示字符数估算，也不要把代理客户端 cost=0 理解为免费。

接口、报错与价格

Chat SSE 参考 · Responses 流式参考 · 401 与模型权限 · Token、缓存与倍率计算 · 真实调用范围。

开始实践：注册并创建 API Key · 已有账号，进入控制台 · 查看当前模型价格。