Agent 的手脚:工具调用

作者:toy


一、工具调用是什么:让模型伸手摸真实世界

从纯文本生成到”有手有脚”

大语言模型在工具调用出现之前,本质上是一台极其复杂的文字处理机器。你问它”今天北京的 PM2.5 是多少”,它会根据训练数据里的历史统计给你一个模糊的答案,而不是告诉你此刻的实测值。模型没有眼睛、没有手、没有网络连接,活在训练数据的时间截止点那一刻。

工具调用(Function Calling / Tool Use)打破了这个边界。它让模型不再只是”写字”,而是能够发出调用指令,访问搜索引擎、查询数据库、执行代码、发送邮件、控制外部 API。从架构视角看,工具是 Harness 给模型装上的”设备驱动”:Harness Engineering 概念里有一句经典比喻,”LLM = CPU,上下文窗口 = RAM,外部数据库 = 磁盘,工具 = 设备驱动”。没有设备驱动的 CPU 只能自言自语。

Function Calling 最初由 OpenAI 在 2023 年 6 月的 gpt-3.5-turbo 更新中引入,Anthropic 随后推出格式略有差异的 tool_use 机制。两家的核心设计哲学一致:模型不直接执行工具,而是通过结构化输出声明”我要调用哪个工具、传什么参数”,执行权留在调用方。这一设计既保留了模型的推理能力,又把危险操作的执行权交还给开发者控制的代码层。

工具调用的标准流程

一次完整的工具调用循环经历五个阶段。用户发起请求,模型分析需求后决策是否需要工具、需要哪个工具;模型以结构化格式输出工具调用请求(而非直接回答);调用方代码层执行对应工具并收集结果;结果被注入到对话上下文;模型基于工具结果继续生成最终回答。

这个循环在 Harness 里被称为 TAO(Thought-Action-Observation)。模型的每一轮推理是 Thought,工具调用请求是 Action,工具执行结果是 Observation,然后进入下一轮 Thought。一次复杂任务可能经历十几轮 TAO 循环,每一轮模型都要判断:现有信息是否足够,还是需要再调一个工具?

关键细节:模型在每一步只看到被注入上下文的工具结果,看不到工具的内部实现。工具的安全性、鉴权逻辑、异常处理全部在调用方代码里,模型无法绕过,这是工具调用安全模型的核心支撑点。

工具调用不是简单的 API 封装

把工具调用理解成”给 API 加一层包装”是常见的误解。真正的难点不在技术整合,而在模型决策的准确性:模型需要理解何时应该调用工具(而不是用训练知识回答)、调用哪个工具(当有几十个工具时)、传什么参数(格式、类型、必填项是否齐全)。

三个问题任何一个出错,工具调用就失败。模型以为自己知道答案而跳过工具调用,会产生幻觉;工具选错,会执行无关操作;参数传错,工具执行报错,触发后续的错误处理链路。

工具定义的质量直接决定模型的决策准确性。工具的 name、description、parameters 的 JSON Schema,都是模型用来判断”这个工具是干什么的,我应不应该用它”的依据。描述写得含糊,模型就可能选错工具或漏掉必填参数。这是一个软件接口设计问题,不是提示词工程问题。工具定义就是工具对外暴露的接口规范,应当像设计和维护 REST API 文档一样认真对待它。


二、Function Calling 的实现细节

OpenAI 的消息格式

OpenAI 的工具调用围绕三个核心字段展开。请求体里的 tools 数组定义可用工具列表,每个工具是一个 JSON 对象,包含 type(通常是 "function")、function.namefunction.descriptionfunction.parameters(JSON Schema 格式)。tool_choice 字段控制工具选择策略:"auto" 让模型自主决定是否调用,"required" 强制模型必须调用至少一个工具,也可以传对象 {"type": "function", "function": {"name": "具体函数名"}} 强制指定工具。

当模型决定调用工具时,响应体里的 choices[0].message.tool_calls 是一个数组,每个元素包含三个字段:id(唯一标识这次调用,后续 tool 角色消息用它关联结果)、function.name(工具名)、function.arguments(JSON 字符串,注意不是对象)。调用方把 argumentsjson.loads() 解析后执行工具,再把结果用 role: "tool" 的消息带上对应的 tool_call_id 拼回对话。

Responses API(2025 年推出)与 Chat Completions API 在工具调用格式上有细节差异,但核心数据结构相同。Responses API 默认对所有函数调用启用 strict mode。

Anthropic 的格式差异

Anthropic 的格式在概念上相同,但命名和结构不同。请求里工具定义用 tools 数组,每个工具包含 namedescriptioninput_schema(对应 OpenAI 的 parameters)。模型的工具调用响应出现在 content 数组里,以 type: "tool_use" 的块存在,包含 idnameinput(已是对象,不是字符串,这一点与 OpenAI 不同)。

回传工具结果时,Anthropic 用 role: "user" 的消息(不是 "tool" 角色),消息的 content 是一个数组,包含 type: "tool_result" 的块,每个块携带 tool_use_id(关联工具调用)、content(结果文本)、以及可选的 is_error: true(标记执行失败)。

两者最实质的差别在并行工具调用时的 tool_result 拼接方式(详见第三章),以及 input 字段的类型差异(Anthropic 直接给对象,OpenAI 给字符串,调用方需要额外 json.loads)。

工具定义的最佳实践

工具的 JSON Schema 是模型理解工具能力的唯一依据。几个实践原则能显著提升调用准确率。

name 要动宾结构、语义明确。get_weatherweather 好,search_product_catalogsearch 好。当工具数量超过 5 个时,相似名称的工具会混淆模型,清晰的动宾命名可以减少误选。

