企业官网建设流程全解析

ChatDev 动态执行模式实战指南：边级 Map 扇出与 Tree 归约详解

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

Dynamic 执行模式是 ChatDev 工作流引擎中在边（Edge）级别定义并行处理行为的高级能力，支持 Map（扇出）与 Tree（扇出+归约）两种模式。当消息通过配置了dynamic的边时，目标节点会根据拆分结果被"虚拟"扩展为多个并行实例，适用于批量处理、并行查询、长文本摘要与层级聚合等场景。读完本文，你将掌握 Dynamic 配置结构、三种 Split 拆分策略、Map/Tree 两种模式的完整 YAML 写法与底层执行原理，并能直接复用仓库中开箱即用的真实示例。

1. 概述

在 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.py：

• type：必填字符串，只能是map或tree，二者通过注册表register_dynamic_edge_type注册，分别绑定MapDynamicConfig与TreeDynamicConfig；
• split：可选，SplitConfig类型，缺省时（如 YAML 中写split: null或不写）自动回退为message模式；
• config：模式特定配置，Map 下解析为MapDynamicConfig（含max_parallel），Tree 下解析为TreeDynamicConfig（含group_size、max_parallel）。

2.1 核心概念

• 动态边：配置了dynamic的边，其传递的消息会触发目标节点的动态扩展。
• 静态边：未配置dynamic的边，其传递的消息会复制到所有动态扩展实例。
• 目标节点扩展：目标节点根据 split 结果被"虚拟"扩展为多个并行实例。

在执行层，当节点存在动态入边时，workflow/graph.py 会调用_get_dynamic_config_for_node取得统一的动态配置，随后经_execute_with_dynamic_config（workflow/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）：

注意无匹配时的兜底逻辑：若所有输入均无匹配且on_no_match=pass，splitter 会回退为"每条输入消息一个单元"（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 配置项

MapDynamicConfig定义在 entity/configs/dynamic_base.py，是"最简配置"型模式：config可整体省略，缺省即max_parallel=10。

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 配置项

TreeDynamicConfig定义在 entity/configs/dynamic_base.py，其中group_size在校验阶段即要求>= 2，否则抛出ConfigError("group_size must be at least 2")。

5.2 执行流程

源码实现：DynamicEdgeExecutor._execute_tree 的执行逻辑值得细读：

1. 先将所有拆分单元打平为消息列表current_messages；
2. 进入归约循环：只要消息数> 1就继续分层，每层调用group_messages(current_messages, group_size)（runtime/node/splitter.py 按group_size滑窗分批，最后一组可能不满）；
3. 每层并行度同样受max_parallel限制（min(组数, max_parallel)）；
4. 每组的输入输出消息会打上dynamic_edge_tree_layer、dynamic_edge_tree_group、dynamic_edge_instance_id等元数据，且归约输出被标记为MessageRole.USER（视为用户生成内容）；
5. 循环以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: map（max_parallel: 5），Z → Y边配置dynamic: tree（group_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_size=2-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