GraphWorkflow - swarms

Overview

The GraphWorkflow class represents a workflow graph where each node is an agent. It provides sophisticated capabilities for building, executing, and visualizing Directed Acyclic Graph (DAG) workflows with automatic parallel execution, topological sorting, and compilation optimization.

Key Features

DAG-Based Workflows: Build complex directed acyclic graphs of agent execution
Automatic Parallelization: Automatically execute independent agents in parallel
Topological Execution: Agents execute in topologically sorted layers
Compilation Optimization: Pre-compute execution plans for faster multi-loop execution
Multiple Graph Backends: Support for NetworkX and Rustworkx backends
Visualization: Generate visual representations of workflows using Graphviz
Fan-Out/Fan-In Patterns: Easy creation of parallel processing patterns
JSON Serialization: Save and load workflows to/from JSON
Async Support: Asynchronous execution for non-blocking operations
Cycle Detection: Automatic detection and reporting of cycles
Nested Workflows: Run a whole GraphWorkflow as a single node
Checkpoints: Resume an interrupted run from its last completed layer
Token Usage: Sum token spend across every node and subgraph with usage

Installation

pip install -U swarms

# For visualization support
pip install graphviz

# For Rustworkx backend (optional, faster for large graphs)
pip install rustworkx

Class Definition

class GraphWorkflow:
    def __init__(
        self,
        id: Optional[str] = None,
        name: Optional[str] = "Graph-Workflow-01",
        description: Optional[str] = "A customizable workflow system for orchestrating and coordinating multiple agents.",
        nodes: Optional[Dict[str, Node]] = None,
        edges: Optional[List[Edge]] = None,
        entry_points: Optional[List[str]] = None,
        end_points: Optional[List[str]] = None,
        max_loops: int = 1,
        task: Optional[str] = None,
        auto_compile: bool = True,
        verbose: bool = False,
        backend: str = "networkx",
        checkpoint_dir: Optional[str] = None,
        on_node_complete: Optional[Callable[[str, Any], None]] = None,
        max_parallel_nodes: Optional[int] = None,
    )

Parameters

Optional[str]

default:"auto-generated"

Unique identifier for the workflow. Auto-generated via generate_id("graph-workflow") if not provided, producing graph-workflow-<32 hex chars>.

Optional[str]

default:"Graph-Workflow-01"

Human-readable name for the workflow

Optional[str]

default:"A customizable workflow system..."

Description of the workflow’s purpose

Optional[Dict[str, Node]]

default:"None"

Dictionary of nodes in the graph, where the key is the node ID and the value is the Node object

Optional[List[Edge]]

default:"None"

List of edges in the graph, where each edge is represented by an Edge object

Optional[List[str]]

default:"None"

List of node IDs that serve as entry points to the graph. Auto-detected if not provided.

Optional[List[str]]

default:"None"

List of node IDs that serve as end points of the graph. Auto-detected if not provided.

int

default:"1"

Maximum number of times to execute the workflow

Optional[str]

default:"None"

The task to be executed by the workflow

bool

default:"True"

Whether to automatically compile the graph for optimization

bool

default:"False"

Whether to enable verbose logging

str

default:"networkx"

Graph backend to use. Options: “networkx”, “rustworkx” (Rustworkx is faster for large graphs)

Optional[str]

default:"None"

Directory for per-layer checkpoints. After each layer finishes, its outputs are written to {checkpoint_dir}/{task_key}_layer_{n}.json, where task_key is a hash of the task string. Running the same task again skips every layer that already has a checkpoint. Subgraph nodes checkpoint under {checkpoint_dir}/{node_id} unless they set their own directory. Remove old files with clear_checkpoints(task).

Optional[Callable[[str, Any], None]]

default:"None"

Callback fired after each node finishes, receiving (node_id, result). Can also be passed per-call to run().

Optional[int]

default:"None"

Maximum number of nodes to execute concurrently within a layer. Defaults to max(1, int(cpu_cores * 0.95)) when None. The run’s thread pool is also capped at the size of the widest layer.

