AI Agent 编排与工作流引擎：从 DAG 到状态机的编排模式对比
一、Agent 编排的"混乱协作"：多 Agent 谁来指挥？
单个 AI Agent 的能力已经很强，但复杂任务需要多个 Agent 协作。问题在于：谁来协调多个 Agent 的执行顺序？谁处理 Agent 间的数据传递？谁在 Agent 失败时决定重试还是降级？没有编排引擎，多 Agent 系统就像没有指挥的交响乐团——每个乐手都很强，但合奏一片混乱。
Agent 编排的核心挑战是"选择合适的编排模式"。DAG（有向无环图）适合流水线式的任务，状态机适合需要条件分支和错误恢复的场景，而混合模式则兼顾两者的优势。
二、编排模式对比
graph TB
subgraph DAG模式
A1[任务A] –> A2[任务B]
A1 –> A3[任务C]
A2 –> A4[任务D]
A3 –> A4
end
subgraph 状态机模式
B1[初始化] –> B2{分析结果}
B2 –>|简单| B3[单Agent处理]
B2 –>|复杂| B4[多Agent协作]
B3 –> B5[验证]
B4 –> B5
B5 –>|通过| B6[完成]
B5 –>|失败| B7[重试/降级]
B7 –> B2
end
subgraph 选择依据
C[任务顺序固定<br/>无分支] –> DAG
D[需要条件判断<br/>错误恢复] –> 状态机
E[混合场景] –> 混合模式
end
三、编排引擎实现
3.1 DAG 编排器
from dataclasses import dataclass, field
from typing import Dict, List, Callable, Any
@dataclass
class TaskNode:
name: str
agent: Callable
dependencies: List[str] = field(default_factory=list)
retry_count: int = 0
max_retries: int = 2
class DAGOrchestrator:
"""DAG 编排器：按依赖关系顺序执行任务"""
def __init__(self):
self.nodes: Dict[str, TaskNode] = {}
def add_task(self, name: str, agent: Callable,
dependencies: List[str] = None) -> None:
self.nodes[name] = TaskNode(
name=name,
agent=agent,
dependencies=dependencies or [],
)
def execute(self, initial_input: Any) -> Dict[str, Any]:
"""按拓扑顺序执行所有任务"""
results = {}
executed = set()
# 拓扑排序
order = self._topological_sort()
for task_name in order:
task = self.nodes[task_name]
# 收集依赖的输出
dep_results = {
dep: results[dep] for dep in task.dependencies
}
# 执行任务（带重试）
for attempt in range(task.max_retries + 1):
try:
result = task.agent(
input_data=initial_input,
dependencies=dep_results,
)
results[task_name] = result
executed.add(task_name)
break
except Exception as e:
if attempt == task.max_retries:
raise RuntimeError(
f"任务 {task_name} 执行失败: {e}"
)
return results
def _topological_sort(self) -> List[str]:
"""Kahn 算法拓扑排序"""
in_degree = {n: 0 for n in self.nodes}
for node in self.nodes.values():
for dep in node.dependencies:
in_degree[dep] = in_degree.get(dep, 0)
for node in self.nodes.values():
in_degree[node.name] = len(node.dependencies)
queue = [n for n, d in in_degree.items() if d == 0]
result = []
while queue:
node = queue.pop(0)
result.append(node)
for other in self.nodes.values():
if node in other.dependencies:
in_degree[other.name] -= 1
if in_degree[other.name] == 0:
queue.append(other.name)
return result
3.2 状态机编排器
from enum import Enum
from typing import Optional
class AgentState(Enum):
INIT = "init"
ANALYZE = "analyze"
SIMPLE_TASK = "simple_task"
COMPLEX_TASK = "complex_task"
VERIFY = "verify"
COMPLETE = "complete"
RETRY = "retry"
FAILED = "failed"
class StateMachineOrchestrator:
"""状态机编排器：支持条件分支和错误恢复"""
def __init__(self):
self.transitions = {
AgentState.INIT: [AgentState.ANALYZE],
AgentState.ANALYZE: [AgentState.SIMPLE_TASK, AgentState.COMPLEX_TASK],
AgentState.SIMPLE_TASK: [AgentState.VERIFY],
AgentState.COMPLEX_TASK: [AgentState.VERIFY],
AgentState.VERIFY: [AgentState.COMPLETE, AgentState.RETRY],
AgentState.RETRY: [AgentState.ANALYZE, AgentState.FAILED],
}
self.handlers = {}
self.max_retries = 3
self.retry_count = 0
def register_handler(self, state: AgentState, handler: Callable):
self.handlers[state] = handler
def execute(self, initial_input: Any) -> Any:
"""按状态机逻辑执行"""
state = AgentState.INIT
context = {'input': initial_input, 'results': {}}
while state != AgentState.COMPLETE and state != AgentState.FAILED:
handler = self.handlers.get(state)
if not handler:
raise RuntimeError(f"未注册状态处理器: {state}")
# 执行当前状态的处理逻辑
result = handler(context)
context['results'][state.value] = result
# 根据结果决定下一状态
next_state = self._decide_next(state, result, context)
state = next_state
if state == AgentState.FAILED:
raise RuntimeError("任务执行失败，重试次数耗尽")
return context['results']
def _decide_next(
self, current: AgentState, result: Any, context: dict
) -> AgentState:
"""根据当前状态和结果决定下一状态"""
if current == AgentState.ANALYZE:
complexity = result.get('complexity', 'simple')
return (AgentState.COMPLEX_TASK
if complexity == 'complex'
else AgentState.SIMPLE_TASK)
if current == AgentState.VERIFY:
if result.get('passed', False):
return AgentState.COMPLETE
self.retry_count += 1
if self.retry_count >= self.max_retries:
return AgentState.FAILED
return AgentState.RETRY
if current == AgentState.RETRY:
return AgentState.ANALYZE
# 默认：按转移表顺序取第一个
transitions = self.transitions.get(current, [])
return transitions[0] if transitions else AgentState.FAILED
四、编排模式的 Trade-offs 分析
DAG 的确定性 vs. 灵活性：DAG 编排在编译期确定执行顺序，执行过程可预测、易调试。但无法处理运行时的条件分支和错误恢复。适合"步骤固定"的流水线场景（数据处理管线、内容生成管线）。
状态机的灵活性 vs. 复杂度：状态机支持条件分支、循环和错误恢复，适合"需要决策"的场景（客服对话、故障诊断）。但状态爆炸问题——状态数随条件数指数增长——使得复杂场景的状态机难以维护。
混合模式：在 DAG 的节点内部使用状态机处理错误恢复，在 DAG 层面保持顺序执行的确定性。这种"外层 DAG + 内层状态机"的混合模式，兼顾了可预测性和灵活性。
适用边界：简单顺序任务用 DAG，需要条件判断和重试的任务用状态机，复杂多步骤任务用混合模式。不要为了"灵活性"在简单场景中使用状态机——额外的复杂度没有收益。
五、总结
Agent 编排的核心是"选择合适的编排模式"。DAG 适合步骤固定的流水线，状态机适合需要条件分支和错误恢复的场景，混合模式兼顾两者。选择的关键是理解任务的特性——步骤是否固定、是否需要条件判断、是否需要错误恢复。
落地建议：先用 DAG 实现基本的顺序执行，验证多 Agent 协作的可行性；然后在关键节点加入状态机处理错误恢复；最后根据实际需求决定是否需要更复杂的混合模式。编排模式应该随任务复杂度演进，而非一开始就选择最复杂的方案。
