ChatDev 2.0 Dynamic 执行模式深度指南：边级 Map 扇出与 Tree 归约的并行编排实战

ChatDev 2.0 Dynamic 执行模式深度指南：边级 Map 扇出与 Tree 归约的并行编排实战

【免费下载链接】ChatDevChatDev 2.0: Dev All through LLM-powered Multi-Agent Collaboration项目地址: https://gitcode.com/Dennis_Huang/ChatDev

ChatDev 2.0 的 Dynamic 执行模式允许在**边（edge）**级别定义并行处理行为，支持 Map（扇出）与 Tree（扇出 + 归约）两种模式，是批量处理、并行查询、长文本摘要与层级聚合场景下的核心能力。本指南以 docs/user_guide/zh/dynamic_execution.md 为主干，结合 entity/configs/edge/dynamic_edge_config.py、workflow/executor/dynamic_edge_executor.py 等源码与 yaml_instance/demo_dynamic.yaml、yaml_instance/demo_dynamic_tree.yaml 真实示例，带你掌握从配置书写到底层执行原理的完整链路。

1. 概述：两种动态执行模式

Dynamic 执行模式允许在边级别定义并行处理行为，支持Map（扇出）和Tree（扇出 + 归约）两种模式。当消息通过配置了dynamic的边传递时，目标节点会根据拆分结果动态扩展为多个并行实例。下表是两种模式的定位对比：

模式	描述	输出	适用场景
Map	扇出执行，将消息拆分为多个单元并行处理	List[Message]（打平结果）	批量处理、并行查询
Tree	扇出 + 归约，并行处理后按组递归合并	单个Message	长文本摘要、层级聚合

两种模式对应源码中的两个配置类：MapDynamicConfig（仅max_parallel一个字段）与 TreeDynamicConfig（group_size+max_parallel），二者在 dynamic_edge_config.py 末尾通过register_dynamic_edge_type("map", ...)与register_dynamic_edge_type("tree", ...)注册为边级动态类型，并对外暴露为type字段的可选项枚举。

2. 配置结构：定义在边上的dynamic块

Dynamic 配置定义在边上，而非节点。在边配置中新增dynamic字段即可启用：

edges: - from: Source Node to: Target Node trigger: true carry_data: true dynamic: # 边级动态执行配置 type: map # map 或 tree split: # 消息拆分策略 type: message # message | regex | json_path # pattern: "..." # regex 模式必填 # json_path: "..." # json_path 模式必填 config: # 模式特定配置 max_parallel: 5 # 最大并发数

这段配置对应 EdgeConfig 中的dynamic: DynamicEdgeConfig | None = None字段。边加载时若存在dynamic键，会调用DynamicEdgeConfig.from_dict完成解析：DynamicEdgeConfig 内部通过注册表按type找到MapDynamicConfig或TreeDynamicConfig，再分别解析split与config。若type不在已注册的动态类型中，会抛出ConfigError，提示可用的类型列表。

2.1 核心概念

• 动态边：配置了dynamic的边，其传递的消息会触发目标节点的动态扩展。
• 静态边：未配置dynamic的边，其传递的消息会复制到所有动态扩展实例。
• 目标节点扩展：目标节点根据 split 结果被"虚拟"扩展为多个并行实例——节点本身不变，执行器在运行时为每个拆分单元实例化一次执行。

2.2 多入边一致性规则

[!IMPORTANT] 当一个节点有多条入边配置了dynamic时，所有动态边的配置必须完全一致（type、split、config），否则执行时会报错。

该规则在 workflow/graph.py 的_get_dynamic_config_for_node方法中落地：执行前会遍历目标节点的所有前驱出边，收集dynamic_config，当发现多于一份动态配置时逐项比对，任何不一致都会抛出WorkflowExecutionError：