Methods

Graph Construction

`add_node(agent, **kwargs)`

Add an agent, or a nested GraphWorkflow, as a node in the workflow graph. An agent node’s ID is its agent_name. A nested workflow becomes a SUBGRAPH node whose ID is the inner workflow’s name; it is compiled on insertion if needed. Raises ValueError if a node with the same ID already exists.

Union[Agent, GraphWorkflow]

required

The agent or inner workflow to add. Pass the object itself, not a Node; to supply prebuilt Node objects, use the nodes constructor argument or from_spec().

Any

Additional keyword arguments for the node, such as metadata

`add_nodes(agents, batch_size=10, **kwargs)`

Add multiple agents to the workflow graph concurrently in batches.

List[Agent]

required

List of agents to add

int

default:"10"

Number of agents to add concurrently in a batch

`add_edge(edge_or_source, target=None, **kwargs)`

Add an edge by Edge object or by passing node objects/ids.

Union[Edge, Node, Agent, str]

required

Either an Edge object or the source node/id

Optional[Union[Node, Agent, str]]

Target node/id (required if edge_or_source is not an Edge)

`add_edges_from_source(source, targets, **kwargs)`

Add multiple edges from a single source to multiple targets (fan-out pattern).

Union[Node, Agent, str]

required

Source node/id that will send output to multiple targets

List[Union[Node, Agent, str]]

required

List of target node/ids that will receive the source output in parallel

List[Edge]

List of created Edge objects

`add_edges_to_target(sources, target, **kwargs)`

Add multiple edges from multiple sources to a single target (fan-in pattern).

List[Union[Node, Agent, str]]

required

List of source node/ids that will send output to the target

Union[Node, Agent, str]

required

Target node/id that will receive all source outputs

List[Edge]

List of created Edge objects

`add_parallel_chain(sources, targets, **kwargs)`

Create a parallel processing chain (full mesh connection).

List[Union[Node, Agent, str]]

required

List of source node/ids

List[Union[Node, Agent, str]]

required

List of target node/ids

List[Edge]

List of created Edge objects

Execution

`run(task=None, img=None, on_node_complete=None, streaming_callback=None, *args, **kwargs)`

Run the workflow graph with optimized parallel agent execution. The graph is compiled first if it is not already. Each node receives the task plus one chat turn per predecessor output. When max_loops > 1, end-point outputs from each loop are fed as additional context into the next loop so agents can iteratively refine their results. A node that raises does not stop the run: its output becomes an "[ERROR] Agent <name> failed: ..." string, and downstream nodes still run. A subgraph node’s output is the inner workflow’s results joined as [node_id]: output blocks.

Optional[str]

Task to execute. Uses self.task if not provided.

Optional[str]

Optional image path for multimodal tasks

Optional[Callable[[str, Any], None]]

default:"None"

Callback fired immediately when each agent finishes, before the layer completes. Receives (node_id, output). Takes precedence over the instance-level callback set in __init__.

Optional[Callable[[str, str], None]]

default:"None"

Callback fired for every token as agents generate output in real time. Receives (node_id, token).

Dict[str, Any]

Execution results for every node, not only the end points, keyed by node ID. When max_loops == 1, returns the single loop’s results. When max_loops > 1, returns a dict with per-loop results keyed as {node_id}_loop_{loop_number} plus the final loop’s results under the plain node_id keys.

`arun(task=None, *args, **kwargs)`

Async version of run(). Runs run() in a worker thread with asyncio.to_thread, forwarding any extra arguments.

Optional[str]

Task to execute. Uses self.task if not provided.

Dict[str, Any]

Execution results from all nodes

`usage` (property)

Provider token usage summed over every agent in the graph, including the agents inside nested subgraphs at any depth. An agent that backs several nodes, or appears in both this graph and a subgraph, is counted once. See Token usage.

dict

