anycable-coder

AnyCable Coder

Action Cable ではなく AnyCable を選ぶ理由

Action Cable が提供するのは「最大 1 回」の配信です。再接続時にメッセージが失われることがあります。すべてのチャンクが重要な LLM のストリーミングには不十分です。

AnyCable が提供する機能：

• 最低 1 回の配信 - メッセージは確実に届きます
• メッセージの順序保証 - チャンクは正しい順序で届きます
• 自動再接続 - 履歴の復元にも対応しています
• Action Cable 拡張プロトコル - WebSocket 上で信頼性を強化します

インストール

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

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

サーバーサイドチャンネル

基本的な LLM ストリーミングチャンネル

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()

LLM ストリーミングパターン

完全な実装

# 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

ケーブル 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
配信保証	最大 1 回	最低 1 回
メッセージの順序	保証なし	保証あり
再接続時の履歴	なし	あり（設定可能）
プレゼンス追跡	手動	組み込み
パフォーマンス	Ruby スレッド	Go サーバー
LLM ストリーミング	信頼性なし	信頼性あり

アンチパターン

アンチパターン	問題	解決策
LLM ストリーミングに Action Cable を使う	再接続時にチャンクが失われる	AnyCable を使う
メッセージの順序を無視する	応答が文字化けする	AnyCable が自動的に処理します
再接続ロジックを手動で実装する	複雑で、エラーが発生しやすい	AnyCable クライアントを使う
プレゼンス追跡を行わない	ユーザーの状態が不明になる	組み込みのプレゼンス API を使う

出力形式

AnyCable でリアルタイム機能を実装する場合：

1. チャンネル - ブロードキャストを行うサーバーサイドの Ruby チャンネル
2. クライアント - メッセージ処理を行う JavaScript のサブスクリプション
3. 設定 - anycable.yml とデプロイ設定
4. エラー処理 - 適切に機能を縮退させるパターン