リアルタイム通信を彩るサーバ送信イベント(SSE)の魅力と可能性

<div class="form">
  <label for="message">メッセージを入力</label>
  <textarea id="message" name="message" data-sse-target="question" rows="1" placeholder="Message" class="scrollbar"></textarea>

  <button type="submit" aria-label="送信" data-sse-target="button" data-action="click->sse#generate">

-snip-

  </button>
  <div class="text-small">Enter で改行 / Ctrl + Enter または Command + Enter で送信</div>
</div>

import {
  Controller
} from "@hotwired/stimulus"

export default class extends Controller {
  connect() {
  }
  static targets = ["question", "button"]
  static values = {
    room: String,
    chat: String
  }

  async generate() {
    const content = this.questionTarget.value
    let message_id = null

    this.#disableSubmitButton()
    this.#writeQuestionMessage()

    const response = await fetch(`/rooms/${this.roomValue}/chats/${this.chatValue}/messages`, {
      method: "POST",
      cache: "no-cache",
      keepalive: true,
      headers: {
        "Content-Type": "application/json",
        "Accept": "text/event-stream",
      },
      body: JSON.stringify({
        message: {
          content: content
        }
      }),
    })

    const reader = response.body.pipeThrough(new TextDecoderStream()).getReader()
    this.#writeAnswerMessage(reader)
  }

-snip-

  async #writeAnswerMessage(reader) {
    const templateMessage = document.getElementById('template-message-you');
    let documentFragment = templateMessage.content.cloneNode(true);
    document.getElementById('messages').appendChild(documentFragment);

    this.#disableActionLinks()

    let lastMessageText = document.querySelector('.messages .message:last-child .text');

    while (true) {
      let {
        value,
        done
      } = await reader.read();
      if (!done) {
        lastMessageText.innerText = lastMessageText.innerText + value
        this.#scrollToBottom()
      } else {
        break
      }
    }

-snip-

class MessagesController < ApplicationController
  include ActionController::Live

  def create  
    -snip-

    query = (message_params)

    response.headers['Content-Type'] = 'text/event-stream'
    response.headers['Last-Modified'] = Time.now.httpdate

    Ai.new.chat(query) do |fragment|
      response.stream.write(fragment)
    end

    -snip-

  rescue ActionController::Live::ClientDisconnected
    response.stream.close
  ensure
    response.stream.close
  end

class Ai
  def initialize
    access_key_id = Rails.application.credentials.dig(:aws, :access_key_id)
    secret_access_key = Rails.application.credentials.dig(:aws, :secret_access_key)
    credentials = Aws::Credentials.new(access_key_id, secret_access_key)
    @client = Aws::BedrockRuntime::Client.new(region: "us-east-1", credentials: credentials)
  end

  def chat(query)
    body = {
      "max_tokens": 1000,
      "anthropic_version": "bedrock-2023-05-31",
      "messages": [
        {
          "role": "user",
          "content": [
            {
              "type": "text",
              "text": query
            }
          ]
        }
      ]
    }

    params = {
     "model_id": "anthropic.claude-3-haiku-20240307-v1:0",
     "content_type": "application/json",
     "accept": "application/json",
      body: body.to_json
    }


    @client.invoke_model_with_response_stream(params) do |stream|
      stream.on_error_event do |event|
        raise event 
      end
      stream.on_event do |event|
        hash_event = JSON.parse(event.bytes)
        case hash_event['type']
        when 'content_block_delta'
          fragment = hash_event['delta']['text']
          yield(fragment)
        when 'message_stop'
          return
        end
      end
    end
  end
end