写了个 AI 聊天页面，被 5 种流式格式折腾了一整天 😭

掘金前端

ofox

2026年3月2日 15:29

周末想给自己的小项目加个 AI 聊天功能，本来以为流式输出很简单：建个 SSE 连接，解析 data: 开头的行，拼接文本，搞定。

结果...一天下来，我被五种不同的流式响应格式彻底搞麻了。

起因：一个"简单"的需求

需求其实很朴素：做一个聊天页面，让用户选择不同的 AI 模型，支持流式输出（打字机效果）。

我先接了 OpenAI 的 GPT-4o，大概十分钟就写完了：

import httpx
import json

async def stream_chat(messages):
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST", "https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {api_key}"},
            json={"model": "gpt-4o", "messages": messages, "stream": True}
        ) as resp:
            async for line in resp.aiter_lines():
                if line.startswith("data: ") and line != "data: [DONE]":
                    chunk = json.loads(line[6:])
                    content = chunk["choices"][0]["delta"].get("content", "")
                    if content:
                        yield content

完美运行，前端效果很丝滑。然后我就想：加几个模型选项呗，Claude、Gemini、DeepSeek、Kimi，让用户自己切换。

于是噩梦开始了。

第一个坑：Claude 的 SSE 完全不一样

Claude 的流式 API 返回格式长这样：

event: message_start
data: {"type": "message_start", "message": {"id": "msg_xxx", ...}}

event: content_block_start
data: {"type": "content_block_start", "index": 0}

event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "你好"}}

event: message_stop
data: {"type": "message_stop"}

注意看，和 OpenAI 完全不同：

多了 event: 行，不能直接忽略
文本在 delta.text 里，不是 choices[0].delta.content
结束信号是 message_stop 事件，不是 data: [DONE]

我只好加了一套专门的解析逻辑：

elif model.startswith("claude"):
    async for line in resp.aiter_lines():
        if line.startswith("data: "):
            chunk = json.loads(line[6:])
            if chunk.get("type") == "content_block_delta":
                text = chunk["delta"].get("text", "")
                if text:
                    yield text

行，能用。继续加模型。

第二个坑：Gemini 压根不是 SSE

Google Gemini 的流式 API，用的不是标准 SSE，而是流式 JSON 数组：

[{"candidates": [{"content": {"parts": [{"text": "你"}]}, "index": 0}}]}
,{"candidates": [{"content": {"parts": [{"text": "好"}]}, "index": 0}}]}
]

对，它返回的是一个 JSON 数组，通过 chunked transfer 分块传输。每个 chunk 可能包含不完整的 JSON 片段，你得自己攒 buffer 然后尝试解析。

我当时的心情：🤯

elif model.startswith("gemini"):
    buffer = ""
    async for chunk in resp.aiter_text():
        buffer += chunk
        while True:
            try:
                # 尝试找到完整的 JSON 对象
                idx = buffer.index("\n")
                line = buffer[:idx].strip().strip(",[]")
                buffer = buffer[idx+1:]
                if line:
                    obj = json.loads(line)
                    text = obj["candidates"][0]["content"]["parts"][0]["text"]
                    yield text
            except (ValueError, json.JSONDecodeError, KeyError, IndexError):
                break

写完这段代码我就知道，这条路走不通了 😅

第三个坑：DeepSeek 的 99% 兼容

DeepSeek 号称 OpenAI 兼容，实际上确实 99% 是。但那 1% 够你 debug 一下午的：

坑1：reasoning_content 字段

DeepSeek-R1 会返回一个额外的推理过程字段：

{
  "choices": [{
    "delta": {
      "reasoning_content": "让我想想这个问题...",
      "content": ""
    }
  }]
}

如果你只读 content，用户会看到 AI "卡住了"半天，然后突然蹦出一大段回复。得把 reasoning_content 也展示出来，或者至少给个 loading 状态。

坑2：空 choices

偶尔会遇到 choices 是空列表的情况：

content = chunk["choices"][0]["delta"].get("content", "")
# IndexError: list index out of range
# 因为 choices 是 []  😭

第四个坑：Kimi/Moonshot 的小细节

Moonshot API 基本兼容 OpenAI，但有两个小坑：

结束信号不一定是 data: [DONE]，有时候直接断开连接，你的代码要能处理这种情况
偶尔返回的 JSON 里 choices[0].delta 是 null 而不是空对象 {}

