anycable-coder

AnyCable Coder

为什么选择 AnyCable 而不是 Action Cable

Action Cable 提供“最多一次”投递——重新连接时消息可能丢失。对于每个数据块都至关重要的大模型流式传输来说，这还不够。

AnyCable 提供：

• 至少一次投递 - 保证消息送达
• 消息顺序 - 数据块按正确顺序送达
• 自动重新连接 - 支持历史记录恢复
• Action Cable 扩展协议 - 在 WebSockets 之上增强可靠性

安装

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

# Client (replace @rails/actioncable)
npm install @anycable/web

服务端频道

基本的大模型流式传输频道

class LlmStreamChannel < ApplicationCable::Channel
  def subscribed
    stream_for current_user
  end

  def generate(data)
    prompt = data["prompt"]

    llm_client.stream(prompt) do |chunk|
      LlmStreamChannel.broadcast_to(
        current_user,
        { type: "chunk", content: chunk }
      )
    end

    LlmStreamChannel.broadcast_to(
      current_user,
      { type: "complete" }
    )
  end
end

使用在线状态跟踪

class ChatChannel < ApplicationCable::Channel
  include AnyCable::Rails::Channel::Presence

  def subscribed
    stream_from "chat_#{params[:room_id]}"
    presence.join(current_user.id, name: current_user.name)
  end

  def unsubscribed
    presence.leave
  end
end

客户端实现

从 Action Cable 迁移

// Before (Action Cable)
import { createConsumer } from "@rails/actioncable"

// After (AnyCable) - same API!
import { createConsumer } from "@anycable/web"

export default createConsumer()

现代 AnyCable 客户端

import { createCable } from "@anycable/web"

const cable = createCable()

// Class-based channel
import { Channel } from "@anycable/web"

class LlmStreamChannel extends Channel {
  static identifier = "LlmStreamChannel"

  async generate(prompt) {
    return this.perform("generate", { prompt })
  }
}

// Subscribe and handle chunks
const channel = new LlmStreamChannel()
cable.subscribe(channel)

await channel.ensureSubscribed()

channel.on("message", (msg) => {
  if (msg.type === "chunk") {
    appendToResponse(msg.content)
  } else if (msg.type === "complete") {
    finishResponse()
  }
})

channel.generate("Explain WebSockets")

直接流（已签名）

// Subscribe directly to a stream without channel class
const cable = createCable()
const stream = cable.streamFrom("llm_response/user_123")

stream.on("message", (msg) => console.log(msg))

在线状态 API

const chatChannel = cable.subscribeTo("ChatChannel", { roomId: "42" })

// Join presence
chatChannel.presence.join(user.id, { name: user.name })

// Listen for presence events
chatChannel.presence.on("presence", (event) => {
  if (event.type === "join") {
    console.log("User joined:", event.id, event.info)
  } else if (event.type === "leave") {
    console.log("User left:", event.id)
  }
})

// Get current presence
const users = await chatChannel.presence.info()

// Leave presence
chatChannel.presence.leave()

大模型流式传输模式

完整实现

# app/channels/assistant_channel.rb
class AssistantChannel < ApplicationCable::Channel
  def subscribed
    stream_for current_user
  end

  def ask(data)
    conversation_id = data["conversation_id"]
    message = data["message"]

    # Broadcast start
    broadcast_event("start", conversation_id:)

    # Stream LLM response
    response = ""
    llm.chat(message) do |chunk|
      response += chunk
      broadcast_event("chunk", conversation_id:, content: chunk)
    end

    # Save and broadcast completion
    Message.create!(conversation_id:, role: "assistant", content: response)
    broadcast_event("complete", conversation_id:)
  rescue => e
    broadcast_event("error", conversation_id:, message: e.message)
  end

  private

  def broadcast_event(type, **payload)
    AssistantChannel.broadcast_to(current_user, { type:, **payload })
  end

  def llm
    @llm ||= OpenAI::Client.new
  end
end

// app/javascript/channels/assistant_channel.js
import { Channel } from "@anycable/web"

export default class AssistantChannel extends Channel {
  static identifier = "AssistantChannel"

  constructor() {
    super()
    this.responseBuffer = ""
  }

  async ask(conversationId, message) {
    this.responseBuffer = ""
    return this.perform("ask", { conversation_id: conversationId, message })
  }

  // Override to handle message types
  receive(message) {
    switch (message.type) {
      case "start":
        this.onStart?.(message.conversation_id)
        break
      case "chunk":
        this.responseBuffer += message.content
        this.onChunk?.(message.content, this.responseBuffer)
        break
      case "complete":
        this.onComplete?.(this.responseBuffer, message.conversation_id)
        break
      case "error":
        this.onError?.(message.message)
        break
    }
  }
}

配置

AnyCable 服务端

# config/anycable.yml
production:
  broadcast_adapter: nats
  redis_url: <%= ENV.fetch("REDIS_URL") %>

  # Enable reliable streams
  streams_history_size: 100
  streams_history_ttl: 300

Cable URL

<%= action_cable_meta_tag %>

// Auto-detects from meta tag, or specify explicitly
import { createCable } from "@anycable/web"
createCable("wss://cable.example.com/cable")

部署

Procfile

web: bundle exec puma -C config/puma.rb
anycable: bundle exec anycable
ws: anycable-go

Docker Compose

services:
  web:
    command: bundle exec puma
  anycable:
    command: bundle exec anycable
  ws:
    image: anycable/anycable-go:1.6
    environment:
      ANYCABLE_RPC_HOST: anycable:50051
      ANYCABLE_REDIS_URL: redis://redis:6379

Action Cable 与 AnyCable 对比

特性	Action Cable	AnyCable
投递保证	最多一次	至少一次
消息顺序	不保证	保证
重新连接时的历史记录	无	有（可配置）
在线状态跟踪	手动	内置
性能	Ruby 线程	Go 服务端
大模型流式传输	不可靠	可靠

反模式

反模式	问题	解决方案
使用 Action Cable 进行大模型流式传输	重新连接时数据块丢失	使用 AnyCable
忽略消息顺序	响应混乱	AnyCable 会自动处理
手动处理重新连接逻辑	复杂且容易出错	使用 AnyCable 客户端
不跟踪在线状态	用户状态未知	使用内置在线状态 API

输出格式

使用 AnyCable 实现实时功能时：

1. 频道 - 服务端 Ruby 频道，支持广播
2. 客户端 - JavaScript 订阅及消息处理
3. 配置 - anycable.yml 和部署设置
4. 错误处理 - 优雅降级模式