AI流式输出优先用SSE，何时改用WebSocket

掘金热榜文章《 AI 流式输出都用 SSE，什么时候必须换 WebSocket？》讨论了一个常见选型问题。仅凭“实时”或“流式”这两个标签无法确定方案，还要根据消息方向、交互频率、数据类型和运维条件选择。

SSE 与 WebSocket 如何选，取决于业务链路的具体形态。对于“提交一次 prompt，服务端持续返回 token、状态或工具事件”的 AI 应用，通常优先选择 SSE；只有在同一会话需要持续高频双向控制、协作同步或传输二进制数据时，才更适合改用 WebSocket。浏览器间低延迟音视频或 P2P 通信还应单独评估 WebRTC。

一次 AI 问答，为什么通常先选 SSE？

一次普通 AI 问答的主链路是：客户端提交问题，服务端调用大模型，再把增量结果持续返回。数据方向主要是服务端到客户端。用户上行发送 prompt 仍可通过普通 HTTP 请求完成，这种单向文本流适合 SSE。

单轮问答、摘要生成和代码补全都可以用同一条链路实现。客户端通过 HTTPS 提交任务，服务端在同一个流式响应中输出 token、status 和 done 等事件，前端根据事件类型更新界面。这种模式没有持续双向通信，不必为了一个停止按钮或状态查询就引入全站 WebSocket。

多轮聊天也不天然意味着需要 WebSocket。如果每一轮都是“提交一条消息，接收一条完整或流式回答”，客户端仍然可以重复建立 HTTP 或 SSE 会话。只要后续增加了高频控制、实时协作或双向同步，才更接近 WebSocket 的设计范围。

SSE 也不只适用于 AI。新闻、股票行情、社交更新、通知和日志监控，本质上都是服务端主动向客户端推送数据，可以使用同一套事件流机制。

SSE 和 WebSocket 到底差在哪里？

SSE 建立在 HTTP 响应模型之上。客户端发起请求后，服务器返回 Content-Type: text/event-stream，保持响应持续打开，并不断写入事件。原生 EventSource 会发起持久的 GET 请求，事件可以携带 data、event、id 和 retry 等字段。

WebSocket 通常先通过 HTTP Upgrade 握手建立连接。常见握手包含 Upgrade: websocket 和 Connection: Upgrade，服务器返回 101 Switching Protocols 后，双方进入 WebSocket 协议栈，可以同时发送文本或二进制帧。

下表从通信方向、连接建立、响应类型、重连机制、部署方式和典型场景六个维度对比 SSE 与 WebSocket。

从协议能力看，SSE 以标准 HTTP 长响应承载单向文本事件，在数据格式处理和代理兼容上通常更简便，适合 LLM token 推送；WebSocket 通过握手进入独立协议栈，支持全双工通信和二进制帧，适合高频双向同步和复杂数据传输。

WebSocket 的协议优势是全双工和二进制帧能力，但不能仅凭协议名称就断定它一定拥有更低的端到端延迟。模型首 token 时间、网络路径、代理缓冲和后端调度仍需要实际测量。SSE 的优势更多体现在 HTTP 兼容和较少的连接管理逻辑上。

哪些 AI 场景适合 SSE，哪些适合 WebSocket？

如果只需要服务端持续输出 token 或状态，选择 SSE。偶尔需要停止、重试或确认，但不需要持续双向通信时，“SSE 结合普通 HTTP”通常已经足够。如果要在同一连接内频繁双向控制，应考虑使用 WebSocket。需要传输二进制、音频或其他复杂载荷时，优先评估 WebSocket。浏览器间低延迟媒体或 P2P 通信则应另行评估 WebRTC，不要只比较 SSE 和 WebSocket。

在多轮聊天和 Agent 场景中，关键要看谁在发起控制。如果模型在服务端调用搜索、数据库或其他工具，再把 tool、status 和 token 事件推给客户端，SSE 仍然够用。只有当客户端需要在同一会话里频繁发送审批、取消、工具选择或人工干预消息时，WebSocket 的持续双向能力才真正产生价值。

“停止生成”按钮本身不足以成为 WebSocket 的理由。POST 流可以通过 fetch 的 AbortController 取消，也可以额外调用一个普通 HTTP 中断接口。客户端停止读取响应并不等于服务端和上游模型一定立即停止，后端仍要接收取消信号并向模型服务传播取消。

