工具编排与组合：Agent的乐高玩法

可视化演示

展示四种工具组合模式、数据流转格式转换和并行调用聚合策略。

组合模式设计

拼乐高时你会发现，高手从来不是随手乱拼。他们嘴里念叨的“先搭底盘、再装传动、最后上外观”，其实就是一套组合模式。工具编排的第一课，就是让你从工具箱里挑出积木后，先想清楚“按什么结构往上搭”。最常见的三种搭法，比乐高说明书还要简单直观。

第一种叫管道流水线，像工厂传送带：A 工具的输出，直接变成 B 工具的输入。比如“查天气 -> 查穿衣建议 -> 生成出行清单”，每个工具只干一件事，干完往传送带上放。第二种叫扇出分发，像一条转接头分出多根电线：拿到一个问题，同时派给多个工具去各查各的。第三种叫汇聚合并，像多条滑梯最后落进同一个沙坑：把几个工具的结果收拢起来，合成一份总报告。

这三种模式的数学本质其实是一种函数复合——把前一个函数的输出作为后一个函数的输入，层层嵌套：

回看那箱乐高：单块积木只是静物，但用管道把它们串成传动轴，用扇出让一个指令同时驱动多个齿轮，用汇聚把零散零件拼成车身——组合模式就是你手里的“说明书”。

抽象层次划分

搭过大型乐高套装的人都知道，最顶层的说明书写的是“搭城堡塔尖”，中间的写“搭城门框架”，最底层的才写“把这块 2×4 的砖嵌到那块 2×2 上”。如果所有步骤都只描写到单块积木，说明书会比字典还厚，你也根本找不到塔尖在哪。工具编排同样需要抽象层次。

最底层叫原子工具，对应单块积木——查一次天气、算一个数、搜一段资料，功能单一不可再分。往上一层叫场景服务，相当于“城门框架”——把“查天气、查穿衣、查路线”打包成一个“出行准备”服务，调用方不用关心里面用了几个工具。最顶层是能力总线，像套装里的总装图——所有场景服务统一挂到一条总线上，Agent 只需要向总线发出需求，总线自动路由到合适的服务。

分层让每个层级只关心自己的事：顶层只管“要什么”，中间层只管“拼哪些”，底层只管“怎么拼”。这种分层结构可以用一个简洁的层级公式描述：

乐高说明书之所以能一页只画三步，正因为它做了分层抽象。你的 Agent 也一样——把原子工具封装成场景服务，再挂到能力总线上，整个系统的复杂度才不会被一眼看穿，改造起来才不至于牵一发而动全身。

数据流转设计

乐高积木之间靠凸点和凹槽咬合。如果两块砖的凸点长得不一样，它们根本拼不到一起。工具编排里的数据流转，做的就是“凸点对齐”这件事——规定前一个工具吐出来的数据长什么样，后一个工具才接得住。我们把这个约定叫做输入输出契约。

假设天气工具输出的是 {"temperature": 26, "condition": "晴"}，而后面的穿衣工具期待的是 {"temp": 26, "weather": "晴"}——字段名对不上，就像凸点尺寸不合，硬拼只会散架。解决的办法是让所有工具共享一份上下文（Context）：一个流动的公共数据包。每个工具从中读自己需要的字段，再往里写入自己的输出，像流水线上的零件盒，每个工位往盒里放一个零件，也从中取走需要的零件。

在工程上，这个共享上下文本质上是一个可传递的数据对象，前一个工具的输出经过适配，就能变成后一个工具可消费的输入：

这就是为什么编排高手总在强调“先定契约、再写工具”。就像乐高套装里所有积木必须遵循统一的凸点标准，你的工具只要都遵循同一个 Context 数据契约，任意两块“积木”都能严丝合缝地咬合在一起，不必担心谁配合不上谁。

格式转换适配

如果说数据流转解决了“积木怎么咬合”，那格式转换适配解决的是“两块积木的凸点对不上时怎么办”。玩乐高时你偶尔会遇到两块来自不同系列的砖，接口形状不一样。这时你不会把砖扔掉，而是翻出一块适配器零件——一头是新式凸点，一头是旧式凹槽，往中间一卡，两块砖就顺利连上了。工具编排中，我们把这个中间零件叫做适配器（Adapter）。

具体怎么做？第一步是统一数据规范：团队约定好一种“标准格式”，比如所有工具的时间都用 ISO 格式、经纬度都用小数。第二步是字段映射：写一份“翻译表”，把工具 A 的 temp 字段翻译成标准格式的 temperature，再把标准格式翻译成工具 B 需要的 temp。第三步是适配器封装：把这些翻译逻辑集中到一个包裹层里，工具本身不用改一行代码，只要给它套上适配器，它就能接入整个编排系统。

