Agent01：工具调用 | 旋大爷的博客

工具调用是 Agent 的核心能力之一，它让 Agent 能够超越纯文本交互，与外部系统、API 和数据库进行交互。理解工具调用的机制，才能构建真正实用的 Agent 系统。

工具调用的基本原理

什么是工具调用

工具调用是指 Agent 根据用户需求，自动选择并执行外部工具或 API 的能力：

组件	说明	示例
Agent	决策中心，决定调用哪个工具	分析用户需求，选择合适的工具
工具	可执行的外部能力	API、函数、数据库查询
参数	工具执行所需的输入	用户输入、上下文信息
返回值	工具执行的结果	API 响应、函数输出

工具调用的流程

用户查询 → Agent 分析 → 选择工具 → 生成参数 → 执行工具 → 获取结果 → 总结回答

工具调用 vs 纯文本生成

维度	纯文本生成	工具调用
信息来源	模型内部知识	外部实时数据
执行能力	只能生成文本	可以执行操作
时效性	依赖训练数据	实时获取最新信息
可靠性	可能产生幻觉	结果可验证
应用范围	受限	可扩展

函数调用机制

Function Calling 原理

Function Calling 是大模型的一项能力，它能理解何时需要调用工具，并生成结构化的调用请求：

步骤	说明	输出
工具描述	定义工具的名称、参数和用途	JSON Schema
模型推理	模型分析是否需要调用工具	调用决策
参数生成	模型生成工具调用所需的参数	参数 JSON
执行调用	执行工具并获取结果	工具返回值
结果总结	模型总结工具返回结果	自然语言回答

工具定义规范

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如北京、上海"
                    },
                    "date": {
                        "type": "string",
                        "description": "日期，格式为YYYY-MM-DD，默认为今天"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_flights",
            "description": "搜索航班信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "from_city": {"type": "string", "description": "出发城市"},
                    "to_city": {"type": "string", "description": "到达城市"},
                    "date": {"type": "string", "description": "出发日期"}
                },
                "required": ["from_city", "to_city", "date"]
            }
        }
    }
]

函数调用实现

import openai

def call_tool(tool_name, arguments):
    """执行工具调用"""
    tools_map = {
        "get_weather": get_weather_api,
        "search_flights": search_flights_api,
        "query_database": query_database
    }
    
    if tool_name not in tools_map:
        return {"error": f"工具 {tool_name} 不存在"}
    
    return tools_map[tool_name](**arguments)

def agent_tool_calling(user_query):
    """Agent 工具调用流程"""
    # 1. 调用模型，获取工具调用决策
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_query}],
        tools=tools,
        tool_choice="auto"
    )
    
    # 2. 检查是否需要调用工具
    if response.choices[0].message.tool_calls:
        tool_call = response.choices[0].message.tool_calls[0]
        tool_name = tool_call.function.name
        arguments = json.loads(tool_call.function.arguments)
        
        # 3. 执行工具
        tool_result = call_tool(tool_name, arguments)
        
        # 4. 将结果返回给模型，生成最终回答
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "user", "content": user_query},
                {"role": "assistant", "content": "", "tool_calls": [tool_call]},
                {"role": "tool", "content": json.dumps(tool_result)}
            ]
        )
        
        return response.choices[0].message.content
    
    return response.choices[0].message.content

API 编排

API 编排的场景

Agent 常常需要调用多个 API 来完成复杂任务：

场景	说明	示例
数据聚合	调用多个 API 获取数据并汇总	整合天气、航班、酒店信息
流程编排	按顺序调用多个 API	创建订单 → 支付 → 发货
条件分支	根据前一个 API 的结果决定后续调用	检查库存 → 有库存则下单，无库存则推荐替代
循环调用	多次调用同一个 API	分页查询所有数据

API 编排实现

class APIOrchestrator:
    def __init__(self):
        self.tools = {
            "get_user_info": self.get_user_info,
            "get_order_list": self.get_order_list,
            "get_order_detail": self.get_order_detail,
            "send_notification": self.send_notification
        }
    
    async def execute_workflow(self, user_id):
        """执行复杂的 API 编排工作流"""
        # 1. 获取用户信息
        user_info = await self.tools["get_user_info"](user_id)
        
        # 2. 获取用户订单列表
        orders = await self.tools["get_order_list"](user_id)
        
        # 3. 获取每个订单的详情
        order_details = []
        for order in orders[:3]:  # 只获取前3个订单
            detail = await self.tools["get_order_detail"](order["id"])
            order_details.append(detail)
        
        # 4. 发送通知
        notification = {
            "user": user_info["name"],
            "orders": len(orders),
            "recent_orders": order_details
        }
        await self.tools["send_notification"](user_id, notification)
        
        return notification

API 编排模式

模式	说明	适用场景
顺序编排	按顺序执行多个 API	流程化任务
并行编排	同时执行多个 API	数据聚合
条件编排	根据条件选择执行路径	需要决策的场景
循环编排	重复执行某个 API	批量处理

参数校验

参数校验的必要性

