企业官网建设流程全解析

Agent 任务调度引擎：DAG 图驱动 vs LLM 动态决策的工程权衡

一、深度引言与场景痛点

Agent 的任务调度方式，决定了 Agent 的"智商上限"和"稳定性"之间的平衡。目前业界有两种主流的调度范式：

DAG 图驱动：在 Agent 开始工作之前，你就定义好了它该先做什么、后做什么、什么条件下走哪条分支。工作流是预先编排好的，Agent 只是按照图结构执行节点。代表框架：LangGraph、Temporal、Prefect。

LLM 动态决策：Agent 没有预设的工作流，它每走一步都靠 LLM 自己决定"下一步该做什么"。代表框架：ReAct Agent、AutoGPT 的思维链循环。

两种方式的对立非常鲜明：DAG 图驱动像工厂的流水线——精确、高效、可控，但缺乏灵活性；LLM 动态决策像创业公司的 CEO——自由、智能、适应性强，但容易跑偏、成本高、结果不可预测。

选择哪一种，本质上是在回答一个问题：你的任务流程有多确定？

如果你的 Agent 就是"搜文档 → 分析 → 写报告"，每一步都明确且步骤间依赖关系清晰，那 DAG 图驱动是最佳选择——稳定、可追踪、成本可控。但如果你的 Agent 面对的是"帮我做一个竞品分析"这种开放性任务（需要它自己决定搜哪些竞品、从哪些维度分析、如何组织报告），LLM 动态决策的能力就必不可少。

本文不站队，而是深入分析两种调度范式的工程差异、适用场景和混合方案，帮你做出理性的架构决策。

二、底层机制与原理深度剖析

flowchart TD A[Agent 任务调度引擎] A --> B[DAG 图驱动] A --> C[LLM 动态决策] A --> D[混合调度] subgraph DAG_Detail [DAG 图驱动] B1[预定义工作流] --> B2[节点 = 确定性函数] B2 --> B3[边 = 数据依赖/条件] B3 --> B4[运行时 = 拓扑排序 + 并行执行] B4 --> B5[确定性: 相同输入 → 相同输出] end subgraph LLM_Detail [LLM 动态决策] C1[无预定义流程] --> C2[每步: 观察 → 思考 → 行动] C2 --> C3[LLM 自主选择工具/路径] C3 --> C4[循环直到目标达成或超时] C4 --> C5[非确定性: 相同输入 → 可能不同输出] end subgraph Hybrid_Detail [混合调度] D1[高层: DAG 定义主流程] --> D2[节点内部: LLM 自由发挥] D2 --> D3[运行时: DAG 保证结构 + LLM 提供灵活性] D3 --> D4[代表: LangGraph with LLM nodes] end style DAG_Detail fill:#e3f2fd style LLM_Detail fill:#fce4ec style Hybrid_Detail fill:#e8f5e9

关键的差异维度对比：

三、生产级代码实现

下面分别实现 DAG 图驱动和 LLM 动态决策两种调度引擎，然后展示混合方案。

3.1 DAG 图驱动调度器