description 是最重要的字段。要说清楚工具做什么、什么时候应该用它、什么情况不应该用它。尤其是后者,告诉模型”当用户只需要历史数据时不要调用此工具”,比单纯描述功能更能减少不必要的调用。

parameters 里的每个字段都应该有 description。枚举字段要列出所有合法值。可选字段要有 default 值说明。OpenAI 的 strict mode 要求所有字段必须在 required 中列出,可选字段用 type: ["string", "null"] 表示。Anthropic 提供 strict: true 的 strict tool use 模式达到同样效果:从根本上消除无效工具调用,保证 tool input 与 JSON Schema 完全一致。

官方建议单次对话注册的工具数量保持在 20 个以下(OpenAI 的 soft limit)。超过这个数量时,模型从大量工具里选择正确工具的准确率会下降。

大量工具时的路由问题

当工具数量超过 20 个(例如企业级 Agent 集成了几十个内部服务),全量注入工具定义会消耗大量 token,同时降低模型选择准确率。这时需要工具路由层:在把工具定义发给模型之前,先用向量搜索或关键字匹配,根据用户输入从工具库里召回最相关的 5-10 个工具,再注入上下文。

向量搜索(把工具描述编码为向量,与用户请求做相似度匹配)适合语义相关的路由;关键字匹配(BM25 或全文搜索)适合有明确关键词的路由。两者结合(倒排+向量的混合检索)是生产环境的常见方案。工具路由本质上是一个小型 RAG 问题,工具定义的质量同样决定召回质量。

工具路由引入了一个新的失败模式:路由层没有召回用户真正需要的工具。这时模型看到的工具集里根本没有能完成任务的工具,它要么用不相关的工具硬凑答案,要么告知用户”我无法完成这个任务”。监控工具路由的召回质量(路由层实际召回的工具是否包含了模型后来使用的工具)是路由层必须配套的可观测性指标。

完整代码示例:工具注册与调用循环

# 完整的 OpenAI 工具调用循环
# 演示:注册工具、接收 tool_call、执行并回传结果
import json
import openai

client = openai.OpenAI()

# 1. 工具实现(实际中这里是真实业务逻辑)
def get_current_weather(location: str, unit: str = "celsius") -> str:
    # 模拟调用气象 API
    return json.dumps({"location": location, "temperature": 22, "unit": unit})

def search_database(query: str, table: str) -> str:
    # 模拟数据库查询
    return json.dumps({"results": [{"id": 1, "content": f"查询 '{query}' 在 {table} 的结果"}]})

# 工具函数注册表
TOOLS = {
    "get_current_weather": get_current_weather,
    "search_database": search_database,
}

# 2. 工具定义(JSON Schema),description 是关键
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取指定城市的实时天气信息。仅当用户需要当前/今天的天气时调用,不适用于历史或预报数据。",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市名称,如'北京'或'上海'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位,默认 celsius(摄氏度)"
                    }
                },
                "required": ["location"],
                "additionalProperties": False  # strict mode 要求
            },
            "strict": True
        }
    }
]

def run_tool_loop(user_message: str) -> str:
    """带工具调用的完整对话循环"""
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )

        message = response.choices[0].message
        messages.append(message)

        # 判断是否需要调用工具
        if response.choices[0].finish_reason != "tool_calls":
            return message.content  # 模型直接回答,对话结束

        # 执行所有工具调用并收集结果
        for tool_call in message.tool_calls:
            fn_name = tool_call.function.name
            fn_args = json.loads(tool_call.function.arguments)

            try:
                result = TOOLS[fn_name](**fn_args)
            except Exception as e:
                result = f"工具执行错误: {str(e)}"

            # 把工具结果用 tool 角色拼回对话
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

        # 循环继续,让模型基于工具结果生成下一步

三、并行工具调用:效率提升还是混乱来源

parallel_tool_calls 开关

单轮对话里注册多个工具,模型可以在一次响应里同时发起多个工具调用,而不是等第一个调用的结果回来再决定调第二个。这在查询多个独立数据源时能显著减少延迟。

OpenAI 的 parallel_tool_calls 参数默认值为 true,模型在单次响应中可以返回包含多个 tool call 对象的数组。设置为 false 可确保每次调用中最多只触发一个工具,适合有严格执行顺序要求的场景。

Anthropic 的设计略有不同。Claude 在单次 stop_reason=tool_use 的响应中可以输出多个 tool_use block,每个 block 带独立的 idnameinput。禁用并行的方式是在 tool_choiceauto 时设置 disable_parallel_tool_use=true

两家的关键差异体现在度量上。Anthropic 官方文档提供了”每条消息平均工具调用数”指标:如果这个数值 >1.0,说明并行工具调用确实在生效。如果模型一直在串行调用,可以通过在系统提示里显式注入并行指令来引导。OpenAI 没有提供对应的度量指标。

并发执行的正确实现

并行工具调用的最大价值在执行层,不在模型层。模型可以一次性输出三个工具调用请求,但如果调用方代码用串行 for 循环执行,实际时间节省为零。必须在执行层用 asyncio.gather(Python)或 Promise.all(JavaScript)并发发出请求。

Anthropic 的关键约束:所有工具结果必须合并到同一条 user 消息的 content 数组中。如果分多条消息分别返回,Claude 会退化为串行调用。这个约束在代码里容易被忽略,导致并行效果完全失效。

# Anthropic 并行工具调用 - asyncio 并发执行
# 来源:改编自 Anthropic 官方文档 parallel-tool-use
import asyncio
import anthropic

client = anthropic.Anthropic()

async def get_weather(location: str) -> str:
    await asyncio.sleep(0.1)  # 实际中调用外部 API
    return f"{location}: 22°C, 晴"

