Agent工作流：工具集成Tool Integration

可视化演示

观察MCP框架下Agent调用外部工具、结构化解析结果，以及多工具依赖图拓扑排序编排的过程，可切换三种模式。

工具集成概述

回到我们的办公室。小智第一天报到时，桌面上只有一支笔和一张纸——他能写出漂亮的方案，却查不了明天的航班、调不出上季度的销售数据。作为主管的你很快意识到：光有聪明的大脑远远不够，得给他配装备。

在AI领域，这个“配装备”的动作就叫工具集成。一个纯粹的LLM（大语言模型）本质上是个“文本预测器”，它只擅长读完上文写下文。一旦涉及实时信息、精确计算、外部系统操作，它就会露怯：3月27日的天气它不知道，23×47它可能算错，发一封真实邮件它更做不到。工具集成把这些能力以外挂的形式交给Agent，让模型输出“调用指令”，再由代码真正执行。

从技术架构看，工具集成可以划分为几个层次：最底层是“函数调用协议”（模型怎么表达“我想用某件装备”），中层是“工具注册”（把装备信息登记成模型能看懂的目录）与“结果解析”（把装备返回的原始数据翻译回模型能懂的文本），上层是“多工具编排”（多个装备协同工作）与“错误处理”（装备失灵时的应急预案）。

$$\displaystyle \mathcal{T} = \{T_1, T_2, \ldots, T_n\}, \quad T_i = (name_i, desc_i, schema_i)$$

1. 工具集合：$\mathcal{T}$ 表示 Agent 可调用的全部工具构成的“工具箱”，整体记为集合。

$$\mathcal{T} = \{T_1, T_2, \ldots, T_n\}$$

2. 单个工具三元组：每个工具 $T_i$ 由名称、功能描述、参数规范三个要素构成，就像一件装备有名字、说明书和操作手册。

$$T_i = (name_i, desc_i, schema_i)$$

3. 下标与总数：$i$ 是工具在注册表中的序号，$n$ 是工具总数，$n$ 越大，Agent 的“武器库”越丰富。

把小智的装备清单写下来，就得到了一个工具集合。每一件装备都有名字（比如“天气查询”）、用途说明（“查询某城市当日天气”）和操作规范（“需要一个城市名字符串参数”）。有了这份清单，下一步就是让模型学会“阅读说明书并下达调用指令”。

函数调用协议

装备配齐了，可小智该怎么“使用”它们呢？总不能让他直接动手拆开电话机吧——他一个AI模型，本质上没有手。解决办法是：让模型用自然语言写一张“工单”，描述他想用哪件装备、传什么参数，然后由真正的代码去执行。

这就是函数调用协议（Function Calling）的核心思想：LLM不直接执行任何外部操作，它只输出一个结构化的调用请求。以OpenAI的function calling为代表，模型会在正常的回复之上，额外生成形如 {"name": "get_weather", "arguments": {"city": "beijing"}} 的结构化指令。这里的 arguments 必须是合法的JSON对象，键名要与工具注册时的参数定义严格一致。

协议本身并不复杂，但它解决了一个根本问题：把模型的“意图”翻译成机器可执行的“指令”。没有这个翻译层，模型只能“说”不能“做”；有了它，小智每想用一个装备，就写一张工单交给执行员，干净利落。协议还催生了“流式输出”等多种演进形态——模型可以一边思考一边生成多张工单，为后续的多工具编排埋下伏笔。

$$\displaystyle \text{LLM}(P, \mathcal{T}) \rightarrow \{ \text{name}, \text{arguments} \}$$

1. 输入上下文：$P$ 是用户的问题或当前对话内容，比如“北京今天冷吗？”。

$$P = \text{用户输入}$$

2. 参考工具集：$\mathcal{T}$ 是已注册的工具集合，模型会逐个“翻阅”这些工具的说明书来决定调用谁。

$$\mathcal{T} = \{T_1, T_2, \ldots, T_n\}$$

3. 输出调用指令：$\{ \text{name}, \text{arguments} \}$ 是模型输出的结构化对象——选中的工具名 + 参数表。整个过程模型只负责“写工单”，不负责“干体力活”。