A new dict with input_tokens, output_tokens, cached_tokens, reasoning_tokens and total_tokens. Agent totals are lifetime totals, so the value grows across repeated run() / arun() calls and includes anything a shared agent spent in other workflows.

results = workflow.run("Research AI trends in healthcare")
print(workflow.usage["total_tokens"])

Compilation and Optimization

`compile()`

Pre-compute expensive operations for faster execution: auto-set missing entry and end points, compute topological layers and run the structural checks from validate(), logging any problems without raising. Results are cached until the graph changes; adding a node or edge, or setting entry or end points, invalidates the cache and the next run() recompiles.

`get_compilation_status()`

Get detailed compilation status information.

Dict[str, Any]

Compilation status including cache state, timestamps, and performance metrics

Entry/Exit Points

`set_entry_points(entry_points)`

Set the entry points for the workflow.

List[str]

required

List of node IDs to serve as entry points

`set_end_points(end_points)`

Set the end points for the workflow.

List[str]

required

List of node IDs to serve as end points

`auto_set_entry_points()`

Automatically set entry points to nodes with no incoming edges.

`auto_set_end_points()`

Automatically set end points to nodes with no outgoing edges.

Visualization

`visualize(format="png", view=True, engine="dot", show_summary=False)`

Visualize the workflow graph using Graphviz.

str

default:"png"

Output format: ‘png’, ‘svg’, ‘pdf’, ‘dot’

bool

default:"True"

Whether to open the visualization after creation

str

default:"dot"

Graphviz layout engine: ‘dot’, ‘neato’, ‘fdp’, ‘sfdp’, ‘twopi’, ‘circo’

bool

default:"False"

Whether to print parallel processing summary

str

Path to the generated visualization file

`visualize_simple()`

Simple text-based visualization for environments without Graphviz.

str

Text representation of the workflow

Serialization

Workflows serialize in two shapes. A shallow spec stores the topology only: each node names its agent by agent_name, so you supply the live agents when loading. A deep export embeds each agent’s to_dict() output plus export metadata and metrics.

`to_dict(shallow=True, include_conversation=False, include_runtime_state=False)`

Serialize the workflow to a dict. This is the builder behind to_spec(), to_json(), save() and save_to_file().

bool

default:"True"

True for the sorted, diffable topology spec. False for a deep export with agents embedded.

bool

default:"False"

Deep shape only. Embed the conversation history.

bool

default:"False"

Deep shape only. Embed the compilation state.

`save(path, shallow=True, overwrite=True, include_conversation=False, include_runtime_state=False)`

Write to_dict(...) to a JSON file, creating parent directories. Raises FileExistsError if the file exists and overwrite=False. Pair with load().

str

The path written

`load(path, agent_registry=None, restore_runtime_state=False, **kwargs)` (classmethod)

Read a file written by save(), save_spec() or save_to_file(), detecting which shape it holds. Raises FileNotFoundError if the file does not exist.

Optional[Dict[str, Agent]]

default:"None"

Mapping from agent_name to live Agent. Required for a shallow spec (ValueError without it). Optional for a deep export; without it, deep-loaded nodes hold plain dicts instead of runnable agents.

bool

default:"False"

Deep exports only. Restore the saved compilation state.

GraphWorkflow

The reconstructed workflow

`to_json(fast=True, include_conversation=False, include_runtime_state=False)`

Serialize the workflow to a deep JSON export with agents embedded.

bool

default:"True"

Ignored; kept for backward compatibility

bool

default:"False"

Whether to include conversation history

bool

default:"False"

Whether to include runtime state like compilation info

str

JSON representation of the workflow

`from_json(json_str, restore_runtime_state=False, agent_registry=None)` (classmethod)

Deserialize a workflow from a deep JSON export. Nodes keep their exported IDs, so edges resolve. Raises ValueError if json_str is not valid JSON.

str

required

JSON string produced by to_json()

bool

default:"False"

Whether to restore runtime state

Optional[Dict[str, Agent]]

