Graph DSL

Use a graph when your application decides the workflow: fixed steps, conditional branches, parallel checks or a review loop. Use an agent when the model should choose its own actions. A graph can contain model calls, agents and other graphs.

This first example is ordinary Elixir and needs no provider credentials. Run the blocks in order in IEx after installing BeamWeaver.

Build and run a graph

defmodule MyApp.TicketRouter do
  use BeamWeaver.Graph

  graph do
    node :classify, &__MODULE__.classify/1
    node :billing, &__MODULE__.billing/1
    node :shipping, &__MODULE__.shipping/1

    edge start(), :classify
    edge :classify, :billing, when: %{category: :billing}
    edge :classify, :shipping, default: true
    edge :billing, finish()
    edge :shipping, finish()
  end

  def classify(%{subject: subject}),
    do: %{category: if(subject =~ ~r/refund/i, do: :billing, else: :shipping)}

  def billing(_state), do: %{queue: "billing"}
  def shipping(_state), do: %{queue: "shipping"}
end

{:ok, state} = MyApp.TicketRouter.invoke(%{subject: "Refund for order A-1001"})
"billing" = state.queue
"Refund for order A-1001" = state.subject

graph do describes the topology; node functions do the work. start() and finish() are boundary markers, not application nodes. The module supplies invoke/2, stream_events/2, resume/2, get_state/1 and compiled_graph/1. Definitions are built/compiled with each call's options. Storage must be owned outside the invocation, just as for agents .

Nodes

A node receives state and returns only changed keys. Unchanged keys remain. A two-argument node also receives BeamWeaver.Graph.Runtime with context, configuration, execution metadata and a custom-event writer.

A Core.Error propagates as an error; other reasons normalize to :node_error. Raised exceptions become :node_exception. Return an error rather than inventing success when an external dependency fails.

Node options

Unknown options fail compilation. Durable Execution explains retries and timeouts; Caching covers cache keys.

A node can combine input state with trusted runtime context:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled

greet = fn state, runtime ->
  %{greeting: "Hi #{runtime.context.customer}, about order #{state.order_id}:"}
end

greeter =
  Graph.new()
  |> Graph.add_node(:greet, greet)
  |> Graph.add_edge(Graph.start(), :greet)
  |> Graph.add_edge(:greet, Graph.end_node())
  |> Graph.compile!()

{:ok, state} = Compiled.invoke(greeter, %{order_id: "A-1001"}, context: %{customer: "Ada"})
state.greeting
#=> "Hi Ada, about order A-1001:"

Return a tagged error when the action fails:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled

lookup = fn _state -> {:error, "order service unavailable"} end

failing =
  Graph.new()
  |> Graph.add_node(:lookup, lookup)
  |> Graph.add_edge(Graph.start(), :lookup)
  |> Graph.add_edge(:lookup, Graph.end_node())
  |> Graph.compile!()

Compiled.invoke(failing, %{})
#=> {:error, %BeamWeaver.Core.Error{type: :node_error, message: "order service unavailable"}}

Edges and routing

An edge runs a destination after its source. when: normally examines the source update, not the entire accumulated state. Use a two-argument predicate to inspect both update and state. Map guards support literal values, allowed-value lists, ranges and predicates.

All matching guarded edges run; this can create parallel branches. A default edge runs only if none match. Without a match or default, that path ends. An exception in a predicate becomes :edge_exception.

Loops

Back-edges create loops. max_runs: bounds traversal of a guarded edge and then lets a default route handle the exhausted case. Counts belong to one run and survive continuation; a new run resets them. Always distinguish "accepted" from "revision budget exhausted" in final state.

This workflow routes a large amount to review:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled

refunds =
  Graph.new()
  |> Graph.add_node(:check_refund, fn %{amount: amount} -> %{large: amount > 100} end)
  |> Graph.add_node(:auto_approve, fn _state -> %{decision: :approved} end)
  |> Graph.add_node(:manager_review, fn _state -> %{decision: :needs_manager} end)
  |> Graph.add_edge(Graph.start(), :check_refund)
  |> Graph.add_edge(:check_refund, :manager_review, when: %{large: true})
  |> Graph.add_edge(:check_refund, :auto_approve, default: true)
  |> Graph.add_edge(:auto_approve, Graph.end_node())
  |> Graph.add_edge(:manager_review, Graph.end_node())
  |> Graph.compile!()

{:ok, state} = Compiled.invoke(refunds, %{amount: 250})
state.decision
#=> :needs_manager

This local reviewer always requests revision, so the bounded loop reaches escalation:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled

drafting =
  Graph.new()
  |> Graph.add_reducer(:drafts, &+/2)
  |> Graph.add_node(:draft, fn _state -> %{drafts: 1} end)
  |> Graph.add_node(:review, fn _state -> %{verdict: :revise} end)
  |> Graph.add_node(:escalate, fn _state -> %{escalated: true} end)
  |> Graph.add_edge(Graph.start(), :draft)
  |> Graph.add_edge(:draft, :review)
  |> Graph.add_edge(:review, :draft, when: %{verdict: :revise}, max_runs: 2)
  |> Graph.add_edge(:review, :escalate, default: true)
  |> Graph.add_edge(:escalate, Graph.end_node())

Compiled.invoke(Graph.compile!(drafting), %{drafts: 0})
#=> {:ok, %{drafts: 3, verdict: :revise, escalated: true}}

The same saved thread retains accumulated state; the next invocation gets a new traversal budget:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled

drafting = Graph.compile!(drafting, checkpointer: BeamWeaver.Checkpoint.ETS.new())
thread = [config: %{"configurable" => %{"thread_id" => "ticket-42"}}]

{:ok, %{drafts: 3}} = Compiled.invoke(drafting, %{drafts: 0}, thread)
Compiled.invoke(drafting, %{drafts: 0}, thread)
#=> {:ok, %{drafts: 6, verdict: :revise, escalated: true}}

Parallel branches and joins

Branches execute in the same superstep and merge their updates after finishing. Use reducers when several branches write the same key. A join waits for all listed branches, even when they take different numbers of steps:

defmodule MyApp.OrderChecks do
  use BeamWeaver.Graph

  graph do
    state do
      channel :checks, merge: :map
    end

    node :order, fn _state -> %{checks: %{order_exists: true}} end
    node :policy, fn %{amount: amount} -> %{checks: %{within_limit: amount <= 100}} end
    node :decide, fn %{checks: checks} -> %{eligible: checks.order_exists and checks.within_limit} end

    edge start(), :order
    edge start(), :policy
    join [:order, :policy], :decide
    edge :decide, finish()
  end
end

{:ok, checked} = MyApp.OrderChecks.invoke(%{amount: 40})
%{order_exists: true, within_limit: true} = checked.checks
true = checked.eligible

These are local example checks, not a payment policy. deps: on a node is an alternative way to declare its join. Join arrivals are checkpointed so completed branches still count after another branch pauses or fails.

Update and route with Command

%Graph.Command{update: changes, goto: destination} combines data and control flow. goto can be a node, a list, sends, or the end marker. It replaces the returning node's ordinary outgoing edges. Declare possible targets in destinations: for static validation. Use a command when destination selection belongs inside the node; use guarded edges when it belongs to the topology.

This node returns both the category and the next destination:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled

alias BeamWeaver.Graph.Command

route = fn %{subject: subject} ->
  category = if subject =~ ~r/refund/i, do: :billing, else: :shipping
  %Command{update: %{category: category}, goto: category}
end

router =
  Graph.new()
  |> Graph.add_node(:classify, route, destinations: [:billing, :shipping])
  |> Graph.add_node(:billing, fn _ -> %{reply: "Our billing team is checking your refund."} end)
  |> Graph.add_node(:shipping, fn _ -> %{reply: "Our shipping team is checking your order."} end)
  |> Graph.add_edge(Graph.start(), :classify)
  |> Graph.add_edge(:billing, Graph.end_node())
  |> Graph.add_edge(:shipping, Graph.end_node())
  |> Graph.compile!()

{:ok, state} = Compiled.invoke(router, %{subject: "Where is order A-1001?"})
state.reply
#=> "Our shipping team is checking your order."

Map-reduce with Send

Use Send when the number of tasks is known only at runtime:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.{Compiled, Send}

batch =
  Graph.new()
  |> Graph.add_reducer(:results, &Kernel.++/2)
  |> Graph.add_node(
    :dispatch,
    fn %{values: values} ->
      Enum.map(values, &%Send{node: :double, update: %{value: &1}})
    end,
    destinations: [:double]
  )
  |> Graph.add_node(:double, fn %{value: value} -> %{results: [value * 2]} end)
  |> Graph.add_node(:total, fn %{results: results} -> %{total: Enum.sum(results)} end)
  |> Graph.add_edge(Graph.start(), :dispatch)
  |> Graph.add_edge(:double, :total)
  |> Graph.add_edge(:total, Graph.end_node())
  |> Graph.compile!()