即时通信、协作编辑和在线游戏需要持续双向消息交换，通常更适合 WebSocket。SSE 更适合服务端主动推送，WebSocket 则适合双方都能随时发起消息的场景。

二进制和多模态场景同样不能只看“流式”二字。SSE 原生传输的是 UTF-8 文本事件；图片上传可以走普通 HTTP，生成结果也可以通过 SSE 返回 URL 或结构化描述。WebSocket 支持原生二进制帧，更适合连续音频、实时控制指令或体积敏感的数据块。浏览器间低延迟音视频还涉及 NAT 穿透和 P2P 媒体传输，应评估 WebRTC，而不是把 SSE 或 WebSocket 当成完整替代品。

AI 流式接口如何把一次请求接到 token 流？

一个便于扩展的混合链路可以这样设计：

客户端                  API / 流服务                 模型服务
  |                          |                          |
  |-- POST prompt ----------->|                          |
  |                          |----- 建立流式调用 ------>|
  |                          |<---- 增量 chunk ---------|
  |<-- SSE token/status 事件 -|                          |
  |<-- tool/done/error 事件 --|                          |
  |                          |                          |
  |-- POST cancel ----------->|----- 取消生成 ---------->|

这里有两个关键点：上行任务和下行结果不一定要绑定在 WebSocket 上；一次性停止操作也不一定要占用同一条流。

SSE 事件可以统一成这样的格式：

retry: 3000

event: token
id: 1042
data: {"delta":"增量文本"}

event: status
id: 1043
data: {"state":"tool_running"}

event: tool
id: 1044
data: {"name":"search","status":"running"}

event: done
id: 1045
data: {"finishReason":"stop"}

data: 用于承载事件内容，event: 标识应用事件类型，id: 为事件提供恢复标识，retry: 指定重连等待时间，事件之间用空行分隔。模型原始 chunk 不宜直接透传，最好先归一化为 token、status、tool、done 和 error 等稳定事件，方便前端处理和断线重放。

如果采用“POST 创建任务，再通过 EventSource 订阅结果”的方式，原生 EventSource 适合 GET 请求。prompt 应放在创建任务的 POST 中，订阅地址只携带任务 ID：

const source = new EventSource(
  `/api/ai/tasks/${taskId}/events`,
  { withCredentials: true }
);

source.addEventListener("token", (event) => {
  const payload = JSON.parse(event.data);
  appendMessage(payload.delta);
});

source.addEventListener("tool", (event) => {
  renderToolStatus(JSON.parse(event.data));
});

source.addEventListener("done", () => {
  source.close();
});

这种方式把普通 HTTPS 和 SSE 分开，适合任务创建、订阅、取消都需要独立鉴权和重试的架构。

如果 prompt 本身需要通过 POST 发送，可以使用 fetch 读取响应流，再按 SSE 格式解析。下面的代码也展示了客户端中断机制：

function parseSseBlock(block) {
  const result = {
    type: "message",
    id: "",
    retry: 0,
    data: []
  };

  for (const line of block.split(/\r?\n/)) {
    if (!line || line.startsWith(":")) continue;

    const colon = line.indexOf(":");
    const field = colon === -1 ? line : line.slice(0, colon);
    let value = colon === -1 ? "" : line.slice(colon + 1);

    if (value.startsWith(" ")) value = value.slice(1);

    if (field === "event") result.type = value;
    if (field === "id") result.id = value;
    if (field === "retry") result.retry = Number(value);
    if (field === "data") result.data.push(value);
  }

  result.data = result.data.join("\n");
  return result;
}

async function streamAnswer(prompt, signal) {
  const response = await fetch("/api/ai/stream", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "Accept": "text/event-stream"
    },
    body: JSON.stringify({ prompt, taskId }),
    signal
  });

  if (!response.ok || !response.body) {
    throw new Error(`stream failed: ${response.status}`);
  }

  const contentType = response.headers.get("content-type") || "";
  if (!contentType.includes("text/event-stream")) {
    throw new Error("unexpected content type");
  }

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

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

      buffer += decoder.decode(value, { stream: true });
      buffer = buffer.replace(/\r\n/g, "\n");

      let boundary;
      while ((boundary = buffer.indexOf("\n\n")) !== -1) {
        const block = buffer.slice(0, boundary);
        buffer = buffer.slice(boundary + 2);
        handleSseEvent(parseSseBlock(block));
      }
    }
  } finally {
    reader.releaseLock();
  }
}

const controller = new AbortController();

