从 Tool 到 ToolChain：Agent 工具编排的工程化落地指南

一、深度引言与场景痛点

前段时间有个朋友找我吐槽，说他写了一个 AI Agent，配了 12 个工具——搜索、Tavily、SQL 查询、邮件发送、日历操作、PDF 解析……每个工具单独测试都很丝滑，但一旦放进 Agent 里让 LLM 自己调度，问题就来了：工具调用串行执行时，一个搜索等 3 秒，一个 PDF 解析等 5 秒，12 个工具即使只调用其中 4 个，总延迟也飙到了十几秒。更要命的是，工具 A 的返回值需要喂给工具 B 做参数，工具 C 和 D 可以并行但框架默认串行调度，白白浪费了一半的响应时间。

这其实是一个很普遍的困境：从单 Tool 到 ToolChain，不仅仅是把函数注册给 Agent 就完事了。它涉及依赖解析、拓扑排序、并行执行、错误传递、结果缝合等一系列工程问题。大部分框架（包括 LangChain、AutoGen）的默认调度器都是"LLM 说调谁就调谁，一个一个来"，这在真实生产场景下完全不可接受。

打个比方，单 Tool 像是一个人在厨房里拿锅铲炒菜，ToolChain 则是一个后厨团队——有人切菜、有人备料、有人掌火，还得有人负责传菜。如果厨师长（LLM）一道一道地下指令，等切完菜再点火，等火烧热了再找调料，这顿饭天黑也端不上桌。我们需要的是一个厨房领班系统——它能理解菜谱（任务定义）、分配人力（并行调度）、监控进度（状态追踪）、并在出问题时及时补救（异常恢复）。

本文就来聊聊，怎么从零搭建一个生产级的 Agent ToolChain 编排引擎。我们不谈概念，直接上架构设计和代码。

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

ToolChain 的核心问题可以抽象为一个 DAG（有向无环图）调度问题：

• 每个 Tool 是图中的一个节点
• 如果 Tool B 依赖 Tool A 的输出，则有一条 A → B 的有向边
• 没有依赖关系的 Tool 可以在不同批次中并行执行
• 调度器需要按拓扑层级逐层推进

下面这张图描述了整个编排流程：

flowchart TD
    A[用户请求进入] --> B{LLM 规划阶段}
    B --> C[解析 Tool 调用序列]
    C --> D[构建依赖 DAG]
    D --> E{依赖分析}
    E -->|无依赖| F1[批次1: 并行执行]
    E -->|有依赖| F2[拓扑排序]
    F2 --> G1[第1层并行]
    G1 --> G2[第2层: 消费上层输出]
    G2 --> G3[第N层: 最终工具]
    F1 --> H[结果收集器]
    G3 --> H
    H --> I{完整性检查}
    I -->|缺失| J[重试缺失节点]
    I -->|完成| K[缝合结果]
    J --> H
    K --> L[返回最终响应]

这个架构的关键在于三件事：

第一，依赖声明。 每个 Tool 不仅要声明自己能做什么，还要声明它需要什么输入、产出什么输出。这很像函数签名，但比函数签名多了一层语义标注——因为 LLM 需要理解哪个工具的输出可以用作另一个工具的输入。

第二，拓扑分层。 依赖 DAG 构建完成后，调度器通过 Kahn 算法或 DFS 进行拓扑分层，同一层的节点没有相互依赖，可以安全并行。

第三，错误传播与隔离。 生产环境中，工具调用失败是常态而非异常。ToolChain 的调度器需要区分"可重试错误"（如网络超时）和"不可重试错误"（如权限不足），并且在单个工具失败时，能决定是终止整条链路还是走降级路径。

三、生产级代码实现

下面是一套完整的 ToolChain 编排引擎的核心实现。注意：这不是玩具代码，它包含了依赖解析、拓扑调度、并行执行、超时控制和错误分类。

import asyncio
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Set
from enum import Enum
from collections import defaultdict, deque


class ToolStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"


class ErrorCategory(Enum):
    RETRYABLE = "retryable"      # 网络超时、临时故障
    FATAL = "fatal"               # 权限不足、参数错误
    DEGRADABLE = "degradable"    # 可降级处理


@dataclass
class ToolNode:
    """工具节点定义"""
    name: str
    func: Callable
    description: str
    # 声明式依赖：指明本工具需要哪些上游工具的输出
    depends_on: List[str] = field(default_factory=list)
    # 产出的语义键，供下游工具引用
    produces: str = ""
    # 超时时间（秒）
    timeout: float = 30.0
    # 最大重试次数
    max_retries: int = 2
    # 是否可选（失败不影响后续）
    optional: bool = False


