ChatGPT streaming with ruby-openai, Rails 7, Hotwire, Turbostream, Sidekiq and Tailwind!

#ChatGPT Streaming.md

How to add ChatGPT streaming to your Ruby on Rails 7 app!

This guide will walk you through adding a ChatGPT-like messaging stream to your Ruby on Rails 7 app using ruby-openai, Rails 7, Hotwire, Turbostream, Sidekiq and Tailwind. All code included below!

• Follow me on Twitter for more Ruby AI at https://twitter.com/alexrudall
• Released under the MIT License - use as you wish :)

First, add the ruby-openai gem! Needs to be at least version 4. Add Sidekiq too.

# Gemfile
gem "ruby-openai", "~> 4.0.0"

# Simple, efficient background processing using Redis.
# https://github.com/sidekiq/sidekiq
gem "sidekiq", "~> 7.0.9"

Install Redis on your machine.

brew install redis

Add Redis and Sidekiq to your Procfile so they run when you run bin/dev.

# Procfile.dev
web: bin/rails server -p 3000
css: bin/rails tailwindcss:watch
sidekiq: bundle exec sidekiq -c 2
queue: redis-server

Add your secret OpenAI token to your .env file. Get one from OpenAI here.

OPENAI_ACCESS_TOKEN=abc123

Add the new routes:

# config/routes.rb
resources :chats, only: %i[create show] do
  resources :messages, only: %i[create]
end

Generate the migrations:

bin/rails generate migration CreateChats user:references
bin/rails generate migration CreateMessages chat:references role:integer content:string

Add the rest of the code, full example files below!

# Controllers.
app/controllers/chats_controller.rb
app/controllers/messages_controller.rb

# Sidekiq job to stream the data from the OpenAI API.
app/jobs/get_ai_response.rb

# Migrations
db/migrate/20230427131800_create_chats.rb
db/migrate/20230427131900_create_messages.rb

# Models
app/models/chat.rb
app/models/message.rb

# Views
app/views/chats/show.html.erb
app/views/messages/_form.html.erb
app/views/messages/_message.html.erb
app/views/messages/create.turbo_stream.erb

20230427131800_create_chats.rb

# db/migrate/20230427131800_create_chats.rb
# bin/rails generate migration CreateChats user:references
class CreateChats < ActiveRecord::Migration[7.0]
  def change
    create_table :chats do |t|
      t.references :user, null: false, foreign_key: true

      t.timestamps
    end
  end
end

20230427131900_create_messages.rb

# db/migrate/20230427131900_create_messages.rb
# bin/rails generate migration CreateMessages chat:references role:integer content:string
class CreateMessages < ActiveRecord::Migration[7.0]
  def change
    create_table :messages do |t|
      t.references :chat, foreign_key: true
      t.integer :role, null: false, default: 0
      t.string :content, null: false
      t.integer :response_number, null: false, default: 0

      t.timestamps
    end
  end
end

chat.rb

# app/models/chat.rb
class Chat < ApplicationRecord
  belongs_to :user
  has_many :messages, dependent: :destroy
end

chats_controller.rb

# app/controllers/chats_controller.rb
class ChatsController < ApplicationController
  before_action :authenticate_user!
  before_action :set_chat, only: %i[show]

  def show
    respond_with(@chat)
  end

  def create
    @chat = Chat.create(user: current_user)
    respond_with(@chat)
  end

  private

  def set_chat
    @chat = Chat.find(params[:id])
  end
end

get_ai_response.rb

# app/jobs/get_ai_response.rb
class GetAiResponse < SidekiqJob
  RESPONSES_PER_MESSAGE = 1
  
  def perform(chat_id)
    chat = Chat.find(chat_id)
    call_openai(chat: chat)
  end

  private

  def call_openai(chat:)
    OpenAI::Client.new.chat(
      parameters: {
        model: "gpt-3.5-turbo",
        messages: Message.for_openai(chat.messages),
        temperature: 0.8,
        stream: stream_proc(chat: chat),
        n: RESPONSES_PER_MESSAGE
      }
    )
  end
  
  def create_messages(chat:)
    messages = []
    RESPONSES_PER_MESSAGE.times do |i|
      message = chat.messages.create(role: "assistant", content: "", response_number: i)
      message.broadcast_created
      messages << message
    end
    messages
  end

  def stream_proc(chat:)
    messages = create_messages(chat: chat)
    proc do |chunk, _bytesize|
      new_content = chunk.dig("choices", 0, "delta", "content")
      message = messages.find { |m| m.response_number == chunk.dig("choices", 0, "index") }
      message.update(content: message.content + new_content) if new_content
    end
  end
