from typing import Dict, Literal, Optional, Union
from autogen_agentchat.base import ChatAgent
from ._digraph_group_chat import DiGraph, DiGraphEdge, DiGraphNode
[文档]
class DiGraphBuilder:
"""
用于构建 :class:`GraphFlow` 中使用的 :class:`DiGraph` 执行图的流式构建器。
.. warning::
这是实验性功能，API 将在未来版本中变更。
该工具提供了一种便捷的编程方式来构建智能体交互图，支持包括以下复杂执行流程：
- 顺序链式执行
- 并行扇出
- 条件分支
- 带安全出口的循环
图中每个节点代表一个智能体。边定义了智能体间的执行路径，并可选择性地基于消息内容进行条件控制。
该构建器兼容 `Graph` 运行器，同时支持标准智能体和过滤型智能体。
方法说明：
- add_node(agent, activation): 向图中添加智能体节点
- add_edge(source, target, condition): 连接两个节点，可选添加条件
- add_conditional_edges(source, condition_to_target): 从源节点添加多个条件边
- set_entry_point(agent): 设置默认起始节点（可选）
- build(): 生成验证过的 `DiGraph`
- get_participants(): 返回已添加的智能体列表
示例 — 顺序流 A → B → C:
>>> builder = GraphBuilder()
>>> 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)
>>> team = Graph(
... participants=builder.get_participants(),
... graph=builder.build(),
... termination_condition=MaxMessageTermination(5),
... )
示例 — 并行扇出 A → (B, C):
>>> builder = GraphBuilder()
>>> builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
>>> builder.add_edge(agent_a, agent_b).add_edge(agent_a, agent_c)
示例 — 条件分支 A → B ("yes"), A → C ("no"):
>>> builder = GraphBuilder()
>>> builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
>>> builder.add_conditional_edges(agent_a, {"yes": agent_b, "no": agent_c})
示例 — 循环: A → B → A ("loop"), B → C ("exit"):
>>> builder = GraphBuilder()
>>> builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
>>> builder.add_edge(agent_a, agent_b)
>>> builder.add_conditional_edges(agent_b, {"loop": agent_a, "exit": agent_c})
"""
def __init__(self) -> None:
self.nodes: Dict[str, DiGraphNode] = {}
self.agents: Dict[str, ChatAgent] = {}
self._default_start_node: Optional[str] = None
def _get_name(self, obj: Union[str, ChatAgent]) -> str:
return obj if isinstance(obj, str) else obj.name
[文档]
def add_node(self, agent: ChatAgent, activation: Literal["all", "any"] = "all") -> "DiGraphBuilder":
"""向图中添加节点并注册其智能体。"""
name = agent.name
if name not in self.nodes:
self.nodes[name] = DiGraphNode(name=name, edges=[], activation=activation)
self.agents[name] = agent
return self
[文档]
def add_edge(
self, source: Union[str, ChatAgent], target: Union[str, ChatAgent], condition: Optional[str] = None
) -> "DiGraphBuilder":
"""添加从源节点到目标节点的有向边，可选择性地添加条件。"""
source_name = self._get_name(source)
target_name = self._get_name(target)
if source_name not in self.nodes:
raise ValueError(f"Source node '{source_name}' must be added before adding an edge.")
if target_name not in self.nodes:
raise ValueError(f"Target node '{target_name}' must be added before adding an edge.")
self.nodes[source_name].edges.append(DiGraphEdge(target=target_name, condition=condition))
return self
[文档]
def add_conditional_edges(
self, source: Union[str, ChatAgent], condition_to_target: Dict[str, Union[str, ChatAgent]]
) -> "DiGraphBuilder":
"""根据条件字符串从源节点添加多条条件边。"""
for condition, target in condition_to_target.items():
self.add_edge(source, target, condition)
return self
[文档]
def set_entry_point(self, name: Union[str, ChatAgent]) -> "DiGraphBuilder":
"""设置图的默认起始节点。"""
node_name = self._get_name(name)
if node_name not in self.nodes:
raise ValueError(f"Start node '{node_name}' must be added before setting as entry point.")
self._default_start_node = node_name
return self
[文档]
def build(self) -> DiGraph:
"""构建并验证有向图。"""
graph = DiGraph(
nodes=self.nodes,
default_start_node=self._default_start_node,
)
graph.graph_validate()
return graph
[文档]
def get_participants(self) -> list[ChatAgent]:
"""返回构建器中按插入顺序排列的代理列表。"""
return list(self.agents.values())
