Implementation:Microsoft Autogen DiGraphBuilder

Knowledge Sources	Microsoft AutoGen
Domains	Multi-Agent Systems, Builder Pattern, Fluent API, Graph Construction, Software Design Patterns
Last Updated	2026-02-11 00:00 GMT

Overview

Concrete tool for programmatically constructing validated directed graph topologies using a fluent builder API, provided by Microsoft AutoGen.

Description

DiGraphBuilder is a fluent builder class for constructing DiGraph execution graphs used by GraphFlow. It provides a chainable method API for adding agent nodes, connecting them with directed edges (optionally conditioned on message content), setting entry points, and producing a validated DiGraph on build().

The builder maintains an internal registry of both graph nodes and their corresponding agent objects. This dual tracking means that after constructing the graph, the list of participants can be retrieved directly via get_participants() without maintaining a separate list.

Edges can reference agents by either their name string or by the agent object directly. Conditions can be string-based (substring matching) or callable-based (arbitrary predicates). The builder supports activation groups and activation conditions for advanced fan-in patterns in cyclic graphs.

Usage

Use DiGraphBuilder whenever you need to construct a DiGraph for GraphFlow. It is the recommended way to build graphs, as it provides:

Fluent method chaining for concise graph definitions.
Automatic name resolution from agent objects.
Early validation of node existence when adding edges.
Deferred structural validation on build().
Participant list extraction via get_participants().

Code Reference

Source Location

Repository: Microsoft AutoGen
File: python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_graph/_graph_builder.py (Lines 104-209)

Signature

class DiGraphBuilder:
    def __init__(self) -> None: ...

    def add_node(
        self,
        agent: ChatAgent,
        activation: Literal["all", "any"] = "all",
    ) -> "DiGraphBuilder": ...

    def add_edge(
        self,
        source: Union[str, ChatAgent],
        target: Union[str, ChatAgent],
        condition: Optional[Union[str, Callable[[BaseChatMessage], bool]]] = None,
        activation_group: Optional[str] = None,
        activation_condition: Optional[Literal["all", "any"]] = None,
    ) -> "DiGraphBuilder": ...

    def add_conditional_edges(
        self,
        source: Union[str, ChatAgent],
        condition_to_target: Dict[str, Union[str, ChatAgent]],
    ) -> "DiGraphBuilder": ...

    def set_entry_point(
        self,
        name: Union[str, ChatAgent],
    ) -> "DiGraphBuilder": ...

    def build(self) -> DiGraph: ...

    def get_participants(self) -> list[ChatAgent]: ...

Import

from autogen_agentchat.teams import DiGraphBuilder

I/O Contract

Inputs (add_node)

Name	Type	Required	Description
agent	ChatAgent	Yes	The agent to register as a graph node. The agent's name is used as the node identifier.
activation	Literal["all", "any"]	No	Activation mode for the node when it has multiple incoming edges. "all" waits for all parents; "any" fires on the first. Defaults to "all".

Inputs (add_edge)

Name	Type	Required	Description
source	str or ChatAgent	Yes	The source node. Can be an agent name string or agent object.
target	str or ChatAgent	Yes	The target node. Can be an agent name string or agent object.
condition	str, Callable, or None	No	Edge condition. If string, the edge activates when the string is a substring of the last message. If callable, activates when the function returns True for the message. If None, unconditional.
activation_group	str or None	No	Group identifier for forward dependencies. Defaults to the target node name. Used to distinguish edges from different paths to the same target.
activation_condition	Literal["all", "any"] or None	No	How edges in this activation group are evaluated. Defaults to "all".

Inputs (set_entry_point)

Name	Type	Required	Description
name	str or ChatAgent	Yes	The node to designate as the default start node.

Outputs

Name	Type	Description
build()	DiGraph	A validated DiGraph ready for use with GraphFlow. Raises ValueError if validation fails.
get_participants()	list[ChatAgent]	The list of agents registered in the builder, in insertion order.

Usage Examples

Basic Example: Sequential Chain

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import DiGraphBuilder
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")

agent_a = AssistantAgent("A", model_client=model_client, system_message="Summarize input.")
agent_b = AssistantAgent("B", model_client=model_client, system_message="Translate to French.")
agent_c = AssistantAgent("C", model_client=model_client, system_message="Translate to German.")

# Build a sequential graph: A -> B -> C
builder = DiGraphBuilder()
builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
builder.add_edge(agent_a, agent_b).add_edge(agent_b, agent_c)
graph = builder.build()

participants = builder.get_participants()  # [agent_a, agent_b, agent_c]

Fan-Out with Conditional Branching

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import DiGraphBuilder
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")

classifier = AssistantAgent("classifier", model_client=model_client,
                             system_message="Classify input as 'positive' or 'negative'.")
pos_handler = AssistantAgent("pos_handler", model_client=model_client,
                              system_message="Handle positive sentiment.")
neg_handler = AssistantAgent("neg_handler", model_client=model_client,
                              system_message="Handle negative sentiment.")

builder = DiGraphBuilder()
builder.add_node(classifier).add_node(pos_handler).add_node(neg_handler)

# Conditional branching using callable conditions
builder.add_edge(classifier, pos_handler,
                 condition=lambda msg: "positive" in msg.to_model_text())
builder.add_edge(classifier, neg_handler,
                 condition=lambda msg: "negative" in msg.to_model_text())

graph = builder.build()

Cyclic Loop with Activation Groups

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import DiGraphBuilder
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4.1")

writer = AssistantAgent("writer", model_client=model_client,
                         system_message="Write content based on feedback.")
reviewer = AssistantAgent("reviewer", model_client=model_client,
                           system_message="Review and say APPROVE or provide feedback.")
publisher = AssistantAgent("publisher", model_client=model_client,
                            system_message="Publish the approved content.")

builder = DiGraphBuilder()
builder.add_node(writer).add_node(reviewer).add_node(publisher)

# Forward path: writer -> reviewer
builder.add_edge(writer, reviewer)

# Conditional exit: reviewer -> publisher (if approved)
builder.add_edge(reviewer, publisher,
                 condition=lambda msg: "APPROVE" in msg.to_model_text())

# Loop back: reviewer -> writer (if not approved), with distinct activation group
builder.add_edge(reviewer, writer,
                 condition=lambda msg: "APPROVE" not in msg.to_model_text(),
                 activation_group="feedback_loop")

builder.set_entry_point(writer)
graph = builder.build()

Related Pages

Implements Principle

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment