工具调用实战

工具（Tools）是 AI Agent 与外部世界交互的桥梁。通过工具，Agent 可以搜索信息、执行代码、调用 API，真正"行动"起来。

为什么需要工具

大模型本身有局限性：

知识有截止日期
无法访问实时数据
不能执行任意代码
无法操作外部系统

工具让 AI 能够：

获取实时信息（搜索、API）
执行计算（代码解释器）
操作文件系统
与第三方服务交互

工具设计原则

1. 清晰的描述

工具的名称和描述要明确，让模型知道何时使用：

python

# ❌ 不好：描述模糊
def search(query):
    """搜索"""
    pass

# ✅ 好：描述清晰
def search_weather(query):
    """搜索天气信息。输入：城市名称（如"北京"），输出：该城市的当前天气"""
    pass

2. 简洁的输入输出

python

# ❌ 复杂参数
def send_email(to, cc, bcc, subject, body, attachments, priority, ...):
    pass

# ✅ 简单直接
def send_email(to, subject, body):
    """发送邮件。参数：收件人、主题、正文"""
    pass

3. 单一职责

每个工具只做一件事：

python

# ❌ 一个工具做多件事
def handle_request(request):
    if request.type == "weather":
        return get_weather(request.city)
    elif request.type == "news":
        return get_news(request.topic)
    ...

# ✅ 拆分为独立工具
def get_weather(city):
    """查询城市天气"""

def search_news(topic):
    """搜索新闻"""

常用工具类型

1. 搜索工具

python

def web_search(query):
    """搜索互联网信息。用于获取实时数据、新闻等。"""
    import requests
    # 实际实现可接入 Google、Bing API 等
    return f"搜索结果：{query}"

def wiki_search(query):
    """搜索维基百科。用于获取权威词条信息。"""
    pass

2. 计算工具

python

def calculate(expression):
    """数学计算。用于精确的数学运算。"""
    import ast
    import operator

    ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
    }

    def eval_expr(node):
        if isinstance(node, ast.Constant):
            return node.value
        elif isinstance(node, ast.BinOp):
            left = eval_expr(node.left)
            right = eval_expr(node.right)
            return ops[type(node.op)](left, right)

    tree = ast.parse(expression, mode='eval')
    return str(eval_expr(tree.body))

3. 代码执行工具

python

def execute_code(code):
    """执行 Python 代码。用于运行计算、分析数据等。"""
    import subprocess
    result = subprocess.run(
        ["python", "-c", code],
        capture_output=True,
        text=True
    )
    return result.stdout or result.stderr

4. API 调用工具

python

def get_weather(city):
    """查询天气。输入城市名，返回天气信息。"""
    import requests

    # 示例：调用天气 API
    api_key = "your-api-key"
    url = f"https://api.weather.com/v3/weather"
    response = requests.get(url, params={"city": city, "key": api_key})
    return response.json()

def fetch_stock_price(symbol):
    """查询股票价格。输入股票代码（如 AAPL）。"""
    pass

def send_slack_message(channel, message):
    """发送 Slack 消息。用于通知。"""
    pass

5. 文件操作工具

python

def read_file(path):
    """读取文件内容。输入：文件路径。"""
    with open(path, 'r', encoding='utf-8') as f:
        return f.read()

def write_file(path, content):
    """写入文件。参数：文件路径、内容。"""
    with open(path, 'w', encoding='utf-8') as f:
        f.write(content)
        return "写入成功"

def list_files(directory):
    """列出目录下的文件。"""
    import os
    return os.listdir(directory)

完整示例：天气预报 Agent

python

from openai import OpenAI

client = OpenAI()

# 定义工具
def get_weather(city):
    """获取城市天气。输入：城市名称（如北京、上海）"""
    # 模拟天气数据
    weather_data = {
        "北京": "晴，15°C，PM2.5: 45",
        "上海": "多云，18°C，PM2.5: 62",
        "广州": "小雨，22°C，PM2.5: 88"
    }
    return weather_data.get(city, "未知城市")

def search_city_info(city):
    """搜索城市信息。用于获取城市的基本介绍。"""
    info = {
        "北京": "中国的首都，政治、文化中心",
        "上海": "中国的经济、金融中心",
        "广州": "中国南方的重要城市，商贸中心"
    }
    return info.get(city, "未知")