import asyncio import time import logging from enum import Enum from typing import Any, Callable, Dict, List, Optional from dataclasses import dataclass, field from collections import defaultdict, deque logger = logging.getLogger(__name__) class NodeType(Enum): TASK = "task" # 普通任务节点 LLM = "llm" # LLM 调用节点（需要模型推理） CONDITION = "condition" # 条件分支节点 MERGE = "merge" # 汇聚节点 class NodeStatus(Enum): PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" SKIPPED = "skipped" @dataclass class GraphNode: """DAG 节点""" name: str node_type: NodeType func: Optional[Callable] = None # 同步或异步函数 depends_on: List[str] = field(default_factory=list) timeout: float = 30.0 retry: int = 0 # 失败重试次数 def __hash__(self): return hash(self.name) class DAGScheduler: """DAG 图驱动任务调度器 核心算法： 1. 拓扑排序找出所有节点的执行层级 2. 同层节点并行执行 3. 条件节点根据上游结果选择下游路径 """ def __init__(self): self.nodes: Dict[str, GraphNode] = {} self._results: Dict[str, Any] = {} self._status: Dict[str, NodeStatus] = {} def add_node(self, node: GraphNode): if node.name in self.nodes: raise ValueError(f"节点名重复: {node.name}") self.nodes[node.name] = node def add_edge(self, from_node: str, to_node: str): """添加依赖边：to_node 依赖 from_node""" if to_node not in self.nodes: raise ValueError(f"目标节点不存在: {to_node}") if from_node not in self.nodes: raise ValueError(f"源节点不存在: {from_node}") if from_node not in self.nodes[to_node].depends_on: self.nodes[to_node].depends_on.append(from_node) def _validate(self): """验证 DAG 合法性（检测循环依赖）""" in_degree = {name: len(n.depends_on) for name, n in self.nodes.items()} adj = defaultdict(list) for name, node in self.nodes.items(): for dep in node.depends_on: adj[dep].append(name) zero_queue = deque([n for n, d in in_degree.items() if d == 0]) visited = 0 while zero_queue: current = zero_queue.popleft() visited += 1 for downstream in adj[current]: in_degree[downstream] -= 1 if in_degree[downstream] == 0: zero_queue.append(downstream) if visited != len(self.nodes): raise ValueError("DAG 包含循环依赖") def _topological_layers(self) -> List[List[str]]: """拓扑分层""" in_degree = {name: len(n.depends_on) for name, n in self.nodes.items()} adj = defaultdict(list) for name, node in self.nodes.items(): for dep in node.depends_on: adj[dep].append(name) layers = [] zero_queue = deque([n for n, d in in_degree.items() if d == 0]) while zero_queue: layer = list(zero_queue) layers.append(layer) zero_queue.clear() for node_name in layer: for downstream in adj[node_name]: in_degree[downstream] -= 1 if in_degree[downstream] == 0: zero_queue.append(downstream) return layers async def _execute_node(self, node: GraphNode) -> Any: """执行单个节点（带重试）""" for attempt in range(node.retry + 1): try: # 准备上游结果 upstream = { dep: self._results.get(dep) for dep in node.depends_on } result = await asyncio.wait_for( node.func(upstream), timeout=node.timeout, ) if asyncio.iscoroutine(result): result = await result return result except asyncio.TimeoutError: if attempt == node.retry: raise logger.warning(f"节点 {node.name} 超时，第 {attempt+1}/{node.retry+1} 次重试") except Exception as e: if attempt == node.retry: raise logger.warning(f"节点 {node.name} 失败: {e}，第 {attempt+1}/{node.retry+1} 次重试") await asyncio.sleep(1) async def run(self) -> Dict[str, Any]: """执行 DAG""" self._validate() layers = self._topological_layers() for node_name in self.nodes: self._status[node_name] = NodeStatus.PENDING for layer_idx, layer in enumerate(layers): logger.info(f"执行第 {layer_idx+1} 层: {layer}") # 并行执行同层所有节点 tasks = [ self._execute_node(self.nodes[name]) for name in layer ] results = await asyncio.gather(*tasks, return_exceptions=True) for node_name, result in zip(layer, results): if isinstance(result, Exception): self._status[node_name] = NodeStatus.FAILED self._results[node_name] = str(result) logger.error(f"节点 {node_name} 执行失败: {result}") else: self._status[node_name] = NodeStatus.COMPLETED self._results[node_name] = result return self._results # ================== DAG 使用示例 ================== async def dag_demo(): """演示 DAG 图驱动的搜索-分析-报告流程""" scheduler = DAGScheduler() async def search_docs(ctx: dict) -> list: logger.info("搜索文档...") await asyncio.sleep(0.5) return [{"id": "doc1", "title": "Python 异步编程"}, {"id": "doc2", "title": "协程原理"}] async def analyze_docs(ctx: dict) -> dict: docs = ctx.get("search", []) logger.info(f"分析 {len(docs)} 篇文档...") await asyncio.sleep(0.3) return {"keywords": ["asyncio", "协程"], "summary": "核心概念分析"} async def generate_report(ctx: dict) -> str: analysis = ctx.get("analyze", {}) search_result = ctx.get("search", []) logger.info(f"基于分析结果生成报告...") await asyncio.sleep(0.8) return f"报告：共分析 {len(search_result)} 篇文档，关键词: {analysis.get('keywords')}" # 构建 DAG scheduler.add_node(GraphNode("search", NodeType.TASK, search_docs, depends_on=[])) scheduler.add_node(GraphNode("analyze", NodeType.TASK, analyze_docs, depends_on=["search"])) scheduler.add_node(GraphNode("report", NodeType.TASK, generate_report, depends_on=["search", "analyze"])) import time start = time.monotonic() results = await scheduler.run() elapsed = (time.monotonic() - start) * 1000 print(f"DAG 执行完成 ({elapsed:.0f}ms):") for name, result in results.items(): print(f" [{name}] {str(result)[:80]}...") ### 3.2 LLM 动态决策调度器 class DynamicAgent: """LLM 动态决策 Agent 特点： - 没有预定义工作流 - 每步由 LLM 决定下一个动作 - 循环直到任务完成或达到上限 """ def __init__(self, max_steps: int = 10, llm_call=None): self.max_steps = max_steps self.llm_call = llm_call or self._mock_llm self.tools = { "search": self._search, "analyze": self._analyze, "python": self._run_python, } self.trace: List[Dict] = [] async def _mock_llm(self, context: str) -> dict: """模拟 LLM 决策（真实场景替换为 LLM API 调用）""" # 实际中，LLM 返回 JSON: {"action": "tool_name", "params": {...}, "thought": "..."} return {"action": "search", "params": {"query": "Python 异步"}, "thought": "需要搜索"} async def _search(self, params: dict) -> str: return "搜索到 3 篇相关文档" async def _analyze(self, params: dict) -> str: return "分析结论：重点在协程和事件循环" async def _run_python(self, params: dict) -> str: return "代码执行结果：[0, 1, 2]" async def run(self, task: str) -> Dict: """执行 LLM 动态决策循环""" context = f"任务: {task}\n\n" self.trace = [] for step in range(self.max_steps): # 1. LLM 思考 + 决策 decision = await self.llm_call(context) thought = decision.get("thought", "") action = decision.get("action", "finish") params = decision.get("params", {}) self.trace.append({ "step": step + 1, "thought": thought, "action": action, "params": params, }) # 2. 执行动作 if action == "finish": return {"status": "completed", "trace": self.trace} if action not in self.tools: context += f"\n错误：工具 '{action}' 不存在" continue try: result = await self.tools[action](params) if asyncio.iscoroutine(result): result = await result context += f"\n[工具 {action} 返回] {result}" except Exception as e: context += f"\n[工具 {action} 错误] {e}" return {"status": "max_steps_reached", "trace": self.trace} ### 3.3 混合调度方案 class HybridScheduler: """混合调度器 高层用 DAG 保证主流程结构， 节点内部用 LLM 提供灵活性 """ def __init__(self): self.dag = DAGScheduler() async def run_structured_task(self, task: str) -> Dict: """执行半结构化任务：DAG 主流程 + LLM 子节点""" dynamic_agent = DynamicAgent(max_steps=5) async def explore_node(ctx: dict): """探索节点：LLM 自由发挥""" return await dynamic_agent.run("探索相关技术主题") async def structured_search(ctx: dict): """结构化搜索节点：确定性逻辑""" results = ctx.get("explore", {}) return f"基于探索结果 {results} 的结构化搜索" # DAG 主流程 self.dag.add_node(GraphNode("explore", NodeType.LLM, explore_node, depends_on=[])) self.dag.add_node(GraphNode("search", NodeType.TASK, structured_search, depends_on=["explore"])) return await self.dag.run() # ================== 运行示例 ================== async def main(): print("=== DAG 图驱动 ===") await dag_demo() print("\n=== LLM 动态决策 ===") agent = DynamicAgent(max_steps=3) async def mock_llm(context: str) -> dict: # 模拟 LLM 两步完成任务 if "搜索到" not in context: return {"action": "search", "params": {"query": "test"}, "thought": "先搜索"} else: return {"action": "finish", "params": {}, "thought": "搜索完成，结束"} agent.llm_call = mock_llm result = await agent.run("分析 Python 异步编程") for step in result.get("trace", []): print(f" 步骤 {step['step']}: {step['thought']} → {step['action']}") if __name__ == "__main__": asyncio.run(main())

