AnyCable、Rails与LLM流式传输的陷阱

到2025年底，很难找到一个没有AI功能（或未曾尝试整合AI功能）的Web应用程序。在用户界面中呈现AI生成内容时，LLM响应流式传输能力至关重要。它们使我们能够快速提供反馈，减少AI的感知延迟，从而改善用户体验。但尽管框架和库为实现流式更新提供了现成的解决方案，实时世界却隐藏着许多陷阱。让我在Ruby on Rails的AI驱动应用程序中揭示这些陷阱（以及如何避免它们）。

Ruby on Rails包含一个向用户提供实时更新的组件：Action Cable。它使用WebSocket作为传输层，并可以依赖各种后端实现水平扩展（数据库、Redis或通过自定义适配器实现的任何后端）。再加上Hotwire和RubyLLM，您只需几行代码就能获得一个向用户流式传输LLM响应的完整解决方案。

一个最小示例如下，包含一个HTML模板和一些Ruby代码：

1
2
3
4
5


<%= turbo_stream_from "chat_42" %>

<div id="chat">
  思考中...
</div>

1
2
3


RubyLLM.chat.ask("实时HTTP传输有哪些陷阱？") do |chunk|
  Turbo::StreamsChannel.broadcast_append_to("chat_42", target: "chat", html: chunk.content)
end

方法很简单，但结果……嗯，不那么可预测。为了更好地演示潜在的UI“幻觉”（乱序显示），我创建了一个简单的AI驱动Rails应用程序——Proposer，它可以帮助遵循选定最佳实践来准备会议演讲提案。这是一个纯粹的rails new（版本8.1，带Hotwire）项目，加入了RubyLLM。目前还没有什么花哨的东西。

想直接看代码吗？它已经在GitHub上了：palkan/proposer。

以Proposer作为试验场，我们将探讨以下主题：

流式传输“脱轨”
用AnyCable恢复秩序
展望未来：持久流（Durable Streams）

流式传输“脱轨”

首先快速概述一下我们的应用程序是如何工作的。流程如下：

用户通过HTML表单提交提案生成请求。
触发一个生成工作流，在后台执行（通过Active Job）。该工作流由新的Active Job Continuation API支持，包含几个步骤，每个步骤对应提案的一个部分（标题、摘要、详细信息）。
在每个工作流步骤中，我们向LLM执行流式请求，并通过Turbo Stream将数据块广播到用户的浏览器。

最后一步（对我们来说最有趣的部分）如下所示：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


def generate_field(id, prompt)
  response = chat.ask(prompt) do |chunk|
    next if chunk.content.blank?

    Turbo::StreamsChannel.broadcast_append_to(
      [proposal, field],
      target: dom_id(proposal, field),
      html: chunk.content
    )
  end

  proposal.update!(id => response.content)
end

我们按收到数据块的速度进行广播，并将最终生成结果存储在数据库中（#update调用）。

用户在浏览器中会看到什么？让我们看几个例子。

无法无“序”

不，上面的视频不是AI生成的，它是真实的：数据以错误的顺序到达浏览器，没有先入先出的保证。这就是Action Cable暴露其多线程本质的方式。在底层，Action Cable使用线程池来分发广播工作（通过对每个订阅的连接调用connection#transmit）。当您连续向同一个客户端广播100条消息时，4个Ruby线程会拾取它们并同时传输给客户端——糟糕！

事实证明，即使数据块之间存在自然延迟（它们并非同时发送），也不足以保证正确的顺序。我们如何实现有序呢？有很多选项可以尝试：

不发送数据块，而是发送累积的数据。 在我们的例子中，可以使用#broadcast_update_to代替#broadcast_append_to。这并不能完全消除“幻觉”（用户仍会看到一些抖动），但结果看起来会是正确的。然而，出于环境考虑，我仍然不愿采用这种方式：当只添加了几个字节时发送完整消息是一种资源浪费（并增加网络流量）。
节流。 累积数据块，最多每100毫秒广播一次。然而，在更高负载下，100毫秒可能不够（当广播消息可能堆积时）。
使用更快的发布/订阅适配器。 如今，Rails默认建议使用Solid Cable，这是一个基于数据库的Action Cable适配器。上面的视频就是使用这种设置录制的。当使用Redis进行发布/订阅时，这种“幻觉”在每个客户端上发生频率较低，但当广播速率较高时仍然明显。
使用Action Cable Next或Async Cable。 Action Cable的下一代版本actioncable-next目前可以作为gem使用（但预计最终会合并到Rails中）。它为广播提供了一种适合Turbo的快速通道模式，消除其中一个线程池，并使广播速度大大提高（因此可以在新数据块到达前被处理）。类似地，async-cable也实现了更快的广播循环。

然而，以上方法都不能提供100%的顺序保证，因为“线程的诅咒”依然存在，并可能在负载下显现。在客户端实现重新排序会更健壮（参见RubyLLM文档中的示例），但同时需要更多的工作（尽管自定义Turbo Stream操作听起来是个不错的主意🤔）。