async def get_stock_price(symbol: str) -> str:
    await asyncio.sleep(0.1)
    return f"{symbol}: $185.20"

async def execute_parallel_tools(tool_uses: list) -> list:
    """并发执行所有工具调用,消除串行延迟"""
    tasks = []
    for tool_use in tool_uses:
        if tool_use.name == "get_weather":
            tasks.append(get_weather(tool_use.input["location"]))
        elif tool_use.name == "get_stock_price":
            tasks.append(get_stock_price(tool_use.input["symbol"]))

    results = await asyncio.gather(*tasks)

    # 关键:所有 tool_result 必须合并到同一条 user 消息
    # 分多条消息返回会让 Claude 退化为串行调用
    return [
        {
            "type": "tool_result",
            "tool_use_id": tool_use.id,
            "content": result
        }
        for tool_use, result in zip(tool_uses, results)
    ]

async def main():
    tools = [
        {
            "name": "get_weather",
            "description": "获取指定城市的天气",
            "input_schema": {
                "type": "object",
                "properties": {"location": {"type": "string"}},
                "required": ["location"]
            }
        },
        {
            "name": "get_stock_price",
            "description": "获取股票价格",
            "input_schema": {
                "type": "object",
                "properties": {"symbol": {"type": "string"}},
                "required": ["symbol"]
            }
        }
    ]

    messages = [{"role": "user", "content": "北京的天气和 AAPL 的股价分别是多少?"}]

    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        tools=tools,
        messages=messages
    )

    if response.stop_reason == "tool_use":
        tool_uses = [b for b in response.content if b.type == "tool_use"]
        print(f"Claude 发起了 {len(tool_uses)} 个并行工具调用")

        tool_results = await execute_parallel_tools(tool_uses)

        # 续接对话:assistant 响应 + 所有结果在同一 user 消息中
        messages.extend([
            {"role": "assistant", "content": response.content},
            {"role": "user", "content": tool_results}  # 单条消息包含所有结果
        ])

        final = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )
        print(final.content[0].text)

asyncio.run(main())

竞争条件与依赖关系

并行工具调用的危险场景:模型一次性发起两个工具调用,但这两个工具实际上存在依赖关系,工具 B 需要工具 A 的结果作为输入。这时并发执行会让工具 B 用空数据运行,产生错误结果。

Anthropic 的处理协议相对明确:如果批次中某个调用因前置依赖缺失而失败,只需在 tool_result 里返回 is_error: true 加自然语言错误信息(如 "Error: 需要先调用 get_user_id 获取用户 ID"),Claude 会自动识别依赖关系并在下一 turn 重新发起,无需应用层手动切换为串行模式。OpenAI 在文档里没有给出同等明确的依赖失败恢复协议,依赖检测需要在应用层自行处理。

哪些工具必须串行?涉及数据库写操作的工具(后者读取前者写入的数据)、有事务语义的操作序列、需要鉴权令牌的级联请求(先获取令牌再用令牌调用)。并行安全的工具:读取不同独立数据源的查询、互不依赖的计算任务、向不同 API 发出的并行搜索。

一个可防御的做法是在工具定义的 description 里显式声明依赖关系:"调用此工具前必须先调用 get_access_token,将令牌作为 token 参数传入"。这条描述对模型的推理有实质引导效果,能让模型在多数场景下主动串行。当模型仍然并行调用时,执行层检测到 token 字段为空,返回 is_error: true 加明确的错误说明,模型在下一轮修正调用顺序。两层结合,描述层引导加执行层兜底,比单独依赖任何一层都稳定。


四、参数纠错:当模型传错参数时

常见错误类型

模型在生成工具参数时会产生几类典型错误。类型不匹配最常见:工具期望整数,模型传了字符串 "5";工具期望布尔值,模型传了字符串 "true"。必填项缺失也很普遍,尤其是上下文信息不足时,模型没有收到用户的 location 信息,却仍然调用了需要 location 的工具,导致参数里缺失 location 字段。枚举值越界也会出现,模型传了 "fahrenheit " 而不是 "fahrenheit"(多了空格),或者传了不在枚举列表里的值。格式错误(日期格式、电话号码格式、URL 格式)是第四类。

这四类错误的共同点:都是模型在推理时产生的语义错误,不是网络错误或服务不可用。处理策略与暂态基础设施错误完全不同,不能做简单重试,而是要把错误信息明确反馈给模型,让模型重新生成修正后的参数。

三层参数校验架构

生产环境建议用三层架构处理参数问题。

第一层:Pydantic 类型强制转换。在工具函数入口用 Pydantic 模型做参数解析,Pydantic 会自动尝试类型转换(字符串 "5" 转成整数 5"true" 转成布尔 True),并对枚举值做规范化处理。这一层能处理大量轻微的格式问题,不需要触发模型重试。

第二层:校验失败时构建结构化错误信息。如果 Pydantic 校验失败,把错误详情格式化成清晰的自然语言,例如 "参数验证失败: location: 必填字段缺失; unit: 值 'kelvin' 不在合法枚举 [celsius, fahrenheit] 中"。错误信息要有指导性,让模型知道具体哪个字段出了什么问题。

第三层:is_error: true 反馈。将错误信息作为 tool_result 回传,设置 is_error: true。Anthropic 的 Claude 接收到标记了 is_error: true 的工具结果后,会自动尝试 2-3 次参数修正重试。OpenAI 的处理方式类似,模型会在后续 turn 里重新生成参数。

# 三层参数校验:Pydantic 类型强制 + 结构化错误反馈
import json
import pydantic
from typing import Optional