• type不一致：报 "inconsistent dynamic configurations"；
• split.type/split.pattern/split.json_path不一致：报 "inconsistent split configurations"；
• max_parallel不一致：单独报错；
• Tree 模式下group_size不一致：单独报错。

设计意图很清晰：同一节点的所有动态入边共享同一套拆分与并发参数，才能保证"拆分单元 → 并行实例 → 结果合并"的语义唯一且可预期。因此当多个上游节点都需要扇出到同一个下游时，请在每条动态边上书写完全相同的dynamic配置。

3. Split 拆分策略

Split 定义如何将通过边的消息拆分为并行执行单元。三种策略在 runtime/node/splitter.py 中分别对应MessageSplitter、RegexSplitter、JsonPathSplitter，由工厂函数create_splitter_from_config按split.type实例化。

3.1 message 模式（默认）

每条通过边的消息作为独立执行单元。这是最常用的模式。

split: type: message

执行行为：

• 源节点输出 4 条消息通过动态边
• 拆分为 4 个并行单元，目标节点执行 4 次

在配置层面，SplitConfig.type的默认值即为"message"（见 entity/configs/dynamic_base.py），即使完全不写split字段，也会回退到 message 拆分。源码实现为MessageSplitter.split：对每条输入消息包装成[msg]作为独立单元。

3.2 regex 模式

使用正则表达式从文本内容中提取匹配项。

split: type: regex pattern: "(?s).{1,2000}(?:\\s|$)" # 每 2000 字符切分

典型用例：

• 按段落拆分：pattern: "\\n\\n"
• 按行拆分：pattern: ".+"
• 按固定长度：pattern: "(?s).{1,N}"

正则策略由 RegexSplitter 实现：对每条消息的文本内容执行finditer，每个匹配项生成一个执行单元；没有匹配时默认按原消息整体作为单元（on_no_match="pass"）。除pattern外，RegexSplitConfig 还支持以下可选参数（均带默认值）：

字段	类型	默认值	说明
group	str/int	None	捕获组名或索引，默认取整个匹配（组 0）
case_sensitive	bool	true	是否区分大小写（对应re.IGNORECASE）
multiline	bool	false	启用re.MULTILINE
dotall	bool	false	启用re.DOTALL
on_no_match	enum	"pass"	无匹配时行为："pass"原样保留消息，"empty"返回空内容单元

拆分粒度直接决定并发单元数量，后续在第 8 节性能建议中会展开说明取舍。

3.3 json_path 模式

从 JSON 格式输出中按路径提取数组元素。

split: type: json_path json_path: "$.items[*]" # JSONPath 表达式

需要说明的是：JsonPathSplitter 的底层实现采用简单点号记法（源码注释示例为'items'、'data.results'），按.切分后逐层从 dict/list 中取值，找到 list 后展开为单元；若消息文本无法解析为 JSON 则整体作为一个单元。因此如果目标输出 JSON 是{"items": [...]}结构，建议配置json_path: "items"；若你的运行环境支持文档示例中的标准 JSONPath 写法（$.items[*]），请以实际验证结果为准——配置前最好用一条真实消息快速验证提取结果。解析出的数组元素若为 dict/list 会以 JSON 字符串形式作为单元内容，否则转为普通字符串。

4. Map 模式详解

Map 模式将消息拆分后并行执行目标节点，输出结果打平为List[Message]。

4.1 配置项

字段	类型	默认值	说明
max_parallel	int	10	最大并发执行数

MapDynamicConfig是配置最轻的动态类型，语义上"类似 passthrough，仅需极简配置"（源码类注释原文），max_parallel缺失时默认取 10。

4.2 执行流程

对应源码实现位于 dynamic_edge_executor.py 的_execute_map：