四、边界分析与架构权衡

1. 选型决策树

你的任务流程有多确定？ ├── 完全确定（固定步骤、固定顺序） │ └── → DAG 图驱动（最优：零浪费） │ ├── 半确定（主流程固定，但某些步骤需要灵活性） │ └── → 混合调度（DAG 主流程 + LLM 子节点） │ └── 高度不确定（无法预知需要哪些步骤） └── → LLM 动态决策（灵活性优先） └── 但必须设置 max_steps 和 Token 预算上限

2. LLM 动态决策的 Token 陷阱

LLM 动态决策每步都需要把整个历史上下文（包括之前的所有思考、所有工具调用结果）传回 LLM，这导致 Token 消耗随着步数呈超线性增长。一个 10 步的 Agent 任务，可能消耗 5000-20000 Token。如果 QPS = 100，一天的 Token 消耗能轻松达到数千万。解决办法是：只传最近 N 步的上下文（如最近 3 步），更早的历史用摘要代替。

3. DAG 的"脆弱性"

DAG 图驱动最大的弱点是缺乏适应性。如果某个节点失败了，DAG 只会按照预设的重试策略处理，但无法"换一个思路"——比如搜索文档没搜到，DAG 不能自主决定"换个关键词再搜"。这在结构化任务中通常不是问题（因为你对流程已经足够了解），但确实限制了 Agent 的"智能感"。

