Class DynamicEdgeExecutor

workflow/executor/dynamic_edge_executor.py:18–399

Execute edge-level dynamic expansion. When an edge has dynamic configuration, this executor: 1. Splits the payload passing through the edge 2. Executes the target node for each split unit 3. Collects and returns results (flat for Map, reduced for Tree)

class DynamicEdgeExecutor:
    """Execute edge-level dynamic expansion.
    
    When an edge has dynamic configuration, this executor:
    1. Splits the payload passing through the edge
    2. Executes the target node for each split unit
    3. Collects and returns results (flat for Map, reduced for Tree)
    """
    
    def __init__(
        self,
        log_manager: LogManager,
        node_executor_func: Callable[[Node, List[Message]], List[Message]],
    ):
        """Initialize the dynamic edge executor.
        
        Args:
            log_manager: Logger instance
            node_executor_func: Function to execute a node with inputs
        """
        self.log_manager = log_manager
        self.node_executor_func = node_executor_func
    
    def execute(
        self,
        target_node: Node,
        payload: Message,
        dynamic_config: DynamicEdgeConfig,
        static_inputs: Optional[List[Message]] = None,
    ) -> List[Message]:
        """Execute dynamic expansion for an edge.
        
        Args:
            target_node: The node to execute (will be used as template)
            payload: The message passing through the edge
            dynamic_config: Edge dynamic configuration
            static_inputs: Optional static inputs from non-dynamic edges
            
        Returns:
            List of output messages from all executions
        """
        split_config = dynamic_config.split
        
        # Create splitter based on config
        splitter = create_splitter_from_config(split_config)
        
        # Split the payload into execution units
        execution_units = splitter.split([payload])
        
        if not execution_units:
            self.log_manager.debug(
                f"Dynamic edge -> {target_node.id}: no execution units after split"
            )
            return []
        
        self.log_manager.info(
            f"Dynamic edge -> {target_node.id}: splitting into {len(execution_units)} parallel units"
        )