工具调用机制

概述

工具调用（Tool Calling）是Agent与外部世界交互的核心机制。通过工具调用，Agent可以执行搜索、计算、API请求、文件操作等实际任务，突破了大模型只能生成文本的限制。现代大模型（如GPT-4、Claude等）都支持原生工具调用功能。

工具调用基础

什么是工具调用

工具调用是指LLM在生成回复时，能够识别出需要使用外部工具，并生成结构化的工具调用请求。应用系统执行工具后，将结果返回给LLM继续处理。

工作流程

用户输入 → LLM分析 → 决定调用工具 → 生成工具调用参数
    ↓
执行工具 → 获取结果 → 返回LLM → 生成最终回复

核心概念

1. 工具定义 - 描述工具的功能、参数和返回值
2. 工具选择 - LLM根据用户需求选择合适的工具
3. 参数构造 - LLM生成符合工具要求的参数
4. 结果处理 - 将工具执行结果整合到回复中

工具定义规范

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"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "在网络上搜索信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"
                    },
                    "num_results": {
                        "type": "integer",
                        "description": "返回结果数量",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        }
    }
]

Claude工具定义格式

tools = [
    {
        "name": "get_weather",
        "description": "获取指定城市的天气信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"]
                }
            },
            "required": ["city"]
        }
    }
]

工具定义最佳实践

class ToolDefinition:
    def __init__(self, name: str, description: str, parameters: dict):
        self.name = name
        self.description = description
        self.parameters = parameters
    
    def to_openai_format(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters
            }
        }
    
    def validate(self):
        assert self.name.isidentifier(), "工具名称必须是有效的标识符"
        assert len(self.description) > 10, "描述应该详细说明工具功能"
        assert "properties" in self.parameters, "必须定义参数属性"

class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register(self, tool_def: ToolDefinition, handler: callable):
        tool_def.validate()
        self.tools[tool_def.name] = {
            "definition": tool_def,
            "handler": handler
        }
    
    def get_definitions(self) -> list:
        return [
            t["definition"].to_openai_format() 
            for t in self.tools.values()
        ]
    
    def get_handler(self, name: str) -> callable:
        return self.tools[name]["handler"]

工具调用实现

OpenAI工具调用

from openai import OpenAI
import json

class OpenAIToolCaller:
    def __init__(self, api_key: str, model: str = "gpt-4o"):
        self.client = OpenAI(api_key=api_key)
        self.model = model
        self.registry = ToolRegistry()
    
    def register_tool(self, name: str, description: str, 
                      parameters: dict, handler: callable):
        tool_def = ToolDefinition(name, description, parameters)
        self.registry.register(tool_def, handler)
    
    def run(self, user_message: str) -> str:
        messages = [{"role": "user", "content": user_message}]
        
        while True:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=self.registry.get_definitions(),
                tool_choice="auto"
            )
            
            message = response.choices[0].message
            
            if message.content:
                return message.content
            
            if message.tool_calls:
                messages.append(message)
                
                for tool_call in message.tool_calls:
                    result = self.execute_tool_call(tool_call)
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(result, ensure_ascii=False)
                    })
            else:
                return "无法处理该请求"
    
    def execute_tool_call(self, tool_call) -> dict:
        handler = self.registry.get_handler(tool_call.function.name)
        arguments = json.loads(tool_call.function.arguments)
        
        try:
            result = handler(**arguments)
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}

caller = OpenAIToolCaller(api_key="your-api-key")

caller.register_tool(
    name="get_weather",
    description="获取城市天气信息",
    parameters={
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "城市名称"}
        },
        "required": ["city"]
    },
    handler=lambda city: f"{city}今天晴天，温度25°C"
)

response = caller.run("北京今天天气怎么样？")
print(response)

Claude工具调用

from anthropic import Anthropic

