ChatDev 动态执行模式实战指南边级 Map 扇出与 Tree 归约详解【免费下载链接】ChatDevChatDev 2.0: Dev All through LLM-powered Multi-Agent Collaboration项目地址: https://gitcode.com/GitHub_Trending/ch/ChatDevDynamic 执行模式是 ChatDev 工作流引擎中在边Edge级别定义并行处理行为的高级能力支持 Map扇出与 Tree扇出归约两种模式。当消息通过配置了dynamic的边时目标节点会根据拆分结果被虚拟扩展为多个并行实例适用于批量处理、并行查询、长文本摘要与层级聚合等场景。读完本文你将掌握 Dynamic 配置结构、三种 Split 拆分策略、Map/Tree 两种模式的完整 YAML 写法与底层执行原理并能直接复用仓库中开箱即用的真实示例。1. 概述模式描述输出适用场景Map扇出执行将消息拆分为多个单元并行处理List[Message]打平结果批量处理、并行查询Tree扇出归约并行处理后按组递归合并单个Message长文本摘要、层级聚合在 ChatDev 中动态执行既可以配置在节点上也可以配置在边上。本文聚焦边级配置动态配置定义在边上而非节点通过边的dynamic字段驱动下游目标节点的并行扩展。边级动态执行的核心实现位于 workflow/executor/dynamic_edge_executor.py其职责是拆分经边传递的负载 → 对每个拆分单元执行目标节点 → 收集并返回结果Map 打平、Tree 归约。2. 配置结构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 # 最大并发数从配置解析源码看dynamic是 EdgeConfig 的一个可选字段DynamicEdgeConfig | None与trigger、condition、carry_data、process等并列只有当dynamic存在且非空时才会解析为动态配置。其底层数据结构定义在 entity/configs/edge/dynamic_edge_config.pytype必填字符串只能是map或tree二者通过注册表register_dynamic_edge_type注册分别绑定MapDynamicConfig与TreeDynamicConfigsplit可选SplitConfig类型缺省时如 YAML 中写split: null或不写自动回退为message模式config模式特定配置Map 下解析为MapDynamicConfig含max_parallelTree 下解析为TreeDynamicConfig含group_size、max_parallel。2.1 核心概念动态边配置了dynamic的边其传递的消息会触发目标节点的动态扩展。静态边未配置dynamic的边其传递的消息会复制到所有动态扩展实例。目标节点扩展目标节点根据 split 结果被虚拟扩展为多个并行实例。在执行层当节点存在动态入边时workflow/graph.py 会调用_get_dynamic_config_for_node取得统一的动态配置随后经_execute_with_dynamic_configworkflow/graph.py把输入消息按来源分为两类元数据带_from_dynamic_edge标记的动态输入参与拆分与其余静态输入复制到每个单元再交给DynamicEdgeExecutor.execute_from_inputs执行。2.2 多入边一致性规则[!IMPORTANT] 当一个节点有多条入边配置了dynamic时所有动态边的配置必须完全一致type、split、config否则执行时会报错。这一规则在 workflow/graph.py 中实现为逐项校验系统会以第一条动态边为准逐一比对后续动态边的type、split含split.type、pattern、json_path、max_parallel以及 Tree 模式下的group_size任何不一致都会抛出WorkflowExecutionError并在报错信息中明确提示是哪两条入边配置冲突。实际使用中如果同一目标节点需要不同的拆分策略应拆分为独立的中间节点或子图而不是在同一节点上混用多套动态配置。3. Split 拆分策略Split 定义如何将通过边的消息拆分为并行执行单元。三种策略的配置类集中在 entity/configs/dynamic_base.py实际拆分逻辑则在 runtime/node/splitter.py 中由Splitter抽象基类的三个子类实现。3.1 message 模式默认每条通过边的消息作为独立执行单元。这是最常用的模式无需任何额外参数。split: type: message执行行为源节点输出 4 条消息通过动态边拆分为 4 个并行单元目标节点执行 4 次源码实现MessageSplitter.split 将每条输入消息包装为单个单元[[msg] for msg in inputs]即一消息一单元。SplitConfig.from_dict对message类型允许完全省略config字段entity/configs/dynamic_base.py所以最小写法就是type: message一行。3.2 regex 模式使用正则表达式从文本内容中提取匹配项每个匹配结果成为一个执行单元。split: type: regex pattern: (?s).{1,2000}(?:\\s|$) # 每 2000 字符切分典型用例按段落拆分pattern: \\n\\n按行拆分pattern: .按固定长度pattern: (?s).{1,N}源码实现RegexSplitter.split 对每条消息的文本调用pattern.finditer每个 match 生成一个单元除pattern外RegexSplitConfig还支持以下进阶选项见 entity/configs/dynamic_base.py字段类型默认值说明groupstr/int无整段匹配 group 0指定提取的捕获组名称或索引case_sensitivebooltrue是否区分大小写false时编译re.IGNORECASEmultilineboolfalse启用re.MULTILINEdotallboolfalse启用re.DOTALL让.匹配换行on_no_matchenumpass无匹配时行为pass保留原文作为单元empty返回空内容单元注意无匹配时的兜底逻辑若所有输入均无匹配且on_no_matchpasssplitter 会回退为每条输入消息一个单元runtime/node/splitter.py避免出现零单元导致目标节点完全不执行的情况。3.3 json_path 模式从 JSON 格式输出中按路径提取数组元素每个数组元素成为一个执行单元。split: type: json_path json_path: $.items[*] # JSONPath 表达式源码实现JsonPathSplitter.split 先用json.loads解析消息文本再通过简化点分路径如items、data.results逐层定位数组runtime/node/splitter.py路径为空时若数据本身是列表则直接使用路径中的数字段可索引列表。元素为 dict/list 时序列化为 JSON 字符串其余转为字符串。若消息不是合法 JSON则整条消息作为一个单元兜底。提示$.items[*]这类标准 JSONPath 语法在语义上等同于源码中的items点分写法实际路径解析采用split(.)的点分遍历因此配置时建议直接写items、data.results这类点分路径。4. Map 模式详解Map 模式将消息拆分后并行执行目标节点输出结果打平为List[Message]。4.1 配置项字段类型默认值说明max_parallelint10最大并发执行数MapDynamicConfig定义在 entity/configs/dynamic_base.py是最简配置型模式config可整体省略缺省即max_parallel10。4.2 执行流程源码实现DynamicEdgeExecutor._execute_map 使用ThreadPoolExecutor并行执行工作线程数为min(拆分单元数, max_parallel)当只有 1 个单元时直接同步执行避免线程池开销。结果按单元原始顺序合并results_by_idx按索引归位因此输出顺序与拆分顺序一致。每个单元的输入与输出消息都会被打上dynamic_edge_unit_index元数据标记便于日志追踪与后续处理。可结合仓库示例 yaml_instance/demo_dynamic.yaml节点 Z 的dynamic.type: map配置理解完整图结构。以下截图展示了 ChatDev 前端工作台中 Map 模式的配置界面动态类型选择、Split 策略与 Max Parallel 参数5. Tree 模式详解Tree 模式在 Map 基础上增加归约层将并行结果按组递归合并最终输出单个结果。5.1 配置项字段类型默认值说明group_sizeint3每组归约的元素数量最小为 2max_parallelint10每层最大并发执行数TreeDynamicConfig定义在 entity/configs/dynamic_base.py其中group_size在校验阶段即要求 2否则抛出ConfigError(group_size must be at least 2)。5.2 执行流程源码实现DynamicEdgeExecutor._execute_tree 的执行逻辑值得细读先将所有拆分单元打平为消息列表current_messages进入归约循环只要消息数 1就继续分层每层调用group_messages(current_messages, group_size)runtime/node/splitter.py 按group_size滑窗分批最后一组可能不满每层并行度同样受max_parallel限制min(组数, max_parallel)每组的输入输出消息会打上dynamic_edge_tree_layer、dynamic_edge_tree_group、dynamic_edge_instance_id等元数据且归约输出被标记为MessageRole.USER视为用户生成内容循环以layer 100为安全熔断避免异常场景下无限分层。Tree 模式的一个关键语义静态边消息只在第一层参与归约输入if is_first_layer: group_inputs list(static_inputs) group_inputs后续各层只聚合上一层输出避免静态指令在多轮归约中被重复注入。以下截图展示了 Tree 模式的配置界面包含 regex 拆分策略、Group Size 与 Max Parallel 参数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 条任务 请使用简洁的语言源码依据在执行入口处workflow/graph.py 将输入按_from_dynamic_edge标记区分为动态/静态两类随后 DynamicEdgeExecutor 在 Map 模式中把静态输入拼接在每个单元输入之前unit_inputs list(static_inputs) unit见 workflow/executor/dynamic_edge_executor.py 与 L180实现复制到所有实例。并行执行时各单元会先clone()消息再打标确保多线程下不共享可变状态。这一模式非常适合批量任务 统一约束指令的组合例如并行处理多份文档的同时每条都附带相同的格式要求。7. 完整示例7.1 旅行规划Map Tree 组合以下示例展示 Map 扇出与 Tree 归约的典型串联三个独立规划请求经 Map 并行执行再由 Tree 归约为一份完整旅行计划。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仓库中有一个与该示例同构的完整可运行配置yaml_instance/demo_dynamic.yaml。该图包含 8 个节点多个 literal 规划请求、passthrough 收集器、Map 模式的 agent 节点 Z、Tree 归约的 agent 节点 Y其中P → Z边配置dynamic: mapmax_parallel: 5Z → Y边配置dynamic: treegroup_size: 2同时E → Z与F → Z也是动态边演示了多条动态边指向同一节点的场景。注意该示例中Z节点自身也配置了dynamic字段属于边级 节点级并存的图读者可将边级配置作为主路径理解。7.2 长文档摘要Tree 模式将超长文本按 2000 字符切块并行摘要后再树状归约最终得到整体摘要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 完整演示了这一场景一个超长小说文本作为literal节点输出通过A → B边上的 Tree 动态配置regex 按 2000 字符切分、group_size: 3、max_parallel: 10交给 agent 节点 B角色为请总结小说内容进行分层并行摘要与归约。该文件可直接用于本地验证 Tree 模式的分块并行与递归合并行为。8. 性能建议控制并发设置合理的max_parallel避免触发 API 限流。从源码看max_parallel直接决定ThreadPoolExecutor的工作线程数Map 为min(单元数, max_parallel)Tree 为每层min(组数, max_parallel)值过大会在单次执行中同时发起大量 LLM 调用。优化拆分粒度过细的拆分增加开销每条消息都要经过一次节点执行与上下文打包过粗则无法充分并行。message 模式下拆分单元数即消息数regex/json_path 模式下由匹配数或数组长度决定。Tree 组大小group_size2-4通常是较好的选择。它决定每层归约合并的条数值越小归约层数越多、单次归约上下文越短值越大每层输出越少但单次归约输入越长。源码允许的最小值为 2。监控成本Dynamic 模式会显著增加 API 调用次数。Map 模式的总调用量等于拆分单元数Tree 模式约为第一层单元数 各归约层组数之和并在每个单元上都可能产生 LLM 推理开销需结合max_parallel与group_size综合评估预算。9. 相关文档工作流编排指南理解图、节点与边编排的整体方法论Agent 节点配置动态扩展的目标节点agent 类型的完整配置说明执行逻辑指南了解节点执行、消息传递与触发机制配置结构契约Dynamic 相关配置字段的模式校验与字段规格【免费下载链接】ChatDevChatDev 2.0: Dev All through LLM-powered Multi-Agent Collaboration项目地址: https://gitcode.com/GitHub_Trending/ch/ChatDev创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考