Record a tool invocation.
(self, node_id: str, tool_name: str, tool_result: str,
duration: float = None, success: bool | None = True,
details: Dict[str, Any] = None,
stage: CallStage | str | None = None)
| 246 | ) |
| 247 | |
| 248 | def record_tool_call(self, node_id: str, tool_name: str, tool_result: str, |
| 249 | duration: float = None, success: bool | None = True, |
| 250 | details: Dict[str, Any] = None, |
| 251 | stage: CallStage | str | None = None) -> None: |
| 252 | """Record a tool invocation.""" |
| 253 | stage_value = stage.value if isinstance(stage, CallStage) else stage |
| 254 | tool_details = { |
| 255 | "tool_result": tool_result, |
| 256 | "tool_name": tool_name, |
| 257 | "success": success, |
| 258 | **(details or {}) |
| 259 | } |
| 260 | if stage_value: |
| 261 | tool_details["stage"] = stage_value |
| 262 | |
| 263 | level = LogLevel.INFO if success is not False else LogLevel.ERROR |
| 264 | self.add_log( |
| 265 | level, |
| 266 | f"Tool call {tool_name} for node {node_id}", |
| 267 | node_id=node_id, |
| 268 | event_type=EventType.TOOL_CALL, |
| 269 | details=tool_details, |
| 270 | duration=duration |
| 271 | ) |
| 272 | |
| 273 | def record_thinking_process(self, node_id: str, thinking_mode: str, thinking_result: str, stage: str, |
| 274 | duration: float = None, details: Dict[str, Any] = None) -> None: |
no test coverage detected