JSON流式协议设计

要点

  • 上一篇讲了 ReadableStream，知道怎样一块一块地往外推数据。但客户端怎么知道每块数据的边界在哪里？这就是协议要解决的问题
  • JSON 流式协议的核心思路：每条消息是一个完整的 JSON 对象，消息之间用约定的分隔符隔开
  • 和 SSE 相比，JSON 流式协议更灵活——可以携带任意结构的 JSON 数据，包括嵌套对象、数组、多字段
  • 协议设计要回答三个问题：分隔符用什么、每条消息长什么样、怎么标识流的开始和结束
  • 服务端实现的关键是把 JSON 对象序列化后拼上分隔符，写入 ReadableStream
  • 客户端解析的关键是攒缓冲区，按分隔符切割，每切出一段就 JSON.parse 一段
  • 错误处理不能只靠 HTTP 状态码——流已经开始之后出错，需要在消息流里嵌入错误类型的消息

1. 先接住上一篇

上一篇讲了 ReadableStream，我们知道可以这样把数据一块一块推出去：

// src/index.ts
app.get('/stream', (c) => {
  const stream = new ReadableStream({
    start(controller) {
      controller.enqueue(new TextEncoder().encode('hello '))
      controller.enqueue(new TextEncoder().encode('world'))
      controller.close()
    }
  })
  return c.body(stream)
})

客户端收到的是连续字节流 hello world，纯文本直接拼起来就行。但如果同一个流里要发送不同类型的消息呢？比如一个 AI 对话接口，流里可能有：

  • 对话内容片段（一个字、一个词）
  • token 用量统计
  • 模型名称、请求 ID 这类元信息
  • 错误信息

每段数据的结构不同。如果服务端连续推送了 {"type":"token","text":"你"}{"type":"done","usage":50}，客户端怎么知道这是两条消息，还是一条消息里恰好包含两个 JSON 对象？

这就是协议要解决的事情。

2. 什么是 JSON 流式协议

JSON 流式协议不是新技术标准，而是一种约定。核心规则只有一条：

每条消息是一个完整的、合法的 JSON 值，消息之间用一个固定的分隔符（delimiter）隔开。

整条流就是用分隔符串起来的一串 JSON：

{"type":"token","data":{"content":"你"}}\n{"type":"token","data":{"content":"好"}}\n{"type":"done"}\n

\n（换行符）是最常见的分隔符。选择它的原因：

  • JSON 规范里字符串值内部的换行必须转义，所以合法 JSON 内部不会出现裸换行符
  • 几乎所有语言都方便按行切割
  • 调试时一行一条消息，可读性好

3. 与 SSE 的对比

SSE（Server-Sent Events）是最常见的流式推送协议，OpenAI 的流式响应就基于 SSE：

// sse-response.txt
event: message
data: {"content": "你"}
 
event: message
data: {"content": "好"}
 
event: done
data: [DONE]

SSE 够用，但有几个限制：

  • data: 字段只能放纯文本，消息体复杂时要在 data 里塞 JSON 字符串，客户端收到还得再 parse
  • SSE 协议比较重，有 event:、data:、id:、retry: 这套字段约定
对比维度SSEJSON 流式协议
消息格式event: + data: + 空行{json}\n
数据嵌套data 只能放字符串直接放任意 JSON 对象
协议开销需要写 event/data/id 字段只有 JSON + 分隔符
自动重连浏览器 EventSource 自带需要自己实现
适用场景浏览器单向接收任意 HTTP 客户端

如果你需要在浏览器里做单向推送、不想自己写重连逻辑，SSE 的 EventSource API 确实方便。但消息体结构复杂、客户端不一定是浏览器、或者想自己控制协议细节时，JSON 流式协议更合适。

4. 协议格式设计

分隔符

常见选择：

  • \n（换行符）：最通用，推荐默认
  • \r\n（CRLF）：兼容 Windows 风格行尾
  • 自定义单字节字符：如 ASCII Record Separator (0x1E)，不会出现在正常 JSON 里

AI 对话场景用 \n 就够了。

消息结构

一条消息通常包含这几个字段：

// message-shape.json
{
  "type": "token",
  "data": { "content": "你" },
  "metadata": { "request_id": "req_abc123", "model": "claude-sonnet-4-20250514" }
}
  • type：消息类型，客户端靠这个字段做分发
  • data：消息的实际内容，不同类型的 data 结构可以不同
  • metadata：可选，放请求级别的公共信息

常见消息类型：

  • token：对话内容片段
  • start：流开始，携带 request_id、model 等元信息
  • meta：统计信息（总 token 数、停止原因）
  • error：流中间出错
  • done：流正常结束

流的开始和结束

方式一：用特殊消息标识。 第一条是 {"type":"start",...}，最后一条是 {"type":"done"}。

