magnus919_agent-skills/pydanticai/references/graph.md

PydanticGraph — Graph & State Machine Engine

PydanticGraph is a type-hint-driven async graph and state machine library for Python. It has zero dependency on pydantic-ai — use it standalone for any workflow.

Install: pip install pydantic-graph (or included with pydantic-ai)

Two APIs: BaseNode (class-based) and GraphBuilder (function-based)

Both are fully supported and interoperable.

---

API 1: BaseNode (Class-based)

Nodes are dataclasses subclassing BaseNode[StateT, DepsT, RunEndT].

Core Types

Type	Purpose
BaseNode[StateT, DepsT, RunEndT]	Abstract node class. Subclass and implement run().
End[RunEndT]	Return from run() to signal graph completion.
GraphRunContext[StateT, DepsT]	Holds ctx.state and ctx.deps during execution.
GraphBuilder[StateT, DepsT, InputT, OutputT]	Builder for constructing executable graphs.
Graph[StateT, DepsT, InputT, OutputT]	Ready-to-run graph.

Simple Node

from dataclasses import dataclass
from pydantic_graph import BaseNode, GraphRunContext

@dataclass
class MyNode(BaseNode[MyState]):          # State generic parameter
    foo: int                               # Node parameters (dataclass fields)

    async def run(self, ctx: GraphRunContext[MyState]) -> AnotherNode:
        # Business logic
        ctx.state.counter += 1
        return AnotherNode()               # Return type = next node

Node That Can End the Graph

@dataclass
class MaybeEndNode(BaseNode[MyState, None, int]):  # RunEndT = int
    foo: int

    async def run(self, ctx: GraphRunContext[MyState]) -> AnotherNode | End[int]:
        if self.foo % 5 == 0:
            return End(self.foo)           # End with value int
        return AnotherNode(self.foo + 1)

Full Graph Example

from __future__ import annotations
from dataclasses import dataclass
from pydantic_graph import BaseNode, End, GraphBuilder, GraphRunContext, StepContext

@dataclass
class DivisibleBy5(BaseNode[None, None, int]):
    foo: int
    async def run(self, ctx: GraphRunContext) -> Increment | End[int]:
        if self.foo % 5 == 0:
            return End(self.foo)
        return Increment(self.foo)

@dataclass
class Increment(BaseNode):
    foo: int
    async def run(self, ctx: GraphRunContext) -> DivisibleBy5:
        return DivisibleBy5(self.foo + 1)

g = GraphBuilder(input_type=int, output_type=int)

@g.step
async def start(ctx: StepContext[None, None, int]) -> DivisibleBy5:
    return DivisibleBy5(ctx.inputs)

g.add(
    g.node(DivisibleBy5),
    g.node(Increment),
    g.edge_from(g.start_node).to(start),
)

graph = g.build()
result = await graph.run(inputs=4)
print(result)  # 5

Stateful Graph Example

from dataclasses import dataclass
from pydantic_graph import BaseNode, End, GraphBuilder, GraphRunContext, StepContext

@dataclass
class VendingState:
    user_balance: float = 0.0
    product: str | None = None

@dataclass
class InsertCoin(BaseNode[VendingState]):
    async def run(self, ctx: GraphRunContext[VendingState]) -> CoinsInserted:
        return CoinsInserted(float(input('Insert coins: ')))

@dataclass
class CoinsInserted(BaseNode[VendingState]):
    amount: float
    async def run(self, ctx: GraphRunContext[VendingState]) -> SelectProduct | Purchase:
        ctx.state.user_balance += self.amount
        if ctx.state.product:
            return Purchase(ctx.state.product)
        return SelectProduct()

g = GraphBuilder(state_type=VendingState)
@g.step
async def start(ctx: StepContext[VendingState, None, None]) -> InsertCoin:
    return InsertCoin()
g.add(
    g.node(InsertCoin), g.node(CoinsInserted),
    g.edge_from(g.start_node).to(start),
)
graph = g.build()
result = await graph.run(state=VendingState())

Dependency Injection

from dataclasses import dataclass
from pydantic_graph import BaseNode, End, GraphBuilder, GraphRunContext, StepContext

@dataclass
class GraphDeps:
    db_url: str

@dataclass
class QueryNode(BaseNode[None, GraphDeps, str]):
    async def run(self, ctx: GraphRunContext[None, GraphDeps]) -> End[str]:
        return End(f"Connected to {ctx.deps.db_url}")

g = GraphBuilder(deps_type=GraphDeps, output_type=str)
@g.step
async def start(ctx: StepContext[None, GraphDeps, None]) -> QueryNode:
    return QueryNode()
g.add(g.node(QueryNode), g.edge_from(g.start_node).to(start))
graph = g.build()
result = await graph.run(deps=GraphDeps(db_url='postgres://...'))

