The Graph
is the central orchestration component in GraphAI. It connects Nodes
and Routers
into a coherent workflow and manages the execution flow.
Graph Basics
A graph consists of:
- Nodes: Processing units that perform specific tasks
- Edges: Connections between nodes that define the flow of data
- State: Shared context that persists throughout the execution
Creating a Graph
from graphai import Graph
# Create a graph with default settings
graph = Graph()
# Create a graph with custom max steps and initial state
graph = Graph(max_steps=20, initial_state={"history": []})
Parameters
max_steps
(int, default=10): Maximum number of steps to prevent infinite loops
initial_state
(Dict[str, Any], optional): Initial state for the graph execution
Adding Nodes
Nodes are the building blocks of your graph. Each node represents a discrete processing step:
# Add a node to the graph
graph.add_node(my_node())
# Add multiple nodes
graph.add_node(node_a())
graph.add_node(node_b())
graph.add_node(node_c())
Nodes can be:
- Start nodes: Entry points to the graph (only one allowed)
- End nodes: Exit points from the graph (multiple allowed)
- Regular nodes: Intermediate processing steps
- Router nodes: Decision points that determine execution flow
Connecting Nodes with Edges
Edges define how data flows between nodes:
# Connect two nodes
graph.add_edge(source_node, destination_node)
# Can use node names instead of node objects
graph.add_edge("node_a", "node_b")
For linear workflows, you simply connect nodes in sequence:
graph.add_edge(node_a, node_b)
graph.add_edge(node_b, node_c)
Working with Routers
Routers are special nodes that determine the next node to execute based on their output:
# Add a router with its sources and destinations
graph.add_router(
sources=[node_a], # Nodes that can lead to the router
router=my_router(), # The router node itself
destinations=[node_b, node_c] # Possible destinations from the router
)
The router must return a dictionary containing a "choice"
key with the name of the next node to execute:
@router
async def my_router(input: dict):
# Decision logic
if some_condition:
return {"choice": "node_b", "data": processed_data}
else:
return {"choice": "node_c", "data": processed_data}
Graph Execution
To execute a graph:
import asyncio
async def run_graph():
# Define initial input
input_data = {"query": "Hello, world!"}
# Execute the graph
result = await graph.execute(input_data)
return result
# Run the async function
result = asyncio.run(run_graph())
Execution Flow
- The graph starts execution at the designated start node
- Each node processes the input and returns an output
- The output is merged with the current state and passed to the next node
- If a router node is encountered, its
"choice"
output determines the next node
- Execution continues until an end node is reached or max_steps is exceeded
State Management
The graph maintains a state dictionary that persists throughout execution:
# Get the current state
state = graph.get_state()
# Set the state
graph.set_state({"history": [], "context": "some context"})
# Update the state
graph.update_state({"new_key": "new_value"})
# Reset the state
graph.reset_state()
Each node receives the current state as an optional parameter:
@node
async def my_node(input: dict, state: dict):
# Access the state
history = state.get("history", [])
# Update the state (changes won't persist outside this node)
# For persistent changes, return them in the output
return {"output": result, "history": history + [result]}
Graph Validation
Before execution, you can validate that your graph is properly configured:
# Compile will raise exceptions if the graph is invalid
graph.compile()
The compile method checks for:
- Presence of a start node
- Presence of at least one end node
- Graph validity (e.g., no disconnected nodes)
Visualization
GraphAI provides a method to visualize your graph (requires matplotlib and networkx):
# Visualize the graph
graph.visualize()
This generates a visual representation of your graph, making it easier to understand complex workflows.
Next Steps
- Learn about Nodes to understand how to build processing units
- Explore State management for maintaining context
- Check out Callbacks for implementing streaming