Graphs

Graphs are the foundation of Mesh. They define the structure and flow of your multi-agent workflows.

What is a Graph?

A graph in Mesh is a directed graph composed of:

• Nodes: Units of execution (agents, LLMs, tools, etc.)
• Edges: Connections between nodes defining execution flow
• Entry Point: The starting node for execution
• Controlled Cycles: Optional loops with exit conditions or iteration limits

Graph Structure

START (implicit)
  ↓
[entry_node]
  ↓
[node_a]
  ↓
[node_b]
  ↓
END (implicit)

Key Properties

1. Directed: Edges have a direction (source → target)
2. Controlled Cycles: Loops are allowed with proper controls (conditions or max iterations)
3. Connected: All nodes must be reachable from START

Building Graphs

Programmatic API (StateGraph)

The recommended way to build graphs is using the StateGraph builder:

from mesh import StateGraph

graph = StateGraph()

# Add nodes
graph.add_node("agent", my_agent, node_type="agent")
graph.add_node("tool", my_function, node_type="tool")

# Add edges
graph.add_edge("START", "agent")
graph.add_edge("agent", "tool")

# Set entry point
graph.set_entry_point("agent")

# Compile
compiled = graph.compile()

React Flow JSON (Declarative)

Parse Flowise-compatible JSON:

from mesh import ReactFlowParser, NodeRegistry

registry = NodeRegistry()
parser = ReactFlowParser(registry)

flow_json = {
    "nodes": [
        {"id": "agent_1", "type": "agentAgentflow", "data": {...}},
        {"id": "llm_1", "type": "llmAgentflow", "data": {...}}
    ],
    "edges": [
        {"source": "agent_1", "target": "llm_1"}
    ]
}

graph = parser.parse(flow_json)

Graph Lifecycle

Build → Validate → Compile → Execute

1. Build

Add nodes and edges:

graph = StateGraph()
graph.add_node("step1", None, node_type="llm")
graph.add_node("step2", None, node_type="llm")
graph.add_edge("START", "step1")
graph.add_edge("step1", "step2")

2. Validate

Check for common issues:

graph.set_entry_point("step1")
compiled = graph.compile()  # Validates automatically

Validation checks:

• Entry point is set
• No uncontrolled cycles (all cycles must have loop controls)
• All nodes are connected
• No orphaned nodes
• Loop edges have proper controls (condition or max_iterations)

3. Compile

Creates an executable ExecutionGraph:

compiled = graph.compile()
# Returns: ExecutionGraph with dependency info

4. Execute

Run the graph:

from mesh import Executor, ExecutionContext, MemoryBackend

executor = Executor(compiled, MemoryBackend())
context = ExecutionContext(
    graph_id="my-graph",
    session_id="session-1",
    chat_history=[],
    variables={},
    state={}
)

async for event in executor.execute("input", context):
    print(event)

Graph Patterns

Sequential Flow

# A → B → C
graph.add_node("A", None, node_type="llm")
graph.add_node("B", None, node_type="llm")
graph.add_node("C", None, node_type="llm")
graph.add_edge("START", "A")
graph.add_edge("A", "B")
graph.add_edge("B", "C")

Parallel Branches

# A → [B, C] (both execute after A)
graph.add_node("A", None, node_type="llm")
graph.add_node("B", None, node_type="llm")
graph.add_node("C", None, node_type="llm")
graph.add_edge("START", "A")
graph.add_edge("A", "B")
graph.add_edge("A", "C")

Conditional Branching

from mesh.nodes import Condition

# A → [condition] → [B or C]
graph.add_node("A", None, node_type="llm")
graph.add_node("condition", [
    Condition("success", lambda x: "success" in str(x), "B"),
    Condition("failure", lambda x: "failure" in str(x), "C")
], node_type="condition")
graph.add_node("B", None, node_type="llm")
graph.add_node("C", None, node_type="llm")
graph.add_edge("START", "A")
graph.add_edge("A", "condition")
graph.add_edge("condition", "B")
graph.add_edge("condition", "C")

Multi-Input Nodes

# [A, B] → C (C waits for both A and B)
graph.add_node("A", None, node_type="llm")
graph.add_node("B", None, node_type="llm")
graph.add_node("C", None, node_type="llm")
graph.add_edge("START", "A")
graph.add_edge("START", "B")
graph.add_edge("A", "C")
graph.add_edge("B", "C")  # C waits for both

Controlled Loops

Mesh supports loops with proper controls to prevent infinite execution:

Loop with Max Iterations

# Self-loop with fixed iteration count
graph.add_node("process", process_fn, node_type="tool")
graph.add_edge("START", "process")
graph.add_edge(
    "process",
    "process",  # Loop back to itself
    is_loop_edge=True,
    max_iterations=10  # Run at most 10 times
)

Loop with Condition

# Loop until condition is met
def should_continue(state, output):
    return output.get("value", 0) < 100

graph.add_node("increment", increment_fn, node_type="tool")
graph.add_edge("START", "increment")
graph.add_edge(
    "increment",
    "increment",
    is_loop_edge=True,
    loop_condition=should_continue  # Exit when returns False
)

Loop with Both Controls