$$\{ \text{name}, \text{arguments} \}$$

协议让小智从“想做事”跨到了“下指令”。下一节我们来看，那份被模型翻阅的“装备说明书”——工具注册机制——到底长什么样。

{  "name": "get_weather",  "arguments": {"city": "beijing"}}

工具注册机制

小智怎么知道办公室里有哪些装备、每件装备怎么用？靠的是一份装备登记表——我们把每件工具的信息结构化地告诉模型。这个过程在工程上叫工具注册（Tool Registration）。

每件工具注册时，要填写一张“说明书卡”，核心是一段JSON Schema。它告诉模型三件事：这个工具接受什么类型的参数（type）、每个参数叫什么名字、有什么含义（properties）、哪些参数是必须提供的（required）。描述文本（description）也极其关键——它的质量直接决定模型能否在关键时刻想起“还有这么一件装备可用”。

工程上，所有注册工具会被集中管理在一个“注册表”里。这个注册表既可以是代码里的一个数组，也可以是动态的服务——支持在运行期间增删工具（动态注册），比如管理员临时给Agent加一个“查快递”的能力。注册表越规范，模型的选择越准确；反之，如果说明书写得含糊，模型就会像小智面对一台没有标签的陌生机器一样不知所措。

$$\displaystyle schema_i = \{ \text{type}, \text{properties}, \text{required} \}$$

1. 整体类型：type 声明参数对象的结构类型，通常是 object。

$$type = \text{object}$$

2. 属性定义：properties 逐个描述每个参数的名称、类型和含义，相当于说明书里的“操作项”。

$$properties = \{ \text{city}: \{type: string\} \}$$

3. 必填约束：required 是一个数组，列出调用该工具时必须给出的参数名；缺了它，工具执行就会报错。

$$required = [\text{city}]$$

装备登记完毕，说明书也写好了。但小智真正把工具用起来，还要学会一件事：看懂工具返回的“执行结果报告单”。

{  "name": "get_weather",  "description": "查询指定城市的当前天气",  "parameters": {    "type": "object",    "properties": {"city": {"type": "string", "description": "城市名，如beijing"}},    "required": ["city"]  }}

结果解析处理

工具执行完毕，外部系统会吐出一堆杂乱的数据。比如天气API返回的可能是一段带状态码、时间戳、多层嵌套的JSON；数据库查询返回的是原始行集；计算器返回的则是一个浮点数。这些原始输出不能直接塞回对话上下文——模型需要的是简洁、准确、可理解的信息。于是就有了结果解析（Result Parsing）这一环。

工程实践上，结果解析通常分两步：第一步是格式正规化，把各种来源的返回值统一转成标准JSON；第二步是语义提取，从JSON中摘出关键字段，去掉噪音。比如天气接口返回 {"code":200, "data": {"temp": 25, "cond": "sunny"}}，解析器会提取成一句“北京当前25℃，晴天”再交给模型。

解析环节还有一个容易被忽视的职责：内容校验与异常剔除。外部系统的数据可能超长、字段缺失、甚至格式错误。一个合格的解析器会做长度截断、缺省值填充、类型转换，并在数据不可用时返回一个明确的错误标记，而不是把乱糟糟的原始堆栈抛给模型。这就像小智拿到一份写得乱七八糟的设备报告，他自己先整理成“一页纸摘要”再向领导汇报。

$$\displaystyle r_{parsed} = \text{Parse}(f_{exec}(name, args))$$

1. 工具执行：$f_{exec}(name, args)$ 表示真正调用外部函数或API，按照注册表里的函数名和参数执行。

$$f_{exec}(name, args) = \text{执行结果}$$

2. 原始返回：这一步得到的是外部系统的原生态输出，可能包含大量冗余信息。

$$\text{raw output}$$

3. 解析动作：Parse 把原始输出清洗、提取、标准化为模型可用的紧凑结果。

$$r_{parsed} = \text{标准化结果}$$

单件工具的“看报告”问题解决了，可真实任务往往不止用一件装备。下一节我们看小智如何同时指挥多台设备协同作战。