这种“不改工具、只加中间层”的思想，在数学上可以看作两套数据结构之间的映射函数。给定输入格式的集合 F_A 和输出需要的格式集合 F_B，适配器就是两个集合之间的双射转换：

有了适配器，你的工具库就像乐高混搭玩法一样——不同系列的积木也能强行合体。今天接入一个第三方天气 API，明天换掉一个老旧的数据库查询工具，只要适配器写得好，整个组合体不会因为换了一块砖而散架。

并行调用策略

搭乐高时，如果城堡的四面墙互不影响，你不会傻等一面墙完工再去搭下一面。你会同时动手，四面墙一起往上长，最后再合拢。Agent 编排也如此：当多个工具之间没有依赖关系时，串行执行是浪费时间，并行调用才是高手的节奏。这就是乐高玩法的第三条心法。

并行调用的核心操作叫扇出（Fan-out）：拿到用户请求后，编排层把它同时广播给多个工具。比如用户问“帮我安排这周末的短途游”，你可以同时调用天气查询工具、火车票查询工具、景点信息工具——三个工具谁也不依赖谁，谁先出结果都不影响别人。但并行不是无限开的，还得配上超时与降级：某个工具卡住了，超过 2 秒没返回，就先跳过它，不让一颗老鼠屎坏了一锅汤。同时要设置并发上限，就像家里电闸不能同时开所有大功率电器，防止下游 API 被瞬间请求冲垮。

并行带来的性能提升不是线性的，因为每多一个并行任务，通讯和调度都会产生额外开销。Amdahl 定律精确描述了这种加速比上限：

这就像一个乐高工厂：四条流水线同时开工，比一条流水线串行快得多，但每条流水线之间的调度、传送带衔接始终有固定耗时。所以聪明的搭建者会先评估“哪些墙能同时砌”，把并行路数控制到恰到好处——既充分利用了时间，又不至于让调度开销吃掉所有收益。

结果聚合策略

四面墙分别搭好后，最激动人心的时刻到了——把墙体合拢，城堡封顶。工具编排中的结果聚合，就是把这些从不同工具返回的零散结果，拼成一个对用户有用的完整答案。

聚合有三大策略。全量汇合最稳妥：等所有并行工具全部返回，再按模板统一整理。比如旅行规划，等天气、车票、景点三个结果都到齐，一起生成一份完整行程单。竞速取优是急性子首选：多个工具做同样的事（比如两个翻译引擎），谁先返回一个高质量结果，就用谁的结果，其余直接丢弃——像比赛跑步，只认冠军。增量流式则是“边拼边展示”：工具返回一部分数据，就立刻把这一部分展示给用户。比如搜索推荐，先展示第一批结果，后续结果慢慢补充，用户不用干等。

把聚合策略写成公式，就是一个把 k 个并行结果合并成最终响应的归并函数：

就像拼完的乐高城堡，你不会把一堆散落的塔身、墙壁、城门直接递给朋友，而会先统一组装、检查咬合，再交付一个完整成品。结果聚合就是那个“最后一步合拢”的动作——它决定了你的 Agent 给用户的是“零件堆”还是“完整城堡”。

测试调试方法

乐高拼完不是终点，还要摇一摇、推一推，看城堡会不会散架。工具编排同样如此：多个工具组合之后，问题往往不在单个工具内部，而在工具之间的衔接处。于是，前三节讲的组合模式、数据流转、格式转换，全部需要一套专门的测试与调试方法来守门。

第一个武器叫组合测试矩阵：把每两个可能相邻的工具作为一对，逐一测试它们的连接是否顺畅。就像把每两块乐高都试着拼一下，用力拉扯看会不会脱开。第二个武器叫链路日志：给每一次编排执行分配一个唯一的 trace_id，从用户请求进来，到每个工具执行完毕，每一步的输入输出都记录在案。出了问题，顺着日志链路走一遍，一眼就能看出是哪块积木在哪个环节“咔哒”一声脱了钩。第三个武器是可观测性监控：不只测一次，而是持续观察线上运行的组合体，统计每个组合结点的耗时、失败率、数据流量，形成一张实时仪表盘。

组合测试矩阵里，每个“连接对”的通过与否可以形式化为一个二值矩阵 M。行代表上游工具 i，列代表下游工具 j，M(i,j)=1 表示这个组合通过测试：

拼乐高时，我们最担心的不是某块砖坏了，而是两块砖接缝处咬不紧。工具编排的测试调试，本质上就是在逐条检查这些“接缝”。当你把组合测试矩阵填满 1、链路日志随时可查、仪表盘一目了然，你就拥有了一个像搭积木一样稳当又灵活的 Agent 系统——搭得快，散不了架。

AI 应用场景

Python 演示

