pydantic_graph.graph_builder

Builder-based graph API: builder, graph runner, and mermaid rendering.

This module is the canonical home for the builder-based graph API: GraphBuilder for declaratively constructing executable graphs, Graph and GraphRun for executing them, and the mermaid rendering helpers used by Graph.render(). The same public symbols are re-exported from pydantic_graph directly.

StateT module-attribute

StateT = TypeVar('StateT', infer_variance=True)

Type variable for graph state.

DepsT module-attribute

DepsT = TypeVar('DepsT', infer_variance=True)

Type variable for graph dependencies.

InputT module-attribute

InputT = TypeVar('InputT', infer_variance=True)

Type variable for graph inputs.

OutputT module-attribute

OutputT = TypeVar('OutputT', infer_variance=True)

Type variable for graph outputs.

EndMarker dataclass

Bases: Generic[OutputT]

A marker indicating the end of graph execution with a final value.

EndMarker is used internally to signal that the graph has completed execution and carries the final output value.

Class Type Parameters:

Name	Bound or Constraints	Description	Default
OutputT		The type of the final output value	required

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@dataclass(init=False)
class EndMarker(Generic[OutputT]):
    """A marker indicating the end of graph execution with a final value.

    EndMarker is used internally to signal that the graph has completed
    execution and carries the final output value.

    Type Parameters:
        OutputT: The type of the final output value
    """

    _value: OutputT
    """The final output value from the graph execution."""

    def __init__(self, value: OutputT):
        # This manually-defined initializer is necessary due to https://github.com/python/mypy/issues/17623.
        self._value = value

    @property
    def value(self) -> OutputT:
        return self._value

ErrorMarker dataclass

A marker indicating that a graph node raised an exception.

Yielded by the graph iterator instead of raising immediately, allowing the caller to recover by sending new tasks via GraphRun.next() or GraphRun.override_next(). If the caller does not override, the error is re-raised on the next iteration.

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@dataclass
class ErrorMarker:
    """A marker indicating that a graph node raised an exception.

    Yielded by the graph iterator instead of raising immediately, allowing the caller
    to recover by sending new tasks via `GraphRun.next()` or `GraphRun.override_next()`.
    If the caller does not override, the error is re-raised on the next iteration.
    """

    error: BaseException
    """The exception raised by the node."""

error instance-attribute

error: BaseException

The exception raised by the node.

JoinItem dataclass

An item representing data flowing into a join operation.

JoinItem carries input data from a parallel execution path to a join node, along with metadata about which execution 'fork' it originated from.

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@dataclass
class JoinItem:
    """An item representing data flowing into a join operation.

    JoinItem carries input data from a parallel execution path to a join
    node, along with metadata about which execution 'fork' it originated from.
    """

    join_id: JoinID
    """The ID of the join node this item is targeting."""

    inputs: Any
    """The input data for the join operation."""

    fork_stack: ForkStack
    """The stack of ForkStackItems that led to producing this join item."""

join_id instance-attribute

join_id: JoinID

The ID of the join node this item is targeting.

inputs instance-attribute

inputs: Any

The input data for the join operation.

fork_stack instance-attribute

fork_stack: ForkStack

The stack of ForkStackItems that led to producing this join item.

Graph dataclass

Bases: Generic[StateT, DepsT, InputT, OutputT]

A complete graph definition ready for execution.

The Graph class represents a complete workflow graph with typed inputs, outputs, state, and dependencies. It contains all nodes, edges, and metadata needed for execution.

Class Type Parameters:

Name	Bound or Constraints	Description	Default
StateT		The type of the graph state	required
DepsT		The type of the dependencies	required
InputT		The type of the input data	required
OutputT		The type of the output data	required

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@dataclass(repr=False)
class Graph(Generic[StateT, DepsT, InputT, OutputT]):
    """A complete graph definition ready for execution.

    The Graph class represents a complete workflow graph with typed inputs,
    outputs, state, and dependencies. It contains all nodes, edges, and
    metadata needed for execution.

    Type Parameters:
        StateT: The type of the graph state
        DepsT: The type of the dependencies
        InputT: The type of the input data
        OutputT: The type of the output data
    """

    name: str | None
    """Optional name for the graph, if not provided the name will be inferred from the calling frame on the first call to a graph method."""

    state_type: type[StateT]
    """The type of the graph state."""

    deps_type: type[DepsT]
    """The type of the dependencies."""

    input_type: type[InputT]
    """The type of the input data."""

    output_type: type[OutputT]
    """The type of the output data."""

    auto_instrument: bool
    """Whether to automatically create instrumentation spans."""

    nodes: dict[NodeID, AnyNode]
    """All nodes in the graph indexed by their ID."""

    edges_by_source: dict[NodeID, list[Path]]
    """Outgoing paths from each source node."""

    parent_forks: dict[JoinID, ParentFork[NodeID]]
    """Parent fork information for each join node."""

    intermediate_join_nodes: dict[JoinID, set[JoinID]]
    """For each join, the set of other joins that appear between it and its parent fork.

    Used to determine which joins are "final" (have no other joins as intermediates) and
    which joins should preserve fork stacks when proceeding downstream."""

    def get_parent_fork(self, join_id: JoinID) -> ParentFork[NodeID]:
        """Get the parent fork information for a join node.

        Args:
            join_id: The ID of the join node

        Returns:
            The parent fork information for the join

        Raises:
            RuntimeError: If the join ID is not found or has no parent fork
        """
        result = self.parent_forks.get(join_id)
        if result is None:
            raise RuntimeError(f'Node {join_id} is not a join node or did not have a dominating fork (this is a bug)')
        return result

    def is_final_join(self, join_id: JoinID) -> bool:
        """Check if a join is 'final' (has no downstream joins with the same parent fork).

        A join is non-final if it appears as an intermediate node for another join
        with the same parent fork.

        Args:
            join_id: The ID of the join node

        Returns:
            True if the join is final, False if it's non-final
        """
        # Check if this join appears in any other join's intermediate_join_nodes
        for intermediate_joins in self.intermediate_join_nodes.values():
            if join_id in intermediate_joins:
                return False
        return True

    async def run(
        self,
        *,
        state: StateT = None,
        deps: DepsT = None,
        inputs: InputT = None,
        span: AbstractContextManager[AbstractSpan] | None = None,
        infer_name: bool = True,
    ) -> OutputT:
        """Execute the graph and return the final output.

        This is the main entry point for graph execution. It runs the graph
        to completion and returns the final output value.

        Args:
            state: The graph state instance
            deps: The dependencies instance
            inputs: The input data for the graph
            span: Optional span for tracing/instrumentation
            infer_name: Whether to infer the graph name from the calling frame.

        Returns:
            The final output from the graph execution
        """
        if infer_name and self.name is None:
            inferred_name = infer_obj_name(self, depth=2)
            if inferred_name is not None:  # pragma: no branch
                self.name = inferred_name

        async with self.iter(state=state, deps=deps, inputs=inputs, span=span, infer_name=False) as graph_run:
            # Note: This would probably be better using `async for _ in graph_run`, but this tests the `next` method,
            # which I'm less confident will be implemented correctly if not used on the critical path. We can change it
            # once we have tests, etc.
            event: Any = None
            while True:
                try:
                    event = await graph_run.next(event)
                except StopAsyncIteration:
                    assert isinstance(event, EndMarker), 'Graph run should end with an EndMarker.'
                    return cast(EndMarker[OutputT], event).value

    def run_sync(
        self,
        *,
        state: StateT = None,
        deps: DepsT = None,
        inputs: InputT = None,
        span: AbstractContextManager[AbstractSpan] | None = None,
        infer_name: bool = True,
    ) -> OutputT:
        """Synchronously execute the graph and return the final output.

        This is a convenience wrapper around [`run`][pydantic_graph.Graph.run] that runs the coroutine on the
        current event loop via `loop.run_until_complete(...)`. As such, it cannot be called from inside async
        code or when an event loop is already running.

        Args:
            state: The graph state instance
            deps: The dependencies instance
            inputs: The input data for the graph
            span: Optional span for tracing/instrumentation
            infer_name: Whether to infer the graph name from the calling frame.

        Returns:
            The final output from the graph execution
        """
        if infer_name and self.name is None:
            inferred_name = infer_obj_name(self, depth=2)
            if inferred_name is not None:  # pragma: no branch
                self.name = inferred_name
        return _utils.get_event_loop().run_until_complete(
            self.run(state=state, deps=deps, inputs=inputs, span=span, infer_name=False)
        )

    @asynccontextmanager
    async def iter(
        self,
        *,
        state: StateT = None,
        deps: DepsT = None,
        inputs: InputT = None,
        span: AbstractContextManager[AbstractSpan] | None = None,
        infer_name: bool = True,
    ) -> AsyncIterator[GraphRun[StateT, DepsT, OutputT]]:
        """Create an iterator for step-by-step graph execution.

        This method allows for more fine-grained control over graph execution,
        enabling inspection of intermediate states and results.

        Args:
            state: The graph state instance
            deps: The dependencies instance
            inputs: The input data for the graph
            span: Optional span for tracing/instrumentation
            infer_name: Whether to infer the graph name from the calling frame.

        Yields:
            A GraphRun instance that can be iterated for step-by-step execution
        """
        if infer_name and self.name is None:
            inferred_name = infer_obj_name(self, depth=3)  # depth=3 because asynccontextmanager adds one
            if inferred_name is not None:  # pragma: no branch
                self.name = inferred_name

        with ExitStack() as stack:
            entered_span: AbstractSpan | None = None
            if span is None:
                if self.auto_instrument:
                    entered_span = stack.enter_context(logfire_span('run graph {graph.name}', graph=self))
            else:
                entered_span = stack.enter_context(span)  # pragma: lax no cover
            traceparent = None if entered_span is None else get_traceparent(entered_span)
            async with GraphRun[StateT, DepsT, OutputT](
                graph=self,
                state=state,
                deps=deps,
                inputs=inputs,
                traceparent=traceparent,
            ) as graph_run:
                yield graph_run

    def render(self, *, title: str | None = None, direction: StateDiagramDirection | None = None) -> str:
        """Render the graph as a Mermaid diagram string.

        Args:
            title: Optional title for the diagram
            direction: Optional direction for the diagram layout

        Returns:
            A string containing the Mermaid diagram representation
        """
        return build_mermaid_graph(self.nodes, self.edges_by_source).render(title=title, direction=direction)

    def __repr__(self) -> str:
        super_repr = super().__repr__()  # include class and memory address
        # Insert the result of calling `__str__` before the final '>' in the repr
        return f'{super_repr[:-1]}\n{self}\n{super_repr[-1]}'

    def __str__(self) -> str:
        """Return a Mermaid diagram representation of the graph.

        Returns:
            A string containing the Mermaid diagram of the graph
        """
        return self.render()