多工具编排

现实中的任务很少能靠一件工具完成。用户问“北京现在多少度？比上海热吗？”——这需要先调天气API查北京，再查上海，最后对比。小智必须学会多工具编排（Multi-tool Orchestration）：把一个大任务拆解成多个工具调用，并组织好先后顺序。

最简单的编排是顺序链式调用：工具A的输出直接作为工具B的输入。例如先调用“城市定位工具”拿到经纬度，再把经纬度传给“天气查询工具”。更复杂的场景会出现并行分支：多个相互独立的工具可以同时调用，互不等待。而最智能的编排方式是ReAct模式——模型在“思考(Reason)”与“行动(Act)”之间循环：推理下一步需要什么信息→调用工具→观察结果→继续推理，直到完成目标。

$$\displaystyle s_t = s_{t-1} \oplus [ \text{Action}_t, \text{Observation}_t ]$$

1. 历史状态：$s_{t-1}$ 是截止上一步 Agent 积累的全部上下文，包括对话历史、之前的工具结果。

$$s_{t-1} = \text{历史上下文}$$

2. 当前决策：$\text{Action}_t$ 是模型在第 t 轮决定调用的工具及其参数，也就是一张新工单。

$$\text{Action}_t = \{name, arguments\}$$

3. 观察回填：$\text{Observation}_t$ 是工具返回并解析后的结果，被追加进上下文，供下一轮推理使用。

$$s_t = s_{t-1} + \text{Action}_t + \text{Observation}_t$$

这套“决策→行动→观察→再决策”的循环，让小智像一位经验丰富的项目经理：每完成一步，就更新项目看板，再决定下一步派谁上场。但设备再多、流程再顺，也架不住“意外”——下一节，我们聊聊应急预案。

错误处理策略

设备越多，出故障的概率越高。小智打电话时对方占线、查数据库时服务超时、调用支付接口时返回签名错误……任何一环失败，整个任务都可能中断。因此错误处理（Error Handling）是工具集成从demo走向生产的必答题。

工程上有一套成熟的“错误处理阶梯”。第一级是超时与重试：给每次调用设置超时时间（比如5秒），超时后自动重试1-3次，指数退避（第一次等1秒，第二次等2秒，第三次等4秒）。第二级是失败降级：主工具失败时切换备选工具，比如主天气服务挂了，就用备用天气API顶上。第三级是异常分类：把错误分成“可重试”（网络超时、限流）与“不可重试”（参数错误、权限不足），前者自动重试，后者直接向模型汇报错误原因，让模型调整策略或请求用户补充信息。

$$\displaystyle P_{succ}^{(k)} = 1 - (1 - p)^k$$

1. 单次成功率：$p$ 表示单次工具调用成功的概率，它由服务稳定性和网络状况决定。

$$p = \text{单次成功率}$$

2. 连续失败概率：$(1 - p)^k$ 是连续 k 次调用都失败的概率，随着 k 增大而指数级减小。

$$(1-p)^k = \text{全失败概率}$$

3. 整体成功率：$P_{succ}^{(k)}$ 是经过 k 次重试后至少有一次成功的概率，体现了“重试提升可靠性”的量化效果。

$$P_{succ}^{(k)} = 1 - (1-p)^k$$

有了这套应急预案，小智就算遇上设备故障，也不会当场愣住——该重试的重试、该换备胎的换备胎、该上报的上报。不过，装备再多，也得算算成本。最后一节看生产环境中的“精打细算”。

生产实践优化

装备配齐、流程跑通之后，老板开始关注账本了。每次工具调用都有成本：LLM的token费用、外部API的调用费、超时重试消耗的时间……大规模生产环境下，成本与性能优化成为工具集成能否持续运行的关键。

最常见的优化手段是缓存：对相同参数的查询结果做内存或Redis缓存，命中缓存就不需要再调外部API。比如“北京天气”一天内被问20次，第一次实调、后19次直接读缓存。其次是并发与限流：独立的工具调用并行发出，缩短总响应时间；同时给每个外部服务设置速率上限，避免触发对方限流策略。再配合监控告警：记录每次调用的成功率、延迟、花费，设置阈值告警，让工程师能第一时间发现某个工具“烧钱”或“变慢”。