@dataclass
class ExecutionResult:
    """单个工具的执��结果"""
    tool_name: str
    status: ToolStatus
    output: Any = None
    error: Optional[Exception] = None
    error_category: Optional[ErrorCategory] = None
    elapsed: float = 0.0


class ToolChainScheduler:
    """ToolChain 核心调度器

    职责：
    1. 根据依赖声明构建 DAG
    2. 拓扑分层
    3. 按层并行执行
    4. 结果收集与错误处理
    """

    def __init__(self, tools: List[ToolNode]):
        self.tools = {t.name: t for t in tools}
        self.results: Dict[str, ExecutionResult] = {}
        self._validate_dag()

    def _validate_dag(self):
        """校验 DAG 合法性：检测循环依赖和缺失依赖"""
        for tool in self.tools.values():
            for dep in tool.depends_on:
                if dep not in self.tools:
                    raise ValueError(f"工具 '{tool.name}' 依赖了未注册的工具 '{dep}'")

        # 检测循环依赖：拓扑排序验证
        in_degree = {name: len(t.depends_on) for name, t in self.tools.items()}
        zero_queue = deque([name for name, d in in_degree.items() if d == 0])
        visited = 0
        while zero_queue:
            current = zero_queue.popleft()
            visited += 1
            for name, tool in self.tools.items():
                if current in tool.depends_on:
                    in_degree[name] -= 1
                    if in_degree[name] == 0:
                        zero_queue.append(name)

        if visited != len(self.tools):
            raise ValueError("检测到循环依赖，DAG 不合法")

    def _topological_layers(self) -> List[List[str]]:
        """将 DAG 按拓扑层级分组，同层可并行"""
        in_degree = {name: len(t.depends_on) for name, t in self.tools.items()}
        adj = defaultdict(list)
        for name, tool in self.tools.items():
            for dep in tool.depends_on:
                adj[dep].append(name)

        layers = []
        zero_queue = deque([name for name, d in in_degree.items() if d == 0])

        while zero_queue:
            layer = list(zero_queue)
            layers.append(layer)
            zero_queue.clear()
            for node in layer:
                for downstream in adj[node]:
                    in_degree[downstream] -= 1
                    if in_degree[downstream] == 0:
                        zero_queue.append(downstream)

        return layers

    def _classify_error(self, error: Exception) -> ErrorCategory:
        """错误分类：根据异常类型判断可重试性"""
        if isinstance(error, (asyncio.TimeoutError, ConnectionError)):
            return ErrorCategory.RETRYABLE
        if isinstance(error, (PermissionError, ValueError)):
            return ErrorCategory.FATAL
        return ErrorCategory.DEGRADABLE

    async def _execute_single(
        self, tool: ToolNode, upstream_results: Dict[str, Any]
    ) -> ExecutionResult:
        """执行单个工具，含重试逻辑"""
        import time
        start = time.monotonic()

        for attempt in range(tool.max_retries + 1):
            try:
                # 将上游结果注入工具函数
                result = await asyncio.wait_for(
                    tool.func(upstream_results), timeout=tool.timeout
                )
                # 处理异步和同步两种情况
                if asyncio.iscoroutine(result):
                    result = await result
                return ExecutionResult(
                    tool_name=tool.name,
                    status=ToolStatus.SUCCESS,
                    output=result,
                    elapsed=time.monotonic() - start,
                )
            except asyncio.TimeoutError as e:
                if attempt == tool.max_retries:
                    return ExecutionResult(
                        tool_name=tool.name,
                        status=ToolStatus.FAILED,
                        error=e,
                        error_category=ErrorCategory.RETRYABLE,
                        elapsed=time.monotonic() - start,
                    )
            except Exception as e:
                cat = self._classify_error(e)
                if cat == ErrorCategory.FATAL or attempt == tool.max_retries:
                    return ExecutionResult(
                        tool_name=tool.name,
                        status=ToolStatus.FAILED,
                        error=e,
                        error_category=cat,
                        elapsed=time.monotonic() - start,
                    )

        # 理论上不会到这里
        return ExecutionResult(tool_name=tool.name, status=ToolStatus.FAILED)

    async def run(self, initial_context: Optional[Dict] = None) -> Dict[str, ExecutionResult]:
        """主调度入口：按拓扑层级并行执行所有工具"""
        layers = self._topological_layers()
        accumulated_results: Dict[str, Any] = initial_context or {}

        for layer_idx, layer in enumerate(layers):
            # 为当前层每个工具准备上游结果
            tasks = []
            for tool_name in layer:
                tool = self.tools[tool_name]
                # 仅注入已声明的依赖
                upstream = {
                    dep: accumulated_results.get(dep)
                    for dep in tool.depends_on
                }
                tasks.append(self._execute_single(tool, upstream))

            # 同层并行执行
            layer_results = await asyncio.gather(*tasks, return_exceptions=False)

            for result in layer_results:
                if isinstance(result, ExecutionResult):
                    self.results[result.tool_name] = result
                    if result.status == ToolStatus.SUCCESS:
                        accumulated_results[result.tool_name] = result.output
                    elif not self.tools[result.tool_name].optional:
                        # 非可选工具失败则终止
                        return self.results

        return self.results


