Pyantra

Typed, observable, reliable workflows for production AI agents.

Pyantra is a Python framework for building AI agent workflows as typed graphs of nodes and edges. It is designed to be small, composable, and dependency-free, with reliability and observability built in by default.

Agents should be observable, reproducible, testable, reliable, and cost-aware by default.

Features

• Typed workflows — nodes flow state through the graph and are type-checked end to end.
• State merging — per-field reducers (Annotated[list[T], reducer]) and partial updates, so concurrent and sequential nodes can contribute to shared state safely.
• Compile-time validation — malformed graphs fail early with clear errors, not at runtime.
• Parallel fan-out — nodes run concurrently on isolated state copies and merge back with reducers.
• Reliability first-class — per-node retry with backoff, timeouts, and circuit breakers.
• Checkpoints — durable snapshots that let failed runs resume where they left off, backed by memory or SQLite.
• Human-in-the-loop — interrupt() pauses a run for input; resume() continues it.
• Structured observability — every run produces a rich event trace.
• LLM abstraction — a dependency-free provider interface with built-in token/cost tracking.
• Sync + async — one traversal engine exposed through both run() and arun().
• Zero dependencies — pure Python standard library. No databases, no services.

Installation

pip install pyantra

Requires Python 3.10 or later.

Quickstart

from dataclasses import dataclass

from pyantra import Graph


@dataclass
class State:
    value: int


graph = Graph(State)


@graph.node
def increment(state: State) -> State:
    state.value += 1
    return state


@graph.node
def double(state: State) -> State:
    state.value *= 2
    return state


graph.set_entry_point(increment)
graph.add_edge(increment, double)

app = graph.compile()

result = app.run(State(value=1))

assert result.state.value == 4

Core concepts

A workflow is a graph of nodes connected by edges. Nodes receive state and return updated state (or mutate it in place and return None). The graph is compiled — and validated — before it can be executed.

Conditional routing

@graph.node
def classify(state: State) -> State:
    return state

@graph.node
def process_positive(state: State) -> State:
    ...

@graph.node
def process_negative(state: State) -> State:
    ...

def route(state: State) -> str:
    return "positive" if state.value >= 0 else "negative"

graph.set_entry_point(classify)
graph.add_conditional_edges(
    classify,
    route,
    {"positive": process_positive, "negative": process_negative},
)

Nodes can also terminate a workflow explicitly:

from pyantra import END

graph.add_edge(final_node, END)

State merging and reducers

Annotate a state field with a reducer to control how updates combine instead of overwriting:

from typing import Annotated, operator
from dataclasses import dataclass, field

@dataclass
class State:
    messages: Annotated[list[str], operator.add] = field(default_factory=list)

@graph.node
def record(state: State) -> dict[str, list[str]]:
    return {"messages": ["hello"]}

Nodes may return:

• None — the node mutated state in place (reducers do not apply).
• the state type — merged field by field; annotated fields are reduced against the current values, all others replace.
• a dict of field updates — the same merge, applied per key.

Any (current, update) -> new callable works as a reducer; operator.add on lists, operator.or_ on sets, and dict merges are common. State merge works for sequential runs and is what makes parallel fan-out safe.

Parallel execution

Fan out from a node to several branches that run concurrently, then continue at a join node (or end):

graph.set_entry_point(ingest)
graph.add_parallel_edges(ingest, summarize, classify, join=combine)

Each branch executes on an isolated copy of the current state. Results merge back with the field reducers — unannotated fields are last-writer-wins, so use reducers for shared fields you want to accumulate.

Async execution

The exact same graph runs asynchronously:

result = await app.arun(State(value=1))

Async nodes (async def) are supported seamlessly in both modes.

Reliability

Reliability is configured per node with NodeConfig. No configuration means fail-fast: a raised exception fails the node immediately.

from pyantra import Backoff, Graph, NodeConfig

graph = Graph(State)

@graph.node
def fetch(state: State) -> State:
    ...

fetch.config = NodeConfig(
    retries=4,                       # retries after the first attempt
    backoff=Backoff.EXPONENTIAL,     # or Backoff.FIXED / Backoff.NONE
    base_delay=1.0,
    max_delay=30.0,
    timeout=15.0,                    # seconds per attempt
)

graph.set_entry_point(fetch)

NodeConfig can also be passed directly when registering a node:

graph.add_node(fetch, name="fetch", config=NodeConfig(retries=3))

Never retry certain errors

Errors that must not be retried (bad input, schema violations, …) can be marked explicitly — they fail immediately regardless of the retry policy:

from pyantra import non_retryable

@non_retryable
class ValidationError(Exception):
    ...

def fetch(state: State) -> State:
    raise ValidationError("bad request")

Retry only certain errors

By default any retryable failure is retried. Use retry_on to restrict retries to specific exception types — anything else fails immediately. This pairs well with the @non_retryable marker, which always wins:

fetch.config = NodeConfig(
    retries=4,
    retry_on=(ConnectionError, TimeoutError),  # only retry these
)

