Overview
Concrete tool for defining and validating directed graph topologies for agent orchestration provided by Microsoft AutoGen.
Description
DiGraph is a Pydantic model that represents a directed graph structure used by GraphFlow to determine agent execution order and conditions. It stores a mapping of node names to DiGraphNode objects, each containing outgoing DiGraphEdge entries with optional conditions. The model provides comprehensive validation including reachability checks, cycle detection with exit condition enforcement, and activation condition consistency validation.
Each DiGraphNode has a name, a list of outgoing DiGraphEdge objects, and an activation mode ("all" or "any") that controls how the node is triggered when it has multiple incoming edges.
Each DiGraphEdge has a target node name, an optional condition (string for substring matching or callable for arbitrary predicates), an activation group identifier, and an activation condition that determines how edges within the same group are evaluated.
Usage
Use DiGraph when you need to define the execution topology for a GraphFlow team. It is typically constructed either directly or via the DiGraphBuilder fluent API. After construction, call graph_validate() to check structural integrity before passing it to GraphFlow.
Code Reference
Source Location
- Repository: Microsoft AutoGen
- File:
python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_graph/_digraph_group_chat.py (Lines 115-300)
Signature
class DiGraph(BaseModel):
nodes: Dict[str, DiGraphNode] # Node name -> DiGraphNode mapping
default_start_node: str | None = None # Default start node name
class DiGraphNode(BaseModel):
name: str # Agent's name
edges: List[DiGraphEdge] = [] # Outgoing edges
activation: Literal["all", "any"] = "all"
class DiGraphEdge(BaseModel):
target: str # Target node name
condition: Union[str, Callable[[BaseChatMessage], bool], None] = None
activation_group: str = "" # Defaults to target node name
activation_condition: Literal["all", "any"] = "all"
Import
from autogen_agentchat.teams import DiGraph, DiGraphNode, DiGraphEdge
I/O Contract
Inputs
| Name |
Type |
Required |
Description
|
| nodes |
Dict[str, DiGraphNode] |
Yes |
Mapping from agent names to their corresponding DiGraphNode definitions, each containing outgoing edges and activation settings.
|
| default_start_node |
str or None |
No |
Name of the designated entry-point node. If None, all nodes with no incoming edges are treated as start nodes.
|
DiGraphNode Fields
| Name |
Type |
Required |
Description
|
| name |
str |
Yes |
The agent name this node represents.
|
| edges |
List[DiGraphEdge] |
No |
Outgoing edges from this node. Defaults to empty list (leaf node).
|
| activation |
Literal["all", "any"] |
No |
How this node is activated when it has multiple incoming edges. "all" waits for all parents; "any" fires on the first. Defaults to "all".
|
DiGraphEdge Fields
| Name |
Type |
Required |
Description
|
| target |
str |
Yes |
Name of the target node this edge points to.
|
| condition |
str, Callable, or None |
No |
If None, the edge is unconditional. If a string, activates when the string is a substring of the last message. If callable, activates when the function returns True.
|
| activation_group |
str |
No |
Group identifier for forward dependencies. Defaults to the target node name. Used to distinguish different dependency patterns (e.g., initial entry vs. loop-back).
|
| activation_condition |
Literal["all", "any"] |
No |
How edges within the same activation group are evaluated. "all" requires all edges satisfied; "any" requires just one. Defaults to "all".
|
Outputs
| Name |
Type |
Description
|
| DiGraph |
DiGraph (Pydantic model) |
A validated directed graph structure ready for use with GraphFlow.
|
Key Methods
| Method |
Return Type |
Description
|
| graph_validate() |
None |
Validates the graph structure: checks for non-empty nodes, start/leaf node existence, edge consistency (no mixed conditional/unconditional), activation condition consistency, and cycle safety.
|
| get_start_nodes() |
Set[str] |
Returns nodes with no incoming edges, or the single default_start_node if set.
|
| get_leaf_nodes() |
Set[str] |
Returns nodes with no outgoing edges (terminal nodes).
|
| get_parents() |
Dict[str, List[str]] |
Computes a mapping of each node to its parent nodes.
|
| get_has_cycles() |
bool |
Returns True if the graph contains at least one cycle (with valid exit conditions).
|
| has_cycles_with_exit() |
bool |
Checks for cycles and validates that every cycle has at least one conditional edge. Raises ValueError if a cycle lacks an exit condition.
|
Usage Examples
Basic Example: Sequential Graph
from autogen_agentchat.teams import DiGraph, DiGraphNode, DiGraphEdge
# Define a simple sequential graph: A -> B -> C
graph = DiGraph(
nodes={
"A": DiGraphNode(
name="A",
edges=[DiGraphEdge(target="B")]
),
"B": DiGraphNode(
name="B",
edges=[DiGraphEdge(target="C")]
),
"C": DiGraphNode(
name="C",
edges=[]
),
}
)
# Validate the graph structure
graph.graph_validate()
# Inspect start and leaf nodes
print(graph.get_start_nodes()) # {"A"}
print(graph.get_leaf_nodes()) # {"C"}
Conditional Branching Graph
from autogen_agentchat.teams import DiGraph, DiGraphNode, DiGraphEdge
# A -> B (if "yes" in message) or C (if "no" in message)
graph = DiGraph(
nodes={
"A": DiGraphNode(
name="A",
edges=[
DiGraphEdge(target="B", condition="yes"),
DiGraphEdge(target="C", condition="no"),
]
),
"B": DiGraphNode(name="B", edges=[]),
"C": DiGraphNode(name="C", edges=[]),
}
)
graph.graph_validate()
print(graph.get_has_cycles()) # False
Cyclic Graph with Exit Condition
from autogen_agentchat.teams import DiGraph, DiGraphNode, DiGraphEdge
# A -> B -> C (if "APPROVE") or back to A (if not "APPROVE")
graph = DiGraph(
nodes={
"A": DiGraphNode(
name="A",
edges=[DiGraphEdge(target="B")]
),
"B": DiGraphNode(
name="B",
edges=[
DiGraphEdge(target="C", condition="APPROVE"),
DiGraphEdge(target="A", condition="REVISE"),
]
),
"C": DiGraphNode(name="C", edges=[]),
},
default_start_node="A",
)
graph.graph_validate()
print(graph.get_has_cycles()) # True
Related Pages
Implements Principle
