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

> Build powerful graph-based agent workflows using AutoGen's DiGraphBuilder and GraphFlow. Implement complex patterns like fan-outs, branching, and loops with ease.

- Repository: [Microsoft/autogen](https://github.com/microsoft/autogen)
- Tags: tutorial
- Published: 2026-03-07

---

**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 `ChatAgent`s 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`](https://github.com/microsoft/autogen/blob/main/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`](https://github.com/microsoft/autogen/blob/main/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`](https://github.com/microsoft/autogen/blob/main/_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:

```python
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:

```python
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):

```python
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:

```python
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**:

```python
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`](https://github.com/microsoft/autogen/blob/main/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`](https://github.com/microsoft/autogen/blob/main/_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`](https://github.com/microsoft/autogen/blob/main/_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`](https://github.com/microsoft/autogen/blob/main/_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).