default:"None"

Mapping from agent_name to live Agent. Nodes whose agent is not in the registry hold the exported agent dict, which cannot run.

GraphWorkflow

A new GraphWorkflow instance

`save_to_file(filepath, include_conversation=False, include_runtime_state=False, overwrite=False)`

Save a deep export (agents embedded) to a JSON file. Appends .json when the path lacks it.

str

required

Path to save the JSON file

bool

default:"False"

Whether to overwrite an existing file. When False, an existing file raises FileExistsError.

str

Path to the saved file

`load_from_file(filepath, restore_runtime_state=False)` (classmethod)

Load a workflow from a JSON file. Equivalent to load(filepath, restore_runtime_state=...) with no agent registry, so a deep export comes back with plain-dict agents. Use load(path, agent_registry=...) when you need a runnable workflow.

str

required

Path to the JSON file

GraphWorkflow

Loaded workflow instance

Validation

`validate(auto_fix=False, raise_on_error=False)`

Validate the workflow structure, checking for isolated nodes, cyclic dependencies, unreachable nodes, dead-end nodes, and missing entry/end points.

bool

default:"False"

Whether to automatically fix simple issues (e.g. auto-setting missing entry/exit points, adding unreachable nodes to entry points)

bool

default:"False"

When True, raises a ValueError if validation determines the workflow is invalid. Defaults to False so existing callers that inspect the returned dict are unaffected.

Dict[str, Any]

Dictionary containing is_valid, errors, warnings, fixed, and (when cycles are found) cycles

Checkpoints and Spec Serialization

`clear_checkpoints(task)`

Delete all checkpoint files written for a specific task. Call after a workflow run completes successfully to reclaim disk space. Requires checkpoint_dir to have been set.

str

required

The task string whose checkpoints should be removed; must match the string passed to run() exactly

int

Number of checkpoint files deleted

Raises:

ValueError: If checkpoint_dir was not set on this workflow instance

`to_spec()`

Serialize the workflow topology to a lightweight plain-dict spec. Unlike to_json(), this does not serialize the Agent objects themselves — it only records each agent’s agent_name, so the spec can be version-controlled and shared.

Dict[str, Any]

Dict with name, description, max_loops, nodes, edges, entry_points, end_points

`save_spec(path)`

Save the topology spec produced by to_spec() to a JSON file.

str

required

Filesystem path to write the JSON file to

`from_topology_spec(spec, agent_registry, **kwargs)` (classmethod)

Reconstruct a GraphWorkflow from a topology spec (as produced by to_spec()/save_spec()) and an agent registry supplying the live Agent objects for each node.

Dict[str, Any]

required

A topology spec as returned by to_spec() or loaded from a file written by save_spec()

Dict[str, Agent]

required

Mapping from agent_name to live Agent instance

GraphWorkflow

A new GraphWorkflow instance

`export_summary()`

Generate a human-readable summary of the workflow structure for inspection.

Dict[str, Any]

Summary dictionary describing the workflow’s nodes, edges, and layers

Factory Method

`from_spec(agents, edges, entry_points=None, end_points=None, task=None, **kwargs)` (classmethod)

Construct a workflow from a list of agents and connections.

List[Union[Agent, Node]]

required

List of agents or Node objects

List[Union[Edge, Tuple[Any, Any]]]

required

List of edges or edge tuples. Supports fan-out, fan-in, and parallel chain patterns.

Optional[List[str]]

List of entry point node IDs

Optional[List[str]]

List of end point node IDs

Optional[str]

Task to be executed by the workflow

GraphWorkflow

A new GraphWorkflow instance

Usage Examples

Basic Sequential Graph

from swarms import Agent, GraphWorkflow

# Create agents
researcher = Agent(
    agent_name="Researcher",
    model_name="claude-sonnet-4-6",
    system_prompt="Research and gather information"
)

analyzer = Agent(
    agent_name="Analyzer",
    model_name="claude-sonnet-4-6",
    system_prompt="Analyze the research findings"
)