name instance-attribute

name: str | None

Optional name for the graph, if not provided the name will be inferred from the calling frame on the first call to a graph method.

state_type instance-attribute

state_type: type[StateT]

The type of the graph state.

deps_type instance-attribute

deps_type: type[DepsT]

The type of the dependencies.

input_type instance-attribute

input_type: type[InputT]

The type of the input data.

output_type instance-attribute

output_type: type[OutputT]

The type of the output data.

auto_instrument instance-attribute

auto_instrument: bool

Whether to automatically create instrumentation spans.

nodes instance-attribute

nodes: dict[NodeID, AnyNode]

All nodes in the graph indexed by their ID.

edges_by_source instance-attribute

edges_by_source: dict[NodeID, list[Path]]

Outgoing paths from each source node.

parent_forks instance-attribute

parent_forks: dict[JoinID, ParentFork[NodeID]]

Parent fork information for each join node.

intermediate_join_nodes instance-attribute

intermediate_join_nodes: dict[JoinID, set[JoinID]]

For each join, the set of other joins that appear between it and its parent fork.

Used to determine which joins are "final" (have no other joins as intermediates) and which joins should preserve fork stacks when proceeding downstream.

get_parent_fork

get_parent_fork(join_id: JoinID) -> ParentFork[NodeID]

Get the parent fork information for a join node.

Parameters:

Name	Type	Description	Default
join_id	JoinID	The ID of the join node	required

Returns:

Type	Description
ParentFork[NodeID]	The parent fork information for the join

Raises:

Type	Description
RuntimeError	If the join ID is not found or has no parent fork

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def get_parent_fork(self, join_id: JoinID) -> ParentFork[NodeID]:
    """Get the parent fork information for a join node.

    Args:
        join_id: The ID of the join node

    Returns:
        The parent fork information for the join

    Raises:
        RuntimeError: If the join ID is not found or has no parent fork
    """
    result = self.parent_forks.get(join_id)
    if result is None:
        raise RuntimeError(f'Node {join_id} is not a join node or did not have a dominating fork (this is a bug)')
    return result

is_final_join

is_final_join(join_id: JoinID) -> bool

Check if a join is 'final' (has no downstream joins with the same parent fork).

A join is non-final if it appears as an intermediate node for another join with the same parent fork.

Parameters:

Name	Type	Description	Default
join_id	JoinID	The ID of the join node	required

Returns:

Type	Description
bool	True if the join is final, False if it's non-final

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def is_final_join(self, join_id: JoinID) -> bool:
    """Check if a join is 'final' (has no downstream joins with the same parent fork).

    A join is non-final if it appears as an intermediate node for another join
    with the same parent fork.

    Args:
        join_id: The ID of the join node

    Returns:
        True if the join is final, False if it's non-final
    """
    # Check if this join appears in any other join's intermediate_join_nodes
    for intermediate_joins in self.intermediate_join_nodes.values():
        if join_id in intermediate_joins:
            return False
    return True

run async

run(
    *,
    state: StateT = None,
    deps: DepsT = None,
    inputs: InputT = None,
    span: (
        AbstractContextManager[AbstractSpan] | None
    ) = None,
    infer_name: bool = True
) -> OutputT

Execute the graph and return the final output.

This is the main entry point for graph execution. It runs the graph to completion and returns the final output value.

Parameters:

Name	Type	Description	Default
state	StateT	The graph state instance	None
deps	DepsT	The dependencies instance	None
inputs	InputT	The input data for the graph	None
span	AbstractContextManager[AbstractSpan] | None	Optional span for tracing/instrumentation	None
infer_name	bool	Whether to infer the graph name from the calling frame.	True

Returns:

Type	Description
OutputT	The final output from the graph execution

Source code in pydantic_graph/pydantic_graph/graph_builder.py

async def run(
    self,
    *,
    state: StateT = None,
    deps: DepsT = None,
    inputs: InputT = None,
    span: AbstractContextManager[AbstractSpan] | None = None,
    infer_name: bool = True,
) -> OutputT:
    """Execute the graph and return the final output.

    This is the main entry point for graph execution. It runs the graph
    to completion and returns the final output value.

    Args:
        state: The graph state instance
        deps: The dependencies instance
        inputs: The input data for the graph
        span: Optional span for tracing/instrumentation
        infer_name: Whether to infer the graph name from the calling frame.

    Returns:
        The final output from the graph execution
    """
    if infer_name and self.name is None:
        inferred_name = infer_obj_name(self, depth=2)
        if inferred_name is not None:  # pragma: no branch
            self.name = inferred_name

    async with self.iter(state=state, deps=deps, inputs=inputs, span=span, infer_name=False) as graph_run:
        # Note: This would probably be better using `async for _ in graph_run`, but this tests the `next` method,
        # which I'm less confident will be implemented correctly if not used on the critical path. We can change it
        # once we have tests, etc.
        event: Any = None
        while True:
            try:
                event = await graph_run.next(event)
            except StopAsyncIteration:
                assert isinstance(event, EndMarker), 'Graph run should end with an EndMarker.'
                return cast(EndMarker[OutputT], event).value

run_sync

run_sync(
    *,
    state: StateT = None,
    deps: DepsT = None,
    inputs: InputT = None,
    span: (
        AbstractContextManager[AbstractSpan] | None
    ) = None,
    infer_name: bool = True
) -> OutputT

Synchronously execute the graph and return the final output.

This is a convenience wrapper around [run][pydantic_graph.Graph.run] that runs the coroutine on the current event loop via loop.run_until_complete(...). As such, it cannot be called from inside async code or when an event loop is already running.

Parameters:

Name	Type	Description	Default
state	StateT	The graph state instance	None
deps	DepsT	The dependencies instance	None
inputs	InputT	The input data for the graph	None
span	AbstractContextManager[AbstractSpan] | None	Optional span for tracing/instrumentation	None
infer_name	bool	Whether to infer the graph name from the calling frame.	True

Returns:

Type	Description
OutputT	The final output from the graph execution

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def run_sync(
    self,
    *,
    state: StateT = None,
    deps: DepsT = None,
    inputs: InputT = None,
    span: AbstractContextManager[AbstractSpan] | None = None,
    infer_name: bool = True,
) -> OutputT:
    """Synchronously execute the graph and return the final output.

    This is a convenience wrapper around [`run`][pydantic_graph.Graph.run] that runs the coroutine on the
    current event loop via `loop.run_until_complete(...)`. As such, it cannot be called from inside async
    code or when an event loop is already running.

    Args:
        state: The graph state instance
        deps: The dependencies instance
        inputs: The input data for the graph
        span: Optional span for tracing/instrumentation
        infer_name: Whether to infer the graph name from the calling frame.

    Returns:
        The final output from the graph execution
    """
    if infer_name and self.name is None:
        inferred_name = infer_obj_name(self, depth=2)
        if inferred_name is not None:  # pragma: no branch
            self.name = inferred_name
    return _utils.get_event_loop().run_until_complete(
        self.run(state=state, deps=deps, inputs=inputs, span=span, infer_name=False)
    )

iter async

iter(
    *,
    state: StateT = None,
    deps: DepsT = None,
    inputs: InputT = None,
    span: (
        AbstractContextManager[AbstractSpan] | None
    ) = None,
    infer_name: bool = True
) -> AsyncIterator[GraphRun[StateT, DepsT, OutputT]]

Create an iterator for step-by-step graph execution.

This method allows for more fine-grained control over graph execution, enabling inspection of intermediate states and results.

Parameters:

Name	Type	Description	Default
state	StateT	The graph state instance	None
deps	DepsT	The dependencies instance	None
inputs	InputT	The input data for the graph	None
span	AbstractContextManager[AbstractSpan] | None	Optional span for tracing/instrumentation	None
infer_name	bool	Whether to infer the graph name from the calling frame.	True

Yields:

Type	Description
AsyncIterator[GraphRun[StateT, DepsT, OutputT]]	A GraphRun instance that can be iterated for step-by-step execution

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@asynccontextmanager
async def iter(
    self,
    *,
    state: StateT = None,
    deps: DepsT = None,
    inputs: InputT = None,
    span: AbstractContextManager[AbstractSpan] | None = None,
    infer_name: bool = True,
) -> AsyncIterator[GraphRun[StateT, DepsT, OutputT]]:
    """Create an iterator for step-by-step graph execution.

    This method allows for more fine-grained control over graph execution,
    enabling inspection of intermediate states and results.

    Args:
        state: The graph state instance
        deps: The dependencies instance
        inputs: The input data for the graph
        span: Optional span for tracing/instrumentation
        infer_name: Whether to infer the graph name from the calling frame.

    Yields:
        A GraphRun instance that can be iterated for step-by-step execution
    """
    if infer_name and self.name is None:
        inferred_name = infer_obj_name(self, depth=3)  # depth=3 because asynccontextmanager adds one
        if inferred_name is not None:  # pragma: no branch
            self.name = inferred_name

    with ExitStack() as stack:
        entered_span: AbstractSpan | None = None
        if span is None:
            if self.auto_instrument:
                entered_span = stack.enter_context(logfire_span('run graph {graph.name}', graph=self))
        else:
            entered_span = stack.enter_context(span)  # pragma: lax no cover
        traceparent = None if entered_span is None else get_traceparent(entered_span)
        async with GraphRun[StateT, DepsT, OutputT](
            graph=self,
            state=state,
            deps=deps,
            inputs=inputs,
            traceparent=traceparent,
        ) as graph_run:
            yield graph_run

render

render(
    *,
    title: str | None = None,
    direction: StateDiagramDirection | None = None
) -> str

Render the graph as a Mermaid diagram string.

Parameters:

Name	Type	Description	Default
title	str | None	Optional title for the diagram	None
direction	StateDiagramDirection | None	Optional direction for the diagram layout	None

Returns:

Type	Description
str	A string containing the Mermaid diagram representation

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def render(self, *, title: str | None = None, direction: StateDiagramDirection | None = None) -> str:
    """Render the graph as a Mermaid diagram string.

    Args:
        title: Optional title for the diagram
        direction: Optional direction for the diagram layout

    Returns:
        A string containing the Mermaid diagram representation
    """
    return build_mermaid_graph(self.nodes, self.edges_by_source).render(title=title, direction=direction)

__str__

__str__() -> str

Return a Mermaid diagram representation of the graph.

Returns:

Type	Description
str	A string containing the Mermaid diagram of the graph

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def __str__(self) -> str:
    """Return a Mermaid diagram representation of the graph.

    Returns:
        A string containing the Mermaid diagram of the graph
    """
    return self.render()

GraphTaskRequest dataclass