A single type is accepted as shorthand: retry_on=ConnectionError. Timeouts count as retryable failures, so retry_on=(TimeoutError,) retries on timeouts.

Circuit breakers

A circuit breaker stops hammering a node after a run of consecutive failures, then allows a trial call once a reset period elapses:

from pyantra import CircuitBreaker, NodeConfig

breaker = CircuitBreaker(failure_threshold=5, reset_timeout=30.0)

graph.add_node(
    external_api,
    name="external_api",
    config=NodeConfig(breaker=breaker),
)

LLMs

Pyantra ships a dependency-free provider abstraction (LLM) plus Message, Usage, and LLMResponse value types. Any model adapter implements generate() / agenerate(); providers (OpenAI, Anthropic, …) can live as extras. Cost and token usage is aggregated per run with UsageTracker, and MockLLM provides scripted responses for tests.

from pyantra import Message, MockLLM, UsageTracker

llm = MockLLM(responses=["summarized"], input_tokens=3, output_tokens=2)
tracker = UsageTracker()

def summarize(state: State) -> State:
    resp = llm.generate([Message(role="user", content=state.prompt)])
    tracker.add(resp.usage)
    state.summary = resp.content
    return state

tracker.total reports aggregate input/output/cache tokens and cost. See docs/llm.md for the design and roadmap.

Checkpoints and resume

Pass a checkpoint store to run() and a run can resume from its last successful node after a failure:

from pyantra import MemoryCheckpointStore

store = MemoryCheckpointStore()

first = app.run(state, checkpointer=store, run_id="order-123")
assert first.status == RunStatus.FAILED

# Re-run with the same run_id: resumes where it stopped instead of restarting.
second = app.run(state, checkpointer=store, run_id="order-123")

CheckpointStore is an abstract interface; in-memory storage ships by default and a durable SQLiteCheckpointStore is built in:

from pyantra import SQLiteCheckpointStore

store = SQLiteCheckpointStore("checkpoints.db")

State, events, and pending interrupts are serialized with pickle, so SQLiteCheckpointStore survives process restarts. Additional backends (Postgres, Redis) can be added behind the same interface.

Human-in-the-loop

Call interrupt() from a node to pause a run and request input. The run pauses with RunStatus.PAUSED, its payload lands on run.interrupt, and the state is checkpointed. Resume with app.resume(...) — the call to interrupt() then returns the value you provided:

from pyantra import interrupt

@graph.node
def review(state: State) -> State:
    decision = interrupt({"question": "approve this change?", "draft": state.draft})
    state.decision = decision
    return state

run = app.run(state, checkpointer=store, run_id="review-7")
assert run.status == RunStatus.PAUSED
print(run.interrupt)          # {"question": "...", "draft": ...}

resumed = app.resume("review-7", "approved", checkpointer=store)
assert resumed.status == RunStatus.COMPLETED

interrupt() raises a BaseException-derived signal, so a node's own except Exception cannot swallow it. Multiple sequential interruptions in one run are supported; each resume() answers the most recent one.

Observability

Every run returns a Run object with a structured event trace — no logging parsing required:

result = app.run(state)

result.run_id      # unique id for the run
result.status      # RunStatus (pending, running, completed, failed, paused, ...)
result.state       # final (or last known) state
result.events      # ordered list of RunEvent
result.error       # human-readable failure message, when failed
result.exception   # the underlying exception, when failed
result.interrupt   # the human-in-the-loop payload, when paused

Example events:

run.started        node.started        node.attempt.failed
run.completed      node.completed      node.attempt.timeout
run.failed         node.failed         node.retrying
run.paused         node.interrupted    edge.selected
run.resumed

Errors

All exceptions derive from PyantraError:

PyantraError
├── GraphCompileError      — the graph failed validation at compile time
├── GraphExecutionError
│   ├── NodeExecutionError — a node raised during execution
│   ├── NodeTimeoutError   — a node exceeded its configured timeout
│   ├── RetryExhaustedError— retries were exhausted
│   ├── CircuitOpenError   — a circuit breaker refused execution
│   ├── InvalidRouteError  — a router returned an unknown destination
│   └── MaxIterationsError — a run exceeded max_iterations
├── CheckpointError        — checkpoint storage or resume failed
└── NonRetryableError      — base class for never-retried errors

Examples

End-to-end runnable examples live in examples/:

python examples/basic_workflow.py
python examples/reliability_workflow.py
python examples/advanced_workflow.py   # reducers, parallel, human-in-the-loop

Development

pip install -e ".[dev]"

ruff check .          # lint
mypy pyantra          # type check
pytest                # test suite

This repository uses conventional commits.

Roadmap

• Automatic LLM usage capture with per-run budgets and compression
• LLM caching and model tiering
• Multi-agent delegation and scoped handoffs
• interrupt() defaults and tool/approval-specific helpers
• Deterministic replay and trace-based regression testing

License

MIT