writer = Agent(
    agent_name="Writer",
    model_name="claude-sonnet-4-6",
    system_prompt="Write based on analysis"
)

# Create workflow
workflow = GraphWorkflow(
    name="Research-Analyze-Write",
    description="Sequential research workflow"
)

# Add nodes
workflow.add_node(researcher)
workflow.add_node(analyzer)
workflow.add_node(writer)

# Add edges
workflow.add_edge("Researcher", "Analyzer")
workflow.add_edge("Analyzer", "Writer")

# Compile and run
workflow.compile()
results = workflow.run("Research AI trends in healthcare")

Fan-Out Pattern (Parallel Specialists)

# Create multiple specialist agents
technical = Agent(agent_name="Technical", model_name="gpt-5.4-mini", system_prompt="Technical analysis")
business = Agent(agent_name="Business", model_name="gpt-5.4-mini", system_prompt="Business analysis")
user_exp = Agent(agent_name="UX", model_name="gpt-5.4-mini", system_prompt="User experience analysis")
synthesizer = Agent(agent_name="Synthesizer", model_name="gpt-5.4", system_prompt="Synthesize all analyses")

workflow = GraphWorkflow(name="Multi-Perspective")

# Add all nodes
for agent in [researcher, technical, business, user_exp, synthesizer]:
    workflow.add_node(agent)

# Fan-out from researcher to specialists
workflow.add_edges_from_source(
    "Researcher",
    ["Technical", "Business", "UX"]
)

# Fan-in to synthesizer
workflow.add_edges_to_target(
    ["Technical", "Business", "UX"],
    "Synthesizer"
)

results = workflow.run("Analyze AI chatbot implementation")

Complex DAG Workflow

# Create a complex multi-stage workflow
data_collector = Agent(agent_name="DataCollector", model_name="gpt-5.4-mini")
preprocessor = Agent(agent_name="Preprocessor", model_name="gpt-5.4-mini")
model_a = Agent(agent_name="ModelA", model_name="gpt-5.4-mini")
model_b = Agent(agent_name="ModelB", model_name="gpt-5.4-mini")
model_c = Agent(agent_name="ModelC", model_name="gpt-5.4-mini")
ensemble = Agent(agent_name="Ensemble", model_name="gpt-5.4")
validator = Agent(agent_name="Validator", model_name="gpt-5.4")

workflow = GraphWorkflow(name="ML-Pipeline")

# Add all nodes
agents = [data_collector, preprocessor, model_a, model_b, model_c, ensemble, validator]
workflow.add_nodes(agents, batch_size=5)

# Build DAG
workflow.add_edge("DataCollector", "Preprocessor")
workflow.add_edges_from_source("Preprocessor", ["ModelA", "ModelB", "ModelC"])
workflow.add_edges_to_target(["ModelA", "ModelB", "ModelC"], "Ensemble")
workflow.add_edge("Ensemble", "Validator")

# Visualize before running
workflow.visualize(show_summary=True)

# Execute
results = workflow.run("Predict customer churn")

Using from_spec for Quick Construction

# Quick workflow construction using from_spec
agents = [researcher, analyzer, writer]

edges = [
    ("Researcher", "Analyzer"),
    ("Analyzer", "Writer")
]

workflow = GraphWorkflow.from_spec(
    agents=agents,
    edges=edges,
    name="Quick-Workflow",
    task="Analyze quantum computing trends"
)

results = workflow.run()  # Uses task from initialization

Advanced Edge Patterns in from_spec

agents = [collector, analyst1, analyst2, analyst3, synthesizer]

edges = [
    # Simple edge
    ("Collector", "Analyst1"),
    
    # Fan-out
    ("Collector", ["Analyst2", "Analyst3"]),
    
    # Fan-in
    (["Analyst1", "Analyst2", "Analyst3"], "Synthesizer"),
]

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

Async Execution

import asyncio

