(
app_name: str, user_id: str, session_id: str, event_id: str
)
| 1143 | tags=[TAG_DEBUG], |
| 1144 | ) |
| 1145 | async def get_event_graph( |
| 1146 | app_name: str, user_id: str, session_id: str, event_id: str |
| 1147 | ): |
| 1148 | session = await self.session_service.get_session( |
| 1149 | app_name=app_name, user_id=user_id, session_id=session_id |
| 1150 | ) |
| 1151 | session_events = session.events if session else [] |
| 1152 | event = next((x for x in session_events if x.id == event_id), None) |
| 1153 | if not event: |
| 1154 | return {} |
| 1155 | |
| 1156 | function_calls = event.get_function_calls() |
| 1157 | function_responses = event.get_function_responses() |
| 1158 | agent_or_app = self.agent_loader.load_agent(app_name) |
| 1159 | root_agent = self._get_root_agent(agent_or_app) |
| 1160 | dot_graph = None |
| 1161 | if function_calls: |
| 1162 | function_call_highlights = [] |
| 1163 | for function_call in function_calls: |
| 1164 | from_name = event.author |
| 1165 | to_name = function_call.name |
| 1166 | function_call_highlights.append((from_name, to_name)) |
| 1167 | dot_graph = await agent_graph.get_agent_graph( |
| 1168 | root_agent, function_call_highlights |
| 1169 | ) |
| 1170 | elif function_responses: |
| 1171 | function_responses_highlights = [] |
| 1172 | for function_response in function_responses: |
| 1173 | from_name = function_response.name |
| 1174 | to_name = event.author |
| 1175 | function_responses_highlights.append((from_name, to_name)) |
| 1176 | dot_graph = await agent_graph.get_agent_graph( |
| 1177 | root_agent, function_responses_highlights |
| 1178 | ) |
| 1179 | else: |
| 1180 | from_name = event.author |
| 1181 | to_name = "" |
| 1182 | dot_graph = await agent_graph.get_agent_graph( |
| 1183 | root_agent, [(from_name, to_name)] |
| 1184 | ) |
| 1185 | if dot_graph and isinstance(dot_graph, graphviz.Digraph): |
| 1186 | return GetEventGraphResult(dot_src=dot_graph.source) |
| 1187 | else: |
| 1188 | return {} |
| 1189 | |
| 1190 | def _navigate_to_node(self, app_info: dict, node_path: str) -> dict | None: |
| 1191 | """Navigate to a specific node in the agent hierarchy. |
nothing calls this directly
no test coverage detected