> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/harness_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/harness_engineering_guide/di-san-bu-fen-xi-tong-ji-cheng-yu-gong-cheng-shi-jian/08_orchestration/8.5_miniharness_orchestration.md). # 8.5 实战：为MiniHarness添加编排引擎 本节展示编排引擎的设计原理和关键概念。完整代码见 `lab/mini_harness/orchestration/engine.py`。 ## 8.5.1 架构需求 编排引擎需要解决四个核心问题： 1. **任务生命周期管理**：从注册到执行、完成或失败 2. **依赖关系处理**：确保任务按正确的顺序执行 3. **状态机驱动**：使用FSM定义工作流的阶段和转移条件 4. **子智能体隔离**：为每个子智能体创建独立的执行上下文 ## 8.5.2 数据模型设计 编排引擎使用四个核心枚举和数据类： ```python class TaskType(Enum): LOCAL_BASH = "local_bash" LOCAL_AGENT = "local_agent" IN_PROCESS_TEAMMATE = "in_process_teammate" WORKFLOW = "workflow" class TaskState(Enum): PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" KILLED = "killed" @dataclass class TaskDefinition: task_id: str task_type: TaskType description: str dependencies: List[str] = field(default_factory=list) timeout_seconds: int = 300 max_retries: int = 1 ``` **设计决策**：使用数据类分离「定义」（配置）和「执行」（状态）。TaskDefinition是只读的任务描述，TaskExecution记录运行时的状态变化。这种分离使得同一个任务可以重新执行而不产生冲突。 ## 8.5.3 任务管理器 ### 依赖检查机制 TaskManager的核心是依赖管理。检查任务是否可执行： ```python def can_execute(self, task_id: str) -> Tuple[bool, Optional[str]]: task = self.get_task(task_id) for dep_id in task.dependencies: dep_exec = self.get_execution(dep_id) if not dep_exec or dep_exec.state != TaskState.COMPLETED: return False, f"Dependency {dep_id} not completed" return True, None ``` 此方法在执行主循环中被频繁调用，确保任务只在依赖完全满足时才运行。 ### 通知队列 完整代码见 `lab/mini_harness/orchestration/engine.py`。 当任务完成或失败时，TaskManager发出通知。这允许工作流响应Task状态变化： ```python def mark_completed(self, task_id: str, result: Dict[str, Any]) -> None: exec_record = self.executions[task_id] exec_record.state = TaskState.COMPLETED exec_record.result = result self._emit_notification(task_id, TaskState.COMPLETED, result) def _emit_notification(self, task_id: str, state: TaskState, result: Optional[Dict[str, Any]]) -> None: next_tasks = self._find_dependent_tasks(task_id) notification = TaskNotification( task_id=task_id, state=state, next_tasks=next_tasks, timestamp=datetime.now(), ) self.notification_queue.append(notification) ``` **关键设计**：通知中包含 `next_tasks` 列表，这是现在可以执行的后续任务。这支持高效的事件驱动调度。 ## 8.5.4 工作流状态机 ### FSM基础架构 工作流使用经典的有限状态机模式： ```python class WorkflowStateMachine: def __init__(self): self.states: Dict[str, StateDefinition] = {} self.transitions: List[TransitionDefinition] = [] self.current_state: Optional[str] = None self.context: Dict[str, Any] = {} def find_next_state(self) -> Optional[str]: for transition in self.transitions: if transition.from_state != self.current_state: continue if transition.condition is None: return transition.to_state try: if transition.condition(self.context): return transition.to_state except Exception as e: continue return None ``` **设计考虑**：转移条件是可选的。无条件转移自动执行，有条件转移基于当前上下文计算。如果条件抛出异常，状态机会记录但继续尝试其他转移。 ### 执行日志 状态机记录所有状态变化及其上下文快照： ```python def _log_state_entry(self, state_id: str) -> None: self.execution_log.append({ "timestamp": datetime.now().isoformat(), "event": "state_entry", "state": state_id, "context_snapshot": dict(self.context), }) ``` 这支持调试和可观测性，完整日志保存在执行结果中。 ## 8.5.5 智能体上下文隔离 ### 上下文变量设计 子智能体需要与父Agent隔离，但仍可访问继承的值。使用Python的ContextVar实现线程安全的隔离： ```python _agent_context: ContextVar = ContextVar('agent_context', default=None) class AgentContext: def __init__(self, agent_id: str, parent_context: Optional[Dict] = None): self.agent_id = agent_id self.variables: Dict[str, Any] = {} self.parent_context = parent_context or {} def get(self, key: str, default: Any = None) -> Any: if key in self.variables: return self.variables[key] return self.parent_context.get(key, default) def __enter__(self): self.token = _agent_context.set(self) return self def __exit__(self, exc_type, exc_val, exc_tb): _agent_context.reset(self.token) ``` **关键设计**：双层查找（本地优先，回退到父）实现变量遮蔽。上下文作为context manager使用，确保异步代码中的正确清理。 ## 8.5.6 编排引擎主循环 ### 执行循环 完整代码见 `lab/mini_harness/orchestration/engine.py`。 编排引擎的核心是主执行循环，它驱动Task和状态转移： ```python import asyncio async def execute_workflow(self, initial_state: str, context: Dict[str, Any], max_iterations: int = 100) -> Dict[str, Any]: self.state_machine.initialize(initial_state, context) iterations = 0 while iterations < max_iterations: if self.state_machine.is_final_state(): break if self.state_machine.is_error_state(): break # 执行可执行的Task for task_id, task in self.task_manager.tasks.items(): can_exec, _ = self.task_manager.can_execute(task_id) if can_exec: result = await self._execute_task(task_id, task) if result["success"]: self.task_manager.mark_completed(task_id, result) else: self.task_manager.mark_failed(task_id, result["error"]) # 处理通知并转移状态 # 注:在 asyncio 取消时正确清理通知队列 try: while (notification := self.task_manager.get_notification()): # 处理通知逻辑 pass except asyncio.CancelledError: # 优雅处理任务取消,清空待处理通知 self.task_manager.clear_notifications() raise next_state = self.state_machine.find_next_state() if next_state: self.state_machine.transition(next_state) iterations += 1 ``` 这种「任务优先」循环模式优先执行所有准备好的Task，然后在没有Task可执行时尝试状态转移。 ### 任务类型处理 编排引擎根据任务类型分发执行： ```python async def _execute_task(self, task_id: str, task: TaskDefinition) -> Dict[str, Any]: try: if task.task_type == TaskType.LOCAL_BASH: return {"success": True, "output": f"Executed {task_id}"} elif task.task_type == TaskType.LOCAL_AGENT: return {"success": True, "output": f"Agent executed {task_id}"} elif task.task_type == TaskType.IN_PROCESS_TEAMMATE: if self.subagent_factory: subagent = self.subagent_factory.create_subagent( f"subagent_{task_id}", task.task_type ) result = await subagent.execute(task_id, self.state_machine.context.copy()) return {"success": True, "result": result} return {"success": False, "error": "SubAgent factory not initialized"} else: return {"success": False, "error": f"Unknown task type: {task.task_type}"} except Exception as e: return {"success": False, "error": str(e)} ``` 上面我们详细介绍了编排引擎的各个核心组件和设计模式。现在通过完整的实际示例，展示如何将这些组件组合在一起，构建一个完整的工作流系统。 ## 8.5.7 使用示例 ### 工作流定义 以下示例展示了如何使用编排引擎定义一个简单的状态工作流，包括起点、研究和完成三个状态及其转换。 ```python engine = OrchestrationEngine() states = [ StateDefinition("start", StateType.INITIAL, "起点"), StateDefinition("research", StateType.NORMAL, "研究"), StateDefinition("complete", StateType.FINAL, "完成"), ] transitions = [ TransitionDefinition("start", "research"), TransitionDefinition("research", "complete"), ] engine.setup_workflow(states, transitions) ``` ### 任务依赖 以下示例展示了如何定义带有依赖关系的任务，编排引擎会按照依赖顺序自动调度执行。 ```python tasks = [ TaskDefinition("task_0", TaskType.LOCAL_BASH, "数据收集"), TaskDefinition("task_1", TaskType.LOCAL_AGENT, "数据分析", dependencies=["task_0"]), TaskDefinition("task_2", TaskType.IN_PROCESS_TEAMMATE, "模型训练", dependencies=["task_1"]), ] engine.register_tasks(tasks) engine.initialize_subagent_factory("main_agent") result = await engine.execute_workflow("start", {"data": "sample"}) ``` ## 8.5.8 本小节小结 编排引擎通过以下设计实现高效的工作流： * **清晰的分层**：TaskManager处理依赖，StateMachine处理阶段 * **事件驱动**：通知队列让调度器对完成做出反应 * **上下文隔离**：ContextVar保护子智能体免受状态污染 * **可观测性**：完整的日志和指标收集 完整实现见 `lab/mini_harness/orchestration/engine.py`。