AI Agent如何高效调用工具：以获取数据库Schema和表列表为例

1次阅读

没有评论

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

在现代 AI 应用中，Agent 经常需要与外部工具交互，比如查询数据库 Schema 和表列表。这种交互看似简单，但在实际业务场景中会面临几个核心挑战：

性能瓶颈：同步调用工具可能导致 AI Agent 阻塞，尤其是在高并发场景下
权限管理复杂：不同工具需要不同的访问权限，如何统一管理成为难题
稳定性问题：网络抖动、工具服务不可用等情况需要妥善处理
资源消耗：频繁调用可能导致系统负载过高

以数据库 Schema 查询为例，一个未经优化的实现可能在 100QPS 压力下响应时间超过 2 秒，这显然无法满足生产环境要求。

在工具调用方式上，我们通常有两种选择：

传统 REST API
优点：简单易用，生态成熟
缺点：存在过度获取 (over-fetching) 问题
测试数据：平均响应时间 120ms(查询完整 Schema)
GraphQL
优点：按需查询，减少数据传输量
缺点：实现复杂度较高
测试数据：平均响应时间 75ms(同等查询条件)

选型建议：
– 简单查询场景使用 REST
– 复杂、多变的数据需求使用 GraphQL
– 无论哪种方式，都建议采用异步调用

以下是 Python 实现的核心代码，展示了如何封装一个高效的工具调用 SDK：

import asyncio
from functools import lru_cache
from typing import Optional, Dict, List
import aiohttp
from pydantic import BaseModel

class DatabaseSchema(BaseModel):
    """数据库 Schema 数据结构"""
    db_name: str
    tables: List[str]

class ToolInvocationError(Exception):
    """工具调用异常基类"""
    pass

class DatabaseToolSDK:
    """数据库工具调用 SDK"""

    def __init__(self, base_url: str, auth_token: str):
        self.base_url = base_url
        self.auth_token = auth_token
        self.session = aiohttp.ClientSession()
        # 默认超时设置为 3 秒
        self.timeout = aiohttp.ClientTimeout(total=3)

    @lru_cache(maxsize=128)
    async def get_schema(self, db_id: str) -> DatabaseSchema:
        """获取数据库 Schema(带缓存)"""
        url = f"{self.base_url}/schemas/{db_id}"
        headers = {"Authorization": f"Bearer {self.auth_token}"}

        # 重试逻辑(最多 3 次)
        max_retries = 3
        for attempt in range(max_retries):
            try:
                async with self.session.get(url, headers=headers, timeout=self.timeout) as resp:
                    if resp.status == 200:
                        data = await resp.json()
                        return DatabaseSchema(**data)
                    elif resp.status == 404:
                        raise ToolInvocationError(f"Database {db_id} not found")
                    else:
                        raise ToolInvocationError(f"API error: {resp.status}")
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                if attempt == max_retries - 1:
                    raise ToolInvocationError(f"Request failed after {max_retries} attempts: {str(e)}")
                await asyncio.sleep(1 * (attempt + 1))  # 指数退避

    async def close(self):
        """清理资源"""
        await self.session.close()

关键设计点：

使用 aiohttp 实现异步调用
LRU 缓存减少重复查询
指数退避重试机制
强类型校验(Pydantic)
资源自动清理

在生产环境中，我们需要严格控制工具调用的权限。JWT 鉴权是常用方案：

鉴权流程
Agent 获取 JWT 令牌(包含角色声明)
工具服务验证令牌并检查权限
记录审计日志

权限设计示例

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, "SECRET_KEY", algorithms=["HS256"])
        if "db:read" not in payload.get("scopes", []):
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return payload
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid token")