end

message.rb

# app/models/message.rb
class Message < ApplicationRecord
  include ActionView::RecordIdentifier

  enum role: { system: 0, assistant: 10, user: 20 }

  belongs_to :chat

  after_create_commit -> { broadcast_created }
  after_update_commit -> { broadcast_updated }

  def broadcast_created
    broadcast_append_later_to(
      "#{dom_id(chat)}_messages",
      partial: "messages/message",
      locals: { message: self, scroll_to: true },
      target: "#{dom_id(chat)}_messages"
    )
  end

  def broadcast_updated
    broadcast_append_to(
      "#{dom_id(chat)}_messages",
      partial: "messages/message",
      locals: { message: self, scroll_to: true },
      target: "#{dom_id(chat)}_messages"
    )
  end

  def self.for_openai(messages)
    messages.map { |message| { role: message.role, content: message.content } }
  end
end

messages_controller.rb

# app/controllers/messages_controller.rb
class MessagesController < ApplicationController
  include ActionView::RecordIdentifier

  before_action :authenticate_user!

  def create
    @message = Message.create(message_params.merge(chat_id: params[:chat_id], role: "user"))

    GetAiResponse.perform_async(@message.chat_id)

    respond_to do |format|
      format.turbo_stream
    end
  end

  private

  def message_params
    params.require(:message).permit(:content)
  end
end

views_chat_show.html.erb

# app/views/chats/show.html.erb
<div class="mx-auto w-full flex">
  <div class="mx-auto">

    <div class="bg-white py-8">
      <div class="mx-auto max-w-lg px-6 ">
        <ul role="list" class="overflow-y-auto max-h-[48vh] flex flex-col-reverse">
          <%= turbo_stream_from "#{dom_id(@chat)}_messages" %>
          <div id="<%= dom_id(@chat) %>_messages">
            <%= render @chat.messages %>
          </div>
        </ul>

        <%= render partial: "messages/form", locals: { chat: @chat } %>
      </div>
    </div>

  </div>
</div>

views_message_create.turbo_stream.erb

# app/views/messages/create.turbo_stream.erb
<%= turbo_stream.append "#{dom_id(@message.chat)}_messages" do %>
  <%= render "message", message: @message, scroll_to: true %>
<% end %>
<%= turbo_stream.replace "#{dom_id(@message.chat)}_message_form" do %>
  <%= render "form", chat: @message.chat %>
<% end %>

views_message_form.html.erb

# app/views/messages/_form.html.erb
<%= turbo_frame_tag "#{dom_id(chat)}_message_form" do %>
  <%= form_with(model: Message.new, url: [chat, chat.messages.new]) do |form| %>
    <div class="my-5">
      <%= form.text_area :content, rows: 4, class: "block shadow rounded-md border border-gray-200 outline-none px-3 py-2 mt-2 w-full", autofocus: true, "x-on:keydown.cmd.enter" => "$event.target.form.requestSubmit();" %>
    </div>

    <div class="grid justify-items-end">
      <%= form.button type: :submit, class: "rounded-lg py-3 px-5 bg-blue-600 text-white inline-block font-medium cursor-pointer" do %>
        <i class="fas fa-paper-plane"></i>
        <span class="pl-2">Send</span>
      <% end %>
    </div>
  <% end %>
<% end %>

views_message_message.html.erb

# app/views/messages/_message.html.erb
# Thanks to github.com/fanahova for this template!
<div id="<%= dom_id(message) %>_messages">
  <% if message.user? %>
    <div class="bg-sky-400 rounded-lg m-8 text-white p-4">
      <%= message.content %>
    </div>
  <% else %>
    <div class="bg-gray-200 rounded-lg m-8 p-4">
      <%= message.content %>
    </div>
  <% end %>
</div>

Thanks @mindtonic ! I find this pattern fairly elegant. Translating this to turbo below:

