Agent调用外部工具的方法：从原理到实战的避坑指南

1次阅读

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

电商比价 Agent 场景
当用户搜索商品时，比价 Agent 需要实时调用京东、淘宝等 10+ 个电商平台的 API 获取价格数据。若直接裸调用第三方接口，会遇到：
各平台响应格式不一致（XML/JSON）
部分接口 QPS 限制为 5 次 / 秒
网络波动导致超时率高达 15%
智能客服工单系统
Agent 在处理 ” 物流投诉 ” 时需同时调用：
快递 100 API 查询物流轨迹（REST）
内部 ERP 系统获取订单详情（gRPC）
短信平台发送通知（消息队列）

维度	REST API	gRPC	消息队列（RabbitMQ）
吞吐量	中等（1k QPS）	高（50k QPS）	高（100k QPS）
平均延迟	100-300ms	5-50ms	200-500ms
开发成本	低	中	高
适用场景	跨语言简单交互	高性能内部调用	异步削峰

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

class ExternalToolClient:
    """
    Circuit Breaker(熔断器) 模式实现
    @param max_retries: 最大重试次数
    @param timeout: 超时阈值 (秒)
    """
    def __init__(self, max_retries=3, timeout=5):
        self.session = requests.Session()
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=20,  # 连接池大小
            pool_maxsize=100,
            max_retries=3
        )
        self.session.mount('http://', adapter)

    @retry(stop=stop_after_attempt(3), 
           wait=wait_exponential(multiplier=1, max=10))
    def call_api(self, url, params):
        try:
            resp = self.session.get(url, params=params, timeout=5)
            resp.raise_for_status()  # 自动处理 4xx/5xx
            return resp.json()
        except requests.exceptions.RequestException as e:
            # 记录到 Prometheus 监控
            log_error(f"API 调用失败: {str(e)}")
            raise

def api_call_logger(func):
    """Decorator(装饰器) 模式记录调用日志"""
    def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = func(*args, **kwargs)
            latency = (time.time() - start_time) * 1000
            log_info(f"{func.__name__} 成功 | 耗时:{latency:.2f}ms")
            return result
        except Exception as e:
            log_error(f"{func.__name__} 失败: {str(e)}")
            raise
    return wrapper

requests 调优参数
pool_connections: 每个 host 保持的连接数（建议 20）
pool_maxsize: 连接池最大值（建议 = 并发线程数 *2）
max_retries: 底层 TCP 重试次数（建议 3）

令牌桶限流实现

from ratelimit import limits, sleep_and_retry

class RateLimitedClient:
    """限制每秒不超过 10 次调用"""
    @sleep_and_retry
    @limits(calls=10, period=1)
    def call_api(self, url):
        return requests.get(url)

加密存储方案
使用 AWS KMS 或 Vault 加密 API 密钥
运行时环境变量注入（禁止硬编码）

OAuth2.0 自动刷新

sequenceDiagram
    Agent->>Auth Server: 请求 JWT 令牌（client_credentials）Auth Server-->>Agent: 返回 access_token(1h 有效)
    loop 每小时自动刷新
        Agent->>Auth Server: 用 refresh_token 获取新 token
        Auth Server-->>Agent: 返回新 token
    end