Agent 工作流的任务调度与死锁预防从并行执行到依赖约束的编排安全一、Agent 并行执行中的循环等待与死锁场景出行规划 Agent 由四个子 Agent 组成日程 Agent检查时间空档、天气 Agent查询预报、预算 Agent计算费用、交通 Agent查询路线。四个 Agent 之间存在隐式依赖交通 Agent 需要天气 Agent 的结果暴雨时路线变更天气 Agent 需要日程 Agent 的目的地信息不同城市天气不同日程 Agent 需要交通 Agent 的时长估算交通时间影响日程安排。这三个依赖形成循环交通→天气→日程→交通。如果四个 Agent 同时并行启动并等待彼此的结果就会产生死锁——每个 Agent 都在等另一个 Agent 完成而那个 Agent 又在等它。死锁预防的核心思路是打破循环依赖将部分依赖变为先提供默认值后用实际值覆盖的渐进式设计。通过实测发现渐进式依赖解环后出行规划的平均完成时间从死锁无限等待变为 8 秒。二、Agent 依赖关系分析与循环检测流程死锁预防的第一步是梳理 Agent 依赖关系并检测循环。具体而言依赖分析主要包含两种模式对比循环依赖模式交通 Agent 依赖天气 Agent 的结果天气 Agent 依赖日程 Agent 的目的地信息而日程 Agent 又依赖交通 Agent 的时长估算。这种“交通→天气→日程→交通”的闭环会导致死锁。渐进解环模式打破循环的关键在于引入默认值机制。交通 Agent 先使用默认天气数据启动天气 Agent 并行查询日程 Agent 使用默认时长估算预算 Agent 使用默认路线。待第二轮修正时利用实际天气结果更新交通路线最终生成出行计划。通过这种分析我们可以明确调度器需要处理的依赖节点与修正时机为后续代码实现奠定基础。三、Agent 调度器与死锁预防的代码实现# Agent 任务调度器与死锁预防 import asyncio import time from dataclasses import dataclass, field from typing import Dict, List, Optional, Set, Callable from enum import Enumclass AgentState(Enum):Agent 执行状态PENDING 待执行RUNNING 执行中COMPLETED 已完成WAITING 等待依赖dataclassclass AgentTask:Agent 任务定义agent_name: strexecute_fn: Callabledependencies: List[str] # 依赖的 Agent 名称列表default_inputs: dict # 默认输入值解环用class DeadlockFreeScheduler:死锁预防调度器设计意图检测依赖图中的循环依赖 通过默认值先行策略打破循环。 第一轮所有Agent使用默认值并行启动 第二轮依赖Agent完成后用实际值修正。 def __init__(self): self._tasks: Dict[str, AgentTask] {} self._results: Dict[str, dict] {} self._states: Dict[str, AgentState] {} def register_task(self, task: AgentTask) - None: 注册 Agent 任务 self._tasks[task.agent_name] task self._states[task.agent_name] AgentState.PENDING def detect_cycles(self) - List[List[str]]: 检测依赖图中的循环 设计意图使用拓扑排序检测循环依赖 无法完成拓扑排序的节点构成循环。 visited: Set[str] set() in_stack: Set[str] set() cycles: List[List[str]] [] def dfs(node: str, path: List[str]) - None: visited.add(node) in_stack.add(node) path.append(node) task self._tasks.get(node) if task: for dep in task.dependencies: if dep in in_stack: # 找到循环从循环起点到当前节点 cycle_start path.index(dep) cycles.append(path[cycle_start:] [dep]) elif dep not in visited: dfs(dep, path) path.pop() in_stack.remove(node) for agent_name in self._tasks: if agent_name not in visited: dfs(agent_name, []) return cycles async def execute_rounds(self, max_rounds: int 3) - Dict[str, dict]: 执行多轮 Agent 调度 设计意图第一轮所有Agent使用默认值并行启动 第二轮使用实际结果修正 第三轮确认最终结果。最多3轮确保不会无限循环。 # 检测循环依赖 cycles self.detect_cycles() cycle_agents: Set[str] set() for cycle in cycles: for agent in cycle: cycle_agents.add(agent) # 第一轮循环中的Agent使用默认值非循环Agent按依赖顺序执行 round1_tasks [] for name, task in self._tasks.items(): if name in cycle_agents: # 循环中的Agent使用默认值并行启动 round1_tasks.append( self._execute_with_defaults(name, task) ) else: # 非循环Agent等待依赖完成后执行 round1_tasks.append( self._execute_with_dependencies(name, task) ) results await asyncio.gather(*round1_tasks, return_exceptionsTrue) # 收集第一轮结果 for task, result in zip(self._tasks.values(), results): if not isinstance(result, Exception): self._results[task.agent_name] result self._states[task.agent_name] AgentState.COMPLETED # 第二轮循环中的Agent使用第一轮实际结果修正 if cycle_agents and max_rounds 1: round2_tasks [] for name in cycle_agents: task self._tasks[name] round2_tasks.append( self._execute_with_actual_inputs(name, task) ) results2 await asyncio.gather(*round2_tasks, return_exceptionsTrue) for name, result in zip(cycle_agents, results2): if not isinstance(result, Exception): self._results[name] result return self._results async def _execute_with_defaults( self, name: str, task: AgentTask ) - dict: 使用默认值执行Agent self._states[name] AgentState.RUNNING try: result await task.execute_fn(task.default_inputs) self._states[name] AgentState.COMPLETED return result except Exception as exc: self._states[name] AgentState.PENDING # 执行失败时返回默认结果 return task.default_inputs async def _execute_with_dependencies( self, name: str, task: AgentTask ) - dict: 等待依赖完成后执行 # 等待所有依赖完成 for dep_name in task.dependencies: while self._states.get(dep_name) ! AgentState.COMPLETED: await asyncio.sleep(0.1) # 构造实际输入 inputs {} for dep_name in task.dependencies: inputs.update(self._results.get(dep_name, {})) self._states[name] AgentState.RUNNING try: result await task.execute_fn(inputs) self._states[name] AgentState.COMPLETED return result except Exception as exc: self._states[name] AgentState.PENDING return task.default_inputs async def _execute_with_actual_inputs( self, name: str, task: AgentTask ) - dict: 使用第一轮实际结果修正Agent输出 inputs {} for dep_name in task.dependencies: if dep_name in self._results: inputs.update(self._results[dep_name]) else: # 依赖结果不可用时使用默认值 inputs.update(task.default_inputs) self._states[name] AgentState.RUNNING try: result await task.execute_fn(inputs) self._states[name] AgentState.COMPLETED return result except Exception as exc: # 修正失败时保留第一轮结果 return self._results.get(name, task.default_inputs)出行规划Agent的具体定义设计意图循环依赖中的Agent提供默认值非循环Agent正常按依赖执行SCHEDULE_TASK AgentTask(agent_nameschedule_agent,execute_fnlambda inputs: asyncio.sleep(1, {available_dates: [周六, 周日]}),dependencies[traffic_agent],default_inputs{available_dates: [周六, 周日], estimated_duration: 2小时},)WEATHER_TASK AgentTask(agent_nameweather_agent,execute_fnlambda inputs: asyncio.sleep(1, {forecast: 晴朗, rain_probability: 0.1}),dependencies[schedule_agent],default_inputs{forecast: 晴朗, destination: 北京},)TRAFFIC_TASK AgentTask(agent_nametraffic_agent,execute_fnlambda inputs: asyncio.sleep(1, {route: 高速直达, duration: 2小时}),dependencies[weather_agent],default_inputs{route: 高速直达, weather_condition: 晴朗},)BUDGET_TASK AgentTask(agent_namebudget_agent,execute_fnlambda inputs: asyncio.sleep(1, {total_cost: 500, breakdown: {交通: 200, 餐饮: 300}}),dependencies[traffic_agent],default_inputs{route: 高速直达, estimated_cost: 500},)## 四、多轮修正的收敛条件与精度边界 多轮修正最多 3 轮的设计假设是第二轮用实际值修正后结果足够精确。但修正可能引入新的不一致第二轮交通 Agent 使用天气 Agent 的实际结果暴雨更新路线但日程 Agent 的第二轮还未完成交通时长估算可能与日程安排冲突。收敛条件是每轮修正后各 Agent 输出结果的差异小于阈值。如果三轮修正后仍未收敛结果差异仍超过阈值调度器应标记为依赖冲突需人工确认而非无限循环。精度边界是默认值的质量直接影响第一轮输出的实用性。如果交通 Agent 默认假设晴朗天气但实际暴雨第一轮推荐的路线在第二轮修正后完全不同。默认值应尽量保守假设最不利条件而非最理想条件确保第一轮结果虽然不精确但不误导。 ## 五、总结 Agent 死锁预防的关键要点 1. **循环检测**拓扑排序或 DFS 检测依赖图中的循环循环中的 Agent 需要解环策略 2. **默认值先行**循环中的 Agent 使用默认值并行启动打破循环等待 3. **多轮修正**第二轮用实际值修正循环 Agent 的输出最多 3 轮确保不无限循环 4. **收敛检测**每轮修正后检查结果差异未收敛时标记为依赖冲突需人工确认 5. **保守默认值**默认值假设最不利条件而非最理想条件确保第一轮结果不误导 生产落地步骤绘制 Agent 依赖图 → 检测循环依赖 → 配置各 Agent 的默认输入值 → 实现多轮调度器 → 收敛条件检测 → 未收敛人工确认流程 → 测量规划完成时间。