stopButton.addEventListener("click", () => {
  controller.abort();
}, { once: true });

streamAnswer("解释 SSE 与 WebSocket 的区别", controller.signal)
  .catch((error) => {
    if (error.name !== "AbortError") throw error;
  });

AbortController 解决的是客户端取消消费。服务端是否真正停止模型调用，仍取决于框架是否传播取消信号，以及上游模型接口是否支持取消。

后端可以把模型增量转换为应用事件。Spring WebFlux 可以使用 Flux 和 MediaType.TEXT_EVENT_STREAM_VALUE 输出 SSE：

record StreamEvent(
    long sequence,
    String type,
    Map<String, Object> data
) {}

@PostMapping(
    value = "/api/ai/stream",
    produces = MediaType.TEXT_EVENT_STREAM_VALUE
)
public Flux<ServerSentEvent<StreamEvent>> stream(
    @RequestBody ChatRequest request
) {
    return aiService.stream(request)
        .map(event -> ServerSentEvent.builder(event)
            .id(Long.toString(event.sequence()))
            .event(event.type())
            .build())
        .onErrorResume(error -> Mono.just(
            ServerSentEvent.<StreamEvent>builder(
                new StreamEvent(
                    0L,
                    "error",
                    Map.of("message", "stream failed")
                )
            )
            .event("error")
            .build()
        ));
}

Django 则可以使用 StreamingHttpResponse 迭代输出事件。应用数据先序列化为单行 JSON，可以避免原始换行破坏 SSE 报文格式：

@require_POST
def ai_stream(request):
    payload = json.loads(request.body)

    def generate():
        yield b"retry: 3000\n\n"

        for sequence, event_type, data in ai_service.stream(
            prompt=payload["prompt"],
            request_id=payload["taskId"]
        ):
            body = json.dumps(
                data,
                ensure_ascii=False,
                separators=(",", ":")
            )
            yield (
                f"id: {sequence}\n"
                f"event: {event_type}\n"
                f"data: {body}\n\n"
            ).encode("utf-8")

    response = StreamingHttpResponse(
        generate(),
        content_type="text/event-stream"
    )
    response["Cache-Control"] = "no-cache"
    response["X-Accel-Buffering"] = "no"
    return response

无论使用哪种框架，都不要在模型调用层堆积缓冲。每次拿到增量 chunk 后应尽快生成并 flush 事件，否则前端可能直到回答结束才一次性收到全部内容。

SSE 断线后能继续吗？

原生 EventSource 支持自动重连，但 SSE 断线重连不等于业务状态自动恢复。当事件携带 id 时，浏览器可以在重连时通过 Last-Event-ID 告知服务端最后收到的位置；服务端仍需保存或重放后续事件。retry 只能控制重连等待时间，不能替代服务端事件存储。

比较稳妥的做法是给任务分配稳定的 taskId，为事件维护递增序号。服务端根据 Last-Event-ID 重放后续事件，客户端按序号去重，并处理终态事件、顺序变化和上下文衔接。当恢复失败时，应返回明确的 resync-required 或重新获取完整结果。

原生 EventSource 和 fetch 的重试不是一回事。EventSource 只适用于其 GET 订阅模型，并内置重连；POST 流的 fetch 在连接断开后不会自动恢复，需要应用保存最后事件 ID，并重新发起带 Last-Event-ID 的请求。

WebSocket 则需要应用自己维护心跳、重连退避、会话恢复、消息确认和去重。连接重建后，客户端通常还要重新订阅任务或会话。对于 Agent 工具调用和审批消息，还要判断旧事件是否已经执行，避免重连后重复调用工具。

重连设计不能只看“断线后重新连上”，还要确认重连后是否会重复生成、重复计费、丢失工具结果或破坏消息顺序。

Nginx 配置会如何影响流式输出？

排查流式输出时，可以按“浏览器是否收到首字节 → 响应头和缓冲 → 网关超时 → 上游连接 → 客户端重连”的顺序定位。如果使用 Nginx 或其他网关，SSE 要重点检查响应缓冲和 Nginx idle timeout；WebSocket 则要重点检查 Upgrade、心跳和连接状态。

可以绕过浏览器，直接用 curl 观察事件是否实时到达：

curl --no-buffer --http1.1 -X POST 'http://localhost:8080/api/ai/stream' -H 'Content-Type: application/json' -H 'Accept: text/event-stream' --data '{"prompt":"测试流式输出","taskId":"demo-1"}'