# Loop with condition AND max iterations for safety
graph.add_node("check", check_fn, node_type="tool")
graph.add_node("process", process_fn, node_type="tool")
graph.add_edge("START", "check")
graph.add_edge("check", "process")
graph.add_edge(
    "process",
    "check",  # Loop back
    is_loop_edge=True,
    loop_condition=lambda state, output: not output.get("done", False),
    max_iterations=50  # Safety limit
)

Loop Condition Signature:

def loop_condition(state: Dict, output: Dict) -> bool:
    """Return True to continue loop, False to exit."""
    return some_check(state, output)

Key Requirements:

• Loop edges must have is_loop_edge=True
• Must specify at least one: loop_condition or max_iterations
• Loop conditions receive both shared state and node output
• Conditions that fail (raise exception) safely exit the loop

Execution Model

Mesh uses a queue-based execution model:

1. Initialize queue with entry point
2. Dequeue node and execute
3. Emit events (streaming)
4. Queue children based on dependencies
5. Repeat until queue is empty

Dependency Resolution

When a node has multiple parents:

# A → C
# B → C (C waits for both A and B)

The executor:

1. Tracks received inputs from parents
2. Waits until all parents complete
3. Combines inputs
4. Executes node

Graph State

State flows through the graph via:

1. Node Outputs

Each node produces output that becomes input for children:

# Node A output: {"content": "Hello"}
# Node B receives: {"content": "Hello"}

2. Shared State

Persistent state across all nodes:

context = ExecutionContext(
    graph_id="my-graph",
    session_id="session-1",
    chat_history=[],
    variables={"user_id": "123"},  # Shared variables
    state={"step_count": 0}  # Shared state
)

3. Chat History

Accumulated conversation history:

context.chat_history = [
    {"role": "user", "content": "Hello"},
    {"role": "assistant", "content": "Hi there!"}
]

Best Practices

1. Use Descriptive Node IDs

# ❌ Bad
graph.add_node("n1", None, node_type="llm")
graph.add_node("n2", None, node_type="llm")

# ✅ Good
graph.add_node("analyzer", None, node_type="llm")
graph.add_node("summarizer", None, node_type="llm")

2. Keep Graphs Simple

# ✅ Good: 3-7 nodes
START → analyze → process → respond → END

# ⚠️ Consider splitting: 20+ nodes
# Too complex, hard to maintain

3. Validate Early

graph = StateGraph()
graph.add_node("step1", None, node_type="llm")
graph.add_node("step2", None, node_type="llm")
graph.add_edge("START", "step1")
graph.add_edge("step1", "step2")
graph.set_entry_point("step1")

try:
    compiled = graph.compile()  # Validates
except ValueError as e:
    print(f"Graph invalid: {e}")

4. Use Variables for Flexibility

# Reference previous nodes
graph.add_node("step2", None, node_type="llm",
               system_prompt="Based on , ...")

Common Errors

“No entry point set”

Problem: Forgot to call set_entry_point()

Solution:

graph.set_entry_point("first_node")
compiled = graph.compile()

“Cycle detected in graph”

Problem: Edge creates an uncontrolled cycle (no loop controls)

Solution: Mark the edge as a loop edge with proper controls:

# ❌ Bad: Uncontrolled cycle
graph.add_edge("A", "B")
graph.add_edge("B", "A")  # Error: cycle!

# ✅ Good: Controlled loop
graph.add_edge("A", "B")
graph.add_edge(
    "B",
    "A",
    is_loop_edge=True,
    max_iterations=10  # Or use loop_condition
)

“Orphaned nodes detected”

Problem: Node not connected to START

Solution: Add edge from START or parent node:

graph.add_edge("START", "orphaned_node")

“Node not found”

Problem: Referencing non-existent node in edge

Solution: Ensure node exists before adding edge:

graph.add_node("node_a", None, node_type="llm")  # Create first
graph.add_edge("START", "node_a")  # Then reference

“Loop edge must have loop_condition or max_iterations”

Problem: Marked edge as is_loop_edge=True but didn’t provide controls

Solution: Add at least one control mechanism:

# ❌ Bad: Loop edge without controls
graph.add_edge("A", "A", is_loop_edge=True)

# ✅ Good: With max_iterations
graph.add_edge("A", "A", is_loop_edge=True, max_iterations=10)

# ✅ Good: With condition
graph.add_edge("A", "A", is_loop_edge=True,
               loop_condition=lambda s, o: o.get("count", 0) < 5)

# ✅ Best: With both for safety
graph.add_edge("A", "A", is_loop_edge=True,
               loop_condition=lambda s, o: not o.get("done", False),
               max_iterations=100)

Advanced Topics

Sub-Graphs (Future)

Coming in v2:

# Compose graphs
sub_graph = StateGraph()
# ... build sub_graph

main_graph = StateGraph()
main_graph.add_subgraph("processing", sub_graph)

Dynamic Graphs (Future)

Coming in v2:

# Modify graph during execution
executor.add_node_dynamically("new_node", ...)

See Also

• Nodes - Learn about node types
• Execution - Understand execution flow
• State Management - Managing state
• Variables - Variable resolution