如何实现 SSE 令牌流的断点续传、可取消及多设备支持

# 如何让 SSE 令牌流具备可恢复性、可取消性且支持多设备 —— /dev/knill 来源: https://zknill.io/posts/everyone-said-sse-token-streaming-was-easy/ 过去，代理（Agents）是你同步交互的对象。现在，它们是在你工作时在后台运行的东西。当你做出这种改变时，传输层就会崩溃。 但很多人说：“不，你可以使用带有 Last-Event-ID 的 Server-Sent Events (SSE) 来获得持久流，这很容易”。是的，这一切都是可行的。但我认为这并不*简单*。所以让我们逐步讲解如何做到这一点，由你自己来判断。 回顾之前的文章和讨论 我想逐步讲解的高级聊天机器人功能包括： - **可恢复流**——在响应中途刷新页面，找回正在进行的令牌，而不是等待完整响应存入数据库。 - **取消**——当用户改变主意时，即使在连接允许断开和重连的情况下，也能在响应中途停止大型语言模型（LLM）。 - **多设备**——在第二个设备或浏览器上打开相同的对话，并实时获取进行中的响应和任何新提示。 这些功能在 SSE 上都是可行的。但它们是否*简单*，正是我们要发现的。 ## 令牌与 API 响应 令牌是 LLM 生成的独立文本片段，但从 LLM 提供商那里获得的实际响应包含更多内容。响应具有略微不同的结构和格式，但大致遵循相似的模式。 某种‘开始’事件，包含文本或工具调用请求的‘增量’事件，以及某种‘结束’事件。 要获取完整的响应文本，你可以将文本增量拼接在一起，或者某些 API 会在最后以自身的事件类型提供‘完整’响应。 Vercel AI SDK: `` 1 2 3 4 5 6 7 8 9 10 `` `` {"type":"text-delta","value":"Let me"} {"type":"text-delta","value":" look that up."} {"type":"tool-call-streaming-start","value":{"toolCallId":"call_001","toolName":"search"}} {"type":"tool-call-delta","value":{"toolCallId":"call_001","argsTextDelta":"{\"query\":\"weather Belfast\"}"}} {"type":"tool-call","value":{"toolCallId":"call_001","toolName":"search","args":{"query":"weather Belfast"}}} {"type":"tool-result","value":{"toolCallId":"call_001","result":{"temp":"14°C","condition":"cloudy"}}} {"type":"text-delta","value":"It's currently"} {"type":"text-delta","value":" 14°C and cloudy"} {"type":"text-delta","value":" in Belfast."} {"type":"finish-message","value":{"finishReason":"stop","usage":{"promptTokens":30,"completionTokens":28}}} `` OpenAI Responses API: `` 1 2 3 4 5 6 7 8 9 `` `` {"event":"response.created","data":{"id":"resp_abc123","object":"response","status":"in_progress","model":"gpt-4o"}} {"event":"response.output_item.added","data":{"output_index":0,"item":{"id":"item_001","type":"message","role":"assistant"}}} {"event":"response.content_part.added","data":{"output_index":0,"content_index":0,"part":{"type":"output_text","text":""}}} {"event":"response.output_text.delta","data":{"output_index":0,"content_index":0,"delta":"Hello"}} {"event":"response.output_text.delta","data":{"output_index":0,"content_index":0,"delta":"! How can I"}} {"event":"response.output_text.delta","data":{"output_index":0,"content_index":0,"delta":" help you today?"}} {"event":"response.output_text.done","data":{"output_index":0,"content_index":0,"text":"Hello! How can I help you today?"}} {"event":"response.output_item.done","data":{"output_index":0,"item":{"id":"item_001","type":"message","role":"assistant","content":[{"type":"output_text","text":"Hello! How can I help you today?"}]}}} {"event":"response.completed","data":{"id":"resp_abc123","object":"response","status":"completed","usage":{"input_tokens":25,"output_tokens":12}}} `` Anthropic API: `` 1 2 3 4 5 6 7 8 9 10 `` `` {"event":"message_start","data":{"type":"message_start","message":{"id":"msg_01XFDUDYJgAACzvnptvVoYEL","type":"message","role":"assistant","content":[],"model":"claude-sonnet-4-20250514"}}} {"event":"content_block_start","data":{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}} {"event":"content_block_delta","data":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}} {"event":"content_block_delta","data":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"! How"}}} {"event":"content_block_delta","data":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" can I"}}} {"event":"content_block_delta","data":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" help you"}}} {"event":"content_block_delta","data":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" today?"}}} {"event":"content_block_stop","data":{"type":"content_block_stop","index":0}} {"event":"message_delta","data":{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":12}}} {"event":"message_stop","data":{"type":"message_stop"}} `` 你会发现，响应的每一‘行’包含的元数据相当多，而文本增量数据却很少。 例如，Anthropic API 的单个事件（行）仅包含 5 个字符的文本增量，却占用了 125 个字符： `` 1 2 3 4 5 6 7 8 9 10 11 `` `` { "event": "content_block_delta", "data": { "type": "content_block_delta", "index": 0, "delta": { "type": "text_delta", "text": "Hello" } } } `` 为什么这很重要？一旦你决定存储每个令牌，这就很重要了，我们稍后会谈到这一点。 ## 基准：如何将令牌流式传输到单个客户端 *这是基本门槛，对吧？* 从几乎所有 AI 聊天机器人演示都有的基础开始。用户向服务器发送带有提示和当前对话历史的 HTTP POST 请求。服务器运行代理，客户端保持响应开放，等待来自代理的响应令牌的 SSE 流。 SSE 代理服务器架构 一旦流式传输完成，服务器将完整响应存储在数据库中。存储在数据库中的‘对话历史’实际上是为了在页面刷新时重新填充客户端，服务器本身并不需要它。 这对于一个用户、一个设备、一个不断开的连接来说工作正常。但默认情况下，你会得到如下 gif 中的行为：如果刷新页面，进行中的令牌流会丢失，只有当流完成并将‘完整’响应存储在数据库中时，响应才会再次可用。 Claude.ai 页面刷新 ## 如何使流可恢复 SSE 规范有一种恢复流的机制，基于 HTTP 头 `Last-Event-ID`。 想法是，如果你服务器发送的事件流中的每个‘事件’都有一个唯一 ID，那么客户端可以跟踪它接收到的最后一个事件，当连接断开时，它可以重新连接并告诉服务器从该最后一个事件恢复。 也许你决定每个‘响应’都有一个 id，比如：`abcxyz`。并且每个令牌按顺序有一个索引。所以 `abcxyz:0` 是第一个令牌，`abcxyz:1` 是第二个令牌，依此类推。 水平扩展架构 大多数应用程序构建于类似上述的架构，其中有许多无状态的、可水平扩展的服务器副本来处理客户端请求。数据存储在数据库中。 因此，来自每个 LLM 响应的所有令牌都需要存储在数据库中或某个令牌缓存中，以防带有 `Last-Event-ID` 的‘恢复’请求被路由到与启动流的服务器副本不同的副本。 因为采用无状态服务器副本设计，任何副本都可以处理任何用户请求。所有状态都存在于某个数据库中，副本按需查找。 在不同服务器副本上恢复 问题是你现在需要将每个令牌写入数据库。你是在赌客户端可能会断开的情况下，将这些令牌写入数据库。对于 HTTP SSE 流保持完整并交付所有令牌的成功请求，你永远不需要你写入数据库的每个令牌的数据。 记得每个令牌事件包含大量元数据，但文本增量不多吗？这意味着为了少量文本而进行大量数据库写入。大量写入，价值却不高。 而且一旦 LLM 完成该响应，单个令牌就无用武之地了，因为‘完整’响应取代了它们所有。所以一旦流完成，你需要清理所有单个令牌，转而使用‘完整响应’。 这对于仅在连接断开时才需要的功能来说，是一种相当昂贵的写放大。过去是 1 个请求，1 个响应，和几个数据库查询。现在你每个令牌都要进行一次数据库写入。（也别告诉我你可以批量处理而不深入思考；因为批量处理只是向客户端承诺可以从一个你数据库中可能没有的 ID 恢复）。 ## 如何处理取消 一旦你使流可恢复，取消就变得有点尴尬。也许之前你假设客户端断开的连接意味着服务器可以取消请求。但现在，你假设客户端可能会带着 `Last-Event-ID` 回来恢复流，所以连接断开不再意味着取消。 相反，你需要通过数据库来传递取消信号。一个单独的 `POST /cancel/{response_id}` 端点，将取消标记写入 LLM 推理过程正在写入的同一共享存储中。处理推理的过程在令牌之间检查标记，如果看到标记则中止上游 LLM 调用。 处理 LLM 响应的副本可能不是接收来自客户端取消请求的副本，因此共享存储是传递取消的路由器。 ## 如何支持多设备 多设备是两个问题，而不是一个，而且它们没有相同的答案。 对于第一个问题，你已经做了一部分，因为你正在数据库中存储响应的单个令牌。所以你可以将这些令牌提供给多个设备。第二个设备可以请求相同的对话并接收历史，然后接收任何进行中响应的令牌流。 但还有第二个问题； > 如果设备 A 发送新提示并开始接收令牌流响应，设备 B 如何知道有新的提示和响应需要渲染？ ... 设备 B 并不知道。 人们会告诉你，你需要所有客户端轮询服务器以获取新数据。在“构建实时功能的模式” (https://zknill.io/posts/patterns-for-building-realtime/) 中，我们讨论了为什么轮询很糟糕；如果不频繁轮询，则会牺牲延迟；如果频繁轮询，则会用流量冲击服务器。两者都不好。 当然你可以进行更长时间的轮询，比如，你知道的，长轮询。但那仍然是轮询。 ## 或者，使用为你处理这些的传输层 警告 如果你只是想要操作指南，请在此停止阅读。因为我将谈论我认为更好的东西，这可能对某些人来说太‘商业化’了。 所有这些功能在 SSE 上都是可行的，这就是这篇文章的内容。但我认为它们并不‘简单’。它们是笨拙且低效的解决方案，以解决 HTTP 并不是流式传输 LLM 令牌和构建异步代理应用程序的良好传输层这一事实。 Ably 发布/订阅频道 我在 Ably (https://ably.com/ai-transport) 工作，我正在为 AI 应用程序构建一个专用传输层，很好地支持令牌流式传输。它基于上图中的发布/订阅模式。 与基于 HTTP 的 SSE 流式传输的关键区别在于： 1. 即使客户端断开与发布/订阅频道的连接，该频道仍然存在，所以服务器可以继续发布令牌，并且这些令牌在客户端重新连接的瞬间即可用。这解耦了连接生命周期和代理生命周期。 2. 多个用户或多个设备可以连接到同一个发布/订阅频道并获得完全相同的令牌流。频道确保客户端实时接收令牌，并且 Ably SDK 自动处理重连、回溯错过的令牌以及历史记录。 3. 频道自动将令牌增量压缩为完整响应，所以追赶进度的客户端每个完整响应只收到 1 条消息，而不是流式传输每个单个令牌。 4. 取消、中断和引导很容易，因为频道处理从客户端发布中断到运行代理的服务器进程的路由。不再需要通过数据库路由取消或后续操作，频道会自动为你处理路由。 所以是的，如果你愿意，你可以构建所有这些东西。但我不认为它们‘简单’，或者认为基于 HTTP 的 SSE 是适合从 LLM 流式传输令牌和构建异步代理应用程序的传输层。