• 拆分单元数 = 1：直接执行，不创建线程池；
• 拆分单元数 > 1：创建ThreadPoolExecutor，工作线程数为min(单元数, max_parallel)，每个单元提交一个任务；
• 结果保序：每个任务在metadata中打上dynamic_edge_unit_index标记，全部完成后按单元索引顺序合并输出，保证打平后的List[Message]顺序与拆分顺序一致；
• 失败即抛：任一单元执行失败会记录日志并向上抛出异常，终止整体执行。

Map 模式下，每个单元的输入 =静态边复制来的消息 + 本单元拆分出的消息（顺序为静态输入在前）。

5. Tree 模式详解

Tree 模式在 Map 基础上增加归约层，将并行结果按组递归合并，最终输出单个结果。

5.1 配置项

字段	类型	默认值	说明
group_size	int	3	每组归约的元素数量，最小为 2
max_parallel	int	10	每层最大并发执行数

group_size的"最小为 2"约束由 TreeDynamicConfig.from_dict 强制执行：group_size < 2时抛出ConfigError("group_size must be at least 2")。

5.2 执行流程

对应源码实现为_execute_tree，其归约循环值得逐条对应：

• 先将各拆分单元的消息打平为单个消息列表（current_messages）；
• 若len(current_messages) <= 1，直接返回，不进入归约（避免无意义调用）；
• 循环条件while len(current_messages) > 1：每轮用 group_messages 按group_size切组（最后一组允许不足），每个组交给目标节点执行一次，组数 > 1 时同样用ThreadPoolExecutor并行（并发数为min(组数, max_parallel)）；
• 静态边消息只在第一层拼入各组输入（is_first_layer标志），后续层不再重复注入；
• 每层输出重新赋给current_messages，继续下一轮，直到只剩 1 条消息；
• 安全兜底：层数超过 100 时记录错误并终止循环，防止异常配置导致无限归约。

Tree 模式的输出消息同样会被打上dynamic_edge_tree_layer、dynamic_edge_tree_group、dynamic_edge_instance_id元数据，并被标记为MessageRole.USER，便于下游链路识别归约来源。

6. 静态边消息复制

当目标节点同时有动态入边和静态入边时：

• 动态边消息：按 split 策略拆分，每个单元执行一次目标节点
• 静态边消息：复制到每个动态扩展实例

nodes: - id: Task Generator type: passthrough config: ... - id: Extra Requirement type: literal config: content: "请使用简洁的语言" - id: Processor type: agent config: name: gpt-4o role: 处理任务 edges: - from: Task Generator to: Processor dynamic: # 动态边：4 条任务 → 4 个并行单元 type: map split: type: message config: max_parallel: 10 - from: Extra Requirement to: Processor # 静态边：复制到所有 4 个实例 trigger: true carry_data: true

执行结果：Processor 执行 4 次，每次收到 1 条任务 + "请使用简洁的语言"

源码层面，静态输入在_execute_map中通过unit_inputs = list(static_inputs) + unit注入每个单元；在_execute_unit中，所有输入消息会被clone()后再打标与提交，避免并行线程共享可变消息对象造成数据竞争。Tree 模式下静态输入仅注入第一层各组（见第 5.2 节）。

7. 完整示例

7.1 旅行规划（Map + Tree 组合）

三个规划请求经 passthrough 汇聚后，先由 Map 动态边扇出为 3 个并行执行单元，再由 Tree 动态边归约为 1 份完整旅行计划：

graph: nodes: - id: Eat Planner type: literal config: content: 请规划在上海吃什么 role: user - id: Play Planner type: literal config: content: 请规划在上海玩什么 role: user - id: Stay Planner type: literal config: content: 请规划在上海住哪里 role: user - id: Collector type: passthrough config: only_last_message: false - id: Travel Executor type: agent config: name: gpt-4o role: 你是旅行规划师，请按照用户请求进行规划 - id: Final Aggregator type: agent config: name: gpt-4o role: 请将输入的内容整合成一份完整的旅行计划 edges: - from: Eat Planner to: Collector - from: Play Planner to: Collector - from: Stay Planner to: Collector - from: Collector to: Travel Executor dynamic: # Map 扇出：3 个规划请求 → 3 个并行执行 type: map split: type: message config: max_parallel: 10 - from: Travel Executor to: Final Aggregator dynamic: # Tree 归约：3 个结果 → 1 个最终计划 type: tree split: type: message config: group_size: 2 max_parallel: 10