根据输出表现，可以定位几个关键环节。如果一直没有任何输出，应优先检查应用是否及时 flush、模型调用是否真正以流式方式执行，以及上游连接是否已经建立。如果先返回完整回答，之后才一次性显示，应检查框架缓冲、反向代理缓冲、压缩和网关聚合等因素。如果输出到一半中断，且一段时间内没有新数据，应检查超时配置，以及模型长时间无增量时的保活机制。如果直连正常但经过 Nginx 后异常，应检查代理配置和负载均衡器的空闲超时设置。如果代理日志正常而浏览器异常，则需进一步检查客户端解析、重连和跨域配置。

SSE 不需要把普通 HTTP 连接升级成 WebSocket。一个代表性的 Nginx 配置如下：

location /api/ai/stream {
    proxy_pass http://ai_api;

    proxy_http_version 1.1;
    proxy_set_header Host $host;
    proxy_set_header Connection "";

    proxy_buffering off;
    proxy_cache off;

    proxy_read_timeout 3600s;
}

3600s 只是配置示例，实际值应结合模型可能出现的最长静默时间确定。proxy_read_timeout 控制上游两次读取之间的等待时间，并不是连接创建后的固定总时长。如果模型可能长时间没有 token，应发送可识别的 status 或心跳事件，并确保代理关闭响应缓冲。

WebSocket 则需要显式处理 Upgrade。下面的 map 应放在 Nginx 的 http 上下文：

map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

location /ws/ {
    proxy_pass http://ai_api;

    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection $connection_upgrade;

    proxy_read_timeout 3600s;
}

SSE 更容易接入现有 HTTP 基础设施，不等于“不需要任何长连接配置”。负载均衡器、API 网关、服务网格和防火墙都可能有各自的空闲超时及缓冲策略。WebSocket 还要额外核对 Upgrade 配置、服务器库、心跳和连接级监控。

协议之外，哪些工程条件会改变答案？

安全、会话隔离、连接数和横向扩展能力，都可能推翻“单向流默认 SSE”的判断。比较协议时，应在同一份 PoC 清单中评估性能、安全和架构，而不是只看首屏体验。

安全方面，任务订阅必须绑定用户或租户身份，不能仅凭一个不可猜测的任务 ID 授权。断线重连后需要重新验证会话状态；prompt、工具参数和模型输出也可能包含敏感数据，日志应做脱敏。使用 Cookie 鉴权时，还要验证 CSRF 防护；长连接建立不代表可以绕过原有权限模型。

扩展方面，需要重点评估单实例承载的活跃生成数、断线后的重连风暴以及横向扩容能力。多个流式实例之间应明确任务状态和事件如何共享，避免用户重连到另一实例后丢失上下文。是否采用负载均衡器亲和性、外部事件分发或任务状态存储，应通过 PoC 结合实际架构决定。

评估系统表现时，建议使用同一模型、相近输出长度和一致的网络条件做小规模对比。测试指标应全面覆盖首 token 延迟、token 间隔、完成率、错误率、断线恢复率、重复事件率和资源占用情况。针对 SSE，重点验证 POST 流读取、取消、心跳、代理缓冲、断线重放和重复事件去重；对于 WebSocket，重点考察双向控制、连接升级、应用心跳、异常断线恢复、会话重订阅和二进制消息处理。同时覆盖直连后端和经过网关两条路径，有助于更准确地定位由中间层引发的潜在问题。

用四个问题做初步选择

协议的初步选择主要取决于四个问题：

1. 数据是否仅由服务端持续下行，包括 token、状态、工具进度和完成事件？
2. 客户端是否必须在同一连接中高频上行，即是否存在连续控制、协作同步或实时人工干预需求？
3. 系统是否必须传输二进制或复杂载荷，包括音频帧、实时媒体控制或高频非文本数据？
4. 现有部署和运维体系更适合哪套协议，是否已经具备稳定的 Upgrade、心跳和连接监控能力？

如果前三个问题中只有第一个为“是”，先做 SSE PoC 通常更稳妥；出现持续双向或二进制需求时，再迁移 WebSocket。过渡期可以采用“普通 HTTPS 发起任务、SSE 接收结果、独立 HTTP 取消”的混合架构。下一步先把当前链路的上下行消息、载荷类型和会话生命周期画出来，再用最小实现验证取消、断线和恢复，不必一开始就改造整套通信系统。