Agent如何调用工具：从基础原理到实战避坑指南

1次阅读

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

在构建 Agent 系统时，工具调用往往是开发者最先遇到的硬骨头。以下是两个真实场景：

场景一：某运维 Agent 通过 SSH 批量执行命令时，因未处理连接超时导致线程池耗尽，整个系统瘫痪
场景二：数据处理 Agent 调用图像处理工具时，未做内存限制引发 OOM（Out Of Memory），连带影响宿主机的其他服务

这些问题的核心在于：工具调用不只是简单的函数执行，而是涉及 进程隔离 、 资源管控 和故障恢复 的分布式系统问题。

维度	JSON-RPC	gRPC	REST
延迟	中（ms 级）	低（μs 级）	高（10+ms）
开发成本	低	中	低
适用场景	内部工具调用	高性能微服务	开放 API

对于 Agent 场景，推荐 JSON-RPC：

协议简单，适合工具调用的轻量级场景
天然支持异步（如 JSON-RPC 2.0 批量调用）
跨语言支持优于 gRPC

from functools import lru_cache

class ToolRegistry:
    """工具注册中心（带 LRU 缓存）"""
    def __init__(self, max_size=100):
        self._registry = {}

    @lru_cache(maxsize=100)
    def get_tool(self, tool_name: str) -> Callable:
        return self._registry.get(tool_name)

    def register(self, name: str, tool: Callable):
        self._registry[name] = tool

import asyncio
from typing import Optional

def retry(max_attempts=3, delay=1):
    """重试装饰器（支持指数退避）"""
    def decorator(func):
        async def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    await asyncio.sleep(delay * (2 ** attempt))
            raise last_error
        return wrapper
    return decorator

Timeout：用 asyncio.wait_for 包装调用
CircuitBreaker（熔断器）：失败阈值触发熔断
Fallback：降级返回默认值

from jose import jwt

def create_token(tool_name: str, secret: str) -> str:
    """生成工具调用令牌"""
    return jwt.encode({"tool": tool_name},
        secret,
        algorithm="HS256"
    )

进程级 ：用subprocess 启动工具，通过 cgroups 限制 CPU/ 内存
容器级 ：将工具打包为 Docker 容器，通过--memory 参数限制

from prometheus_client import Counter

TOOL_CALLS = Counter(
    'agent_tool_calls_total', 
    'Total tool calls',
    ['tool_name', 'status']
)

# 在调用处埋点
TOOL_CALLS.labels(tool_name='ssh', status='success').inc()