Record data when entering a node.
(self, node_id: str, inputs: List[Dict[str, str]], node_type: str = None,
details: Dict[str, Any] = None)
| 156 | self.add_log(LogLevel.CRITICAL, message, node_id, event_type, details) |
| 157 | |
| 158 | def enter_node(self, node_id: str, inputs: List[Dict[str, str]], node_type: str = None, |
| 159 | details: Dict[str, Any] = None) -> None: |
| 160 | """Record data when entering a node.""" |
| 161 | self.current_path.append(node_id) |
| 162 | self.info( |
| 163 | f"Entering node {node_id}", |
| 164 | node_id=node_id, |
| 165 | event_type=EventType.NODE_START, |
| 166 | details={ |
| 167 | "inputs": inputs, |
| 168 | # "combined_input": combined_input, |
| 169 | "node_type": node_type, |
| 170 | **(details or {}) |
| 171 | } |
| 172 | ) |
| 173 | |
| 174 | def exit_node(self, node_id: str, output: str, duration: float = None, |
| 175 | output_size: int = None, details: Dict[str, Any] = None) -> None: |