Building Graph-Based Agent Workflows with DiGraphBuilder and GraphFlow in AutoGen

Use AutoGen's DiGraphBuilder to declaratively construct directed agent workflows and GraphFlow to execute them, enabling complex patterns like parallel fan-outs, conditional branching, and cyclic loops with deterministic termination.

AutoGen's graph-based execution engine moves beyond simple round-robin chats, allowing you to orchestrate ChatAgents as structured directed workflows. By leveraging the DiGraphBuilder fluent API and the GraphFlow runtime in the microsoft/autogen repository, you can build sophisticated agent topologies with validation guarantees and fine-grained control over execution paths.

Core Components of Graph-Based Workflows

DiGraphBuilder

Located in python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_graph/_graph_builder.py, the DiGraphBuilder class provides a chainable interface for defining the topology of your agent workflow. It handles adding nodes (agents), connecting them with edges (execution paths), setting activation groups, and defining conditions (string or callable) for conditional routing. The builder also supports optional entry points via set_entry_point().

DiGraph Models

The underlying immutable structure is defined in python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_graph/_digraph_group_chat.py through Pydantic models:

• DiGraph: Container for the entire workflow structure
• DiGraphNode: Represents an agent node with properties like activation ("all" or "any")
• DiGraphEdge: Defines connections with check_condition logic for runtime evaluation

These models contain validation logic for start/leaf nodes, conditional-edge consistency, activation-group uniformity, and cycle-with-exit detection to prevent infinite loops.

GraphFlowManager and GraphFlow

Also in _digraph_group_chat.py, the GraphFlowManager serves as the runtime engine that drives graph execution. It maintains a ready queue of nodes whose dependencies are satisfied, tracks remaining parent counters, manages activation-group bookkeeping ("all" vs "any" logic), and applies edge conditions via DiGraphEdge.check_condition. It also implements termination handling for both natural graph completion and user-provided termination conditions.

The GraphFlow class acts as the high-level team interface that ties together the agents, the validated DiGraph, and the manager. It exposes the standard AutoGen team API including run() and run_stream() methods, making it interchangeable with other team implementations.

Constructing Graph-Based Agent Workflows

Sequential Execution (A → B → C)

The simplest pattern chains agents linearly, where each agent runs only after the previous completes:

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main():
    model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
    a = AssistantAgent("A", model_client=model_client, system_message="You are a helpful assistant.")
    b = AssistantAgent("B", model_client=model_client, system_message="Translate to Spanish.")
    c = AssistantAgent("C", model_client=model_client, system_message="Summarize in bullet points.")

    # Build the graph

    builder = DiGraphBuilder()
    builder.add_node(a).add_node(b).add_node(c)
    builder.add_edge(a, b).add_edge(b, c)
    graph = builder.build()

    # Create the team

    team = GraphFlow(
        participants=[a, b, c],
        graph=graph,
        termination_condition=MaxMessageTermination(5),
    )

    # Run

    result = await team.run(task="Explain quantum computing in one paragraph.")
    for msg in result.messages:
        print(f"{msg.source}: {msg.content}")

import asyncio
asyncio.run(main())

Parallel Fan-Out (A → B, C)

Execute multiple agents simultaneously after a single parent completes. Both agents receive the same context from the parent and run concurrently:

builder = DiGraphBuilder()
builder.add_node(a).add_node(b).add_node(c)
builder.add_edge(a, b).add_edge(a, c)   # Fan-out from A to both B and C

graph = builder.build()

team = GraphFlow(
    participants=[a, b, c],
    graph=graph,
    termination_condition=MaxMessageTermination(5),
)

# Both B and C execute in parallel after A finishes

Conditional Branching

Route execution based on message content using string conditions (substring matching) or callable conditions (lambda/functions):

builder = DiGraphBuilder()
builder.add_node(a).add_node(b).add_node(c)

# Using callable conditions

builder.add_edge(a, b, condition=lambda msg: "yes" in msg.to_model_text())
builder.add_edge(a, c, condition=lambda msg: "yes" not in msg.to_model_text())

# Or use string conditions for simple substring matching

builder.add_edge(a, b, condition="YES")
builder.add_edge(a, c, condition="NO")

graph = builder.build()

Cyclic Workflows with Exit Conditions

Create loops that terminate based on specific agent output. The validation logic requires at least one conditional edge in every cycle to prevent infinite execution:

builder = DiGraphBuilder()
builder.add_node(a).add_node(b).add_node(c)

# A → B (unconditional)

builder.add_edge(a, b)

# B → C when approved, otherwise loop back to A

