AI Agent工作流DAG引擎：从任务编排到并行执行的7种生产模式
AI与大数据
线性Agent流水线已死，DAG才是AI工作流的终极答案
你的AI Agent还在用 input → process → output 的线性流水线？当任务需要3个Agent并行调研、2个Agent串行分析、最后1个Agent汇总——线性编排根本搞不定。2026年，AI Agent工作流DAG引擎已经成为生产级系统的标配：DAG（有向无环图）让任务依赖、并行调度、条件路由从硬编码变成了声明式配置。
本文核心收获：
- 理解DAG工作流引擎的核心概念与架构
- 掌握7种生产级DAG编排模式，从任务定义到监控告警
- 完整Python代码实现，可直接用于生产环境
- 5个常见坑及解决方案、10个报错排查手册
- 自建DAG vs LangGraph vs Prefect对比分析
目录
- DAG工作流核心概念
- Pattern 1：任务定义与依赖图构建
- Pattern 2：拓扑排序与并行调度
- Pattern 3：条件路由与分支合并
- Pattern 4：错误恢复与重试策略
- Pattern 5：状态持久化与断点续跑
- Pattern 6：动态DAG与子图嵌套
- Pattern 7：生产监控与告警
- 5个常见坑及解决方案
- 10个常见报错排查
- 进阶优化技巧
- 对比分析：自建DAG vs LangGraph vs Prefect
- 在线工具推荐
- 总结
DAG工作流核心概念
DAG（Directed Acyclic Graph，有向无环图）是AI Agent工作流引擎的数学基础。每个节点代表一个任务（Agent调用、工具执行、数据转换），每条边代表依赖关系。
┌──────────────────────────────────────────────────────────────┐
│ DAG工作流引擎架构 │
├──────────────────────────────────────────────────────────────┤
│ │
│ ┌─────┐ ┌─────┐ ┌─────┐ │
│ │ A │────▶│ B │────▶│ D │ ← 串行依赖链 │
│ └──┬──┘ └─────┘ └─────┘ │
│ │ ▲ │
│ │ ┌─────┐ │ │
│ └────▶│ C │──────────┘ ← B、C并行，D等待两者完成 │
│ └──┬──┘ │
│ │ ┌─────┐ │
│ └────────▶│ E │ ← 条件路由：C→E 或 C→F │
│ └─────┘ │
│ ┌─────┐ │
│ │ F │ ← 条件分支的另一条路径 │
│ └─────┘ │
│ │
│ 核心保证： │
│ 1. 无环 — 不存在 A→B→C→A 的循环依赖 │
│ 2. 拓扑序 — 存在至少一种合法的执行顺序 │
│ 3. 并行度 — 无依赖关系的节点可同时执行 │
└──────────────────────────────────────────────────────────────┘
关键术语
| 术语 | 说明 |
|---|---|
| Node（节点） | 工作流中的执行单元，如LLM调用、工具执行、数据转换 |
| Edge（边） | 节点间的依赖关系，分为普通边和条件边 |
| DAG（有向无环图） | 节点和边构成的图结构，保证无环路 |
| 拓扑排序 | 将DAG节点排列为合法执行序列的算法 |
| 层级 | 拓扑排序后，同一层级的节点可并行执行 |
| Checkpoint | 工作流执行状态的快照，用于断点恢复 |
| 条件路由 | 根据运行时状态动态选择下一个执行节点 |
为什么DAG比线性流水线更强？
| 维度 | 线性流水线 | DAG工作流 |
|---|---|---|
| 并行执行 | ❌ 只能串行 | ✅ 无依赖节点并行 |
| 条件分支 | ⚠️ 硬编码if-else | ✅ 声明式条件边 |
| 错误恢复 | ❌ 从头重试 | ✅ 断点续跑 |
| 可视化 | ⚠️ 难以理解 | ✅ 图结构直观 |
| 扩展性 | ❌ 改动牵一发动全身 | ✅ 局部修改不影响全局 |
Pattern 1：任务定义与依赖图构建
AI Agent工作流DAG引擎的第一步是定义任务节点和依赖关系。我们用Python实现一个类型安全的DAG定义系统。
基础数据模型
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable
import hashlib
import json
class NodeType(Enum):
LLM_CALL = "llm_call"
TOOL_CALL = "tool_call"
TRANSFORM = "transform"
CONDITION = "condition"
PARALLEL_GROUP = "parallel_group"
SUB_WORKFLOW = "sub_workflow"
HUMAN_APPROVAL = "human_approval"
class EdgeType(Enum):
NORMAL = "normal"
CONDITIONAL = "conditional"
@dataclass
class RetryPolicy:
max_retries: int = 3
base_delay: float = 1.0
max_delay: float = 60.0
backoff_factor: float = 2.0
retryable_exceptions: list[type[Exception]] = field(
default_factory=lambda: [Exception]
)
@dataclass
class NodeDefinition:
node_id: str
node_type: NodeType
handler: Callable[..., Any] | None = None
timeout_seconds: float = 300.0
retry_policy: RetryPolicy = field(default_factory=RetryPolicy)
metadata: dict[str, Any] = field(default_factory=dict)
def __hash__(self):
return hash(self.node_id)
def __eq__(self, other):
if isinstance(other, NodeDefinition):
return self.node_id == other.node_id
return False
@dataclass
class EdgeDefinition:
source_id: str
target_id: str
edge_type: EdgeType = EdgeType.NORMAL
condition: Callable[..., bool] | None = None
condition_name: str = ""
def __hash__(self):
return hash((self.source_id, self.target_id, self.condition_name))
DAG构建器
class DAGBuilder:
def __init__(self, workflow_id: str, name: str = ""):
self.workflow_id = workflow_id
self.name = name
self._nodes: dict[str, NodeDefinition] = {}
self._edges: list[EdgeDefinition] = []
self._entry_node: str | None = None
def add_node(self, node: NodeDefinition) -> DAGBuilder:
if node.node_id in self._nodes:
raise ValueError(f"Node '{node.node_id}' already exists")
self._nodes[node.node_id] = node
return self
def add_edge(
self,
source_id: str,
target_id: str,
edge_type: EdgeType = EdgeType.NORMAL,
condition: Callable[..., bool] | None = None,
condition_name: str = "",
) -> DAGBuilder:
if source_id not in self._nodes:
raise ValueError(f"Source node '{source_id}' not found")
if target_id not in self._nodes:
raise ValueError(f"Target node '{target_id}' not found")
self._edges.append(
EdgeDefinition(
source_id=source_id,
target_id=target_id,
edge_type=edge_type,
condition=condition,
condition_name=condition_name,
)
)
return self
def set_entry(self, node_id: str) -> DAGBuilder:
if node_id not in self._nodes:
raise ValueError(f"Entry node '{node_id}' not found")
self._entry_node = node_id
return self
def build(self) -> DAGDefinition:
if not self._entry_node:
raise ValueError("Entry node not set")
dag = DAGDefinition(
workflow_id=self.workflow_id,
name=self.name,
nodes=dict(self._nodes),
edges=list(self._edges),
entry_node=self._entry_node,
)
dag.validate()
return dag
@dataclass
class DAGDefinition:
workflow_id: str
name: str
nodes: dict[str, NodeDefinition]
edges: list[EdgeDefinition]
entry_node: str
def validate(self):
self._check_cycle()
self._check_reachability()
def _check_cycle(self):
adjacency: dict[str, set[str]] = {
nid: set() for nid in self.nodes
}
for edge in self.edges:
adjacency[edge.source_id].add(edge.target_id)
visited: set[str] = set()
recursion_stack: set[str] = set()
def dfs(node_id: str) -> bool:
visited.add(node_id)
recursion_stack.add(node_id)
for neighbor in adjacency.get(node_id, set()):
if neighbor not in visited:
if dfs(neighbor):
return True
elif neighbor in recursion_stack:
return True
recursion_stack.remove(node_id)
return False
for node_id in self.nodes:
if node_id not in visited:
if dfs(node_id):
raise ValueError(
f"Cycle detected in DAG '{self.workflow_id}'"
)
def _check_reachability(self):
reachable: set[str] = set()
stack = [self.entry_node]
while stack:
current = stack.pop()
if current in reachable:
continue
reachable.add(current)
for edge in self.edges:
if edge.source_id == current:
stack.append(edge.target_id)
unreachable = set(self.nodes.keys()) - reachable
if unreachable:
raise ValueError(
f"Unreachable nodes detected: {unreachable}"
)
def fingerprint(self) -> str:
data = {
"nodes": sorted(self.nodes.keys()),
"edges": [
{"s": e.source_id, "t": e.target_id, "c": e.condition_name}
for e in sorted(
self.edges,
key=lambda e: (e.source_id, e.target_id),
)
],
}
raw = json.dumps(data, sort_keys=True)
return hashlib.sha256(raw.encode()).hexdigest()[:12]
构建示例：内容生成工作流
def fetch_topic(state: dict) -> dict:
return {"topic": state.get("input", "AI技术趋势")}
def research(state: dict) -> dict:
topic = state["topic"]
return {"research_data": f"关于{topic}的深度研究数据..."}
def analyze(state: dict) -> dict:
data = state["research_data"]
return {"analysis": f"基于{data}的分析结论..."}
def write_draft(state: dict) -> dict:
analysis = state["analysis"]
return {"draft": f"基于分析{analysis}的初稿内容..."}
def review(state: dict) -> dict:
draft = state["draft"]
return {"review_result": "approved", "final_content": draft}
def needs_revision(state: dict) -> bool:
return state.get("review_result") == "needs_revision"
def is_approved(state: dict) -> bool:
return state.get("review_result") == "approved"
builder = (
DAGBuilder("content-gen-v1", "内容生成工作流")
.add_node(NodeDefinition("fetch", NodeType.TRANSFORM, handler=fetch_topic))
.add_node(NodeDefinition("research", NodeType.LLM_CALL, handler=research))
.add_node(NodeDefinition("analyze", NodeType.LLM_CALL, handler=analyze))
.add_node(NodeDefinition("write", NodeType.LLM_CALL, handler=write_draft))
.add_node(NodeDefinition("review", NodeType.LLM_CALL, handler=review))
.add_edge("fetch", "research")
.add_edge("research", "analyze")
.add_edge("analyze", "write")
.add_edge("write", "review")
.add_edge(
"review", "write",
EdgeType.CONDITIONAL,
condition=needs_revision,
condition_name="needs_revision",
)
.set_entry("fetch")
)
dag = builder.build()
print(f"DAG fingerprint: {dag.fingerprint()}")
Pattern 2：拓扑排序与并行调度
DAG引擎的核心调度能力来自拓扑排序。排序后，同一层级的节点没有相互依赖，可以并行执行——这是AI工作流引擎性能的关键。
拓扑排序与层级划分
from collections import deque
class TopologicalSorter:
def __init__(self, dag: DAGDefinition):
self.dag = dag
self._adjacency: dict[str, set[str]] = {nid: set() for nid in dag.nodes}
self._in_degree: dict[str, int] = {nid: 0 for nid in dag.nodes}
for edge in dag.edges:
if edge.edge_type == EdgeType.NORMAL:
self._adjacency[edge.source_id].add(edge.target_id)
self._in_degree[edge.target_id] += 1
def sort(self) -> list[str]:
in_degree = dict(self._in_degree)
queue = deque(
nid for nid, deg in in_degree.items() if deg == 0
)
result = []
while queue:
node_id = queue.popleft()
result.append(node_id)
for neighbor in self._adjacency[node_id]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
queue.append(neighbor)
if len(result) != len(self.dag.nodes):
raise ValueError("DAG contains a cycle (should have been caught in validation)")
return result
def compute_levels(self) -> dict[str, int]:
levels: dict[str, int] = {}
order = self.sort()
for node_id in order:
max_parent_level = -1
for edge in self.dag.edges:
if edge.target_id == node_id and edge.edge_type == EdgeType.NORMAL:
parent_level = levels.get(edge.source_id, 0)
max_parent_level = max(max_parent_level, parent_level)
levels[node_id] = max_parent_level + 1
return levels
def get_parallel_groups(self) -> list[list[str]]:
levels = self.compute_levels()
max_level = max(levels.values()) if levels else 0
groups: list[list[str]] = []
for level in range(max_level + 1):
group = [nid for nid, lvl in levels.items() if lvl == level]
if group:
groups.append(group)
return groups
并行调度器
import asyncio
import time
from dataclasses import dataclass, field
@dataclass
class NodeResult:
node_id: str
status: str
output: dict = field(default_factory=dict)
error: str | None = None
start_time: float = 0.0
end_time: float = 0.0
retry_count: int = 0
@dataclass
class WorkflowResult:
workflow_id: str
execution_id: str
status: str
state: dict = field(default_factory=dict)
node_results: dict[str, NodeResult] = field(default_factory=dict)
total_time: float = 0.0
class DAGScheduler:
def __init__(self, dag: DAGDefinition, max_concurrency: int = 10):
self.dag = dag
self.max_concurrency = max_concurrency
self._sorter = TopologicalSorter(dag)
self._semaphore = asyncio.Semaphore(max_concurrency)
async def execute(self, initial_state: dict | None = None) -> WorkflowResult:
execution_id = f"exec-{int(time.time() * 1000)}"
state = dict(initial_state or {})
node_results: dict[str, NodeResult] = {}
start_time = time.time()
parallel_groups = self._sorter.get_parallel_groups()
for group in parallel_groups:
tasks = [
self._execute_node(node_id, state, node_results)
for node_id in group
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
node_id = group[i]
if isinstance(result, Exception):
node_results[node_id] = NodeResult(
node_id=node_id,
status="failed",
error=str(result),
)
return WorkflowResult(
workflow_id=self.dag.workflow_id,
execution_id=execution_id,
status="failed",
state=state,
node_results=node_results,
total_time=time.time() - start_time,
)
node_results[node_id] = result
state.update(result.output)
return WorkflowResult(
workflow_id=self.dag.workflow_id,
execution_id=execution_id,
status="completed",
state=state,
node_results=node_results,
total_time=time.time() - start_time,
)
async def _execute_node(
self,
node_id: str,
state: dict,
node_results: dict[str, NodeResult],
) -> NodeResult:
node = self.dag.nodes[node_id]
start_time = time.time()
async with self._semaphore:
try:
if asyncio.iscoroutinefunction(node.handler):
output = await node.handler(state)
else:
output = await asyncio.to_thread(node.handler, state)
if not isinstance(output, dict):
output = {"result": output}
return NodeResult(
node_id=node_id,
status="completed",
output=output,
start_time=start_time,
end_time=time.time(),
)
except Exception as e:
return NodeResult(
node_id=node_id,
status="failed",
error=str(e),
start_time=start_time,
end_time=time.time(),
)
执行示例
async def parallel_research_a(state: dict) -> dict:
await asyncio.sleep(0.1)
return {"research_a": "技术趋势研究数据"}
async def parallel_research_b(state: dict) -> dict:
await asyncio.sleep(0.1)
return {"research_b": "市场分析研究数据"}
async def merge_research(state: dict) -> dict:
return {
"merged": f"{state.get('research_a', '')} + {state.get('research_b', '')}"
}
builder = (
DAGBuilder("parallel-research", "并行研究工作流")
.add_node(NodeDefinition("start", NodeType.TRANSFORM, handler=lambda s: s))
.add_node(NodeDefinition("research_a", NodeType.LLM_CALL, handler=parallel_research_a))
.add_node(NodeDefinition("research_b", NodeType.LLM_CALL, handler=parallel_research_b))
.add_node(NodeDefinition("merge", NodeType.TRANSFORM, handler=merge_research))
.add_edge("start", "research_a")
.add_edge("start", "research_b")
.add_edge("research_a", "merge")
.add_edge("research_b", "merge")
.set_entry("start")
)
dag = builder.build()
scheduler = DAGScheduler(dag)
result = await scheduler.execute({"input": "AI Agent工作流DAG引擎"})
print(f"Status: {result.status}")
print(f"Total time: {result.total_time:.3f}s")
print(f"Parallel groups: {TopologicalSorter(dag).get_parallel_groups()}")
Pattern 3：条件路由与分支合并
真实AI工作流不会只有一条路径。根据Agent输出、数据质量、用户偏好等条件动态选择执行路径，是DAG编排的核心能力。
条件路由实现
class ConditionalRouter:
def __init__(self, dag: DAGDefinition):
self.dag = dag
self._conditional_edges: dict[str, list[EdgeDefinition]] = {}
for edge in dag.edges:
if edge.edge_type == EdgeType.CONDITIONAL:
self._conditional_edges.setdefault(edge.source_id, []).append(edge)
def resolve_next_nodes(
self, node_id: str, state: dict
) -> list[str]:
next_nodes: list[str] = []
for edge in self.dag.edges:
if edge.source_id != node_id:
continue
if edge.edge_type == EdgeType.NORMAL:
next_nodes.append(edge.target_id)
elif edge.edge_type == EdgeType.CONDITIONAL:
if edge.condition and edge.condition(state):
next_nodes.append(edge.target_id)
return next_nodes
def get_all_branches(self) -> dict[str, list[str]]:
branches: dict[str, list[str]] = {}
for source_id, edges in self._conditional_edges.items():
branches[source_id] = [
f"{e.condition_name} → {e.target_id}" for e in edges
]
return branches
带条件路由的调度器
class ConditionalDAGScheduler(DAGScheduler):
def __init__(self, dag: DAGDefinition, max_concurrency: int = 10):
super().__init__(dag, max_concurrency)
self._router = ConditionalRouter(dag)
async def execute(self, initial_state: dict | None = None) -> WorkflowResult:
execution_id = f"exec-{int(time.time() * 1000)}"
state = dict(initial_state or {})
node_results: dict[str, NodeResult] = {}
start_time = time.time()
completed: set[str] = set()
pending: set[str] = {self.dag.entry_node}
while pending:
ready: list[str] = []
for node_id in list(pending):
deps = self._get_dependencies(node_id)
if deps.issubset(completed):
ready.append(node_id)
if not ready:
raise RuntimeError(
f"Deadlock detected. Pending: {pending}, Completed: {completed}"
)
tasks = [
self._execute_node(node_id, state, node_results)
for node_id in ready
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
node_id = ready[i]
if isinstance(result, Exception):
node_results[node_id] = NodeResult(
node_id=node_id, status="failed", error=str(result)
)
return WorkflowResult(
workflow_id=self.dag.workflow_id,
execution_id=execution_id,
status="failed",
state=state,
node_results=node_results,
total_time=time.time() - start_time,
)
node_results[node_id] = result
state.update(result.output)
completed.add(node_id)
pending.discard(node_id)
next_nodes = self._router.resolve_next_nodes(node_id, state)
for next_id in next_nodes:
if next_id not in completed:
pending.add(next_id)
return WorkflowResult(
workflow_id=self.dag.workflow_id,
execution_id=execution_id,
status="completed",
state=state,
node_results=node_results,
total_time=time.time() - start_time,
)
def _get_dependencies(self, node_id: str) -> set[str]:
deps: set[str] = set()
for edge in self.dag.edges:
if edge.target_id == node_id:
deps.add(edge.source_id)
return deps
条件路由示例：智能客服分流
def classify_intent(state: dict) -> dict:
user_input = state.get("user_input", "")
if "退款" in user_input or "退货" in user_input:
return {"intent": "refund", "confidence": 0.95}
elif "技术" in user_input or "bug" in user_input.lower():
return {"intent": "technical", "confidence": 0.90}
else:
return {"intent": "general", "confidence": 0.80}
def handle_refund(state: dict) -> dict:
return {"response": "退款流程已启动，预计3-5个工作日到账"}
def handle_technical(state: dict) -> dict:
return {"response": "技术支持团队已收到您的问题，将在2小时内响应"}
def handle_general(state: dict) -> dict:
return {"response": "感谢您的咨询，客服将尽快为您解答"}
def is_refund(state: dict) -> bool:
return state.get("intent") == "refund"
def is_technical(state: dict) -> bool:
return state.get("intent") == "technical"
def is_general(state: dict) -> bool:
return state.get("intent") == "general"
builder = (
DAGBuilder("customer-service", "智能客服分流")
.add_node(NodeDefinition("classify", NodeType.LLM_CALL, handler=classify_intent))
.add_node(NodeDefinition("refund_handler", NodeType.LLM_CALL, handler=handle_refund))
.add_node(NodeDefinition("tech_handler", NodeType.LLM_CALL, handler=handle_technical))
.add_node(NodeDefinition("general_handler", NodeType.LLM_CALL, handler=handle_general))
.add_node(NodeDefinition("respond", NodeType.TRANSFORM, handler=lambda s: {"final": s.get("response", "")}))
.add_edge("classify", "refund_handler", EdgeType.CONDITIONAL, is_refund, "is_refund")
.add_edge("classify", "tech_handler", EdgeType.CONDITIONAL, is_technical, "is_technical")
.add_edge("classify", "general_handler", EdgeType.CONDITIONAL, is_general, "is_general")
.add_edge("refund_handler", "respond")
.add_edge("tech_handler", "respond")
.add_edge("general_handler", "respond")
.set_entry("classify")
)
dag = builder.build()
scheduler = ConditionalDAGScheduler(dag)
result = await scheduler.execute({"user_input": "我要退款，商品有质量问题"})
print(f"Response: {result.state.get('final', '')}")
Pattern 4：错误恢复与重试策略
AI工作流中LLM调用、API请求随时可能失败。没有重试和错误恢复的DAG引擎在生产环境中是不可接受的。
重试策略实现
import random
import logging
logger = logging.getLogger(__name__)
class RetryExecutor:
def __init__(self, retry_policy: RetryPolicy):
self.policy = retry_policy
async def execute_with_retry(
self,
handler: Callable[..., Any],
state: dict,
node_id: str,
) -> NodeResult:
last_error: Exception | None = None
retry_count = 0
for attempt in range(self.policy.max_retries + 1):
try:
if asyncio.iscoroutinefunction(handler):
output = await handler(state)
else:
output = await asyncio.to_thread(handler, state)
if not isinstance(output, dict):
output = {"result": output}
return NodeResult(
node_id=node_id,
status="completed",
output=output,
retry_count=retry_count,
)
except tuple(self.policy.retryable_exceptions) as e:
last_error = e
retry_count += 1
if attempt < self.policy.max_retries:
delay = min(
self.policy.base_delay
* (self.policy.backoff_factor ** attempt),
self.policy.max_delay,
)
jitter = random.uniform(0, delay * 0.1)
logger.warning(
f"Node '{node_id}' failed (attempt {attempt + 1}/"
f"{self.policy.max_retries + 1}), "
f"retrying in {delay + jitter:.2f}s: {e}"
)
await asyncio.sleep(delay + jitter)
except Exception as e:
last_error = e
break
return NodeResult(
node_id=node_id,
status="failed",
error=str(last_error),
retry_count=retry_count,
)
带重试和降级的调度器
class ResilientDAGScheduler(ConditionalDAGScheduler):
def __init__(
self,
dag: DAGDefinition,
max_concurrency: int = 10,
fallback_handlers: dict[str, Callable] | None = None,
):
super().__init__(dag, max_concurrency)
self._fallback_handlers = fallback_handlers or {}
async def _execute_node(
self,
node_id: str,
state: dict,
node_results: dict[str, NodeResult],
) -> NodeResult:
node = self.dag.nodes[node_id]
retry_executor = RetryExecutor(node.retry_policy)
result = await retry_executor.execute_with_retry(
node.handler, state, node_id
)
if result.status == "failed" and node_id in self._fallback_handlers:
logger.info(f"Node '{node_id}' failed, executing fallback handler")
try:
fallback = self._fallback_handlers[node_id]
if asyncio.iscoroutinefunction(fallback):
output = await fallback(state)
else:
output = await asyncio.to_thread(fallback, state)
if not isinstance(output, dict):
output = {"result": output}
return NodeResult(
node_id=node_id,
status="completed_with_fallback",
output=output,
retry_count=result.retry_count,
)
except Exception as fallback_error:
result.error = f"Primary: {result.error}; Fallback: {fallback_error}"
return result
使用示例
async def call_llm_with_retry(state: dict) -> dict:
if random.random() < 0.5:
raise ConnectionError("LLM API timeout")
return {"llm_response": "分析结果..."}
def fallback_llm(state: dict) -> dict:
return {"llm_response": "降级：使用缓存结果"}
builder = (
DAGBuilder("resilient-workflow", "容错工作流")
.add_node(
NodeDefinition(
"llm_call",
NodeType.LLM_CALL,
handler=call_llm_with_retry,
retry_policy=RetryPolicy(
max_retries=3,
base_delay=0.5,
retryable_exceptions=[ConnectionError, TimeoutError],
),
)
)
.set_entry("llm_call")
)
dag = builder.build()
scheduler = ResilientDAGScheduler(
dag,
fallback_handlers={"llm_call": fallback_llm},
)
result = await scheduler.execute({"input": "test"})
print(f"Status: {result.status}")
Pattern 5：状态持久化与断点续跑
长时运行的AI工作流（如多轮Agent协作、大规模数据处理）必须支持状态持久化。执行到一半崩溃后能从断点恢复，而不是从头开始。
检查点管理器
import json
import os
from pathlib import Path
from datetime import datetime
class CheckpointManager:
def __init__(self, storage_dir: str = ".checkpoints"):
self._storage = Path(storage_dir)
self._storage.mkdir(parents=True, exist_ok=True)
def save(
self,
workflow_id: str,
execution_id: str,
state: dict,
completed_nodes: set[str],
pending_nodes: set[str],
node_results: dict[str, NodeResult],
) -> str:
checkpoint_id = f"cp-{int(time.time() * 1000)}"
checkpoint_data = {
"checkpoint_id": checkpoint_id,
"workflow_id": workflow_id,
"execution_id": execution_id,
"state": state,
"completed_nodes": list(completed_nodes),
"pending_nodes": list(pending_nodes),
"node_results": {
nid: {
"node_id": r.node_id,
"status": r.status,
"output": r.output,
"error": r.error,
"retry_count": r.retry_count,
}
for nid, r in node_results.items()
},
"saved_at": datetime.now().isoformat(),
}
filepath = self._storage / f"{workflow_id}_{execution_id}.json"
with open(filepath, "w", encoding="utf-8") as f:
json.dump(checkpoint_data, f, ensure_ascii=False, indent=2)
return checkpoint_id
def load(
self, workflow_id: str, execution_id: str
) -> dict | None:
filepath = self._storage / f"{workflow_id}_{execution_id}.json"
if not filepath.exists():
return None
with open(filepath, "r", encoding="utf-8") as f:
return json.load(f)
def list_checkpoints(self, workflow_id: str) -> list[dict]:
checkpoints = []
for fp in self._storage.glob(f"{workflow_id}_*.json"):
with open(fp, "r", encoding="utf-8") as f:
data = json.load(f)
checkpoints.append({
"execution_id": data["execution_id"],
"saved_at": data["saved_at"],
"completed": len(data["completed_nodes"]),
"pending": len(data["pending_nodes"]),
})
return sorted(checkpoints, key=lambda x: x["saved_at"], reverse=True)
def cleanup(self, workflow_id: str, keep_last: int = 5):
checkpoints = self.list_checkpoints(workflow_id)
for cp in checkpoints[keep_last:]:
filepath = self._storage / f"{workflow_id}_{cp['execution_id']}.json"
filepath.unlink(missing_ok=True)
支持断点续跑的调度器
class PersistentDAGScheduler(ResilientDAGScheduler):
def __init__(
self,
dag: DAGDefinition,
checkpoint_manager: CheckpointManager,
max_concurrency: int = 10,
checkpoint_interval: int = 1,
fallback_handlers: dict[str, Callable] | None = None,
):
super().__init__(dag, max_concurrency, fallback_handlers)
self._checkpoint_mgr = checkpoint_manager
self._checkpoint_interval = checkpoint_interval
async def execute(
self,
initial_state: dict | None = None,
execution_id: str | None = None,
) -> WorkflowResult:
if execution_id:
return await self._resume(execution_id, initial_state)
return await self._run_from_start(initial_state)
async def _run_from_start(
self, initial_state: dict | None = None
) -> WorkflowResult:
execution_id = f"exec-{int(time.time() * 1000)}"
state = dict(initial_state or {})
node_results: dict[str, NodeResult] = {}
completed: set[str] = set()
pending: set[str] = {self.dag.entry_node}
start_time = time.time()
steps_since_checkpoint = 0
while pending:
ready = [
nid for nid in pending
if self._get_dependencies(nid).issubset(completed)
]
if not ready:
raise RuntimeError("Deadlock in DAG execution")
tasks = [
self._execute_node(nid, state, node_results)
for nid in ready
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
node_id = ready[i]
if isinstance(result, Exception):
node_results[node_id] = NodeResult(
node_id=node_id, status="failed", error=str(result)
)
self._checkpoint_mgr.save(
self.dag.workflow_id, execution_id,
state, completed, pending, node_results,
)
return WorkflowResult(
workflow_id=self.dag.workflow_id,
execution_id=execution_id,
status="failed",
state=state,
node_results=node_results,
total_time=time.time() - start_time,
)
node_results[node_id] = result
state.update(result.output)
completed.add(node_id)
pending.discard(node_id)
next_nodes = self._router.resolve_next_nodes(node_id, state)
for next_id in next_nodes:
if next_id not in completed:
pending.add(next_id)
steps_since_checkpoint += 1
if steps_since_checkpoint >= self._checkpoint_interval:
self._checkpoint_mgr.save(
self.dag.workflow_id, execution_id,
state, completed, pending, node_results,
)
steps_since_checkpoint = 0
return WorkflowResult(
workflow_id=self.dag.workflow_id,
execution_id=execution_id,
status="completed",
state=state,
node_results=node_results,
total_time=time.time() - start_time,
)
async def _resume(
self,
execution_id: str,
initial_state: dict | None = None,
) -> WorkflowResult:
checkpoint = self._checkpoint_mgr.load(
self.dag.workflow_id, execution_id
)
if not checkpoint:
raise ValueError(
f"No checkpoint found for {self.dag.workflow_id}/{execution_id}"
)
state = checkpoint["state"]
if initial_state:
state.update(initial_state)
completed = set(checkpoint["completed_nodes"])
pending = set(checkpoint["pending_nodes"])
node_results = {
nid: NodeResult(
node_id=r["node_id"],
status=r["status"],
output=r["output"],
error=r.get("error"),
retry_count=r.get("retry_count", 0),
)
for nid, r in checkpoint["node_results"].items()
}
failed_nodes = {
nid for nid, r in node_results.items() if r.status == "failed"
}
pending.update(failed_nodes)
start_time = time.time()
steps_since_checkpoint = 0
while pending:
ready = [
nid for nid in pending
if self._get_dependencies(nid).issubset(completed)
]
if not ready:
break
tasks = [
self._execute_node(nid, state, node_results)
for nid in ready
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
node_id = ready[i]
if isinstance(result, Exception):
node_results[node_id] = NodeResult(
node_id=node_id, status="failed", error=str(result)
)
self._checkpoint_mgr.save(
self.dag.workflow_id, execution_id,
state, completed, pending, node_results,
)
return WorkflowResult(
workflow_id=self.dag.workflow_id,
execution_id=execution_id,
status="failed",
state=state,
node_results=node_results,
total_time=time.time() - start_time,
)
node_results[node_id] = result
state.update(result.output)
completed.add(node_id)
pending.discard(node_id)
next_nodes = self._router.resolve_next_nodes(node_id, state)
for next_id in next_nodes:
if next_id not in completed:
pending.add(next_id)
steps_since_checkpoint += 1
if steps_since_checkpoint >= self._checkpoint_interval:
self._checkpoint_mgr.save(
self.dag.workflow_id, execution_id,
state, completed, pending, node_results,
)
steps_since_checkpoint = 0
return WorkflowResult(
workflow_id=self.dag.workflow_id,
execution_id=execution_id,
status="completed",
state=state,
node_results=node_results,
total_time=time.time() - start_time,
)
Pattern 6：动态DAG与子图嵌套
生产环境中，DAG不是静态的。根据运行时数据动态生成子任务、嵌套子工作流是高级编排的关键能力。
动态DAG生成
class DynamicDAGGenerator:
def __init__(self, base_dag: DAGDefinition):
self.base_dag = base_dag
def generate_dynamic_nodes(
self,
state: dict,
dynamic_node_factory: Callable[[dict], list[NodeDefinition]],
dependency_resolver: Callable[[list[NodeDefinition], dict], list[EdgeDefinition]],
) -> tuple[list[NodeDefinition], list[EdgeDefinition]]:
new_nodes = dynamic_node_factory(state)
new_edges = dependency_resolver(new_nodes, state)
return new_nodes, new_edges
def merge_into_base(
self,
new_nodes: list[NodeDefinition],
new_edges: list[EdgeDefinition],
attach_after: str,
) -> DAGDefinition:
builder = DAGBuilder(
f"{self.base_dag.workflow_id}-dynamic",
f"{self.base_dag.name} (dynamic)",
)
for node in self.base_dag.nodes.values():
builder.add_node(node)
for edge in self.base_dag.edges:
builder.add_edge(
edge.source_id, edge.target_id,
edge.edge_type, edge.condition, edge.condition_name,
)
for node in new_nodes:
builder.add_node(node)
for edge in new_edges:
builder.add_edge(
edge.source_id, edge.target_id,
edge.edge_type, edge.condition, edge.condition_name,
)
builder.set_entry(self.base_dag.entry_node)
return builder.build()
子图嵌套
class SubWorkflowNode:
def __init__(
self,
sub_dag: DAGDefinition,
scheduler_class: type = ConditionalDAGScheduler,
max_concurrency: int = 5,
):
self.sub_dag = sub_dag
self._scheduler_class = scheduler_class
self._max_concurrency = max_concurrency
async def execute(self, state: dict) -> dict:
scheduler = self._scheduler_class(
self.sub_dag, max_concurrency=self._max_concurrency
)
result = await scheduler.execute(state)
if result.status != "completed":
raise RuntimeError(
f"Sub-workflow '{self.sub_dag.workflow_id}' failed: "
f"{[r.error for r in result.node_results.values() if r.error]}"
)
return result.state
def create_sub_workflow_node(
node_id: str,
sub_dag: DAGDefinition,
max_concurrency: int = 5,
) -> NodeDefinition:
sub_executor = SubWorkflowNode(sub_dag, max_concurrency=max_concurrency)
return NodeDefinition(
node_id=node_id,
node_type=NodeType.SUB_WORKFLOW,
handler=sub_executor.execute,
metadata={"sub_workflow_id": sub_dag.workflow_id},
)
动态DAG示例：多源数据采集
def create_data_source_nodes(state: dict) -> list[NodeDefinition]:
sources = state.get("data_sources", ["api", "database", "file"])
nodes = []
for source in sources:
async def fetch_data(s: dict, src=source) -> dict:
await asyncio.sleep(0.1)
return {f"{src}_data": f"来自{src}的数据"}
nodes.append(
NodeDefinition(
f"fetch_{source}",
NodeType.TOOL_CALL,
handler=fetch_data,
)
)
return nodes
def resolve_dynamic_edges(
new_nodes: list[NodeDefinition], state: dict
) -> list[EdgeDefinition]:
edges = []
for node in new_nodes:
edges.append(EdgeDefinition(source_id="start", target_id=node.node_id))
edges.append(EdgeDefinition(source_id=node.node_id, target_id="aggregate"))
return edges
sub_builder = (
DAGBuilder("data-collection", "数据采集子图")
.add_node(NodeDefinition("start", NodeType.TRANSFORM, handler=lambda s: s))
.add_node(NodeDefinition("aggregate", NodeType.TRANSFORM, handler=lambda s: {"aggregated": "all data merged"}))
.set_entry("start")
)
sub_dag = sub_builder.build()
builder = (
DAGBuilder("main-workflow", "主工作流")
.add_node(NodeDefinition("plan", NodeType.LLM_CALL, handler=lambda s: {**s, "data_sources": ["api", "database", "file"]}))
.add_node(create_sub_workflow_node("collect", sub_dag))
.add_node(NodeDefinition("report", NodeType.LLM_CALL, handler=lambda s: {"report": "最终报告"}))
.add_edge("plan", "collect")
.add_edge("collect", "report")
.set_entry("plan")
)
main_dag = builder.build()
scheduler = ConditionalDAGScheduler(main_dag)
result = await scheduler.execute({"input": "生成数据采集报告"})
Pattern 7：生产监控与告警
AI Agent工作流DAG引擎上线后，监控是运维的命脉。你需要知道每个节点的执行时间、成功率、错误分布。
指标收集器
from dataclasses import dataclass, field
from collections import defaultdict
import statistics
@dataclass
class NodeMetrics:
node_id: str
total_executions: int = 0
success_count: int = 0
failure_count: int = 0
fallback_count: int = 0
total_retry_count: int = 0
execution_times: list[float] = field(default_factory=list)
@property
def success_rate(self) -> float:
if self.total_executions == 0:
return 0.0
return self.success_count / self.total_executions
@property
def avg_execution_time(self) -> float:
if not self.execution_times:
return 0.0
return statistics.mean(self.execution_times)
@property
def p95_execution_time(self) -> float:
if len(self.execution_times) < 2:
return self.avg_execution_time
sorted_times = sorted(self.execution_times)
idx = int(len(sorted_times) * 0.95)
return sorted_times[min(idx, len(sorted_times) - 1)]
@property
def p99_execution_time(self) -> float:
if len(self.execution_times) < 2:
return self.avg_execution_time
sorted_times = sorted(self.execution_times)
idx = int(len(sorted_times) * 0.99)
return sorted_times[min(idx, len(sorted_times) - 1)]
class MetricsCollector:
def __init__(self):
self._node_metrics: dict[str, NodeMetrics] = defaultdict(
lambda: NodeMetrics(node_id="")
)
self._workflow_count = 0
self._workflow_success = 0
self._workflow_failure = 0
def record_node_result(self, result: NodeResult):
metrics = self._node_metrics[result.node_id]
metrics.node_id = result.node_id
metrics.total_executions += 1
metrics.total_retry_count += result.retry_count
if result.status == "completed":
metrics.success_count += 1
elif result.status == "completed_with_fallback":
metrics.fallback_count += 1
metrics.success_count += 1
else:
metrics.failure_count += 1
exec_time = result.end_time - result.start_time
if exec_time > 0:
metrics.execution_times.append(exec_time)
def record_workflow_result(self, result: WorkflowResult):
self._workflow_count += 1
if result.status == "completed":
self._workflow_success += 1
else:
self._workflow_failure += 1
for node_result in result.node_results.values():
self.record_node_result(node_result)
def get_node_metrics(self, node_id: str) -> NodeMetrics | None:
return self._node_metrics.get(node_id)
def get_all_metrics(self) -> dict[str, NodeMetrics]:
return dict(self._node_metrics)
def summary(self) -> dict:
return {
"total_workflows": self._workflow_count,
"success_workflows": self._workflow_success,
"failed_workflows": self._workflow_failure,
"workflow_success_rate": (
self._workflow_success / self._workflow_count
if self._workflow_count > 0
else 0.0
),
"nodes": {
nid: {
"success_rate": f"{m.success_rate:.2%}",
"avg_time": f"{m.avg_execution_time:.3f}s",
"p95_time": f"{m.p95_execution_time:.3f}s",
"p99_time": f"{m.p99_execution_time:.3f}s",
"total_retries": m.total_retry_count,
"fallback_count": m.fallback_count,
}
for nid, m in self._node_metrics.items()
},
}
告警规则
class AlertRule:
def __init__(
self,
name: str,
condition: Callable[[NodeMetrics], bool],
severity: str = "warning",
message_template: str = "",
):
self.name = name
self.condition = condition
self.severity = severity
self.message_template = message_template
def check(self, metrics: NodeMetrics) -> str | None:
if self.condition(metrics):
return self.message_template.format(
node_id=metrics.node_id,
success_rate=f"{metrics.success_rate:.2%}",
avg_time=f"{metrics.avg_execution_time:.3f}s",
)
return None
class AlertManager:
def __init__(self):
self._rules: list[AlertRule] = []
self._alerts: list[dict] = []
def add_rule(self, rule: AlertRule):
self._rules.append(rule)
def check_metrics(self, metrics_collector: MetricsCollector):
for node_id, metrics in metrics_collector.get_all_metrics().items():
for rule in self._rules:
alert_msg = rule.check(metrics)
if alert_msg:
self._alerts.append({
"rule": rule.name,
"severity": rule.severity,
"node_id": node_id,
"message": alert_msg,
"timestamp": datetime.now().isoformat(),
})
def get_alerts(self, severity: str | None = None) -> list[dict]:
if severity:
return [a for a in self._alerts if a["severity"] == severity]
return list(self._alerts)
alert_mgr = AlertManager()
alert_mgr.add_rule(AlertRule(
name="low_success_rate",
condition=lambda m: m.total_executions >= 5 and m.success_rate < 0.8,
severity="critical",
message_template="Node {node_id} success rate {success_rate} below 80%",
))
alert_mgr.add_rule(AlertRule(
name="high_latency",
condition=lambda m: m.avg_execution_time > 30.0,
severity="warning",
message_template="Node {node_id} avg execution time {avg_time} exceeds 30s",
))
alert_mgr.add_rule(AlertRule(
name="high_retry_rate",
condition=lambda m: m.total_executions > 0
and m.total_retry_count / m.total_executions > 2.0,
severity="warning",
message_template="Node {node_id} has high retry rate, avg retries per execution > 2",
))
5个常见坑及解决方案
坑1：DAG环检测遗漏条件边
条件边在运行时才决定是否激活，静态环检测可能漏掉运行时环路。
def validate_conditional_cycles(dag: DAGDefinition):
all_edges = list(dag.edges)
for edge in all_edges:
if edge.edge_type == EdgeType.CONDITIONAL:
test_edges = [
e for e in all_edges
if not (e.source_id == edge.source_id
and e.target_id == edge.target_id
and e.edge_type == EdgeType.CONDITIONAL)
]
test_edges.append(EdgeDefinition(
source_id=edge.source_id,
target_id=edge.target_id,
edge_type=EdgeType.NORMAL,
))
test_dag = DAGDefinition(
workflow_id=dag.workflow_id + "-test",
name=dag.name,
nodes=dag.nodes,
edges=test_edges,
entry_node=dag.entry_node,
)
try:
test_dag._check_cycle()
except ValueError:
raise ValueError(
f"Conditional edge '{edge.source_id}' → '{edge.target_id}' "
f"may create a runtime cycle"
)
解决方案：对所有条件边做"假设激活"的环检测，确保任何条件组合下都不会产生环路。
坑2：并行节点写冲突
多个并行节点同时修改state中的同一个key，导致数据覆盖。
def validate_parallel_write_safety(dag: DAGDefinition):
levels = TopologicalSorter(dag).compute_levels()
level_groups: dict[int, list[str]] = {}
for nid, level in levels.items():
level_groups.setdefault(level, []).append(nid)
for level, nodes in level_groups.items():
if len(nodes) <= 1:
continue
output_keys: dict[str, list[str]] = {}
for nid in nodes:
node = dag.nodes[nid]
keys = node.metadata.get("output_keys", [])
for key in keys:
output_keys.setdefault(key, []).append(nid)
conflicts = {k: v for k, v in output_keys.items() if len(v) > 1}
if conflicts:
raise ValueError(
f"Parallel write conflict at level {level}: {conflicts}"
)
解决方案：在DAG验证阶段检查并行节点的输出key是否冲突，或使用命名空间隔离。
坑3：Checkpoint序列化失败
state中包含不可序列化的对象（如数据库连接、文件句柄），导致checkpoint保存失败。
import pickle
def safe_serialize_state(state: dict) -> bytes:
try:
return pickle.dumps(state)
except (pickle.PicklingError, TypeError) as e:
clean_state = {}
for key, value in state.items():
try:
pickle.dumps(value)
clean_state[key] = value
except (pickle.PicklingError, TypeError):
clean_state[key] = f"<non-serializable: {type(value).__name__}>"
return pickle.dumps(clean_state)
解决方案：在handler中只返回JSON可序列化的数据，或使用自定义序列化器。
坑4：条件路由无匹配分支
所有条件边的condition都返回False，工作流卡死。
def ensure_default_branch(dag: DAGDefinition) -> DAGDefinition:
conditional_sources = set()
for edge in dag.edges:
if edge.edge_type == EdgeType.CONDITIONAL:
conditional_sources.add(edge.source_id)
builder = DAGBuilder(
f"{dag.workflow_id}-safe", f"{dag.name} (safe)"
)
for node in dag.nodes.values():
builder.add_node(node)
for edge in dag.edges:
builder.add_edge(
edge.source_id, edge.target_id,
edge.edge_type, edge.condition, edge.condition_name,
)
for source_id in conditional_sources:
has_normal = any(
e.source_id == source_id and e.edge_type == EdgeType.NORMAL
for e in dag.edges
)
if not has_normal:
builder.add_node(
NodeDefinition(
f"{source_id}_default",
NodeType.TRANSFORM,
handler=lambda s: {"routed_to_default": True},
)
)
builder.add_edge(source_id, f"{source_id}_default")
builder.set_entry(dag.entry_node)
return builder.build()
解决方案：为每个条件路由节点添加默认分支，确保至少有一条路径可执行。
坑5：子图状态泄漏
子工作流修改了父工作流的state，导致不可预期的副作用。
def isolate_sub_workflow_state(
parent_state: dict, sub_workflow_input_keys: list[str]
) -> tuple[dict, Callable[[dict], dict]]:
isolated = {k: parent_state[k] for k in sub_workflow_input_keys if k in parent_state}
def merge_back(sub_state: dict) -> dict:
output_keys = set(sub_workflow_input_keys)
return {k: v for k, v in sub_state.items() if k not in output_keys}
return isolated, merge_back
解决方案：子工作流执行时只传入必要的key，返回时只合并新增的key。
10个常见报错排查
| # | 报错信息 | 原因 | 解决方案 |
|---|---|---|---|
| 1 | Cycle detected in DAG |
节点间存在循环依赖 | 检查Edge定义，移除形成环的边 |
| 2 | Unreachable nodes detected |
节点没有从入口可达的路径 | 检查是否缺少Edge连接 |
| 3 | Entry node not found |
set_entry指定的节点不存在 | 确认node_id拼写正确 |
| 4 | Source/Target node not found |
add_edge引用了不存在的节点 | 先add_node再add_edge |
| 5 | Deadlock detected |
条件路由无匹配且无默认分支 | 添加默认分支或检查条件函数 |
| 6 | Node failed after N retries |
LLM API持续超时或报错 | 检查API Key、网络、降级策略 |
| 7 | Sub-workflow failed |
子工作流内部节点失败 | 查看子工作流node_results定位具体节点 |
| 8 | Checkpoint serialization error |
state包含不可序列化对象 | handler只返回dict[str, Any] |
| 9 | Parallel write conflict |
并行节点输出key冲突 | 使用命名空间隔离输出key |
| 10 | Runtime cycle via conditional edge |
条件边在运行时形成环路 | 使用validate_conditional_cycles检查 |
进阶优化技巧
1. 异步预取：提前加载下一层节点依赖
class PrefetchScheduler(PersistentDAGScheduler):
async def _run_from_start(self, initial_state=None):
execution_id = f"exec-{int(time.time() * 1000)}"
state = dict(initial_state or {})
node_results: dict[str, NodeResult] = {}
completed: set[str] = set()
pending: set[str] = {self.dag.entry_node}
start_time = time.time()
while pending:
ready = [
nid for nid in pending
if self._get_dependencies(nid).issubset(completed)
]
if not ready:
break
prefetch_tasks = []
for nid in ready:
node = self.dag.nodes[nid]
if node.node_type == NodeType.LLM_CALL:
prefetch_tasks.append(
asyncio.create_task(self._warmup_llm(nid))
)
tasks = [
self._execute_node(nid, state, node_results)
for nid in ready
]
results = await asyncio.gather(*tasks, return_exceptions=True)
if prefetch_tasks:
await asyncio.gather(*prefetch_tasks, return_exceptions=True)
for i, result in enumerate(results):
node_id = ready[i]
if isinstance(result, Exception):
node_results[node_id] = NodeResult(
node_id=node_id, status="failed", error=str(result)
)
return WorkflowResult(
workflow_id=self.dag.workflow_id,
execution_id=execution_id,
status="failed",
state=state,
node_results=node_results,
total_time=time.time() - start_time,
)
node_results[node_id] = result
state.update(result.output)
completed.add(node_id)
pending.discard(node_id)
next_nodes = self._router.resolve_next_nodes(node_id, state)
for next_id in next_nodes:
if next_id not in completed:
pending.add(next_id)
return WorkflowResult(
workflow_id=self.dag.workflow_id,
execution_id=execution_id,
status="completed",
state=state,
node_results=node_results,
total_time=time.time() - start_time,
)
async def _warmup_llm(self, node_id: str):
logger.info(f"Warming up LLM connection for node '{node_id}'")
await asyncio.sleep(0.01)
2. 超时熔断：防止慢节点拖垮整个工作流
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
half_open_max: int = 1,
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max = half_open_max
self._failure_count = 0
self._last_failure_time: float = 0
self._state = "closed"
self._half_open_count = 0
async def call(self, handler: Callable, state: dict) -> dict:
if self._state == "open":
if time.time() - self._last_failure_time > self.recovery_timeout:
self._state = "half_open"
self._half_open_count = 0
else:
raise RuntimeError("Circuit breaker is OPEN")
try:
result = await handler(state) if asyncio.iscoroutinefunction(handler) else await asyncio.to_thread(handler, state)
if self._state == "half_open":
self._half_open_count += 1
if self._half_open_count >= self.half_open_max:
self._state = "closed"
self._failure_count = 0
return result
except Exception as e:
self._failure_count += 1
self._last_failure_time = time.time()
if self._failure_count >= self.failure_threshold:
self._state = "open"
raise
3. DAG可视化：自动生成Mermaid图
def dag_to_mermaid(dag: DAGDefinition) -> str:
lines = ["graph TD"]
for edge in dag.edges:
style = ""
if edge.edge_type == EdgeType.CONDITIONAL:
style = f"|{edge.condition_name}|"
lines.append(f" {edge.source_id} -->{style} {edge.target_id}")
for nid, node in dag.nodes.items():
label = f"{nid}\\n({node.node_type.value})"
lines.append(f" {nid}[\"{label}\"]")
return "\n".join(lines)
使用在线Mermaid编辑器可以直接渲染DAG可视化图。
对比分析：自建DAG vs LangGraph vs Prefect
| 维度 | 自建DAG引擎 | LangGraph | Prefect |
|---|---|---|---|
| 语言 | Python（可扩展） | Python | Python |
| DAG定义 | 声明式Builder | StateGraph | Flow + Task |
| 并行执行 | ✅ 自动层级并行 | ✅ 基于asyncio | ✅ 原生Dask/Ray |
| 条件路由 | ✅ 条件边 | ✅ conditional_edges | ✅ branch |
| 状态持久化 | ✅ CheckpointManager | ✅ Checkpointer | ✅ Result + Storage |
| 断点续跑 | ✅ 原生支持 | ✅ 需配置 | ⚠️ 需自建 |
| 错误恢复 | ✅ 重试+降级 | ⚠️ 需自建 | ✅ 原生重试 |
| LLM集成 | ⚠️ 需自建 | ✅ LangChain生态 | ⚠️ 需自建 |
| 可视化 | ✅ Mermaid导出 | ✅ LangGraph Studio | ✅ Prefect UI |
| 学习曲线 | 中等 | 中等 | 低 |
| 生产监控 | ✅ 自定义Metrics | ⚠️ LangSmith | ✅ Prefect Cloud |
| 动态DAG | ✅ 运行时生成 | ✅ Command | ✅ 动态任务 |
| 子图嵌套 | ✅ SubWorkflowNode | ✅ Subgraph | ⚠️ 子Flow |
| 社区生态 | ❌ 自维护 | ✅ 活跃 | ✅ 活跃 |
| 适用场景 | 高度定制化需求 | LangChain用户 | 通用任务编排 |
选型建议：
- 自建DAG引擎：需要深度定制、与现有系统紧密集成、对性能有极致要求
- LangGraph：已在LangChain生态中、需要快速原型验证、重视LLM原生支持
- Prefect：通用任务编排、非LLM为主的混合工作流、需要开箱即用的UI
如果你在LangGraph多Agent协作方面有更多需求，推荐阅读 Python LangGraph多Agent协作实战。关于Agent记忆架构，参考 AI Agent记忆架构设计。Agent工具调用方面，参考 Python AI Agent工具使用指南。
在线工具推荐
| 工具 | 用途 | 链接 |
|---|---|---|
| JSON格式化 | 查看和编辑DAG定义JSON | /json/format |
| Mermaid编辑器 | 可视化DAG工作流图 | /dev/mermaid |
| Curl转代码 | 快速生成API调用代码 | /dev/curl-to-code |
总结
AI Agent工作流DAG引擎是2026年生产级AI系统的核心基础设施。本文覆盖了7种生产模式：
- 任务定义与依赖图构建 — 类型安全的DAG Builder，自动环检测和可达性验证
- 拓扑排序与并行调度 — 基于层级划分的自动并行执行，asyncio并发控制
- 条件路由与分支合并 — 声明式条件边，运行时动态路由
- 错误恢复与重试策略 — 指数退避重试、降级handler、熔断器
- 状态持久化与断点续跑 — Checkpoint机制，崩溃后从断点恢复
- 动态DAG与子图嵌套 — 运行时生成子任务，子工作流封装复用
- 生产监控与告警 — 节点级指标收集、成功率/延迟监控、告警规则
核心原则：DAG让AI工作流从"硬编码流水线"进化为"声明式编排"，是Agent系统从原型到生产的必经之路。
更多AI Agent实战内容，推荐阅读：
- Python LangGraph多Agent协作：从状态机到工作流编排的5种实战模式
- AI Agent记忆架构设计：从短期记忆到长期知识的4层体系
- Python AI Agent工具使用：从Function Calling到MCP协议的完整实战
外部参考：
本站提供浏览器本地工具，免注册即可试用 →
#AI Agent#DAG#工作流引擎#任务编排#LangGraph#并行执行#2026#AI与大数据