当前位置：首页 > news >正文

给 Agent 添加工具调用能力：搜索/计算/API

news 2026/5/11 23:58:41

给 Agent 添加工具调用能力：搜索/计算/API

系列文章：《AI Agent 开发实战》第 3 期
难度等级：⭐⭐⭐⭐
预计耗时：45 分钟

🎯 本文目标

学会为 AI Agent 添加各种工具：

✅ 网络搜索
✅ 数学计算
✅ API 调用
✅ 文件操作
✅ 数据库查询

📚 工具基础

什么是工具 (Tool)？

定义：Agent 可以调用的外部函数

作用：

扩展 Agent 能力
执行具体操作
获取实时数据

工具定义格式

fromgoogle.adkimportTool# 简单工具defsearch_web(query:str)->str:"""搜索网络信息"""returnf"搜索结果：{query}"tool=Tool(name="web_search",func=search_web,description="当用户需要查询实时信息时使用")

🔧 常用工具实现

1. 网络搜索工具

依赖：

pipinstallduckduckgo-search

实现：

fromgoogle.adkimportToolfromduckduckgo_searchimportDDGSdefsearch_web(query:str,num_results:int=5)->str:""" 搜索网络获取最新信息 参数: query: 搜索关键词 num_results: 返回结果数量 返回: 搜索结果摘要 """results=[]withDDGS()asddgs:forrinddgs.text(query,max_results=num_results):results.append(f"标题：{r['title']}\n链接：{r['href']}\n摘要：{r['body']}")return"\n\n".join(results)# 创建工具web_search_tool=Tool(name="search",func=search_web,description="搜索网络获取最新信息、新闻、数据等")

使用：

agent=Agent(name="ResearchBot",tools=[web_search_tool])response=agent.run("帮我查一下 2026 年 AI 领域的最新进展")

2. 计算器工具

importmathfromgoogle.adkimportTooldefcalculate(expression:str)->str:""" 计算数学表达式 支持: - 加减乘除 (+, -, *, /) - 幂运算 (**) - 平方根 (sqrt) - 三角函数 (sin, cos, tan) 参数: expression: 数学表达式 返回: 计算结果 """try:# 安全计算环境safe_dict={"sqrt":math.sqrt,"sin":math.sin,"cos":math.cos,"tan":math.tan,"log":math.log,"pi":math.pi,"e":math.e}result=eval(expression,{"__builtins__":{}},safe_dict)returnf"计算结果：{result}"exceptExceptionase:returnf"计算错误：{str(e)}"calc_tool=Tool(name="calculator",func=calculate,description="执行数学计算，包括基础运算和科学计算")

3. API 调用工具

示例：天气查询

importrequestsfromgoogle.adkimportTooldefget_weather(city:str)->str:""" 查询城市天气 参数: city: 城市名称 返回: 天气信息 """url=f"http://wttr.in/{city}?format=3"try:response=requests.get(url,timeout=5)ifresponse.status_code==200:returnresponse.text.strip()else:returnf"天气查询失败：{response.status_code}"exceptExceptionase:returnf"天气查询错误：{str(e)}"weather_tool=Tool(name="weather",func=get_weather,description="查询指定城市的当前天气")

示例：股票价格

defget_stock_price(symbol:str)->str:""" 查询股票价格 参数: symbol: 股票代码 (如：AAPL, 600519) 返回: 股票价格信息 """url=f"https://api.example.com/stock/{symbol}"response=requests.get(url)data=response.json()returnf"{symbol}当前价格：{data['price']}元，涨跌幅：{data['change']}%"stock_tool=Tool(name="stock_price",func=get_stock_price,description="查询 A 股、美股的实时股票价格")

4. 文件操作工具

importosfrompathlibimportPathfromgoogle.adkimportTooldefread_file(file_path:str)->str:"""读取文件内容"""try:withopen(file_path,'r',encoding='utf-8')asf:returnf.read()exceptExceptionase:returnf"读取失败：{str(e)}"defwrite_file(file_path:str,content:str)->str:"""写入文件内容"""try:Path(file_path).parent.mkdir(parents=True,exist_ok=True)withopen(file_path,'w',encoding='utf-8')asf:f.write(content)returnf"文件已保存：{file_path}"exceptExceptionase:returnf"写入失败：{str(e)}"deflist_files(directory:str)->str:"""列出目录内容"""try:files=os.listdir(directory)return"\n".join(files)exceptExceptionase:returnf"列出失败：{str(e)}"file_tools=[Tool(name="read_file",func=read_file,description="读取文本文件内容"),Tool(name="write_file",func=write_file,description="写入内容到文件"),Tool(name="list_files",func=list_files,description="列出目录中的文件")]

5. 数据库查询工具

