LangEx

Graph-based agent orchestration for Elixir. Build stateful, multi-step LLM workflows using nodes, edges, and conditional routing -- with the concurrency, fault tolerance, and streaming you get for free on the BEAM.

tools = [
  %LangEx.Tool{
    name: "get_weather",
    description: "Get current weather for a city",
    parameters: %{type: "object", properties: %{city: %{type: "string"}}, required: ["city"]},
    function: fn %{"city" => city} -> "#{city}: 22°C, sunny" end
  }
]

graph =
  Graph.new(LangEx.MessagesState.schema())
  |> Graph.add_node(:agent, LangEx.LLM.ChatModel.node(model: "claude-opus-4-20250514", tools: tools))
  |> Graph.add_node(:tools, LangEx.Tool.Node.node(tools))
  |> Graph.add_edge(:__start__, :agent)
  |> Graph.add_conditional_edges(:agent, &LangEx.Tool.Node.tools_condition/1, %{
    tools: :tools,
    __end__: :__end__
  })
  |> Graph.add_edge(:tools, :agent)
  |> Graph.compile()

{:ok, result} = LangEx.invoke(graph, %{messages: [Message.human("Weather in Tokyo?")]})

Define a graph. Add nodes for LLM calls and tool execution. Wire them with edges and conditions. Compile. Invoke. The LLM decides when to call tools and when to respond -- LangEx orchestrates the loop.

Why LangEx?

Python has LangGraph. Elixir deserves the same power, built on primitives that actually make sense for long-running, stateful agent workflows:

• Parallel node execution -- tool calls and graph nodes run concurrently via Task.Supervisor, not thread pools or async/await hacks
• Lightweight state machines -- graph state lives in function arguments and checkpoints, not GenServers; thousands of agent threads cost nothing
• Interrupt and resume -- pause execution for human approval, persist state to Redis or Postgres, resume hours later from exactly where you left off
• Streaming for free -- execution events are a lazy Elixir Stream; pipe them to Phoenix channels, LiveView, or Server-Sent Events
• Fault tolerance -- BEAM supervisors and process isolation mean one failing agent doesn't take down the rest

Installation

def deps do
  [
    {:lang_ex, "~> 0.8.0"},

    # Optional: for Redis checkpointing
    {:redix, "~> 1.5"},

    # Optional: for PostgreSQL checkpointing
    {:postgrex, "~> 0.19"},
    {:ecto_sql, "~> 3.12"}
  ]
end

The core library has zero infrastructure dependencies. Add a checkpointer only if you need pause/resume or durability.

Quick Start

A minimal graph that routes messages by intent:

alias LangEx.Graph
alias LangEx.Message

graph =
  Graph.new(messages: {[], &Message.add_messages/2}, intent: nil)
  |> Graph.add_node(:classify, fn state ->
    content = List.last(state.messages).content
    intent = if String.contains?(content, "weather"), do: "weather", else: "greeting"
    %{intent: intent}
  end)
  |> Graph.add_node(:weather, fn _state -> %{messages: [Message.ai("It's sunny today!")]} end)
  |> Graph.add_node(:greet, fn _state -> %{messages: [Message.ai("Hello there!")]} end)
  |> Graph.add_edge(:__start__, :classify)
  |> Graph.add_conditional_edges(:classify, &Map.get(&1, :intent), %{
    "weather" => :weather,
    "greeting" => :greet
  })
  |> Graph.add_edge(:weather, :__end__)
  |> Graph.add_edge(:greet, :__end__)
  |> Graph.compile()

{:ok, result} = LangEx.invoke(graph, %{messages: [Message.human("What's the weather?")]})

How it works:Graph.new defines the state schema (with optional reducers per key). Nodes are functions that receive state and return updates. Edges wire nodes together. Conditional edges route dynamically based on state. compile/1 validates and freezes the graph. invoke/2 runs it.

Configuration

API keys are resolved in order: explicit opts, Application config, environment variables.

# Environment variables (recommended)
# export ANTHROPIC_API_KEY=sk-ant-...

