ChatDev 2.0 Dynamic 执行模式深度指南边级 Map 扇出与 Tree 归约的并行编排实战【免费下载链接】ChatDevChatDev 2.0: Dev All through LLM-powered Multi-Agent Collaboration项目地址: https://gitcode.com/Dennis_Huang/ChatDevChatDev 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一个字段与 TreeDynamicConfiggroup_sizemax_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当发现多于一份动态配置时逐项比对任何不一致都会抛出WorkflowExecutionErrortype不一致报 inconsistent dynamic configurationssplit.type/split.pattern/split.json_path不一致报 inconsistent split configurationsmax_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_matchpass。除pattern外RegexSplitConfig 还支持以下可选参数均带默认值字段类型默认值说明groupstr/intNone捕获组名或索引默认取整个匹配组 0case_sensitivebooltrue是否区分大小写对应re.IGNORECASEmultilineboolfalse启用re.MULTILINEdotallboolfalse启用re.DOTALLon_no_matchenumpass无匹配时行为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_parallelint10最大并发执行数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_sizeint3每组归约的元素数量最小为 2max_parallelint10每层最大并发执行数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: 23 个结果在第一层按 2 组归约2 1得到 2 条中间结果第二轮归约成 1 条最终计划共调用 Final Aggregator 2 次。该拓扑与仓库内置的 yaml_instance/demo_dynamic.yaml 一脉相承——该示例同样以上海旅行规划为题材A/B/C/D/F多个 literal 请求汇入 passthrough 节点PP → Z为map动态边max_parallel: 5Z → 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 是该场景的完整可运行版本节点Aliteral携带一段长篇小说文本经A → B的动态边进入节点Bagent 摘要节点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_size2-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.pydynamic字段解析入口、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.mdAgent 节点配置docs/user_guide/zh/nodes/agent.md【免费下载链接】ChatDevChatDev 2.0: Dev All through LLM-powered Multi-Agent Collaboration项目地址: https://gitcode.com/Dennis_Huang/ChatDev创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考