Raxol.Agent.Stream (Raxol Agent v2.6.0)

Copy Markdown View Source

Stream-first API for agent sessions.

Wraps agent interactions as lazy Elixir Streams with natural backpressure. Each element is a typed event tuple. Compose with standard Stream/Enum functions.

Quick Start

# Stream text deltas from a prompt
Raxol.Agent.Stream.run("Analyze mix.exs", opts)
|> Raxol.Agent.Stream.text_deltas()
|> Enum.each(&IO.write/1)

# Collect final result
{:ok, result} =
  Raxol.Agent.Stream.run("What is 2+2?", opts)
  |> Raxol.Agent.Stream.collect()

# ReAct loop with tools
Raxol.Agent.Stream.react("Count lines in mix.exs", opts)
|> Stream.each(fn
  {:tool_use, %{name: name}} -> IO.puts("Calling #{name}...")
  {:text_delta, text} -> IO.write(text)
  _ -> :ok
end)
|> Stream.run()

Event Types

  • {:text_delta, text} -- streaming text chunk from LLM
  • {:tool_use, %{name, arguments, id}} -- LLM requesting a tool call
  • {:tool_result, %{name, result}} -- result from executing a tool
  • {:turn_complete, %{content, usage, iteration}} -- end of one ReAct turn
  • {:done, %{content, tool_results, usage}} -- final answer
  • {:error, reason} -- error during execution

Options

Common options for run/2 and react/2:

  • :executor -- Raxol.Agent.ExecutorConfig selecting backend + model + auth (takes precedence over :backend)
  • :backend -- AIBackend module (default: Raxol.Agent.Backend.Mock); ignored when :executor is given
  • :backend_opts -- keyword list passed to backend (api_key, model, etc.); merged over the executor's resolved opts
  • :model -- per-request model override (wins over :executor/:backend_opts)
  • :system_prompt -- system message prepended to conversation
  • :messages -- pre-built message list (overrides prompt)
  • :stream -- whether to use streaming backend (default: true)

Additional options for react/2:

  • :actions -- list of Action modules available as tools
  • :max_iterations -- loop guard (default: 10)

Summary

Types

done_info()
event()
tool_result()
tool_use()
turn_info()

Functions

collect(stream)

Collect all events and return the final content.

collect_text(stream)

Collect text from a stream into a single string.

react(prompt_or_messages, opts \\ [])

Stream a ReAct reasoning loop with tool use.

run(prompt_or_messages, opts \\ [])

Stream a single LLM completion.

text_deltas(stream)

Filter stream to only text delta events, unwrapping the text.

tool_results(stream)

Filter stream to only tool result events.

tool_uses(stream)

Filter stream to only tool use events.

Types

done_info()

@type done_info() :: %{
  content: String.t(),
  tool_results: [tool_result()],
  usage: map()
}

event()

@type event() ::
  {:text_delta, String.t()}
  | {:tool_use, tool_use()}
  | {:tool_result, tool_result()}
  | {:turn_complete, turn_info()}
  | {:done, done_info()}
  | {:error, term()}

tool_result()

@type tool_result() :: %{name: String.t(), result: map() | {:error, term()}}

tool_use()

@type tool_use() :: %{name: String.t(), arguments: map(), id: String.t() | nil}

turn_info()

@type turn_info() :: %{
  content: String.t(),
  usage: map(),
  iteration: non_neg_integer()
}

Functions

collect(stream)

@spec collect(Enumerable.t()) :: {:ok, done_info()} | {:error, term()}

Collect all events and return the final content.

Drains the stream and returns {:ok, done_info} or {:error, reason}.

collect_text(stream)

@spec collect_text(Enumerable.t()) :: String.t()

Collect text from a stream into a single string.

Joins all text deltas. If no text deltas were emitted, falls back to the final content from the :done event.

react(prompt_or_messages, opts \\ [])

@spec react(
  String.t() | [map()],
  keyword()
) :: Enumerable.t()

Stream a ReAct reasoning loop with tool use.

The LLM sees available tools (from :actions) and can call them. Each iteration emits tool_use/tool_result events. Continues until the LLM produces a final text answer or :max_iterations is reached.

Examples

Raxol.Agent.Stream.react("Analyze mix.exs", [
  backend: Backend.Mock,
  backend_opts: [response: "The file looks good."],
  actions: [MyAction],
  max_iterations: 5
])
|> Enum.to_list()

run(prompt_or_messages, opts \\ [])

@spec run(
  String.t() | [map()],
  keyword()
) :: Enumerable.t()

Stream a single LLM completion.

Returns a lazy Stream of events. If the backend supports streaming, you get {:text_delta, chunk} events followed by {:done, result}. Otherwise falls back to a single {:done, result}.

Examples

Raxol.Agent.Stream.run("Hello", backend: Backend.Mock, backend_opts: [response: "Hi"])
|> Enum.to_list()
#=> [{:text_delta, "Hi"}, {:done, %{content: "Hi", ...}}]

text_deltas(stream)

@spec text_deltas(Enumerable.t()) :: Enumerable.t()

Filter stream to only text delta events, unwrapping the text.

tool_results(stream)

@spec tool_results(Enumerable.t()) :: Enumerable.t()

Filter stream to only tool result events.

tool_uses(stream)

@spec tool_uses(Enumerable.t()) :: Enumerable.t()

Filter stream to only tool use events.