class WeatherInput(pydantic.BaseModel):
    """第一层:Pydantic 自动类型转换"""
    location: str
    unit: str = "celsius"  # 有默认值的可选参数

    @pydantic.validator('unit')
    def validate_unit(cls, v):
        v = v.strip().lower()  # 去除空格并转小写,修复常见格式问题
        if v not in ['celsius', 'fahrenheit']:
            return 'celsius'  # 枚举值不合法时使用默认值而非报错
        return v

def safe_weather_tool(raw_input: dict) -> tuple[str, bool]:
    """
    返回 (result, is_error) 元组
    第二层:结构化错误信息;第三层:is_error 标记
    """
    try:
        validated = WeatherInput(**raw_input)  # Pydantic 类型修正
        result = f"{validated.location}: 22°C"
        return result, False
    except pydantic.ValidationError as e:
        errors = e.errors()
        error_details = '; '.join(
            f"{'.'.join(str(x) for x in err['loc'])}: {err['msg']}"
            for err in errors
        )
        # 错误信息要有指导性,帮助模型知道如何修正
        error_msg = f"参数验证失败: {error_details}。请在下次调用中补充或修正这些参数。"
        return error_msg, True

def build_tool_result(tool_use_id: str, raw_input: dict) -> dict:
    """构建 Anthropic 格式的 tool_result"""
    content, is_error = safe_weather_tool(raw_input)
    result = {
        "type": "tool_result",
        "tool_use_id": tool_use_id,
        "content": content
    }
    if is_error:
        result["is_error"] = True  # Claude 接收后会尝试修正参数重新调用
    return result

strict mode 的根本性作用

OpenAI Responses API 默认对所有函数调用启用 strict mode,Chat Completions API 需显式设置 strict: true。strict mode 的约束是强性的:每个对象的 additionalProperties 必须为 false,所有字段必须在 required 中列出(可选字段用 type: ["string", "null"] 表示),$ref 中的定义必须内联。满足这些约束后,模型的结构化输出会与 JSON Schema 完全一致,从源头消除参数格式错误,不再需要依赖第二层、第三层的错误反馈循环。

Anthropic 的 strict tool use(strict: true)达到相同效果,同样保证 tool input 与 JSON Schema 完全一致。


五、代码解释器:最强大也最危险的工具

代码执行工具的特殊性

代码解释器(Code Interpreter)与其他工具有本质区别。其他工具执行的是预定义的业务逻辑,代码解释器执行的是模型实时生成的任意代码。工具的行为空间因此从”调用固定函数”扩展到”执行任意程序”,安全风险级别跨越了数个量级。

OpenAI 的 Code Interpreter 是托管式服务,在隔离容器里运行,容器若 20 分钟内无操作则自动过期(元数据保留但数据不可恢复),任何容器操作(上传/下载文件、API 读取)均会刷新计时器。内存限制可选 1g(默认)、4g、16g、64g,支持 30+ 种文件格式(CSV/JSON/XML/图像等)。E2B 代码沙箱在 Pro 层单个沙箱最长持续运行 24 小时,Hobby 层 1 小时。

实际应用场景覆盖数据分析(让模型直接对 CSV 文件进行统计分析)、图表生成(用 matplotlib/seaborn 生成可视化)、数学计算(数值求解、符号计算)、文件格式转换。这些场景的共同点:模型需要执行确定性的计算,而不仅仅是”描述”计算步骤。

沙箱隔离方案对比

执行不可信代码需要严格的隔离边界。当前主流方案有四种,隔离强度和性能开销各不相同。

Firecracker microVM 是 AWS Lambda 使用的方案,采用硬件虚拟化(KVM),每个工作负载运行独立 Linux 内核。启动时间 ≤125ms(从 InstanceStart API 到 guest userspace 启动),每个 VMM 进程内存开销 ≤5 MiB,单主机可以每秒创建最多 150 个 microVM,自 2018 年起在 AWS 生产环境运行,每月处理数万亿次函数调用。攻击面极小:每个 microVM 有独立内核,即使内核漏洞被利用也无法横向移动到其他 VM。

gVisor 是 Google Cloud Run/Functions 使用的方案,用用户态内核(Sentry 进程)通过 ptrace 或 KVM 拦截所有系统调用,用 Go 重新实现了 Linux 系统调用子集。性能开销按工作负载类型分类:CPU 密集型接近原生,系统调用密集型有 10-40% 额外开销,文件系统密集型有 30-80% 额外开销,网络密集型有 5-15% 额外开销。gVisor 启动比 Firecracker 更快(无 VM 引导过程),但隔离强度弱于 Firecracker,共享宿主内核,仍有被 kernel exploit pivot 的理论风险。

WebAssembly(WASM)运行在内存安全虚拟机中,没有 syscall 接口,所有与宿主的交互必须通过显式导入的 host functions 进行。Pyodide(CPython 通过 Emscripten 编译为 WASM)可在浏览器沙箱内运行 Python,支持 NumPy/pandas/SciPy 等包。WASM 与 Docker/gVisor 的核心区别:不是过滤系统调用,而是完全消除了系统调用接口,攻击面从”限制 syscall”降到”无 syscall”。适合浏览器端的代码执行场景,在服务端性能受限。

E2B / Daytona 这类商业沙箱产品基于 OCI 容器,冷启动时间 90-200ms,提供 API 接口管理沙箱生命周期,适合需要快速集成的场景。

对 LLM 生成代码的自动执行场景,Firecracker 是最安全的选择:硬件隔离边界完全阻断内核漏洞横向移动路径,而 gVisor 的共享内核架构仍有 pivot 风险。

代码执行结果的处理

代码执行结果有三种形态:stdout(标准输出,文本数据)、stderr(标准错误,调试信息和错误栈)、文件输出(图表、CSV、二进制文件)。

