⚠️ 待更新·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。
⚠️ Pending Update · 2026-08-29 Verification · Content may be outdated, please refer to official docs
Updated: 2026-08-29 · Status: Pending Verification
Background
OpenAI uses text/event-stream SSE pushing data: {...}\n\n chunks; Anthropic uses event: content_block_delta events; Gemini uses chunked JSON arrays. If clients implement each separately, every new provider forces another rewrite of the streaming parser. The unified gateway translates every upstream streaming protocol into OpenAI SSE so one client parser eats all models.
Core Problems
- Time-to-first-token (TTFT): users are highly sensitive to waits above 1s. Treat TTFT as a routing metric and prefer fast-start providers (e.g. Groq).
- Mid-stream failure: if the upstream returns 5xx halfway through, the client has already started rendering — you cannot just retry. The gateway must heartbeat between chunks and, on failure, emit a
data: {"error": ...} chunk to close the stream gracefully.
- Backpressure: upstream pushes faster than the client reads; a buffer overflow drops chunks. The gateway uses
asyncio.Queue(maxsize=64) as a bounded queue and pauses upstream reads when full.
- Cancel propagation: when the client disconnects, the gateway must immediately cancel the upstream request — otherwise the upstream keeps running and burning tokens.
Code Example
import asyncio, json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
async def sse_translate(upstream_stream, model):
'Translate any upstream stream into 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")
Edge Buffering and CDN Coordination
When clients are on cross-border networks, direct gateway connections suffer high latency and packet loss. Routing SSE responses through a CDN (e.g. Cloudflare) edge node buffer dramatically lowers TTFT. But CDNs impose SSE timeouts (usually 100s); long streams must emit heartbeats to keep the connection alive. The gateway can emit : keep-alive comment chunks every 15 seconds — the CDN will not drop the connection, and clients ignore comment lines during parsing. This CDN coordination lets free models serve global users too.
Best Practices
- Heartbeat: every 15s emit an SSE comment
: ping\n\n to keep CDN/reverse proxies from closing the connection.
- Resume support: implement
Last-Event-ID so the client can resume from a break; the gateway caches the last N chunks for this.
- First-chunk flush: flush the first chunk immediately — do not wait to accumulate 4KB, otherwise TTFT balloons at the proxy layer.
- Cancel propagation: when the client disconnects, the upstream request must be cancelled to avoid burning tokens on dead streams.
Free models can also deliver a smooth typewriter experience — as long as the gateway aligns every streaming protocol to SSE.