839 字
2 分钟
Agent 工具调用：Function Calling 与 Tool Use 实战
如果 Agent 是「大脑」，那工具就是它的「双手」。
没有工具的 Agent 只能进行纯文本交互，无法访问外部信息，无法执行实际操作，无法真正改变世界。
工具调用是 Agent 从「聊天机器人」进化为「智能助手」的关键能力。
本文要点
- Function Calling 核心原理
- OpenAI vs Anthropic 工具调用对比
- Tool Schema 设计最佳实践
- 工具选择与排序策略
- 多工具协同模式
- 常见工具类型与实现
一、工具调用的本质
1.1 为什么需要工具调用？
问题：大模型的局限性
1. 知识截止：训练数据有时效性，无法获取最新信息2. 计算能力：复杂数学计算容易出错3. 无外部访问：无法访问数据库、API、文件系统4. 无实际操作：只能「说」，不能「做」
解决：工具调用
让大模型能够调用外部工具，获取信息、执行操作。1.2 工具调用的基本流程
sequenceDiagram
participant U as 用户
participant A as Agent (LLM)
participant T as 工具
U->>A: 提出请求
Note over A: 分析任务<br/>判断需要调用工具
A->>A: 生成工具调用参数
A->>T: 调用工具
T-->>A: 返回结果
Note over A: 处理工具结果<br/>生成最终回答
A-->>U: 返回结果
1.3 核心概念
┌─────────────────────────────────────────────────────────────┐│ 工具调用核心概念 │├─────────────────────────────────────────────────────────────┤│ ││ Tool（工具） ││ ├── 定义：可供调用的外部功能 ││ ├── 示例：搜索、数据库查询、API调用、代码执行 ││ └── 属性：名称、描述、参数Schema、执行函数 ││ ││ Tool Schema ││ ├── 定义：工具的结构化描述 ││ ├── 内容：名称、功能描述、参数定义 ││ └── 格式：JSON Schema ││ ││ Function Calling ││ ├── 定义：模型生成工具调用参数的能力 ││ ├── 过程：理解需求 → 选择工具 → 构造参数 ││ └── 输出：结构化的工具调用请求 ││ ││ Tool Use ││ ├── 定义：实际执行工具调用 ││ ├── 过程：接收参数 → 执行函数 → 返回结果 ││ └── 实现：开发者提供的执行逻辑 ││ │└─────────────────────────────────────────────────────────────┘二、Function Calling 原理详解
2.1 OpenAI Function Calling
基本使用
from openai import OpenAI
client = OpenAI()
# 1. 定义工具tools = [ { "type": "function", "function": { "name": "get_weather", "description": "获取指定城市的当前天气信息", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称，如：北京、上海" }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位，默认摄氏度" } }, "required": ["city"] } } }]
# 2. 发送请求response = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "user", "content": "北京今天天气怎么样？"} ], tools=tools, tool_choice="auto" # auto: 自动决定是否调用工具)
# 3. 处理响应message = response.choices[0].message
if message.tool_calls: # 模型决定调用工具 for tool_call in message.tool_calls: function_name = tool_call.function.name arguments = json.loads(tool_call.function.arguments)
print(f"调用工具: {function_name}") print(f"参数: {arguments}") # 执行工具并获取结果 result = execute_tool(function_name, arguments)完整的函数调用循环
import jsonfrom openai import OpenAI
class OpenAIFunctionCaller: """OpenAI Function Calling 完整实现"""
def __init__(self, model: str = "gpt-4o"): self.client = OpenAI() self.model = model self.tools = [] self.tool_functions = {}
def register_tool(self, schema: dict, function: callable): """注册工具""" self.tools.append(schema) self.tool_functions[schema["function"]["name"]] = function
def run(self, user_message: str, max_iterations: int = 5) -> str: """执行对话，支持多轮工具调用"""
messages = [ {"role": "user", "content": user_message} ]
for _ in range(max_iterations): # 调用模型 response = self.client.chat.completions.create( model=self.model, messages=messages, tools=self.tools, tool_choice="auto" )
message = response.choices[0].message
# 检查是否需要调用工具 if not message.tool_calls: # 无需调用工具，返回最终回答 return message.content
# 添加助手消息到历史 messages.append(message)
# 处理每个工具调用 for tool_call in message.tool_calls: function_name = tool_call.function.name arguments = json.loads(tool_call.function.arguments)
# 执行工具 result = self._execute_tool(function_name, arguments)
# 添加工具结果到历史 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(result) })
return "达到最大迭代次数，任务未完成"
def _execute_tool(self, name: str, arguments: dict) -> any: """执行工具""" if name not in self.tool_functions: return f"错误：未知工具 {name}"
try: return self.tool_functions[name](**arguments) except Exception as e: return f"工具执行错误：{str(e)}"
# 使用示例caller = OpenAIFunctionCaller()
# 注册天气工具caller.register_tool( schema={ "type": "function", "function": { "name": "get_weather", "description": "获取城市天气", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名称"} }, "required": ["city"] } } }, function=lambda city: f"{city}今天晴，温度15-25°C")
# 执行result = caller.run("北京今天天气怎么样？")print(result)2.2 Anthropic Tool Use
基本使用
from anthropic import Anthropic
client = Anthropic()
# 1. 定义工具（与 OpenAI 略有不同）tools = [ { "name": "get_weather", "description": "获取指定城市的当前天气信息", "input_schema": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称" }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位" } }, "required": ["city"] } }]
# 2. 发送请求response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=1024, messages=[ {"role": "user", "content": "北京今天天气怎么样？"} ], tools=tools)
# 3. 处理响应for block in response.content: if block.type == "tool_use": # 工具调用 tool_name = block.name tool_input = block.input
print(f"调用工具: {tool_name}") print(f"参数: {tool_input}")
# 执行工具 result = execute_tool(tool_name, tool_input)
# 继续对话 response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=1024, messages=[ {"role": "user", "content": "北京今天天气怎么样？"}, {"role": "assistant", "content": response.content}, { "role": "user", "content": [ { "type": "tool_result", "tool_use_id": block.id, "content": str(result) } ] } ], tools=tools )完整实现
import anthropicfrom typing import List, Dict, Any, Callable
class AnthropicToolUser: """Anthropic Tool Use 完整实现"""
def __init__(self, model: str = "claude-sonnet-4-20250514"): self.client = anthropic.Anthropic() self.model = model self.tools = [] self.tool_functions: Dict[str, Callable] = {}
def register_tool(self, name: str, description: str, input_schema: dict, function: Callable): """注册工具""" self.tools.append({ "name": name, "description": description, "input_schema": input_schema }) self.tool_functions[name] = function
def run(self, user_message: str, max_iterations: int = 5) -> str: """执行对话"""
messages = [ {"role": "user", "content": user_message} ]
for _ in range(max_iterations): response = self.client.messages.create( model=self.model, max_tokens=4096, messages=messages, tools=self.tools )
# 检查是否有工具调用 tool_use_blocks = [ block for block in response.content if block.type == "tool_use" ]
if not tool_use_blocks: # 没有工具调用，返回文本结果 text_blocks = [ block for block in response.content if block.type == "text" ] return "".join(block.text for block in text_blocks)
# 添加助手响应 messages.append({ "role": "assistant", "content": response.content })
# 处理工具调用 tool_results = [] for block in tool_use_blocks: result = self._execute_tool(block.name, block.input)
tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": str(result) })
# 添加工具结果 messages.append({ "role": "user", "content": tool_results })
return "达到最大迭代次数"
def _execute_tool(self, name: str, arguments: dict) -> Any: """执行工具""" if name not in self.tool_functions: return {"error": f"未知工具: {name}"}
try: return self.tool_functions[name](**arguments) except Exception as e: return {"error": str(e)}2.3 OpenAI vs Anthropic 对比
┌─────────────────────────────────────────────────────────────┐│ Function Calling 对比 │├─────────────────────────────────────────────────────────────┤│ ││ OpenAI Anthropic ││ ──────────────────────────── ──────────────────────── ││ ││ Schema 格式: Schema 格式: ││ { { ││ "type": "function", "name": "...", ││ "function": { "description": "...", ││ "name": "...", "input_schema": {...} ││ "description": "...", } ││ "parameters": {...} ││ } ││ } │ ││ ││ 工具调用格式: 工具调用格式: ││ message.tool_calls[] response.content[] ││ .function.name .type == "tool_use" ││ .function.arguments .name ││ .id .input ││ ││ 工具结果格式: 工具结果格式: ││ { { ││ "role": "tool", "type": "tool_result", ││ "tool_call_id": "...", "tool_use_id": "...", ││ "content": "..." "content": "..." ││ } } ││ ││ 特点: 特点: ││ • tool_choice 可控制 • 必须在 tools 中定义 ││ • 支持并行调用 • 更严格的 Schema 验证 ││ • 调用更简洁 • 错误处理更详细 ││ │└─────────────────────────────────────────────────────────────┘三、Tool Schema 设计最佳实践
3.1 Schema 设计原则
flowchart TD
A[Tool Schema 设计] --> B[清晰命名]
A --> C[详细描述]
A --> D[合理参数]
A --> E[类型安全]
B --> B1[动词+名词]
B --> B2[避免缩写]
C --> C1[说明功能]
C --> C2[使用场景]
C --> C3[返回内容]
D --> D1[必填/可选]
D --> D2[默认值]
D --> D3[参数校验]
E --> E1[使用枚举]
E --> E2[设置范围]
E --> E3[格式约束]
3.2 好的 Schema 示例
# 好的设计good_weather_tool = { "type": "function", "function": { "name": "get_current_weather", "description": """获取指定城市的实时天气信息。
使用场景：- 用户询问某地天气时调用- 需要天气数据做决策时使用
返回信息：温度、湿度、天气状况、风向风速""", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称，支持中英文，如：北京、Beijing" }, "country": { "type": "string", "description": "国家代码（ISO 3166），如：CN、US。可选，用于区分同名城市" }, "units": { "type": "string", "enum": ["metric", "imperial"], "description": "温度单位：metric(摄氏度)，imperial(华氏度)。默认 metric" } }, "required": ["city"] } }}
# 不好的设计bad_weather_tool = { "type": "function", "function": { "name": "weather", # 不清晰 "description": "获取天气", # 太简单 "parameters": { "type": "object", "properties": { "c": { # 缩写不明确 "type": "string" # 缺少描述 }, "u": { # 缩写不明确 "type": "string" # 没有限制值范围 } } } }}3.3 参数设计技巧
# 1. 使用枚举限制选项status_tool = { "type": "function", "function": { "name": "update_order_status", "parameters": { "type": "object", "properties": { "order_id": {"type": "string"}, "status": { "type": "string", "enum": ["pending", "processing", "shipped", "delivered", "cancelled"], "description": "订单状态" } } } }}
# 2. 使用默认值减少必填参数search_tool = { "type": "function", "function": { "name": "search_products", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词"}, "page": { "type": "integer", "description": "页码，从1开始", "default": 1 }, "page_size": { "type": "integer", "description": "每页数量", "default": 20 }, "sort_by": { "type": "string", "enum": ["relevance", "price_asc", "price_desc", "rating"], "default": "relevance" } }, "required": ["query"] } }}
# 3. 复杂对象使用嵌套结构order_tool = { "type": "function", "function": { "name": "create_order", "parameters": { "type": "object", "properties": { "customer": { "type": "object", "properties": { "name": {"type": "string"}, "email": {"type": "string", "format": "email"}, "phone": {"type": "string"} }, "required": ["name", "email"] }, "items": { "type": "array", "items": { "type": "object", "properties": { "product_id": {"type": "string"}, "quantity": {"type": "integer", "minimum": 1} }, "required": ["product_id", "quantity"] } }, "shipping_address": { "type": "object", "properties": { "street": {"type": "string"}, "city": {"type": "string"}, "zip_code": {"type": "string"} } } }, "required": ["customer", "items"] } }}四、工具选择与排序策略
4.1 tool_choice 参数
# OpenAI 的 tool_choice 选项options = { "auto": "模型自动决定是否调用工具", "none": "强制不调用任何工具", "required": "强制必须调用至少一个工具", {"type": "function", "function": {"name": "xxx"}}: "强制调用指定工具"}
# 示例：强制调用搜索工具response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "搜索最新的AI新闻"}], tools=[search_tool], tool_choice={"type": "function", "function": {"name": "web_search"}})4.2 工具选择策略
flowchart TD
A[用户请求] --> B{分析意图}
B --> C[识别关键实体]
B --> D[理解任务类型]
C --> E{匹配工具}
D --> E
E --> F{单工具?}
F -->|是| G[调用对应工具]
F -->|否| H[工具排序]
H --> I[按相关性排序]
I --> J[按历史成功率]
J --> K[返回排序结果]
4.3 智能工具选择器
from typing import List, Dictimport re
class ToolSelector: """智能工具选择器"""
def __init__(self, tools: List[Dict]): self.tools = tools self.tool_stats = {} # 工具统计信息
def select_tools(self, query: str, max_tools: int = 5) -> List[Dict]: """选择最相关的工具"""
scored_tools = []
for tool in self.tools: score = self._calculate_relevance(tool, query) scored_tools.append((tool, score))
# 按分数排序 scored_tools.sort(key=lambda x: x[1], reverse=True)
return [t[0] for t in scored_tools[:max_tools]]
def _calculate_relevance(self, tool: Dict, query: str) -> float: """计算工具与查询的相关性"""
function = tool.get("function", {}) name = function.get("name", "").lower() description = function.get("description", "").lower()
score = 0.0
# 1. 名称匹配 name_words = name.split("_") for word in name_words: if word in query.lower(): score += 0.3
# 2. 描述关键词匹配 query_words = set(query.lower().split()) desc_words = set(description.split()) overlap = query_words & desc_words score += len(overlap) * 0.1
# 3. 正则模式匹配 patterns = self._extract_patterns(function) for pattern in patterns: if re.search(pattern, query, re.IGNORECASE): score += 0.2
# 4. 历史成功率 stats = self.tool_stats.get(name, {}) success_rate = stats.get("success_rate", 0.5) score *= (0.5 + success_rate * 0.5)
return score
def _extract_patterns(self, function: Dict) -> List[str]: """从描述中提取模式""" description = function.get("description", "") patterns = []
# 简单的模式提取逻辑 # 实际中可以使用更复杂的 NLP 技术 weather_keywords = ["天气", "温度", "weather", "temperature"] search_keywords = ["搜索", "查找", "search", "find"]
for keyword in weather_keywords: if keyword in description.lower(): patterns.append(r"天气|温度|weather")
for keyword in search_keywords: if keyword in description.lower(): patterns.append(r"搜索|查找|search|find")
return patterns
def record_result(self, tool_name: str, success: bool): """记录工具调用结果""" if tool_name not in self.tool_stats: self.tool_stats[tool_name] = { "calls": 0, "successes": 0, "success_rate": 0.5 }
stats = self.tool_stats[tool_name] stats["calls"] += 1 if success: stats["successes"] += 1
stats["success_rate"] = stats["successes"] / stats["calls"]五、多工具协同模式
5.1 并行调用
class ParallelToolCaller: """并行工具调用"""
def __init__(self, caller): self.caller = caller self.executor = ThreadPoolExecutor(max_workers=10)
async def call_parallel(self, tool_calls: List[Dict]) -> List[Any]: """并行执行多个工具调用"""
futures = [] for call in tool_calls: future = self.executor.submit( self.caller._execute_tool, call["name"], call["arguments"] ) futures.append(future)
results = [f.result() for f in futures] return results
# 使用示例async def process_complex_query(query: str): """处理需要多个工具的复杂查询"""
# 假设模型决定需要调用多个工具 tool_calls = [ {"name": "get_weather", "arguments": {"city": "北京"}}, {"name": "get_news", "arguments": {"topic": "科技", "limit": 5}}, {"name": "get_stock", "arguments": {"symbol": "AAPL"}} ]
# 并行执行 results = await parallel_caller.call_parallel(tool_calls)
# 整合结果 return { "weather": results[0], "news": results[1], "stock": results[2] }5.2 工具链式调用
flowchart LR
A[用户请求] --> B[工具1: 搜索]
B --> C[结果1]
C --> D[工具2: 分析]
D --> E[结果2]
E --> F[工具3: 生成]
F --> G[最终结果]
class ChainedToolCaller: """链式工具调用"""
def __init__(self, caller): self.caller = caller
def execute_chain(self, query: str, chain: List[Dict]) -> Any: """执行工具链"""
context = {"query": query, "results": []}
for step in chain: # 准备参数（可以使用前一步的结果） arguments = self._prepare_arguments(step, context)
# 执行工具 result = self.caller._execute_tool( step["tool"], arguments )
context["results"].append({ "tool": step["tool"], "arguments": arguments, "result": result })
return context
def _prepare_arguments(self, step: Dict, context: Dict) -> Dict: """准备工具参数，支持引用前序结果"""
arguments = {} arg_config = step.get("arguments", {})
for key, value in arg_config.items(): if isinstance(value, str) and value.startswith("$"): # 引用前序结果 ref_path = value[1:] # 去掉 $ value = self._resolve_reference(ref_path, context)
arguments[key] = value
return arguments
def _resolve_reference(self, path: str, context: Dict) -> Any: """解析引用路径"""
# 支持格式：$results[0].result.data parts = path.split(".") value = context
for part in parts: if "[" in part: # 数组索引 key = part.split("[")[0] index = int(part.split("[")[1].rstrip("]")) value = value[key][index] else: value = value[part]
return value
# 使用示例chain = [ { "tool": "web_search", "arguments": {"query": "$query"} }, { "tool": "extract_content", "arguments": {"url": "$results[0].result.url"} }, { "tool": "summarize", "arguments": {"content": "$results[1].result.content"} }]
result = chain_caller.execute_chain("AI最新进展", chain)5.3 条件分支
class ConditionalToolCaller: """条件分支工具调用"""
def execute_with_conditions(self, query: str, rules: List[Dict]) -> Any: """根据条件选择执行路径"""
for rule in rules: if self._evaluate_condition(rule["condition"], query): return self._execute_action(rule["action"])
return None
def _evaluate_condition(self, condition: str, query: str) -> bool: """评估条件"""
conditions = { "contains_weather": lambda q: any( word in q for word in ["天气", "温度", "weather"] ), "contains_search": lambda q: any( word in q for word in ["搜索", "查找", "search"] ), "is_question": lambda q: "?" in q or "？" in q }
evaluator = conditions.get(condition, lambda q: False) return evaluator(query)
# 规则配置rules = [ { "condition": "contains_weather", "action": {"tool": "get_weather", "arguments": {"city": "北京"}} }, { "condition": "contains_search", "action": {"tool": "web_search", "arguments": {"query": "$query"}} }, { "condition": "is_question", "action": {"tool": "answer_question", "arguments": {"question": "$query"}} }]六、常见工具类型与实现
6.1 搜索工具
def web_search_tool(query: str, limit: int = 5) -> list: """网络搜索工具"""
tools_schema = { "type": "function", "function": { "name": "web_search", "description": "搜索互联网获取信息", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词" }, "limit": { "type": "integer", "description": "返回结果数量", "default": 5 } }, "required": ["query"] } } }
# 实现搜索逻辑 # 可以使用 SerperAPI、Bing Search API 等 import requests
API_KEY = "your-api-key" url = "https://api.serper.dev/search"
response = requests.post( url, headers={"X-API-KEY": API_KEY}, json={"q": query, "num": limit} )
results = response.json().get("organic", [])
return [ { "title": r["title"], "link": r["link"], "snippet": r.get("snippet", "") } for r in results[:limit] ]6.2 数据库工具
import sqlite3from typing import List, Dict
class DatabaseTool: """数据库查询工具"""
def __init__(self, db_path: str): self.conn = sqlite3.connect(db_path) self.conn.row_factory = sqlite3.Row
def get_schema(self) -> dict: """返回工具 Schema""" return { "type": "function", "function": { "name": "query_database", "description": "执行 SQL 查询获取数据库信息", "parameters": { "type": "object", "properties": { "sql": { "type": "string", "description": "SQL 查询语句（仅支持 SELECT）" } }, "required": ["sql"] } } }
def execute(self, sql: str) -> List[Dict]: """执行查询""" # 安全检查：只允许 SELECT if not sql.strip().upper().startswith("SELECT"): return {"error": "仅支持 SELECT 查询"}
try: cursor = self.conn.execute(sql) rows = cursor.fetchall() return [dict(row) for row in rows] except Exception as e: return {"error": str(e)}
# 使用示例db_tool = DatabaseTool("sales.db")
# 注册到 Agentcaller.register_tool( schema=db_tool.get_schema(), function=db_tool.execute)6.3 API 调用工具
import requestsfrom typing import Dict, Any
class APICallTool: """通用 API 调用工具"""
def __init__(self, base_url: str, headers: Dict = None): self.base_url = base_url.rstrip("/") self.headers = headers or {}
def get_schema(self) -> dict: return { "type": "function", "function": { "name": "api_call", "description": f"调用 {self.base_url} 的 API 接口", "parameters": { "type": "object", "properties": { "method": { "type": "string", "enum": ["GET", "POST", "PUT", "DELETE"], "default": "GET" }, "endpoint": { "type": "string", "description": "API 端点路径" }, "params": { "type": "object", "description": "查询参数" }, "body": { "type": "object", "description": "请求体（POST/PUT）" } }, "required": ["endpoint"] } } }
def execute(self, method: str = "GET", endpoint: str = "", params: Dict = None, body: Dict = None) -> Any:
url = f"{self.base_url}/{endpoint.lstrip('/')}"
try: response = requests.request( method=method, url=url, headers=self.headers, params=params, json=body ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: return {"error": str(e)}6.4 代码执行工具
import subprocessimport tempfileimport os
class CodeExecutionTool: """代码执行工具"""
def get_schema(self) -> dict: return { "type": "function", "function": { "name": "execute_code", "description": "执行 Python 代码并返回结果", "parameters": { "type": "object", "properties": { "code": { "type": "string", "description": "要执行的 Python 代码" }, "timeout": { "type": "integer", "description": "执行超时时间（秒）", "default": 30 } }, "required": ["code"] } } }
def execute(self, code: str, timeout: int = 30) -> dict: """安全执行代码"""
# 使用临时文件执行 with tempfile.NamedTemporaryFile( mode='w', suffix='.py', delete=False ) as f: f.write(code) temp_file = f.name
try: result = subprocess.run( ['python', temp_file], capture_output=True, text=True, timeout=timeout )
return { "success": result.returncode == 0, "stdout": result.stdout, "stderr": result.stderr }
except subprocess.TimeoutExpired: return { "success": False, "error": f"执行超时（{timeout}秒）" }
finally: os.unlink(temp_file)七、工具调用最佳实践
7.1 错误处理
class RobustToolCaller: """健壮的工具调用"""
def execute_with_retry(self, tool_name: str, arguments: dict, max_retries: int = 3) -> dict:
last_error = None
for attempt in range(max_retries): try: result = self._execute_tool(tool_name, arguments)
if isinstance(result, dict) and "error" in result: last_error = result["error"]
# 尝试修复参数 fixed_args = self._try_fix_arguments( tool_name, arguments, last_error ) if fixed_args != arguments: arguments = fixed_args continue
return {"success": True, "result": result}
except Exception as e: last_error = str(e) time.sleep(2 ** attempt) # 指数退避
return { "success": False, "error": last_error, "suggestion": self._suggest_alternative(tool_name, arguments) }
def _try_fix_arguments(self, tool_name: str, arguments: dict, error: str) -> dict: """尝试修复参数"""
# 使用 LLM 分析错误并建议修复 prompt = f"""工具调用失败，请分析错误并建议修复：
工具：{tool_name}参数：{json.dumps(arguments)}错误：{error}
请返回修复后的参数（JSON格式）。"""
fixed = self.llm.generate(prompt) return json.loads(fixed)7.2 安全控制
class SecureToolManager: """安全的工具管理"""
def __init__(self): self.tools = {} self.permissions = {} # 工具权限配置 self.audit_log = [] # 审计日志
def register_tool(self, name: str, function: callable, permission_level: str = "normal"): """注册工具并设置权限级别"""
self.tools[name] = function self.permissions[name] = permission_level
def execute(self, tool_name: str, arguments: dict, user_permission: str = "normal") -> any: """执行工具（带权限检查）"""
# 1. 权限检查 required = self.permissions.get(tool_name, "normal") if not self._check_permission(user_permission, required): self._log(tool_name, arguments, "PERMISSION_DENIED") raise PermissionError( f"权限不足：需要 {required} 级别" )
# 2. 参数校验 validated_args = self._validate_arguments(tool_name, arguments)
# 3. 执行并记录 try: result = self.tools[tool_name](**validated_args) self._log(tool_name, validated_args, "SUCCESS", result) return result except Exception as e: self._log(tool_name, validated_args, "ERROR", str(e)) raise
def _check_permission(self, user_level: str, required: str) -> bool: """检查权限""" levels = ["guest", "normal", "admin", "super"] return levels.index(user_level) >= levels.index(required)
def _validate_arguments(self, tool_name: str, arguments: dict) -> dict: """校验和清理参数""" # 实现 SQL 注入防护、XSS 过滤等 validated = {} for key, value in arguments.items(): if isinstance(value, str): # 清理危险字符 value = self._sanitize_string(value) validated[key] = value return validated
def _log(self, tool_name: str, arguments: dict, status: str, result: any = None): """记录审计日志""" self.audit_log.append({ "timestamp": datetime.now().isoformat(), "tool": tool_name, "arguments": arguments, "status": status, "result": str(result)[:500] # 截断 })常见问题 FAQ
Q1：如何决定何时使用工具调用？
A：考虑以下因素：
- 任务是否需要外部数据？（如搜索、数据库）
- 是否需要执行实际操作？（如发送邮件、创建文件）
- 纯文本推理是否足够？
- 工具调用的收益是否大于成本？
Q2：工具数量有限制吗？
A：
- OpenAI：建议不超过 128 个工具
- Anthropic：建议不超过 64 个工具
- 工具过多会增加选择难度，建议按场景分组
Q3：如何处理工具调用失败？
A：
- 重试机制：自动重试 2-3 次
- 参数修复：使用 LLM 分析错误并修正
- 降级策略：尝试替代工具或方法
- 用户通知：明确告知失败原因
Q4：如何提高工具调用的准确性？
A：
- 优化工具描述，提供详细的使用场景
- 使用枚举限制参数值
- 提供示例调用
- 记录成功案例用于 Few-shot
Q5：工具调用有安全风险吗？
A：主要风险包括：
- 恶意参数注入
- 未授权的敏感操作
- 数据泄露
建议措施：
- 实施权限控制
- 参数校验和清理
- 操作审计日志
- 敏感操作需人工确认
小结
工具调用是 Agent 的核心能力，让 Agent 从「会说」变成「会做」。
核心要点回顾：
┌─────────────────────────────────────────────────────────────┐│ 工具调用核心总结 │├─────────────────────────────────────────────────────────────┤│ ││ 核心原理：LLM 生成结构化参数 → 执行函数 → 返回结果 ││ ││ Schema 设计：清晰命名、详细描述、合理参数、类型安全 ││ ││ 选择策略：语义匹配、历史成功率、用户权限 ││ ││ 协同模式：并行调用、链式调用、条件分支 ││ ││ 最佳实践：错误处理、安全控制、审计日志 ││ │└─────────────────────────────────────────────────────────────┘下一步学习：
下篇预告
《Agent 多智能体协作：从单体到群体智能》
深入解析：
- 多 Agent 协作的必要性
- 流水线、层级、对等三种协作模式
- CrewAI 多 Agent 框架实战
- Agent 间通信与协调
参考资料
支持与分享
如果这篇文章对你有帮助，欢迎支持作者或分享给更多人
Agent 工具调用：Function Calling 与 Tool Use 实战
https://blog.souloss.com/posts/machine-learning/agent-guide/tool-use/ 部分信息可能已经过时
相关文章 智能推荐
1
让AI使用工具：Function Calling实战
AI 让AI使用工具——Function Calling实战
2
MCP 协议解析：Agent 的工具标准
AI MCP（Model Context Protocol）协议深度解析——协议架构、消息格式、工具定义与发现机制，以及与 Function Calling、OpenAPI 的对比分析。
3
Agent 可观测性：日志、追踪与调试
AI 深度解读 Agent 可观测性——Langfuse、OpenTelemetry 追踪、LangSmith 等工具
4
AI Agent 实战指南
AI AI Agent 实战指南系列——从基础概念到项目实战，手把手教你构建智能体应用，涵盖最新大模型架构创新。
5
Agent 安全：提示注入与防御
AI 深度解读 Agent 安全——提示注入攻击、工具投毒、防御策略