处理策略因形态而异。stdout 和 stderr 直接作为字符串拼入 tool_result。文件输出需要存储到临时位置,把 URL 或文件引用回传给模型,模型可在后续对话中引用文件。图表的处理方式是把图像编码为 base64,作为多模态消息内容回传(Anthropic 和 OpenAI 都支持在消息里内嵌 base64 图像)。

stdout 的长度要做截断保护。模型生成的代码可能输出几 MB 的数据,全量回传会撑爆上下文窗口。合理的上限是 10-50KB,超出部分截断并告知模型”输出已截断,共 X 行,显示前 Y 行”。

安全风险清单

代码执行工具的核心安全风险不在于执行本身,而在于执行边界的侵蚀。

代码注入:用户在请求里藏入控制指令,让模型生成恶意代码。缓解:沙箱是最终防线,不依赖模型去判断代码是否安全。资源耗尽:模型生成无限循环、递归炸弹、内存耗尽代码。缓解:强制 CPU 时间限制、内存硬上限、进程树监控。文件系统访问:代码读取宿主的敏感文件。缓解:沙箱文件系统隔离,只挂载工作目录。网络访问:代码向外泄露数据或作为跳板发动攻击。缓解:--unshare-net 断网隔离。依赖包安装:代码在执行时 pip install 恶意包。缓解:预构建包镜像,禁用网络后 pip 无法安装。


六、工具重试与失败处理

错误分类的重要性

工具调用的失败可以分为两个完全不同的类别,处理策略截然不同。

第一类是暂态基础设施错误:网络超时、服务临时不可用、限流(Rate Limit)、服务器内部错误。这类错误与请求内容无关,相同的请求稍后重试就可能成功。对应的 HTTP 状态码:429(限流)、500/502/503/504(服务端错误)、529(Anthropic 的过载状态码)。处理策略是指数退避重试。

第二类是语义错误:请求参数格式错误、必填项缺失、枚举值非法、权限不足、业务逻辑拒绝(如”该用户不存在”)。这类错误重试相同请求不会有不同结果。对应的 HTTP 状态码:400、401、403、404。处理策略是把错误信息反馈给模型,让模型修正参数或选择不同工具。

混淆这两类错误是常见工程失误:对 400 错误做指数退避重试,每次都会失败,白白消耗 token 配额和时间;对 429 错误直接报错给模型,模型没有理由认为是参数问题,可能做出错误的推断。

指数退避的正确实现

指数退避的核心是 Full Jitter 公式:delay = random(0, min(cap, base * 2^attempt))base 是初始等待时间(通常 1 秒),cap 是最大等待时间(通常 60 秒),attempt 是重试次数(从 0 开始)。random 引入随机性(jitter),防止多个客户端在同一时刻同时触发重试(雷群效应,thundering herd)。

两个必须遵守的规则。第一:优先读取 Retry-After 响应头。Anthropic 和 OpenAI 在 429 响应中均会返回 Retry-After 头,标明精确等待秒数。这个数值比退避公式更准确,必须优先使用。第二:设置最大重试次数上限(通常 5 次),达到上限后向上层传播异常,不能无限重试。

# 指数退避重试 + 错误分类
import time
import random
from typing import Any, Callable

# 可重试的暂态错误码
RETRYABLE_HTTP_CODES = {429, 500, 502, 503, 504, 529}
MAX_RETRIES = 5

def exponential_backoff_with_jitter(
    attempt: int,
    base_delay: float = 1.0,
    cap: float = 60.0
) -> float:
    """Full Jitter 退避公式,防止雷群效应"""
    return random.uniform(0, min(cap, base_delay * (2 ** attempt)))

def retry_tool_call(tool_fn: Callable, *args, **kwargs) -> Any:
    """带指数退避的工具调用包装器"""
    for attempt in range(MAX_RETRIES):
        try:
            return tool_fn(*args, **kwargs)
        except Exception as e:
            error_code = getattr(e, 'status_code', 500)

            # 语义错误直接抛出,让 LLM 自修正参数,不重试
            if error_code not in RETRYABLE_HTTP_CODES:
                raise

            if attempt == MAX_RETRIES - 1:
                raise  # 最后一次,向上传播

            # 优先读 Retry-After 响应头
            retry_after = getattr(e, 'retry_after', None)
            wait_time = float(retry_after) if retry_after else exponential_backoff_with_jitter(attempt)

            print(f"第 {attempt+1} 次重试,等待 {wait_time:.2f}s (错误码: {error_code})")
            time.sleep(wait_time)

熔断器:分层防护的最后一道门

指数退避处理的是单次工具调用的失败,熔断器(Circuit Breaker)处理的是工具持续性故障。当某个工具在一段时间内错误率超过阈值(如 5 分钟内失败率 >50%),熔断器进入 open 状态,停止向该工具发送请求,避免级联失败。经过一段冷却期后进入 half-open 状态,允许少量试探性请求,如果成功则恢复 closed 状态。

熔断器与退避重试形成分层防护:退避重试处理短时间的偶发故障,熔断器处理工具长时间不可用的场景。生产环境两者都需要,不能互相替代。

工具降级策略

当主工具失败时,降级到备选方案可以提升系统可用性。典型模式:搜索工具失败时降级到缓存数据;实时 API 失败时降级到本地知识库;高精度计算工具失败时降级到近似计算。

降级策略需要在工具定义层面就设计好。工具的 description 可以说明”此工具偶尔不可用,如果调用失败可以使用 search_local_cache 工具代替”,让模型在遇到 is_error: true 时自主选择降级路径,而不是无限重试或放弃任务。

