Record a human interaction.
(self, node_id: str, input_data: Any = None, output: Any = None,
details: Dict[str, Any] = None)
| 101 | self.logger.record_edge_process(from_node, to_node, details) |
| 102 | |
| 103 | def record_human_interaction(self, node_id: str, input_data: Any = None, output: Any = None, |
| 104 | details: Dict[str, Any] = None) -> None: |
| 105 | """Record a human interaction.""" |
| 106 | input_size = len(str(input_data)) if input_data is not None else 0 |
| 107 | output_size = len(str(output)) if output is not None else 0 |
| 108 | duration = self.logger.get_timer(f"human_{node_id}") |
| 109 | call_details = { |
| 110 | "input_size": input_size, |
| 111 | "output_size": output_size, |
| 112 | **(details or {}) |
| 113 | } |
| 114 | self.logger.record_human_interaction( |
| 115 | node_id, input_data, output, duration, call_details |
| 116 | ) |
| 117 | |
| 118 | def record_model_call(self, node_id: str, model_name: str, |
| 119 | input_data: Any = None, output: Any = None, |