Pydantic AI Graph Agents: Build Stateful Pipelines with pydantic-graph

The standard agent loop — call model, pick a tool, call model again — breaks down for long-horizon tasks. You need a way to declare the stages your agent moves through, share typed state between them, and resume from any checkpoint. That is what pydantic-graph provides.

Effloow Lab installed pydantic-ai 1.97.0 locally, ran a 3-node state-machine pipeline, and documented what the API looks like in 2026 — including the active migration from BaseNode to the new GraphBuilder beta.

What pydantic-graph Actually Is

pydantic-graph ships as part of the pydantic-ai package. It is a typed state machine runtime: you define nodes as Python classes (or functions in the new API), connect them via return types, and run them with shared mutable state.

There is no magic scheduler, no background thread pool, and no vendor lock-in. It is pure Python — you decide whether individual nodes call an LLM, a database, or just compute locally.

Key concepts:

• Node — a unit of work. Receives the current state, does something, returns the next node or End.
• State — a dataclass shared across all nodes and mutated in-place via ctx.state.
• Graph — the container that validates node connections and drives execution.
• Persistence — optional snapshot system to save state before/after each node.

Install and First Run

pip install "pydantic-ai[graph]"
python -c "import pydantic_ai; print(pydantic_ai.__version__)"
# 1.97.0

The [graph] extra installs pydantic-graph alongside pydantic-ai.

A 3-Node Pipeline: Ingest → Process → Report

Effloow Lab confirmed this pipeline runs correctly on v1.97.0:

from pydantic_graph import Graph, BaseNode, End, GraphRunContext
from dataclasses import dataclass, field
from typing import Union
import asyncio

@dataclass
class PipelineState:
    items: list[str] = field(default_factory=list)
    processed: int = 0

@dataclass
class Ingest(BaseNode[PipelineState]):
    items: list[str]

    async def run(self, ctx: GraphRunContext[PipelineState]) -> "Process":
        ctx.state.items = self.items
        print(f"[Ingest] loaded {len(self.items)} items")
        return Process()

@dataclass
class Process(BaseNode[PipelineState]):
    async def run(
        self, ctx: GraphRunContext[PipelineState]
    ) -> Union["Report", End[str]]:
        ctx.state.processed = len(ctx.state.items)
        if ctx.state.processed == 0:
            return End("nothing to report")
        return Report()

@dataclass
class Report(BaseNode[PipelineState]):
    async def run(self, ctx: GraphRunContext[PipelineState]) -> End[str]:
        summary = f"Done: {ctx.state.processed} items processed"
        return End(summary)

graph = Graph(nodes=[Ingest, Process, Report])

async def main():
    result = await graph.run(
        Ingest(items=["task-a", "task-b", "task-c"]),
        state=PipelineState(),
    )
    print(result.output)  # "Done: 3 items processed"

asyncio.run(main())

Three things to notice:

1. Return type declares routing — Process.run returns Union["Report", End[str]]. The graph validates this at build time. You cannot return a node type that is not in the graph.
2. State is a shared dataclass — ctx.state is the same object across all nodes. Mutations persist between nodes automatically.
3. End[T] is the terminal — returning End(value) stops execution and sets result.output.

The Two APIs: Which One to Use?

As of version 1.97.0, pydantic-graph ships two APIs.

When you run the BaseNode API today you will see:

PydanticGraphDeprecationWarning: Importing Graph from pydantic_graph is deprecated.
The BaseNode-based Graph runner ... will be removed (or repurposed) in v2;
use the builder-based GraphBuilder API instead.

For new projects, start with the GraphBuilder beta. For existing code, the migration is mechanical — class-to-function, Graph to g.build().

GraphBuilder Beta: What It Looks Like

from pydantic_graph.beta import GraphBuilder
from dataclasses import dataclass, field

@dataclass
class State:
    results: list[str] = field(default_factory=list)

g = GraphBuilder(State)

@g.step
async def ingest(state: State, items: list[str]) -> str:
    state.results.extend(items)
    return "process"

@g.step
async def process(state: State) -> str:
    state.results = [r.upper() for r in state.results]
    return "report"

@g.step
async def report(state: State) -> None:
    print(f"Done: {state.results}")

pipeline = g.build(start="ingest")

The routing is now string-based (return "process") rather than return-type based. Both approaches have tradeoffs: the BaseNode approach gives compile-time route checking; the builder approach is less boilerplate and easier to dynamically compose.

State Persistence: Snapshot and Resume

pydantic-graph ships SimpleStatePersistence for saving state snapshots before and after each node runs. This matters for long-running pipelines where you want to resume from a checkpoint rather than restart from scratch.

from pydantic_graph import SimpleStatePersistence

persistence = SimpleStatePersistence()
result = await graph.run(
    Ingest(items=["a", "b", "c"]),
    state=PipelineState(),
    persistence=persistence,
)

# Load the last checkpoint
snapshot = await persistence.last_snapshot(graph)
print(snapshot)  # EndSnapshot with final state

The persistence interface defines snapshot_node, snapshot_node_if_new, snapshot_end, and load_all. For production use, you would back this with a database rather than in-memory storage.

When to Use Graph Agents

Graph-based agents add overhead compared to a simple agent loop. The overhead pays off when:

Connecting an LLM Node

The graph does not require an LLM. When you do want one, any BaseNode can call a Pydantic AI agent internally:

from pydantic_ai import Agent

summarizer = Agent("openai:gpt-4.1-mini", result_type=str)

@dataclass
class Summarize(BaseNode[PipelineState]):
    async def run(self, ctx: GraphRunContext[PipelineState]) -> End[str]:
        items_text = "\n".join(ctx.state.items)
        result = await summarizer.run(f"Summarize: {items_text}")
        return End(result.data)

The graph manages state and routing; the Agent manages the model call. They compose without any special glue.

Migration Checklist: BaseNode → GraphBuilder

If you have existing code using BaseNode:

1. Replace class Foo(BaseNode[S]): async def run(...) with @g.step async def foo(...)
2. Replace Graph(nodes=[...]) with g = GraphBuilder(S); ... pipeline = g.build(start="foo")
3. Change return NextNode() to return "next_node" (string routing)
4. Change return End(value) to return None (end of graph) or keep End — check beta docs
5. Run with await pipeline.run(state=S(), input=your_input)

The persistence API remains the same in both versions.

FAQ

Does pydantic-graph work without pydantic-ai agents?
Yes. Nodes are plain async functions and can call any external service. You only need pydantic-ai Agent if you want LLM calls inside a node.

Is the GraphBuilder API production-ready?
It is tagged as beta in the pydantic_graph.beta module. The BaseNode API is more stable right now, but the GraphBuilder is the forward path — use it for new projects.

Can I use pydantic-graph with Anthropic Claude?
Yes. Inside any node, use Agent("anthropic:claude-sonnet-4-6") for the model call. The graph does not care which model your nodes call.

What is the difference between pydantic-graph and LangGraph?
Both are graph-based agent runtimes. pydantic-graph is lighter, more Pythonic, and avoids the Runnable abstraction layer. LangGraph has more built-in integrations and a visual studio. For Python-first teams, pydantic-graph has less ceremony.

When will BaseNode be removed?
The deprecation warning says it will be removed or repurposed in v2. No timeline was specified in v1.97.0 release notes.