Build a graph of the agent and its sub-agents. Args: graph: The graph to build on. agent: The agent to build the graph for. highlight_pairs: A list of pairs of nodes to highlight. parent_agent: The parent agent of the current agent. This is specifically used when building Workfl
(
graph: graphviz.Digraph,
agent: BaseAgent,
highlight_pairs,
parent_agent=None,
)
| 39 | |
| 40 | |
| 41 | async def build_graph( |
| 42 | graph: graphviz.Digraph, |
| 43 | agent: BaseAgent, |
| 44 | highlight_pairs, |
| 45 | parent_agent=None, |
| 46 | ): |
| 47 | """ |
| 48 | Build a graph of the agent and its sub-agents. |
| 49 | Args: |
| 50 | graph: The graph to build on. |
| 51 | agent: The agent to build the graph for. |
| 52 | highlight_pairs: A list of pairs of nodes to highlight. |
| 53 | parent_agent: The parent agent of the current agent. This is specifically used when building Workflow Agents to directly connect a node to nodes inside a Workflow Agent. |
| 54 | |
| 55 | Returns: |
| 56 | None |
| 57 | """ |
| 58 | from ..workflow._base_node import START |
| 59 | from ..workflow._workflow import Workflow |
| 60 | |
| 61 | dark_green = '#0F5223' |
| 62 | light_green = '#69CB87' |
| 63 | light_gray = '#cccccc' |
| 64 | white = '#ffffff' |
| 65 | |
| 66 | def get_node_name(tool_or_agent: Union[BaseAgent, BaseTool]): |
| 67 | if isinstance(tool_or_agent, BaseAgent): |
| 68 | # Added Workflow Agent checks for different agent types |
| 69 | if isinstance(tool_or_agent, SequentialAgent): |
| 70 | return tool_or_agent.name + ' (Sequential Agent)' |
| 71 | elif isinstance(tool_or_agent, LoopAgent): |
| 72 | return tool_or_agent.name + ' (Loop Agent)' |
| 73 | elif isinstance(tool_or_agent, ParallelAgent): |
| 74 | return tool_or_agent.name + ' (Parallel Agent)' |
| 75 | else: |
| 76 | return tool_or_agent.name |
| 77 | elif isinstance(tool_or_agent, BaseTool): |
| 78 | return tool_or_agent.name |
| 79 | elif hasattr(tool_or_agent, 'name'): |
| 80 | return tool_or_agent.name |
| 81 | else: |
| 82 | raise ValueError(f'Unsupported tool type: {tool_or_agent}') |
| 83 | |
| 84 | def get_node_caption(tool_or_agent: Union[BaseAgent, BaseTool]): |
| 85 | |
| 86 | if isinstance(tool_or_agent, BaseAgent): |
| 87 | return '🤖 ' + tool_or_agent.name |
| 88 | elif retrieval_tool_module_loaded and isinstance( |
| 89 | tool_or_agent, BaseRetrievalTool |
| 90 | ): |
| 91 | return '🔎 ' + tool_or_agent.name |
| 92 | elif isinstance(tool_or_agent, FunctionTool): |
| 93 | return '🔧 ' + tool_or_agent.name |
| 94 | elif isinstance(tool_or_agent, AgentTool): |
| 95 | return '🤖 ' + tool_or_agent.name |
| 96 | elif isinstance(tool_or_agent, BaseTool): |
| 97 | return '🔧 ' + tool_or_agent.name |
| 98 | elif hasattr(tool_or_agent, 'name'): |
no test coverage detected