chatgpt-mirai-qq-bot执行器实现：工作流节点执行和依赖管理

chatgpt-mirai-qq-bot执行器实现工作流节点执行和依赖管理概述在现代AI聊天机器人系统中工作流引擎是核心组件之一。chatgpt-mirai-qq-bot项目实现了一个高效的工作流执行器能够处理复杂的节点依赖关系、条件分支和循环控制。本文将深入解析其执行器架构、依赖管理机制和实际应用场景。执行器核心架构WorkflowExecutor 类结构执行流程概览依赖管理机制执行图构建执行器通过_build_execution_graph()方法构建执行依赖图def _build_execution_graph(self): self.execution_graph defaultdict(list) for wire in self.workflow.wires: # 验证数据类型匹配 source_output wire.source_block.outputs[wire.source_output] target_input wire.target_block.inputs[wire.target_input] if not target_input.data_type source_output.data_type: raise TypeError(fType mismatch in wire) # 构建执行图边 self.execution_graph[wire.source_block].append(wire.target_block)执行条件检查_can_execute()方法确保节点只有在满足所有前置条件时才会执行def _can_execute(self, block: Block) - bool: # 检查是否已执行 if block.name in self.results: return False # 获取所有前置blocks predecessor_blocks set() for wire in self.workflow.wires: if wire.target_block block: predecessor_blocks.add(wire.source_block) # 验证前置blocks完成 for pred_block in predecessor_blocks: if pred_block.name not in self.results: return False # 验证输入连接 for input_name in block.inputs: input_satisfied False for wire in self.workflow.wires: if (wire.target_block block and wire.target_input input_name and wire.source_block.name in self.results): input_satisfied True break if not input_satisfied and not block.inputs[input_name].nullable: return False return True节点类型支持基础块Block基础块是所有功能块的基类提供标准的输入输出接口class Block: id: str name: str inputs: Dict[str, Input] {} outputs: Dict[str, Output] {} def execute(self, **kwargs) - Dict[str, Any]: return {output: fProcessed {kwargs} for output in self.outputs}条件块ConditionBlock条件块支持基于输入数据的条件分支class ConditionBlock(Block): outputs {condition_result: Output(condition_result, 条件结果, bool, 条件结果)} def execute(self, **kwargs) - Dict[str, Any]: result self.condition_func(kwargs) return {condition_result: result}循环块LoopBlock循环块支持迭代执行class LoopBlock(Block): outputs { should_continue: Output(should_continue, 是否继续, bool, 是否继续), iteration: Output(iteration, 当前迭代数据, dict, 当前迭代数据) } def execute(self, **kwargs) - Dict[str, Any]: should_continue self.condition_func(kwargs) self.iteration_count 1 return { should_continue: should_continue, iteration: {self.iteration_var: self.iteration_count, **kwargs} }输入输出系统输入定义class Input: def __init__(self, name: str, label: str, data_type: type, description: str, nullable: bool False, default: Optional[Any] None): self.name name self.label label self.data_type data_type self.description description self.nullable nullable self.default default def validate(self, value: Any) - bool: if value is None: return self.nullable return isinstance(value, self.data_type)输出定义class Output: def __init__(self, name: str, label: str, data_type: type, description: str): self.name name self.label label self.data_type data_type self.description description def validate(self, value: Any) - bool: return isinstance(value, self.data_type)执行策略并行执行执行器使用线程池实现并行执行async def run(self) - Dict[str, Any]: loop asyncio.get_event_loop() with ThreadPoolExecutor() as executor: # 从入口节点开始执行 entry_blocks [block for block in self.workflow.blocks if not block.inputs] await self._execute_nodes(entry_blocks, executor, loop) return self.results数据收集_gather_inputs()方法负责从前置节点收集输入数据def _gather_inputs(self, block: Block) - Dict[str, Any]: inputs {} input_wire_map {} # 创建输入映射 for wire in self.workflow.wires: if wire.target_block block: input_wire_map[wire.target_input] wire # 收集输入数据 for input_name in block.inputs: if input_name in input_wire_map: wire input_wire_map[input_name] if wire.source_block.name in self.results: inputs[input_name] self.results[wire.source_block.name][wire.source_output] else: raise RuntimeError(fSource block not executed) elif not block.inputs[input_name].nullable: raise RuntimeError(fMissing wire connection) return inputs错误处理机制类型安全检查执行器在构建执行图时进行严格的类型检查# 验证连线的数据类型是否匹配 source_output wire.source_block.outputs[wire.source_output] target_input wire.target_block.inputs[wire.target_input] if not target_input.data_type source_output.data_type: error_msg (fType mismatch in wire: {wire.source_block.name}.{wire.source_output} f({source_output.data_type}) - {wire.target_block.name}.{wire.target_input} f({target_input.data_type})) raise TypeError(error_msg)执行异常处理async def _execute_normal_block(self, block: Block, executor, loop): try: result await future self.results[block.name] result next_blocks self.execution_graph[block] if next_blocks: await self._execute_nodes(next_blocks, executor, loop) except Exception as e: self.logger.error(fBlock {block.name} execution failed: {str(e)}, exc_infoTrue) raise RuntimeError(fBlock {block.name} execution failed: {e})实际应用场景聊天消息处理流程多步骤任务处理# 创建复杂的工作流示例 workflow Workflow( namecomplex_chat_workflow, blocks[input_block, auth_block, process_block, output_block], wires[ Wire(input_block, message, auth_block, input), Wire(auth_block, authenticated, process_block, input), Wire(process_block, response, output_block, message) ] ) executor WorkflowExecutor(workflow) results await executor.run()性能优化策略1. 执行图缓存执行器在初始化时构建执行图避免重复计算依赖关系。2. 并行执行使用线程池并行执行独立节点提高整体吞吐量。3. 懒加载输入只在需要时收集输入数据减少不必要的内存开销。4. 条件短路条件块根据评估结果只执行一个分支避免不必要的计算。测试验证项目提供了完整的测试套件涵盖各种执行场景测试场景描述验证点正常执行流程线性节点依赖所有节点按顺序执行类型不匹配输入输出类型不一致抛出TypeError异常失败块处理节点执行失败正确传播异常空工作流无节点的工作流返回空结果多输出节点节点有多个输出所有输出正确收集总结chatgpt-mirai-qq-bot的工作流执行器实现了以下核心特性智能依赖管理自动构建执行图确保节点按正确顺序执行多类型支持支持普通块、条件块、循环块等多种节点类型类型安全严格的输入输出类型检查避免运行时错误并行执行利用线程池实现高效并行处理健壮的错误处理完善的异常捕获和传播机制灵活的扩展性易于添加新的节点类型和执行策略该执行器为AI聊天机器人提供了强大的工作流处理能力能够处理从简单消息回复到复杂多步骤任务的各类场景是项目架构中的核心组件之一。创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考