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

Agent 工具调用入门指南

背景与痛点

在自动化流程中，Agent 工具调用是连接不同服务和功能的核心枢纽。想象一下，你正在构建一个智能助手，它需要调用天气 API、发送邮件、查询数据库等。但实际开发中，我们常常遇到这些问题：

• 不同工具的 API 协议五花八门，难以统一调用
• 网络不稳定导致调用超时，没有自动重试机制
• 上下文信息在多个工具间传递困难
• 缺乏有效的错误处理和监控

技术方案对比

目前主流有三种实现方案：

1. OpenAI Function Calling
2. 优点：与 GPT 模型深度集成，自动参数解析

3. 缺点：绑定特定 AI 服务，灵活性较低

4. LangChain Tools

5. 优点：丰富的预置工具，开箱即用

6. 缺点：抽象层级高，自定义困难

7. 自定义 SDK

8. 优点：完全可控，可深度优化
9. 缺点：开发成本高，需要维护

对于大多数场景，我建议从自定义 SDK 开始，因为它能让你真正理解底层原理。

核心实现

1. 工具注册机制

使用 Python 装饰器可以优雅地实现工具注册：

tools = {}

def register_tool(name):
    def decorator(func):
        tools[name] = func
        return func
    return decorator

@register_tool("get_weather")
def get_weather(city: str) -> dict:
    """获取城市天气信息"""
    # 实现代码...

2. 带上下文的调用

加入 JWT 鉴权的示例：

import jwt

def call_with_context(tool_name, params, context):
    try:
        # 验证 JWT
        payload = jwt.decode(context["token"], "secret", algorithms=["HS256"])

        # 调用工具
        if tool_name in tools:
            return tools[tool_name](**params)
    except jwt.ExpiredSignatureError:
        raise Exception("Token expired")

3. 异步调用队列

使用 asyncio 实现并发控制：

import asyncio

semaphore = asyncio.Semaphore(10)  # 限制并发数

async def async_call(tool_name, params):
    async with semaphore:
        try:
            return await tools[tool_name](**params)
        except Exception as e:
            print(f"调用失败: {e}")

生产环境考量

超时与重试

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def call_with_retry(tool_name, params):
    # 实现代码...

结果持久化

建议使用 Redis 作为缓存层，MongoDB 存储完整调用记录。

监控埋点

Prometheus 示例：

from prometheus_client import Counter

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

def call_tool(tool_name, params):
    try:
        result = tools[tool_name](**params)
        TOOL_CALLS.labels(tool_name=tool_name, status='success').inc()
        return result
    except:
        TOOL_CALLS.labels(tool_name=tool_name, status='fail').inc()
        raise

常见问题与优化

1. 避免循环调用
   使用 DAG 检测算法，记录调用链

2. 敏感参数过滤

   import re
   
   def filter_sensitive(data):
       pattern = re.compile(r'(password|token)=[^&]+')
       return pattern.sub(r'\1=*****', data)

3. 冷启动优化

4. 预热常用工具
5. 使用连接池
6. 延迟加载非核心工具

思考题

如何设计工具调用的熔断机制？参考思路：
– 基于错误率自动熔断
– 分级降级策略
– 手动开关

完整实现可以参考：GitHub 示例链接

希望这篇指南能帮助你快速构建可靠的 Agent 工具调用系统。在实际开发中，记得根据业务需求调整方案，并做好监控和日志记录。