LangEx.Graph.Pregel (LangEx v0.11.3)

Copy Markdown View Source

Super-step execution engine inspired by Google's Pregel.

Processes the graph in discrete super-steps: resolve which nodes to run next, execute them, apply state updates via reducers, then repeat until reaching the END node or exhausting the recursion limit.

Supports checkpointing, interrupts (dynamic and static breakpoints), streaming events, runtime context, Send fan-out, and managed values (remaining_steps, is_last_step, and the remaining_ms / remaining_tokens run budgets).

Interrupt model

Every LangEx.Interrupt.interrupt/1 call gets a stable ID derived from the node name and the call order within the node ("node:0", "node:1"). Resume values are keyed by interrupt ID, so a node can interrupt multiple times across resume cycles, and several nodes interrupting in the same parallel super-step each keep their own pending entry. Completed sibling results in a parallel super-step are merged into state before pausing, and their resolved next targets are recorded in the checkpoint — nothing is lost when one branch interrupts.

Error model

Node exceptions (after the node's retry policy, if any, is exhausted) surface as {:error, %LangEx.NodeError{}} rather than raising out of the run. Programmer errors — routing to an undefined node, a missing conditional mapping — still raise.

Summary

Types

entry()
interrupt()
run_opts()

Functions

run(graph, state, limit)

Runs the compiled graph from the start node through to completion.

Types

entry()

@type entry() :: atom() | LangEx.Send.t()

interrupt()

@type interrupt() :: %{id: String.t(), value: term(), node: atom()}

run_opts()

@type run_opts() :: %{
  optional(:start_nodes) => [entry()] | nil,
  optional(:parent_id) => String.t() | nil,
  optional(:bypass_breakpoints) => boolean(),
  optional(:resume_values) => %{required(String.t()) => term()},
  optional(:raw_interrupts) => boolean(),
  optional(:max_concurrency) => pos_integer() | nil,
  optional(:node_timeout) => timeout() | nil,
  optional(:store) => {module(), keyword()} | nil,
  optional(:deferred_backlog) => [entry()],
  optional(:completed_next) => [entry()],
  optional(:durability) => :sync | :async | :exit,
  optional(:deadline) => integer() | nil,
  optional(:token_budget) => pos_integer() | nil,
  recursion_limit: pos_integer(),
  checkpointer: module() | nil,
  config: keyword(),
  context: term(),
  resume: %{nodes: [entry()], values: %{required(String.t()) => term()}} | nil,
  step: non_neg_integer(),
  emit_to: pid() | nil
}

Functions

run(graph, state, limit)

@spec run(LangEx.Graph.Compiled.t(), map(), run_opts() | pos_integer()) ::
  {:ok, map()} | {:interrupt, term(), map()} | {:error, term()}

Runs the compiled graph from the start node through to completion.