# ================== 使用示例 ==================

async def search_tool(ctx: dict) -> str:
    """模拟搜索工具"""
    await asyncio.sleep(0.5)
    return "搜索到 3 篇相关文档"


async def summarize_tool(ctx: dict) -> str:
    """模拟摘要工具，依赖搜索工具的输出"""
    search_result = ctx.get("search_tool", "")
    await asyncio.sleep(0.3)
    return f"基于「{search_result[:20]}...」生成的摘要"


async def translate_tool(ctx: dict) -> str:
    """模拟翻译工具，和摘要工具并行执行"""
    await asyncio.sleep(0.4)
    return "translated content"


# 注册工具
tools = [
    ToolNode(
        name="search_tool",
        func=search_tool,
        description="搜索相关文档",
        depends_on=[],
        produces="search_results",
    ),
    ToolNode(
        name="summarize_tool",
        func=summarize_tool,
        description="对搜索结果生成摘要",
        depends_on=["search_tool"],
        produces="summary",
    ),
    ToolNode(
        name="translate_tool",
        func=translate_tool,
        description="翻译内容",
        depends_on=["search_tool"],
        produces="translation",
    ),
]

# 启动调度
async def main():
    scheduler = ToolChainScheduler(tools)
    results = await scheduler.run()
    for name, r in results.items():
        print(f"[{r.status.value}] {name}: {r.output[:50] if r.output else r.error}")


if __name__ == "__main__":
    asyncio.run(main())

这段代码的核心设计考量：

• 拓扑分层决定了并行度上限。_topological_layers() 方法使用 Kahn 算法的变体，每层独立 gather。
• 错误分类不是形式上的装饰，它直接影响链路能否降级运行。_classify_error 方法将异常分为三类，只有 FATAL 会立即终止。
• 结果传递通过字典解耦，下游工具通过 depends_on 声明的键名从 accumulated_results 中获取上游输出，而非通过位置参数耦合。

四、边界分析与架构权衡

任何架构方案都有其适用范围和代价，ToolChain 编排引擎也不例外：

1. 静态 DAG vs 动态 DAG
本文实现的是静态 DAG——依赖关系在工具注册时就已声明。这在工具集固定的场景下完全够用，性能也最可控。但如果你的 Agent 需要根据中间结果动态决定下一步调用哪些工具（比如"搜索结果不够，再搜一轮"），就需要动态 DAG。动态 DAG 的实现复杂度是指数级上升的，因为它要求在运行时重构图结构并处理潜在的循环。

2. 集中调度 vs 去中心化调度
ToolChainScheduler 采用了集中式调度模式——一个调度器管所有工具。优势是实现简单、状态一致性好。代价是调度器本身成为单点。如果你的工具数量超过 50 个且并发度极高，可以考虑引入 Worker Pool 模式将工具分配到多个调度实例上。

3. 语义依赖 vs 显式依赖
当前实现依赖工具开发者显式声明 depends_on。另一种思路是让 LLM 在规划阶段自动推导依赖关系，这更灵活但也更不可控。我的建议是：工具数量 ≤ 20 时用显式依赖，> 20 时混合使用——核心链路显式声明，边缘工具让 LLM 推导。

4. 重试策略的诅咒
max_retries 设置为 2 是经过生产验证的经验值。重试次数太多，不仅拉长延迟，还可能对下游服务造成雪崩。尤其当上游 API 有速率限制时，无脑重试反而会让问题恶化。

五、总结

从 Tool 到 ToolChain 的转变，本质上是从"让 AI 调用单个函数"升级到"让 AI 编排一个工作流"。这中间的核心工程问题是依赖管理、并行调度和错误传播，而这些问题的解法，其实早在操作系统的进程调度和数据库的事务管理中就有了答案——我们只是把那些成熟的模式搬到了 Agent 的上下文里。

一个生产级的 ToolChain 编排引擎不需要多花哨的抽象，做到三条就够了：依赖声明要简洁（让下游工具知道去哪里拿数据）、调度要分层并行（别让没有依赖的工具互相等待）、错误要分类处理（可重试的重试，不可重试的快失败）。

希望这篇指南能帮你少踩几个坑。下一篇我们来聊聊 Agent 的上下文窗口管理——当 System Prompt 超长时，怎么给 LLM "减肥"。