# Or application config
config :lang_ex, :anthropic, api_key: "sk-ant-..."

Model strings are auto-resolved to providers -- "claude-opus-4-20250514" routes to Anthropic, "gemini-2.0-flash" to Gemini, "gpt-4o" to OpenAI. Register custom providers at runtime:

LangEx.LLM.Registry.register_provider(:groq, MyApp.LLM.Groq)
LangEx.LLM.Registry.register_prefix("llama-", :groq)

Features

Checkpointing

Persist graph state after each step for pause/resume, fault recovery, and time-travel debugging.

graph = Graph.new(...) |> ... |> Graph.compile(checkpointer: LangEx.Checkpointer.Redis)

{:ok, result} = LangEx.invoke(graph, input, config: [thread_id: "my-thread"])

For PostgreSQL, generate a migration and call LangEx.Migration.up():

defmodule MyApp.Repo.Migrations.AddLangEx do
  use Ecto.Migration

  def up, do: LangEx.Migration.up()
  def down, do: LangEx.Migration.down()
end

Checkpoint write timing is controlled by the :durability invoke option:

Under every mode, checkpoints preserve full work entries — including pending %LangEx.Send{} payloads — so crash-continue and interrupt-resume pick up exactly where the run stopped (checkpoint format v2).

Human-in-the-Loop Interrupts

Pause execution at any node, surface a payload to the caller, and resume with a human-provided value. Requires a checkpointer.

graph =
  Graph.new(value: 0, approved: false)
  |> Graph.add_node(:check, fn state ->
    approval = LangEx.Interrupt.interrupt("Approve value #{state.value}?")
    %{approved: approval}
  end)
  |> Graph.add_node(:finalize, fn state -> %{value: state.value * 10} end)
  |> Graph.add_edge(:__start__, :check)
  |> Graph.add_edge(:check, :finalize)
  |> Graph.add_edge(:finalize, :__end__)
  |> Graph.compile(checkpointer: LangEx.Checkpointer.Redis)

# Pauses at the interrupt
{:interrupt, "Approve value 42?", _state} =
  LangEx.invoke(graph, %{value: 42}, config: [thread_id: "approval-1"])

# Resume with the human's decision
{:ok, result} =
  LangEx.invoke(graph, %LangEx.Command{resume: true}, config: [thread_id: "approval-1"])

On resume the interrupted node re-runs from the top: earlier interrupt/1 calls return their recorded answers and execution continues past the pause point. Side effects placed before an interrupt therefore execute again — keep them idempotent or move them to a later node. interrupt/1 must be called from a graph node function; calling it from tool functions or processes spawned inside a node raises.

When a branch of a parallel super-step interrupts, completed siblings keep their results and their routing: their state updates merge before the pause and their next targets are recorded in the checkpoint, so nothing is re-executed or dropped on resume.

Error Handling

Node exceptions never leak as raises out of invoke/3. After a node's retry policy (if any) is exhausted, the run returns a structured error with the failing node and original cause:

{:error, %LangEx.NodeError{node: :fetch, reason: %Req.TransportError{}}} =
  LangEx.invoke(graph, input)

Per-node execution policies compose on Graph.add_node/4:

Graph.add_node(graph, :fetch, &fetch_data/1,
  timeout: 10_000,
  retry: [max_attempts: 4, initial_interval_ms: 200, backoff_factor: 2.0, jitter: true],
  on_error: fn exception, _state -> %{fetch_failed: Exception.message(exception)} end
)

• retry: — exponential backoff with jitter and a max_interval_ms cap; retries exceptions only ({:error, _} returns are ordinary results)
• timeout: — per-attempt budget; a timed-out attempt raises LangEx.NodeTimeoutError, which the retry policy can retry
• on_error: — fallback invoked after retries are exhausted; its return value becomes the node result (a state update or %LangEx.Command{})
• cache: — memoize results by input state (bounded ETS, optional TTL)
• defer: — fan-in barrier for parallel branches converging at different depths

Programmer errors — routing to an undefined node, a missing conditional mapping — still raise with descriptive messages.

Streaming

Get a lazy stream of execution events:

