Graph | Smallest AI Docs

A Graph defines how nodes are connected. Events flow through the graph from parent nodes to child nodes, creating a processing pipeline for your agent logic.

The Atoms graph is a DAG (Directed Acyclic Graph). Events can flow through multiple branches, but never in circles.

How Graphs Work

When you call session.add_edge(parent, child), you’re creating a connection. Events emitted by the parent via send_event() are automatically queued for the child.

User Input → [Root] → [Your Nodes] → [Sink] → User Output

The session automatically creates two special nodes:

Root: Entry point-receives events from the WebSocket
Sink: Exit point-sends events back to the WebSocket

Building a Graph

Step 1: Add Nodes

async def setup(session: CrewSession):
    logger = LoggerNode()
    agent = SalesAgent()
    analytics = AnalyticsNode()
    
    session.add_node(logger)
    session.add_node(agent)
    session.add_node(analytics)

session.add_node(node): Registers a Node instance with the session. The node must inherit from the base Node class.

Step 2: Connect with Edges

    # Define the flow: Logger → Agent → Analytics
    session.add_edge(logger, agent)
    session.add_edge(agent, analytics)
    
    await session.start()

The Resulting Graph

[Root] → [Logger] → [Agent] → [Analytics] → [Sink]

Graph Patterns

# [Root] → [A] → [B] → [C] → [Sink]
# The simplest pattern—events flow sequentially.
session.add_edge(node_a, node_b)
session.add_edge(node_b, node_c)

Automatic Connections

Nodes without explicit parents connect to Root. Nodes without explicit children connect to Sink.

# Just add one node with no edges:
session.add_node(my_agent)
# Automatically becomes:
# [Root] -> [my_agent] -> [Sink]

This means a minimal agent only needs:

async def setup(session: CrewSession):
    session.add_node(SalesAgent())
    await session.start()
    await session.wait_until_complete()

Cycle Detection

Graphs must not contain cycles. The session validates this at startup:

# This will FAIL
session.add_edge(node_a, node_b)
session.add_edge(node_b, node_c)
session.add_edge(node_c, node_a)  # Creates a cycle!
await session.start()  
# Raises: ValueError("Graph contains cycles")

Cycles would cause infinite event loops. The framework prevents this at startup, but design your graphs as DAGs from the start.

Event Flow in Detail

When a node calls send_event():

The event is queued for each child node
Each child’s process_event() is called asynchronously
Children can further propagate via their own send_event()

class ParentNode(Node):
    async def process_event(self, event):
        # Modify the event if needed
        event.metadata["processed_by"] = self.name
        
        # Queue for all children
        await self.send_event(event)

Each node has its own event queue. Multiple events can be queued while a node is processing, and they’ll be handled in order.

Custom Routing

For dynamic routing, don’t use send_event()-directly queue to specific children:

class RouterNode(Node):
    def __init__(self, sales_node, support_node):
        super().__init__(name="router")
        self.sales = sales_node
        self.support = support_node
    
    async def process_event(self, event):
        intent = self.classify(event)
        
        if intent == "sales":
            await self.sales.queue_event(event)
        elif intent == "support":
            await self.support.queue_event(event)
        else:
            # Fallback: send to all children normally
            await self.send_event(event)

Best Practices

Keep graphs shallow

Deeply nested graphs increase latency. Events have to hop through every node.

Bad: A -> B -> C -> D -> E (5 hops) Good: Router -> [A, B, C, D, E] (2 hops)

Use meaningful node names

You will thank yourself when reading logs.

# Good
LoggerNode(name="input_logger")
# Bad
LoggerNode(name="log1")

Avoid fan-out > 3

If a node is sending to more than 3 children, it’s usually better to have a dedicated Router node that decides where the event goes, rather than broadcasting to everyone.

Draw it first

Graphs can get complex. Sketching the flow on paper (or Excalidraw) before coding saves a lot of headaches.