动态边执行器：Map/Tree 分裂与并行调度

动态边执行器：Map/Tree 分裂与并行调度

学习目标

• 理解 DynamicEdgeExecutor 的完整执行管线
• 掌握 Map 和 Tree 模式的实现差异
• 了解并发控制机制和结果收集策略
• 能够诊断动态边执行中的常见问题

前置知识

本章涉及动态 Agent 执行的通用原理，建议先阅读：

• 动态 Agent 执行：Map/Tree 运行时分裂

下文聚焦 ChatDev 的具体动态边执行器实现。

---

项目实践

执行管线

DynamicEdgeExecutor.execute() 的完整流程：

def execute(self, target_node, payload, dynamic_config, static_inputs=None):
    # 1. 创建分裂器
    splitter = create_splitter_from_config(dynamic_config.split)

    # 2. 分裂输入
    execution_units = splitter.split([payload])

    if not execution_units:
        return []  # 无分裂结果，early return

    # 3. 选择模式
    if dynamic_config.is_map():
        return self._execute_map(target_node, execution_units, dynamic_config, static_inputs)
    elif dynamic_config.is_tree():
        return self._execute_tree(target_node, execution_units, dynamic_config, static_inputs)

Map 执行

_execute_map() 的并发调度：

def _execute_map(self, target_node, execution_units, dynamic_config, static_inputs):
    results = []
    with concurrent.futures.ThreadPoolExecutor(
        max_workers=dynamic_config.max_parallel
    ) as executor:
        futures = []
        for unit in execution_units:
            merged_inputs = unit + (static_inputs or [])
            future = executor.submit(
                self.node_executor_func, target_node, merged_inputs
            )
            futures.append(future)

        for future in concurrent.futures.as_completed(futures):
            results.extend(future.result())

    return results

关键设计：

• 线程池限流：max_workers=dynamic_config.max_parallel
• 结果扁平化：results.extend() 将每个实例的输出列表展平
• 输入合并：unit + static_inputs 合并动态单元和静态上下文

Tree 执行

_execute_tree() 的递归分裂：

def _execute_tree(self, target_node, execution_units, dynamic_config, static_inputs):
    results = []
    for unit in execution_units:
        merged_inputs = unit + (static_inputs or [])
        # 执行当前节点
        output = self.node_executor_func(target_node, merged_inputs)
        # 检查输出是否需要进一步分裂
        if dynamic_config.should_split(output):
            # 递归分裂
            sub_results = self.execute(target_node, output, dynamic_config, static_inputs)
            results.extend(sub_results)
        else:
            results.extend(output if isinstance(output, list) else [output])
    return results

从完整输入执行

execute_from_inputs() 方法处理已收集的全部输入：

def execute_from_inputs(self, target_node, inputs, dynamic_config, static_inputs=None):
    # 与 execute() 类似，但输入是已收集的完整列表
    # 适用于 _execute_node 中边收集完成后的统一处理
    splitter = create_splitter_from_config(dynamic_config.split)
    execution_units = splitter.split(inputs)
    # ... 与 execute() 相同的后续流程

分裂器工厂

create_splitter_from_config() 根据 YAML 配置创建对应的分裂器：

def create_splitter_from_config(split_config) -> Splitter:
    if split_config.type == "regex":
        return RegexSplitter(
            pattern=split_config.config.pattern,
            group=split_config.config.group,
            case_sensitive=split_config.config.case_sensitive,
            multiline=split_config.config.multiline,
            dotall=split_config.config.dotall,
            on_no_match=split_config.config.on_no_match,
        )
    elif split_config.type == "json_path":
        return JsonPathSplitter(
            path=split_config.config.json_path,
        )
    elif split_config.type == "message":
        return MessageSplitter()

静态输入与动态输入的合并

在 workflow/graph.py 的 _execute_node() 中：

def _execute_node(node):
    # 收集所有输入边的消息
    inputs = collect_inputs(node)

    # 区分动态边和静态边
    static_inputs = [msg for msg, edge in inputs if not edge.has_dynamic()]
    dynamic_payloads = [msg for msg, edge in inputs if edge.has_dynamic()]

    # 动态边处理
    if dynamic_payloads:
        for payload in dynamic_payloads:
            dynamic_results = dynamic_edge_executor.execute(
                target_node=node,
                payload=payload,
                dynamic_config=edge.dynamic_config,
                static_inputs=static_inputs,
            )
            node.set_output(dynamic_results)
    else:
        # 普通执行
        node.set_output(node_executor.execute(node, inputs))

问题与规避

问题	表现	规避策略
线程池耗尽	max_parallel 过小导致排队	根据 LLM API QPS 调整
结果收集顺序不确定	as_completed 返回顺序随机	下游节点不依赖顺序
递归 Tree 深度过大	栈溢出	限制 Tree 递归深度
分裂后上下文丢失	动态单元不含 System Prompt	确保 static_inputs 包含必要上下文

设计取舍

ThreadPoolExecutor 的 max_workers 选择

ChatDev 没有在 DynamicEdgeExecutor 中硬编码线程池大小，而是从 YAML 配置的 max_parallel 获取：

• 硬编码：简单但不灵活
• 配置化：灵活但需要用户理解并发概念

ChatDev 选择配置化，因为不同场景的并发需求差异巨大（3 个查询 vs 100 个文件处理）。

参考来源

• ChatDev DynamicEdgeExecutor 源码