The LiveView cannot wait

Streaming changes less than it seems: the chain is still the return value, and the pieces arrive on the side, as a side effect. Two lines of configuration:

llm = ChatAnthropic.new!(%{model: "claude-sonnet-5", stream: true})

chain =
  %{llm: llm}
  |> LLMChain.new!()
  |> LLMChain.add_callback(%{
    on_llm_new_delta: fn _chain, deltas -> send(view_pid, {:deltas, deltas}) end,
    on_message_processed: fn _chain, message -> send(view_pid, {:done, message}) end
  })

Each %MessageDelta{} is a chunk of the answer, with content and a status that reads :incomplete until the last one, which comes in :complete. The library accumulates them for you: by the time run/2 returns, the chain already holds the whole assembled message. The deltas exist for the screen, and the logic can ignore them.

Why not inside handle_event

The obvious way to write this is to run the chain right there in the LiveView's handle_event. It doesn't work.

The LiveView is a process, and while it sits inside run/2 waiting on the model, it processes nothing else, not even its own delta sends, which pile up in the mailbox until the call returns. The user stares at a frozen screen for fifteen seconds and then gets the whole answer at once.

The call has to happen in another process:

def handle_event("send", %{"text" => text}, socket) do
  view = self()

  Task.Supervisor.start_child(MyApp.TaskSupervisor, fn ->
    MyApp.Conversation.ask(socket.assigns.conversation_id, text, view)
  end)

  {:noreply, assign(socket, streaming: true, buffer: "")}
end

The ask/2 from the first part gained a third argument, the pid that will receive the deltas, and that is what the GenServer uses to build the add_callback before running the chain. Who is listening changes with every tab opened; the conversation itself does not.

Binding view = self() outside the function is mandatory, because inside the block self() would already be the Task. And start_child/2 rather than Task.async/1 because there is no result to await here: the answer comes through the deltas, and a supervised Task that dies does not take the LiveView with it.

On the LiveView side, two handle_info clauses cover it:

def handle_info({:deltas, deltas}, socket) do
  text = Enum.map_join(deltas, "", &delta_text/1)
  {:noreply, assign(socket, buffer: socket.assigns.buffer <> text)}
end

def handle_info({:done, message}, socket) do
  {:noreply,
   socket
   |> assign(streaming: false, buffer: "")
   |> stream_insert(:messages, message)}
end

# Content arrives as a string or as a list of ContentPart, depending on the
# model and on whether the answer carries reasoning parts.
defp delta_text(%{content: content}) when is_binary(content), do: content
defp delta_text(%{content: parts}) when is_list(parts), do: Enum.map_join(parts, "", & &1.content)
defp delta_text(_), do: ""

The buffer is a plain string in assigns because it is ephemeral content, rewritten many times per second. The finished message goes to a stream, which is where the ever-growing list belongs.

When more than one screen is watching

Sending straight to a pid works while there is exactly one watcher. With two open tabs, or a human agent following the customer's conversation, the way forward is PubSub:

# in the tool, or wherever the conversation runs
Phoenix.PubSub.broadcast(MyApp.PubSub, "conversation:#{id}", {:deltas, deltas})

# in the LiveView's mount
if connected?(socket), do: Phoenix.PubSub.subscribe(MyApp.PubSub, "conversation:#{id}")

The handle_info clauses stay identical, only the delivery changes. Since the swap is cheap, it is worth starting with send and migrating when the second screen shows up.

The whole shape

Putting the three parts together:

LiveView (one per tab)
   │  handle_event
   ▼
Supervised Task ──► Conversation GenServer ──► LLMChain.run/2
                          (the state)             │
   ◄────────── deltas ───────────────────────────-┘
   PubSub or send

The GenServer holds the conversation, the Task absorbs the wait, the LiveView paints. None of the three knows how the others work inside, and any one of them can break without taking the others down.

LangChain is only here for the message formats and the tool loop. The shape around it is plain OTP, and it works the same for any slow external call: a payment that drags, a large upload, a report that takes minutes to come out.