def stream_proc(messages)
  proc do |chunk, _bytesize|
    new_content = chunk.dig("choices", 0, "delta", "content")
    finish_reason = chunk.dig("choices", 0, "finish_reason")

    message = messages.find { |m| m.response_number == chunk.dig("choices", 0, "index") }

    if new_content.present?
      # update the message in memory then broadcast
      message.content += new_content
      message.broadcast_updated
    end

    if finish_reason.present?
      message.save!
    end
  end
end

I have some problems with the order in which the messages are shown. The messages lose their order, so the question ends up AFTER the response, instead of BEFORE.

Does anyone else have that problem? Any suggestions? I've spent hours on this.

One thing that 'slightly' helps is, adjusting use of:
broadcast_append_later_to ()
broadcast_append_to -> and adding later... -> broadcast_append_later_to

@nickkirt I have seen a couple of times the order got messed up. I've only seen it maybe 2-3 times out of 100. I haven't looked into it much yet or tried to reproduce it.

I'm assuming it's some race condition where one of the later jobs is being processed faster.

It's not that - sorry I wasn't very clear.

It always happens - per message (request from the user) / response. So instead of the response being after the message. It's the other way round. Initially, the order is correct. But when the streaming stops, the response ends up before the message (request).

I'm sure that it's a mistake that I've introduced. But do you have an idea of what is causing the switch - which occurs upon completion of the streaming message.

It always happens - per message (request from the user) / response. So instead of the response being after the message. It's the other way round. Initially, the order is correct. But when the streaming stops, the response ends up before the message (request).

Hey, one bright side of it happening every time it makes it easier to debug and narrow down :) My solution is different from the exact code in the gist, but it's the same idea, and I haven't had this issue.

Is the order still messed up after you do a hard refresh of the Chat page?

Once it's refreshed it's in the right order.

It's working OK ish, but it jitters upon completion of the stream - that's the best I can get it working. But I can't pinpoint what is happening upon completion.

If you want, try it at: aymi.ai (it's a work in progress).

Just chat with a character (requires sign-up, but it's really quick).

No worries if you don't have time.

I believe I solved the order issue by adding an chat message order number field and using CSS to reorder them. Note that I had bad luck using the message ID because there seems to be some maximum order number value that CSS will order.

Something like:

<element style="order: <%= message.response_number %>">

Scrolling is a whole other thing: I just had it scroll to the message every time it received an update. If another message had been sent after the message that was updating, then I did not scroll (would be annoying to jump up or back and forth if two are updating). A scroll "lock" behavior that only scrolls when you are at the bottom would be preferred but not as trivial to implement.

Thanks, @DaAwesomeP! This works great for me, but I haven't seen the out of order message since adding it. I changed the container to a flex column and added the inline style to each message.

I'm querying the order number, but I figure I could persist it to save a query, too. I have an index on created_at in my messages table in my implementation.

In models/chatbot_message.rb

  def message_order
    @message_order ||= conversation.messages.where("created_at <= ?", created_at).count
  end

In my chatbot_messages/_chatbot_message.html.erb partial:

  <div id="<%= dom_id(chatbot_message) %>" style="order: <%= chatbot_message.message_order %>;">

Wow, I fixed it!!!

In the Messages Controller, there's the following line:
GetAiResponse.perform_async(@message.chat_id)

Instead, I used the following:
GetAiResponseJob.perform_now(@message.id)

I replaced it with:
GetAiResponseJob.perform_later(@message.id)

Now it works fine!

I'm surprised that's the problem. But I'm happy that it's working.

Did anyone manage to create a run and stream the response? I'm trying to implement streaming for assistants. I would appreciate any insights 🙏

Did anyone manage to create a run and stream the response? I'm trying to implement streaming for assistants. I would appreciate any insights 🙏

+1

Did anyone manage to create a run and stream the response? I'm trying to implement streaming for assistants. I would appreciate any insights 🙏

I used the basic structure mentioned in here with a number of changes to create one and it seems to be working well so far at limited scale. If you have any specific questions I'll try to follow-up and help!

Some other resources that might be helpful:
Drifting Ruby - streaming llm response
How To Integrate Chatgpt With Rails 7: Step-by-step Tutorial