A request to run a task representing the execution of a node in the graph.

GraphTaskRequest encapsulates all the information needed to execute a specific node, including its inputs and the fork context it's executing within.

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@dataclass
class GraphTaskRequest:
    """A request to run a task representing the execution of a node in the graph.

    GraphTaskRequest encapsulates all the information needed to execute a specific
    node, including its inputs and the fork context it's executing within.
    """

    node_id: NodeID
    """The ID of the node to execute."""

    inputs: Any
    """The input data for the node."""

    fork_stack: ForkStack = field(repr=False)
    """Stack of forks that have been entered.

    Used by the GraphRun to decide when to proceed through joins.
    """

node_id instance-attribute

node_id: NodeID

The ID of the node to execute.

inputs instance-attribute

inputs: Any

The input data for the node.

fork_stack class-attribute instance-attribute

fork_stack: ForkStack = field(repr=False)

Stack of forks that have been entered.

Used by the GraphRun to decide when to proceed through joins.

GraphTask dataclass

Bases: GraphTaskRequest

A task representing the execution of a node in the graph.

GraphTask encapsulates all the information needed to execute a specific node, including its inputs and the fork context it's executing within, and has a unique ID to identify the task within the graph run.

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@dataclass
class GraphTask(GraphTaskRequest):
    """A task representing the execution of a node in the graph.

    GraphTask encapsulates all the information needed to execute a specific
    node, including its inputs and the fork context it's executing within,
    and has a unique ID to identify the task within the graph run.
    """

    task_id: TaskID = field(repr=False)
    """Unique identifier for this task."""

    @staticmethod
    def from_request(request: GraphTaskRequest, get_task_id: Callable[[], TaskID]) -> GraphTask:
        # Don't call the get_task_id callable, this is already a task
        if isinstance(request, GraphTask):
            return request
        return GraphTask(request.node_id, request.inputs, request.fork_stack, get_task_id())

task_id class-attribute instance-attribute

task_id: TaskID = field(repr=False)

Unique identifier for this task.

GraphRun

Bases: Generic[StateT, DepsT, OutputT]

A single execution instance of a graph.

GraphRun manages the execution state for a single run of a graph, including task scheduling, fork/join coordination, and result tracking.

Class Type Parameters:

Name	Bound or Constraints	Description	Default
StateT		The type of the graph state	required
DepsT		The type of the dependencies	required
OutputT		The type of the output data	required

Source code in pydantic_graph/pydantic_graph/graph_builder.py

class GraphRun(Generic[StateT, DepsT, OutputT]):
    """A single execution instance of a graph.

    GraphRun manages the execution state for a single run of a graph,
    including task scheduling, fork/join coordination, and result tracking.

    Type Parameters:
        StateT: The type of the graph state
        DepsT: The type of the dependencies
        OutputT: The type of the output data
    """

    def __init__(
        self,
        graph: Graph[StateT, DepsT, InputT, OutputT],
        *,
        state: StateT,
        deps: DepsT,
        inputs: InputT,
        traceparent: str | None,
    ):
        """Initialize a graph run.

        Args:
            graph: The graph to execute
            state: The graph state instance
            deps: The dependencies instance
            inputs: The input data for the graph
            traceparent: Optional trace parent for instrumentation
        """
        self.graph = graph
        """The graph being executed."""

        self.state = state
        """The graph state instance."""

        self.deps = deps
        """The dependencies instance."""

        self.inputs = inputs
        """The initial input data."""

        self._active_reducers: dict[tuple[JoinID, NodeRunID], JoinState] = {}
        """Active reducers for join operations."""

        self._next: EndMarker[OutputT] | ErrorMarker | Sequence[GraphTask] | None = None
        """The next item to be processed."""

        self._next_task_id = 0
        self._next_node_run_id = 0
        initial_fork_stack: ForkStack = (ForkStackItem(StartNode.id, self._get_next_node_run_id(), 0),)
        self._first_task = GraphTask(
            node_id=StartNode.id, inputs=inputs, fork_stack=initial_fork_stack, task_id=self._get_next_task_id()
        )
        self._iterator_task_group = create_task_group()
        self._iterator_instance = _GraphIterator[StateT, DepsT, OutputT](
            self.graph,
            self.state,
            self.deps,
            self._iterator_task_group,
            self._get_next_node_run_id,
            self._get_next_task_id,
        )
        self._iterator = self._iterator_instance.iter_graph(self._first_task)

        self.__traceparent = traceparent
        self._async_exit_stack = AsyncExitStack()

    async def __aenter__(self):
        self._async_exit_stack.enter_context(_unwrap_exception_groups())
        await self._async_exit_stack.enter_async_context(self._iterator_task_group)
        await self._async_exit_stack.enter_async_context(self._iterator_context())
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any):
        await self._async_exit_stack.__aexit__(exc_type, exc_val, exc_tb)

    @asynccontextmanager
    async def _iterator_context(self):
        try:
            yield
        finally:
            self._iterator_instance.iter_stream_sender.close()
            self._iterator_instance.iter_stream_receiver.close()
            await self._iterator.aclose()

    @overload
    def _traceparent(self, *, required: Literal[False]) -> str | None: ...
    @overload
    def _traceparent(self) -> str: ...
    def _traceparent(self, *, required: bool = True) -> str | None:
        """Get the trace parent for instrumentation.

        Args:
            required: Whether to raise an error if no traceparent exists

        Returns:
            The traceparent string, or None if not required and not set

        Raises:
            GraphRuntimeError: If required is True and no traceparent exists
        """
        if self.__traceparent is None and required:  # pragma: no cover
            raise exceptions.GraphRuntimeError('No span was created for this graph run')
        return self.__traceparent

    def __aiter__(self) -> AsyncIterator[EndMarker[OutputT] | Sequence[GraphTask]]:
        """Return self as an async iterator.

        Returns:
            Self for async iteration
        """
        return self

    async def __anext__(self) -> EndMarker[OutputT] | Sequence[GraphTask]:
        """Get the next item in the async iteration.

        Returns:
            The next execution result from the graph

        Raises:
            Exception: If a node raised an error and the caller has not recovered via
                `override_next()`.
        """
        if self._next is None:
            self._next = await anext(self._iterator)
        else:
            self._next = await self._iterator.asend(self._next)
        if isinstance(self._next, ErrorMarker):
            # A node raised an error. Store it so the caller can recover via
            # override_next() before the next __anext__ call re-raises.
            raise self._next.error
        return self._next

    async def next(
        self, value: EndMarker[OutputT] | Sequence[GraphTaskRequest] | None = None
    ) -> EndMarker[OutputT] | Sequence[GraphTask]:
        """Advance the graph execution by one step.

        This method allows for sending a value to the iterator, which is useful
        for resuming iteration or overriding intermediate results.

        Args:
            value: Optional value to send to the iterator

        Returns:
            The next execution result: either an EndMarker, or sequence of GraphTasks
        """
        if self._next is None:
            # Prevent `TypeError: can't send non-None value to a just-started async generator`
            # if `next` is called before the `first_node` has run.
            await anext(self)
        if value is not None:
            self._set_next(value)
        return await anext(self)

    def override_next(self, value: Sequence[GraphTaskRequest] | EndMarker[OutputT]) -> None:
        """Override the next pending step, allowing the graph to continue after an `End` or error.

        This is used by hook systems (like `after_node_run` or `on_node_run_error`) to redirect
        the graph to a new node when the current step produced an `End` result or raised an error,
        or to signal early completion by passing an `EndMarker`.

        Must only be called between iterations (not while an iteration is in flight).

        Args:
            value: New task requests to execute next, or an `EndMarker` to signal completion.
        """
        self._set_next(value)

    def _set_next(self, value: Sequence[GraphTaskRequest] | EndMarker[OutputT]) -> None:
        if isinstance(value, EndMarker):
            self._next = value
        else:
            self._next = [GraphTask.from_request(gtr, self._get_next_task_id) for gtr in value]

    @property
    def next_task(self) -> EndMarker[OutputT] | ErrorMarker | Sequence[GraphTask]:
        """Get the next task(s) to be executed.

        Returns:
            The next execution item, or the initial task if none is set
        """
        return self._next or [self._first_task]

    @property
    def output(self) -> OutputT | None:
        """Get the final output if the graph has completed.

        Returns:
            The output value if execution is complete, None otherwise
        """
        if isinstance(self._next, EndMarker):
            return self._next.value
        return None

    def _get_next_task_id(self) -> TaskID:
        next_id = TaskID(f'task:{self._next_task_id}')
        self._next_task_id += 1
        return next_id

    def _get_next_node_run_id(self) -> NodeRunID:
        next_id = NodeRunID(f'task:{self._next_node_run_id}')
        self._next_node_run_id += 1
        return next_id

__init__

__init__(
    graph: Graph[StateT, DepsT, InputT, OutputT],
    *,
    state: StateT,
    deps: DepsT,
    inputs: InputT,
    traceparent: str | None
)

Initialize a graph run.

Parameters:

Name	Type	Description	Default
graph	Graph[StateT, DepsT, InputT, OutputT]	The graph to execute	required
state	StateT	The graph state instance	required
deps	DepsT	The dependencies instance	required
inputs	InputT	The input data for the graph	required
traceparent	str | None	Optional trace parent for instrumentation	required

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def __init__(
    self,
    graph: Graph[StateT, DepsT, InputT, OutputT],
    *,
    state: StateT,
    deps: DepsT,
    inputs: InputT,
    traceparent: str | None,
):
    """Initialize a graph run.

    Args:
        graph: The graph to execute
        state: The graph state instance
        deps: The dependencies instance
        inputs: The input data for the graph
        traceparent: Optional trace parent for instrumentation
    """
    self.graph = graph
    """The graph being executed."""

    self.state = state
    """The graph state instance."""

    self.deps = deps
    """The dependencies instance."""

    self.inputs = inputs
    """The initial input data."""

    self._active_reducers: dict[tuple[JoinID, NodeRunID], JoinState] = {}
    """Active reducers for join operations."""

    self._next: EndMarker[OutputT] | ErrorMarker | Sequence[GraphTask] | None = None
    """The next item to be processed."""

    self._next_task_id = 0
    self._next_node_run_id = 0
    initial_fork_stack: ForkStack = (ForkStackItem(StartNode.id, self._get_next_node_run_id(), 0),)
    self._first_task = GraphTask(
        node_id=StartNode.id, inputs=inputs, fork_stack=initial_fork_stack, task_id=self._get_next_task_id()
    )
    self._iterator_task_group = create_task_group()
    self._iterator_instance = _GraphIterator[StateT, DepsT, OutputT](
        self.graph,
        self.state,
        self.deps,
        self._iterator_task_group,
        self._get_next_node_run_id,
        self._get_next_task_id,
    )
    self._iterator = self._iterator_instance.iter_graph(self._first_task)

    self.__traceparent = traceparent
    self._async_exit_stack = AsyncExitStack()

