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

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

---

🎯 本文目标

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

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

---

📚 工具基础

什么是工具 (Tool)？

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

作用：

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

---

工具定义格式

from google.adk import Tool

# 简单工具
def search_web(query: str) -> str:
    """搜索网络信息"""
    return f"搜索结果：{query}"

tool = Tool(
    name="web_search",
    func=search_web,
    description="当用户需要查询实时信息时使用"
)

---

🔧 常用工具实现

1. 网络搜索工具

依赖：

pip install duckduckgo-search

实现：

from google.adk import Tool
from duckduckgo_search import DDGS

def search_web(query: str, num_results: int = 5) -> str:
    """
    搜索网络获取最新信息
    
    参数:
        query: 搜索关键词
        num_results: 返回结果数量
    
    返回:
        搜索结果摘要
    """
    results = []
    with DDGS() as ddgs:
        for r in ddgs.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. 计算器工具

import math
from google.adk import Tool

def calculate(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)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

calc_tool = Tool(
    name="calculator",
    func=calculate,
    description="执行数学计算，包括基础运算和科学计算"
)

---

3. API 调用工具

示例：天气查询

import requests
from google.adk import Tool

def get_weather(city: str) -> str:
    """
    查询城市天气
    
    参数:
        city: 城市名称
    
    返回:
        天气信息
    """
    url = f"http://wttr.in/{city}?format=3"
    try:
        response = requests.get(url, timeout=5)
        if response.status_code == 200:
            return response.text.strip()
        else:
            return f"天气查询失败：{response.status_code}"
    except Exception as e:
        return f"天气查询错误：{str(e)}"

weather_tool = Tool(
    name="weather",
    func=get_weather,
    description="查询指定城市的当前天气"
)

示例：股票价格

def get_stock_price(symbol: str) -> str:
    """
    查询股票价格
    
    参数:
        symbol: 股票代码 (如：AAPL, 600519)
    
    返回:
        股票价格信息
    """
    url = f"https://api.example.com/stock/{symbol}"
    response = requests.get(url)
    data = response.json()
    
    return f"{symbol} 当前价格：{data['price']}元，涨跌幅：{data['change']}%"

stock_tool = Tool(
    name="stock_price",
    func=get_stock_price,
    description="查询 A 股、美股的实时股票价格"
)

---

4. 文件操作工具

import os
from pathlib import Path
from google.adk import Tool

def read_file(file_path: str) -> str:
    """读取文件内容"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        return f"读取失败：{str(e)}"

def write_file(file_path: str, content: str) -> str:
    """写入文件内容"""
    try:
        Path(file_path).parent.mkdir(parents=True, exist_ok=True)
        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)}"

def list_files(directory: str) -> str:
    """列出目录内容"""
    try:
        files = os.listdir(directory)
        return "\n".join(files)
    except Exception as e:
        return f"列出失败：{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. 数据库查询工具

import sqlite3
from google.adk import Tool

def query_database(sql: str) -> str:
    """
    执行 SQL 查询（只读）
    
    参数:
        sql: SELECT 语句
    
    返回:
        查询结果
    """
    # 安全检查
    if not sql.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) for row in results]
        return "\n".join(output)
    except Exception as e:
        return f"查询错误：{str(e)}"

db_tool = Tool(
    name="database_query",
    func=query_database,
    description="查询数据库获取数据，仅支持 SELECT 语句"
)

---

💻 综合案例

案例 1：数据分析助手

from google.adk import Agent

# 创建工具集
data_tools = [
    web_search_tool,      # 搜索数据
    calc_tool,            # 计算统计
    read_file,            # 读取 CSV
    query_database        # 查询数据库
]

# 创建 Agent
data_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)

def analyze_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. 条件工具调用

def smart_search(query: str) -> str:
    """智能搜索：根据查询类型选择工具"""
    
    # 判断查询类型
    if any(k in query.lower() for k in ["天气", "weather"]):
        return get_weather(query.replace("天气", "").strip())
    elif any(k in query.lower() for k in ["股价", "股票", "stock"]):
        return get_stock_price(query)
    else:
        return search_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. 工具安全性

# ❌ 危险：直接执行用户输入
def dangerous_exec(code: str):
    return eval(code)

# ✅ 安全：限制可用函数
def safe_calculate(expression: str):
    safe_dict = {"sqrt": math.sqrt, "pi": math.pi}
    return eval(expression, {"__builtins__": {}}, safe_dict)

2. 错误处理

def robust_tool(param: str) -> str:
    try:
        result = do_something(param)
        return f"成功：{result}"
    except TimeoutError:
        return "错误：操作超时"
    except ConnectionError:
        return "错误：网络连接失败"
    except Exception as e:
        return f"错误：{str(e)}"

3. 性能考虑

# 添加超时限制
from functools import wraps

def timeout(seconds=5):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            import signal
            def handler(signum, frame):
                raise TimeoutError()
            signal.signal(signal.SIGALRM, handler)
            signal.alarm(seconds)
            try:
                return func(*args, **kwargs)
            finally:
                signal.alarm(0)
        return wrapper
    return decorator

@timeout(seconds=10)
def api_call(url: str) -> str:
    return requests.get(url).text

---

📚 系列导航

期数	主题	状态
第 3 期	工具调用能力	✅
第 4 期	多 Agent 协作	下一篇

---

觉得有用？点赞 👍 收藏 ⭐ 关注 ➕ 三连支持一下！