importsqlite3fromgoogle.adkimportTooldefquery_database(sql:str)->str:""" 执行 SQL 查询（只读） 参数: sql: SELECT 语句 返回: 查询结果 """# 安全检查ifnotsql.strip().upper().startswith("SELECT"):return"错误：只允许执行 SELECT 查询"try:conn=sqlite3.connect("database.db")cursor=conn.cursor()cursor.execute(sql)results=cursor.fetchall()conn.close()# 格式化输出output=[str(row)forrowinresults]return"\n".join(output)exceptExceptionase:returnf"查询错误：{str(e)}"db_tool=Tool(name="database_query",func=query_database,description="查询数据库获取数据，仅支持 SELECT 语句")

💻 综合案例

案例 1：数据分析助手

fromgoogle.adkimportAgent# 创建工具集data_tools=[web_search_tool,# 搜索数据calc_tool,# 计算统计read_file,# 读取 CSVquery_database# 查询数据库]# 创建 Agentdata_assistant=Agent(name="DataAnalyst",instruction="你是一个数据分析助手，帮助用户分析数据、生成报告",tools=data_tools)# 使用示例response=data_assistant.run(""" 帮我分析销售数据： 1. 读取 sales.csv 文件 2. 计算总销售额 3. 找出最畅销的产品 4. 生成简要报告 """)

案例 2：研究助手

research_tools=[web_search_tool,# 搜索文献weather_tool,# 查询天气stock_tool,# 查询股价write_file# 保存报告]research_assistant=Agent(name="ResearchAssistant",instruction="你是一个研究助手，帮助用户收集信息、整理资料",tools=research_tools,memory=Memory(max_turns=20))# 使用示例response=research_assistant.run(""" 研究主题：2026 年 AI 发展趋势 任务： 1. 搜索最新的 AI 新闻 2. 查找相关论文 3. 整理关键观点 4. 保存为研究报告 """)

🎓 高级技巧

1. 工具链 (Tool Chain)

defanalyze_and_report(data_file:str)->str:"""工具链：读取 → 分析 → 报告"""# 步骤 1：读取数据data=read_file(data_file)# 步骤 2：分析analysis=calculate_statistics(data)# 步骤 3：搜索对比数据industry_avg=search_web("行业平均水平")# 步骤 4：生成报告report=f""" # 数据分析报告 ## 数据概览{analysis}## 行业对比{industry_avg}## 结论 ... """# 步骤 5：保存write_file("report.md",report)return"报告已生成并保存"chain_tool=Tool(name="analyze_and_report",func=analyze_and_report,description="完整的数据分析流程：读取→分析→对比→生成报告")

2. 条件工具调用

defsmart_search(query:str)->str:"""智能搜索：根据查询类型选择工具"""# 判断查询类型ifany(kinquery.lower()forkin["天气","weather"]):returnget_weather(query.replace("天气","").strip())elifany(kinquery.lower()forkin["股价","股票","stock"]):returnget_stock_price(query)else:returnsearch_web(query)smart_tool=Tool(name="smart_search",func=smart_search,description="智能搜索：自动识别查询类型并调用相应工具")

3. 工具组合使用

# Agent 自动决定使用哪些工具agent=Agent(name="SuperAssistant",tools=[web_search_tool,calc_tool,weather_tool,stock_tool,read_file,write_file])# Agent 会根据任务自动选择合适的工具response=agent.run(""" 帮我做以下事情： 1. 查询北京的天气 2. 搜索 AI 相关新闻 3. 计算 1234 * 5678 4. 把结果保存到 result.txt """)# Agent 会自动调用：weather → search → calculate → write_file

⚠️ 注意事项

1. 工具安全性

# ❌ 危险：直接执行用户输入defdangerous_exec(code:str):returneval(code)# ✅ 安全：限制可用函数defsafe_calculate(expression:str):safe_dict={"sqrt":math.sqrt,"pi":math.pi}returneval(expression,{"__builtins__":{}},safe_dict)

2. 错误处理

defrobust_tool(param:str)->str:try:result=do_something(param)returnf"成功：{result}"exceptTimeoutError:return"错误：操作超时"exceptConnectionError:return"错误：网络连接失败"exceptExceptionase:returnf"错误：{str(e)}"

3. 性能考虑

# 添加超时限制fromfunctoolsimportwrapsdeftimeout(seconds=5):defdecorator(func):@wraps(func)defwrapper(*args,**kwargs):importsignaldefhandler(signum,frame):raiseTimeoutError()signal.signal(signal.SIGALRM,handler)signal.alarm(seconds)try:returnfunc(*args,**kwargs)finally:signal.alarm(0)returnwrapperreturndecorator@timeout(seconds=10)defapi_call(url:str)->str:returnrequests.get(url).text