Graph Workflow

The GraphWorkflow orchestrates agents using a Directed Acyclic Graph (DAG) structure, enabling complex workflows with dependencies, parallel branches, and convergence points. Ideal for sophisticated pipelines with intricate task relationships.

When to Use

Complex dependencies: Tasks with intricate dependency graphs
Parallel branches: Multiple independent paths that converge
Pipeline optimization: Maximize parallelism while respecting dependencies
Software builds: Compile, test, and deploy pipelines
Data processing: ETL workflows with multiple stages

Key Features

DAG-based execution with topological sorting
Automatic parallelization of independent nodes
Support for NetworkX and Rustworkx backends
Fan-out and fan-in patterns
Entry and exit point management
Graph visualization (with Graphviz)
Auto-compilation for performance
Graph validation and cycle detection
Nested workflows: a GraphWorkflow can run as a single node
Per-layer checkpoints for resuming interrupted runs
Token usage summed across every node and subgraph

Basic Example

from swarms import Agent, GraphWorkflow, Node, Edge, NodeType

# Define agents
data_collector = Agent(
    agent_name="DataCollector",
    system_prompt="Collect and validate data from sources.",
    model_name="gpt-5.4",
)

processor = Agent(
    agent_name="Processor",
    system_prompt="Process and clean collected data.",
    model_name="gpt-5.4",
)

analyzer = Agent(
    agent_name="Analyzer",
    system_prompt="Analyze processed data for insights.",
    model_name="gpt-5.4",
)

# Create graph workflow
workflow = GraphWorkflow(
    name="Data-Pipeline",
    description="ETL workflow for data processing",
)

# Add agents as nodes
workflow.add_node(data_collector)
workflow.add_node(processor)
workflow.add_node(analyzer)

# Define dependencies
workflow.add_edge("DataCollector", "Processor")  # Collector -> Processor
workflow.add_edge("Processor", "Analyzer")       # Processor -> Analyzer

# Compile and run
workflow.compile()
result = workflow.run("Process Q4 sales data")
print(result)

Creating from Spec

Simplified workflow creation:

from swarms import GraphWorkflow

# Define agents
agents = [collector, processor, analyzer, reporter]

# Define edges (dependencies)
edges = [
    ("DataCollector", "Processor"),
    ("Processor", "Analyzer"),
    ("Analyzer", "Reporter"),
]

# Create workflow
workflow = GraphWorkflow.from_spec(
    agents=agents,
    edges=edges,
    task="Process data",
)

result = workflow.run()

Advanced Edge Patterns

Fan-Out Pattern

One node distributes to multiple nodes:

# Single edge to multiple targets
workflow.add_edges_from_source(
    source="DataCollector",
    targets=["TechnicalAnalyst", "FundamentalAnalyst", "SentimentAnalyst"]
)

# Equivalent to:
# DataCollector -> TechnicalAnalyst
# DataCollector -> FundamentalAnalyst
# DataCollector -> SentimentAnalyst
# (All three analysts run in parallel)

Fan-In Pattern

Multiple nodes converge to one:

# Multiple sources to single target
workflow.add_edges_to_target(
    sources=["TechnicalAnalyst", "FundamentalAnalyst", "SentimentAnalyst"],
    target="SynthesisAgent"
)

# Equivalent to:
# TechnicalAnalyst -> SynthesisAgent
# FundamentalAnalyst -> SynthesisAgent
# SentimentAnalyst -> SynthesisAgent
# (Synthesis waits for all three)

Parallel Chain Pattern

Full mesh connection:

# Multiple sources to multiple targets
workflow.add_parallel_chain(
    sources=["Collector1", "Collector2"],
    targets=["Analyst1", "Analyst2", "Analyst3"]
)

# Creates:
# Collector1 -> Analyst1, Analyst2, Analyst3
# Collector2 -> Analyst1, Analyst2, Analyst3

Tuple-Based Patterns

Simplified edge definitions:

edges = [
    # Simple edge
    ("agent1", "agent2"),
    
    # Fan-out
    ("agent1", ["agent2", "agent3"]),
    
    # Fan-in
    (["agent1", "agent2"], "agent3"),
    
    # Parallel chain
    (["agent1", "agent2"], ["agent3", "agent4"]),
]

workflow = GraphWorkflow.from_spec(
    agents=agents,
    edges=edges,
)

Key Parameters

str

Name for the workflow

str

Description of the workflow’s purpose

Dict[str, Node]

Dictionary of nodes (agents)