async def run_workflow():
    workflow = GraphWorkflow(name="Async-Workflow")
    workflow.add_nodes([agent1, agent2, agent3])
    workflow.add_edge("agent1", "agent2")
    workflow.add_edge("agent2", "agent3")
    
    results = await workflow.arun("Process this task asynchronously")
    return results

results = asyncio.run(run_workflow())

Save and Load Workflows

Agents hold live LLM clients, so a saved file can’t recreate them. Save the topology and pass the live agents back in when you load:

# Topology only: each node stores its agent_name
workflow.save("my_workflow.json")

loaded_workflow = GraphWorkflow.load(
    "my_workflow.json",
    agent_registry={"Researcher": researcher, "Analyzer": analyzer, "Writer": writer},
)

results = loaded_workflow.run("New task")

A deep export also round-trips, with the same registry needed to make it runnable:

workflow.save_to_file(
    "my_workflow_full.json",
    include_runtime_state=True,
    overwrite=True,
)

loaded_workflow = GraphWorkflow.load(
    "my_workflow_full.json",
    agent_registry={"Researcher": researcher, "Analyzer": analyzer, "Writer": writer},
    restore_runtime_state=True,
)

load_from_file() takes no agent registry, so the workflow it returns holds the exported agent dicts rather than agents. Running it produces an [ERROR] output for every node. Use load(path, agent_registry=...) to get a runnable workflow.

Using Rustworkx Backend

# Use Rustworkx for better performance on large graphs
workflow = GraphWorkflow(
    name="Large-Workflow",
    backend="rustworkx",  # Faster for large graphs
    verbose=True
)

# Add many nodes
workflow.add_nodes(list_of_100_agents, batch_size=20)

# Build complex graph
# ...

results = workflow.run("Complex task")

Workflow Validation

# Validate workflow before execution
validation = workflow.validate(auto_fix=True)

if validation["is_valid"]:
    print("Workflow is valid!")
    results = workflow.run(task)
else:
    print("Errors:", validation["errors"])
    print("Warnings:", validation["warnings"])

Compilation for Multi-Loop Execution

workflow = GraphWorkflow(
    name="Multi-Loop-Workflow",
    max_loops=10,  # Will run 10 times
    auto_compile=True  # Compilation cached for all loops
)

workflow.add_nodes([agent1, agent2, agent3])
workflow.add_edges_from_source("agent1", ["agent2", "agent3"])

# Compilation is cached and reused across all loops
results = workflow.run("Task to repeat 10 times")

# Check compilation status
status = workflow.get_compilation_status()
print(f"Compiled: {status['is_compiled']}")
print(f"Layers: {status['cached_layers_count']}")

Custom Visualization

# Generate different visualization formats
workflow.visualize(
    format="svg",
    engine="fdp",  # Force-directed layout
    view=False,
    show_summary=True
)

# Simple text visualization (no Graphviz needed)
text_viz = workflow.visualize_simple()
print(text_viz)

Node and Edge Classes

Node

A node wraps an Agent (NodeType.AGENT) or a nested GraphWorkflow (NodeType.SUBGRAPH). When id is omitted, it defaults to the agent’s agent_name or the workflow’s name.

from swarms import Node, NodeType

# Create node from agent
node = Node.from_agent(agent)

# Create node from a nested workflow
sub_node = Node.from_subgraph(inner_workflow, node_id="research-stage")

# Create node manually
node = Node(
    id="my-node",
    type=NodeType.AGENT,
    agent=agent,
    metadata={"custom_key": "custom_value"}
)

# Prebuilt nodes go in through the constructor; add_node() takes the agent itself
workflow = GraphWorkflow(nodes={node.id: node})

Edge

from swarms import Edge

# Create edge from nodes
edge = Edge.from_nodes(source_agent, target_agent)

# Create edge from IDs
edge = Edge(source="agent1", target="agent2")

# With metadata
edge = Edge(
    source="agent1",
    target="agent2",
    metadata={"weight": 1.0}
)