降级设计还需要考虑用户感知。如果主工具失败后降级到精度更低的备用工具,最终回答应该透明地说明这一点(”实时数据暂时不可用,以下基于最近一次缓存数据,时间戳为 2026-05-29 08:00″),而不是用降级后的低质量数据假装是准确的实时回答。这是工具调用系统的诚实性要求,也是用户信任的基础。


七、权限白名单:最小权限原则

全权 Agent 的安全风险

一个能调用任意工具、访问任意文件、连接任意网络地址的 Agent 是安全噩梦。提示词注入(Prompt Injection)已被确认为 LLM 架构的基因级缺陷:系统提示词和用户输入在模型眼里是一锅粥,攻击者可以在工具的返回结果里藏入控制指令,让 Agent 执行未授权的操作。微软 Copilot 和 Google 日历的真实案例都验证了这个攻击路径的有效性:微软 Copilot 因白色字体隐藏指令被成功操控,危险评分达 9.3/10;Google 日历的邀请事件里藏入指令,成功操控了 IoT 设备。

更系统性的数据:91% 的企业已经在使用 AI Agent,其中 88% 报告了安全事件。这不是小概率事件,而是当前工具调用生态的普遍现状。

在这个威胁模型下,权限控制不能依赖模型的”判断力”,必须在执行层强制执行。模型决定”应该调用哪个工具”,权限系统决定”这个调用是否被允许执行”,两者独立。安全架构的核心原则:把模型视为不可信的决策方,把执行层的权限控制视为最终防线,不假设模型会做出”正确”的安全判断。

NVIDIA 最小权限四项策略

NVIDIA 在 Agent 安全实践指南里提出了四项策略性控制,覆盖了权限管理的核心维度。

Task-scoped access(任务范围访问):仅授予当前任务所需权限,而非所有潜在任务的权限。处理日志分析任务的 Agent 只需要读取日志目录的权限,不应该同时拥有发送邮件的能力。

Tool-level authorization(工具级别鉴权):每次工具调用独立鉴权,而不是在 Agent 启动时一次性授权所有工具。即使 Agent 已经通过身份验证,每次高风险操作仍需要单独鉴权检查。

Execution-time policy evaluation(执行时策略评估):在操作发起时基于当前上下文做授权决策,而不是在注册工具时静态声明。当前用户是谁、当前会话的上下文是什么、操作的目标资源是否在允许范围内,这些需要在执行时动态判断。

文件系统保护:绝对禁止 Agent 修改配置文件(.zshrc.gitconfig.ssh/、MCP config 等),出站网络只允许到显式白名单目的地。这是最容易被忽视、后果最严重的权限边界。

工具权限分级注册

把工具按操作风险分级注册,是权限管理的基础数据结构。

# 工具权限白名单注册与执行时鉴权
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, Set
import subprocess
import os

class RiskLevel(Enum):
    SAFE = 1       # 只读,无副作用
    MODERATE = 2   # 有限写,可撤销
    DANGEROUS = 3  # 不可逆,需审批

@dataclass
class ToolManifest:
    """每个工具的权限声明"""
    name: str
    risk_level: RiskLevel
    allowed_paths: Set[str] = field(default_factory=set)   # 允许访问的文件路径
    allowed_hosts: Set[str] = field(default_factory=set)   # 允许的出站网络目标
    requires_sandbox: bool = False                          # 是否需要沙箱隔离
    timeout_seconds: int = 30                               # 执行超时

class ToolRegistry:
    """工具白名单注册与执行时鉴权"""

    def __init__(self):
        self._registry: dict[str, ToolManifest] = {}
        # 绝对禁止修改的路径(来自 NVIDIA 安全指南)
        self._forbidden_paths = {
            os.path.expanduser("~/.zshrc"),
            os.path.expanduser("~/.gitconfig"),
            os.path.expanduser("~/.ssh"),
            ".cursorrules",
        }

    def register(self, manifest: ToolManifest):
        self._registry[manifest.name] = manifest

    def authorize(
        self,
        tool_name: str,
        requested_path: Optional[str] = None,
        requested_host: Optional[str] = None
    ) -> tuple[bool, str]:
        """执行时鉴权——Execution-time policy evaluation"""
        if tool_name not in self._registry:
            return False, f"工具 '{tool_name}' 未注册"

        manifest = self._registry[tool_name]

        if requested_path:
            # 绝对禁止修改敏感配置文件
            for forbidden in self._forbidden_paths:
                if requested_path.startswith(forbidden):
                    return False, f"绝对禁止:不允许访问 {forbidden}"

            if manifest.allowed_paths:
                allowed = any(requested_path.startswith(p) for p in manifest.allowed_paths)
                if not allowed:
                    return False, f"路径 {requested_path} 不在工具允许范围内"

        if requested_host and manifest.allowed_hosts:
            if requested_host not in manifest.allowed_hosts:
                return False, f"主机 {requested_host} 不在出站白名单"

        return True, "authorized"

    def execute_with_sandbox(self, tool_name: str, code: str, workspace: str = "/tmp/sandbox") -> str:
        """用 bubblewrap 实现 OS 级沙箱执行(适用于 DANGEROUS 级工具)"""
        manifest = self._registry.get(tool_name)
        if not manifest or not manifest.requires_sandbox:
            raise ValueError(f"工具 {tool_name} 未配置沙箱")

        # bubblewrap 沙箱:只读挂载系统库,读写工作目录,完全断网
        bwrap_cmd = [
            "bwrap",
            "--ro-bind", "/usr", "/usr",
            "--ro-bind", "/lib", "/lib",
            "--bind", workspace, "/workspace",
            "--unshare-net",       # 断网隔离,防止数据外泄
            "--unshare-pid",       # PID 命名空间隔离
            "--die-with-parent",   # 父进程退出时沙箱自动清理
            "--chdir", "/workspace",
            "python3", "-c", code
        ]

        try:
            result = subprocess.run(
                bwrap_cmd, capture_output=True, text=True, timeout=manifest.timeout_seconds
            )
            return result.stdout
        except subprocess.TimeoutExpired:
            return f"执行超时(>{manifest.timeout_seconds}s)"

