LangEx.Graph.Stream (LangEx v0.11.3)

Copy Markdown View Source

Streaming graph execution.

Returns an Elixir Stream that lazily yields events as the graph executes. Execution runs in a supervised, monitored task that sends events to the consumer via the mailbox. The stream blocks until the next event arrives — long-running nodes (e.g. LLM calls) never cause the stream to halt early. If the runner crashes, the crash is surfaced as a {:done, {:error, {:runner_exit, reason}}} event. Halting the stream early shuts the runner down.

Streaming accepts the same inputs as LangEx.invoke/3: a map to start a run, %{} to continue a crashed thread, or %LangEx.Command{resume: value} to resume an interrupted one.

Stream modes

Pass modes: [...] to select event granularity (default [:updates]):

  • :updates — per-node/per-step events (the event list below)
  • :values{:values, state} with the full state after each super-step
  • :messages{:message_delta, %{node: name, kind: :content, text: chunk}} token deltas forwarded from streaming LLM adapters via ChatModel
  • :custom{:custom, term} events emitted by nodes with emit/1

{:interrupt, value} and the final {:done, result} are always delivered regardless of mode.

:updates events

  • {:step_start, step, active_nodes} - a super-step begins
  • {:node_start, node_name} - a node is about to execute
  • {:node_end, node_name, update} - a node finished with this update
  • {:step_end, step, state} - a super-step completed

Summary

Functions

Emits a custom event into the enclosing run's event stream.

Returns a lazy stream of execution events from the compiled graph.

Functions

emit(event)

@spec emit(term()) :: :ok

Emits a custom event into the enclosing run's event stream.

Call from inside a node function; consumers see {:custom, event} when streaming with the :custom mode. A no-op when the graph is executed with invoke/3 instead of stream/3.

stream(graph, input, opts \\ [])

Returns a lazy stream of execution events from the compiled graph.

Accepts the same options as LangEx.invoke/3 plus :modes.