Nested workflows (subgraphs)

Pass a GraphWorkflow to add_node() to run it as a single node. The inner workflow receives the outer node’s prompt as its task, and its results are joined into one output for downstream nodes. usage and to_spec() include subgraphs.

research = GraphWorkflow(name="ResearchStage")
research.add_node(researcher)
research.add_node(analyzer)
research.add_edge("Researcher", "Analyzer")

pipeline = GraphWorkflow(name="Pipeline")
pipeline.add_node(research)  # node id "ResearchStage"
pipeline.add_node(writer)
pipeline.add_edge("ResearchStage", "Writer")

results = pipeline.run("Research AI trends in healthcare")

Best Practices

Compilation: run() compiles on demand; call compile() yourself to surface validation warnings before the first run
Visualization: Use visualize() to understand your workflow structure
Validation: Validate workflows before production deployment
Entry/Exit Points: Let the system auto-detect entry/exit points or set them explicitly
Backend Selection: Use Rustworkx for large graphs (>100 nodes)
Error Handling: Wrap execution in try-except blocks
Batching: Use add_nodes() with appropriate batch sizes for many agents
Caching: Enable auto_compile for multi-loop workflows
Save Progress: Save complex workflows with save() and reload them with load(path, agent_registry=...)
Cost Tracking: Read workflow.usage after a run to see token spend across every node

Performance Characteristics

Parallel Execution: Independent agents in the same layer run concurrently
Thread Pool: One pool per run, sized to max_parallel_nodes (default ~95% of CPU cores) and capped at the widest layer
Compilation: Pre-computes execution plan, cached for multi-loop runs
Memory: Each agent maintains its own context
Graph Backend: Rustworkx is faster for large graphs (>100 nodes)

Common Patterns

Research Pipeline

edges = [
    ("DataCollector", "Researcher"),
    ("Researcher", ["Analyst1", "Analyst2", "Analyst3"]),
    (["Analyst1", "Analyst2", "Analyst3"], "Synthesizer")
]

Ensemble Decision Making

edges = [
    ("Coordinator", ["Expert1", "Expert2", "Expert3", "Expert4"]),
    (["Expert1", "Expert2", "Expert3", "Expert4"], "Aggregator"),
    ("Aggregator", "DecisionMaker")
]

Multi-Stage Review

edges = [
    ("Creator", "TechnicalReviewer"),
    ("TechnicalReviewer", "BusinessReviewer"),
    ("BusinessReviewer", "LegalReviewer"),
    ("LegalReviewer", "FinalApprover")
]

SequentialWorkflow: For simple sequential execution
ConcurrentWorkflow: For parallel execution on same task
AgentRearrange: For flexible flow-based orchestration
Agent: The base agent class used in workflows

​Overview

​Key Features

​Installation

​Class Definition

​Parameters

​Methods

​Graph Construction

​add_node(agent, **kwargs)

​add_nodes(agents, batch_size=10, **kwargs)

​add_edge(edge_or_source, target=None, **kwargs)

​add_edges_from_source(source, targets, **kwargs)

​add_edges_to_target(sources, target, **kwargs)

​add_parallel_chain(sources, targets, **kwargs)

​Execution

​run(task=None, img=None, on_node_complete=None, streaming_callback=None, *args, **kwargs)

​arun(task=None, *args, **kwargs)

​usage (property)

​Compilation and Optimization

​compile()

​get_compilation_status()

​Entry/Exit Points

​set_entry_points(entry_points)

​set_end_points(end_points)

​auto_set_entry_points()

​auto_set_end_points()

​Visualization

​visualize(format="png", view=True, engine="dot", show_summary=False)

​visualize_simple()

​Serialization

​to_dict(shallow=True, include_conversation=False, include_runtime_state=False)

​save(path, shallow=True, overwrite=True, include_conversation=False, include_runtime_state=False)

​load(path, agent_registry=None, restore_runtime_state=False, **kwargs) (classmethod)

