graphai.graph

NodeProtocol Objects

class NodeProtocol(Protocol)

Protocol defining the interface of a decorated node.

BranchError Objects

@dataclass
class BranchError()

A failure inside a branch pipeline, recorded instead of raised so the main path is never taken down by a side pipeline.

Graph Objects

class Graph()

get_state

def get_state() -> dict[str, Any]

Get the current graph state. Returns: The current graph state.

set_state

def set_state(state: dict[str, Any]) -> Graph

Set the graph state. Arguments:

state - The new state to set for the graph.

Returns: The graph instance.

update_state

def update_state(values: dict[str, Any]) -> Graph

Update the graph state with new values. Arguments:

values - The new values to update the graph state with.

Returns: The graph instance.

reset_state

def reset_state() -> Graph

Reset the graph state to an empty dict.

add_node

def add_node(node: NodeProtocol) -> Graph

Adds a node to the graph. Arguments:

node - The node to add to the graph.

Raises:

Exception - If a node with the same name already exists in the graph.

add_edge

def add_edge(source: NodeProtocol | str,
             destination: NodeProtocol | str) -> Graph

Adds an edge between two nodes that already exist in the graph. Arguments:

source - The source node or its name.
destination - The destination node or its name.

add_branch

def add_branch(source: NodeProtocol | str,
               destination: NodeProtocol | str,
               *,
               condition: BranchCondition | None = None,
               wait: bool = True) -> Graph

Adds a branch edge: a side pipeline that starts at destination once source has run, without becoming a fork of the main path. Regular edges out of a node with two or more successors fork the main path and expect a join. A branch edge instead runs destination and whatever regular edges follow it as an isolated pipeline: it needs no join and no end node, it stops at the first node with no successors (or one whose output has success: False), and an exception inside it is recorded in branch_errors rather than raised. Branches may themselves have branches. Arguments:

source - The node whose completion starts the branch.
destination - The first node of the branch pipeline.
condition - Optional callable (output, state) -> bool (may be async) evaluated with the source node’s output; the branch is skipped when it returns a falsy value.
wait - When True (default) the main path waits for the branch to finish before moving on. When False the branch runs as a detached task; source0 still waits for all detached branches before returning.

add_router

def add_router(sources: list[NodeProtocol], router: NodeProtocol,
               destinations: list[NodeProtocol]) -> Graph

Adds a router node, allowing for a decision to be made on which branch to follow based on the choice output of the router node. Arguments:

sources - The list of source nodes for the router.
router - The router node.
destinations - The list of destination nodes for the router.

compile

def compile(*, strict: bool = False) -> Graph

Validate the graph:

exactly one start node present (or Graph.start_node set)
at least one end node present
all edges reference known nodes
all nodes reachable from the start (optional) no cycles when strict=True Returns self on success; raises GraphCompileError otherwise.

wait_for_branches

async def wait_for_branches() -> None

Wait for every detached branch pipeline started by this graph. Called by execute() before it returns; exposed for callers that drive _execute_branch themselves.

execute_many

async def execute_many(inputs: Iterable[dict[str, Any]],
                       *,
                       concurrency: int = 5) -> list[Any]

Execute the graph on many inputs concurrently. Arguments:

inputs (Iterable[dict]): An iterable of input dicts to feed into the graph.
concurrency (int): Maximum number of graph executions to run at once.
state (Optional[Any]): Optional shared state to pass to each execution. If you want isolated state per execution, pass None and the graph’s normal semantics will apply.

Returns: list[Any]: The list of results in the same order as inputs.

get_callback

def get_callback()

Get a new instance of the callback class. Returns: Callback: A new instance of the callback class.

set_callback

def set_callback(callback_class: type[Callback]) -> "Graph"

Set the callback class that is returned by the get_callback method and used as the default callback when no callback is passed to the execute method. Arguments:

callback_class (type[Callback]): The callback class to use as the default callback.

add_parallel

def add_parallel(source: NodeProtocol | str,
                 destinations: list[NodeProtocol | str])

Add multiple outgoing edges from a single source node to be executed in parallel. Arguments:

source - The source node for the parallel branches.
destinations - The list of destination nodes for the parallel branches.

add_join

def add_join(sources: list[NodeProtocol | str],
             destination: NodeProtocol | str)

Joins multiple parallel branches into a single branch. Arguments:

sources - The list of source nodes for the join.
destination - The destination node for the join.

visualize

def visualize(*, save_path: str | None = None)

Render the current graph. If matplotlib is not installed, raise a helpful error telling users to install the viz extra. Optionally save to a file via save_path.

BranchEdge Objects

class BranchEdge()

Side edge from source to the first node of an isolated pipeline. See Graph.add_branch.

Get Started

Client Reference

User Guide

NodeProtocol Objects

BranchError Objects

Graph Objects

get_state

set_state

update_state

reset_state

add_node

add_edge

add_branch

add_router

compile

wait_for_branches

execute_many

get_callback

set_callback

add_parallel

add_join

visualize

BranchEdge Objects

​NodeProtocol Objects

​BranchError Objects

​Graph Objects

​get_state

​set_state

​update_state

​reset_state

​add_node

​add_edge

​add_branch

​add_router

​compile

​wait_for_branches

​execute_many

​get_callback

​set_callback

​add_parallel

​add_join

​visualize

​BranchEdge Objects

NodeProtocol Objects

BranchError Objects

Graph Objects

get_state

set_state

update_state

reset_state

add_node

add_edge

add_branch

add_router

compile

wait_for_branches

execute_many

get_callback

set_callback

add_parallel

add_join

visualize

BranchEdge Objects