$$\displaystyle C_{total} = \sum_{j=1}^{m} N_j \times c_j$$

1. 分类调用次数：对第 $j$ 个工具统计它的调用次数 $N_j$，不同工具的使用频率差别可能很大。

$$N_j = \text{工具j的调用次数}$$

2. 单次成本：$c_j$ 是第 $j$ 个工具每次调用的费用，比如一次天气API调用0.01元。

$$c_j = \text{第j个工具的单次成本}$$

3. 总成本汇总：每个工具的花费累加，得到整体成本。这个公式是“成本优化”的起点——先量化，再优化。

$$C_{total} = \sum_{j=1}^{m} N_j \times c_j$$

从配装备、写说明书、看反馈、搞协同、做预案再到算成本，小智终于从“纸上谈兵”进化成了“能打硬仗”的全能助理。工具集成这门手艺，让每个Agent开发者都能亲手完成这次进化。

AI 应用场景

Python 演示

import jsonimport timefrom typing import Any, Callable, Dict # The mock weather tool, pretending to call an external APIdef get_weather(city: str) -> Dict[str, Any]:    weather_map = {"beijing": "sunny 25C", "shanghai": "rainy 22C"}    return {"city": city, "weather": weather_map.get(city, "unknown")} # A simple calculator tooldef calculate(expression: str) -> float:    return eval(expression) # The "tool registry", mapping names to real functionsTOOL_REGISTRY: Dict[str, Callable] = {    "get_weather": get_weather,    "calculate": calculate,} # The "instruction manuals" that would be sent to the LLMTOOL_SCHEMAS = [    {        "name": "get_weather",        "description": "Query current weather of a city",        "parameters": {            "type": "object",            "properties": {"city": {"type": "string"}},            "required": ["city"],        },    },    {        "name": "calculate",        "description": "Evaluate a math expression",        "parameters": {            "type": "object",            "properties": {"expression": {"type": "string"}},            "required": ["expression"],        },    },] def execute_tool_call(call: Dict[str, Any]) -> Dict[str, Any]:    """Execute a single function-call request from the LLM."""    name = call["name"]    args = call.get("arguments", {})    if name not in TOOL_REGISTRY:        return {"error": f"Unknown tool: {name}"}    try:        result = TOOL_REGISTRY[name](**args)        return {"tool": name, "result": result}    except Exception as exc:        return {"tool": name, "error": str(exc)} def run_agent_loop(user_question: str) -> None:    """Simulate an agent loop: LLM decides which tool to call."""    # In a real app this JSON would come from the LLM response    mock_llm_calls = [        {"name": "get_weather", "arguments": {"city": "beijing"}},        {"name": "calculate", "arguments": {"expression": "25 * 4 + 10"}},    ]    print(f"User: {user_question}")    for call in mock_llm_calls:        print(f"LLM asks -> {call['name']}({call['arguments']})")        response = execute_tool_call(call)        print(f"Tool returns: {json.dumps(response, ensure_ascii=False)}")        time.sleep(0.3) if __name__ == "__main__":    run_agent_loop("How is the weather in beijing and what is 25*4+10?")

1. TOOL_REGISTRY: Dict[str, Callable] = {...}这就是“装备登记表”，把工具名字映射到真实函数
2. TOOL_SCHEMAS = [...]这些字典是“使用说明书”，生产环境中会被随提示词一起发给LLM
3. def execute_tool_call(call: Dict[str, Any]) -> Dict[str, Any]统一的“工单执行入口”，所有调用都走这里
4. if name not in TOOL_REGISTRY:查不到工具的降级处理，避免程序直接崩溃
5. result = TOOL_REGISTRY[name](**args)用参数解包的方式调用函数，一行代码完成动态分发
6. except Exception as exc: return {"tool": name, "error": str(exc)}捕获异常并结构化成错误信息，方便LLM理解

常见误区

面试题