Agent调用工具Tool的实战指南：从基础原理到生产环境避坑

1次阅读

没有评论

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

在实际开发中，Agent 调用 Tool 时常常会遇到各种问题，特别是当系统规模扩大后，这些问题会变得更加明显。

接口耦合严重：Agent 直接调用 Tool 的具体实现，导致 Tool 的变更会直接影响到 Agent 的稳定性。
超时失控：Tool 执行时间过长，导致 Agent 阻塞，甚至引发雪崩效应。
错误处理缺失：Tool 的异常没有被妥善处理，导致 Agent 无法感知故障，继续发送请求。
并发竞争：多个 Agent 同时调用同一个 Tool，可能导致资源竞争或数据不一致。

这些问题在生产环境中尤为致命，轻则导致服务降级，重则引发系统崩溃。

为了规避上述问题，我们需要设计一个合理的架构来解耦 Agent 和 Tool 的调用。以下是三种常见的调用方式及其适用场景：

RPC（远程过程调用）
适用于高性能、低延迟的场景。
需要 Tool 和 Agent 在同一内网环境。
典型框架：gRPC、Thrift。
HTTP
通用性强，跨语言支持好。
适用于 Tool 和 Agent 部署在不同环境的情况。
典型框架：Flask、FastAPI。
消息队列
适用于异步、高吞吐的场景。
Tool 的处理时间较长，且不需要立即返回结果。
典型框架：RabbitMQ、Kafka。

以下是分层架构的示意图：

+-------------------+       +-------------------+       +-------------------+
|      Agent        |       |    Tool Proxy     |       |       Tool         |
+-------------------+       +-------------------+       +-------------------+
| 发起 Tool 调用请求  | ----> | 路由 / 负载均衡    | ----> | 执行具体任务      |
| 接收 Tool 返回结果  | <---- | 错误处理 / 重试    | <---- | 返回执行结果      |
+-------------------+       +-------------------+       +-------------------+

接下来，我们通过 Python 代码展示如何实现一个带有重试机制的 Tool 调用封装类。

import requests
from retrying import retry
from contextlib import contextmanager

class ToolClient:
    def __init__(self, tool_url):
        self.tool_url = tool_url

    @retry(
        stop_max_attempt_number=3,
        wait_exponential_multiplier=1000,
        wait_exponential_max=10000,
        retry_on_exception=lambda e: isinstance(e, (requests.exceptions.Timeout, requests.exceptions.ConnectionError))
    )
    def call_tool(self, params):
        response = requests.post(self.tool_url, json=params, timeout=5)
        response.raise_for_status()
        return response.json()

    @contextmanager
    def tool_context(self, params):
        try:
            result = self.call_tool(params)
            yield result
        except Exception as e:
            print(f"Tool 调用失败: {str(e)}")
            raise

# 使用示例
if __name__ == "__main__":
    client = ToolClient("http://example.com/tool")
    with client.tool_context({"param1": "value1"}) as result:
        print(f"Tool 返回结果: {result}")

import aiohttp
import asyncio
from async_retrying import retry

class AsyncToolClient:
    def __init__(self, tool_url):
        self.tool_url = tool_url

    @retry(attempts=3, delay=1, backoff=2)
    async def call_tool(self, params):
        async with aiohttp.ClientSession() as session:
            async with session.post(self.tool_url, json=params, timeout=5) as response:
                response.raise_for_status()
                return await response.json()

# 使用示例
async def main():
    client = AsyncToolClient("http://example.com/tool")
    try:
        result = await client.call_tool({"param1": "value1"})
        print(f"Tool 返回结果: {result}")
    except Exception as e:
        print(f"Tool 调用失败: {str(e)}")

if __name__ == "__main__":
    asyncio.run(main())

在生产环境中，我们需要监控 Tool 的调用情况，包括成功率、响应时间等指标。可以通过埋点来实现：

from prometheus_client import Counter, Histogram

TOOL_CALL_TOTAL = Counter('tool_call_total', 'Total tool calls', ['tool_name', 'status'])
TOOL_CALL_DURATION = Histogram('tool_call_duration_seconds', 'Tool call duration', ['tool_name'])

class InstrumentedToolClient(ToolClient):
    def call_tool(self, params):
        start_time = time.time()
        try:
            result = super().call_tool(params)
            TOOL_CALL_TOTAL.labels(tool_name=self.tool_url, status='success').inc()
            return result
        except Exception as e:
            TOOL_CALL_TOTAL.labels(tool_name=self.tool_url, status='failure').inc()
            raise
        finally:
            TOOL_CALL_DURATION.labels(tool_name=self.tool_url).observe(time.time() - start_time)

Tool 在冷启动时可能会有较高的延迟，可以通过预热来解决：

在系统启动时，预先调用 Tool 的简单接口。
对于长时间不用的 Tool，定期发送心跳请求保持活跃。
使用连接池管理 HTTP 长连接。

from functools import wraps

def check_permission(role):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            current_role = get_current_role()  # 获取当前用户角色
            if current_role not in role:
                raise PermissionError(f"Role {current_role} not allowed")
            return func(*args, **kwargs)
        return wrapper
    return decorator

class SecureToolClient(ToolClient):
    @check_permission(['admin', 'tool_operator'])
    def call_tool(self, params):
        return super().call_tool(params)

以下是三个真实生产环境中的故障案例：