Streaming

Streaming lets your application show progress while an agent works: text, tool activity, state changes and your own events. It also changes ownership: the process consuming a live stream owns the run.

Complete Installation . Run these blocks in order in IEx.

Stream an answer

This real OpenAI call uses a local policy tool that emits a progress event:

defmodule MyApp.PolicySearch do
  use BeamWeaver.Tool

  name "policy_search"
  description "Read the refund timing policy."
  injected :runtime, :tool_runtime

  schema do
    field :query, :string
  end

  @impl true
  def invoke(_tool, %{runtime: runtime}, _opts) do
    BeamWeaver.Core.ToolRuntime.emit_output_delta(runtime, "Reading the refund policy")
    {:ok, "Refunds take five business days."}
  end
end

defmodule MyApp.StreamingAgent do
  use BeamWeaver.Agent

  model "openai:gpt-5.4-mini", stream: true, max_tokens: 256, timeout: 60_000
  system_prompt "Call policy_search for refund questions, then explain its result."
  tools [MyApp.PolicySearch]
end

alias BeamWeaver.Stream.Events
{:ok, stream} = MyApp.StreamingAgent.stream_events(%{messages: ["How long do refunds take?"]}, live: true)
events = Enum.to_list(stream)
false = Enum.any?(events, &match?(%{event: %Events.Error{}}, &1))

%{event: %{tool_call_id: call_id}} =
  Enum.find(events, &match?(%{event: %Events.ToolStart{tool_name: "policy_search"}}, &1))

true = Enum.any?(events, &match?(%{event: %Events.ToolFinish{tool_call_id: ^call_id}}, &1))
true = Enum.any?(events, &match?(%{event: %Events.ToolDelta{}}, &1))
text = for %{event: %Events.Token{text: text}} <- events, into: "", do: text
true = byte_size(text) > 0
IO.puts(text)

Use Enum.each/2 and IO.write/1 on token events to display text as it arrives. This example collects once so subsequent inspections do not repeat the run. Token boundaries vary by provider; a token event need not contain a whole word.

Live streams

Two options control different things: stream: true requests incremental model output; live: true returns an enumerable that runs while you consume it. Without model streaming, the agent can still report tools and state changes, but model text arrives as a whole reply.

A live stream is lazy. Each enumeration starts a run. Enumerating it twice can make two sets of API calls and repeat tool side effects. If you need several views, materialize it once or dispatch events from a single consumer. Stopping enumeration early cancels the run; so does the consumer process exiting. Cancellation cannot undo an external side effect already performed.

Without live: true, the call runs to completion and returns a collected list. An interrupted collected run returns {:interrupted, interrupt} with interrupt.events; failure returns {:error, error}.

Events

Each item is a BeamWeaver.Stream.Envelope with an event, run_id, node, namespace and step. Use identifiers to correlate concurrent work.

BeamWeaver.Stream.event_mode/1 classifies events. stream_mode: limits emission to channels such as :messages, :tools, :updates, :values, :custom, :tasks, :checkpoints, :lifecycle and :debug. The default :events includes all channels. Terminal failures remain visible when filtering. Restrict channels when you do not need full state snapshots.

Filter the already-collected events from the first example without starting another run:

tool_events = Enum.filter(events, &(BeamWeaver.Stream.event_mode(&1.event) == :tools))
true = Enum.any?(tool_events, &match?(%{event: %Events.ToolStart{}}, &1))
true = Enum.any?(tool_events, &match?(%{event: %Events.ToolFinish{}}, &1))
IO.inspect(Enum.map(tool_events, & &1.event.__struct__), label: "Tool event types")

The final answer

Use tokens for provisional UI text and completed messages as the authoritative copy. Do not append a complete message after already displaying its tokens.

answers =
  for %{event: %Events.Message{message: %{role: :assistant} = message}} <- events,
      message.tool_calls in [nil, []],
      do: BeamWeaver.Core.Message.text(message)

answer = List.last(answers)
true = is_binary(answer) and byte_size(answer) > 0

End of enumeration means the stream ended; inspect terminal errors and interrupts before treating that as successful completion. A Done event belongs to a model call, not necessarily the whole agent run.

Errors

A live run can fail after stream_events/2 returned successfully. Handle Events.Error inside the consumer. Invalid setup can instead return an error before a stream is created.

