大模型传输协议对比

大模型 API 主要使用以下传输协议，选择合适的协议对用户体验和系统性能至关重要。

为什么传输协议很重要？

传输协议决定了客户端和服务器之间如何交换数据。对于大模型 API，协议选择直接影响：

核心权衡：

flowchart LR
    A[简单性] <--> B[实时性]
    C[兼容性] <--> D[性能]

协议概览

选择建议：

HTTP REST

HTTP REST 是最基础的协议，所有大模型 API 都支持。

特点

• 请求-响应模式
• 无状态
• 简单易用
• 广泛支持

工作流程

Client                    Server
  |                         |
  |--- POST /v1/messages -->|
  |                         | (处理中... 可能数秒)
  |<-- 200 OK + JSON -------|
  |                         |

关键特征：

• 客户端发送请求后阻塞等待
• 服务器处理完成后一次性返回全部结果
• 连接在响应后关闭

示例

curl -X POST https://api.anthropic.com/v1/messages \
  -H "Content-Type: application/json" \
  -H "x-api-key: $API_KEY" \
  -d '{"model": "claude-sonnet-4-20250514", "max_tokens": 100, "messages": [{"role": "user", "content": "Hi"}]}'

优缺点

适用场景：

• 批量处理任务
• 后台异步任务
• 简单的 API 集成
• 对实时性要求不高的场景

Server-Sent Events (SSE)

SSE 是大模型流式输出的主流协议，让用户实时看到生成内容。

特点

• 服务器单向推送
• 基于 HTTP 长连接
• 自动重连机制
• 文本格式传输

SSE vs HTTP REST：

工作流程

Client                    Server
  |                         |
  |--- POST (stream=true) ->|
  |                         |
  |<-- event: message_start |
  |<-- event: content_delta |
  |<-- event: content_delta |
  |<-- event: message_stop  |
  |                         |

数据格式

SSE 使用简单的文本格式，每条消息包含可选的 event 和必需的 data：

event: message_start
data: {"type":"message_start","message":{"id":"msg_01"}}

event: content_block_delta
data: {"type":"content_block_delta","delta":{"text":"Hello"}}

event: content_block_delta
data: {"type":"content_block_delta","delta":{"text":"!"}}

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

格式规范：

• 每个字段占一行
• 字段格式：field: value
• 消息之间用空行分隔
• 支持的字段：event、data、id、retry

各厂商 SSE 实现对比

不同厂商的 SSE 实现有细微差异，需要注意：

OpenAI

data: {"id":"chatcmpl-abc","choices":[{"delta":{"content":"Hello"}}]}

data: {"id":"chatcmpl-abc","choices":[{"delta":{"content":"!"}}]}

data: [DONE]

特点：

• 只使用 data: 前缀，无 event: 字段
• 以 data: [DONE] 结束
• 内容在 delta.content 中

Claude

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

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

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

特点：

• 使用 event: + data: 完整格式
• 事件类型明确，便于处理
• 包含 index 标识内容块

Gemini

{"candidates":[{"content":{"parts":[{"text":"Hello"}]}}]}
{"candidates":[{"content":{"parts":[{"text":"!"}]}}]}

特点：

• 每行一个 JSON 对象（NDJSON 格式）
• 无标准 SSE 格式
• 使用独立的流式端点 :streamGenerateContent

解析差异总结：

客户端实现

JavaScript (浏览器)

const eventSource = new EventSource('/api/stream');

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data);
};

eventSource.onerror = (error) => {
  console.error('SSE Error:', error);
  eventSource.close();
};

JavaScript (fetch + ReadableStream)

const response = await fetch('/v1/chat/completions', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'Authorization': `Bearer ${apiKey}`
  },
  body: JSON.stringify({
    model: 'gpt-4o',
    messages: [{role: 'user', content: 'Hello'}],
    stream: true
  })
});

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

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

  const chunk = decoder.decode(value);
  const lines = chunk.split('\n');

  for (const line of lines) {
    if (line.startsWith('data: ')) {
      const data = line.slice(6);
      if (data === '[DONE]') break;
      console.log(JSON.parse(data));
    }
  }
}

Python

import httpx

with httpx.stream('POST', url, json=payload, headers=headers) as response:
    for line in response.iter_lines():
        if line.startswith('data: '):
            data = line[6:]
            if data == '[DONE]':
                break
            print(json.loads(data))

优缺点

WebSocket

特点

• 全双工通信
• 持久连接
• 低延迟
• 支持二进制

工作流程

Client                    Server
  |                         |
  |--- WebSocket Upgrade -->|
  |<-- 101 Switching -------|
  |                         |
  |<===== 双向通信 ========>|
  |                         |
  |--- Close Frame -------->|
  |<-- Close Frame ---------|