​to_json(fast=True, include_conversation=False, include_runtime_state=False)

​from_json(json_str, restore_runtime_state=False, agent_registry=None) (classmethod)

​save_to_file(filepath, include_conversation=False, include_runtime_state=False, overwrite=False)

​load_from_file(filepath, restore_runtime_state=False) (classmethod)

​Validation

​validate(auto_fix=False, raise_on_error=False)

​Checkpoints and Spec Serialization

​clear_checkpoints(task)

​to_spec()

​save_spec(path)

​from_topology_spec(spec, agent_registry, **kwargs) (classmethod)

​export_summary()

​Factory Method

​from_spec(agents, edges, entry_points=None, end_points=None, task=None, **kwargs) (classmethod)

​Usage Examples

​Basic Sequential Graph

​Fan-Out Pattern (Parallel Specialists)

​Complex DAG Workflow

​Using from_spec for Quick Construction

​Advanced Edge Patterns in from_spec

​Async Execution

​Save and Load Workflows

​Using Rustworkx Backend

​Workflow Validation

​Compilation for Multi-Loop Execution

​Custom Visualization

​Node and Edge Classes

​Node

​Edge

​Nested workflows (subgraphs)

​Best Practices

​Performance Characteristics

​Common Patterns

​Research Pipeline

​Ensemble Decision Making

​Multi-Stage Review

​Related Classes

Overview

Key Features

Installation

Class Definition

Parameters

Methods

Graph Construction

`add_node(agent, **kwargs)`

`add_nodes(agents, batch_size=10, **kwargs)`

`add_edge(edge_or_source, target=None, **kwargs)`

`add_edges_from_source(source, targets, **kwargs)`

`add_edges_to_target(sources, target, **kwargs)`

`add_parallel_chain(sources, targets, **kwargs)`

Execution

`run(task=None, img=None, on_node_complete=None, streaming_callback=None, *args, **kwargs)`

`arun(task=None, *args, **kwargs)`

`usage` (property)

Compilation and Optimization

`compile()`

`get_compilation_status()`

Entry/Exit Points

`set_entry_points(entry_points)`

`set_end_points(end_points)`

`auto_set_entry_points()`

`auto_set_end_points()`

Visualization

`visualize(format="png", view=True, engine="dot", show_summary=False)`

`visualize_simple()`

Serialization

`to_dict(shallow=True, include_conversation=False, include_runtime_state=False)`

`save(path, shallow=True, overwrite=True, include_conversation=False, include_runtime_state=False)`

`load(path, agent_registry=None, restore_runtime_state=False, **kwargs)` (classmethod)

`to_json(fast=True, include_conversation=False, include_runtime_state=False)`

`from_json(json_str, restore_runtime_state=False, agent_registry=None)` (classmethod)

`save_to_file(filepath, include_conversation=False, include_runtime_state=False, overwrite=False)`

`load_from_file(filepath, restore_runtime_state=False)` (classmethod)

Validation

`validate(auto_fix=False, raise_on_error=False)`

Checkpoints and Spec Serialization

`clear_checkpoints(task)`

`to_spec()`

`save_spec(path)`

`from_topology_spec(spec, agent_registry, **kwargs)` (classmethod)

`export_summary()`

Factory Method

`from_spec(agents, edges, entry_points=None, end_points=None, task=None, **kwargs)` (classmethod)

Usage Examples

Basic Sequential Graph

Fan-Out Pattern (Parallel Specialists)

Complex DAG Workflow

Using from_spec for Quick Construction

Advanced Edge Patterns in from_spec

Async Execution

Save and Load Workflows

Using Rustworkx Backend

Workflow Validation

Compilation for Multi-Loop Execution

Custom Visualization

Node and Edge Classes

Node

Edge

Nested workflows (subgraphs)

Best Practices

Performance Characteristics

Common Patterns

Research Pipeline

Ensemble Decision Making

Multi-Stage Review

Related Classes