流式响应统一处理

⚠️ 待更新·2026-08-29核验 · 更新时间待核验 · 本文信息可能已过期，请以官方文档为准更新时间：2026-08-29 · 核验状态：待更新 · 官方溯源待补

背景

OpenAI 用 text/event-stream SSE 推 data: {...}\n\n chunk;Anthropic 用 event: content_block_delta 事件;Gemini 用 chunked JSON array。客户端若各自实现,每加一家就得重写一遍流式解析。统一网关把所有上游流式协议翻译成 OpenAI SSE,客户端用一套解析器吃所有模型。

核心问题

首 token 延迟(TTFT):用户对 >1s 的等待极其敏感。把 TTFT 设为路由指标,优先选首 token 快的 Provider(如 Groq)。
中途失败:流到一半上游 5xx,客户端已经开始渲染,不能简单 retry。网关要在 chunk 之间做心跳检测,失败时发一个 data: {"error": ...} chunk 优雅收尾。
背压(Backpressure):上游推得快、客户端读得慢,缓冲区溢出会丢 chunk。网关用 asyncio.Queue(maxsize=64) 做有界队列,满了就暂停读上游。
取消传播:客户端断连时,网关要立刻取消对上游的请求,不能让上游继续跑烧 token。

代码示例

import asyncio, json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def sse_translate(upstream_stream, model):
    '把任意上游流式翻译成 OpenAI SSE。'
    queue = asyncio.Queue(maxsize=64)
    async def producer():
        try:
            async for chunk in upstream_stream:
                # normalize each chunk to OpenAI shape
                data = {
                    "id": "chatcmpl-x", "object": "chat.completion.chunk",
                    "model": model,
                    "choices": [{"index": 0, "delta": {"content": chunk.text},
                                 "finish_reason": None}]}
                await queue.put(f"data: {json.dumps(data)}\n\n")
        except Exception as e:
            await queue.put(f"data: {json.dumps({'error': str(e)})}\n\n")
        finally:
            await queue.put("data: [DONE]\n\n")

    task = asyncio.create_task(producer())
    try:
        while True:
            item = await queue.get()
            if item == "data: [DONE]\n\n":
                break
            yield item
    finally:
        task.cancel()  # propagate cancellation upstream

@app.get("/v1/chat")
async def chat():
    return StreamingResponse(sse_translate(..., "llama-3.3-70b"),
                              media_type="text/event-stream")

边缘缓冲与 CDN 协同

当客户端在跨国网络下,直连网关会出现高延迟和丢包。把 SSE 流式响应通过 CDN(如 Cloudflare)边缘节点缓冲,可以显著降低 TTFT。但 CDN 对 SSE 有超时限制(通常 100s),长流式响应必须分块发心跳保持连接。网关可以发 : keep-alive 注释 chunk(15 秒一次),CDN 不会掐连接,客户端解析时也忽略注释行。这种 CDN 协同让免费模型也能服务全球用户。

最佳实践

心跳:每 15s 发一个 SSE 注释 : ping\n\n,防止 CDN/反向代理掐连接。
断线续传:支持 Last-Event-ID,客户端重连时从断点续传,网关要缓存最近 N chunk。
首 chunk 缓冲:接到的第一个 chunk 立即 flush,不要等到攒够 4KB,否则 TTFT 会被代理层拖大。
取消传播:客户端断连后,上游请求必须取消,避免空烧 token。

免费模型也可以拥有顺滑的打字机体验,只要网关把所有流式协议都对齐到 SSE。

🚀 立即开始：免费 API 一键调用

想要无需逐一注册、统一鉴权调用以上全部免费模型？Apishare.cc 提供统一 API Key，一个 Key 调用 100+ 模型，免费模型零成本直连。

👉 立即注册 Apishare.cc → 获取你的统一 API Key

📊 想看更多免费模型排行？查看 2026年9月免费 LLM API 综合实力榜单 →

立即上手：APIShare 免费 API 目录

🆓 注册领取免费额度：立即注册 APIShare · 登录控制台
🔍 浏览全部免费 API 与实时榜单：APIShare 免费 API 目录
📊 查看免费 LLM API 排行：Free LLM API Rankings

免费 API 聚合平台说明

本文涉及的模型均由 APIShare 免费 API 聚合平台统一接入，一份 Key 调用全站模型。

完整模型目录：APIShare 免费 API 目录
注册即送免费额度：注册领取 API Key