A failed model attempt emits a debug payload with type: :model_call_failed. Retry or fallback middleware can then start another attempt. Already emitted tokens from the failed attempt are provisional and must not be presented as part of the final answer. Clear or replace that attempt's UI text. A tool failure is different: ToolError normally goes back to the model for recovery.

Built-in Middleware describes retry policies. Testing shows how to reproduce failure paths deterministically.

Tool progress and custom events

The tool above uses BeamWeaver.Core.ToolRuntime.emit_output_delta/2. For arbitrary progress payloads, a graph node's second argument is a runtime whose stream_writer emits custom events:

defmodule MyApp.ProgressGraph do
  use BeamWeaver.Graph

  graph do
    node :classify, &__MODULE__.classify/2
    edge start(), :classify
    edge :classify, finish()
  end

  def classify(_state, runtime) do
    runtime.stream_writer.(%{stage: "classifying"})
    %{category: "billing"}
  end
end

{:ok, stream} = MyApp.ProgressGraph.stream_events(%{}, live: true)
progress = for %{event: %Events.Custom{payload: payload}} <- stream, do: payload
[%{stage: "classifying"}] = progress

Tools can inject :runtime for the same writer. Middleware state hooks receive it directly; wrapper hooks read request.runtime.

Tool-call arguments

MessageChunk can contain partial JSON arguments in tool_call_chunks. They are display data until complete; never execute an action from a partial argument string. Agents assemble the call for you. Direct model consumers can use BeamWeaver.Stream.Finalize.finalize/1 to merge chunks into a message. Use call IDs and indices to keep simultaneous calls separate.

Tool arguments can span arbitrary chunk boundaries. Assemble them before parsing/executing:

alias BeamWeaver.Core.Messages