graph instance-attribute

graph = graph

The graph being executed.

state instance-attribute

state = state

The graph state instance.

deps instance-attribute

deps = deps

The dependencies instance.

inputs instance-attribute

inputs = inputs

The initial input data.

__aiter__

__aiter__() -> (
    AsyncIterator[EndMarker[OutputT] | Sequence[GraphTask]]
)

Return self as an async iterator.

Returns:

Type	Description
AsyncIterator[EndMarker[OutputT] | Sequence[GraphTask]]	Self for async iteration

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def __aiter__(self) -> AsyncIterator[EndMarker[OutputT] | Sequence[GraphTask]]:
    """Return self as an async iterator.

    Returns:
        Self for async iteration
    """
    return self

__anext__ async

__anext__() -> EndMarker[OutputT] | Sequence[GraphTask]

Get the next item in the async iteration.

Returns:

Type	Description
EndMarker[OutputT] | Sequence[GraphTask]	The next execution result from the graph

Raises:

Type	Description
Exception	If a node raised an error and the caller has not recovered via override_next().

Source code in pydantic_graph/pydantic_graph/graph_builder.py

async def __anext__(self) -> EndMarker[OutputT] | Sequence[GraphTask]:
    """Get the next item in the async iteration.

    Returns:
        The next execution result from the graph

    Raises:
        Exception: If a node raised an error and the caller has not recovered via
            `override_next()`.
    """
    if self._next is None:
        self._next = await anext(self._iterator)
    else:
        self._next = await self._iterator.asend(self._next)
    if isinstance(self._next, ErrorMarker):
        # A node raised an error. Store it so the caller can recover via
        # override_next() before the next __anext__ call re-raises.
        raise self._next.error
    return self._next

next async

next(
    value: (
        EndMarker[OutputT]
        | Sequence[GraphTaskRequest]
        | None
    ) = None,
) -> EndMarker[OutputT] | Sequence[GraphTask]

Advance the graph execution by one step.

This method allows for sending a value to the iterator, which is useful for resuming iteration or overriding intermediate results.

Parameters:

Name	Type	Description	Default
value	EndMarker[OutputT] | Sequence[GraphTaskRequest] | None	Optional value to send to the iterator	None

Returns:

Type	Description
EndMarker[OutputT] | Sequence[GraphTask]	The next execution result: either an EndMarker, or sequence of GraphTasks

Source code in pydantic_graph/pydantic_graph/graph_builder.py

async def next(
    self, value: EndMarker[OutputT] | Sequence[GraphTaskRequest] | None = None
) -> EndMarker[OutputT] | Sequence[GraphTask]:
    """Advance the graph execution by one step.

    This method allows for sending a value to the iterator, which is useful
    for resuming iteration or overriding intermediate results.

    Args:
        value: Optional value to send to the iterator

    Returns:
        The next execution result: either an EndMarker, or sequence of GraphTasks
    """
    if self._next is None:
        # Prevent `TypeError: can't send non-None value to a just-started async generator`
        # if `next` is called before the `first_node` has run.
        await anext(self)
    if value is not None:
        self._set_next(value)
    return await anext(self)

override_next

override_next(
    value: Sequence[GraphTaskRequest] | EndMarker[OutputT],
) -> None

Override the next pending step, allowing the graph to continue after an End or error.

This is used by hook systems (like after_node_run or on_node_run_error) to redirect the graph to a new node when the current step produced an End result or raised an error, or to signal early completion by passing an EndMarker.

Must only be called between iterations (not while an iteration is in flight).

Parameters:

Name	Type	Description	Default
value	Sequence[GraphTaskRequest] | EndMarker[OutputT]	New task requests to execute next, or an EndMarker to signal completion.	required

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def override_next(self, value: Sequence[GraphTaskRequest] | EndMarker[OutputT]) -> None:
    """Override the next pending step, allowing the graph to continue after an `End` or error.

    This is used by hook systems (like `after_node_run` or `on_node_run_error`) to redirect
    the graph to a new node when the current step produced an `End` result or raised an error,
    or to signal early completion by passing an `EndMarker`.

    Must only be called between iterations (not while an iteration is in flight).

    Args:
        value: New task requests to execute next, or an `EndMarker` to signal completion.
    """
    self._set_next(value)

next_task property

next_task: (
    EndMarker[OutputT] | ErrorMarker | Sequence[GraphTask]
)

Get the next task(s) to be executed.

Returns:

Type	Description
EndMarker[OutputT] | ErrorMarker | Sequence[GraphTask]	The next execution item, or the initial task if none is set

output property

output: OutputT | None

Get the final output if the graph has completed.

Returns:

Type	Description
OutputT | None	The output value if execution is complete, None otherwise

GraphBuilder dataclass

Bases: Generic[StateT, DepsT, GraphInputT, GraphOutputT]

A builder for constructing executable graph definitions.

GraphBuilder provides a fluent interface for defining nodes, edges, and routing in a graph workflow. It supports typed state, dependencies, and input/output validation.

Class Type Parameters:

Name	Bound or Constraints	Description	Default
StateT		The type of the graph state	required
DepsT		The type of the dependencies	required
GraphInputT		The type of the graph input data	required
GraphOutputT		The type of the graph output data	required

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@dataclass(init=False)
class GraphBuilder(Generic[StateT, DepsT, GraphInputT, GraphOutputT]):
    """A builder for constructing executable graph definitions.

    GraphBuilder provides a fluent interface for defining nodes, edges, and
    routing in a graph workflow. It supports typed state, dependencies, and
    input/output validation.

    Type Parameters:
        StateT: The type of the graph state
        DepsT: The type of the dependencies
        GraphInputT: The type of the graph input data
        GraphOutputT: The type of the graph output data
    """

    name: str | None
    """Optional name for the graph, if not provided the name will be inferred from the calling frame on the first call to a graph method."""

    state_type: TypeOrTypeExpression[StateT]
    """The type of the graph state."""

    deps_type: TypeOrTypeExpression[DepsT]
    """The type of the dependencies."""

    input_type: TypeOrTypeExpression[GraphInputT]
    """The type of the graph input data."""

    output_type: TypeOrTypeExpression[GraphOutputT]
    """The type of the graph output data."""

    auto_instrument: bool
    """Whether to automatically create instrumentation spans."""

    _nodes: dict[NodeID, AnyNode]
    """Internal storage for nodes in the graph."""

    _edges_by_source: dict[NodeID, list[Path]]
    """Internal storage for edges by source node."""

    _decision_index: int
    """Counter for generating unique decision node IDs."""

    Source = TypeAliasType('Source', SourceNode[StateT, DepsT, OutputT], type_params=(OutputT,))
    Destination = TypeAliasType('Destination', DestinationNode[StateT, DepsT, InputT], type_params=(InputT,))

    def __init__(
        self,
        *,
        name: str | None = None,
        state_type: TypeOrTypeExpression[StateT] = NoneType,
        deps_type: TypeOrTypeExpression[DepsT] = NoneType,
        input_type: TypeOrTypeExpression[GraphInputT] = NoneType,
        output_type: TypeOrTypeExpression[GraphOutputT] = NoneType,
        auto_instrument: bool = True,
    ):
        """Initialize a graph builder.

        Args:
            name: Optional name for the graph, if not provided the name will be inferred from the calling frame on the first call to a graph method.
            state_type: The type of the graph state
            deps_type: The type of the dependencies
            input_type: The type of the graph input data
            output_type: The type of the graph output data
            auto_instrument: Whether to automatically create instrumentation spans
        """
        self.name = name

        self.state_type = state_type
        self.deps_type = deps_type
        self.input_type = input_type
        self.output_type = output_type

        self.auto_instrument = auto_instrument

        self._nodes = {}
        self._edges_by_source = defaultdict(list)
        self._decision_index = 1

        self._start_node = StartNode[GraphInputT]()
        self._end_node = EndNode[GraphOutputT]()

    # Node building
    @property
    def start_node(self) -> StartNode[GraphInputT]:
        """Get the start node for the graph.

        Returns:
            The start node that receives the initial graph input
        """
        return self._start_node

    @property
    def end_node(self) -> EndNode[GraphOutputT]:
        """Get the end node for the graph.

        Returns:
            The end node that produces the final graph output
        """
        return self._end_node

    @overload
    def step(
        self,
        *,
        node_id: str | None = None,
        label: str | None = None,
    ) -> Callable[[StepFunction[StateT, DepsT, InputT, OutputT]], Step[StateT, DepsT, InputT, OutputT]]: ...
    @overload
    def step(
        self,
        call: StepFunction[StateT, DepsT, InputT, OutputT],
        *,
        node_id: str | None = None,
        label: str | None = None,
    ) -> Step[StateT, DepsT, InputT, OutputT]: ...
    def step(
        self,
        call: StepFunction[StateT, DepsT, InputT, OutputT] | None = None,
        *,
        node_id: str | None = None,
        label: str | None = None,
    ) -> (
        Step[StateT, DepsT, InputT, OutputT]
        | Callable[[StepFunction[StateT, DepsT, InputT, OutputT]], Step[StateT, DepsT, InputT, OutputT]]
    ):
        """Create a step from a step function.

        This method can be used as a decorator or called directly to create
        a step node from an async function.

        Args:
            call: The step function to wrap
            node_id: Optional ID for the node
            label: Optional human-readable label

        Returns:
            Either a Step instance or a decorator function
        """
        if call is None:

            def decorator(
                func: StepFunction[StateT, DepsT, InputT, OutputT],
            ) -> Step[StateT, DepsT, InputT, OutputT]:
                return self.step(call=func, node_id=node_id, label=label)

            return decorator

        node_id = node_id or get_callable_name(call)

        step = Step[StateT, DepsT, InputT, OutputT](id=NodeID(node_id), call=call, label=label)

        return step

    @overload
    def stream(
        self,
        *,
        node_id: str | None = None,
        label: str | None = None,
    ) -> Callable[
        [StreamFunction[StateT, DepsT, InputT, OutputT]], Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]
    ]: ...
    @overload
    def stream(
        self,
        call: StreamFunction[StateT, DepsT, InputT, OutputT],
        *,
        node_id: str | None = None,
        label: str | None = None,
    ) -> Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]: ...
    @overload
    def stream(
        self,
        call: StreamFunction[StateT, DepsT, InputT, OutputT] | None = None,
        *,
        node_id: str | None = None,
        label: str | None = None,
    ) -> (
        Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]
        | Callable[
            [StreamFunction[StateT, DepsT, InputT, OutputT]],
            Step[StateT, DepsT, InputT, AsyncIterable[OutputT]],
        ]
    ): ...
    def stream(
        self,
        call: StreamFunction[StateT, DepsT, InputT, OutputT] | None = None,
        *,
        node_id: str | None = None,
        label: str | None = None,
    ) -> (
        Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]
        | Callable[
            [StreamFunction[StateT, DepsT, InputT, OutputT]],
            Step[StateT, DepsT, InputT, AsyncIterable[OutputT]],
        ]
    ):
        """Create a step from an async iterator (which functions like a "stream").

        This method can be used as a decorator or called directly to create
        a step node from an async function.

        Args:
            call: The step function to wrap
            node_id: Optional ID for the node
            label: Optional human-readable label

        Returns:
            Either a Step instance or a decorator function
        """
        if call is None:

            def decorator(
                func: StreamFunction[StateT, DepsT, InputT, OutputT],
            ) -> Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]:
                return self.stream(call=func, node_id=node_id, label=label)

            return decorator

        # We need to wrap the call so that we can call `await` even though the result is an async iterator
        async def wrapper(ctx: StepContext[StateT, DepsT, InputT]):
            return call(ctx)

        node_id = node_id or get_callable_name(call)

        return self.step(call=wrapper, node_id=node_id, label=label)

    @overload
    def join(
        self,
        reducer: ReducerFunction[StateT, DepsT, InputT, OutputT],
        *,
        initial: OutputT,
        node_id: str | None = None,
        parent_fork_id: str | None = None,
        preferred_parent_fork: Literal['farthest', 'closest'] = 'farthest',
    ) -> Join[StateT, DepsT, InputT, OutputT]: ...
    @overload
    def join(
        self,
        reducer: ReducerFunction[StateT, DepsT, InputT, OutputT],
        *,
        initial_factory: Callable[[], OutputT],
        node_id: str | None = None,
        parent_fork_id: str | None = None,
        preferred_parent_fork: Literal['farthest', 'closest'] = 'farthest',
    ) -> Join[StateT, DepsT, InputT, OutputT]: ...

    def join(
        self,
        reducer: ReducerFunction[StateT, DepsT, InputT, OutputT],
        *,
        initial: OutputT | Unset = UNSET,
        initial_factory: Callable[[], OutputT] | Unset = UNSET,
        node_id: str | None = None,
        parent_fork_id: str | None = None,
        preferred_parent_fork: Literal['farthest', 'closest'] = 'farthest',
    ) -> Join[StateT, DepsT, InputT, OutputT]:
        if initial_factory is UNSET:
            initial_factory = lambda: initial  # pyright: ignore[reportAssignmentType]  # noqa: E731

        return Join[StateT, DepsT, InputT, OutputT](
            id=JoinID(NodeID(node_id or generate_placeholder_node_id(get_callable_name(reducer)))),
            reducer=reducer,
            initial_factory=cast(Callable[[], OutputT], initial_factory),
            parent_fork_id=ForkID(parent_fork_id) if parent_fork_id is not None else None,
            preferred_parent_fork=preferred_parent_fork,
        )

    # Edge building
    def add(self, *edges: EdgePath[StateT, DepsT]) -> None:  # noqa: C901
        """Add one or more edge paths to the graph.

        This method processes edge paths and automatically creates any necessary
        fork nodes for broadcasts and maps.

        Args:
            *edges: The edge paths to add to the graph
        """

        def _handle_path(p: Path):
            """Process a path and create necessary fork nodes.

            Args:
                p: The path to process
            """
            for item in p.items:
                if isinstance(item, BroadcastMarker):
                    new_node = Fork[Any, Any](id=item.fork_id, is_map=False, downstream_join_id=None)
                    self._insert_node(new_node)
                    for path in item.paths:
                        _handle_path(Path(items=[*path.items]))
                elif isinstance(item, MapMarker):
                    new_node = Fork[Any, Any](id=item.fork_id, is_map=True, downstream_join_id=item.downstream_join_id)
                    self._insert_node(new_node)
                elif isinstance(item, DestinationMarker):
                    pass

        def _handle_destination_node(d: AnyDestinationNode):
            if id(d) in destination_ids:
                return  # prevent infinite recursion if there is a cycle of decisions

            destination_ids.add(id(d))
            destinations.append(d)
            self._insert_node(d)
            if isinstance(d, Decision):
                for branch in d.branches:
                    _handle_path(branch.path)
                    for d2 in branch.destinations:
                        _handle_destination_node(d2)

        destination_ids = set[int]()
        destinations: list[AnyDestinationNode] = []
        for edge in edges:
            for source_node in edge.sources:
                self._insert_node(source_node)
                self._edges_by_source[source_node.id].append(edge.path)
            for destination_node in edge.destinations:
                _handle_destination_node(destination_node)
            _handle_path(edge.path)

        # Automatically create edges from step function return hints including `BaseNode`s
        for destination in destinations:
            if not isinstance(destination, Step) or isinstance(destination, NodeStep):
                continue
            parent_namespace = _utils.get_parent_namespace(inspect.currentframe())
            type_hints = get_type_hints(destination.call, localns=parent_namespace, include_extras=True)
            try:
                return_hint = type_hints['return']
            except KeyError:
                pass
            else:
                edge = self._edge_from_return_hint(destination, return_hint)
                if edge is not None:
                    self.add(edge)

    def add_edge(self, source: Source[T], destination: Destination[T], *, label: str | None = None) -> None:
        """Add a simple edge between two nodes.

        Args:
            source: The source node
            destination: The destination node
            label: Optional label for the edge
        """
        builder = self.edge_from(source)
        if label is not None:
            builder = builder.label(label)
        self.add(builder.to(destination))

    def add_mapping_edge(
        self,
        source: Source[Iterable[T]],
        map_to: Destination[T],
        *,
        pre_map_label: str | None = None,
        post_map_label: str | None = None,
        fork_id: ForkID | None = None,
        downstream_join_id: JoinID | None = None,
    ) -> None:
        """Add an edge that maps iterable data across parallel paths.

        Args:
            source: The source node that produces iterable data
            map_to: The destination node that receives individual items
            pre_map_label: Optional label before the map operation
            post_map_label: Optional label after the map operation
            fork_id: Optional ID for the fork node produced for this map operation
            downstream_join_id: Optional ID of a join node that will always be downstream of this map.
                Specifying this ensures correct handling if you try to map an empty iterable.
        """
        builder = self.edge_from(source)
        if pre_map_label is not None:
            builder = builder.label(pre_map_label)
        builder = builder.map(fork_id=fork_id, downstream_join_id=downstream_join_id)
        if post_map_label is not None:
            builder = builder.label(post_map_label)
        self.add(builder.to(map_to))

    # TODO(DavidM): Support adding subgraphs; I think this behaves like a step with the same inputs/outputs but gets rendered as a subgraph in mermaid

    def edge_from(self, *sources: Source[SourceOutputT]) -> EdgePathBuilder[StateT, DepsT, SourceOutputT]:
        """Create an edge path builder starting from the given source nodes.

        Args:
            *sources: The source nodes to start the edge path from

        Returns:
            An EdgePathBuilder for constructing the complete edge path
        """
        return EdgePathBuilder[StateT, DepsT, SourceOutputT](
            sources=sources, path_builder=PathBuilder(working_items=[])
        )

    def decision(self, *, note: str | None = None, node_id: str | None = None) -> Decision[StateT, DepsT, Never]:
        """Create a new decision node.

        Args:
            note: Optional note to describe the decision logic
            node_id: Optional ID for the node produced for this decision logic

        Returns:
            A new Decision node with no branches
        """
        return Decision(id=NodeID(node_id or generate_placeholder_node_id('decision')), branches=[], note=note)

    def match(
        self,
        source: TypeOrTypeExpression[SourceT],
        *,
        matches: Callable[[Any], bool] | None = None,
    ) -> DecisionBranchBuilder[StateT, DepsT, SourceT, SourceT, Never]:
        """Create a decision branch matcher.

        Args:
            source: The type or type expression to match against
            matches: Optional custom matching function

        Returns:
            A DecisionBranchBuilder for constructing the branch
        """
        # Note, the following node_id really is just a placeholder and shouldn't end up in the final graph
        # This is why we don't expose a way for end users to override the value used here.
        node_id = NodeID(generate_placeholder_node_id('match_decision'))
        decision = Decision[StateT, DepsT, Never](id=node_id, branches=[], note=None)
        new_path_builder = PathBuilder[StateT, DepsT, SourceT](working_items=[])
        return DecisionBranchBuilder(decision=decision, source=source, matches=matches, path_builder=new_path_builder)

    def match_node(
        self,
        source: type[SourceNodeT],
        *,
        matches: Callable[[Any], bool] | None = None,
    ) -> DecisionBranch[SourceNodeT]:
        """Create a decision branch for `BaseNode` subclasses.

        This is similar to `match()` but specifically designed for matching against `BaseNode` types.

        Args:
            source: The `BaseNode` subclass to match against
            matches: Optional custom matching function

        Returns:
            A `DecisionBranch` for the `BaseNode` type
        """
        node = NodeStep(source)
        path = Path(items=[DestinationMarker(node.id)])
        return DecisionBranch(source=source, matches=matches, path=path, destinations=[node])

    def node(
        self,
        node_type: type[BaseNode[StateT, DepsT, GraphOutputT]],
    ) -> EdgePath[StateT, DepsT]:
        """Create an edge path from a `BaseNode` class.

        This method integrates a `BaseNode` subclass into the builder graph by
        analyzing its `run` return type hints and creating appropriate edges.

        Args:
            node_type: The `BaseNode` subclass to integrate

        Returns:
            An `EdgePath` representing the node and its connections

        Raises:
            GraphSetupError: If the node type is missing required type hints
        """
        parent_namespace = _utils.get_parent_namespace(inspect.currentframe())
        type_hints = get_type_hints(node_type.run, localns=parent_namespace, include_extras=True)
        try:
            return_hint = type_hints['return']
        except KeyError as e:  # pragma: no cover
            raise exceptions.GraphSetupError(
                f'Node {node_type} is missing a return type hint on its `run` method'
            ) from e

        node = NodeStep(node_type)

        edge = self._edge_from_return_hint(node, return_hint)
        if not edge:  # pragma: no cover
            raise exceptions.GraphSetupError(f'Node {node_type} is missing a return type hint on its `run` method')

        return edge

    # Helpers
    def _insert_node(self, node: AnyNode) -> None:
        """Insert a node into the graph, checking for ID conflicts.

        Args:
            node: The node to insert

        Raises:
            ValueError: If a different node with the same ID already exists
        """
        existing = self._nodes.get(node.id)
        if existing is None:
            self._nodes[node.id] = node
        elif isinstance(existing, NodeStep) and isinstance(node, NodeStep) and existing.node_type is node.node_type:
            pass
        elif existing is not node:
            raise GraphBuildingError(
                f'All nodes must have unique node IDs. {node.id!r} was the ID for {existing} and {node}'
            )

    def _edge_from_return_hint(
        self, node: SourceNode[StateT, DepsT, Any], return_hint: TypeOrTypeExpression[Any]
    ) -> EdgePath[StateT, DepsT] | None:
        """Create edges from a return type hint.

        This method analyzes return type hints from step functions or node methods
        to automatically create appropriate edges in the graph.

        Args:
            node: The source node
            return_hint: The return type hint to analyze

        Returns:
            An EdgePath if edges can be inferred, None otherwise

        Raises:
            GraphSetupError: If the return type hint is invalid or incomplete
        """
        destinations: list[AnyDestinationNode] = []
        union_args = _utils.get_union_args(return_hint)
        for return_type in union_args:
            return_type, annotations = _utils.unpack_annotated(return_type)
            return_type_origin = get_origin(return_type) or return_type
            if return_type_origin is End:
                destinations.append(self.end_node)
            elif return_type_origin is BaseNode:
                raise exceptions.GraphSetupError(  # pragma: no cover
                    f'Node {node} return type hint includes a plain `BaseNode`. '
                    'Edge inference requires each possible returned `BaseNode` subclass to be listed explicitly.'
                )
            elif return_type_origin is StepNode:
                step = cast(
                    Step[StateT, DepsT, Any, Any] | None,
                    next((a for a in annotations if isinstance(a, Step)), None),  # pyright: ignore[reportUnknownArgumentType]
                )
                if step is None:
                    raise exceptions.GraphSetupError(  # pragma: no cover
                        f'Node {node} return type hint includes a `StepNode` without a `Step` annotation. '
                        'When returning `my_step.as_node()`, use `Annotated[StepNode[StateT, DepsT], my_step]` as the return type hint.'
                    )
                destinations.append(step)
            elif return_type_origin is JoinNode:
                join = cast(
                    Join[StateT, DepsT, Any, Any] | None,
                    next((a for a in annotations if isinstance(a, Join)), None),  # pyright: ignore[reportUnknownArgumentType]
                )
                if join is None:
                    raise exceptions.GraphSetupError(  # pragma: no cover
                        f'Node {node} return type hint includes a `JoinNode` without a `Join` annotation. '
                        'When returning `my_join.as_node()`, use `Annotated[JoinNode[StateT, DepsT], my_join]` as the return type hint.'
                    )
                destinations.append(join)
            elif inspect.isclass(return_type_origin) and issubclass(return_type_origin, BaseNode):
                destinations.append(NodeStep(return_type))

        if len(destinations) < len(union_args):
            # Only build edges if all the return types are nodes
            return None

        edge = self.edge_from(node)
        if len(destinations) == 1:
            return edge.to(destinations[0])
        else:
            decision = self.decision()
            for destination in destinations:
                # We don't actually use this decision mechanism, but we need to build the edges for parent-fork finding
                decision = decision.branch(self.match(NoneType).to(destination))
            return edge.to(decision)

    # Graph building
    def build(self, validate_graph_structure: bool = True) -> Graph[StateT, DepsT, GraphInputT, GraphOutputT]:
        """Build the final executable graph from the accumulated nodes and edges.

        This method performs validation, normalization, and analysis of the graph
        structure to create a complete, executable graph instance.

        Args:
            validate_graph_structure: whether to perform validation of the graph structure
                See the docstring of _validate_graph_structure below for more details.

        Returns:
            A complete Graph instance ready for execution

        Raises:
            ValueError: If the graph structure is invalid (e.g., join without parent fork)
        """
        nodes = self._nodes
        edges_by_source = self._edges_by_source

        nodes, edges_by_source = _replace_placeholder_node_ids(nodes, edges_by_source)
        nodes, edges_by_source = _flatten_paths(nodes, edges_by_source)
        nodes, edges_by_source = _normalize_forks(nodes, edges_by_source)
        if validate_graph_structure:
            _validate_graph_structure(nodes, edges_by_source)
        parent_forks = _collect_dominating_forks(nodes, edges_by_source)
        intermediate_join_nodes = _compute_intermediate_join_nodes(nodes, parent_forks)

        return Graph[StateT, DepsT, GraphInputT, GraphOutputT](
            name=self.name,
            state_type=unpack_type_expression(self.state_type),
            deps_type=unpack_type_expression(self.deps_type),
            input_type=unpack_type_expression(self.input_type),
            output_type=unpack_type_expression(self.output_type),
            nodes=nodes,
            edges_by_source=edges_by_source,
            parent_forks=parent_forks,
            intermediate_join_nodes=intermediate_join_nodes,
            auto_instrument=self.auto_instrument,
        )