GenAI Example (Agent Inside Graph)

from dataclasses import dataclass
from pydantic_ai import Agent
from pydantic_graph import BaseNode, End, GraphBuilder, GraphRunContext, StepContext

@dataclass
class State:
    messages: list = None

email_writer = Agent('openai:gpt-5.2', output_type=str,
    instructions='Write a welcome email.')
feedback_agent = Agent('openai:gpt-5.2', output_type=str,
    instructions='Review the email and provide feedback.')

@dataclass
class WriteEmail(BaseNode[State]):
    feedback: str | None = None
    async def run(self, ctx: GraphRunContext[State]) -> Feedback:
        prompt = f"Write email. Feedback: {self.feedback}" if self.feedback else "Write welcome email."
        result = await email_writer.run(prompt)
        return Feedback(result.output)

@dataclass
class Feedback(BaseNode[State, None, str]):
    email: str
    async def run(self, ctx: GraphRunContext[State]) -> WriteEmail | End[str]:
        result = await feedback_agent.run(f"Review: {self.email}")
        if 'approve' in result.output.lower():
            return End(self.email)
        return WriteEmail(feedback=result.output)

g = GraphBuilder(state_type=State, output_type=str)
@g.step
async def start(ctx: StepContext[State, None, None]) -> WriteEmail:
    return WriteEmail()
g.add(g.node(WriteEmail), g.node(Feedback), g.edge_from(g.start_node).to(start))
graph = g.build()

Mermaid Diagrams

print(graph)
# or
mermaid = graph.render(title='My Graph', direction='LR')  # TB, LR, RL, BT

Step-by-step execution with iter()

async with graph.iter(state=state) as run:
    async for event in run:
        print(f"Node: {event}")
        if run.output is not None:
            break

---

API 2: GraphBuilder (Function-based)

The builder API provides concise syntax with parallel execution primitives.

Simple Chain

from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext

@dataclass
class CounterState:
    value: int = 0

g = GraphBuilder(state_type=CounterState, output_type=int)

@g.step
async def increment(ctx: StepContext[CounterState, None, None]) -> int:
    ctx.state.value += 1
    return ctx.state.value

@g.step
async def double_it(ctx: StepContext[CounterState, None, int]) -> int:
    return ctx.inputs * 2

g.add(
    g.edge_from(g.start_node).to(increment),
    g.edge_from(increment).to(double_it),
    g.edge_from(double_it).to(g.end_node),
)

graph = g.build()
result = await graph.run(state=CounterState())

Parallel Processing with Map

from pydantic_graph import GraphBuilder, StepContext, reduce_list_append

g = GraphBuilder(state_type=CounterState, input_type=list[int], output_type=list[int])

@g.step
async def square(ctx: StepContext[CounterState, None, int]) -> int:
    ctx.state.items_processed += 1
    return ctx.inputs * ctx.inputs

collect = g.join(reduce_list_append, initial_factory=list[int])

g.add(
    g.edge_from(g.start_node).map().to(square),        # Fan-out: one per input item
    g.edge_from(square).to(collect),                    # Collect results
    g.edge_from(collect).to(g.end_node),
)

graph = g.build()
result = await graph.run(state=CounterState(), inputs=[1, 2, 3])

Parallel with Broadcast

# Same input goes to multiple parallel paths
g.edge_from(g.start_node).broadcast().to(square, cube, sqrt)

Decisions (Conditional Branching)

# Use a decision node to route based on a condition
g.edge_from(check_node).to(
    on_true=process_positive,
    on_false=process_negative,
)

Joins and Reducers

Reducer	Behavior
reduce_list_append	Collect items into list
reduce_list_extend	Extend list with items
reduce_sum	Sum numeric values
reduce_dict_update	Merge dicts
reduce_first_value	Take first value
reduce_null	Discard values

my_join = g.join(reduce_sum, initial_factory=lambda: 0)

StepContext Fields

Field	Type	Description
ctx.state	StateT	Mutable shared state
ctx.deps	DepsT	Injected dependencies
ctx.inputs	InputT	Input data for this step

Graph generics

GraphBuilder is generic over [StateT, DepsT, InputT, OutputT]:

• Omit state_type for stateless graphs (defaults to None)
• Omit deps_type for dependency-free graphs
• Omit input_type/output_type for simple graphs

Comparison: BaseNode vs GraphBuilder

Aspect	BaseNode API	GraphBuilder API
Style	Class-based (OO)	Function-based (builder pattern)
State mutation	Via dataclass fields + ctx.state	Via ctx.state
Parallelism	Manual fork/join	Built-in .map(), .broadcast(), joins
Complexity	Better for complex node logic	Better for simple linear flows
Graph structure	Inferred from run() return types	Explicit via g.edge_from()