OpenAI Realtime API 示例

const ws = new WebSocket('wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview', {
  headers: {
    'Authorization': `Bearer ${apiKey}`,
    'OpenAI-Beta': 'realtime=v1'
  }
});

ws.onopen = () => {
  // 发送会话配置
  ws.send(JSON.stringify({
    type: 'session.update',
    session: {
      modalities: ['text', 'audio'],
      voice: 'alloy'
    }
  }));

  // 发送用户消息
  ws.send(JSON.stringify({
    type: 'conversation.item.create',
    item: {
      type: 'message',
      role: 'user',
      content: [{type: 'input_text', text: 'Hello!'}]
    }
  }));

  // 请求响应
  ws.send(JSON.stringify({type: 'response.create'}));
};

ws.onmessage = (event) => {
  const data = JSON.parse(event.data);

  switch (data.type) {
    case 'response.text.delta':
      console.log(data.delta);
      break;
    case 'response.audio.delta':
      // 处理音频数据
      break;
    case 'response.done':
      console.log('Response complete');
      break;
  }
};

优缺点

gRPC

特点

• 基于 HTTP/2
• Protocol Buffers 序列化
• 强类型
• 高性能

工作流程

Client                    Server
  |                         |
  |--- HTTP/2 Stream ------>|
  |<-- Protobuf Response ---|
  |<-- Protobuf Response ---|
  |<-- Trailers ------------|

Google Vertex AI 示例

from google.cloud import aiplatform
from google.protobuf import json_format

# 初始化
aiplatform.init(project="your-project", location="us-central1")

# 创建端点
endpoint = aiplatform.Endpoint("projects/.../endpoints/...")

# 预测请求
response = endpoint.predict(
    instances=[{"content": "Hello, Gemini!"}],
    parameters={"temperature": 0.7}
)

print(response.predictions)

Proto 定义示例

service PredictionService {
  rpc Predict(PredictRequest) returns (PredictResponse);
  rpc StreamPredict(PredictRequest) returns (stream PredictResponse);
}

message PredictRequest {
  string endpoint = 1;
  repeated Value instances = 2;
  Value parameters = 3;
}

message PredictResponse {
  repeated Value predictions = 1;
  Value metadata = 2;
}

优缺点

协议选择指南

按场景选择

按延迟要求选择

按客户端类型选择

实现对比

连接管理

错误处理

数据格式

混合使用模式

常见架构

┌─────────────┐     SSE      ┌─────────────┐
│   Browser   │ ◄──────────► │   Gateway   │
└─────────────┘              └──────┬──────┘
                                    │
                              gRPC  │
                                    ▼
                             ┌─────────────┐
                             │  LLM Server │
                             └─────────────┘

协议转换

# SSE → WebSocket 转换示例
async def sse_to_ws(sse_response, websocket):
    async for line in sse_response.aiter_lines():
        if line.startswith('data: '):
            data = line[6:]
            await websocket.send_json({
                'type': 'delta',
                'content': json.loads(data)
            })

常见问题

Q: 为什么聊天应用推荐 SSE 而不是 WebSocket？

SSE 更简单，基于 HTTP 协议，防火墙友好，且有自动重连机制。WebSocket 的双向通信能力在纯文本聊天场景中用不上。

Q: gRPC 为什么浏览器支持差？

gRPC 基于 HTTP/2 的二进制帧，浏览器的 fetch API 无法直接处理。需要使用 gRPC-Web 代理或 Envoy 网关转换。

Q: 如何处理 SSE 连接断开？

const eventSource = new EventSource('/api/stream');

eventSource.onerror = (error) => {
  // EventSource 会自动重连
  // 如需自定义重连逻辑：
  if (eventSource.readyState === EventSource.CLOSED) {
    setTimeout(() => {
      // 重新创建连接
    }, 3000);
  }
};

Q: WebSocket 如何实现心跳保活？

const ws = new WebSocket(url);
const heartbeat = setInterval(() => {
  if (ws.readyState === WebSocket.OPEN) {
    ws.send(JSON.stringify({type: 'ping'}));
  }
}, 30000);

ws.onclose = () => clearInterval(heartbeat);

参考资料

• SSE 规范: https://html.spec.whatwg.org/multipage/server-sent-events.html
• WebSocket 规范: https://datatracker.ietf.org/doc/html/rfc6455
• gRPC 文档: https://grpc.io/docs/
• HTTP/2 规范: https://datatracker.ietf.org/doc/html/rfc7540
• OpenAI Realtime API: https://platform.openai.com/docs/guides/realtime