__init__

__init__(
    *,
    name: str | None = None,
    state_type: TypeOrTypeExpression[StateT] = NoneType,
    deps_type: TypeOrTypeExpression[DepsT] = NoneType,
    input_type: TypeOrTypeExpression[
        GraphInputT
    ] = NoneType,
    output_type: TypeOrTypeExpression[
        GraphOutputT
    ] = NoneType,
    auto_instrument: bool = True
)

Initialize a graph builder.

Parameters:

Name	Type	Description	Default
name	str | None	Optional name for the graph, if not provided the name will be inferred from the calling frame on the first call to a graph method.	None
state_type	TypeOrTypeExpression[StateT]	The type of the graph state	NoneType
deps_type	TypeOrTypeExpression[DepsT]	The type of the dependencies	NoneType
input_type	TypeOrTypeExpression[GraphInputT]	The type of the graph input data	NoneType
output_type	TypeOrTypeExpression[GraphOutputT]	The type of the graph output data	NoneType
auto_instrument	bool	Whether to automatically create instrumentation spans	True

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def __init__(
    self,
    *,
    name: str | None = None,
    state_type: TypeOrTypeExpression[StateT] = NoneType,
    deps_type: TypeOrTypeExpression[DepsT] = NoneType,
    input_type: TypeOrTypeExpression[GraphInputT] = NoneType,
    output_type: TypeOrTypeExpression[GraphOutputT] = NoneType,
    auto_instrument: bool = True,
):
    """Initialize a graph builder.

    Args:
        name: Optional name for the graph, if not provided the name will be inferred from the calling frame on the first call to a graph method.
        state_type: The type of the graph state
        deps_type: The type of the dependencies
        input_type: The type of the graph input data
        output_type: The type of the graph output data
        auto_instrument: Whether to automatically create instrumentation spans
    """
    self.name = name

    self.state_type = state_type
    self.deps_type = deps_type
    self.input_type = input_type
    self.output_type = output_type

    self.auto_instrument = auto_instrument

    self._nodes = {}
    self._edges_by_source = defaultdict(list)
    self._decision_index = 1

    self._start_node = StartNode[GraphInputT]()
    self._end_node = EndNode[GraphOutputT]()

name instance-attribute

name: str | None = name

Optional name for the graph, if not provided the name will be inferred from the calling frame on the first call to a graph method.

state_type instance-attribute

state_type: TypeOrTypeExpression[StateT] = state_type

The type of the graph state.

deps_type instance-attribute

deps_type: TypeOrTypeExpression[DepsT] = deps_type

The type of the dependencies.

input_type instance-attribute

input_type: TypeOrTypeExpression[GraphInputT] = input_type

The type of the graph input data.

output_type instance-attribute

output_type: TypeOrTypeExpression[GraphOutputT] = (
    output_type
)

The type of the graph output data.

auto_instrument instance-attribute

auto_instrument: bool = auto_instrument

Whether to automatically create instrumentation spans.

start_node property

start_node: StartNode[GraphInputT]

Get the start node for the graph.

Returns:

Type	Description
StartNode[GraphInputT]	The start node that receives the initial graph input

end_node property

end_node: EndNode[GraphOutputT]

Get the end node for the graph.

Returns:

Type	Description
EndNode[GraphOutputT]	The end node that produces the final graph output

step

step(
    *, node_id: str | None = None, label: str | None = None
) -> Callable[
    [StepFunction[StateT, DepsT, InputT, OutputT]],
    Step[StateT, DepsT, InputT, OutputT],
]

step(
    call: StepFunction[StateT, DepsT, InputT, OutputT],
    *,
    node_id: str | None = None,
    label: str | None = None
) -> Step[StateT, DepsT, InputT, OutputT]

step(
    call: (
        StepFunction[StateT, DepsT, InputT, OutputT] | None
    ) = None,
    *,
    node_id: str | None = None,
    label: str | None = None
) -> (
    Step[StateT, DepsT, InputT, OutputT]
    | Callable[
        [StepFunction[StateT, DepsT, InputT, OutputT]],
        Step[StateT, DepsT, InputT, OutputT],
    ]
)

Create a step from a step function.

This method can be used as a decorator or called directly to create a step node from an async function.

Parameters:

Name	Type	Description	Default
call	StepFunction[StateT, DepsT, InputT, OutputT] | None	The step function to wrap	None
node_id	str | None	Optional ID for the node	None
label	str | None	Optional human-readable label	None

Returns:

Type	Description
Step[StateT, DepsT, InputT, OutputT] | Callable[[StepFunction[StateT, DepsT, InputT, OutputT]], Step[StateT, DepsT, InputT, OutputT]]	Either a Step instance or a decorator function

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def step(
    self,
    call: StepFunction[StateT, DepsT, InputT, OutputT] | None = None,
    *,
    node_id: str | None = None,
    label: str | None = None,
) -> (
    Step[StateT, DepsT, InputT, OutputT]
    | Callable[[StepFunction[StateT, DepsT, InputT, OutputT]], Step[StateT, DepsT, InputT, OutputT]]
):
    """Create a step from a step function.

    This method can be used as a decorator or called directly to create
    a step node from an async function.

    Args:
        call: The step function to wrap
        node_id: Optional ID for the node
        label: Optional human-readable label

    Returns:
        Either a Step instance or a decorator function
    """
    if call is None:

        def decorator(
            func: StepFunction[StateT, DepsT, InputT, OutputT],
        ) -> Step[StateT, DepsT, InputT, OutputT]:
            return self.step(call=func, node_id=node_id, label=label)

        return decorator

    node_id = node_id or get_callable_name(call)

    step = Step[StateT, DepsT, InputT, OutputT](id=NodeID(node_id), call=call, label=label)

    return step

stream

stream(
    *, node_id: str | None = None, label: str | None = None
) -> Callable[
    [StreamFunction[StateT, DepsT, InputT, OutputT]],
    Step[StateT, DepsT, InputT, AsyncIterable[OutputT]],
]