问题	说明	后果
参数缺失	缺少必需参数	API 调用失败
参数类型错误	参数类型不匹配	运行时错误
参数格式错误	参数格式不正确	数据处理错误
参数越界	参数值超出范围	异常结果或安全问题
恶意参数	参数包含恶意内容	安全风险

参数校验实现

from pydantic import BaseModel, ValidationError, Field

class GetWeatherRequest(BaseModel):
    city: str = Field(..., description="城市名称", min_length=1, max_length=50)
    date: str = Field(None, description="日期，格式为YYYY-MM-DD")
    
    @validator("date")
    def validate_date(cls, v):
        if v:
            try:
                datetime.strptime(v, "%Y-%m-%d")
            except ValueError:
                raise ValueError("日期格式必须为YYYY-MM-DD")
        return v

def validate_parameters(tool_name, arguments):
    """参数校验"""
    validators = {
        "get_weather": GetWeatherRequest,
        "search_flights": SearchFlightsRequest
    }
    
    if tool_name in validators:
        try:
            validators[tool_name](**arguments)
        except ValidationError as e:
            return False, str(e)
    
    return True, ""

参数校验策略

策略	说明	实现方式
类型校验	检查参数类型	Pydantic、TypeScript 类型
格式校验	检查参数格式	正则表达式
范围校验	检查参数值范围	最小值、最大值限制
枚举校验	检查参数是否在允许范围内	枚举类型
依赖校验	检查参数之间的依赖关系	自定义校验逻辑

权限边界

权限控制的重要性

风险	说明	后果
越权访问	Agent 调用了无权限的工具	数据泄露
敏感操作	Agent 执行了危险操作	系统损坏
资源滥用	Agent 频繁调用资源密集型工具	资源耗尽
数据篡改	Agent 修改了不该修改的数据	数据错误

权限模型

class PermissionManager:
    def __init__(self):
        self.permissions = {
            "admin": ["get_weather", "search_flights", "query_database", "modify_data"],
            "user": ["get_weather", "search_flights"],
            "guest": ["get_weather"]
        }
    
    def check_permission(self, user_role, tool_name):
        """检查用户是否有权限调用工具"""
        if user_role not in self.permissions:
            return False
        
        return tool_name in self.permissions[user_role]
    
    def get_available_tools(self, user_role):
        """获取用户可用的工具列表"""
        return self.permissions.get(user_role, [])

权限控制策略

策略	说明	适用场景
角色权限	根据用户角色分配权限	用户分级场景
资源权限	根据资源类型分配权限	多租户场景
操作权限	根据操作类型分配权限	需要精细控制的场景
上下文权限	根据上下文动态分配权限	临时授权场景

工具调用的常见问题

问题1：工具选择错误

表现：Agent 选择了不合适的工具

解决方案：

优化工具描述，使其更清晰
添加工具选择的示例
使用更强大的模型进行工具选择

问题2：参数生成错误

表现：Agent 生成的参数不正确

解决方案：

提供详细的参数描述
添加参数生成的示例
实现严格的参数校验

问题3：工具调用失败

表现：工具执行过程中发生错误

解决方案：

添加错误处理和重试机制
记录详细的错误日志
提供友好的错误提示

问题4：权限不足

表现：Agent 尝试调用无权限的工具

解决方案：

实现严格的权限检查
在工具描述中注明权限要求
提供权限不足的友好提示

工具调用的最佳实践

工具设计原则

原则	说明	示例
单一职责	每个工具只做一件事	`get_weather` 只获取天气
接口清晰	参数和返回值定义清晰	使用 JSON Schema
幂等性	多次调用结果一致	查询类工具天然幂等
容错性	处理异常情况	返回明确的错误信息
安全性	考虑权限和注入风险	参数校验、权限控制

工具调用流程

1. 工具注册
   - 定义工具描述（JSON Schema）
   - 实现工具执行逻辑
   - 配置权限控制

2. 工具选择
   - 模型分析用户需求
   - 匹配最合适的工具
   - 生成调用参数

3. 参数校验
   - 检查参数完整性
   - 验证参数类型和格式
   - 检查参数范围

4. 权限检查
   - 验证用户权限
   - 检查资源访问权限
   - 确认操作权限

5. 工具执行
   - 调用工具函数
   - 处理异常和重试
   - 返回执行结果

6. 结果总结
   - 将结果传递给模型
   - 生成自然语言回答
   - 提供来源引用

评估指标

指标	定义	目标值
工具选择准确率	正确选择工具的比例	> 90%
参数生成准确率	参数正确的比例	> 85%
工具调用成功率	工具执行成功的比例	> 95%
用户满意度	用户对结果的满意度	> 4.5/5

项目判断清单

需要实时数据 → 使用工具调用获取外部数据
需要执行操作 → 设计相应的工具函数
需要调用多个 API → 实现 API 编排
参数容易出错 → 添加严格的参数校验
有安全风险 → 实现权限控制
工具选择错误 → 优化工具描述和示例
需要重试机制 → 添加错误处理和重试
需要监控 → 记录工具调用日志和指标