# 注册示例
registry = ToolRegistry()

registry.register(ToolManifest(
    name="read_file",
    risk_level=RiskLevel.SAFE,
    allowed_paths={"/workspace/data/", "/tmp/readonly/"}
))

registry.register(ToolManifest(
    name="execute_code",
    risk_level=RiskLevel.DANGEROUS,
    allowed_paths={"/workspace/"},
    allowed_hosts=set(),        # 完全断网
    requires_sandbox=True,
    timeout_seconds=30
))

# 执行前鉴权
ok, reason = registry.authorize("read_file", requested_path="/workspace/data/report.csv")
# ok=True

ok, reason = registry.authorize("read_file", requested_path=os.path.expanduser("~/.ssh/id_rsa"))
# ok=False, reason="绝对禁止:不允许访问 /Users/xxx/.ssh"

审计日志

权限管理不能只靠预防,还需要事后可追溯。每次工具调用都应该记录:调用时间戳、工具名称、完整参数(脱敏处理密钥/密码字段)、执行结果(成功/失败/被拒绝)、执行耗时、触发此次调用的消息 ID。

审计日志有两个实用价值。第一:安全事件调查,当发生异常操作时能还原完整的工具调用链。第二:性能优化,通过统计哪些工具被频繁调用、哪些工具调用经常失败,识别工具定义的改进点。审计日志的存储应该独立于工具执行逻辑,避免被恶意代码修改或清除。


八、MCP:工具标准化的协议层

为什么需要 MCP

在 MCP(Model Context Protocol)出现之前,每个 Agent 框架都有自己的工具集成方式。LangChain 有 LangChain Tools,LlamaIndex 有自己的工具接口,Claude Code 有内置工具集,各自为政。同一个数据库查询工具,要给每个框架写一个适配层,维护成本随工具数量和框架数量的乘积增长。

MCP 是 Anthropic 在 2024 年推出的工具连接协议,目标是把工具集成标准化。一个工具(MCP Server)实现一次,所有支持 MCP 的客户端(Claude Desktop、Claude Code、各种第三方框架)都能直接使用。从架构类比来看,MCP 在 Agent 生态里的角色类似于 USB 协议在硬件生态里的角色:定义接口规范,让外设(工具)和主机(Agent)之间的连接标准化。

MCP 定义了三类资源:Tools(工具,模型可调用的函数)、Resources(资源,模型可读取的数据)、Prompts(提示词模板)。工具是其中最核心的部分,也是本篇文章关注的焦点。

MCP 的权限问题

MCP 解决了集成标准化问题,但引入了新的安全挑战。MCP 协议本身没有内置权限验证机制,这是 Agent 安全领域的研究者和从业者普遍指出的缺陷。在 Agent 安全领域的调查中,MCP 被列为”连接工具但缺乏权限验证”的典型协议:A2A(Google 的 Agent 通信协议)解决 Agent 间通信,MCP(Anthropic 的工具协议)解决 Agent 使用工具,但这两个协议都缺少跨平台的身份验证机制。攻击者可以发布恶意的 MCP Server,伪装成合法工具,诱导用户安装后劫持 Agent 的工具调用。

MCP 的 tool poisoning 攻击已经在安全研究中被证实可行:在 MCP Server 的工具定义里藏入隐藏指令(类似提示词注入,但出现在工具的 description 字段里),让模型在调用工具时受到操控。

应对策略:只使用来源可信的 MCP Server(官方或经过审计的第三方),在生产环境里对 MCP Server 的工具定义做静态审计(扫描 description 里的异常指令模式),用本章介绍的 ToolRegistry 对 MCP 工具的实际执行权限做二次约束。

从 Function Calling 到 MCP 的演进逻辑

Function Calling 是模型层面的能力(模型知道如何生成结构化的工具调用请求),MCP 是系统层面的协议(定义工具如何注册、发现、调用)。两者不是替代关系,而是不同层次的规范。

实际部署里,一个完整的工具调用链路通常是:MCP Server 定义并暴露工具 → MCP 客户端把工具注册信息转换成 tools 数组 → 发送给模型(使用 Function Calling)→ 模型返回 tool_call → MCP 客户端把调用路由到对应的 MCP Server 执行 → 结果回传给模型。MCP 是工具调用的”分发层”,Function Calling 是”调用层”,两者协作完成端到端的工具使用链路。


十、工具调用的结构性问题

工具定义的维护成本被严重低估

工具定义(JSON Schema + description)是工具调用系统里最容易腐败的资产。业务 API 在迭代,工具定义却可能没有同步更新,导致模型用过期的参数描述去调用已经变更了接口的工具。

这个问题没有技术银弹,只有工程规范:工具定义和 API 接口在同一代码仓库里管理,每次 API 变更必须同步更新工具定义,CI/CD 流程里加入工具定义的自动化校验(用真实调用测试工具定义的准确性)。考虑到工具定义是模型决策的直接输入,它的正确性至少应该与单元测试覆盖率同等重视。

Vercel 有一个极端案例值得关注:他们把工具数量从 20+ 砍到 4 个,Agent 的成功率从 80% 提升到 100%,速度提升 3.5 倍。少即是多。每个工具的定义越清晰、职责越单一,模型的选择准确率越高。当你发现模型频繁选错工具,先考虑减少工具数量或重写 description,而不是去调整提示词。工具数量是工具调用系统里最直接影响准确率的单一变量,在增加新工具之前,先问:现有工具能不能组合解决这个问题?