import timefrom concurrent.futures import ThreadPoolExecutor, as_completedfrom dataclasses import dataclass, fieldfrom typing import Callable, Dict, List, Any  @dataclassclass ToolResult:    """原子工具执行后写入上下文的结果包裹"""    data: Dict[str, Any]    latency_ms: float = 0.0  def tool(name: str):    """把普通函数注册为一个原子工具，附带工具元信息"""    def decorator(fn: Callable) -> Callable:        fn.tool_name = name  # type: ignore        return fn    return decorator  class ToolPipeline:    """管道流水线：按顺序执行一串工具，前一个输出喂给下一个"""     def __init__(self, steps: List[Callable]):        self.steps_list = steps     def run(self, ctx: Dict[str, Any]) -> Dict[str, Any]:        for step_fn in self.steps_list:            result = step_fn(ctx)            ctx.update(result.data)        return ctx  class FanOutGate:    """扇出分发：把同一份上下文并行广播给多个独立工具"""     def __init__(self, branches: List[Callable], timeout: float = 3.0):        self.branch_list = branches        self.timeout = timeout     def run(self, ctx: Dict[str, Any]) -> List[Dict[str, Any]]:        results = []        with ThreadPoolExecutor(max_workers=len(self.branch_list)) as pool:            futures = [pool.submit(fn, ctx) for fn in self.branch_list]            for future in as_completed(futures, timeout=self.timeout):                results.append(future.result().data)        return results  class Aggregator:    """汇聚合并：把多个并行结果按自定义规则合并成一份输出"""     def __init__(self, merge_fn: Callable[[List[Dict[str, Any]]], Dict[str, Any]]):        self.merge_fn = merge_fn     def run(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:        return self.merge_fn(results)  @tool("查天气")def fetch_weather(ctx: Dict[str, Any]) -> ToolResult:    time.sleep(0.2)    return ToolResult(data={"weather": "多云", "temp": 24})  @tool("查高铁")def fetch_train(ctx: Dict[str, Any]) -> ToolResult:    time.sleep(0.3)    return ToolResult(data={"train": "G7301", "depart": "08:30"})  @tool("查景点")def fetch_spot(ctx: Dict[str, Any]) -> ToolResult:    time.sleep(0.1)    return ToolResult(data={"spot": "西湖", "open": "06:30"})  def merge_plan(parts: List[Dict[str, Any]]) -> Dict[str, Any]:    """把三个并行结果合并成一份完整行程单"""    plan = {}    for part in parts:        plan.update(part)    plan["summary"] = "08:30 高铁出发，西湖游览，注意带伞"    return plan  def main():    # 1. 管道流水线：先查天气，再根据天气生成穿衣建议    def dress_advice(ctx: Dict[str, Any]) -> ToolResult:        if "晴" in ctx.get("weather", ""):            advice = "短袖"        else:            advice = "外套"        return ToolResult(data={"advice": advice})     pipeline = ToolPipeline([fetch_weather, dress_advice])    ctx_out = pipeline.run({})    print("管道输出:", ctx_out)     # 2. 扇出 + 聚合：并行三个查询，再合拢成出行计划    gate = FanOutGate([fetch_weather, fetch_train, fetch_spot])    parallel_data = gate.run({})    print("并行结果:", parallel_data)     agg = Aggregator(merge_plan)    final_plan = agg.run(parallel_data)    print("聚合计划:", final_plan)  if __name__ == "__main__":    main()

1. def tool(name: str)一个装饰器，给普通函数打上“工具”标签，相当于给乐高积木贴上型号标签，方便编排层识别。
2. self.steps_list = steps流水线的步骤清单，记录先拼哪块、后拼哪块，对应乐高说明书的步骤顺序。
3. ctx.update(result.data)把每个工具的输出去写入共享上下文，后面的工具就能读到前面工具的结果，如同料盒在流水线上传递。
4. ThreadPoolExecutor(max_workers=len(self.branch_list))创建与分支数等量的线程池，实现扇出并行，相当于多条流水线同时开工。
5. as_completed(futures, timeout=self.timeout)等待所有并行任务完成，超时则丢弃未完成结果，对应超时降级策略。
6. plan.update(part)将多个并行返回的字典合并成一个大的字典，实现全量汇合聚合。
7. merge_plan用户自定义的聚合规则函数，把“拼好的零件”组装成完整城堡（出行计划）。
8. pipeline.run({})管道模式的实际调用入口，注意传入的是一个空字典——所有中间数据都在工具之间自动流转。
9. Aggregator(merge_plan)把聚合规则封装成对象，方便复用不同策略（全量、竞速、流式）。
10. parallel_data扇出后收集的所有结果列表，它是聚合器的输入，也是“组装台”上的零件。

常见误区

面试题