TOOLS = {
    "get_weather": get_weather,
    "search_city_info": search_city_info
}

TOOL_DESCRIPTIONS = """
- get_weather: 获取城市天气。输入：城市名称
- search_city_info: 搜索城市信息。输入：城市名称
"""

def weather_agent(user_input):
    messages = [
        {"role": "system", "content": f"""你是一个天气助手，可以使用工具来回答用户问题。

可用工具：
{TOOL_DESCRIPTIONS}

回答格式要求：
1. 先思考需要什么信息
2. 调用必要的工具
3. 根据结果回答用户"""},
        {"role": "user", "content": user_input}
    ]

    for _ in range(10):
        response = client.chat.completions.create(
            model="gpt-4",
            messages=messages
        )

        content = response.choices[0].message.content
        messages.append({"role": "assistant", "content": content})

        # 解析工具调用
        tool_calls = parse_tool_calls(content)
        if not tool_calls:
            return content

        # 执行工具
        for tool_name, tool_input in tool_calls:
            if tool_name in TOOLS:
                result = TOOLS[tool_name](tool_input)
                messages.append({
                    "role": "user",
                    "content": f"工具 {tool_name} 返回：{result}"
                })

    return "已达到最大迭代次数"

def parse_tool_calls(content):
    """从模型输出中解析工具调用"""
    import re

    # 简单解析：查找 [TOOL:tool_name]input[/TOOL] 格式
    pattern = r'\[TOOL:(\w+)\](.*?)\[/TOOL\]'
    matches = re.findall(pattern, content, re.DOTALL)

    if matches:
        return matches

    # 检查是否直接返回答案
    return None

# 测试
print(weather_agent("北京今天天气怎么样？"))

工具调用进阶

1. 工具选择策略

python

def select_best_tool(query, tools):
    """让模型选择最合适的工具"""
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "根据用户问题，选择最合适的工具。"},
            {"role": "user", "content": f"问题：{query}\n工具：{list(tools.keys())}"}
        ]
    )
    return response.choices[0].message.content

2. 并行工具调用

python

import concurrent.futures

def execute_tools_parallel(tool_calls):
    """并行执行多个工具"""
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = {
            executor.submit(tool, input): (name, input)
            for name, tool in TOOLS.items()
            for input in tool_calls.get(name, [])
        }

        results = {}
        for future in concurrent.futures.as_completed(futures):
            name, input = futures[future]
            results[f"{name}:{input}"] = future.result()

        return results

3. 工具链

python

def tool_chain(query, chain):
    """顺序执行工具链"""
    result = query
    for tool_name in chain:
        result = TOOLS[tool_name](result)
    return result

安全注意事项

输入验证：验证工具输入，防止注入攻击
权限控制：限制工具的操作范围
日志记录：记录工具调用历史
超时处理：设置工具调用超时
错误处理：优雅处理工具异常

python

def safe_execute(tool, input_data, timeout=30):
    """安全执行工具"""
    import signal

    def timeout_handler(signum, frame):
        raise TimeoutError("工具执行超时")

    try:
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(timeout)
        result = tool(input_data)
        signal.alarm(0)
        return result
    except TimeoutError:
        return "执行超时"
    except Exception as e:
        return f"执行错误：{str(e)}"

下一步

多代理系统 - 了解多个 Agent 如何协作

工具调用实战 ​

为什么需要工具 ​

工具设计原则 ​

1. 清晰的描述 ​

2. 简洁的输入输出 ​

3. 单一职责 ​

常用工具类型 ​

1. 搜索工具 ​

2. 计算工具 ​

3. 代码执行工具 ​

4. API 调用工具 ​

5. 文件操作工具 ​

完整示例：天气预报 Agent ​

工具调用进阶 ​

1. 工具选择策略 ​

2. 并行工具调用 ​

3. 工具链 ​

安全注意事项 ​

下一步 ​

工具调用实战

为什么需要工具

工具设计原则

1. 清晰的描述

2. 简洁的输入输出

3. 单一职责

常用工具类型

1. 搜索工具

2. 计算工具

3. 代码执行工具

4. API 调用工具

5. 文件操作工具

完整示例：天气预报 Agent

工具调用进阶

1. 工具选择策略

2. 并行工具调用

3. 工具链

安全注意事项

下一步