Agent如何调用工具：从原理到实战的技术解析

1次阅读

没有评论

共计 2378 个字符，预计需要花费 6 分钟才能阅读完成。

在现代智能 Agent 系统中，调用外部工具（如数据库、API、计算服务等）是核心能力之一。典型的应用场景包括：

数据查询与分析（调用数据库或第三方 API）
复杂计算任务（调用专门的算法服务）
系统集成（与企业内部系统交互）

然而，这些场景也带来了一系列挑战：

延迟问题：外部调用往往比本地操作慢数个数量级，可能阻塞 Agent 执行流程
错误处理：网络波动、服务不可用、数据格式不符等情况需要妥善处理
权限控制：不同工具可能需要不同的认证机制和访问权限
依赖管理：工具版本变更可能影响 Agent 稳定性

开发者通常有几种主要的技术选择来实现工具调用：

直接 API 调用
优点：实现简单，无需额外抽象层
缺点：难以统一管理，错误处理分散
适用场景：简单系统或原型开发
工具封装层
优点：统一接口，集中错误处理
缺点：增加开发成本
适用场景：中型以上系统
异步调用
优点：提高吞吐量，避免阻塞
缺点：增加代码复杂度
适用场景：高并发或延迟敏感场景

下面我们实现一个基础但可扩展的工具调用框架：

from typing import Callable, Dict, Any, Optional
from functools import wraps
import time
import random

# 工具注册表
tool_registry: Dict[str, Dict[str, Any]] = {}

def register_tool(name: str, max_retries: int = 3, timeout: int = 10):
    """工具装饰器，用于注册新工具"""
    def decorator(func: Callable):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 实现重试逻辑
            last_error = None
            for attempt in range(max_retries + 1):
                try:
                    start_time = time.time()
                    result = func(*args, **kwargs)
                    # 简单的执行时间记录
                    execution_time = time.time() - start_time
                    if execution_time > timeout:
                        raise TimeoutError(f"Tool {name} timed out")
                    return result
                except Exception as e:
                    last_error = e
                    if attempt == max_retries:
                        break
                    time.sleep(2 ** attempt)  # 指数退避
            raise last_error

        # 注册工具元数据
        tool_registry[name] = {
            "function": wrapper,
            "description": func.__doc__,  # 使用函数文档字符串作为描述
            "max_retries": max_retries,
            "timeout": timeout
        }
        return wrapper
    return decorator

# 示例工具定义
@register_tool(name="weather_query", max_retries=2, timeout=5)
def get_weather(city: str) -> str:
    """查询指定城市的天气情况"""
    # 模拟 API 调用
    if random.random() < 0.3:  # 30% 概率模拟失败
        raise ConnectionError("Weather service unavailable")
    return f"{city}: Sunny, 25°C"

# 工具调用函数
def call_tool(tool_name: str, *args, **kwargs) -> Any:
    """统一工具调用入口"""
    if tool_name not in tool_registry:
        raise ValueError(f"Tool {tool_name} not registered")

    tool_info = tool_registry[tool_name]
    return tool_info["function"](*args, **kwargs)

# 使用示例
if __name__ == "__main__":
    try:
        print(call_tool("weather_query", "Beijing"))
    except Exception as e:
        print(f"Error: {e}")

这个实现包含以下关键特性：