4.2 工具调用机制

4.2 工具调用机制

导读：工具调用（Tool Calling / Function Calling）是Agent系统的核心能力，使大模型能够突破纯文本生成的限制，与外部世界进行交互。本节将深入讲解工具调用的原理、实现方式和最佳实践。

4.2.1 工具调用原理

Function Calling机制

现代大模型（如GPT-4、Claude 3、GLM-4等）原生支持Function Calling能力。其工作原理是：模型不再只输出纯文本，而是可以选择调用一个或多个预定义的函数，并生成结构化的参数。

# OpenAI Function Calling基础示例
from openai import OpenAI
import json

client = 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": "calculate",
            "description": "执行数学计算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "数学表达式"}
                },
                "required": ["expression"]
            }
        }
    }
]

# 让模型决定是否需要工具
response = client.chat.completions.create(
    model="gpt-4-turbo-preview",
    messages=[
        {"role": "user", "content": "北京今天多少度？另外帮我算一下 25 * 48"}
    ],
    tools=tools,
    tool_choice="auto"
)

# 解析工具调用
if response.choices[0].message.tool_calls:
    for tool_call in response.choices[0].message.tool_calls:
        function_name = tool_call.function.name
        function_args = json.loads(tool_call.function.arguments)
        print(f"模型请求调用: {function_name}({function_args})")
​

工具调用流程

用户输入 → LLM判断是否需要工具 → 
    ├── 不需要 → 直接生成文本回答
    └── 需要 → 生成工具调用请求 → 
        ├── 执行工具 → 将结果返回LLM → LLM生成最终回答
        └── 可能需要多轮工具调用
​

4.2.2 工具定义与注册

统一工具注册框架

# 统一的工具注册框架
from typing import Dict, List, Any, Callable
from dataclasses import dataclass, field

@dataclass
class ToolDefinition:
    """工具定义"""
    name: str
    description: str
    handler: Callable
    parameters: Dict
    is_dangerous: bool = False
    rate_limit: int = 60

class ToolRegistry:
    """工具注册中心"""
    
    def __init__(self):
        self._tools: Dict[str, ToolDefinition] = {}
    
    def register(self, name: str, description: str, parameters: Dict,
                 handler: Callable, is_dangerous: bool = False, rate_limit: int = 60):
        """注册工具"""
        tool_def = ToolDefinition(
            name=name, description=description, handler=handler,
            parameters=parameters, is_dangerous=is_dangerous, rate_limit=rate_limit
        )
        self._tools[name] = tool_def
    
    def get_tool_specs(self) -> List[Dict]:
        """获取所有工具的OpenAI格式规范"""
        return [
            {"type": "function", "function": {
                "name": t.name, "description": t.description, "parameters": t.parameters
            }}
            for t in self._tools.values()
        ]
    
    def execute(self, tool_name: str, arguments: Dict) -> Any:
        """执行工具"""
        if tool_name not in self._tools:
            raise ValueError(f"未知工具: {tool_name}")
        return self._tools[tool_name].handler(**arguments)

# 注册示例
registry = ToolRegistry()
registry.register(
    name="get_weather",
    description="获取指定城市的天气信息",
    parameters={
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "城市名称"},
            "date": {"type": "string", "description": "日期"}
        },
        "required": ["city"]
    },
    handler=lambda city, date="today": f"{city}天气: 晴, 25°C"
)
​

4.2.3 常用工具类型

API调用工具

# HTTP API调用工具
import httpx
from typing import Dict

class APICallTool:
    """通用API调用工具"""
    
    def __init__(self, timeout: float = 10.0):
        self.timeout = timeout
    
    async def call(self, url: str, method: str = "GET", 
                   headers: Dict = None, body: Dict = None) -> Dict:
        """执行API调用"""
        try:
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                if method.upper() == "GET":
                    response = await client.get(url, headers=headers)
                else:
                    response = await client.post(url, json=body, headers=headers)
                return {
                    "status_code": response.status_code,
                    "body": response.text[:2000],
                    "success": 200 <= response.status_code < 300
                }
        except httpx.TimeoutException:
            return {"success": False, "error": "请求超时"}
        except Exception as e:
            return {"success": False, "error": str(e)}
​

代码执行工具

# 安全的代码执行工具
import subprocess
import tempfile
from pathlib import Path

class SafeCodeExecutor:
    """安全代码执行器"""
    
    def __init__(self, timeout: int = 30, max_output: int = 5000):
        self.timeout = timeout
        self.max_output = max_output
        self.blocked_modules = ["os.system", "subprocess.call", "eval", "exec"]
    
    def execute(self, code: str) -> Dict:
        """安全执行Python代码"""
        for module in self.blocked_modules:
            if module in code:
                return {"success": False, "error": f"禁止使用: {module}"}
        
        try:
            with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
                f.write(code)
                temp_path = f.name
            
            result = subprocess.run(
                ["python3", temp_path],
                capture_output=True, text=True, timeout=self.timeout
            )
            return {
                "success": result.returncode == 0,
                "stdout": result.stdout[:self.max_output],
                "stderr": result.stderr[:self.max_output] if result.stderr else None
            }
        except subprocess.TimeoutExpired:
            return {"success": False, "error": f"执行超时({self.timeout}秒)"}
        finally:
            Path(temp_path).unlink(missing_ok=True)
​

文件操作工具

