LLM 多工具链式调用：从并行规划到依赖感知的执行引擎

LLM 多工具链式调用从并行规划到依赖感知的执行引擎一、单工具调用的天花板复杂任务中的编排困境当 LLM 通过 Function Calling 调用单个工具时流程相对简单模型输出工具名和参数运行时执行后返回结果。但真实业务场景中一个用户请求往往需要多个工具协同完成。例如帮我查一下上周销售额最高的产品然后给它的负责人发一封邮件这个请求隐含了查询数据库、筛选排序、查找负责人、发送邮件四个步骤且后三步依赖前一步的输出。如果让 LLM 逐轮调用工具每轮都要把完整的上下文重新输入Token 消耗会随工具数量线性增长延迟也成倍增加。更关键的是LLM 在长上下文中容易遗忘早期工具的返回结果导致后续调用参数错误。某客服 Agent 系统在连续调用 5 个工具后参数正确率从首轮的 94% 下降到 61%。二、DAG 编排模型与依赖感知调度多工具链式调用的核心是将 LLM 规划的工具调用序列建模为有向无环图DAG通过拓扑排序实现无依赖工具的并行执行有依赖工具的顺序执行。graph TD A[用户请求] -- B[LLM 规划工具调用] B -- C[query_sales_db] B -- D[get_current_date] C -- E[find_product_owner] D -- C E -- F[send_email] C -- G[generate_report] G -- F style A fill:#e1f5fe style B fill:#fff3e0 style F fill:#e8f5e9上图中get_current_date和query_sales_db存在依赖查询需要日期参数find_product_owner和generate_report都依赖query_sales_db的输出但两者之间无依赖可以并行最终send_email依赖两者的输出。三、依赖感知执行引擎的实现# tool_dag.py # 基于 DAG 的多工具依赖感知调度引擎 import asyncio from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional, Set dataclass class ToolNode: DAG 中的工具节点 name: str func: Callable # 实际执行函数 params: Dict[str, Any] # 静态参数 dependencies: List[str] field(default_factorylist) # 依赖的工具名 output_key: str # 输出结果的引用键 _result: Any None _error: Optional[Exception] None class ToolDAGExecutor: DAG 执行器拓扑排序 并行调度 依赖注入 def __init__(self, max_concurrency: int 5): self.nodes: Dict[str, ToolNode] {} self.max_concurrency max_concurrency self._semaphore asyncio.Semaphore(max_concurrency) def add_node(self, node: ToolNode) - None: self.nodes[node.name] node def _topological_levels(self) - List[List[str]]: 按拓扑层级分组同层节点可并行执行 in_degree {name: 0 for name in self.nodes} for name, node in self.nodes.items(): for dep in node.dependencies: in_degree[name] 1 levels [] remaining set(self.nodes.keys()) while remaining: # 当前层入度为 0 的节点 level [n for n in remaining if in_degree[n] 0] if not level: raise ValueError(检测到循环依赖DAG 不合法) levels.append(level) for n in level: remaining.remove(n) # 更新下游节点入度 for name in remaining: if n in self.nodes[name].dependencies: in_degree[name] - 1 return levels def _resolve_params(self, node: ToolNode) - Dict[str, Any]: 将依赖节点的输出注入到当前节点的参数中 resolved dict(node.params) for dep_name in node.dependencies: dep_node self.nodes[dep_name] if dep_node.output_key and dep_node._result is not None: resolved[dep_node.output_key] dep_node._result return resolved async def _execute_node(self, node: ToolNode) - None: 执行单个工具节点带并发控制和超时 async with self._semaphore: try: params self._resolve_params(node) # 支持同步和异步函数 if asyncio.iscoroutinefunction(node.func): result await asyncio.wait_for( node.func(**params), timeout30.0 ) else: result node.func(**params) node._result result except asyncio.TimeoutError: node._error TimeoutError(f工具 {node.name} 执行超时) except Exception as e: node._error e async def execute(self) - Dict[str, Any]: 按拓扑层级执行 DAG同层并行跨层串行 levels self._topological_levels() for level in self._topological_levels(): # 同层节点并行执行 tasks [self._execute_node(self.nodes[name]) for name in level] await asyncio.gather(*tasks) # 检查本层是否有执行失败失败则终止下游 for name in level: if self.nodes[name]._error is not None: raise self.nodes[name]._error return {name: node._result for name, node in self.nodes.items()}# llm_planner.py # LLM 规划器将用户请求解析为工具调用 DAG from pydantic import BaseModel from typing import List, Optional class ToolCall(BaseModel): 单个工具调用的规划结果 name: str params: dict dependencies: List[str] [] # 依赖哪些工具的输出 output_key: str # 本工具输出供下游引用的键名 class ToolPlan(BaseModel): LLM 规划的完整工具调用方案 calls: List[ToolCall] reasoning: str # LLM 的规划思路用于可解释性 TOOL_PLAN_PROMPT 你是一个工具调用规划器。根据用户请求和可用工具列表规划需要调用的工具序列。 规则 1. 明确标注每个工具调用的依赖关系 2. 无依赖的工具会被并行执行有依赖的按顺序执行 3. 每个工具的 output_key 用于将结果传递给下游工具 可用工具 {tools_description} 用户请求{user_request} async def plan_tool_calls( llm_client, user_request: str, tools_description: str ) - ToolPlan: 调用 LLM 规划工具调用 DAG prompt TOOL_PLAN_PROMPT.format( tools_descriptiontools_description, user_requestuser_request, ) response await llm_client.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}], response_format{type: json_object}, ) return ToolPlan.model_validate_json(response.choices[0].message.content)四、DAG 编排的架构权衡与局限LLM 规划的不可靠性。LLM 生成的 DAG 可能存在逻辑错误遗漏依赖、创建循环依赖、参数类型不匹配。实测中 GPT-4o 在 4 工具场景下的规划正确率为 89%7 工具场景下降至 71%。必须在执行前增加 DAG 校验层检查循环依赖、参数类型匹配、必填参数完整性。校验不通过时将错误信息反馈给 LLM 重新规划最多重试 2 次。并行度的上限。虽然 DAG 允许同层工具并行但 LLM API 的并发限制和下游服务的限流策略会约束实际并行度。当并行工具超过 10 个时LLM 的上下文窗口可能无法容纳所有工具的返回结果导致后续推理质量下降。错误传播与部分失败。DAG 中某个节点失败后所有下游节点都无法执行。对于部分失败场景需要设计降级策略用默认值替代失败节点的输出或跳过非关键路径的节点继续执行。但这增加了业务逻辑的复杂度需要在每个工具节点上定义失败时是否可跳过的属性。五、总结多工具链式调用的 DAG 编排模式通过依赖感知调度实现了并行执行与顺序依赖的统一。核心要点LLM 负责规划执行引擎负责调度两者解耦使得规划错误不影响执行安全性拓扑层级分组是并行调度的关键同层并行、跨层串行执行前必须增加 DAG 校验层防止循环依赖和参数不匹配。落地建议从 2-3 个工具的简单链式调用开始验证 LLM 规划准确率后再扩展到更复杂的 DAG 结构始终保留LLM 规划 → 人工确认 → 自动执行的过渡期。