i;dZddlZddlmZmZmZmZddlm Z ddl m Z ddl m Z mZddlmZmZddlmZGd d Zy) aDynamic edge executor for edge-level Map and Tree execution. Handles dynamic node expansion based on edge-level dynamic configuration. When a message passes through an edge with dynamic config, the target node is virtually expanded into multiple instances based on split results. N)CallableDictListOptional)Node)DynamicEdgeConfig)Message MessageRole)create_splitter_from_configgroup_messages) LogManagerc leZdZdZdedeeeegeeffdZ ddedede d e eed eef d Z dded eede d e eed eef d Z ddedeeede d e eed eef dZ ddedeeede d e eed eef dZdedeeded eefdZdedeededed eef dZy)DynamicEdgeExecutoraExecute 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) log_managernode_executor_funcc ||_||_y)zInitialize the dynamic edge executor. Args: log_manager: Logger instance node_executor_func: Function to execute a node with inputs N)rr)selfrrs T/Users/bowang/.openclaw/workspace/ChatDev/workflow/executor/dynamic_edge_executor.py__init__zDynamicEdgeExecutor.__init__s'"4N target_nodepayloaddynamic_config static_inputsreturnc|j}t|}|j|g}|s+|jjd|jdgS|jj d|jdt |d|jr|j||||S|jr|j||||Std|j)aExecute 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 Dynamic edge -> : no execution units after splitz: splitting into z parallel unitsUnknown dynamic type: ) splitr rdebugidinfolenis_map _execute_mapis_tree _execute_tree ValueErrortype)rrrrr split_configsplitterexecution_unitss rexecutezDynamicEdgeExecutor.execute)s$&++ /|<#..'3    " "";>>"22RS I {~~..?O@T?UUd e   "$$_nm  # # %%%_nm 5n6I6I5JKL Lrinputsc d|j}|xsg}t|}|j|}|s?|jjd|jd|r|j ||SgS|jj d|jdt|dt|d|jd |rdt|dnd z|jr|j||||S|jr|j||||Std |j) aCExecute dynamic expansion using all collected inputs. This method is called from _execute_node when a node has incoming edges with dynamic configuration. All inputs are already collected and passed here. Args: target_node: The node to execute inputs: Dynamic edge inputs to be split dynamic_config: Edge dynamic configuration static_inputs: Non-dynamic edge inputs to be replicated to all units Returns: List of output messages from all executions z Dynamic node rz : splitting z dynamic inputs into z parallel units (z mode)z, with z! static inputs replicated to eachr)r r rr!r"rr#r$r*r%r&r'r(r))rrr/rrr+r,r-s rexecute_from_inputsz'DynamicEdgeExecutor.execute_from_inputsXsW*&++ %+ /|<#..0    " " //OP ..{MJJI KNN+<F }DY?#$$5n6I6I5J& RR_]+,,MNeg i  "$$_nm  # # %%%_nm 5n6I6I5JKL Lrr-c :|j}|j}g}|xsg}t|dk(r7t||dz}|j ||d} |j | n0t t||} tjj| 5} i} t|D]6\} }t||z}| j|j||| }| | |<8i}tjj| D]U}| |} |j}||| <|jjd|j d| dt|dWt't|D]} | |vs|j ||  d d d |jj)d|j d t|d |S#t"$r4}|jj%d|j d| d|d }~wwxYw#1swYxYw) a]Execute in Map mode (fan-out only). Args: target_node: Target node template execution_units: Split message units dynamic_config: Dynamic configuration static_inputs: Static inputs to copy to all units Returns: Flat list of all output messages r max_workersr#z: completed with  outputs: failed with error: Nz: Map completed with z total outputs) as_map_config max_parallelr$list _execute_unitextendmin concurrentfuturesThreadPoolExecutor enumeratesubmit as_completedresultrr!r" Exceptionerrorranger#)rrr-rr map_configr; all_outputs unit_inputsoutputseffective_workersexecutorrAidxunitfutureresults_by_idxrFes rr&z DynamicEdgeExecutor._execute_mapsA$$113 !.. %' %+  1 $}-0BBK((k1EG   w '!$C$8, G ##66CT6UYa@B!*?!;IC"&}"5"(00==gFF!&/C !'.4s+((...{~~.>auE..1&k](D G"!_!56Cn,#**>#+>?79V@ {~~./""%k"2!3> C #%((...{~~.>auE2236 )VUs8A0HA GH;H H/H  HHHc |j}|td|j|j}|j}|xsg}g}|D]} |j | |sgS|j jd|jdt|d|d} d} t|dkDr | dz } t||} |j jd|jd | d t| d g} t| dk(r;| d}| rt||z}|j||| d}| j |ntt| |}tjj!| 5}i}t#| D];\}}|}| rt||z}|j%|j||| |}|||<=i}tjj'|D]}||} |j)}|||<t/t| D]}||vs| j || ddd|j jd|jd | dt| d| }d} | dkDr*|j j-d|jdnt|dkDr |j jd|jd| dt|d|S#t*$r7}|j j-d|jd | d|d|d}~wwxYw#1swYxYw)ayExecute in Tree mode (fan-out + reduce). Args: target_node: Target node template execution_units: Split message units dynamic_config: Dynamic configuration static_inputs: Static inputs (used in first layer only) Returns: Single-element list with the final reduced result Nz'Invalid tree configuration for edge -> rz: Tree starting with z inputs, group_size=rTr4z layer z : processing z groupsr5r7-r9z : produced r8Fdz: exceeded maximum layersz: Tree completed after z layers with z output(s))as_tree_configr)r" group_sizer;r>rr#r$r r!r<_execute_groupr?r@rArBrCrDrErFrGrHrI)rrr-rr tree_configrYr;current_messagesrQlayeris_first_layergroups layer_outputs group_inputsrMrNrOrArPgrouprRrSrFrTs rr(z!DynamicEdgeExecutor._execute_trees$%335  F{~~FVWX X ++ "// %+ +-#D  # #D )$ I {~~./""%&6"7!88LZL Z "#a' QJE$$4jAF    " "";>>"2'%A!&k]'3  ,.M6{a%ay !#' #6#EL--k<PQR$$W-%(F \$B!''::GX:Y]eDFG&/&7 U', )+/ +>+ML!) //lESV"+.'8@BN","4"4"A"A'"J%fo"%+]]_F28N3/ #K %S[1.0)001DE 23Z:    " "";>>"2'%A ./x9  - "Ns{  &&&{~~&66OPE"#a'H {~~./$$)7->2B!E7!C5Q667S!:" "%ZYs7:A5L:0K7L:#L:7 L72L22L77L::MnoderL unit_indexc ~|jjd|jd|dt|d|Dcgc]}|j }}|D]#}t |j }||d<||_%|j||}|D]#}t |j }||d<||_%|Scc}w)zExecute a single map unit.rr7: executing with  inputsdynamic_edge_unit_index)rr!r"r$clonedictmetadatar)rrcrLrdmsgrkrMs rr=z!DynamicEdgeExecutor._execute_unitMs twwiq 5!+./w 8 /::kssyy{k :CCLL)H2H/ 0#CL"''CH '=sC%)N)__name__ __module__ __qualname____doc__r rrrr rrrr.r2r&r(intr=rZrrrrs 5 5%dDM%:DM%IJ 5&26 -M-M-M* -M  W . -M g -Mh26 6M6MW 6M* 6M  W . 6M g 6Mz26 EEd7m,E* E  W . E g EX26 t t d7m,t * t  W . t g t l']  g >##7m# #  # g #rr)rxconcurrent.futuresr@typingrrrrentity.configsr'entity.configs.edge.dynamic_edge_configrentity.messagesr r runtime.node.splitterr r utils.log_managerr rrzrrrs011E0M(}}r