{:ok, reduced} = Compiled.invoke(batch, %{values: [2, 3, 5], results: []})
[4, 6, 10] = Enum.sort(reduced.results)
20 = reduced.total

Each send runs the worker with state plus its update. Worker results merge through reducers; the following node runs after that batch. Do not depend on completion order when order is not part of the task contract.

Static graphs as modules

Inside graph do, use node, edge, join, reducer and state do. A state channel can use merge: :last, :list, :map or a reducer function, plus visibility options. Graph State explains channels.

Use the builder (Graph.new, add_node, add_edge, add_join) for topology computed from data. A graph module can return that builder from graph/0 instead of declaring nodes. Choose one style per module. A compiled value runs through BeamWeaver.Graph.Compiled; a module runs through its generated functions.

Call a model or an agent from a node

A node can call ChatModel.invoke/3 and return the reply as an update. Model initialization creates a value; network work happens at invocation. Set a node timeout that accommodates the provider timeout and retries.

Adding an agent module as the node makes it a subgraph with integrated checkpoints, interrupts and events. Calling Agent.invoke inside a plain function creates a separate run. See Subgraphs and the real-model Workflow Patterns .

Batch independent runs

Compiled.batch/3 returns sequential outcomes in input order. batch_as_completed/3 returns a lazy concurrent stream with input indices; set max concurrency and task timeouts. Keep independent thread IDs for concurrent runs. bulk_update_state/4 is a different operation: it applies checkpointed supersteps of state edits without executing business nodes.

For independent inputs, consume the indexed completion stream and correlate results by index:

{:ok, router} = MyApp.TicketRouter.compiled_graph([])
inputs = [%{subject: "Refund A-1001"}, %{subject: "Delivery A-1002"}]
[{:ok, %{queue: "billing"}}, {:ok, %{queue: "shipping"}}] = BeamWeaver.Graph.Compiled.batch(router, inputs)
{:ok, completions} = BeamWeaver.Graph.Compiled.batch_as_completed(router, inputs, max_concurrency: 2)
completed = Enum.to_list(completions)
true = length(completed) == 2
IO.inspect(completed, label: "Results with input indices")

Stream a run

stream_events/2 reports typed events; live: true makes it lazy. Consume once, handle failures and correlate subgraph namespaces. Nodes emit custom progress through runtime.stream_writer. Streaming covers lifecycle and cancellation.

Interrupts

Call Graph.interrupt/2 inside a node or configure static before/after breakpoints. Both need a checkpointer for continuation. Dynamic interrupt nodes rerun from the beginning on resume, so earlier side effects can repeat. See Human Review .

Checkpoints

Graph modules accept storage and thread configuration per invocation. Compiled graphs take the saver at compilation and thread configuration per call. A checkpoint preserves values and pending execution, not arbitrary process state. Checkpoints explains inspection and replay.

Recursion limit

A superstep runs all currently ready nodes. The default graph budget is 25 steps; reaching it returns :recursion_limit. Set a per-call or declared budget. RemainingSteps and IsLastStep managed values let nodes choose a controlled exit before exhaustion.

An unbounded polling edge is stopped by the run budget:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled

poller =
  Graph.new()
  |> Graph.add_node(:poll, fn %{polls: polls} -> %{polls: polls + 1} end)
  |> Graph.add_edge(Graph.start(), :poll)
  |> Graph.add_edge(:poll, :poll)
  |> Graph.compile!()

Compiled.invoke(poller, %{polls: 0}, recursion_limit: 5)
#=> {:error, %BeamWeaver.Core.Error{type: :recursion_limit, details: %{limit: 5}}}

Validate a graph

Graph.compile/2 returns a result tuple; compile!/2 raises. validation_report/2 reports all detected problems. With validate_static: true, validation also checks reachability and paths to an end. Dynamic routing is visible only when destinations are declared.

Compiled.get_graph/2 returns the topology as data: nodes, edges, branches, and the sorted entry_points and finish_points, the nodes that edges from the start lead to and the nodes with an edge to the end. draw_mermaid/2 and draw_ascii/2 render it.

%{diagnostics: []} = Graph.validation_report(MyApp.TicketRouter.graph(), validate_static: true)
{:ok, compiled} = MyApp.TicketRouter.compiled_graph([])
topology = Compiled.get_graph(compiled)
["classify"] = topology.entry_points
["billing", "shipping"] = topology.finish_points
diagram = Compiled.draw_mermaid(compiled)
true = is_binary(diagram) and byte_size(diagram) > 0

Static validity does not prove external services or node business rules work. Exercise representative inputs and failure paths in your application tests.