LangEx.Graph.Compiled (LangEx v0.11.3)

A compiled, executable graph.

Created by LangEx.Graph.compile/1. Use invoke/2 to run the graph with an initial state input.

Summary

Functions

Deletes every checkpoint for the thread in :config — e.g. when a conversation is closed or a user requests data removal.

get_state(compiled, opts)

Returns the latest checkpoint for a thread, or a specific one when :checkpoint_id is present in :config.

get_state_history(compiled, opts)

Returns the checkpoint history for a thread, most recent first.

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

Executes the compiled graph with the given input state.

update_state(graph, update, opts)

Applies an update to a thread's checkpointed state and saves it as a new checkpoint whose parent is the loaded one.

Types

t()

@type t() :: %LangEx.Graph.Compiled{
  checkpointer: module() | nil,
  conditional_edges: %{
    required(atom()) => {(map() -> atom() | String.t()), map() | nil}
  },
  edges: %{required(atom()) => [atom()]},
  initial_state: map(),
  interrupt_after: [atom()],
  interrupt_before: [atom()],
  name: atom() | String.t() | nil,
  node_opts: %{required(atom()) => keyword()},
  nodes: %{required(atom()) => (map() -> map()) | t()},
  reducers: LangEx.Graph.State.reducers(),
  store: {module(), keyword()} | nil
}

Functions

delete_thread(compiled, opts)

@spec delete_thread(
  t(),
  keyword()
) :: :ok | {:error, term()}

Deletes every checkpoint for the thread in :config — e.g. when a conversation is closed or a user requests data removal.

get_state(compiled, opts)

@spec get_state(
  t(),
  keyword()
) :: {:ok, LangEx.Checkpoint.t()} | :none | {:error, term()}

Returns the latest checkpoint for a thread, or a specific one when :checkpoint_id is present in :config.

Compiled.get_state(graph, config: [thread_id: "t-1"])
Compiled.get_state(graph, config: [thread_id: "t-1", checkpoint_id: "abc"])

get_state_history(compiled, opts)

@spec get_state_history(
  t(),
  keyword()
) :: [LangEx.Checkpoint.t()]

Returns the checkpoint history for a thread, most recent first.

Each checkpoint carries parent_id, so the full lineage (including forks created by update_state/3) can be reconstructed.

Options: :config (with :thread_id), :limit.

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

@spec invoke(t(), map() | LangEx.Command.t(), keyword()) ::
  {:ok, map()} | {:interrupt, term(), map()} | {:error, term()}

Executes the compiled graph with the given input state.

With a checkpointer and a :thread_id, invoking with an empty input (%{}) continues an unfinished run from the last checkpoint's pending nodes instead of restarting from :__start__ — this is how a crashed run is recovered. Non-empty input always starts a fresh pass from :__start__, merging the input into the latest checkpointed state.

Options:

:recursion_limit - max super-steps before raising (default: 25)
:config - keyword with :thread_id for checkpointing / resume
:context - runtime context passed to arity-2 node functions
:max_concurrency - cap on parallel node/Send tasks per super-step (default: System.schedulers_online())
:node_timeout - per-node timeout in ms for parallel super-steps (default: :infinity)
:deadline_ms - wall-clock budget for the whole run. Exposes a :remaining_ms managed value to nodes and flips :is_last_step to true once the deadline passes, so a node can conclude gracefully instead of the engine raising (default: no deadline)
:token_budget - cumulative token budget for the run. Exposes a :remaining_tokens managed value and flips :is_last_step once the budget is spent. Usage is read from the :llm_usage state key (the reducer convention ChatModel.merge_usage/2 populates) (default: none)
:durability - checkpoint write mode (default :sync):
- :sync - write after every super-step, on the hot path
- :async - write after every super-step in a supervised task (lower latency; a crash may lose the most recent step)
- :exit - skip per-step checkpoints; only interrupts persist (pause/resume works, crash recovery restarts from :__start__)

update_state(graph, update, opts)

@spec update_state(t(), map(), keyword()) ::
  {:ok, LangEx.Checkpoint.t()} | {:error, term()}

Applies an update to a thread's checkpointed state and saves it as a new checkpoint whose parent is the loaded one.

The update goes through the graph's reducers, exactly as a node result would. Loading a historical checkpoint via :checkpoint_id in :config forks the thread from that point. Returns the new checkpoint.