List[Edge]

List of edges (dependencies)

List[str]

Node IDs with no predecessors (auto-detected if not set)

List[str]

Node IDs with no successors (auto-detected if not set)

int

default:"1"

Maximum execution loops

bool

default:"True"

Automatically compile on initialization

str

default:"networkx"

Graph backend (“networkx” or “rustworkx”)

bool

default:"False"

Enable verbose logging

str

Default task used by run() when no task argument is passed.

str

Directory for per-layer checkpoints. Rerunning the same task skips every layer that already has one. See Checkpoints.

Callable[[str, Any], None]

Instance-level callback fired as (node_id, output) immediately after each node finishes. A callback passed directly to run() takes precedence over this one.

int

Caps how many nodes execute concurrently within a layer. Defaults to max(1, int(get_cpu_cores() * 0.95)) when not set.

Methods

add_node()

Add an agent to the graph. The node ID is the agent’s agent_name, and a duplicate ID raises ValueError:

workflow.add_node(
    agent=my_agent,
    metadata={"priority": "high"}  # Optional metadata
)

add_node() also accepts a whole GraphWorkflow, which becomes a subgraph node. Pass the agent or workflow itself, not a Node object.

add_nodes()

Add multiple agents concurrently:

workflow.add_nodes(
    agents=[agent1, agent2, agent3],
    batch_size=10,
)

add_edge()

Add a dependency between nodes:

# Using agent objects
workflow.add_edge(source_agent, target_agent)

# Using node IDs
workflow.add_edge("Agent1", "Agent2")

# Using Edge object
from swarms import Edge
edge = Edge(source="Agent1", target="Agent2", metadata={"type": "data"})
workflow.add_edge(edge)

compile()

Pre-compute expensive operations:

# Manual compilation
workflow.compile()

# Auto-compilation (default)
workflow = GraphWorkflow(auto_compile=True)

Compilation:

Auto-sets entry/exit points
Computes topological layers
Caches for performance until the graph changes
Validates graph structure, logging problems without raising

run() compiles automatically when needed, so calling compile() yourself is optional.

run()

Execute the workflow:

result = workflow.run(
    task="Process data pipeline",
    img=None,  # Optional image input
)

run() also accepts on_node_complete (fired as (node_id, output) right after each node finishes, before its layer completes) and streaming_callback (fired as (node_id, token) for every token an agent generates). A callback passed to run() overrides the instance-level on_node_complete set in the constructor.

def on_done(node_id: str, output) -> None:
    print(f"[{node_id}] finished")

result = workflow.run(
    task="Process data pipeline",
    on_node_complete=on_done,
    streaming_callback=lambda node_id, token: print(token, end=""),
)

Return shape: with max_loops == 1 (the default), run() returns a Dict[str, Any] with every node’s output, keyed by node ID. With max_loops > 1, it returns per-loop results keyed as {node_id}_loop_{loop_number} plus the final loop’s results under the plain node_id keys. A node that raises does not stop the run. Its output becomes an "[ERROR] Agent <name> failed: ..." string and downstream nodes still execute.

Nested workflows

Add a GraphWorkflow as a node to reuse a whole sub-pipeline as one step. Its node ID is the inner workflow’s name. The inner workflow receives the outer node’s prompt as its task, and its results are joined into a single output for downstream nodes.

research = GraphWorkflow(name="ResearchStage")
research.add_node(data_collector)
research.add_node(processor)
research.add_edge("DataCollector", "Processor")

pipeline = GraphWorkflow(name="Pipeline")
pipeline.add_node(research)
pipeline.add_node(analyzer)
pipeline.add_edge("ResearchStage", "Analyzer")

result = pipeline.run("Process Q4 sales data")

Token usage

workflow.usage sums provider token usage over every agent in the graph, subgraphs included. An agent that backs several nodes is counted once. Totals are lifetime totals, so they grow across runs. See Token usage.

result = workflow.run("Process Q4 sales data")
print(workflow.usage)
# {"input_tokens": ..., "output_tokens": ..., "cached_tokens": ..., "reasoning_tokens": ..., "total_tokens": ...}

Checkpoints

Set checkpoint_dir to write each layer’s outputs to disk as it finishes. If a run crashes, run the same task again and completed layers are restored instead of re-executed:

workflow = GraphWorkflow(name="Data-Pipeline", checkpoint_dir="./checkpoints")
# ... add nodes and edges ...

result = workflow.run("Process Q4 sales data")