class ClaudeToolCaller:
    def __init__(self, api_key: str, model: str = "claude-3-5-sonnet-20241022"):
        self.client = Anthropic(api_key=api_key)
        self.model = model
        self.registry = ToolRegistry()
    
    def register_tool(self, name: str, description: str,
                      parameters: dict, handler: callable):
        tool_def = ToolDefinition(name, description, parameters)
        self.registry.register(tool_def, handler)
    
    def run(self, user_message: str) -> str:
        messages = [{"role": "user", "content": user_message}]
        
        while True:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=4096,
                messages=messages,
                tools=self.registry.get_definitions()
            )
            
            if response.stop_reason == "tool_use":
                messages.append({
                    "role": "assistant",
                    "content": response.content
                })
                
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        result = self.execute_tool(block)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps(result, ensure_ascii=False)
                        })
                
                messages.append({
                    "role": "user",
                    "content": tool_results
                })
            
            else:
                for block in response.content:
                    if hasattr(block, "text"):
                        return block.text
                
                return "无法处理该请求"
    
    def execute_tool(self, tool_use) -> dict:
        handler = self.registry.get_handler(tool_use.name)
        
        try:
            result = handler(**tool_use.input)
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}

工具设计模式

1. 简单工具模式

适用于单一功能、参数简单的工具。

def calculate(expression: str) -> float:
    """
    计算数学表达式
    
    Args:
        expression: 数学表达式，如 "2 + 3 * 4"
    
    Returns:
        计算结果
    """
    import ast
    import operator
    
    ops = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv
    }
    
    node = ast.parse(expression, mode='eval')
    
    def eval_node(node):
        if isinstance(node, ast.Num):
            return node.n
        elif isinstance(node, ast.BinOp):
            left = eval_node(node.left)
            right = eval_node(node.right)
            return ops[type(node.op)](left, right)
        else:
            raise ValueError("不支持的表达式")
    
    return eval_node(node.body)

2. 复合工具模式

组合多个工具完成复杂任务。

class CompositeTool:
    def __init__(self, tools: list):
        self.tools = tools
    
    def run(self, **kwargs):
        results = {}
        
        for tool in self.tools:
            if tool.can_run(kwargs):
                result = tool.run(**kwargs)
                results[tool.name] = result
                kwargs.update(result)
        
        return results

class ResearchTool(CompositeTool):
    def __init__(self):
        super().__init__([
            SearchTool(),
            ExtractTool(),
            SummarizeTool()
        ])
    
    def research(self, topic: str) -> dict:
        return self.run(topic=topic)

3. 条件工具模式

根据条件选择执行不同的工具。

class ConditionalTool:
    def __init__(self):
        self.conditions = []
    
    def add_condition(self, condition: callable, tool: callable):
        self.conditions.append((condition, tool))
    
    def run(self, **kwargs):
        for condition, tool in self.conditions:
            if condition(kwargs):
                return tool(**kwargs)
        
        raise ValueError("没有匹配的工具")

router = ConditionalTool()
router.add_condition(
    condition=lambda x: "天气" in x.get("query", ""),
    tool=get_weather
)
router.add_condition(
    condition=lambda x: "计算" in x.get("query", ""),
    tool=calculate
)

4. 链式工具模式

工具的输出作为下一个工具的输入。

class ToolChain:
    def __init__(self, tools: list):
        self.tools = tools
    
    def run(self, initial_input: any) -> any:
        result = initial_input
        
        for tool in self.tools:
            result = tool(result)
        
        return result

pipeline = ToolChain([
    lambda x: search_web(x),
    lambda x: extract_key_info(x),
    lambda x: summarize(x),
    lambda x: format_output(x)
])

result = pipeline.run("人工智能最新进展")

工具调用优化

1. 参数验证

from pydantic import BaseModel, ValidationError
from typing import Optional

class WeatherParams(BaseModel):
    city: str
    unit: Optional[str] = "celsius"
    
    class Config:
        extra = "forbid"

