(self)
| 82 | self.payload_to_text = payload_to_text_func |
| 83 | |
| 84 | def run(self) -> str: |
| 85 | self.log_manager.info("Executing graph with majority voting approach") |
| 86 | all_nodes = list(self.nodes.values()) |
| 87 | if not all_nodes: |
| 88 | self.log_manager.error("No nodes to execute in majority voting mode") |
| 89 | return "" |
| 90 | |
| 91 | for node in all_nodes: |
| 92 | node.clear_input() |
| 93 | for message in self.initial_messages: |
| 94 | node.append_input(message.clone()) |
| 95 | |
| 96 | node_ids = [node.id for node in all_nodes] |
| 97 | |
| 98 | def _execute(node_id: str) -> None: |
| 99 | self.execute_node_func(self.nodes[node_id]) |
| 100 | |
| 101 | parallel_executor = ParallelExecutor(self.log_manager, self.nodes) |
| 102 | parallel_executor.execute_nodes_parallel(node_ids, _execute) |
| 103 | |
| 104 | return self._collect_majority_result() |
| 105 | |
| 106 | def _collect_majority_result(self) -> str: |
| 107 | node_outputs: List[Dict[str, str]] = [] |
nothing calls this directly
no test coverage detected