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 = "等待依赖"
@dataclass
class AgentTask:
"""Agent 任务定义"""
agent_name: str
execute_fn: Callable
dependencies: 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_exceptions=True) # 收集第一轮结果 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_exceptions=True) 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_name="schedule_agent",
execute_fn=lambda inputs: asyncio.sleep(1, {"available_dates": ["周六", "周日"]}),
dependencies=["traffic_agent"],
default_inputs={"available_dates": ["周六", "周日"], "estimated_duration": "2小时"},
)
WEATHER_TASK = AgentTask(
agent_name="weather_agent",
execute_fn=lambda inputs: asyncio.sleep(1, {"forecast": "晴朗", "rain_probability": 0.1}),
dependencies=["schedule_agent"],
default_inputs={"forecast": "晴朗", "destination": "北京"},
)
TRAFFIC_TASK = AgentTask(
agent_name="traffic_agent",
execute_fn=lambda inputs: asyncio.sleep(1, {"route": "高速直达", "duration": "2小时"}),
dependencies=["weather_agent"],
default_inputs={"route": "高速直达", "weather_condition": "晴朗"},
)
BUDGET_TASK = AgentTask(
agent_name="budget_agent",
execute_fn=lambda 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 的默认输入值 → 实现多轮调度器 → 收敛条件检测 → 未收敛人工确认流程 → 测量规划完成时间。