class ValidatedTool:
    def __init__(self, param_model: type, handler: callable):
        self.param_model = param_model
        self.handler = handler
    
    def run(self, **kwargs):
        try:
            params = self.param_model(**kwargs)
            return self.handler(**params.dict())
        except ValidationError as e:
            return {"error": f"参数验证失败: {e}"}

weather_tool = ValidatedTool(
    param_model=WeatherParams,
    handler=lambda city, unit: get_weather_data(city, unit)
)

2. 结果缓存

from functools import lru_cache
import hashlib
import json

class CachedTool:
    def __init__(self, tool: callable, maxsize: int = 128):
        self.tool = tool
        self.cache = {}
        self.maxsize = maxsize
    
    def _hash_args(self, **kwargs) -> str:
        args_str = json.dumps(kwargs, sort_keys=True)
        return hashlib.md5(args_str.encode()).hexdigest()
    
    def run(self, **kwargs):
        cache_key = self._hash_args(**kwargs)
        
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        result = self.tool(**kwargs)
        
        if len(self.cache) >= self.maxsize:
            self.cache.pop(next(iter(self.cache)))
        
        self.cache[cache_key] = result
        return result

3. 并行调用

import asyncio
from concurrent.futures import ThreadPoolExecutor

class ParallelToolExecutor:
    def __init__(self, max_workers: int = 5):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
    
    async def execute_parallel(self, tool_calls: list) -> list:
        loop = asyncio.get_event_loop()
        
        tasks = [
            loop.run_in_executor(
                self.executor,
                self.execute_single,
                call
            )
            for call in tool_calls
        ]
        
        results = await asyncio.gather(*tasks)
        return results
    
    def execute_single(self, tool_call: dict) -> dict:
        handler = self.get_handler(tool_call["name"])
        return handler(**tool_call["arguments"])

4. 超时控制

import signal
from contextlib import contextmanager

class TimeoutError(Exception):
    pass

@contextmanager
def timeout(seconds: int):
    def timeout_handler(signum, frame):
        raise TimeoutError("工具执行超时")
    
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(seconds)
    
    try:
        yield
    finally:
        signal.alarm(0)

class TimeoutTool:
    def __init__(self, tool: callable, timeout_seconds: int = 30):
        self.tool = tool
        self.timeout_seconds = timeout_seconds
    
    def run(self, **kwargs):
        try:
            with timeout(self.timeout_seconds):
                return self.tool(**kwargs)
        except TimeoutError:
            return {"error": "工具执行超时"}

常用工具实现

搜索工具

import requests
from bs4 import BeautifulSoup

class SearchTool:
    def __init__(self, api_key: str = None):
        self.api_key = api_key
    
    def search(self, query: str, num_results: int = 5) -> list:
        if self.api_key:
            return self._search_with_api(query, num_results)
        else:
            return self._search_with_scraping(query, num_results)
    
    def _search_with_api(self, query: str, num_results: int) -> list:
        url = "https://api.search-engine.com/search"
        params = {
            "q": query,
            "num": num_results,
            "api_key": self.api_key
        }
        
        response = requests.get(url, params=params)
        data = response.json()
        
        return [
            {
                "title": item["title"],
                "url": item["url"],
                "snippet": item["snippet"]
            }
            for item in data["results"]
        ]
    
    def _search_with_scraping(self, query: str, num_results: int) -> list:
        pass

search_tool = SearchTool()
results = search_tool.search("Python教程")

文件操作工具

import os
from pathlib import Path