stream(
    call: StreamFunction[StateT, DepsT, InputT, OutputT],
    *,
    node_id: str | None = None,
    label: str | None = None
) -> Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]

stream(
    call: (
        StreamFunction[StateT, DepsT, InputT, OutputT]
        | None
    ) = None,
    *,
    node_id: str | None = None,
    label: str | None = None
) -> (
    Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]
    | Callable[
        [StreamFunction[StateT, DepsT, InputT, OutputT]],
        Step[StateT, DepsT, InputT, AsyncIterable[OutputT]],
    ]
)

stream(
    call: (
        StreamFunction[StateT, DepsT, InputT, OutputT]
        | None
    ) = None,
    *,
    node_id: str | None = None,
    label: str | None = None
) -> (
    Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]
    | Callable[
        [StreamFunction[StateT, DepsT, InputT, OutputT]],
        Step[StateT, DepsT, InputT, AsyncIterable[OutputT]],
    ]
)

Create a step from an async iterator (which functions like a "stream").

This method can be used as a decorator or called directly to create a step node from an async function.

Parameters:

Name	Type	Description	Default
call	StreamFunction[StateT, DepsT, InputT, OutputT] | None	The step function to wrap	None
node_id	str | None	Optional ID for the node	None
label	str | None	Optional human-readable label	None

Returns:

Type	Description
Step[StateT, DepsT, InputT, AsyncIterable[OutputT]] | Callable[[StreamFunction[StateT, DepsT, InputT, OutputT]], Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]]	Either a Step instance or a decorator function

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def stream(
    self,
    call: StreamFunction[StateT, DepsT, InputT, OutputT] | None = None,
    *,
    node_id: str | None = None,
    label: str | None = None,
) -> (
    Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]
    | Callable[
        [StreamFunction[StateT, DepsT, InputT, OutputT]],
        Step[StateT, DepsT, InputT, AsyncIterable[OutputT]],
    ]
):
    """Create a step from an async iterator (which functions like a "stream").

    This method can be used as a decorator or called directly to create
    a step node from an async function.

    Args:
        call: The step function to wrap
        node_id: Optional ID for the node
        label: Optional human-readable label

    Returns:
        Either a Step instance or a decorator function
    """
    if call is None:

        def decorator(
            func: StreamFunction[StateT, DepsT, InputT, OutputT],
        ) -> Step[StateT, DepsT, InputT, AsyncIterable[OutputT]]:
            return self.stream(call=func, node_id=node_id, label=label)

        return decorator

    # We need to wrap the call so that we can call `await` even though the result is an async iterator
    async def wrapper(ctx: StepContext[StateT, DepsT, InputT]):
        return call(ctx)

    node_id = node_id or get_callable_name(call)

    return self.step(call=wrapper, node_id=node_id, label=label)

add

add(*edges: EdgePath[StateT, DepsT]) -> None

Add one or more edge paths to the graph.

This method processes edge paths and automatically creates any necessary fork nodes for broadcasts and maps.

Parameters:

Name	Type	Description	Default
*edges	EdgePath[StateT, DepsT]	The edge paths to add to the graph	()

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def add(self, *edges: EdgePath[StateT, DepsT]) -> None:  # noqa: C901
    """Add one or more edge paths to the graph.

    This method processes edge paths and automatically creates any necessary
    fork nodes for broadcasts and maps.

    Args:
        *edges: The edge paths to add to the graph
    """

    def _handle_path(p: Path):
        """Process a path and create necessary fork nodes.

        Args:
            p: The path to process
        """
        for item in p.items:
            if isinstance(item, BroadcastMarker):
                new_node = Fork[Any, Any](id=item.fork_id, is_map=False, downstream_join_id=None)
                self._insert_node(new_node)
                for path in item.paths:
                    _handle_path(Path(items=[*path.items]))
            elif isinstance(item, MapMarker):
                new_node = Fork[Any, Any](id=item.fork_id, is_map=True, downstream_join_id=item.downstream_join_id)
                self._insert_node(new_node)
            elif isinstance(item, DestinationMarker):
                pass

    def _handle_destination_node(d: AnyDestinationNode):
        if id(d) in destination_ids:
            return  # prevent infinite recursion if there is a cycle of decisions

        destination_ids.add(id(d))
        destinations.append(d)
        self._insert_node(d)
        if isinstance(d, Decision):
            for branch in d.branches:
                _handle_path(branch.path)
                for d2 in branch.destinations:
                    _handle_destination_node(d2)

    destination_ids = set[int]()
    destinations: list[AnyDestinationNode] = []
    for edge in edges:
        for source_node in edge.sources:
            self._insert_node(source_node)
            self._edges_by_source[source_node.id].append(edge.path)
        for destination_node in edge.destinations:
            _handle_destination_node(destination_node)
        _handle_path(edge.path)

    # Automatically create edges from step function return hints including `BaseNode`s
    for destination in destinations:
        if not isinstance(destination, Step) or isinstance(destination, NodeStep):
            continue
        parent_namespace = _utils.get_parent_namespace(inspect.currentframe())
        type_hints = get_type_hints(destination.call, localns=parent_namespace, include_extras=True)
        try:
            return_hint = type_hints['return']
        except KeyError:
            pass
        else:
            edge = self._edge_from_return_hint(destination, return_hint)
            if edge is not None:
                self.add(edge)

add_edge

add_edge(
    source: Source[T],
    destination: Destination[T],
    *,
    label: str | None = None
) -> None

Add a simple edge between two nodes.

Parameters:

Name	Type	Description	Default
source	Source[T]	The source node	required
destination	Destination[T]	The destination node	required
label	str | None	Optional label for the edge	None

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def add_edge(self, source: Source[T], destination: Destination[T], *, label: str | None = None) -> None:
    """Add a simple edge between two nodes.

    Args:
        source: The source node
        destination: The destination node
        label: Optional label for the edge
    """
    builder = self.edge_from(source)
    if label is not None:
        builder = builder.label(label)
    self.add(builder.to(destination))

add_mapping_edge

add_mapping_edge(
    source: Source[Iterable[T]],
    map_to: Destination[T],
    *,
    pre_map_label: str | None = None,
    post_map_label: str | None = None,
    fork_id: ForkID | None = None,
    downstream_join_id: JoinID | None = None
) -> None

Add an edge that maps iterable data across parallel paths.

Parameters:

Name	Type	Description	Default
source	Source[Iterable[T]]	The source node that produces iterable data	required
map_to	Destination[T]	The destination node that receives individual items	required
pre_map_label	str | None	Optional label before the map operation	None
post_map_label	str | None	Optional label after the map operation	None
fork_id	ForkID | None	Optional ID for the fork node produced for this map operation	None
downstream_join_id	JoinID | None	Optional ID of a join node that will always be downstream of this map. Specifying this ensures correct handling if you try to map an empty iterable.	None

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def add_mapping_edge(
    self,
    source: Source[Iterable[T]],
    map_to: Destination[T],
    *,
    pre_map_label: str | None = None,
    post_map_label: str | None = None,
    fork_id: ForkID | None = None,
    downstream_join_id: JoinID | None = None,
) -> None:
    """Add an edge that maps iterable data across parallel paths.

    Args:
        source: The source node that produces iterable data
        map_to: The destination node that receives individual items
        pre_map_label: Optional label before the map operation
        post_map_label: Optional label after the map operation
        fork_id: Optional ID for the fork node produced for this map operation
        downstream_join_id: Optional ID of a join node that will always be downstream of this map.
            Specifying this ensures correct handling if you try to map an empty iterable.
    """
    builder = self.edge_from(source)
    if pre_map_label is not None:
        builder = builder.label(pre_map_label)
    builder = builder.map(fork_id=fork_id, downstream_join_id=downstream_join_id)
    if post_map_label is not None:
        builder = builder.label(post_map_label)
    self.add(builder.to(map_to))

edge_from

edge_from(
    *sources: Source[SourceOutputT],
) -> EdgePathBuilder[StateT, DepsT, SourceOutputT]

Create an edge path builder starting from the given source nodes.

Parameters:

Name	Type	Description	Default
*sources	Source[SourceOutputT]	The source nodes to start the edge path from	()

Returns:

Type	Description
EdgePathBuilder[StateT, DepsT, SourceOutputT]	An EdgePathBuilder for constructing the complete edge path

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def edge_from(self, *sources: Source[SourceOutputT]) -> EdgePathBuilder[StateT, DepsT, SourceOutputT]:
    """Create an edge path builder starting from the given source nodes.

    Args:
        *sources: The source nodes to start the edge path from

    Returns:
        An EdgePathBuilder for constructing the complete edge path
    """
    return EdgePathBuilder[StateT, DepsT, SourceOutputT](
        sources=sources, path_builder=PathBuilder(working_items=[])
    )

decision

decision(
    *, note: str | None = None, node_id: str | None = None
) -> Decision[StateT, DepsT, Never]

Create a new decision node.

Parameters:

Name	Type	Description	Default
note	str | None	Optional note to describe the decision logic	None
node_id	str | None	Optional ID for the node produced for this decision logic	None

Returns:

Type	Description
Decision[StateT, DepsT, Never]	A new Decision node with no branches

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def decision(self, *, note: str | None = None, node_id: str | None = None) -> Decision[StateT, DepsT, Never]:
    """Create a new decision node.

    Args:
        note: Optional note to describe the decision logic
        node_id: Optional ID for the node produced for this decision logic

    Returns:
        A new Decision node with no branches
    """
    return Decision(id=NodeID(node_id or generate_placeholder_node_id('decision')), branches=[], note=note)

match

match(
    source: TypeOrTypeExpression[SourceT],
    *,
    matches: Callable[[Any], bool] | None = None
) -> DecisionBranchBuilder[
    StateT, DepsT, SourceT, SourceT, Never
]

Create a decision branch matcher.

Parameters:

Name	Type	Description	Default
source	TypeOrTypeExpression[SourceT]	The type or type expression to match against	required
matches	Callable[[Any], bool] | None	Optional custom matching function	None

Returns:

Type	Description
DecisionBranchBuilder[StateT, DepsT, SourceT, SourceT, Never]	A DecisionBranchBuilder for constructing the branch

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def match(
    self,
    source: TypeOrTypeExpression[SourceT],
    *,
    matches: Callable[[Any], bool] | None = None,
) -> DecisionBranchBuilder[StateT, DepsT, SourceT, SourceT, Never]:
    """Create a decision branch matcher.

    Args:
        source: The type or type expression to match against
        matches: Optional custom matching function

    Returns:
        A DecisionBranchBuilder for constructing the branch
    """
    # Note, the following node_id really is just a placeholder and shouldn't end up in the final graph
    # This is why we don't expose a way for end users to override the value used here.
    node_id = NodeID(generate_placeholder_node_id('match_decision'))
    decision = Decision[StateT, DepsT, Never](id=node_id, branches=[], note=None)
    new_path_builder = PathBuilder[StateT, DepsT, SourceT](working_items=[])
    return DecisionBranchBuilder(decision=decision, source=source, matches=matches, path_builder=new_path_builder)

