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 动态决策混合调度流程确定性高预定义低动态主流程高、细节低Token 消耗低LLM 只用于节点内部极高每步都调用 LLM中执行延迟低并行、无多余思考高逐步决策中灵活性低改流程 改代码高LLM 适应新情况中可观测性高每个节点状态可见低中间决策不透明中适合任务类型结构化、可预见的开放式、探索性的半结构化三、生产级代码实现下面分别实现 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_factorylist) 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), timeoutnode.timeout, ) if asyncio.iscoroutine(result): result await result return result except asyncio.TimeoutError: if attempt node.retry: raise logger.warning(f节点 {node.name} 超时第 {attempt1}/{node.retry1} 次重试) except Exception as e: if attempt node.retry: raise logger.warning(f节点 {node.name} 失败: {e}第 {attempt1}/{node.retry1} 次重试) 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_idx1} 层: {layer}) # 并行执行同层所有节点 tasks [ self._execute_node(self.nodes[name]) for name in layer ] results await asyncio.gather(*tasks, return_exceptionsTrue) 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(fDAG 执行完成 ({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_callNone): 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_steps5) 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_steps3) 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 工程化的道路上少走弯路。
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 动态决策混合调度流程确定性高预定义低动态主流程高、细节低Token 消耗低LLM 只用于节点内部极高每步都调用 LLM中执行延迟低并行、无多余思考高逐步决策中灵活性低改流程 改代码高LLM 适应新情况中可观测性高每个节点状态可见低中间决策不透明中适合任务类型结构化、可预见的开放式、探索性的半结构化三、生产级代码实现下面分别实现 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_factorylist) 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), timeoutnode.timeout, ) if asyncio.iscoroutine(result): result await result return result except asyncio.TimeoutError: if attempt node.retry: raise logger.warning(f节点 {node.name} 超时第 {attempt1}/{node.retry1} 次重试) except Exception as e: if attempt node.retry: raise logger.warning(f节点 {node.name} 失败: {e}第 {attempt1}/{node.retry1} 次重试) 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_idx1} 层: {layer}) # 并行执行同层所有节点 tasks [ self._execute_node(self.nodes[name]) for name in layer ] results await asyncio.gather(*tasks, return_exceptionsTrue) 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(fDAG 执行完成 ({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_callNone): 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_steps5) 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_steps3) 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 工程化的道路上少走弯路。