class FileTool:
    def __init__(self, base_dir: str = "."):
        self.base_dir = Path(base_dir)
    
    def read(self, filepath: str) -> str:
        full_path = self.base_dir / filepath
        self._validate_path(full_path)
        
        with open(full_path, 'r', encoding='utf-8') as f:
            return f.read()
    
    def write(self, filepath: str, content: str) -> bool:
        full_path = self.base_dir / filepath
        self._validate_path(full_path)
        
        full_path.parent.mkdir(parents=True, exist_ok=True)
        
        with open(full_path, 'w', encoding='utf-8') as f:
            f.write(content)
        
        return True
    
    def list_files(self, directory: str = ".") -> list:
        full_path = self.base_dir / directory
        self._validate_path(full_path)
        
        return [
            {
                "name": item.name,
                "type": "directory" if item.is_dir() else "file",
                "size": item.stat().st_size if item.is_file() else None
            }
            for item in full_path.iterdir()
        ]
    
    def _validate_path(self, path: Path):
        resolved = path.resolve()
        if not str(resolved).startswith(str(self.base_dir.resolve())):
            raise ValueError("路径不在允许范围内")

API调用工具

import requests
from typing import Dict, Any

class APITool:
    def __init__(self, base_url: str, headers: dict = None):
        self.base_url = base_url.rstrip('/')
        self.headers = headers or {}
    
    def get(self, endpoint: str, params: dict = None) -> Dict[str, Any]:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = requests.get(url, headers=self.headers, params=params)
        return self._handle_response(response)
    
    def post(self, endpoint: str, data: dict = None) -> Dict[str, Any]:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = requests.post(url, headers=self.headers, json=data)
        return self._handle_response(response)
    
    def _handle_response(self, response: requests.Response) -> dict:
        if response.status_code >= 400:
            return {
                "success": False,
                "error": f"HTTP {response.status_code}: {response.text}"
            }
        
        return {
            "success": True,
            "data": response.json()
        }

工具调用调试

日志记录

import logging
from datetime import datetime

class ToolLogger:
    def __init__(self, log_file: str = "tool_calls.log"):
        self.logger = logging.getLogger("ToolLogger")
        self.logger.setLevel(logging.INFO)
        
        handler = logging.FileHandler(log_file)
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
    
    def log_call(self, tool_name: str, arguments: dict):
        self.logger.info(f"调用工具: {tool_name}, 参数: {arguments}")
    
    def log_result(self, tool_name: str, result: any, duration: float):
        self.logger.info(
            f"工具返回: {tool_name}, "
            f"耗时: {duration:.2f}s, "
            f"结果: {result}"
        )
    
    def log_error(self, tool_name: str, error: Exception):
        self.logger.error(f"工具错误: {tool_name}, 错误: {error}")

class LoggedTool:
    def __init__(self, tool: callable, logger: ToolLogger):
        self.tool = tool
        self.logger = logger
    
    def run(self, **kwargs):
        tool_name = self.tool.__name__
        self.logger.log_call(tool_name, kwargs)
        
        start_time = datetime.now()
        
        try:
            result = self.tool(**kwargs)
            duration = (datetime.now() - start_time).total_seconds()
            self.logger.log_result(tool_name, result, duration)
            return result
        except Exception as e:
            self.logger.log_error(tool_name, e)
            raise

调试模式

class DebugToolCaller:
    def __init__(self, caller):
        self.caller = caller
        self.debug_mode = False
        self.call_history = []
    
    def enable_debug(self):
        self.debug_mode = True
    
    def run(self, user_message: str) -> str:
        if self.debug_mode:
            print(f"[DEBUG] 用户输入: {user_message}")
        
        result = self.caller.run(user_message)
        
        if self.debug_mode:
            print(f"[DEBUG] 最终回复: {result}")
        
        return result
    
    def get_call_history(self) -> list:
        return self.call_history

小结

工具调用是Agent的核心能力，通过合理设计和实现工具系统，可以让Agent具备强大的执行能力：

1. 规范定义 - 使用标准的工具定义格式，提供清晰的描述
2. 灵活实现 - 支持多种工具模式，适应不同场景需求
3. 性能优化 - 参数验证、结果缓存、并行调用、超时控制
4. 安全可靠 - 权限控制、路径验证、错误处理
5. 易于调试 - 日志记录、调试模式、历史追踪

下一章我们将探讨Agent的记忆系统设计，这是实现多轮对话和知识积累的关键。