match_node

match_node(
    source: type[SourceNodeT],
    *,
    matches: Callable[[Any], bool] | None = None
) -> DecisionBranch[SourceNodeT]

Create a decision branch for BaseNode subclasses.

This is similar to match() but specifically designed for matching against BaseNode types.

Parameters:

Name	Type	Description	Default
source	type[SourceNodeT]	The BaseNode subclass to match against	required
matches	Callable[[Any], bool] | None	Optional custom matching function	None

Returns:

Type	Description
DecisionBranch[SourceNodeT]	A DecisionBranch for the BaseNode type

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def match_node(
    self,
    source: type[SourceNodeT],
    *,
    matches: Callable[[Any], bool] | None = None,
) -> DecisionBranch[SourceNodeT]:
    """Create a decision branch for `BaseNode` subclasses.

    This is similar to `match()` but specifically designed for matching against `BaseNode` types.

    Args:
        source: The `BaseNode` subclass to match against
        matches: Optional custom matching function

    Returns:
        A `DecisionBranch` for the `BaseNode` type
    """
    node = NodeStep(source)
    path = Path(items=[DestinationMarker(node.id)])
    return DecisionBranch(source=source, matches=matches, path=path, destinations=[node])

node

node(
    node_type: type[BaseNode[StateT, DepsT, GraphOutputT]],
) -> EdgePath[StateT, DepsT]

Create an edge path from a BaseNode class.

This method integrates a BaseNode subclass into the builder graph by analyzing its run return type hints and creating appropriate edges.

Parameters:

Name	Type	Description	Default
node_type	type[BaseNode[StateT, DepsT, GraphOutputT]]	The BaseNode subclass to integrate	required

Returns:

Type	Description
EdgePath[StateT, DepsT]	An EdgePath representing the node and its connections

Raises:

Type	Description
GraphSetupError	If the node type is missing required type hints

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def node(
    self,
    node_type: type[BaseNode[StateT, DepsT, GraphOutputT]],
) -> EdgePath[StateT, DepsT]:
    """Create an edge path from a `BaseNode` class.

    This method integrates a `BaseNode` subclass into the builder graph by
    analyzing its `run` return type hints and creating appropriate edges.

    Args:
        node_type: The `BaseNode` subclass to integrate

    Returns:
        An `EdgePath` representing the node and its connections

    Raises:
        GraphSetupError: If the node type is missing required type hints
    """
    parent_namespace = _utils.get_parent_namespace(inspect.currentframe())
    type_hints = get_type_hints(node_type.run, localns=parent_namespace, include_extras=True)
    try:
        return_hint = type_hints['return']
    except KeyError as e:  # pragma: no cover
        raise exceptions.GraphSetupError(
            f'Node {node_type} is missing a return type hint on its `run` method'
        ) from e

    node = NodeStep(node_type)

    edge = self._edge_from_return_hint(node, return_hint)
    if not edge:  # pragma: no cover
        raise exceptions.GraphSetupError(f'Node {node_type} is missing a return type hint on its `run` method')

    return edge

build

build(
    validate_graph_structure: bool = True,
) -> Graph[StateT, DepsT, GraphInputT, GraphOutputT]

Build the final executable graph from the accumulated nodes and edges.

This method performs validation, normalization, and analysis of the graph structure to create a complete, executable graph instance.

Parameters:

Name	Type	Description	Default
validate_graph_structure	bool	whether to perform validation of the graph structure See the docstring of _validate_graph_structure below for more details.	True

Returns:

Type	Description
Graph[StateT, DepsT, GraphInputT, GraphOutputT]	A complete Graph instance ready for execution

Raises:

Type	Description
ValueError	If the graph structure is invalid (e.g., join without parent fork)

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def build(self, validate_graph_structure: bool = True) -> Graph[StateT, DepsT, GraphInputT, GraphOutputT]:
    """Build the final executable graph from the accumulated nodes and edges.

    This method performs validation, normalization, and analysis of the graph
    structure to create a complete, executable graph instance.

    Args:
        validate_graph_structure: whether to perform validation of the graph structure
            See the docstring of _validate_graph_structure below for more details.

    Returns:
        A complete Graph instance ready for execution

    Raises:
        ValueError: If the graph structure is invalid (e.g., join without parent fork)
    """
    nodes = self._nodes
    edges_by_source = self._edges_by_source

    nodes, edges_by_source = _replace_placeholder_node_ids(nodes, edges_by_source)
    nodes, edges_by_source = _flatten_paths(nodes, edges_by_source)
    nodes, edges_by_source = _normalize_forks(nodes, edges_by_source)
    if validate_graph_structure:
        _validate_graph_structure(nodes, edges_by_source)
    parent_forks = _collect_dominating_forks(nodes, edges_by_source)
    intermediate_join_nodes = _compute_intermediate_join_nodes(nodes, parent_forks)

    return Graph[StateT, DepsT, GraphInputT, GraphOutputT](
        name=self.name,
        state_type=unpack_type_expression(self.state_type),
        deps_type=unpack_type_expression(self.deps_type),
        input_type=unpack_type_expression(self.input_type),
        output_type=unpack_type_expression(self.output_type),
        nodes=nodes,
        edges_by_source=edges_by_source,
        parent_forks=parent_forks,
        intermediate_join_nodes=intermediate_join_nodes,
        auto_instrument=self.auto_instrument,
    )

DEFAULT_HIGHLIGHT_CSS module-attribute

DEFAULT_HIGHLIGHT_CSS = 'fill:#fdff32'

The default CSS to use for highlighting nodes.

StateDiagramDirection module-attribute

StateDiagramDirection = Literal['TB', 'LR', 'RL', 'BT']

Used to specify the direction of the state diagram generated by mermaid.

• 'TB': Top to bottom, this is the default for mermaid charts.
• 'LR': Left to right
• 'RL': Right to left
• 'BT': Bottom to top

MermaidNode dataclass

A mermaid node.

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@dataclass
class MermaidNode:
    """A mermaid node."""

    id: str
    kind: NodeKind
    label: str | None
    note: str | None

MermaidEdge dataclass

A mermaid edge.

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@dataclass
class MermaidEdge:
    """A mermaid edge."""

    start_id: str
    end_id: str
    label: str | None

build_mermaid_graph

build_mermaid_graph(
    graph_nodes: dict[NodeID, AnyNode],
    graph_edges_by_source: dict[NodeID, list[Path]],
) -> MermaidGraph

Build a mermaid graph.

Source code in pydantic_graph/pydantic_graph/graph_builder.py

def build_mermaid_graph(  # noqa: C901
    graph_nodes: dict[NodeID, AnyNode], graph_edges_by_source: dict[NodeID, list[Path]]
) -> MermaidGraph:
    """Build a mermaid graph."""
    nodes: list[MermaidNode] = []
    edges_by_source: dict[str, list[MermaidEdge]] = defaultdict(list)

    def _collect_edges(path: Path, last_source_id: NodeID) -> None:
        working_label: str | None = None
        for item in path.items:
            assert not isinstance(item, MapMarker | BroadcastMarker), 'These should be removed during Graph building'
            if isinstance(item, LabelMarker):
                working_label = item.label
            elif isinstance(item, DestinationMarker):
                edges_by_source[last_source_id].append(MermaidEdge(last_source_id, item.destination_id, working_label))

    for node_id, node in graph_nodes.items():
        kind: NodeKind
        label: str | None = None
        note: str | None = None
        if isinstance(node, StartNode):
            kind = 'start'
        elif isinstance(node, EndNode):
            kind = 'end'
        elif isinstance(node, Step):
            kind = 'step'
            label = node.label
        elif isinstance(node, Join):
            kind = 'join'
        elif isinstance(node, Fork):
            kind = 'map' if node.is_map else 'broadcast'
        elif isinstance(node, Decision):
            kind = 'decision'
            note = node.note
        else:
            assert_never(node)

        source_node = MermaidNode(id=node_id, kind=kind, label=label, note=note)
        nodes.append(source_node)

    for k, v in graph_edges_by_source.items():
        for path in v:
            _collect_edges(path, k)

    for node in graph_nodes.values():
        if isinstance(node, Decision):
            for branch in node.branches:
                _collect_edges(branch.path, node.id)

    # Add edges in the same order that we added nodes
    edges: list[MermaidEdge] = sum([edges_by_source.get(node.id, []) for node in nodes], list[MermaidEdge]())
    return MermaidGraph(nodes, edges)

MermaidGraph dataclass

A mermaid graph.

Source code in pydantic_graph/pydantic_graph/graph_builder.py

@dataclass
class MermaidGraph:
    """A mermaid graph."""

    nodes: list[MermaidNode]
    edges: list[MermaidEdge]

    title: str | None = None
    direction: StateDiagramDirection | None = None

    def render(
        self,
        direction: StateDiagramDirection | None = None,
        title: str | None = None,
        edge_labels: bool = True,
    ):
        lines: list[str] = []
        if title:
            lines = ['---', f'title: {title}', '---']
        lines.append('stateDiagram-v2')
        if direction is not None:
            lines.append(f'  direction {direction}')

        nodes, edges = _topological_sort(self.nodes, self.edges)
        for node in nodes:
            # List all nodes in order they were created
            node_lines: list[str] = []
            if node.kind == 'start' or node.kind == 'end':
                pass  # Start and end nodes use special [*] syntax in edges
            elif node.kind == 'step':
                line = f'  {node.id}'
                if node.label:
                    line += f': {node.label}'
                node_lines.append(line)
            elif node.kind == 'join':
                node_lines = [f'  state {node.id} <<join>>']
            elif node.kind == 'broadcast' or node.kind == 'map':
                node_lines = [f'  state {node.id} <<fork>>']
            elif node.kind == 'decision':
                node_lines = [f'  state {node.id} <<choice>>']
                if node.note:
                    node_lines.append(f'  note right of {node.id}\n    {node.note}\n  end note')
            else:  # pragma: no cover
                assert_never(node.kind)
            lines.extend(node_lines)

        lines.append('')

        for edge in edges:
            # Use special [*] syntax for start/end nodes
            render_start_id = '[*]' if edge.start_id == StartNode.id else edge.start_id
            render_end_id = '[*]' if edge.end_id == EndNode.id else edge.end_id
            edge_line = f'  {render_start_id} --> {render_end_id}'
            if edge.label and edge_labels:
                edge_line += f': {edge.label}'
            lines.append(edge_line)

        return '\n'.join(lines)