图执行器与策略模式

图执行器与策略模式

学习目标

• 理解 GraphExecutor 的构建链与运行时上下文组装
• 掌握三种执行策略的实现细节与切换逻辑
• 了解取消信号（threading.Event）的优雅处理
• 能够分析 ChatDev 的图执行策略选择

---

项目实践

GraphExecutor 构建链

ChatDev 的图构建是一条清晰的管线：

核心入口：

executor = GraphExecutor(
    graph,
    session_id=session_id,
    workspace_hook_factory=workspace_hook_factory,
    cancel_event=cancel_event,
)
executor._build_memories_and_thinking()  # 初始化记忆和思考
executor.run(task_prompt)                 # 执行

RuntimeBuilder 组装

RuntimeBuilder 是一个 dataclass，负责构建 RuntimeContext：

@dataclass
class RuntimeBuilder:
    graph: GraphContext

    def build(self, logger=None, *, session_id=None) -> RuntimeContext:
        tool_manager = ToolManager()                                    # 工具管理
        function_manager = get_function_manager(EDGE_FUNCTION_DIR)      # 函数工具
        processor_function_manager = get_function_manager(EDGE_PROCESSOR_FUNCTION_DIR)  # 边处理器
        log_manager = LogManager(logger)                                # 日志管理
        token_tracker = TokenTracker(workflow_id=self.graph.name)       # Token 追踪
        attachment_store = AttachmentStore(attachments_dir)             # 附件存储

        return RuntimeContext(
            tool_manager=tool_manager,
            function_manager=function_manager,
            edge_processor_function_manager=processor_function_manager,
            log_manager=log_manager,
            token_tracker=token_tracker,
            attachment_store=attachment_store,
            code_workspace=code_workspace,
            global_state=global_state,
        )

三种执行策略的实现

1. DAGExecutor

DAGExecutor 负责无环图的层间屏障执行：

class DAGExecutor:
    def __init__(self, log_manager, nodes, layers, execute_node_func):
        self.layers = layers  # [[node_ids_layer0], [layer1], ...]

    def execute(self):
        for layer in self.layers:
            # 层内并行
            for node_id in layer:
                self.execute_node_func(nodes[node_id])
            # 层间屏障（隐式：for 循环等待所有节点完成）

拓扑分层使用 Kahn 算法，由 TopologyBuilder 预先计算。

2. CycleExecutor

CycleExecutor 处理含环图的迭代执行：

class CycleExecutor:
    def __init__(self, log_manager, nodes, cycle_execution_order, cycle_manager, execute_node_func):
        self.cycle_execution_order = cycle_execution_order  # 环内节点执行顺序
        self.cycle_manager = cycle_manager                  # 环检测与管理

    def execute(self):
        for cycle_step in self.cycle_execution_order:
            node_id = cycle_step["node_id"]
            # 执行节点
            self.execute_node_func(nodes[node_id])
            # 检查是否满足退出条件
            if self._should_exit_cycle(cycle_step):
                break
            # 安全退出检查
            if not self.cycle_manager.is_within_iteration_limit():
                break

3. ParallelExecutor + MajorityVoteStrategy

并行执行器 + 多数投票策略：

class MajorityVoteStrategy:
    def run(self) -> str:
        # 1. 所有节点接收相同初始消息
        for node in all_nodes:
            node.clear_input()
            for message in self.initial_messages:
                node.append_input(message.clone())

        # 2. 并行执行
        parallel_executor = ParallelExecutor(self.log_manager, self.nodes)
        parallel_executor.execute_nodes_parallel(node_ids, _execute)

        # 3. 聚合结果
        return self._collect_majority_result()

    def _collect_majority_result(self) -> str:
        # Counter.most_common(1) 获取多数结果
        outputs = [node.output_text for node in self.nodes.values()]
        counter = Counter(outputs)
        return counter.most_common(1)[0][0]

取消信号处理

GraphExecutor 通过 threading.Event 支持优雅取消：

def request_cancel(self, reason: Optional[str] = None) -> None:
    self._cancel_reason = reason
    self._cancel_event.set()

def _raise_if_cancelled(self) -> None:
    if self.is_cancelled():
        raise WorkflowCancelledError(
            self._cancel_reason or "Workflow execution cancelled",
            workflow_id=self.graph.name,
        )

每个节点执行前调用 _raise_if_cancelled()，确保取消信号在节点级别生效。

问题与规避

问题	表现	规避策略
DAG 层内节点失败	屏障等待中某个节点抛出异常	配置节点 retry 策略
Cycle 退出条件永远不满足	迭代到 max_iterations 强制退出	确保边的 condition 设计合理
取消信号丢失	cancel_event 未被传递到子线程	确保所有执行器共享同一 Event
Token 追踪不准确	并发节点的 Token 被重复计算	TokenTracker 使用线程安全的累加器

设计取舍

策略模式 vs 统一执行器

维度	策略模式	统一执行器
代码清晰度	高（每种策略独立）	低（if-elif 分支）
扩展性	高（新增策略类即可）	低（需修改主执行器）
运行开销	低（策略选择一次）	相同
ChatDev 的选择	✅ 使用	❌ 未采用

Kahn vs Tarjan 的选择

• DAG 分层：Kahn（入度法）天然产出层次结构
• 环检测：Tarjan（SCC）输出具体环成员

ChatDev 在拓扑分析中两者结合使用：先用 Tarjan 检测环，无环时用 Kahn 分层。

参考来源

• ChatDev GraphExecutor 源码
• ChatDev RuntimeBuilder 源码