builder.add_edge(b, c, condition=lambda msg: "APPROVE" in msg.to_model_text())
builder.add_edge(b, a, condition=lambda msg: "APPROVE" not in msg.to_model_text())

builder.set_entry_point(a)  # Optional; defaults to nodes without parents

graph = builder.build()

If you attempt to create a cycle without a conditional exit, graph.graph_validate() raises:

ValueError: Cycle detected without exit condition: A -> B -> A

Join Patterns with Activation Groups

Control when a node with multiple incoming edges executes using activation groups and activation conditions:

builder = DiGraphBuilder()
builder.add_node(a).add_node(b).add_node(c)

# D fires when ANY parent is ready (race condition)

builder.add_node(d, activation="any")

builder.add_edge(a, d, activation_group="g1")
builder.add_edge(b, d, activation_group="g1")
graph = builder.build()

Both edges share activation_group="g1" while node D specifies activation="any". The GraphFlowManager enqueues D as soon as either A or B finishes. Use activation="all" (the default) to require all parents to complete before the node runs.

Validation and Runtime Execution

The DiGraphBuilder produces a DiGraph model that undergoes strict validation via graph_validate() before execution. According to the source code in python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_graph/_digraph_group_chat.py, this validation enforces:

• Start nodes: At least one node must have no incoming edges (or an explicit entry point set via set_entry_point()).
• Leaf nodes: At least one node must have no outgoing edges to ensure natural termination.
• Edge consistency: No node can mix conditional and unconditional outgoing edges.
• Activation group consistency: All edges sharing an activation_group must have the same activation_condition ("all" or "any").
• Cycle safety: Any cycle in the graph must contain at least one conditional edge to prevent infinite loops.

During runtime, the GraphFlowManager (also in _digraph_group_chat.py) maintains:

• Ready queue: Nodes whose parent dependencies are satisfied.
• Remaining parent counters: Tracks how many parent edges still need to complete for each node.
• Activation group bookkeeping: Manages "all" vs "any" logic for multi-parent nodes.
• Condition evaluation: Applies DiGraphEdge.check_condition using either string substring matching or callable predicates.

When the ready queue empties, the manager emits a StopMessage indicating digraph execution is complete, and resets internal state to allow the team to be reused.

Summary

• DiGraphBuilder provides a fluent API in _graph_builder.py for constructing agent workflows with nodes (agents) and edges (execution paths).
• GraphFlow serves as the high-level team class that executes validated DiGraph structures using the GraphFlowManager runtime logic.
• Validation enforces structural integrity: start/leaf nodes, consistent activation groups, and mandatory exit conditions for cycles.
• Execution patterns include sequential chains, parallel fan-outs, conditional branching, cyclic loops with termination, and join patterns with activation groups ("all" vs "any").
• Condition evaluation supports both string substring matching and callable predicates for dynamic routing.

Frequently Asked Questions

What is the difference between DiGraphBuilder and GraphFlow?

DiGraphBuilder is a construction-time utility class that provides a fluent interface for defining the topology of your agent workflow. It handles adding nodes, connecting them with edges, setting activation groups, and defining conditions for conditional routing, ultimately producing a validated DiGraph model. GraphFlow is the runtime team class that actually executes the workflow; it takes the validated graph and participant agents, then orchestrates their execution using the GraphFlowManager to manage state, conditions, and termination.

How do I prevent infinite loops in cyclic graph workflows?

The validation logic in DiGraph.graph_validate() (located in _digraph_group_chat.py) automatically prevents infinite loops by requiring that every cycle in the graph contains at least one conditional edge. When building your graph, ensure that at least one edge in every loop uses a condition parameter (either a string for substring matching or a callable function). If you attempt to build a cycle without such an exit condition, the builder raises ValueError: Cycle detected without exit condition.

Can I use callable conditions for edge routing in GraphFlow?

Yes, DiGraphEdge supports both string conditions and callable conditions. For string conditions, the runtime checks if the string appears as a substring in the message content. For callable conditions, you provide a function (or lambda) that receives the message object and returns a boolean. Note that callable conditions are evaluated at runtime by GraphFlowManager and are not serialized with the graph structure, whereas string conditions are stored directly in the DiGraph model.

What are activation groups and when should I use them?

Activation groups control how a node with multiple incoming edges determines when it is ready to execute. By assigning the same activation_group name to multiple edges pointing to the same target node, you can specify whether all parents must complete (default, activation_condition="all") or any single parent must complete (activation_condition="any"). Use activation groups for join patterns where you need to synchronize multiple parallel branches (requiring all) or for race conditions where the first completing branch should trigger the next step (requiring any).

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}