4. 混合方案中的"切分点"选择

不是所有节点都适合用 LLM。判断标准很简单：花 3 秒思考就能确定最优方案的事情，用确定性函数；需要推理、创造、判断的事情，用 LLM。比如"从数据库查数据"是确定性的（SQL 写好就行），但"根据数据写结论"是需要 LLM 的。

五、总结

DAG 图驱动和 LLM 动态决策不是"二选一"的问题，而是"各自在最合适的场景下发挥优势"的问题。我的建议是：

• 新项目从 DAG 开始。因为 DAG 让你想清楚业务流程，即使后来发现需要更多灵活性，DAG 的拓扑结构也能作为混合调度的主流程骨架。
• 在 DAG 的末端节点引入 LLM。比如最后的"生成报告"节点用 LLM，而前面的"搜索数据 → 清洗数据 → 聚合统计"用确定性函数。这样既保证了流程的可靠性，又获得了 LLM 的生成能力。
• 只在探索性任务中放开 LLM 的决策自由度。并且一定要设置步数上限和 Token 预算——没有边界的 LLM 动态决策，就是一颗没有引信的定时炸弹。

到这里，0706 和 0707 两天的 20 篇文章全部完成。从 ToolChain 编排到混合调度引擎，从 Multi-Level Cache 到向量库选型，从火焰图定位瓶颈到时序图可视化——这 20 篇文章串起了 Agent & RAG 工程化的完整知识图谱。希望这一系列能帮你在 AI 工程化的道路上少走弯路。