Pydantic Ai Graph Agent State Machine Poc 2026

Date: 2026-05-18 Track: sandbox-poc Slug: pydantic-ai-graph-agent-state-machine-poc-2026

Environment

• Python 3.12
• pydantic-ai 1.97.0 (installed via pip)
• pydantic-graph 1.97.0 (bundled with pydantic-ai)
• macOS (local, no API key required for graph execution)

Commands Run

pip3 install "pydantic-ai[graph]"
python3 -c "import pydantic_ai; print(pydantic_ai.__version__)"  # → 1.97.0

PoC 1: 3-Node Linear Pipeline

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
        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)
        return Report()

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

graph = Graph(nodes=[Ingest, Process, Report])
result = asyncio.run(graph.run(Ingest(items=['a', 'b', 'c']), state=PipelineState()))
# Output: "Done: 3 items processed"

Output confirmed: All three nodes executed, state shared correctly across nodes.

Key Observations

• pydantic_graph.BaseNode is deprecated in v1.97.0; migration path is the builder-based GraphBuilder API
• Deprecation warning: PydanticGraphDeprecationWarning: Importing Graph from pydantic_graph is deprecated
• Graph.run() returns a GraphRunResult object (not a tuple) — .output attribute holds the result
• State is a dataclass passed at run-start and mutated in-place through each node's ctx.state
• SimpleStatePersistence provides last_snapshot(), record_run(), load_all(), load_next() methods

API Migration Status

What Worked

1. 3-node stateful pipeline: Ingest → Process → Report — ran correctly
2. Shared mutable state via ctx.state — confirmed working
3. Conditional branching via return type Union[NodeA, End[T]] — API confirmed
4. GraphRunResult.output — output access confirmed

Limitations

• API is in active migration; BaseNode API will be removed in v2
• SimpleStatePersistence.last_snapshot() signature changed in recent releases — EndSnapshot is not callable
• LLM-backed nodes require OPENAI_API_KEY or ANTHROPIC_API_KEY; not tested here (not required for graph structure PoC)
• The new GraphBuilder / @g.step API documentation is sparse at this version

Verdict for Article

Strong sandbox-poc candidate. Local execution confirmed with real version numbers. Three concrete findings:

1. pydantic-graph ships a proper state machine with typed nodes
2. API is migrating from BaseNode → GraphBuilder (useful migration guide angle)
3. State persistence system exists with snapshot/resume capability

Write as a practical PoC guide: install, build a 3-node pipeline, understand state flow, note migration path to v2 API.