graph
|> LangEx.stream(%{value: 0})
|> Enum.each(fn
  {:node_start, name} -> IO.puts("Starting #{name}...")
  {:node_end, name, _update} -> IO.puts("Finished #{name}")
  {:done, {:ok, result}} -> IO.inspect(result, label: "Final")
  _ -> :ok
end)

State Reducers

Each state key can have a custom merge function. The built-in Message.add_messages/2 appends and deduplicates by ID; write your own for counters, sets, or domain-specific logic.

Graph.new(
  messages: {[], &Message.add_messages/2},
  total: {0, fn old, new -> old + new end}
)

Subgraphs

Use a compiled graph as a node inside another graph for composable, nested workflows:

inner = Graph.new(value: 0) |> ... |> Graph.compile()

outer =
  Graph.new(value: 0, label: "")
  |> Graph.add_node(:sub, inner)
  |> Graph.add_node(:tag, fn _state -> %{label: "done"} end)
  |> Graph.add_edge(:__start__, :sub)
  |> Graph.add_edge(:sub, :tag)
  |> Graph.add_edge(:tag, :__end__)
  |> Graph.compile()

Interrupts, errors, context, and stream events propagate through subgraph boundaries. Resuming an interrupt that fired inside a subgraph depends on the subgraph's own checkpointing:

• Subgraph compiled with its own checkpointer — checkpoints are namespaced under "{thread_id}/{node_name}" and the subgraph resumes from its saved position; nodes before the interrupt do not re-run.
• No subgraph checkpointer — the subgraph re-runs from :__start__ with resume values injected, so all pre-interrupt subgraph nodes execute again and their side effects must be idempotent.

Multi-Agent Teams

Build teams of cooperating agents that pass the conversation between one another. A swarm lets any agent hand off to any peer; the active agent is tracked in state and persisted across turns:

graph =
  LangEx.Prebuilt.Swarm.create(
    agents: [
      [model: "gpt-4o", name: :router, system_prompt: "Route the user to the right specialist."],
      [model: "gpt-4o", name: :refunds, system_prompt: "Handle refund requests."]
    ],
    default_active_agent: :router,
    checkpointer: LangEx.Checkpointer.Memory
  )

{:ok, state} =
  LangEx.invoke(graph, %{messages: [Message.human("I want a refund")]},
    config: [thread_id: "t-1"]
  )

Each agent automatically gets a transfer_to_<peer> tool for every other agent. A supervisor instead delegates to workers that return control when done:

graph =
  LangEx.Prebuilt.Supervisor.create(
    model: "gpt-4o",
    prompt: "You manage a research agent and a math agent. Delegate to them.",
    agents: [
      [model: "gpt-4o", name: :research, tools: [search_tool]],
      [model: "gpt-4o", name: :math, tools: [calc_tool]]
    ],
    checkpointer: LangEx.Checkpointer.Memory
  )

Handoffs are ordinary tools: a tool function returning a %LangEx.Command{} updates state (e.g. the active agent) and steers routing. Build one directly with LangEx.Prebuilt.Handoff.tool/2.

Runtime Context

Inject dependencies into nodes without closures:

Graph.add_node(:greet, fn _state, context ->
  %{greeting: "Hello from #{context.provider}!"}
end)

LangEx.invoke(graph, %{}, context: %{provider: "Anthropic"})

Send (Fan-Out)

Dynamic map-reduce patterns from conditional edges using %LangEx.Send{}.

Telemetry

All LLM calls and graph executions emit :telemetry events for observability.

Extending LangEx

Custom LLM provider -- implement LangEx.LLM behaviour (chat/2):

defmodule MyApp.LLM.Groq do
  @behaviour LangEx.LLM

  @impl true
  def chat(messages, opts) do
    # Call the Groq API
    {:ok, LangEx.Message.ai("response")}
  end
end

LangEx.LLM.Registry.register_provider(:groq, MyApp.LLM.Groq)

Custom checkpointer -- implement LangEx.Checkpointer behaviour (save/2, load/1, list/2, delete_thread/1).

Examples

License

MIT