# Delete this task's checkpoint files once you no longer need them
workflow.clear_checkpoints("Process Q4 sales data")

Checkpoints are keyed on the exact task string, so a different task starts from scratch.

Use Cases

Software Build Pipeline

compile_agent = Agent(agent_name="Compiler", ...)
test_agent = Agent(agent_name="Tester", ...)
lint_agent = Agent(agent_name="Linter", ...)
security_agent = Agent(agent_name="SecurityScanner", ...)
package_agent = Agent(agent_name="Packager", ...)
deploy_agent = Agent(agent_name="Deployer", ...)

build_pipeline = GraphWorkflow(name="CI-CD-Pipeline")

# Add all nodes
for agent in [compile_agent, test_agent, lint_agent, security_agent, package_agent, deploy_agent]:
    build_pipeline.add_node(agent)

# Define dependencies
build_pipeline.add_edge("Compiler", "Tester")
build_pipeline.add_edges_from_source("Compiler", ["Linter", "SecurityScanner"])
build_pipeline.add_edges_to_target(["Tester", "Linter", "SecurityScanner"], "Packager")
build_pipeline.add_edge("Packager", "Deployer")

build_pipeline.compile()
result = build_pipeline.run("Build and deploy version 2.0")

Data Science Pipeline

edges = [
    # Data collection phase
    ("APICollector", "DataValidator"),
    ("DatabaseCollector", "DataValidator"),
    
    # Parallel processing
    ("DataValidator", ["Cleaner", "Transformer", "FeatureEngineer"]),
    
    # Convergence for modeling
    (["Cleaner", "Transformer", "FeatureEngineer"], "ModelTrainer"),
    
    # Evaluation and deployment
    ("ModelTrainer", "ModelEvaluator"),
    ("ModelEvaluator", "Deployer"),
]

ml_pipeline = GraphWorkflow.from_spec(
    agents=[api_collector, db_collector, validator, cleaner, transformer, 
            feature_eng, trainer, evaluator, deployer],
    edges=edges,
)

result = ml_pipeline.run("Train and deploy customer churn model")

Content Creation Workflow

# Diamond pattern: Research -> (Write, Design) -> Review
edges = [
    ("Researcher", ["Writer", "Designer"]),
    (["Writer", "Designer"], "Reviewer"),
    ("Reviewer", "Publisher"),
]

content_workflow = GraphWorkflow.from_spec(
    agents=[researcher, writer, designer, reviewer, publisher],
    edges=edges,
)

result = content_workflow.run("Create marketing campaign for product launch")

Entry and Exit Points

Auto-Detection

workflow = GraphWorkflow()
workflow.add_node(agent1)
workflow.add_node(agent2)
workflow.add_node(agent3)
workflow.add_edge("agent1", "agent2")
workflow.add_edge("agent2", "agent3")

workflow.compile()

# Automatically detects:
# - entry_points: ["agent1"] (no incoming edges)
# - end_points: ["agent3"] (no outgoing edges)

Manual Setting

workflow.set_entry_points(["StartAgent"])
workflow.set_end_points(["FinalAgent"])

Backend Selection

NetworkX (Default)

workflow = GraphWorkflow(backend="networkx")

Benefits:

Pure Python
Rich ecosystem
Extensive algorithms
Easy debugging

Rustworkx (Performance)

workflow = GraphWorkflow(backend="rustworkx")

Benefits:

Rust-based performance
Faster graph operations
Lower memory usage
Better for large graphs

Requires: pip install rustworkx

Topological Execution

The workflow executes in topological layers:

# Graph:
# A -> B, C
# B -> D
# C -> D

# Execution layers:
# Layer 0: [A]
# Layer 1: [B, C]  (parallel)
# Layer 2: [D]

Parallel execution within layers using ThreadPoolExecutor.

Graph Validation

Cycle Detection

compile() never raises on a cycle by itself. It runs the same structural checks as validate() and only logs a warning; the cyclic nodes are placed in a final layer. To turn cycle detection into an exception, call validate(raise_on_error=True) explicitly: a detected cycle counts as a “serious warning” that marks the workflow invalid (unless auto_fix=True), so raise_on_error=True will raise ValueError for it.

workflow.add_edge("A", "B")
workflow.add_edge("B", "C")
workflow.add_edge("C", "A")  # Creates cycle
workflow.compile()  # succeeds; cycles only warn here

result = workflow.validate()
# "cycles" is present only when at least one cycle was found
if result.get("cycles"):
    print(f"Cycle(s) detected: {result['cycles']}")