方式二：靠流的生命周期。 流被 close 就意味着结束，不需要额外的 done 消息。

方式一更灵活，推荐在 AI 场景下使用。

5. 服务端实现

用 Hono + Cloudflare Workers 实现 AI 对话的流式接口。

定义消息类型和编码函数

// src/protocols/types.ts
export type StreamMessage =
  | { type: 'start'; data: { request_id: string; model: string } }
  | { type: 'token'; data: { content: string } }
  | { type: 'meta'; data: { total_tokens: number; finish_reason: string } }
  | { type: 'error'; data: { code: string; message: string } }
  | { type: 'done' }
 
const DELIMITER = '\n'
 
export function encodeMessage(msg: StreamMessage): string {
  return JSON.stringify(msg) + DELIMITER
}

encodeMessage 把一条消息转成 JSON 字符串，再拼上换行符。

创建流式响应

// src/routes/chat.ts
import { Hono } from 'hono'
import { encodeMessage } from '../protocols/types'
import type { StreamMessage } from '../protocols/types'
 
const app = new Hono()
 
app.post('/api/chat', async (c) => {
  const body = await c.req.json<{ prompt: string }>()
  const requestId = crypto.randomUUID()
  const encoder = new TextEncoder()
 
  const stream = new ReadableStream({
    async start(controller) {
      const send = (msg: StreamMessage) => {
        controller.enqueue(encoder.encode(encodeMessage(msg)))
      }
 
      send({ type: 'start', data: { request_id: requestId, model: 'my-llm-v1' } })
 
      try {
        const llmStream = await callLLMStream(body.prompt)
        for await (const token of llmStream) {
          send({ type: 'token', data: { content: token } })
        }
        send({ type: 'meta', data: { total_tokens: 128, finish_reason: 'stop' } })
        send({ type: 'done' })
        controller.close()
      } catch (err) {
        send({ type: 'error', data: { code: 'LLM_ERROR', message: (err as Error).message } })
        controller.close()
      }
    }
  })
 
  return c.body(stream, 200, { 'Content-Type': 'application/x-ndjson' })
})

几个细节：

  • Content-Type 用 application/x-ndjson（Newline Delimited JSON），也可以用 application/stream+json
  • 错误发生在流中间时，HTTP 状态码已经发出去了，只能用 error 类型的消息通知客户端

6. 客户端解析

服务端把消息用换行符连成一串发出去，客户端收到的是连续字节流。要还原成一条条消息：

  1. 把收到的字节拼成字符串，放进缓冲区
  2. 按换行符切割缓冲区
  3. 对切出来的每一段做 JSON.parse
// src/client/parse.ts
import type { StreamMessage } from '../protocols/types'
 
export function createStreamParser() {
  let buffer = ''
  const decoder = new TextDecoder()
 
  return {
    feed(chunk: Uint8Array): StreamMessage[] {
      buffer += decoder.decode(chunk, { stream: true })
      const lines = buffer.split('\n')
 
      // 最后一段可能不完整，留在缓冲区等下一次 feed
      buffer = lines.pop() ?? ''
 
      const messages: StreamMessage[] = []
      for (const line of lines) {
        const trimmed = line.trim()
        if (!trimmed) continue
        try {
          messages.push(JSON.parse(trimmed) as StreamMessage)
        } catch {
          console.warn('Failed to parse stream message:', trimmed)
        }
      }
      return messages
    },
 
    // 流结束后调用，处理缓冲区里剩余的数据
    flush(): StreamMessage | null {
      const trimmed = buffer.trim()
      if (!trimmed) return null
      buffer = ''
      try {
        return JSON.parse(trimmed) as StreamMessage
      } catch {
        return null
      }
    }
  }
}

关键点：

  • buffer.split('\n') 之后，最后一段不一定是完整消息。比如 &#123;"type":"token"&#125;\n&#123;"type":"tok，第二段还不完整。所以最后一段留在缓冲区，等下一次 feed 拼接
  • JSON.parse 外面包 try-catch，不能让一条坏消息打断整个解析流程
  • flush() 在流结束后调用，处理缓冲区里可能残留的最后一条消息

在 fetch 里使用

// src/client/consume.ts
async function consumeChatStream(prompt: string) {
  const response = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ prompt })
  })
 
  if (!response.ok || !response.body) throw new Error(`Request failed: ${response.status}`)
 
  const reader = response.body.getReader()
  const parser = createStreamParser()
 
  while (true) {
    const { done, value } = await reader.read()
    if (done) break
    for (const msg of parser.feed(value)) {
      handleMessage(msg)
    }
  }
 
  const lastMsg = parser.flush()
  if (lastMsg) handleMessage(lastMsg)
}
 