chunks = [
  Messages.ai_chunk("",
    tool_call_chunks: [Messages.tool_call_chunk(id: "call_1", index: 0, name: "lookup_order", args: ~s({"order_id))]
  ),
  Messages.ai_chunk("", tool_call_chunks: [Messages.tool_call_chunk(id: "call_1", index: 0, args: ~s(": "A-1001"}))])
]

[call] = BeamWeaver.Stream.Finalize.finalize(chunks).tool_calls
{call.name, call.args}
#=> {"lookup_order", %{"order_id" => "A-1001"}}

When the run pauses

A live interrupt arrives as a debug payload with type: :interrupt and an interrupt value, then the stream ends. Display its review request and resume using the original thread/checkpointer plus fresh caller context. When provided, interrupt.response_schema describes the expected form data. See Human Review .

Redact a stream

A sensitive value can span tokens. Replacing text inside each token separately can leak it. BeamWeaver.Agent.Middleware.PII.stream_transform/1 buffers text until it can apply the configured policy and also transforms message chunks, whole messages, tool data and state snapshots.

Apply the transform before events leave your server. PII middleware on saved messages alone does not protect already-emitted tokens. Guardrails explains the distinction and redaction strategies.

The transform buffers text across chunk boundaries. Supply a synthetic email in two chunks to see that boundary handled:

alias BeamWeaver.Stream.{Envelope, Events}
{transform, pending} = BeamWeaver.Agent.Middleware.PII.stream_transform(type: :email, strategy: :redact)

source = [
  %Envelope{event: %Events.Token{text: "Contact ada@"}},
  %Envelope{event: %Events.Token{text: "example.com for help."}},
  %Envelope{event: %Events.Done{}}
]

safe =
  source
  |> Stream.transform(pending, fn envelope, acc ->
    {{:ok, emitted}, acc} = transform.(envelope, acc)
    {List.wrap(emitted), acc}
  end)
  |> Enum.to_list()

text = for %{event: %Events.Token{text: part}} <- safe, into: "", do: part
false = text =~ "[email protected]"
true = text =~ "[REDACTED_EMAIL]"

Subgraphs

Subgraph events retain the path of node names in namespace. Parent events use []. A subgraph failure is first a node failure; retry or an error handler may recover it. Only an unrecovered top-level failure ends with Error.

BeamWeaver.Stream.Subgraphs.from_events/2 summarizes paths and statuses. BeamWeaver.Stream.Lifecycle.from_events/2 produces lifecycle events from those summaries. Keep Task events when using either helper.

Push tokens to a LiveView

Consume the stream in a supervised task rather than in the LiveView callback. Send only UI events to the LiveView: provisional text, completed messages, review requests and failures. Correlate messages with a run ID so events from an older request cannot overwrite a newer answer.

Tie task cancellation explicitly to your application's lifecycle. A normal process exit does not automatically terminate every linked process. Cancel work when the view disconnects if it belongs to that view; use a persistent job owner when it must continue. Keep final state in a checkpointer rather than only in a socket assignment.

This complete bridge uses a supervised task and monitors the receiving process. A LiveView can handle the tagged messages in handle_info/2; on disconnect, the monitor stops the worker. Only one process enumerates the stream:

defmodule MyApp.ChatStream do
  def start(supervisor, receiver, question) do
    owner = make_ref()

    {:ok, worker} =
      Task.Supervisor.start_child(supervisor, fn ->
        receiver_ref = Process.monitor(receiver)

        task =
          Task.Supervisor.async_nolink(supervisor, fn ->
            case MyApp.StreamingAgent.stream_events(%{messages: [question]}, live: true) do
              {:ok, stream} ->
                Enum.each(stream, &send(receiver, {:agent_event, owner, &1}))
                :stream_ended

              other ->
                other
            end
          end)

        receive do
          {ref, result} when ref == task.ref ->
            Process.demonitor(ref, [:flush])
            Process.demonitor(receiver_ref, [:flush])
            send(receiver, {:agent_stream_finished, owner, result})

          {:DOWN, ^receiver_ref, :process, ^receiver, _} ->
            Task.shutdown(task, :brutal_kill)

          {:DOWN, ref, :process, _pid, reason} when ref == task.ref ->
            send(receiver, {:agent_stream_finished, owner, {:error, reason}})
        end
      end)

    {:ok, worker, owner}
  end
end

{:ok, supervisor} = Task.Supervisor.start_link()

try do
  {:ok, _worker, run_ref} = MyApp.ChatStream.start(supervisor, self(), "How long do refunds take?")

  receive do
    {:agent_stream_finished, ^run_ref, :stream_ended} -> :ok
    {:agent_stream_finished, ^run_ref, other} -> raise inspect(other)
  after
    120_000 -> raise "Stream did not finish"
  end

  receive do
    {:agent_event, ^run_ref, %{event: %BeamWeaver.Stream.Events.ToolFinish{}}} -> :ok
  after
    0 -> raise "No tool completion was delivered"
  end
after
  Supervisor.stop(supervisor)
end

Custom producers and backpressure

BeamWeaver.Stream.Mux combines caller-owned producers through a bounded queue. A Stream.Sink lets a producer emit and check cancellation; child sinks attach nested namespaces. Stream.Policy chooses blocking, drop-newest, drop-oldest or error behavior on overflow. Size the buffer and emission timeout for the consumer; dropping events can make an otherwise complete run impossible to reconstruct from its stream.

Heartbeats are opt-in progress signals, not evidence that a model/tool finished. Read-side MessagesTransformer, Transformers and lifecycle/subgraph helpers project collected events without starting work. A synthesized message from partial deltas is not a substitute for a completed provider outcome.

Producer forms select the callback argument:

max_buffer: defaults to 256 items, emit_timeout: to 5,000 ms, and the whole-stream timeout: to infinity. Choose explicit budgets for application work. Two producers can share a bounded stream while retaining separate namespaces. All events below are consumed once:

alias BeamWeaver.Stream.{Mux, Sink, Events}

producers =
  for name <- ["orders", "policy"] do
    {:sink, name,
     fn sink ->
       child = Sink.child(sink, namespace: [name])
       :ok = Sink.emit(child, %Events.Custom{payload: %{source: name, status: "ready"}})
     end}
  end

items = Mux.stream(producers, max_buffer: 4, overflow: :block, timeout: 5_000) |> Enum.to_list()

ready =
  for %{event: %Events.Custom{payload: %{source: source}}, namespace: namespace} <- items,
      do: {source, namespace}

true = {"orders", ["orders"]} in ready
true = {"policy", ["policy"]} in ready
false = Enum.any?(items, &match?(%{event: %Events.Error{}}, &1))