8.5 实战：为MiniHarness添加编排引擎

本节展示编排引擎的设计原理和关键概念。完整代码见 lab/mini_harness/orchestration/engine.py。

8.5.1 架构需求

编排引擎需要解决四个核心问题：

1. 任务生命周期管理：从注册到执行、完成或失败

2. 依赖关系处理：确保任务按正确的顺序执行

3. 状态机驱动：使用FSM定义工作流的阶段和转移条件

4. 子Agent隔离：为每个子Agent创建独立的执行上下文

8.5.2 数据模型设计

编排引擎使用四个核心枚举和数据类：

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的核心是依赖管理。检查任务是否可执行：

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状态变化：

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基础架构

工作流使用经典的有限状态机模式：

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

设计考虑：转移条件是可选的。无条件转移自动执行，有条件转移基于当前上下文计算。如果条件抛出异常，状态机会记录但继续尝试其他转移。

执行日志

状态机记录所有状态变化及其上下文快照：

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需要与父Agent隔离，但仍可访问继承的值。使用Python的ContextVar实现线程安全的隔离：

_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和状态转移：

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可执行时尝试状态转移。

任务类型处理

编排引擎根据任务类型分发执行：

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.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,
                    self.state_machine.context.copy())
                return {"success": True, "result": result}
        else:
            return {"success": False, "error": "Unknown task type"}
    except Exception as e:
        return {"success": False, "error": str(e)}

上面我们详细介绍了编排引擎的各个核心组件和设计模式。现在通过完整的实际示例，展示如何将这些组件组合在一起，构建一个完整的工作流系统。

8.5.7 使用示例

工作流定义

以下示例展示了如何使用编排引擎定义一个简单的状态工作流，包括起点、研究和完成三个状态及其转换。

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)

任务依赖

以下示例展示了如何定义带有依赖关系的任务，编排引擎会按照依赖顺序自动调度执行。

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保护子Agent免受状态污染

• 可观测性：完整的日志和指标收集

完整实现见 lab/mini_harness/orchestration/engine.py。