顺序并不是Action Cable唯一缺失的保证。

没有可靠的网络这回事

第二个例子展示了实时通信中一个非常典型的陷阱——未能考虑连接丢失。看看在线状态指示器（绿圈表示“在线”，红色表示无信号），当它变红时会发生什么：数据块丢失，UI状态损坏。

网络问题是不可避免的。包括Action Cable在内的大多数WebSocket客户端都支持自动重新连接，但它们不会自动捕获错过的消息。LLM响应的数据块永远丢失了，因为我们不存储它们；它们是短暂的。然而，从UI的角度来看，我们需要一种方法来让短暂断开连接的客户端访问它们——我们该怎么做呢？

Action Cable没有为此提供任何解决方案（尽管理论上Solid Cable可以，因为它将消息存储在数据库中）。换句话说，Action Cable提供的是“最多一次”的交付保证——对于可靠的LLM响应流式传输来说不够。我们至少需要“至少一次”的保证！

我们可以通过发送累积数据而不是数据块，并在重新连接时请求当前已提交的状态，来模拟“至少一次”的交付。然而，有一个更好的替代方案，它不需要重新思考流式传输的实现或担心潜在的边缘情况——我们可以使用一个更合适的实时服务器，它开箱即用地提供可靠性。

用AnyCable恢复秩序

tl;dr AnyCable同时提供消息排序和“至少一次”的交付保证。

是的，我们可以通过简单地从一个服务器实现切换到另一个，从Action Cable切换到AnyCable，一石二鸟。无需技巧，无需变通——只需几个终端命令，就能可靠地将LLM响应流式传输到客户端：

1
2


bundle add anycable-rails
bin/rails g anycable:setup

让我们看看在Proposer应用程序中实施AnyCable的过程：

就是这样！（嗯，差不多：您可以看到需要自己处理或委托给AI伙伴的TODO项目列表。）

现在，我们不再需要担心实时数据流传输的复杂性——服务器会处理它：

应用程序代码保持不变，并且它以所见即所得的方式工作：“我逐个广播LLM响应数据块，客户端按相同顺序接收所有数据块”。

想了解更多关于AnyCable可靠性的技术细节吗？去看看我们的文档吧！好吧，我们知道我们的文档有点糟糕🫤（升级版即将到来🤞）。让我以列表形式强调最重要的几点：

服务器将发布（广播的消息）存储在类似日志的结构中（单节点安装时存储在内存中，Pro集群中存储在Redis Streams中）。
服务器为每条消息附加位置元数据。
客户端（我们钟爱的@anycable/web）跟踪订阅的流和位置，并在重新连接时自动从最后看到的位置开始追赶。
（可选）客户端可以在初始订阅时请求给定时间范围内的历史消息（这样我们就可以消除页面加载期间的竞争条件，或者在Proposer的情况下，在页面重新加载后存活下来）。

因此，这些保证被内置于AnyCable服务器和Action Cable Extended通信协议中。是的，我们不得不在WebSocket和现有的Action Cable协议之上发明一种自定义协议，以提供更好的交付保证。（为什么火星上没有实现可靠实时通信的通用标准？我不知道。我知道的是，未来这种情况有可能改变。）

展望未来：持久流（Durable Streams）

最近，我们在ElectricSQL的朋友宣布了一项新倡议：一种用于可靠数据流式传输的HTTP协议，称为持久流（Durable Streams）。

其目标是标准化客户端和服务器之间分别消费和生产数据流的通信语言。对数据格式没有限制（只要可以表示为字节的仅追加日志），没有强制的认证/授权模式（使用您自己的），没有自定义的应用层协议（好吧，就像我们在AnyCable做的那样😄）。唯一的要求是传输层：HTTP（一次性、轮询或SSE）。

持久流协议和AnyCable的可靠性设计有很多共同点：

两者都假设将流存储为日志（有序且可从任何位置访问），每个流一个日志（而不是每个客户端-流对）。
两者都使用由客户端跟踪的流偏移量来实现可恢复性。

如果我们已经有现有的解决方案（比如AnyCable），持久流有什么意义呢？互操作性。在这篇文章中，我们展示了AnyCable与Action Cable的互操作性如何使其成为在Rails应用中实现可靠实时流式传输的理所当然的选择。尽管如此，我们仍然必须切换客户端实现——虽然只需要几行代码，但毕竟还是需要改。

现在想象一下，您不必担心客户端或服务器的供应商锁定，并且可以在开放标准上设计您的应用程序。您可以在任何平台上使用任何兼容的客户端，并根据需要升级服务器：开始时使用嵌入式实现和内存存储，随着负载增加切换到托管解决方案，或者在达到更高负载时构建自己的高可用集群。

这种理念与我们在AnyCable的理念非常一致。我们名字中的“Any”意味着对任何人都要有帮助：任何语言、任何平台，以及……任何协议！

是的，您猜对了！AnyCable正在逐步采用持久流。我们将从实现协议的“读取”部分开始（这样您将能够消费持久流，但仍然使用AnyCable API发布数据），然后看看效果如何。

敬请期待！