大模型Agent工具调用实战：从原理到生产环境部署

1次阅读

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

在实际业务中集成大模型 Agent 时，工具调用环节往往会遇到以下典型问题：

接口超时不可控 ：模型 API 响应时间波动大，简单设置固定超时会导致大量误判
结果解析复杂度高 ：大模型返回的非结构化数据需要多层清洗和类型转换
状态维护成本高 ：长周期会话需要保持工具调用的上下文一致性

适用场景：
需要立即获得结果的短流程
业务逻辑简单的轻量级应用
优缺点：
实现简单，代码直观
吞吐量受限于网络延迟
线程阻塞影响整体性能

适用场景：
高并发请求处理
允许延迟响应的业务流程
性能对比（实测数据）：
同步调用 QPS 约 50-100
异步方案可达 1000+（取决于队列深度）

from tenacity import retry, stop_after_attempt, wait_exponential
import httpx
from typing import Optional, Dict, Any

class ToolInvoker:
    def __init__(self, base_url: str, api_key: str):
        self.client = httpx.AsyncClient(base_url=base_url)
        self.api_key = api_key

    @retry(stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10)
    )
    async def call_tool(
        self,
        tool_name: str,
        params: Dict[str, Any],
        timeout: float = 30.0
    ) -> Optional[Dict[str, Any]]:
        """
        带自动重试的工具调用方法
        :param tool_name: 工具端点名称
        :param params: 调用参数
        :param timeout: 单次请求超时 (秒)
        """
        try:
            headers = self._build_auth_headers()
            resp = await self.client.post(f"/tools/{tool_name}",
                json=params,
                headers=headers,
                timeout=timeout
            )
            resp.raise_for_status()
            return self._parse_response(resp.json())
        except httpx.RequestError as e:
            logger.error(f"Request failed: {str(e)}")
            raise

    def _build_auth_headers(self) -> Dict[str, str]:
        """生成包含 JWT 的鉴权头"""
        token = jwt.encode({"key": self.api_key}, "secret", algorithm="HS256")
        return {"Authorization": f"Bearer {token}",
            "X-Request-Sign": self._generate_signature()}

JWT 鉴权 ：每个请求携带时效性 token
请求签名 ：防止参数篡改
IP 白名单 ：生产环境建议额外配置

# 推荐客户端配置
client = httpx.AsyncClient(
    limits=httpx.Limits(
        max_connections=100,  # 根据服务端承受能力调整
        max_keepalive_connections=20,
        keepalive_expiry=60
    ),
    timeout=httpx.Timeout(30.0, connect=5.0)
)

连接超时：5 秒（避免 TCP 握手阻塞）
读超时：动态调整（根据历史响应 P99 值×1.5）
总超时：各阶段超时之和×1.2

使用 orjson 替代标准 json 库（性能提升 3 - 5 倍）

处理 datetime 对象：

import orjson

def default(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError

orjson.dumps(data, default=default)

设置输出 token 上限（避免过量计费）

实现结果校验器：

def validate_output(output: str, schema: dict) -> bool:
    try:
        jsonschema.validate(json.loads(output), schema)
        return True
    except (ValueError, jsonschema.ValidationError):
        return False

本文实现的方案已经过生产环境验证，能支撑日均百万级调用。留两个开放问题供读者探讨：