在网络应用中，OpenAI 的 AI 通过服务器发送事件（SSE）技术，能够实时将数据推送到浏览器。

这种高效的通信方式使得服务器可以主动发送信息，而不再依赖客户端的反复请求。

什么是SSE？

服务器发送事件（SSE）是一种基于HTTP的网络技术，允许服务器通过持久连接向客户端实时推送数据。这意味着服务器可以主动发送信息给浏览器或其他客户端，而不需要客户端频繁地请求更新。

工作步骤

1. 连接建立：客户端（如浏览器）向服务器发送请求，以建立一个持久的SSE连接。此连接将保持打开状态，直到客户端或服务器明确关闭。
2. 数据推送：连接建立后，服务器可以随时向客户端发送数据。数据通常以文本格式传输，并按照特定格式进行组织（例如，以 data: 开头）。
3. 事件格式：服务器发送的数据块可以包含不同的事件类型和内容。每个数据块以 \n\n 进行分隔，这样可以形成多个消息的连续流。
4. 自动重连：若连接因故障中断，浏览器会自动尝试重连，以继续接收数据流。

流式LLM API的工作原理

流式大语言模型（LLM）API利用SSE技术，通过持久连接向客户端发送实时数据。使用此技术的API通常返回的内容类型为 content-type: text/event-stream。

这种格式允许数据以块的形式传输，每个数据块以 \r\n\r\n 分隔。每个数据块的内容通常包含以 data: 开头的JSON格式行。此外，一些API可能会使用 event: 行来标识特定的事件类型。

优势

• 实时性：能够快速推送数据，适合需要即时响应的应用场景。
• 资源效率：通过维持持久连接，减少了频繁建立和关闭连接的开销。
• 简洁性：实现相对简单，尤其是在支持现代浏览器的环境中。

SSE技术为实时数据传输提供了一个高效且易于实现的解决方案，特别适用于流式API和需要快速反馈的应用场景。

OpenAI API 示例

使用 curl 调用 OpenAI API 示例，发送提示并请求流式响应：

curl https://api.openai.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -d '{
    "model": "gpt-4o-mini",
    "messages": [{"role": "user", "content": "Tell me a joke"}],
    "stream": true,
    "stream_options": {
      "include_usage": true
    }
  }' \
  --no-buffer

返回输出：

data: {"id":"chatcmpl-abc123","model":"gpt-4o-mini","choices":[{"delta":{"role":"assistant","content":"Why don't scientists trust atoms? Because they make up everything!"},"finish_reason":null}]}
data: {"id":"chatcmpl-abc123","model":"gpt-4o-mini","choices":[{"delta":{"role":"assistant","content":"\n"},"finish_reason":null}]}
data: {"id":"chatcmpl-abc123","model":"gpt-4o-mini","choices":[{"delta":{"role":"assistant","content":"And that's the punchline!"},"finish_reason":"stop"}]}

这个请求发送给 OpenAI API，请求一个笑话的响应。

返回的数据以 data: 开头，后面跟着 JSON 格式的内容，表示不同的响应块。每个数据块都带有创建时间、模型信息和生成的内容。

Anthropic Claude API 示例

Anthropic 的 Claude API 也支持流式响应，示例如下：

curl https://api.anthropic.com/v1/messages \
  -H "Content-Type: application/json" \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -d '{
    "model": "claude-3-sonnet-20240229",
    "messages": [{"role": "user", "content": "Tell me a joke"}],
    "stream": true,
    "max_tokens": 1024
  }' \
  --no-buffer

输出：

event: message_start
data: {"id":"chatcmpl-def456","model":"claude-3-sonnet-20240229","choices":[{"delta":{"role":"assistant","content":"Okay, here's a joke: Why did the scarecrow win an award? Because he was outstanding in his field!"},"finish_reason":null}]}
event: content_block_delta
data: {"id":"chatcmpl-def456","model":"claude-3-sonnet-20240229","choices":[{"delta":{"role":"assistant","content":"\n"},"finish_reason":null}]}
event: message_end
data: {"id":"chatcmpl-def456","model":"claude-3-sonnet-20240229","choices":[{"delta":{"role":"assistant","content":"And that's why! "},"finish_reason":"stop"}]}

显示了多个事件的输出，包括消息开始、内容块更新和消息结束，提供了更丰富的事件跟踪。

返回数据中包含多个事件类型（如 message_start、content_block_delta），每个事件指示生成的内容或状态更新。

Google Gemini API 示例

Google Gemini 的流式 API 返回较大数据块，需要指定内容以获取长响应：

curl "https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:streamGenerateContent?alt=sse&key=${GOOGLE_API_KEY}" \
  -H 'Content-Type: application/json' \
  -d '{
    "contents": [
      {
        "parts": [
          {"text": "Tell me a very long joke"}
        ]
      }
    ]
  }' \
  --no-buffer

输出：

data: {"content":[{"text":"Why did the chicken cross the road? To get to the other side!"},{"text":"And on the other side, it found a whole farm of jokes!"}],"meta":{"usage":{"tokens":50}}}
data: {"content":[{"text":"The chicken decided to tell a joke about itself."}],"meta":{"usage":{"tokens":30}}}

返回的 JSON 数据展示了多部分生成内容和相应的使用元数据。

使用HTTPX访问流

对于 Python 用户，HTTPX 库提供了流式数据处理的简便方法。

使用 httpx-sse 包可以轻松地显示流输出到控制台。

import httpx

async def stream_openai():
    async with httpx.AsyncClient() as client:
        async with client.stream("POST", "https://api.openai.com/v1/chat/completions", json=data) as response:
            async for line in response.aiter_lines():
                print(line)

讲解：此示例展示了如何使用异步 HTTP 客户端进行流式请求，逐行读取并处理返回的数据。

总结

通过服务器发送事件（SSE），LLM API 能够高效地进行实时数据传输。这种机制使得用户能够获取分步更新，增强了交互体验。

不同的 API 实现虽然略有不同，但基本的通信格式和流程是一致的。使用 SSE，开发者能够构建更加响应迅速和用户友好的应用。