工具定义还有一个常被忽视的质量维度:负面说明。大部分工具定义只说”这个工具能做什么”,不说”什么时候不应该用它”。对语义相近的工具(例如 search_websearch_knowledge_base),明确写出两个工具的适用边界差异,能显著减少模型的混淆。

工具调用与模型推理的边界模糊问题

工具调用让模型有了执行能力,但也模糊了一个重要边界:什么应该由模型推理解决,什么应该调用工具解决。

让模型用工具查询它其实”知道”的信息(用工具查询通用知识),会消耗不必要的延迟和 API 配额。让模型用训练知识回答它”不知道”的实时信息(用记忆回答实时问题),会产生幻觉。正确的边界划分:训练截止日之前的稳定事实用模型知识,实时数据/用户私有数据/需要确定性计算的场景用工具。

这个边界不是一次配置好就不变的,随着模型更新(训练截止日变化)和业务需求变化需要持续校准。把这个决策写成可测试的规则(哪类问题应该用哪类工具),并用 LLM-as-judge 定期评估工具选择准确率,是更可靠的工程方案。

还有一类更微妙的边界问题:工具的返回结果里包含了足够做推理的信息,模型应该基于结果回答,还是应该再调用更多工具确认?过度调用工具会让 Agent 反应迟钝,但过早停止工具调用会导致回答不完整。通常在工具定义的 description 里加入”调用此工具后如果结果置信度不足,应调用 verify_result 工具进行交叉验证”这类指导信息,比依赖模型自己判断更稳定。

反射式自修复:工具失败的学习机会

一篇发布于 arxiv 的研究(2509.18847)提出了一个有价值的思路:把工具调用失败从启发式重试转化为可训练的反射动作策略。四个步骤:检测失败、结构化分析具体失败原因(错误的参数字段、不存在的工具名、任务分解错误)、基于分析调整策略、带修正参数重新执行。

在 ToolBench、APIBench、ACEBench 上对 Llama 3、Qwen、GPT-4 系列的实验均显示准确率有显著提升。核心洞察:错误信息本身就是训练信号,结构化的错误反馈比泛化的”失败”信号更能帮助模型学习如何修正。

在生产系统里,不要只记录工具调用失败的次数,要记录失败的具体原因分布(哪类参数错误最常见)。这份统计可以直接反馈到工具定义的优化上,如果”location 字段缺失”是最常见的错误,就在工具 description 里更明确地说明什么情况下 location 是必须的。

多步骤任务的工具调用编排问题

当一个用户请求需要多个工具按顺序执行(比如”查出销售额最高的产品,然后生成该产品的月度趋势图”),编排逻辑放在哪里是一个架构决策。

方案一:让模型自主编排。给模型提供所有相关工具,让它自己规划调用顺序。这是 ReAct 模式的核心思路,灵活但不可预测,适合开放式探索任务。方案二:用代码显式编排。把”先查数据再画图”的逻辑写成硬编码的工作流,工具调用只负责执行每个步骤。适合有固定流程的业务场景,可预测但不灵活。方案三:混合编排。用模型做高层规划(生成执行计划),用代码执行确定性步骤,对不确定的分支用模型判断。这是生产环境里最常见的选择。

Plan-and-Execute 模式(先让模型生成完整计划,再逐步执行)比纯粹的 ReAct 模式快 3.6 倍,因为它减少了每一步都需要请求模型判断下一步的延迟。代价是灵活性:如果中途发现原计划有误,需要额外的重新规划步骤。对延迟敏感的场景,Plan-and-Execute 更合适;对任务不确定性高的场景,ReAct 更合适。


十一、从这里开始:一个可落地的行动路径

理解工具调用的技术细节之后,最容易陷入的陷阱是过度设计,在第一个工具都没注册之前就开始担心沙箱隔离方案的性能开销。

实际的落地路径应该是这样的:从一个工具开始,把工具定义写得极其清晰,测试模型在各种变体请求下的调用准确率。准确率达到 90% 以上,再加第二个工具。超过 5 个工具时开始考虑工具路由层;超过 10 个工具时把权限白名单注册器放进去;第一次需要执行代码时才引入沙箱方案,先用最简单的 Docker 隔离,性能瓶颈出现后再评估 Firecracker 或 gVisor。

并行工具调用不是默认要开启的优化,而是在测量到串行瓶颈之后才引入的特性。先把单工具调用的准确率和可靠性跑稳,再追求并发。错误处理也是同样的逻辑:先把指数退避做对,再根据实际错误分布决定要不要加熔断器。

度量驱动每一步决策。每次工具调用的成功率、平均延迟、错误类型分布,这三个指标可以告诉你当前系统的真实瓶颈在哪,引导你把有限的工程投入放在最值得的地方。

具体来说,可以从一个最小评估集开始:收集 50 个有代表性的用户请求,手工标注每个请求的”正确工具调用序列”,然后用 Agent 跑一遍,计算工具选择准确率、参数正确率、任务完成率。这个评估集就是你的验收标准。每次修改工具定义、调整提示词或升级模型,都在这个集合上跑一遍,确认没有退化。没有这个评估集,你无法知道某次”优化”是真的让系统变好,还是在一个方面好了另一个方面坏了。

工具调用系统的成熟标志不是工具数量,不是并发能力,而是:每次工具调用失败,你能在 5 分钟内准确判断是哪层出了问题(工具定义、参数生成、执行逻辑、权限控制),并给出可验证的修复方案。做到这一点,系统才算真正可维护。

抢沙发

评论前必须登录!

立即登录   注册