动态边 Map/Tree 运行时分裂

动态边 Map/Tree 运行时分裂

学习目标

• 理解 ChatDev 的动态边执行管线：分裂 → 调度 → 收集
• 掌握 Map 和 Tree 两种动态模式的具体实现
• 了解 RegexSplitter 和 JSONPathSplitter 的分裂策略
• 能够在 YAML 中配置动态边并评估并行数量

前置知识

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

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

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

---

项目实践

动态边在 YAML 中的配置

ChatDev 的动态边配置挂载在边的 dynamic 字段上：

- from: Planner
  to: Executor
  trigger: true
  condition: 'true'
  dynamic:
    type: map                    # map 或 tree
    split:
      type: regex                # 分裂策略
      config:
        pattern: "<Query>:\\s*(.*)"   # 从 Planner 输出中提取 Query
    config:
      max_parallel: 3            # 最大并行数

典型场景：Planner Agent 输出多个 <Query>: xxx 行，动态边按正则分裂，每个 Query 启动一个 Executor 实例并行执行。

分裂策略实现

ChatDev 在 runtime/node/splitter.py 中实现了三种分裂策略：

RegexSplitter

class RegexSplitter(Splitter):
    def split(self, inputs: List[Message]) -> List[List[Message]]:
        for msg in inputs:
            text = msg.text_content()
            matches = list(self.pattern.finditer(text))

            if not matches:
                if self.on_no_match == "pass":
                    units.append([msg])          # 原样传递
                elif self.on_no_match == "empty":
                    units.append([empty_msg])    # 空消息
                continue

            for match in matches:
                # 提取捕获组 → 创建新消息
                extracted = match.group(self.group) if self.group else match.group(0)
                unit_msg = Message(role=msg.role, content=extracted)
                units.append([unit_msg])

关键点：on_no_match 行为可控——pass（原样传递）适合 fallback 场景，empty（空消息）适合静默跳过。

JSONPathSplitter

针对结构化 JSON 输出的分裂，适用于 Agent 以 JSON 数组形式返回任务的场景。

MessageSplitter

最基础的策略——每条消息一个执行单元。适用于批量消息场景。

Map 执行模式

DynamicEdgeExecutor._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  # 扁平化

Tree 执行模式

Tree 是 Map 的递归版本。配置方式相同，但 split 策略会在每个子节点输出上递归应用。

dynamic:
  type: tree
  split:
    type: regex
    config:
      pattern: "<Item>:\\s*(.*)"
  config:
    max_parallel: 5

执行流程：

输入 → Split → [A, B, C]
  A → Target Node → Split → [A1, A2] → Target Node
  B → Target Node → Split → [B1, B2, B3] → Target Node
  C → Target Node → 无分裂 → 直接输出
       ↓
  归约所有结果

执行上下文隔离

每个动态实例拥有独立的 ExecutionContext：

原始 ExecutionContext
    ↓ (复制)
实例 1 ExecutionContext
实例 2 ExecutionContext
实例 3 ExecutionContext

隔离保证：

• 记忆读写不互相干扰
• 工具调用历史独立
• 错误不传播到其他实例

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

动态节点可以同时接收两类输入：

# 来自动态边的输入（变化的）
<Query>: 搜索 Python 异步编程最佳实践

# 来自非动态边的输入（共享上下文）
[System Prompt] 你是一个技术研究员，请根据以下 Query 进行深度调研...

在代码中，execute_from_inputs() 方法将所有输入合并后传递给节点执行器。

问题与规避

问题	表现	规避策略
分裂数量爆炸	正则匹配到数百个子项	max_parallel 限流 + 精确正则
正则不匹配	输入格式变化导致分裂为空	测试 on_no_match 行为
静态输入缺失	动态实例缺少 System Prompt	确保有非动态边携带静态输入
结果顺序不确定	并发执行导致结果随机排序	下游节点不依赖结果顺序
实例间竞争	多个实例写入同一文件	每个实例独立工作目录
线程池阻塞	LLM 调用慢导致线程池占满	增大 max_parallel 或使用异步执行器

设计取舍

ThreadPoolExecutor vs Asyncio

ChatDev 选择 concurrent.futures.ThreadPoolExecutor 而非 asyncio：

维度	ThreadPoolExecutor	Asyncio
LLM I/O 阻塞	线程池隔离（阻塞不影响其他线程）	需确保 LLM SDK 支持 async
代码复杂度	低（标准库）	高（需全链路 async）
调试友好	高（每个线程独立）	低（事件循环共享）
ChatDev 的选择	✅ 使用	❌ 未采用

Map vs 预定义并行节点

维度	动态边 Map	预定义 N 个并行节点
YAML 行数	~10 行	~N × 30 行
数量变更	无需修改 YAML	需增删节点
调试难度	高（实例 ID 动态）	低（固定 ID）
LLM 可生成	是（Agent 输出 Query 即可）	否（需修改 YAML）

参考来源

• ChatDev DynamicEdgeExecutor 源码
• ChatDev Splitter 源码