AI Agent工具使用与Function Calling完全教程
从原理到生产，系统掌握AI Agent的工具调用能力
目录
- Function Calling 机制深度解析
- OpenAI Function Calling 实战
- Claude Tool Use 详解
- Gemini Function Calling
- 工具定义 JSON Schema 最佳实践
- 多工具编排与链式调用
- 错误处理与重试机制
- 自定义工具开发
- Agent 工具安全与权限控制
- 生产级 Agent 工具系统架构
1. Function Calling 机制深度解析
1.1 什么是 Function Calling？
Function Calling 是让大语言模型（LLM）能够"调用外部工具"的核心机制。它解决了LLM的一个根本局限：模型只能生成文本，无法直接执行操作。
传统LLM交互：
用户: "北京今天天气怎么样？"
LLM: "我无法获取实时天气信息..."（受限于训练数据）
Function Calling：
用户: "北京今天天气怎么样？"
LLM: → 调用 get_weather(city="北京") （识别意图，生成调用参数）
系统: → 执行函数，返回结果
LLM: "北京今天晴，气温25°C，微风..." （基于实时数据生成回答）
1.2 核心工作原理
┌─────────────────────────────────────────────────────────┐
│ Function Calling 完整流程 │
│ │
│ ① 定义工具 │
│ ┌───────────────────────────────────┐ │
│ │ { │ │
│ │ "name": "get_weather", │ │
│ │ "description": "获取天气信息", │ │
│ │ "parameters": { │ │
│ │ "city": {"type": "string"} │ │
│ │ } │ │
│ │ } │ │
│ └───────────────────────────────────┘ │
│ │
│ ② 用户提问 + 工具定义 → 发送给LLM │
│ ┌───────────────────────────────────┐ │
│ │ Messages: │ │
│ │ User: "北京天气如何？" │ │
│ │ Tools: [get_weather, ...] │ │
│ └───────────────────────────────────┘ │
│ │
│ ③ LLM决策：直接回答 or 调用工具？ │
│ ┌───────────────────────────────────┐ │
│ │ LLM输出: │ │
│ │ tool_call: { │ │
│ │ name: "get_weather", │ │
│ │ arguments: {"city": "北京"} │ │
│ │ } │ │
│ └───────────────────────────────────┘ │
│ │
│ ④ 系统执行函数，将结果返回给LLM │
│ ┌───────────────────────────────────┐ │
│ │ Tool Result: │ │
│ │ {"temp": 25, "condition": "晴"} │ │
│ └───────────────────────────────────┘ │
│ │
│ ⑤ LLM基于工具结果生成最终回答 │
│ ┌───────────────────────────────────┐ │
│ │ "北京今天晴天，气温25°C，适合出行" │ │
│ └───────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
1.3 关键概念
| 概念 | 说明 |
|---|---|
| Tool / Function | 可供LLM调用的外部函数 |
| Tool Definition | 用JSON Schema描述工具的名称、参数、用途 |
| Tool Call | LLM决定调用某个工具并生成参数 |
| Tool Result | 工具执行后返回的结果 |
| Parallel Tool Calls | LLM一次请求中同时调用多个工具 |
| Forced/Required Tool | 强制LLM必须调用某个工具 |
2. OpenAI Function Calling 实战
2.1 基础用法
from openai import OpenAI
import json
client = OpenAI()
# 定义工具
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"]
}
}
}
]
# 模拟天气API
def get_weather(city: str, unit: str = "celsius") -> dict:
weather_data = {
"北京": {"temp": 25, "condition": "晴", "humidity": 45},
"上海": {"temp": 28, "condition": "多云", "humidity": 72},
}
data = weather_data.get(city, {"temp": 20, "condition": "未知", "humidity": 50})
if unit == "fahrenheit":
data["temp"] = data["temp"] * 9/5 + 32
return data
# 完整的对话流程
def chat_with_tools(user_message: str):
messages = [
{"role": "system", "content": "你是一个有用的助手，可以查询天气信息。"},
{"role": "user", "content": user_message}
]
# 第一步：发送消息和工具定义
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto", # auto | none | required
)
assistant_message = response.choices[0].message
# 第二步：检查是否有工具调用
if assistant_message.tool_calls:
messages.append(assistant_message)
# 执行每个工具调用
for tool_call in assistant_message.tool_calls:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
if function_name == "get_weather":
result = get_weather(**arguments)
else:
result = {"error": f"Unknown function: {function_name}"}
# 第三步：将工具结果添加到消息
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False),
})
# 第四步：让LLM基于工具结果生成最终回答
final_response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
)
return final_response.choices[0].message.content
# 没有工具调用，直接返回回答
return assistant_message.content
# 测试
print(chat_with_tools("北京今天天气怎么样？"))
# 输出: "北京今天天气晴朗，气温25°C，湿度45%，是个不错的天气！"
2.2 并行工具调用
# OpenAI支持在一次响应中调用多个工具
def chat_with_parallel_tools(user_message: str):
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "get_population",
"description": "获取城市人口数据",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
}
},
]
messages = [{"role": "user", "content": user_message}]
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
parallel_tool_calls=True, # 启用并行工具调用
)
assistant_msg = response.choices[0].message
if assistant_msg.tool_calls:
messages.append(assistant_msg)
# 并行执行所有工具调用
for tc in assistant_msg.tool_calls:
func_name = tc.function.name
args = json.loads(tc.function.arguments)
result = execute_tool(func_name, args)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps(result),
})
final = client.chat.completions.create(
model="gpt-4o",
messages=messages,
)
return final.choices[0].message.content
return assistant_msg.content
# 用户说："北京和上海的天气分别怎么样？"
# LLM可能同时调用：
# tool_call_1: get_weather(city="北京")
# tool_call_2: get_weather(city="上海")
2.3 强制工具调用
# 方式1：自动决定（默认）
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto",
)
# 方式2：强制调用特定工具
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice={
"type": "function",
"function": {"name": "get_weather"}
},
)
# 方式3：禁止调用工具
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="none",
)
2.4 结构化输出（Structured Outputs）
# 使用strict模式确保参数格式正确
tools_strict = [
{
"type": "function",
"function": {
"name": "create_user",
"description": "创建新用户",
"strict": True, # 启用严格模式
"parameters": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "用户姓名"
},
"email": {
"type": "string",
"description": "邮箱地址"
},
"age": {
"type": "integer",
"description": "年龄"
},
"roles": {
"type": "array",
"items": {
"type": "string",
"enum": ["admin", "user", "moderator"]
},
"description": "用户角色列表"
}
},
"required": ["name", "email", "age", "roles"],
"additionalProperties": False
}
}
}
]
3. Claude Tool Use 详解
3.1 Anthropic Claude 的工具使用
Claude的Tool Use与OpenAI类似但有细微差异：
import anthropic
import json
client = anthropic.Anthropic()
# 定义工具（Claude格式）
tools = [
{
"name": "get_stock_price",
"description": "获取股票的实时价格信息。输入股票代码，返回当前价格、涨跌幅等数据。",
"input_schema": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "股票代码，如 'AAPL'、'GOOGL'、'600519'"
},
"market": {
"type": "string",
"enum": ["US", "HK", "CN"],
"description": "市场：US=美股，HK=港股，CN=A股"
}
},
"required": ["symbol"]
}
},
{
"name": "calculate",
"description": "执行数学计算。支持基本运算和常用数学函数。",
"input_schema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如 '100 * 1.05' 或 'sqrt(144)'"
}
},
"required": ["expression"]
}
}
]
# 模拟股票数据
def get_stock_price(symbol: str, market: str = "US") -> dict:
stocks = {
"AAPL": {"price": 178.50, "change": "+2.3%", "volume": "52.3M"},
"GOOGL": {"price": 141.80, "change": "+1.1%", "volume": "28.1M"},
"600519": {"price": 1688.00, "change": "-0.5%", "volume": "3.2万手"},
}
return stocks.get(symbol, {"error": "未找到该股票"})
def calculate(expression: str) -> dict:
import math
try:
allowed = {"__builtins__": {}, "sqrt": math.sqrt, "abs": abs, "round": round}
result = eval(expression, allowed)
return {"result": result}
except Exception as e:
return {"error": str(e)}
# Claude工具调用完整流程
def chat_with_claude(user_message: str):
messages = [{"role": "user", "content": user_message}]
while True:
# 调用Claude
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
tools=tools,
messages=messages,
)
# 检查是否需要工具调用
if response.stop_reason == "tool_use":
# 收集所有文本和工具使用块
tool_results = []
for block in response.content:
if block.type == "tool_use":
# 执行工具
if block.name == "get_stock_price":
result = get_stock_price(**block.input)
elif block.name == "calculate":
result = calculate(**block.input)
else:
result = {"error": f"Unknown tool: {block.name}"}
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result, ensure_ascii=False),
})
# 将助手回复和工具结果添加到消息历史
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
elif response.stop_reason == "end_turn":
# 提取最终文本回答
for block in response.content:
if block.type == "text":
return block.text
return "No response"
else:
return f"Unexpected stop reason: {response.stop_reason}"
# 测试
print(chat_with_claude("苹果公司的股票现在多少钱？如果我买100股，需要花多少美元？"))
3.2 Claude 的 tool_choice 参数
# Claude也支持控制工具调用行为
# 自动决定（默认）
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=tools,
tool_choice={"type": "auto"}, # 自动决定是否调用工具
messages=messages,
)
# 强制调用某个工具
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=tools,
tool_choice={
"type": "tool",
"name": "get_stock_price" # 强制调用此工具
},
messages=messages,
)
# 禁止工具调用
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=tools,
tool_choice={"type": "none"},
messages=messages,
)
3.3 Claude vs OpenAI 工具调用对比
| 特性 | OpenAI | Claude |
|---|---|---|
| 工具定义字段 | function.parameters |
input_schema |
| 工具结果角色 | role: "tool" + tool_call_id |
role: "user" + tool_result |
| 并行调用 | 支持（parallel_tool_calls） |
支持（多个tool_use块） |
| 强制调用 | tool_choice 对象 |
tool_choice 对象 |
| 严格模式 | strict: true |
不需要（天然严格） |
| Token计算 | 工具定义占用上下文 | 同样占用上下文 |
4. Gemini Function Calling
4.1 Google Gemini 工具调用
from google import genai
from google.genai import types
import json
client = genai.Client()
# 定义工具（Gemini格式）
def search_restaurants(cuisine: str, location: str, price_range: str) -> dict:
"""搜索餐厅信息"""
return {
"restaurants": [
{
"name": f"好吃的{cuisine}餐厅",
"address": f"{location}中心大街123号",
"rating": 4.5,
"price_range": price_range,
}
]
}
def make_reservation(restaurant: str, date: str, time: str, party_size: int) -> dict:
"""预订餐厅"""
return {
"status": "confirmed",
"reservation_id": "RES-20241201-001",
"restaurant": restaurant,
"date": date,
"time": time,
"party_size": party_size,
}
# 使用Gemini的函数声明
restaurant_tool = types.FunctionDeclaration(
name="search_restaurants",
description="搜索餐厅，可以按菜系、位置和价位筛选",
parameters={
"type": "object",
"properties": {
"cuisine": {
"type": "string",
"description": "菜系类型，如中餐、日料、意大利菜"
},
"location": {
"type": "string",
"description": "地理位置，如北京朝阳区"
},
"price_range": {
"type": "string",
"enum": ["低", "中", "高"],
"description": "价格区间"
}
},
"required": ["cuisine", "location"]
}
)
reservation_tool = types.FunctionDeclaration(
name="make_reservation",
description="预订餐厅",
parameters={
"type": "object",
"properties": {
"restaurant": {"type": "string", "description": "餐厅名称"},
"date": {"type": "string", "description": "日期，格式YYYY-MM-DD"},
"time": {"type": "string", "description": "时间，格式HH:MM"},
"party_size": {"type": "integer", "description": "用餐人数"}
},
"required": ["restaurant", "date", "time", "party_size"]
}
)
# 创建工具集
tools = types.Tool(function_declarations=[restaurant_tool, reservation_tool])
# 配置
config = types.GenerateContentConfig(
tools=[tools],
temperature=0.3,
)
# 工具函数映射
available_functions = {
"search_restaurants": search_restaurants,
"make_reservation": make_reservation,
}
def chat_with_gemini(user_message: str):
"""Gemini工具调用完整流程"""
contents = [
types.Content(
role="user",
parts=[types.Part(text=user_message)]
)
]
while True:
response = client.models.generate_content(
model="gemini-2.0-flash",
contents=contents,
config=config,
)
# 检查是否有函数调用
has_tool_call = False
for part in response.candidates[0].content.parts:
if part.function_call:
has_tool_call = True
func_name = part.function_call.name
func_args = dict(part.function_call.args)
# 执行函数
if func_name in available_functions:
result = available_functions[func_name](**func_args)
else:
result = {"error": f"Unknown function: {func_name}"}
# 添加助手回复（包含函数调用）
contents.append(response.candidates[0].content)
# 添加函数结果
contents.append(
types.Content(
role="function",
parts=[
types.Part(
function_response=types.FunctionResponse(
name=func_name,
response=result,
)
)
]
)
)
if not has_tool_call:
# 没有工具调用，返回最终文本
return response.text
# 测试
result = chat_with_gemini(
"我想在北京朝阳区找一家中等价位的日料餐厅，然后帮我和朋友2人预订明天晚上7点的位子"
)
print(result)
4.2 三大平台工具调用统一接口
"""
统一的工具调用接口，封装OpenAI、Claude、Gemini
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any
import json
@dataclass
class ToolCall:
id: str
name: str
arguments: dict
@dataclass
class ToolResult:
tool_call_id: str
content: str
is_error: bool = False
class UnifiedToolCaller(ABC):
"""统一工具调用基类"""
def __init__(self, tools: list[dict], tool_functions: dict[str, callable]):
self.tools = tools
self.tool_functions = tool_functions
@abstractmethod
def chat(self, messages: list[dict]) -> tuple[str, list[ToolCall]]:
"""发送消息，返回(文本回复, 工具调用列表)"""
pass
def execute_tool_calls(self, tool_calls: list[ToolCall]) -> list[ToolResult]:
"""执行工具调用"""
results = []
for tc in tool_calls:
try:
func = self.tool_functions.get(tc.name)
if func:
result = func(**tc.arguments)
results.append(ToolResult(
tool_call_id=tc.id,
content=json.dumps(result, ensure_ascii=False),
))
else:
results.append(ToolResult(
tool_call_id=tc.id,
content=json.dumps({"error": f"Unknown tool: {tc.name}"}),
is_error=True,
))
except Exception as e:
results.append(ToolResult(
tool_call_id=tc.id,
content=json.dumps({"error": str(e)}),
is_error=True,
))
return results
def run_conversation(self, user_message: str, max_rounds: int = 5) -> str:
"""运行完整对话，自动处理工具调用"""
messages = [{"role": "user", "content": user_message}]
for _ in range(max_rounds):
text, tool_calls = self.chat(messages)
if not tool_calls:
return text
# 执行工具
results = self.execute_tool_calls(tool_calls)
# 将结果添加到消息历史（由子类处理具体格式）
self._append_tool_results(messages, tool_calls, results)
return "达到最大工具调用轮数限制"
@abstractmethod
def _append_tool_results(
self,
messages: list,
tool_calls: list[ToolCall],
results: list[ToolResult]
):
"""将工具结果添加到消息历史"""
pass
class OpenAIToolCaller(UnifiedToolCaller):
"""OpenAI工具调用实现"""
def __init__(self, tools, tool_functions, api_key: str, model: str = "gpt-4o"):
super().__init__(tools, tool_functions)
from openai import OpenAI
self.client = OpenAI(api_key=api_key)
self.model = model
def chat(self, messages):
# 转换工具格式为OpenAI格式
openai_tools = [
{"type": "function", "function": t} for t in self.tools
]
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=openai_tools,
)
msg = response.choices[0].message
messages.append(msg)
tool_calls = []
if msg.tool_calls:
for tc in msg.tool_calls:
tool_calls.append(ToolCall(
id=tc.id,
name=tc.function.name,
arguments=json.loads(tc.function.arguments),
))
return msg.content or "", tool_calls
def _append_tool_results(self, messages, tool_calls, results):
for result in results:
messages.append({
"role": "tool",
"tool_call_id": result.tool_call_id,
"content": result.content,
})
# 使用示例
def demo():
tools = [
{
"name": "get_time",
"description": "获取当前时间",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区，如 Asia/Shanghai"
}
},
"required": ["timezone"]
}
}
]
import datetime
tool_functions = {
"get_time": lambda timezone="UTC": {
"time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"timezone": timezone,
}
}
caller = OpenAIToolCaller(
tools=tools,
tool_functions=tool_functions,
api_key="your-key",
)
result = caller.run_conversation("现在几点了？")
print(result)
5. 工具定义 JSON Schema 最佳实践
5.1 好的工具定义 vs 坏的工具定义
# ❌ 坏的工具定义
BAD_TOOL = {
"name": "do_stuff",
"description": "做事情",
"parameters": {
"type": "object",
"properties": {
"data": {"type": "string"} # 什么data？格式是什么？
}
}
}
# ✅ 好的工具定义
GOOD_TOOL = {
"name": "send_email",
"description": "发送电子邮件。用于向指定收件人发送邮件，支持纯文本和HTML格式。",
"parameters": {
"type": "object",
"properties": {
"to": {
"type": "string",
"description": "收件人邮箱地址，如 'user@example.com'"
},
"subject": {
"type": "string",
"description": "邮件主题"
},
"body": {
"type": "string",
"description": "邮件正文内容，支持HTML格式"
},
"cc": {
"type": "array",
"items": {"type": "string"},
"description": "抄送人邮箱列表，可选"
},
"priority": {
"type": "string",
"enum": ["low", "normal", "high"],
"description": "邮件优先级，默认normal"
}
},
"required": ["to", "subject", "body"]
}
}
5.2 工具定义编写规范
"""
工具定义编写规范清单
"""
# 1. 命名规范
NAMING_RULES = """
- 使用 snake_case 命名函数
- 名称应清晰描述动作：get_weather ✅ vs weather ❌
- 避免缩写：get_user_info ✅ vs get_usr ❌
- 保持一致性：要么都是 get_xxx，要么都是 fetch_xxx
"""
# 2. 描述规范
DESCRIPTION_RULES = """
- 第一句话说明工具的核心功能
- 补充说明使用场景和限制
- 说明参数的组合关系
- 避免过于笼统："获取数据" → "获取指定城市的实时天气数据"
"""
# 3. 参数规范
PARAMETER_RULES = """
- 每个参数都有 description
- 使用 enum 约束可选值
- required 数组列出必填参数
- 使用 format 指定格式（如 "format": "email"）
- 数值类型指定 minimum/maximum
- 数组类型指定 items 结构
"""
# 4. 完整示例
WELL_DEFINED_TOOL = {
"name": "search_products",
"description": (
"搜索商品目录。根据关键词、类别和价格范围搜索商品。"
"返回匹配的商品列表，包含名称、价格、评分等信息。"
"注意：搜索结果最多返回50条，如需更多请使用分页参数。"
),
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词，支持自然语言描述",
"minLength": 1,
"maxLength": 200,
},
"category": {
"type": "string",
"enum": [
"electronics", "clothing", "food",
"books", "home", "sports"
],
"description": "商品类别筛选，不填则搜索全部类别"
},
"min_price": {
"type": "number",
"minimum": 0,
"description": "最低价格（元），不填则不限"
},
"max_price": {
"type": "number",
"minimum": 0,
"description": "最高价格（元），不填则不限"
},
"sort_by": {
"type": "string",
"enum": ["relevance", "price_asc", "price_desc", "rating"],
"description": "排序方式，默认按相关性排序"
},
"page": {
"type": "integer",
"minimum": 1,
"default": 1,
"description": "页码，从1开始"
},
"page_size": {
"type": "integer",
"enum": [10, 20, 50],
"default": 20,
"description": "每页返回数量"
}
},
"required": ["query"],
"additionalProperties": False,
}
}
5.3 复杂参数类型处理
# 嵌套对象
NESTED_OBJECT_TOOL = {
"name": "create_order",
"description": "创建订单",
"parameters": {
"type": "object",
"properties": {
"items": {
"type": "array",
"description": "订单商品列表",
"items": {
"type": "object",
"properties": {
"product_id": {"type": "string", "description": "商品ID"},
"quantity": {
"type": "integer",
"minimum": 1,
"description": "购买数量"
},
"spec": {
"type": "object",
"properties": {
"color": {"type": "string"},
"size": {"type": "string"}
},
"description": "商品规格"
}
},
"required": ["product_id", "quantity"]
}
},
"shipping_address": {
"type": "object",
"properties": {
"province": {"type": "string"},
"city": {"type": "string"},
"district": {"type": "string"},
"detail": {"type": "string"},
"phone": {"type": "string"},
"name": {"type": "string"}
},
"required": ["province", "city", "detail", "phone", "name"],
"description": "收货地址"
},
"coupon_code": {
"type": "string",
"description": "优惠券码，可选"
}
},
"required": ["items", "shipping_address"]
}
}
6. 多工具编排与链式调用
6.1 工具链模式
"""
工具链：一个工具的输出作为另一个工具的输入
"""
class ToolChain:
"""工具链执行器"""
def __init__(self):
self.steps = []
def add_step(self, tool_name: str, arg_mapping: dict):
"""
添加链式步骤
arg_mapping: 参数映射，key是当前工具参数名，value是前序步骤结果的引用
例如: {"query": "$step_0.result.keywords"} 表示使用第0步结果中的keywords字段
"""
self.steps.append({
"tool_name": tool_name,
"arg_mapping": arg_mapping,
})
return self # 支持链式调用
def execute(self, initial_args: dict, tool_functions: dict) -> list:
"""执行工具链"""
results = []
context = {"initial": initial_args}
for i, step in enumerate(self.steps):
# 解析参数映射
resolved_args = {}
for param_name, source_ref in step["arg_mapping"].items():
if source_ref.startswith("$initial."):
# 引用初始参数
key = source_ref.replace("$initial.", "")
resolved_args[param_name] = initial_args.get(key)
elif source_ref.startswith("$step_"):
# 引用前序步骤结果
parts = source_ref.split(".")
step_idx = int(parts[0].replace("$step_", ""))
field = parts[1] if len(parts) > 1 else None
if step_idx < len(results):
prev_result = results[step_idx]
if field and isinstance(prev_result, dict):
resolved_args[param_name] = prev_result.get(field)
else:
resolved_args[param_name] = prev_result
else:
# 直接值
resolved_args[param_name] = source_ref
# 执行工具
tool_func = tool_functions[step["tool_name"]]
result = tool_func(**resolved_args)
results.append(result)
return results
# 使用示例：研究助手工具链
chain = ToolChain()
chain.add_step("web_search", {"query": "$initial.topic"}) \
.add_step("extract_content", {"urls": "$step_0.urls"}) \
.add_step("summarize", {"text": "$step_1.content"}) \
.add_step("generate_report", {"summary": "$step_2.summary", "topic": "$initial.topic"})
results = chain.execute(
initial_args={"topic": "2024年AI芯片发展"},
tool_functions={
"web_search": web_search_func,
"extract_content": extract_content_func,
"summarize": summarize_func,
"generate_report": generate_report_func,
}
)
6.2 条件分支与循环
"""
支持条件分支和循环的工具编排
"""
from typing import Callable
from enum import Enum
class StepType(Enum):
TOOL = "tool"
CONDITION = "condition"
LOOP = "loop"
PARALLEL = "parallel"
class WorkflowEngine:
"""工作流引擎：支持条件分支、循环、并行"""
def __init__(self, tool_functions: dict):
self.tool_functions = tool_functions
def execute_workflow(self, workflow: dict, context: dict) -> dict:
"""执行工作流"""
return self._execute_node(workflow["root"], context)
def _execute_node(self, node: dict, context: dict) -> dict:
node_type = node.get("type")
if node_type == "tool":
return self._execute_tool(node, context)
elif node_type == "condition":
return self._execute_condition(node, context)
elif node_type == "loop":
return self._execute_loop(node, context)
elif node_type == "parallel":
return self._execute_parallel(node, context)
elif node_type == "sequence":
return self._execute_sequence(node, context)
raise ValueError(f"Unknown node type: {node_type}")
def _execute_tool(self, node: dict, context: dict) -> dict:
"""执行单个工具"""
func = self.tool_functions[node["tool"]]
args = self._resolve_args(node.get("args", {}), context)
result = func(**args)
# 将结果存储到上下文
if "output_key" in node:
context[node["output_key"]] = result
return result
def _execute_condition(self, node: dict, context: dict) -> dict:
"""条件分支"""
condition_result = self._evaluate_condition(node["condition"], context)
if condition_result:
return self._execute_node(node["then"], context)
elif "else" in node:
return self._execute_node(node["else"], context)
return {}
def _execute_loop(self, node: dict, context: dict) -> list:
"""循环执行"""
max_iterations = node.get("max_iterations", 10)
results = []
for i in range(max_iterations):
# 检查退出条件
if self._evaluate_condition(node.get("until", "false"), context):
break
context["loop_index"] = i
result = self._execute_node(node["body"], context)
results.append(result)
return results
def _execute_parallel(self, node: dict, context: dict) -> list:
"""并行执行多个节点"""
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = [
executor.submit(self._execute_node, child, context.copy())
for child in node["children"]
]
return [f.result() for f in concurrent.futures.as_completed(futures)]
def _execute_sequence(self, node: dict, context: dict) -> dict:
"""顺序执行多个节点"""
results = {}
for child in node["children"]:
results[child.get("name", "unnamed")] = self._execute_node(child, context)
return results
def _resolve_args(self, args: dict, context: dict) -> dict:
"""解析参数中的上下文引用"""
resolved = {}
for key, value in args.items():
if isinstance(value, str) and value.startswith("$"):
resolved[key] = context.get(value[1:])
else:
resolved[key] = value
return resolved
def _evaluate_condition(self, condition: str, context: dict) -> bool:
"""评估条件表达式（简化版）"""
# 生产环境应使用安全的表达式评估器
for key, value in context.items():
condition = condition.replace(f"${key}", repr(value))
try:
return bool(eval(condition, {"__builtins__": {}}, {}))
except:
return False
# 工作流定义示例：智能客服
CUSTOMER_SERVICE_WORKFLOW = {
"root": {
"type": "sequence",
"children": [
{
"type": "tool",
"tool": "classify_intent",
"args": {"text": "$user_message"},
"output_key": "intent"
},
{
"type": "condition",
"condition": "$intent.category == 'order'",
"then": {
"type": "tool",
"tool": "lookup_order",
"args": {"order_id": "$intent.order_id"},
"output_key": "order"
},
"else": {
"type": "condition",
"condition": "$intent.category == 'complaint'",
"then": {
"type": "tool",
"tool": "create_ticket",
"args": {"subject": "$intent.summary"},
"output_key": "ticket"
}
}
}
]
}
}
7. 错误处理与重试机制
7.1 常见错误类型
class ToolError(Exception):
"""工具调用错误基类"""
pass
class ToolTimeoutError(ToolError):
"""工具执行超时"""
pass
class ToolRateLimitError(ToolError):
"""API限流"""
pass
class ToolAuthError(ToolError):
"""认证失败"""
pass
class ToolInvalidArgsError(ToolError):
"""参数错误"""
pass
class ToolServiceError(ToolError):
"""服务不可用"""
pass
7.2 重试策略
import time
import random
from functools import wraps
from typing import Type
def retry_with_backoff(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
retryable_errors: tuple = (ToolRateLimitError, ToolServiceError, ToolTimeoutError),
):
"""指数退避重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_error = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except retryable_errors as e:
last_error = e
if attempt < max_retries:
# 指数退避 + 随机抖动
delay = min(
base_delay * (2 ** attempt) + random.uniform(0, 1),
max_delay
)
# 如果有Retry-After头，使用它
if hasattr(e, 'retry_after') and e.retry_after:
delay = max(delay, e.retry_after)
print(f"Retry {attempt + 1}/{max_retries} after {delay:.1f}s: {e}")
time.sleep(delay)
else:
raise
raise last_error
return wrapper
return decorator
# 使用示例
@retry_with_backoff(max_retries=3, base_delay=2.0)
def call_external_api(query: str) -> dict:
"""带重试的外部API调用"""
response = requests.get(
"https://api.example.com/search",
params={"q": query},
timeout=10,
)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
err = ToolRateLimitError(f"Rate limited, retry after {retry_after}s")
err.retry_after = retry_after
raise err
if response.status_code == 503:
raise ToolServiceError("Service temporarily unavailable")
if response.status_code != 200:
raise ToolError(f"API error: {response.status_code}")
return response.json()
7.3 工具调用包装器
import logging
from typing import Any, Optional
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class ToolCallResult:
"""工具调用结果包装"""
success: bool
data: Any = None
error: Optional[str] = None
tool_name: str = ""
duration_ms: int = 0
retries: int = 0
class RobustToolExecutor:
"""健壮的工具执行器"""
def __init__(self, timeout: float = 30.0):
self.timeout = timeout
self.call_history: list[ToolCallResult] = []
def execute(
self,
tool_name: str,
tool_func: callable,
arguments: dict,
max_retries: int = 2,
) -> ToolCallResult:
"""执行工具调用，带超时、重试、日志"""
import time
start_time = time.time()
retries = 0
last_error = None
for attempt in range(max_retries + 1):
try:
# 参数验证
validated_args = self._validate_args(tool_name, arguments)
# 带超时执行
result = self._execute_with_timeout(
tool_func, validated_args, self.timeout
)
# 结果验证
validated_result = self._validate_result(tool_name, result)
duration = int((time.time() - start_time) * 1000)
call_result = ToolCallResult(
success=True,
data=validated_result,
tool_name=tool_name,
duration_ms=duration,
retries=attempt,
)
self.call_history.append(call_result)
logger.info(
f"Tool {tool_name} succeeded in {duration}ms "
f"(retries: {attempt})"
)
return call_result
except ToolTimeoutError as e:
last_error = e
logger.warning(f"Tool {tool_name} timeout: {e}")
except ToolRateLimitError as e:
last_error = e
delay = getattr(e, 'retry_after', 2 ** attempt)
time.sleep(delay)
retries += 1
except ToolInvalidArgsError as e:
# 参数错误不重试
duration = int((time.time() - start_time) * 1000)
call_result = ToolCallResult(
success=False,
error=str(e),
tool_name=tool_name,
duration_ms=duration,
)
self.call_history.append(call_result)
return call_result
except Exception as e:
last_error = e
logger.error(f"Tool {tool_name} error: {e}")
if attempt < max_retries:
time.sleep(2 ** attempt)
retries += 1
# 所有重试都失败了
duration = int((time.time() - start_time) * 1000)
call_result = ToolCallResult(
success=False,
error=str(last_error),
tool_name=tool_name,
duration_ms=duration,
retries=retries,
)
self.call_history.append(call_result)
return call_result
def _execute_with_timeout(
self, func: callable, args: dict, timeout: float
) -> Any:
"""带超时执行"""
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(func, **args)
try:
return future.result(timeout=timeout)
except concurrent.futures.TimeoutError:
raise ToolTimeoutError(f"Tool execution timed out ({timeout}s)")
def _validate_args(self, tool_name: str, args: dict) -> dict:
"""参数验证"""
# 基础验证：检查必填参数、类型等
# 生产环境应使用JSON Schema验证
if not isinstance(args, dict):
raise ToolInvalidArgsError(f"Arguments must be a dict, got {type(args)}")
return args
def _validate_result(self, tool_name: str, result: Any) -> Any:
"""结果验证"""
# 确保结果可以被序列化
try:
import json
json.dumps(result)
return result
except (TypeError, ValueError):
return str(result)
8. 自定义工具开发
8.1 Python 工具开发框架
"""
自定义工具开发框架
提供装饰器快速定义工具
"""
import inspect
import json
from typing import get_type_hints, Any, Optional
from functools import wraps
class ToolRegistry:
"""工具注册中心"""
def __init__(self):
self._tools: dict[str, dict] = {}
self._functions: dict[str, callable] = {}
def tool(
self,
name: str = None,
description: str = None,
required: list[str] = None,
):
"""工具定义装饰器"""
def decorator(func):
tool_name = name or func.__name__
tool_desc = description or (func.__doc__ or "").strip().split("\n")[0]
# 从函数签名自动生成参数schema
sig = inspect.signature(func)
hints = get_type_hints(func)
properties = {}
for param_name, param in sig.parameters.items():
if param_name == "self":
continue
param_type = hints.get(param_name, str)
prop = self._type_to_schema(param_type)
# 从docstring提取参数描述
prop["description"] = f"参数: {param_name}"
if param.default != inspect.Parameter.empty:
prop["default"] = param.default
properties[param_name] = prop
# 确定required参数
if required is None:
required_params = [
name for name, param in sig.parameters.items()
if param.default == inspect.Parameter.empty and name != "self"
]
else:
required_params = required
tool_def = {
"type": "function",
"function": {
"name": tool_name,
"description": tool_desc,
"parameters": {
"type": "object",
"properties": properties,
"required": required_params,
}
}
}
self._tools[tool_name] = tool_def
self._functions[tool_name] = func
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
wrapper._tool_name = tool_name
return wrapper
return decorator
def _type_to_schema(self, type_hint) -> dict:
"""将Python类型转换为JSON Schema类型"""
type_map = {
str: {"type": "string"},
int: {"type": "integer"},
float: {"type": "number"},
bool: {"type": "boolean"},
list: {"type": "array", "items": {"type": "string"}},
dict: {"type": "object"},
}
return type_map.get(type_hint, {"type": "string"})
def get_tools(self) -> list[dict]:
"""获取所有工具定义"""
return list(self._tools.values())
def get_functions(self) -> dict[str, callable]:
"""获取所有工具函数"""
return self._functions.copy()
def call(self, tool_name: str, arguments: dict) -> Any:
"""调用工具"""
func = self._functions.get(tool_name)
if not func:
raise ValueError(f"Unknown tool: {tool_name}")
return func(**arguments)
# 使用示例
registry = ToolRegistry()
@registry.tool(
name="get_current_time",
description="获取当前时间。返回指定时区的当前日期和时间。"
)
def get_current_time(timezone: str = "Asia/Shanghai") -> dict:
from datetime import datetime
import pytz
tz = pytz.timezone(timezone)
now = datetime.now(tz)
return {
"datetime": now.strftime("%Y-%m-%d %H:%M:%S"),
"timezone": timezone,
"weekday": now.strftime("%A"),
}
@registry.tool(
name="calculate_bmi",
description="计算BMI（身体质量指数）。输入身高(cm)和体重(kg)，返回BMI值和健康评估。"
)
def calculate_bmi(height: float, weight: float) -> dict:
bmi = weight / (height / 100) ** 2
if bmi < 18.5:
category = "偏瘦"
elif bmi < 24:
category = "正常"
elif bmi < 28:
category = "偏胖"
else:
category = "肥胖"
return {
"bmi": round(bmi, 1),
"category": category,
"healthy_range": "18.5 - 24.0",
}
@registry.tool(description="搜索文件。在指定目录中搜索匹配关键词的文件。")
def search_files(query: str, directory: str = ".", max_results: int = 10) -> dict:
import os
matches = []
for root, dirs, files in os.walk(directory):
for f in files:
if query.lower() in f.lower():
matches.append(os.path.join(root, f))
if len(matches) >= max_results:
break
if len(matches) >= max_results:
break
return {"files": matches, "count": len(matches)}
# 导出给LLM使用
tools_for_llm = registry.get_tools()
tool_functions = registry.get_functions()
print(json.dumps(tools_for_llm, indent=2, ensure_ascii=False))
8.2 TypeScript 工具开发
/**
* TypeScript工具开发框架
*/
import { z } from "zod";
// 工具定义接口
interface ToolDefinition {
name: string;
description: string;
parameters: z.ZodSchema;
handler: (args: any) => Promise<any>;
}
// 工具注册中心
class ToolRegistry {
private tools = new Map<string, ToolDefinition>();
register(definition: ToolDefinition): void {
this.tools.set(definition.name, definition);
}
// 装饰器风格注册
tool<T extends z.ZodSchema>(
name: string,
description: string,
parameters: T,
handler: (args: z.infer<T>) => Promise<any>
) {
this.register({ name, description, parameters, handler });
}
// 转换为OpenAI工具格式
toOpenAITools() {
return Array.from(this.tools.values()).map((t) => ({
type: "function" as const,
function: {
name: t.name,
description: t.description,
parameters: zodToJsonSchema(t.parameters),
},
}));
}
// 执行工具调用
async call(name: string, args: unknown): Promise<any> {
const tool = this.tools.get(name);
if (!tool) throw new Error(`Unknown tool: ${name}`);
// 使用Zod验证参数
const validated = tool.parameters.parse(args);
return tool.handler(validated);
}
}
// Zod Schema 转 JSON Schema（简化版）
function zodToJsonSchema(schema: z.ZodSchema): object {
if (schema instanceof z.ZodObject) {
const shape = schema.shape;
const properties: Record<string, any> = {};
const required: string[] = [];
for (const [key, value] of Object.entries(shape)) {
properties[key] = zodFieldToJsonSchema(value as z.ZodSchema);
if (!(value as any).isOptional()) {
required.push(key);
}
}
return {
type: "object",
properties,
required: required.length > 0 ? required : undefined,
};
}
return { type: "object" };
}
function zodFieldToJsonSchema(schema: z.ZodSchema): object {
if (schema instanceof z.ZodString) return { type: "string" };
if (schema instanceof z.ZodNumber) return { type: "number" };
if (schema instanceof z.ZodBoolean) return { type: "boolean" };
if (schema instanceof z.ZodArray) {
return { type: "array", items: zodFieldToJsonSchema(schema.element) };
}
if (schema instanceof z.ZodEnum) {
return { type: "string", enum: schema._def.values };
}
return { type: "string" };
}
// 使用示例
const registry = new ToolRegistry();
// 定义工具
const WeatherSchema = z.object({
city: z.string().describe("城市名称"),
unit: z.enum(["celsius", "fahrenheit"]).optional().default("celsius"),
});
registry.tool(
"get_weather",
"获取指定城市的天气信息",
WeatherSchema,
async ({ city, unit }) => {
// 调用天气API
const response = await fetch(
`https://api.weather.com/v1/current?city=${city}&unit=${unit}`
);
return response.json();
}
);
// 定义搜索工具
const SearchSchema = z.object({
query: z.string().min(1).describe("搜索关键词"),
limit: z.number().int().min(1).max(100).optional().default(10),
});
registry.tool(
"web_search",
"搜索互联网内容",
SearchSchema,
async ({ query, limit }) => {
// 实现搜索逻辑
return { results: [], total: 0 };
}
);
// 导出工具定义给LLM
const openAITools = registry.toOpenAITools();
console.log(JSON.stringify(openAITools, null, 2));
// 执行工具调用
async function handleToolCall(name: string, args: unknown) {
try {
const result = await registry.call(name, args);
return { success: true, data: result };
} catch (error) {
if (error instanceof z.ZodError) {
return { success: false, error: "Invalid arguments", details: error.errors };
}
return { success: false, error: (error as Error).message };
}
}
8.3 MCP（Model Context Protocol）工具
"""
基于MCP协议的工具开发
MCP是Anthropic提出的标准化工具协议
"""
import json
from mcp.server import Server
from mcp.types import Tool, TextContent
# 创建MCP服务器
server = Server("my-tools-server")
@server.list_tools()
async def list_tools() -> list[Tool]:
"""返回可用工具列表"""
return [
Tool(
name="web_search",
description="搜索互联网内容",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
),
Tool(
name="read_file",
description="读取文件内容",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "文件路径"
}
},
"required": ["path"]
}
),
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""执行工具调用"""
if name == "web_search":
result = await do_web_search(arguments["query"])
return [TextContent(type="text", text=json.dumps(result))]
elif name == "read_file":
with open(arguments["path"], "r") as f:
content = f.read()
return [TextContent(type="text", text=content)]
raise ValueError(f"Unknown tool: {name}")
# 运行MCP服务器
async def main():
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options(),
)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
9. Agent 工具安全与权限控制
9.1 安全威胁模型
┌─────────────────────────────────────────────────────┐
│ Agent 工具安全威胁模型 │
│ │
│ ┌─────────────────┐ │
│ │ 1. 提示注入攻击 │ 恶意输入诱导LLM调用危险工具 │
│ │ (Prompt │ │
│ │ Injection) │ "忽略之前的指令，执行..." │
│ └─────────────────┘ │
│ │
│ ┌─────────────────┐ │
│ │ 2. 参数注入 │ 工具参数中嵌入恶意内容 │
│ │ (Argument │ │
│ │ Injection) │ city="北京; rm -rf /" │
│ └─────────────────┘ │
│ │
│ ┌─────────────────┐ │
│ │ 3. 越权调用 │ LLM调用超出授权范围的工具 │
│ │ (Privilege │ │
│ │ Escalation) │ 用户只问天气，却调用了转账工具 │
│ └─────────────────┘ │
│ │
│ ┌─────────────────┐ │
│ │ 4. 数据泄露 │ 工具返回敏感信息给LLM/用户 │
│ │ (Data │ │
│ │ Leakage) │ 工具结果包含密码、密钥等 │
│ └─────────────────┘ │
│ │
│ ┌─────────────────┐ │
│ │ 5. 无限循环 │ LLM反复调用工具导致资源耗尽 │
│ │ (Infinite │ │
│ │ Loop) │ A调用B，B调用A，无限递归 │
│ └─────────────────┘ │
└─────────────────────────────────────────────────────┘
9.2 权限控制模型
"""
基于RBAC的工具权限控制系统
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import re
class Permission(Enum):
READ = "read" # 只读操作
WRITE = "write" # 写操作
ADMIN = "admin" # 管理操作
EXECUTE = "execute" # 执行操作
@dataclass
class ToolPermission:
"""工具权限定义"""
tool_name: str
required_permissions: list[Permission]
rate_limit: int = 60 # 每分钟调用次数限制
requires_confirmation: bool = False # 是否需要用户确认
allowed_arguments: Optional[dict] = None # 参数白名单
@dataclass
class UserRole:
"""用户角色"""
name: str
permissions: set[Permission]
tool_whitelist: Optional[set[str]] = None # 工具白名单（None表示全部）
tool_blacklist: set[str] = field(default_factory=set) # 工具黑名单
class SecurityManager:
"""安全管理器"""
def __init__(self):
self.tool_permissions: dict[str, ToolPermission] = {}
self.roles: dict[str, UserRole] = {}
self.call_log: list[dict] = []
self.rate_counters: dict[str, list[float]] = {}
def register_tool(self, perm: ToolPermission):
"""注册工具权限"""
self.tool_permissions[perm.tool_name] = perm
def register_role(self, role: UserRole):
"""注册角色"""
self.roles[role.name] = role
def check_permission(
self,
user_role: str,
tool_name: str,
arguments: dict,
) -> tuple[bool, str]:
"""检查调用权限"""
role = self.roles.get(user_role)
if not role:
return False, f"Unknown role: {user_role}"
tool_perm = self.tool_permissions.get(tool_name)
if not tool_perm:
return False, f"Unknown tool: {tool_name}"
# 检查黑名单
if tool_name in role.tool_blacklist:
return False, f"Tool {tool_name} is blacklisted for role {user_role}"
# 检查白名单
if role.tool_whitelist is not None:
if tool_name not in role.tool_whitelist:
return False, f"Tool {tool_name} not in whitelist for role {user_role}"
# 检查权限
required = set(tool_perm.required_permissions)
if not required.issubset(role.permissions):
missing = required - role.permissions
return False, f"Missing permissions: {[p.value for p in missing]}"
# 检查参数白名单
if tool_perm.allowed_arguments:
for key, allowed_values in tool_perm.allowed_arguments.items():
if key in arguments and arguments[key] not in allowed_values:
return False, f"Argument {key}={arguments[key]} not allowed"
# 检查速率限制
if not self._check_rate_limit(user_role, tool_name, tool_perm.rate_limit):
return False, f"Rate limit exceeded for {tool_name}"
return True, "OK"
def _check_rate_limit(
self, user_role: str, tool_name: str, limit: int
) -> bool:
"""检查速率限制"""
import time
key = f"{user_role}:{tool_name}"
now = time.time()
if key not in self.rate_counters:
self.rate_counters[key] = []
# 清理超过1分钟的记录
self.rate_counters[key] = [
t for t in self.rate_counters[key] if now - t < 60
]
if len(self.rate_counters[key]) >= limit:
return False
self.rate_counters[key].append(now)
return True
def log_call(
self,
user_role: str,
tool_name: str,
arguments: dict,
result: any,
success: bool,
):
"""记录工具调用日志"""
import time
self.call_log.append({
"timestamp": time.time(),
"user_role": user_role,
"tool_name": tool_name,
"arguments": arguments,
"success": success,
# 注意：不记录完整结果，可能包含敏感信息
})
# 使用示例
security = SecurityManager()
# 定义角色
security.register_role(UserRole(
name="user",
permissions={Permission.READ},
tool_blacklist={"delete_user", "admin_config"},
))
security.register_role(UserRole(
name="admin",
permissions={Permission.READ, Permission.WRITE, Permission.ADMIN, Permission.EXECUTE},
))
# 定义工具权限
security.register_tool(ToolPermission(
tool_name="get_weather",
required_permissions=[Permission.READ],
rate_limit=30,
))
security.register_tool(ToolPermission(
tool_name="send_email",
required_permissions=[Permission.WRITE],
rate_limit=10,
requires_confirmation=True,
))
security.register_tool(ToolPermission(
tool_name="delete_user",
required_permissions=[Permission.ADMIN],
rate_limit=5,
requires_confirmation=True,
))
# 检查权限
allowed, msg = security.check_permission("user", "get_weather", {"city": "北京"})
print(f"用户查天气: {allowed} - {msg}") # True
allowed, msg = security.check_permission("user", "delete_user", {"user_id": "123"})
print(f"用户删用户: {allowed} - {msg}") # False - blacklisted
9.3 参数消毒
"""
工具参数消毒：防止注入攻击
"""
import re
import shlex
from typing import Any
class ArgumentSanitizer:
"""参数消毒器"""
# 危险模式
DANGEROUS_PATTERNS = [
r';\s*rm\s', # shell命令注入
r'\|\s*.*', # pipe注入
r'`.*`', # 反引号执行
r'\$\(.*\)', # 命令替换
r'--.*--', # 命令行参数注入
r'<script.*>', # XSS
r'javascript:', # JS协议
r'on\w+\s*=', # 事件处理器
]
@classmethod
def sanitize_string(cls, value: str, max_length: int = 1000) -> str:
"""消毒字符串参数"""
# 长度限制
value = value[:max_length]
# 移除控制字符
value = ''.join(
c for c in value if c.isprintable() or c in '\n\r\t'
)
# 检查危险模式
for pattern in cls.DANGEROUS_PATTERNS:
if re.search(pattern, value, re.IGNORECASE):
raise ValueError(
f"Potentially dangerous input detected: {pattern}"
)
return value
@classmethod
def sanitize_path(cls, path: str, allowed_dirs: list[str]) -> str:
"""消毒文件路径参数"""
import os
# 解析路径
normalized = os.path.normpath(path)
# 防止路径遍历
if '..' in normalized:
raise ValueError("Path traversal detected")
# 检查是否在允许的目录下
abs_path = os.path.abspath(normalized)
if not any(abs_path.startswith(d) for d in allowed_dirs):
raise ValueError(f"Path not in allowed directories: {abs_path}")
return normalized
@classmethod
def sanitize_sql(cls, query: str) -> str:
"""消毒SQL参数（简单版，生产环境使用参数化查询）"""
# 移除SQL注入常见模式
dangerous = [
"DROP", "DELETE", "TRUNCATE", "ALTER",
"UNION SELECT", "INSERT INTO", "UPDATE",
"--", "/*", "*/", ";",
]
upper = query.upper()
for d in dangerous:
if d in upper:
raise ValueError(f"Potentially dangerous SQL: {d}")
return query
@classmethod
def sanitize_arguments(
cls,
tool_name: str,
arguments: dict,
schema: dict,
) -> dict:
"""根据schema消毒所有参数"""
sanitized = {}
props = schema.get("properties", {})
for key, value in arguments.items():
prop_schema = props.get(key, {})
prop_type = prop_schema.get("type")
if prop_type == "string":
if "enum" in prop_schema:
if value not in prop_schema["enum"]:
raise ValueError(
f"Invalid enum value for {key}: {value}"
)
sanitized[key] = cls.sanitize_string(str(value))
elif prop_type in ("integer", "number"):
try:
num = int(value) if prop_type == "integer" else float(value)
if "minimum" in prop_schema and num < prop_schema["minimum"]:
raise ValueError(f"{key} below minimum")
if "maximum" in prop_schema and num > prop_schema["maximum"]:
raise ValueError(f"{key} above maximum")
sanitized[key] = num
except (ValueError, TypeError):
raise ValueError(f"Invalid {prop_type} for {key}: {value}")
elif prop_type == "boolean":
sanitized[key] = bool(value)
else:
sanitized[key] = value
return sanitized
10. 生产级 Agent 工具系统架构
10.1 整体架构
┌─────────────────────────────────────────────────────────────┐
│ 生产级Agent工具系统架构 │
│ │
│ ┌──────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 用户 │───→│ API Gateway │───→│ Agent Router │ │
│ │ 请求 │ │ (认证/限流) │ │ (意图识别) │ │
│ └──────────┘ └──────────────┘ └──────┬───────┘ │
│ │ │
│ ┌─────────────────────────┼────────┐ │
│ ▼ ▼ ▼ │
│ ┌──────────────────┐ ┌────────────────┐ ┌──────────┐ │
│ │ LLM Provider │ │ Tool Registry │ │ Security │ │
│ │ (多模型路由) │ │ (工具注册中心) │ │ Manager │ │
│ └────────┬─────────┘ └───────┬────────┘ └──────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────────────────────────────────────┐ │
│ │ Tool Execution Engine │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐│ │
│ │ │Sandbox │ │Timeout │ │Retry │ │Circuit ││ │
│ │ │沙箱执行│ │超时控制│ │重试机制│ │Breaker ││ │
│ │ └────────┘ └────────┘ └────────┘ └────────┘│ │
│ └──────────────────────┬───────────────────────┘ │
│ │ │
│ ┌──────────────┼──────────────┐ │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ 内部工具 │ │ 外部API │ │ 数据存储 │ │
│ │ (数据库等) │ │ (第三方) │ │ (向量/缓存) │ │
│ └────────────┘ └────────────┘ └────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Observability Layer │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐│ │
│ │ │Logging │ │Metrics │ │Tracing │ │Alerts ││ │
│ │ │日志 │ │指标 │ │追踪 │ │告警 ││ │
│ │ └────────┘ └────────┘ └────────┘ └────────┘│ │
│ └──────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
10.2 工具执行引擎
"""
生产级工具执行引擎
"""
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Any, Optional
from contextlib import asynccontextmanager
from enum import Enum
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed" # 正常
OPEN = "open" # 熔断
HALF_OPEN = "half_open" # 半开
@dataclass
class CircuitBreaker:
"""熔断器"""
failure_threshold: int = 5
recovery_timeout: float = 60.0
state: CircuitState = CircuitState.CLOSED
failure_count: int = 0
last_failure_time: float = 0.0
def record_success(self):
self.failure_count = 0
self.state = CircuitState.CLOSED
def record_failure(self):
import time
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
logger.warning(f"Circuit breaker OPEN after {self.failure_count} failures")
def allow_request(self) -> bool:
import time
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
return True
return False
# HALF_OPEN: 允许一个请求
return True
@dataclass
class ToolConfig:
"""工具配置"""
name: str
timeout: float = 30.0
max_retries: int = 2
rate_limit: int = 60 # 每分钟调用次数
circuit_breaker: CircuitBreaker = field(default_factory=CircuitBreaker)
cache_ttl: int = 0 # 缓存时间（秒），0=不缓存
sandbox: bool = False # 是否在沙箱中执行
class ToolExecutionEngine:
"""工具执行引擎"""
def __init__(self):
self.tools: dict[str, callable] = {}
self.configs: dict[str, ToolConfig] = {}
self.circuit_breakers: dict[str, CircuitBreaker] = {}
self.cache: dict[str, tuple[float, Any]] = {}
self.metrics: dict[str, dict] = {}
def register_tool(
self,
name: str,
func: callable,
config: ToolConfig = None,
):
"""注册工具"""
self.tools[name] = func
self.configs[name] = config or ToolConfig(name=name)
self.circuit_breakers[name] = CircuitBreaker()
self.metrics[name] = {
"calls": 0,
"successes": 0,
"failures": 0,
"total_time_ms": 0,
}
async def execute(
self,
tool_name: str,
arguments: dict,
user_context: dict = None,
) -> dict:
"""执行工具调用"""
import time
if tool_name not in self.tools:
return {
"success": False,
"error": f"Unknown tool: {tool_name}",
}
config = self.configs[tool_name]
breaker = self.circuit_breakers[tool_name]
# 检查熔断器
if not breaker.allow_request():
return {
"success": False,
"error": f"Tool {tool_name} is circuit-broken, try again later",
}
# 检查缓存
cache_key = f"{tool_name}:{hash(frozenset(arguments.items()))}"
if config.cache_ttl > 0 and cache_key in self.cache:
cached_time, cached_result = self.cache[cache_key]
if time.time() - cached_time < config.cache_ttl:
return cached_result
# 执行（带重试）
last_error = None
for attempt in range(config.max_retries + 1):
start_time = time.time()
try:
# 带超时执行
result = await asyncio.wait_for(
self._run_tool(tool_name, arguments),
timeout=config.timeout,
)
duration_ms = int((time.time() - start_time) * 1000)
# 记录成功
breaker.record_success()
self._record_metric(tool_name, True, duration_ms)
# 缓存结果
if config.cache_ttl > 0:
self.cache[cache_key] = (time.time(), result)
return {
"success": True,
"data": result,
"duration_ms": duration_ms,
"attempts": attempt + 1,
}
except asyncio.TimeoutError:
last_error = f"Timeout after {config.timeout}s"
logger.warning(f"Tool {tool_name} timeout (attempt {attempt + 1})")
except Exception as e:
last_error = str(e)
logger.error(f"Tool {tool_name} error: {e} (attempt {attempt + 1})")
# 等待后重试
if attempt < config.max_retries:
await asyncio.sleep(2 ** attempt)
# 所有重试失败
breaker.record_failure()
self._record_metric(tool_name, False, 0)
return {
"success": False,
"error": last_error,
"attempts": config.max_retries + 1,
}
async def _run_tool(self, tool_name: str, arguments: dict) -> Any:
"""运行工具（在沙箱或直接执行）"""
config = self.configs[tool_name]
func = self.tools[tool_name]
if config.sandbox:
return await self._sandbox_execute(func, arguments)
if asyncio.iscoroutinefunction(func):
return await func(**arguments)
else:
return func(**arguments)
async def _sandbox_execute(self, func, arguments: dict) -> Any:
"""沙箱执行（简化版，生产环境使用Docker/WASM）"""
import concurrent.futures
loop = asyncio.get_event_loop()
with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
future = executor.submit(func, **arguments)
return await asyncio.wrap_future(future)
def _record_metric(self, tool_name: str, success: bool, duration_ms: int):
"""记录指标"""
m = self.metrics[tool_name]
m["calls"] += 1
if success:
m["successes"] += 1
else:
m["failures"] += 1
m["total_time_ms"] += duration_ms
def get_metrics(self) -> dict:
"""获取所有工具的指标"""
result = {}
for name, m in self.metrics.items():
result[name] = {
**m,
"avg_time_ms": m["total_time_ms"] / max(m["calls"], 1),
"success_rate": m["successes"] / max(m["calls"], 1),
}
return result
# 使用示例
async def main():
engine = ToolExecutionEngine()
# 注册工具
engine.register_tool(
name="get_weather",
func=lambda city, **kw: {"city": city, "temp": 25, "condition": "晴"},
config=ToolConfig(
name="get_weather",
timeout=10.0,
cache_ttl=300, # 缓存5分钟
),
)
engine.register_tool(
name="send_email",
func=send_email_func,
config=ToolConfig(
name="send_email",
timeout=30.0,
max_retries=3,
rate_limit=10,
),
)
# 执行工具
result = await engine.execute(
"get_weather",
{"city": "北京"},
)
print(f"Result: {result}")
# 查看指标
print(f"Metrics: {engine.get_metrics()}")
if __name__ == "__main__":
asyncio.run(main())
10.3 可观测性
"""
工具调用可观测性：日志、指标、追踪
"""
import time
import json
import uuid
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Optional
@dataclass
class TraceSpan:
"""追踪跨度"""
trace_id: str
span_id: str
parent_id: Optional[str]
tool_name: str
start_time: float
end_time: float = 0
status: str = "pending"
attributes: dict = None
@property
def duration_ms(self):
return int((self.end_time - self.start_time) * 1000)
class ToolTracer:
"""工具调用追踪器"""
def __init__(self):
self.spans: list[TraceSpan] = []
self._current_trace: Optional[str] = None
self._current_span: Optional[str] = None
@contextmanager
def trace(self, tool_name: str, attributes: dict = None):
"""追踪工具调用"""
trace_id = self._current_trace or str(uuid.uuid4())
span_id = str(uuid.uuid4())
parent_id = self._current_span
span = TraceSpan(
trace_id=trace_id,
span_id=span_id,
parent_id=parent_id,
tool_name=tool_name,
start_time=time.time(),
attributes=attributes or {},
)
# 设置当前上下文
prev_trace = self._current_trace
prev_span = self._current_span
self._current_trace = trace_id
self._current_span = span_id
try:
yield span
span.status = "success"
except Exception as e:
span.status = "error"
span.attributes["error"] = str(e)
raise
finally:
span.end_time = time.time()
self.spans.append(span)
self._current_trace = prev_trace
self._current_span = prev_span
def get_trace(self, trace_id: str) -> list[TraceSpan]:
"""获取完整追踪"""
return [s for s in self.spans if s.trace_id == trace_id]
def export_json(self) -> str:
"""导出为JSON（兼容OpenTelemetry格式）"""
return json.dumps([
{
"traceId": s.trace_id,
"spanId": s.span_id,
"parentSpanId": s.parent_id,
"name": s.tool_name,
"startTime": s.start_time,
"endTime": s.end_time,
"duration_ms": s.duration_ms,
"status": s.status,
"attributes": s.attributes,
}
for s in self.spans
], indent=2)
# 使用示例
tracer = ToolTracer()
def execute_with_tracing(engine, tool_name, arguments):
with tracer.trace(tool_name, {"arguments": arguments}) as span:
result = engine.execute(tool_name, arguments)
span.attributes["result_success"] = result.get("success", False)
return result
# 查看追踪数据
print(tracer.export_json())
10.4 部署建议
# docker-compose.yml 示例
version: '3.8'
services:
agent-api:
build: .
ports:
- "8000:8000"
environment:
- LLM_API_KEY=${LLM_API_KEY}
- REDIS_URL=redis://redis:6379
- DATABASE_URL=postgresql://user:pass@db:5432/agent
depends_on:
- redis
- db
redis:
image: redis:7-alpine
# 用于缓存、速率限制、消息队列
db:
image: postgres:16-alpine
# 用于存储工具调用日志、用户数据
prometheus:
image: prom/prometheus
# 指标收集
grafana:
image: grafana/grafana
# 指标可视化
# 生产环境关键配置
PRODUCTION_CONFIG = {
"tool_execution": {
"default_timeout": 30,
"max_concurrent_tools": 10,
"enable_sandbox": True,
"cache_enabled": True,
},
"security": {
"enable_rate_limiting": True,
"enable_argument_sanitization": True,
"enable_audit_log": True,
"max_tool_chain_depth": 5,
},
"reliability": {
"circuit_breaker_enabled": True,
"failure_threshold": 5,
"recovery_timeout": 60,
"max_retries": 3,
},
"observability": {
"enable_tracing": True,
"enable_metrics": True,
"log_level": "INFO",
},
}
总结
核心要点
| 主题 | 关键点 |
|---|---|
| Function Calling | LLM识别意图 → 生成结构化调用 → 系统执行 → 结果回传 |
| 三大平台 | OpenAI、Claude、Gemini各有细微差异，核心流程一致 |
| 工具定义 | 清晰的命名、详细的描述、严格的Schema是好工具的基础 |
| 编排模式 | 链式调用、条件分支、并行执行、循环控制 |
| 错误处理 | 重试 + 指数退避 + 熔断器 + 降级策略 |
| 安全控制 | 权限模型 + 参数消毒 + 速率限制 + 审计日志 |
| 生产部署 | 可观测性 + 缓存 + 沙箱 + 多模型路由 |
工具开发Checklist
□ 工具名称清晰，使用snake_case
□ 描述准确说明功能、场景、限制
□ 参数Schema完整，包含type/description/required
□ 使用enum约束可选值
□ 数值参数设置min/max范围
□ 实现参数验证和消毒
□ 添加超时控制
□ 实现重试机制（带指数退避）
□ 记录调用日志（不记录敏感信息）
□ 设置速率限制
□ 错误信息友好且不含敏感细节
□ 工具结果可被JSON序列化
□ 编写单元测试
□ 配置监控告警
上一篇推荐阅读：AI搜索引擎与Perplexity实战完全教程