function handleMessage(msg: StreamMessage) {
  switch (msg.type) {
    case 'start':
      console.log('Stream started:', msg.data.request_id)
      break
    case 'token':
      process.stdout.write(msg.data.content)
      break
    case 'meta':
      console.log('\nTokens used:', msg.data.total_tokens)
      break
    case 'error':
      console.error('Stream error:', msg.data.message)
      break
    case 'done':
      console.log('Stream finished')
      break
  }
}

7. 错误处理

流式响应的错误处理比普通接口复杂，错误可能发生在三个阶段。

流开始之前

请求参数错误、鉴权失败——在 HTTP 响应头发出之前就能检测到，直接用普通 JSON 错误响应：

// src/routes/chat.ts
app.post('/api/chat', async (c) => {
  const body = await c.req.json<{ prompt?: string }>()
  if (!body.prompt) {
    return c.json({ error: { code: 'INVALID_INPUT', message: 'prompt is required' } }, 400)
  }
  // ... 正常流式逻辑
})

客户端在 fetch 之后先检查 response.ok，false 就走普通错误分支。

流传输中间

LLM 调用超时、模型返回异常——HTTP 状态码已经发出去了（通常是 200），没法再改。在消息流里嵌入 error 消息：

// src/protocols/stream-error.ts
catch (err) {
  const errorMsg: StreamMessage = {
    type: 'error',
    data: { code: classifyError(err), message: getSafeMessage(err) }
  }
  controller.enqueue(encoder.encode(encodeMessage(errorMsg)))
  controller.close()
}
 
function classifyError(err: unknown): string {
  if (err instanceof DOMException && err.name === 'AbortError') return 'CANCELLED'
  if (err instanceof TypeError) return 'NETWORK_ERROR'
  return 'INTERNAL_ERROR'
}
 
function getSafeMessage(err: unknown): string {
  if (err instanceof TypeError) return 'Network connection failed'
  return 'An unexpected error occurred'
}

getSafeMessage 不暴露内部错误细节，只返回安全的提示文案。

流意外中断

网络断开、Worker 被回收、代理超时——流突然断了，没有 error 也没有 done 消息。客户端检测方式：reader.read() 返回 &#123; done: true &#125; 时，检查是否收到过 done 类型的消息。

// src/client/detect-interruption.ts
let streamCompleted = false
 
while (true) {
  const { done, value } = await reader.read()
  if (done) break
  for (const msg of parser.feed(value)) {
    if (msg.type === 'done') streamCompleted = true
    handleMessage(msg)
  }
}
 
if (!streamCompleted) {
  console.error('Stream interrupted: no done message received')
  // 触发重试或展示错误提示
}

8. 与 OpenAI / Claude 流式响应的对比

OpenAI 的格式

OpenAI 使用 SSE，data 字段里放 JSON 对象：

// openai-stream.txt
data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"你"},"index":0}]}
 
data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"好"},"index":0}]}
 
data: [DONE]
  • 对话内容在 choices[0].delta.content 里，流结束用 [DONE] 字符串标识

Claude 的格式

Claude 也用 SSE，消息结构更细粒度：

// claude-stream.txt
event: message_start
data: {"type":"message_start","message":{"id":"msg_xxx","model":"claude-sonnet-4-20250514"}}
 
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"你"}}
 
event: message_stop
data: {"type":"message_stop"}

用不同的 event 类型区分消息，区分了 message 级别和 content_block 级别的开始/结束。

从中能学到什么

把两家的设计放一起看，可以归纳出几个规律：

  1. 流的第一条消息总是携带元信息——请求 ID、模型名称，客户端用来做日志关联
  2. 内容片段是最频繁的消息类型，每条只携带几个字
  3. 结束信号明确标识——OpenAI 用 [DONE]，Claude 用 message_stop 事件
  4. 统计信息放在结束之前——token 用量、停止原因在流的最后阶段发出

我们前面设计的协议遵循了这些规律。如果要和 OpenAI 或 Claude 的格式兼容，在 encodeMessage 里按它们的字段结构来组织 data 就行，协议层不用改。

总结

  • JSON 流式协议：每条消息是一个完整 JSON，消息之间用分隔符隔开
  • \n 是最常用的分隔符，因为合法 JSON 内部不会有裸换行
  • 和 SSE 相比，JSON 流式协议更轻量，支持复杂消息结构
  • 协议设计要确定三件事：分隔符、消息结构（type + data）、开始和结束标识
  • 服务端用 encodeMessage 把结构化消息转成 JSON + 分隔符，写入 ReadableStream
  • 客户端解析的核心是缓冲区机制——收到的数据可能在不完整的位置被截断
  • 错误处理分三个阶段：流开始之前用 HTTP 状态码，流中间用 error 消息，流中断靠客户端检测 done
  • OpenAI 和 Claude 的协议核心模式一致：元信息开头、内容片段为主、明确结束信号