Record a model invocation.
(self, node_id: str, model_name: str,
input_data: str = None, output: str = None,
duration: float = None, details: Dict[str, Any] = None,
stage: CallStage | str | None = None)
| 223 | ) |
| 224 | |
| 225 | def record_model_call(self, node_id: str, model_name: str, |
| 226 | input_data: str = None, output: str = None, |
| 227 | duration: float = None, details: Dict[str, Any] = None, |
| 228 | stage: CallStage | str | None = None) -> None: |
| 229 | """Record a model invocation.""" |
| 230 | stage_value = stage.value if isinstance(stage, CallStage) else stage |
| 231 | call_details = { |
| 232 | "model_name": model_name, |
| 233 | "input_data": input_data, |
| 234 | "output": output, |
| 235 | **(details or {}) |
| 236 | } |
| 237 | if stage_value: |
| 238 | call_details["stage"] = stage_value |
| 239 | |
| 240 | self.info( |
| 241 | f"Model call for node {node_id}", |
| 242 | node_id=node_id, |
| 243 | event_type=EventType.MODEL_CALL, |
| 244 | details=call_details, |
| 245 | duration=duration |
| 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, |
no test coverage detected