一个个小问题，每个都得加 defensive code。

最后代码变成了灾难

写完所有模型的适配，我的流式解析函数长这样：

async def stream_chat(model, messages):
    if model.startswith("gpt"):
        # OpenAI 格式
        ...  # 20行
    elif model.startswith("claude"):
        # Anthropic 格式
        ...  # 25行
    elif model.startswith("gemini"):
        # Google 流式 JSON
        ...  # 40行（最复杂）
    elif model.startswith("deepseek"):
        # OpenAI 兼容但要处理 reasoning_content 和空 choices
        ...  # 30行
    elif model.startswith("moonshot"):
        # OpenAI 兼容但结束信号不同
        ...  # 20行

一个 200 多行的函数，5 种 if-else 分支，每种都有自己的 edge case。这代码谁维护谁头疼。

而且最恶心的是：每当某个模型 API 更新格式，你就得重新测一遍所有分支。

转折点：统一成一种格式

写到第二天我实在受不了了。核心痛点其实很清楚：

如果所有模型的流式响应都是 OpenAI 格式，我只需要维护一套代码。

方案有两个：

方案 A：自己写转换代理

起一个中间服务，把各家 API 的响应统一转成 OpenAI SSE 格式再返回给前端。能行，但：

每种格式写一个 adapter，维护成本高
模型 API 更新你得跟着改
还得处理认证、限流、重试...

方案 B：用现成的 API 聚合平台

我后来发现 ofox.ai 这类平台已经把这事干了——所有模型走同一个 /v1/chat/completions endpoint，统一返回 OpenAI 兼容的 SSE 格式。

改完之后的代码：

async def stream_chat(model, messages):
    """一套代码搞定所有模型"""
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST", "https://api.ofox.ai/v1/chat/completions",
            headers={"Authorization": f"Bearer {OFOX_KEY}"},
            json={"model": model, "messages": messages, "stream": True}
        ) as resp:
            async for line in resp.aiter_lines():
                if line.startswith("data: ") and line != "data: [DONE]":
                    chunk = json.loads(line[6:])
                    choices = chunk.get("choices", [])
                    if choices and choices[0]["delta"].get("content"):
                        yield choices[0]["delta"]["content"]

200 行 → 15 行。不管前端选什么模型，后端就这一套解析逻辑。

前端也简单了：

const response = await fetch('/api/chat', {
  method: 'POST',
  body: JSON.stringify({ model: selectedModel, messages })
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;

  const text = decoder.decode(value);
  const lines = text.split('\n').filter(l => l.startsWith('data: '));

  for (const line of lines) {
    if (line === 'data: [DONE]') continue;
    const { choices } = JSON.parse(line.slice(6));
    if (choices?.[0]?.delta?.content) {
      appendMessage(choices[0].delta.content);
    }
  }
}

一套代码，所有模型通用。不管是 GPT-4o、Claude 3.5、Gemini、DeepSeek 还是 Kimi，前端不用改一行。

最终效果对比

模型	原生流式格式	统一后
GPT-4o	OpenAI SSE ✅	一致
Claude 3.5 Sonnet	Anthropic SSE（event + data）	→ OpenAI SSE
Gemini 2.0	流式 JSON 数组	→ OpenAI SSE
DeepSeek-R1	近似 OpenAI（有 quirks）	→ 标准 OpenAI SSE
Kimi K2.5	近似 OpenAI（小坑）	→ 标准 OpenAI SSE

指标	优化前	优化后
解析代码行数	~200 行	~15 行
if-else 分支	5 个	0 个
新增模型适配时间	半天起步	改个 model 名就行
API Key 数量	5 个	1 个

给想做多模型应用的同学几点建议

别假设 "OpenAI 兼容" 就是 100% 兼容：至少留 defensive code
Gemini 的流式格式和其他家完全不同：如果你要支持 Gemini 流式，做好心理准备
DeepSeek-R1 的 reasoning_content 要单独处理：不然用户体验会很奇怪
一开始就用统一格式的方案：别像我一样先踩坑再回头
测试时用真实的长对话：很多 edge case 只在长文本输出时出现

如果你也在做多模型 AI 应用，强烈建议一开始就走统一 API 的路线。我现在所有项目的模型调用都走 ofox.ai，一个 key 搞定 50 多个模型，流式格式统一，省了太多事。

踩坑一天，回头一看，大部分坑根本不用踩 🤦‍♂️

阅读视图