Agent工具系统 - 自定义工具开发
**前置知识**：需要先掌握 AI Agent基础
前置知识：需要先掌握 AI Agent基础
本文重点：Agent工具开发与集成
一、工具系统概述
Agent工具类型：
1. 搜索工具
- 网络搜索
- 数据库查询
- API调用
2. 执行工具
- 代码执行
- 文件操作
- 系统命令
3. 分析工具
- 数据分析
- 图像处理
- 文本处理
4. 自定义工具
- 业务逻辑封装
- 领域特定功能二、LangChain工具开发
2.1 使用装饰器
from langchain.tools import tool
from typing import List, Optional
@tool
def search_web(query: str) -> str:
"""搜索网络获取信息
Args:
query: 搜索关键词
Returns:
搜索结果摘要
"""
# 实际实现
import requests
response = requests.get(f"https://api.example.com/search?q={query}")
return response.json().get("summary", "未找到结果")
@tool
def calculate(expression: str) -> str:
"""计算数学表达式
Args:
expression: 数学表达式，如 "2 + 3 * 4"
Returns:
计算结果
"""
try:
result = eval(expression)
return str(result)
except Exception as e:
return f"计算错误: {str(e)}"
@tool
def query_database(sql: str) -> str:
"""执行SQL查询
Args:
sql: SQL查询语句
Returns:
查询结果JSON
"""
# 示例：连接SQLite
import sqlite3
import json
conn = sqlite3.connect("app.db")
cursor = conn.cursor()
cursor.execute(sql)
columns = [desc[0] for desc in cursor.description]
results = [dict(zip(columns, row)) for row in cursor.fetchall()]
conn.close()
return json.dumps(results, ensure_ascii=False)2.2 使用Tool类
from langchain.tools import Tool
def get_weather(city: str) -> str:
"""获取天气"""
# 模拟天气API
weather_data = {
"北京": "晴，25°C",
"上海": "多云，28°C",
"深圳": "雨，30°C"
}
return weather_data.get(city, "未知城市")
weather_tool = Tool(
name="get_weather",
description="获取指定城市的天气信息。输入城市名称。",
func=get_weather
)三、OpenAI Function Calling
3.1 定义工具
import json
from openai import OpenAI
client = OpenAI()
# 定义工具
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如北京、上海"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索网络信息",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
}
}
]
def get_weather(city):
return f"{city}今天天气晴朗，温度25度"
def search_web(query):
return f"关于'{query}'的搜索结果..."
# 工具映射
tool_functions = {
"get_weather": get_weather,
"search_web": search_web
}3.2 执行工具调用
def run_with_tools(user_message):
messages = [{"role": "user", "content": user_message}]
# 第一次调用
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
# 检查是否需要调用工具
if message.tool_calls:
# 执行工具
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
# 调用函数
result = tool_functions[func_name](**func_args)
# 添加工具结果
messages.append(message)
messages.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": func_name,
"content": result
})
# 再次调用获取最终回答
final_response = client.chat.completions.create(
model="gpt-4",
messages=messages
)
return final_response.choices[0].message.content
return message.content
# 使用
result = run_with_tools("北京今天天气怎么样？")
print(result)四、复杂工具示例
4.1 文件操作工具
@tool
def read_file(file_path: str) -> str:
"""读取文件内容
Args:
file_path: 文件路径
Returns:
文件内容
"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
return f.read()
except Exception as e:
return f"读取失败: {str(e)}"
@tool
def write_file(file_path: str, content: str) -> str:
"""写入文件
Args:
file_path: 文件路径
content: 要写入的内容
Returns:
操作结果
"""
try:
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
return f"成功写入 {file_path}"
except Exception as e:
return f"写入失败: {str(e)}"4.2 数据分析工具
@tool
def analyze_csv(file_path: str, analysis_type: str) -> str:
"""分析CSV文件
Args:
file_path: CSV文件路径
analysis_type: 分析类型 (summary/correlation/distribution)
Returns:
分析结果
"""
import pandas as pd
import json
df = pd.read_csv(file_path)
if analysis_type == "summary":
result = df.describe().to_dict()
elif analysis_type == "correlation":
result = df.corr().to_dict()
else:
result = {"columns": list(df.columns), "shape": list(df.shape)}
return json.dumps(result, default=str)五、工具安全与限制
class SafeToolExecutor:
"""安全的工具执行器"""
def __init__(self):
self.allowed_functions = {}
self.banned_operations = ["rm", "delete", "format", "shutdown"]
def register_tool(self, name, func, description):
"""注册工具"""
self.allowed_functions[name] = {
"func": func,
"description": description
}
def execute(self, name, **kwargs):
"""执行工具"""
if name not in self.allowed_functions:
return f"工具 '{name}' 不存在"
# 检查参数
for key, value in kwargs.items():
if isinstance(value, str):
for banned in self.banned_operations:
if banned in value.lower():
return f"禁止的操作: {banned}"
try:
return self.allowed_functions[name]["func"](**kwargs)
except Exception as e:
return f"执行错误: {str(e)}"参考资源
返回：AI Agent智能体 最后更新: 2026年4月20日
讨论与反馈