# 文件操作工具集
class FileTools:
    """文件操作工具"""
    
    ALLOWED_EXTENSIONS = {".txt", ".md", ".csv", ".json", ".py"}
    
    @staticmethod
    def read_file(file_path: str, max_size: int = 10000) -> Dict:
        """读取文件内容"""
        path = Path(file_path)
        if path.suffix.lower() not in FileTools.ALLOWED_EXTENSIONS:
            return {"success": False, "error": "不支持的文件类型"}
        if path.stat().st_size > max_size * 1024:
            return {"success": False, "error": f"文件超过{max_size}KB限制"}
        try:
            content = path.read_text(encoding="utf-8")
            return {"success": True, "content": content, "size": len(content)}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    @staticmethod
    def list_directory(dir_path: str) -> Dict:
        """列出目录内容"""
        path = Path(dir_path)
        items = []
        for item in sorted(path.iterdir()):
            items.append({
                "name": item.name,
                "type": "directory" if item.is_dir() else "file",
                "size": item.stat().st_size if item.is_file() else None
            })
        return {"success": True, "items": items, "total": len(items)}
​

4.2.4 工具调用编排

多工具并行调用

# 多工具并行调用编排器
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict

class ToolCallOrchestrator:
    """工具调用编排器"""
    
    def __init__(self, registry: ToolRegistry, max_concurrent: int = 5):
        self.registry = registry
        self.max_concurrent = max_concurrent
        self.executor = ThreadPoolExecutor(max_workers=max_concurrent)
    
    async def execute_calls(self, tool_calls: List[Dict]) -> List[Dict]:
        """并行执行多个工具调用"""
        semaphore = asyncio.Semaphore(self.max_concurrent)
        
        async def execute_one(call: Dict) -> Dict:
            async with semaphore:
                tool_name = call["name"]
                arguments = call.get("arguments", {})
                try:
                    loop = asyncio.get_event_loop()
                    result = await loop.run_in_executor(
                        self.executor, self.registry.execute, tool_name, arguments
                    )
                    return {"tool": tool_name, "success": True, "result": result}
                except Exception as e:
                    return {"tool": tool_name, "success": False, "error": str(e)}
        
        tasks = [execute_one(call) for call in tool_calls]
        return await asyncio.gather(*tasks)
​

错误处理与重试

# 工具调用错误处理与重试
from tenacity import retry, stop_after_attempt, wait_exponential

class ResilientToolExecutor:
    """带重试和降级的工具执行器"""
    
    def __init__(self, registry: ToolRegistry, max_retries: int = 3):
        self.registry = registry
        self.max_retries = max_retries
    
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
    def execute_with_retry(self, tool_name: str, arguments: Dict) -> Any:
        """带重试的工具执行"""
        try:
            return self.registry.execute(tool_name, arguments)
        except ConnectionError:
            raise  # 触发重试
        except ValueError as e:
            return {"success": False, "error": f"参数错误: {e}"}  # 不重试参数错误
    
    def execute_with_fallback(self, tool_name: str, arguments: Dict, fallback: Any = None) -> Any:
        """带降级的工具执行"""
        try:
            return self.registry.execute(tool_name, arguments)
        except Exception as e:
            print(f"工具 {tool_name} 执行失败: {e}，使用降级方案")
            return fallback if fallback is not None else {"success": False, "error": str(e)}
​

4.2.5 工具调用的Prompt设计

好的工具描述是工具调用成功的保障。工具的description和parameters描述必须清晰准确。

# 工具描述最佳实践
GOOD_TOOL_DESCRIPTIONS = [
    {
        "name": "search_knowledge_base",
        "description": """在内部知识库中搜索相关文档。适用于回答关于公司产品、
政策、流程等问题。返回相关度最高的文档片段列表。
注意：此工具只搜索公司内部文档，不搜索互联网。""",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词，建议使用具体的术语而非模糊描述。示例：'年假政策' 而非 '假期相关'"
                },
                "top_k": {
                    "type": "integer",
                    "description": "返回结果数量，默认5。建议3-10之间",
                    "default": 5
                },
                "department": {
                    "type": "string",
                    "description": "可选的部门过滤，如'HR'、'技术'、'财务'",
                    "default": None
                }
            },
            "required": ["query"]
        }
    }
]
​

本章小结

通过本节的学习，你将：

✅ 理解Function Calling的工作原理和调用流程
✅ 掌握统一的工具注册和管理框架
✅ 学会实现API调用、代码执行、文件操作等常用工具
✅ 掌握多工具并行调用和错误处理机制
✅ 了解工具描述设计的最佳实践

在4.3节中，我们将基于工具调用能力，学习如何编排复杂的工作流。

[FAQ] 常见问题

Q: Function Calling和手动解析Prompt中的工具指令有什么区别？
A: Function Calling是模型原生支持的结构化输出，保证参数格式正确。手动解析需要从文本中提取工具名和参数，容易出错且不稳定。推荐始终使用原生Function Calling。

Q: 一个Agent最多可以注册多少个工具？
A: 理论上没有硬性限制，但工具过多会降低模型选择正确工具的准确率。建议控制在15-20个以内。如果工具很多，可以分组管理，按需加载。

Q: 工具调用的延迟如何优化？
A: 可以通过以下方式优化：并行执行独立的工具调用、缓存频繁调用的结果、使用流式输出减少等待感知、预加载常用工具的资源。

Q: 如何防止Agent恶意使用工具？
A: 多层防护：工具注册时标记危险等级、执行前检查参数合理性、设置资源使用上限（调用次数、执行时间）、完整的审计日志、高风险操作需人工确认。

Q: 不同模型的Function Calling兼容吗？
A: 基本兼容但细节有差异。OpenAI和Anthropic的Function Calling格式最为成熟，开源模型的支持程度不一。建议使用LangChain等框架做抽象层，屏蔽底层差异。