I’ve updated this post to use the ChatGPT API. You can find the updated version here: https://www.hackwithgpt.com/blog/streaming-chatgpt-responses-with-phoenix-liveview/
I was recently messing around with the GPT-3 API and wanted to recreate the streaming generation effect that OpenAI has implemented with ChatGPT in a LiveView. I’m not sure if this the best way to do it, but it works!
Streaming GPT-3 API Responses
If you’ve worked with the GPT-3 API in the past (especially the larger models), request latency can be a significant detriment to user experience. One way to overcome this is taking advantage of GPT-3’s ability to stream tokens in batches as data-only server-sent events.
After a bit of digging, I found this article for using HTTPoison to stream large HTTP responses. Fortunately, this same paradigm works for streaming server-sent events from the OpenAI API. We can rework the logic in the linked article to convert OpenAI responses into streams of tokens:
defmodule StreamingText.OpenAI do
require Logger
def stream(prompt) do
url = "https://api.openai.com/v1/completions"
body = Jason.encode!(body(prompt, true))
headers = headers()
Stream.resource(
fn -> HTTPoison.post!(url, body, headers, stream_to: self(), async: :once) end,
&handle_async_response/1,
&close_async_response/1
)
end
defp close_async_response(resp) do
:hackney.stop_async(resp)
end
defp handle_async_response({:done, resp}) do
{:halt, resp}
end
defp handle_async_response(%HTTPoison.AsyncResponse{id: id} = resp) do
receive do
%HTTPoison.AsyncStatus{id: ^id, code: code} ->
Logger.info("openai,request,status,#{inspect(code)}")
HTTPoison.stream_next(resp)
{[], resp}
%HTTPoison.AsyncHeaders{id: ^id, headers: headers} ->
Logger.info("openai,request,headers,#{inspect(headers)}")
HTTPoison.stream_next(resp)
{[], resp}
%HTTPoison.AsyncChunk{id: ^id, chunk: chunk} ->
HTTPoison.stream_next(resp)
parse_chunk(chunk, resp)
%HTTPoison.AsyncEnd{id: ^id} ->
{:halt, resp}
end
end
defp parse_chunk(chunk, resp) do
{chunk, done?} =
chunk
|> String.split("data:")
|> Enum.map(&String.trim/1)
|> Enum.reject(&(&1 == ""))
|> Enum.reduce({"", false}, fn trimmed, {chunk, is_done?} ->
case Jason.decode(trimmed) do
{:ok, %{"choices" => [%{"text" => text}]}} ->
{chunk <> text, is_done? or false}
{:error, %{data: "[DONE]"}} ->
{chunk, is_done? or true}
end
end)
if done? do
{[chunk], {:done, resp}}
else
{[chunk], resp}
end
end
defp headers() do
[
Accept: "application/json",
"Content-Type": "application/json",
Authorization: "Bearer #{System.get_env("OPENAI_API_KEY")}"
]
end
defp body(prompt, streaming?) do
%{
model: "text-davinci-003",
prompt: prompt,
stream: streaming?,
max_tokens: 1024
}
end
end
This logic more-or-less follows from the article; however, we need some special logic for handling data returned from the API. Each event from the API is prefixed with a data: <data> where <data> can be [DONE] or JSON data. During my experiments, I noticed that at times the stream receives multiple messages at once, so just calling Jason.decode! on the response will break. parse_chunk/2 splits responses on data and then handles both additional tokens and termination tokens.
Streaming Text in LiveView
The stream/1 function returns a stream from an HTTP request to the OpenAI API. We can use this stream in a LiveView to render the stream as the API returns tokens. Start by adding a new live route:
live "/answer", AnswerLive
In a new LiveView, add the following mount and render callbacks:
defmodule StreamingTextWeb.AnswerLive do
use StreamingTextWeb, :live_view
import StreamingTextWeb.CoreComponents
@impl true
def mount(_session, _params, socket) do
socket =
socket
|> assign(:question, "")
|> assign(:answer, "")
|> assign(:state, :waiting_for_question)
{:ok, socket}
end
@impl true
def render(assigns) do
~H"""
<div class="flex flex-col max-w-4xl min-h-screen items-center">
<h1 class="text-2xl">Ask Me Anything</h1>
<.simple_form
for={%{}}
as={:question}
:let={f}
phx-submit="answer_question"
class="w-full"
>
<.input
disabled={@state != :waiting_for_question}
field={{f, :question}}
value={@question}
type="text"
/>
<.button
type="submit"
disabled={@state != :waiting_for_question}
phx-disabled-with="Answering..."
>
Answer Question
</.button>
</.simple_form>
<div class="mt-4 text-md">
<p><span class="font-semibold">Question:</span> <%= @question %></p>
<p><span class="font-semibold">Answer:</span><%= @answer %></p>
</div>
</div>
"""
end
end
This will create a simple form which we can use to submit queries to GPT-3 and stream responses back. Next, add the following event handler to handle form submission:
@impl true
def handle_event("answer_question", %{"question" => %{"question" => question}}, socket) do
prompt = prompt(question)
stream = StreamingText.OpenAI.stream(prompt)
socket =
socket
|> assign(:question, question)
|> assign(:state, :answering_question)
|> assign(:response_task, stream_response(stream))
{:noreply, socket}
end
defp prompt(question) do
"""
Answer the following question.
Question: #{question}
Answer:
"""
end
This will build a prompt, create a response stream, change the current state and invoke a stream_response/1 function and assign the result to :response_task. Next, implement stream_response/1 like this:
defp stream_response(stream) do
target = self()
Task.Supervisor.async(StreamingText.TaskSupervisor, fn ->
for chunk <- stream, into: <<>> do
send(target, {:render_response_chunk, chunk})
chunk
end
end)
end
This will spin up a new task to consume the OpenAI stream and send chunks to the LiveView process. You’ll need to start your task supervisor to your supervision tree:
children = [
# Start the Telemetry supervisor
StreamingTextWeb.Telemetry,
# Start the Ecto repository
StreamingText.Repo,
# Start the PubSub system
{Phoenix.PubSub, name: StreamingText.PubSub},
# Start Finch
{Finch, name: StreamingText.Finch},
{Task.Supervisor, name: StreamingText.TaskSupervisor},
# Start the Endpoint (http/https)
StreamingTextWeb.Endpoint
# Start a worker by calling: StreamingText.Worker.start_link(arg)
# {StreamingText.Worker, arg}
]
Next, you need to handle the :render_response_chunk messages:
@impl true
def handle_info({:render_response_chunk, chunk}, socket) do
answer = socket.assigns.answer <> chunk
{:noreply, assign(socket, :answer, answer)}
end
This will receive the :render_response_chunk messages and append the next chunk to the current answer, which will then re-render in the browser. Finally, you need to await on the result of the task in a separate handler:
def handle_info({ref, answer}, socket) when socket.assigns.response_task.ref == ref do
socket =
socket
|> assign(:answer, answer)
|> assign(:state, :waiting_for_question)
{:noreply, socket}
end
def handle_info(_message, socket) do
{:noreply, socket}
end
And that’s all you need. Now you can open the browser and try it out:
Success!
Sean Moriarity 
Nice
LikeLike