try:
    workflow.validate(raise_on_error=True)
except ValueError as e:
    print(f"Invalid workflow: {e}")

Dependency Validation

validate() checks for (each reported as a warning or error in the returned dict):

Referenced nodes existing / valid agent instances on every node (error)
Isolated nodes (no incoming or outgoing edges)
Cyclic dependencies (detected via simple_cycles(), returned under result["cycles"])
Unreachable nodes (not reachable from any entry point)
Dead-end nodes (cannot reach any end point)
Missing entry points / end points

result = workflow.validate(auto_fix=True)
print(result["errors"], result["warnings"], result["fixed"])

Performance Optimization

Compilation Caching

# First run: compiles
result1 = workflow.run("Task 1")

# Subsequent runs: uses cache
result2 = workflow.run("Task 2")
result3 = workflow.run("Task 3")

# Changing the graph invalidates the cache; the next run recompiles
workflow.add_node(reporter)
workflow.add_edge("Analyzer", "Reporter")
result4 = workflow.run("Task 4")

Concurrent Node Addition

# Add many nodes efficiently
workflow.add_nodes(
    agents=agent_list,
    batch_size=10,  # Process in batches
)

Best Practices

Graph Design: Keep graphs acyclic - use multiple workflows for iterative processes

Clear Dependencies: Only add edges for true dependencies
Maximize Parallelism: Let independent nodes run concurrently
Compilation: Call compile() early to surface validation warnings; run() compiles on its own otherwise
Entry/Exit Points: Let auto-detection work unless specific control needed
Backend Choice: Use Rustworkx for large graphs (>100 nodes)

Graph compilation is cached. Adding nodes or edges, or setting entry or end points, invalidates the cache, and the next run() recompiles automatically.

Error Handling

try:
    workflow.add_edge("NonExistentAgent", "TargetAgent")
except ValueError as e:
    print(f"Invalid edge: {e}")
    # Source or target node doesn't exist

try:
    workflow.validate(raise_on_error=True)
except ValueError as e:
    print(f"Invalid workflow: {e}")
    # Cycle detected or invalid graph structure

Visualization

With Graphviz installed:

# Requires: pip install graphviz

path = workflow.visualize(
    format="png",     # "png", "svg", "pdf", or "dot"
    view=True,         # open the file after generating it
    engine="dot",      # graphviz layout engine
    show_summary=False,
)

visualize() renders nodes and edges with Graphviz (auto-detecting fan-out/fan-in patterns for clearer styling) and returns the path to the generated file. It raises ImportError if graphviz is not installed. For a dependency-free text view, use workflow.visualize_simple(), which returns an ASCII representation of the graph.

Agent Rearrange - Simpler flow patterns
Sequential Workflow - Linear execution
Concurrent Workflow - Pure parallel
Hierarchical Swarm - Director coordination

​When to Use

​Key Features

​Basic Example

​Creating from Spec

​Advanced Edge Patterns

​Fan-Out Pattern

​Fan-In Pattern

​Parallel Chain Pattern

​Tuple-Based Patterns

​Key Parameters

​Methods

​add_node()

​add_nodes()

​add_edge()

​compile()

​run()

​Nested workflows

​Token usage

​Checkpoints

​Use Cases

​Software Build Pipeline

​Data Science Pipeline

​Content Creation Workflow

​Entry and Exit Points

​Auto-Detection

​Manual Setting

​Backend Selection

​NetworkX (Default)

​Rustworkx (Performance)

​Topological Execution

​Graph Validation

​Cycle Detection

​Dependency Validation

​Performance Optimization

​Compilation Caching

​Concurrent Node Addition

​Best Practices

​Error Handling

​Visualization

​Related Architectures

When to Use

Key Features

Basic Example

Creating from Spec

Advanced Edge Patterns

Fan-Out Pattern

Fan-In Pattern

Parallel Chain Pattern

Tuple-Based Patterns

Key Parameters

Methods

add_node()

add_nodes()

add_edge()

compile()

run()

Nested workflows

Token usage

Checkpoints

Use Cases

Software Build Pipeline

Data Science Pipeline

Content Creation Workflow

Entry and Exit Points

Auto-Detection

Manual Setting

Backend Selection

NetworkX (Default)

Rustworkx (Performance)

Topological Execution

Graph Validation

Cycle Detection

Dependency Validation

Performance Optimization

Compilation Caching

Concurrent Node Addition

Best Practices

Error Handling

Visualization

Related Architectures