注意这里group_size: 2：3 个结果在第一层按 2 组归约（2 + 1），得到 2 条中间结果，第二轮归约成 1 条最终计划，共调用 Final Aggregator 2 次。该拓扑与仓库内置的 yaml_instance/demo_dynamic.yaml 一脉相承——该示例同样以"上海旅行规划"为题材：A/B/C/D/F多个 literal 请求汇入 passthrough 节点P，P → Z为map动态边（max_parallel: 5），Z → Y为tree动态边（group_size: 3）完成聚合。

7.2 长文档摘要（Tree 模式）

长文本先由 regex 按 2000 字符切块，再以 Tree 模式分层并行摘要并归约：

edges: - from: Document Source to: Summarizer dynamic: type: tree split: type: regex pattern: "(?s).{1,2000}(?:\\s|$)" # 2000 字符切分 config: group_size: 3 max_parallel: 10

仓库中的 yaml_instance/demo_dynamic_tree.yaml 是该场景的完整可运行版本：节点A（literal，携带一段长篇小说文本）经A → B的动态边进入节点B（agent 摘要节点），dynamic配置为tree + regex(pattern: "(?s).{1,2000}(?:\\s|$)") + group_size: 3 + max_parallel: 10。文本被切为若干 2000 字符块后并行摘要，再分层归约，最终得到整篇小说的单一摘要。这也是 Web 端配置界面中dynamic_tree.png截图的真实映射。

8. 性能建议

• 控制并发：设置合理的max_parallel避免触发 API 限流。并发是全局性资源，max_parallel过高会同时放大对上游 LLM 服务的请求压力，执行器实际工作线程数为min(单元数, max_parallel)。
• 优化拆分粒度：过细的拆分增加开销，过粗则无法充分并行。regex 的(?s).{1,N}每块长度 N 直接决定单元数量：N 越小块越多、并发越高、但每单元上下文越短（摘要质量可能下降），需要结合模型上下文窗口权衡。
• Tree 组大小：group_size=2-4通常是较好的选择。更大的组减少归约层数、节省调用次数，但每层单次归约的输入更多，对模型的整合能力要求更高；注意最小值为 2。
• 监控成本：Dynamic 模式会显著增加 API 调用次数。Map 至少执行"单元数"次，Tree 为"每层组数之和"次，归约层会反复调用目标节点，调试与成本核算时应以日志中的Dynamic edge -> {node.id}: splitting into N parallel units与Tree completed after L layers等信息为准。

9. 相关文档与源码导航

• 边配置实现：entity/configs/edge/edge.py（dynamic字段解析入口）、entity/configs/edge/dynamic_edge_config.py（动态边配置类与注册表）
• 拆分策略与动态配置基类：runtime/node/splitter.py、entity/configs/dynamic_base.py
• 动态执行器（Map/Tree 并发实现）：workflow/executor/dynamic_edge_executor.py
• 多入边一致性校验：workflow/graph.py（_get_dynamic_config_for_node）
• 可运行示例：yaml_instance/demo_dynamic.yaml、yaml_instance/demo_dynamic_tree.yaml、yaml_instance/deep_research_v1.yaml、yaml_instance/MACNet_v1.yaml
• 工作流编排指南：docs/user_guide/zh/workflow_authoring.md
• Agent 节点配置：docs/user_guide/zh/nodes/agent.md

【免费下载链接】ChatDevChatDev 2.0: Dev All through LLM-powered Multi-Agent Collaboration项目地址: https://gitcode.com/Dennis_Huang/ChatDev