Execute DAG workflows. Features: - Execute layer by layer following the topology - Support parallel execution inside a layer - Serialize Human nodes automatically
| 8 | |
| 9 | |
| 10 | class DAGExecutor: |
| 11 | """Execute DAG workflows. |
| 12 | |
| 13 | Features: |
| 14 | - Execute layer by layer following the topology |
| 15 | - Support parallel execution inside a layer |
| 16 | - Serialize Human nodes automatically |
| 17 | """ |
| 18 | |
| 19 | def __init__( |
| 20 | self, |
| 21 | log_manager: LogManager, |
| 22 | nodes: Dict[str, Node], |
| 23 | layers: List[List[str]], |
| 24 | execute_node_func: Callable[[Node], None] |
| 25 | ): |
| 26 | """Initialize the executor. |
| 27 | |
| 28 | Args: |
| 29 | log_manager: Logger instance |
| 30 | nodes: Mapping of node ids to ``Node`` objects |
| 31 | layers: Topological layers |
| 32 | execute_node_func: Callable used to execute a single node |
| 33 | """ |
| 34 | self.log_manager = log_manager |
| 35 | self.nodes = nodes |
| 36 | self.layers = layers |
| 37 | self.execute_node_func = execute_node_func |
| 38 | self.parallel_executor = ParallelExecutor(log_manager, nodes) |
| 39 | |
| 40 | def execute(self) -> None: |
| 41 | """Execute the DAG workflow.""" |
| 42 | for layer_idx, layer_nodes in enumerate(self.layers): |
| 43 | self.log_manager.debug(f"Executing Layer {layer_idx} with nodes: {layer_nodes}") |
| 44 | self._execute_layer(layer_nodes) |
| 45 | |
| 46 | def _execute_layer(self, layer_nodes: List[str]) -> None: |
| 47 | """Execute a single topological layer.""" |
| 48 | def execute_if_triggered(node_id: str) -> None: |
| 49 | node = self.nodes[node_id] |
| 50 | if node.is_triggered(): |
| 51 | self.execute_node_func(node) |
| 52 | else: |
| 53 | self.log_manager.debug(f"Node {node_id} skipped - not triggered") |
| 54 | |
| 55 | self.parallel_executor.execute_nodes_parallel(layer_nodes, execute_if_triggered) |