自定义工具开发：让AI Agent连接你的业务系统

一、开场：AI Agent不能只会”聊天”

大家好，我是老金。

上周有个做电商的朋友问我：”你们那个AI Agent能帮我查订单吗？”

我说：”能啊，给它加个工具就行。”

他：”那能帮我发短信通知客户吗？”

我：”能，加个工具。”

他：”能对接我们ERP系统吗？”

我：”能…等等，你这需求有点多啊。”

他笑了：”不就是写几个API调用吗？”

好吧，他说得对。AI Agent的核心能力之一，就是通过工具连接外部系统。今天就来聊聊自定义工具开发的最佳实践。

二、AI Agent工具的本质

工具是什么？

从AI的角度看，工具就是一个函数：

def query_order(order_id: str) -> dict:
    """查询订单状态"""
    # 调用后端API
    response = requests.get(f"/api/orders/{order_id}")
    return response.json()

从LLM的角度看，工具是一个描述：

{
  "name": "query_order",
  "description": "查询订单状态，需要订单号",
  "parameters": {
    "type": "object",
    "properties": {
      "order_id": {
        "type": "string",
        "description": "订单号，格式为ORD开头"
      }
    },
    "required": ["order_id"]
  }
}

LLM通过描述理解工具，生成调用参数，执行后获得结果。

工具调用流程

1. 用户提问："帮我查订单ORD12345"
2. LLM理解意图，选择工具：query_order
3. LLM提取参数：{"order_id": "ORD12345"}
4. 执行工具，获得结果
5. LLM基于结果生成回复

三、工具描述的艺术

描述决定了AI能不能用好工具

看两个例子：

# ❌ 糟糕的描述
{
    "name": "search",
    "description": "搜索",
    "parameters": {
        "type": "object",
        "properties": {
            "q": {"type": "string"}
        }
    }
}

# AI不知道：
# - 搜索什么？商品？订单？用户？
# - q是什么格式？
# - 什么时候该用这个工具？

# ✅ 好的描述
{
    "name": "search_product",
    "description": "搜索商品信息。当用户询问商品价格、规格、库存时使用。返回匹配的商品列表。",
    "parameters": {
        "type": "object",
        "properties": {
            "keyword": {
                "type": "string",
                "description": "搜索关键词，可以是商品名称、品牌或型号"
            },
            "category": {
                "type": "string",
                "description": "商品分类，可选值：electronics, clothing, food",
                "enum": ["electronics", "clothing", "food"]
            },
            "limit": {
                "type": "integer",
                "description": "返回结果数量，默认10",
                "default": 10
            }
        },
        "required": ["keyword"]
    }
}

描述优化清单

元素	要点	示例
名称	动词+名词，语义清晰	query_order而非order
描述	何时使用、返回什么	“查询订单状态，返回物流信息”
参数名	自解释	order_id而非id
参数描述	格式、约束、默认值	“订单号，ORD开头，10位”
枚举值	明确可选范围	"enum": ["pending", "shipped"]

复杂参数的处理

# 嵌套对象参数
{
    "name": "create_order",
    "parameters": {
        "type": "object",
        "properties": {
            "items": {
                "type": "array",
                "description": "商品列表",
                "items": {
                    "type": "object",
                    "properties": {
                        "product_id": {"type": "string"},
                        "quantity": {"type": "integer", "minimum": 1}
                    }
                }
            },
            "shipping_address": {
                "type": "object",
                "description": "收货地址",
                "properties": {
                    "province": {"type": "string"},
                    "city": {"type": "string"},
                    "detail": {"type": "string"}
                }
            }
        }
    }
}

四、工具开发模式

模式1：简单封装

直接封装现有API：

import requests
from typing import Optional

class OrderTools:
    """订单相关工具"""

    BASE_URL = "https://api.example.com"

    @staticmethod
    def query_order(order_id: str) -> dict:
        """查询订单

        Args:
            order_id: 订单号

        Returns:
            订单信息
        """
        response = requests.get(
            f"{OrderTools.BASE_URL}/orders/{order_id}",
            headers={"Authorization": f"Bearer {API_TOKEN}"}
        )
        return response.json()

    @staticmethod
    def cancel_order(order_id: str, reason: Optional[str] = None) -> dict:
        """取消订单"""
        response = requests.post(
            f"{OrderTools.BASE_URL}/orders/{order_id}/cancel",
            json={"reason": reason}
        )
        return response.json()

模式2：智能处理

增加智能处理逻辑：

class SmartOrderTools:
    """智能订单工具"""

    @staticmethod
    def query_order(query: str) -> dict:
        """智能查询订单

        支持多种查询方式：
        - 订单号：ORD12345
        - 手机号：13800138000
        - 用户名：张三
        """
        # 自动识别查询类型
        if query.startswith("ORD"):
            return OrderTools.query_by_order_id(query)
        elif query.isdigit() and len(query) == 11:
            return OrderTools.query_by_phone(query)
        else:
            return OrderTools.query_by_username(query)

    @staticmethod
    def query_by_order_id(order_id):
        # ...
        pass

    @staticmethod
    def query_by_phone(phone):
        # ...
        pass

模式3：组合工具

多个API组合成一个工具：

async def process_return(order_id: str, item_index: int, reason: str) -> dict:
    """处理退货申请

    自动完成退货流程：
    1. 查询订单
    2. 检查退货条件
    3. 创建退货单
    4. 发送通知
    """
    # 1. 查询订单
    order = await get_order(order_id)

    # 2. 检查退货条件
    if order.status not in ["delivered", "received"]:
        return {"error": "订单状态不支持退货"}

    item = order.items[item_index]
    if item.return_deadline  dict:
    """安全的订单查询"""
    try:
        # 验证输入
        validated = OrderQueryInput(order_id=order_id)
        # 执行查询
        return query_order(validated.order_id)
    except ValidationError as e:
        return {"error": f"输入验证失败: {e}"}

权限控制

from functools import wraps

def require_permission(permission: str):
    """权限装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, user_context=None, **kwargs):
            if not user_context:
                raise PermissionError("缺少用户上下文")

            if permission not in user_context.permissions:
                raise PermissionError(f"需要权限: {permission}")

            return func(*args, user_context=user_context, **kwargs)
        return wrapper
    return decorator

@require_permission("order:write")
def cancel_order(order_id: str, user_context=None) -> dict:
    """取消订单（需要写权限）"""
    # ...
    pass

敏感信息处理

def mask_sensitive_data(data: dict) -> dict:
    """脱敏处理"""
    masked = data.copy()

    # 手机号脱敏
    if "phone" in masked:
        masked["phone"] = masked["phone"][:3] + "****" + masked["phone"][-4:]

    # 身份证脱敏
    if "id_card" in masked:
        masked["id_card"] = masked["id_card"][:6] + "********" + masked["id_card"][-4:]

    # 银行卡脱敏
    if "bank_card" in masked:
        masked["bank_card"] = "****" + masked["bank_card"][-4:]

    return masked

def query_user_info(user_id: str) -> dict:
    """查询用户信息"""
    user_info = db.get_user(user_id)
    return mask_sensitive_data(user_info)

调用限制

from collections import defaultdict
from time import time

class RateLimiter:
    """速率限制器"""

    def __init__(self, max_calls: int, period: int):
        self.max_calls = max_calls
        self.period = period
        self.calls = defaultdict(list)

    def is_allowed(self, key: str) -> bool:
        """检查是否允许调用"""
        now = time()
        calls = self.calls[key]

        # 清理过期记录
        self.calls[key] = [t for t in calls if now - t = self.max_calls:
            return False

        self.calls[key].append(now)
        return True

# 使用示例
limiter = RateLimiter(max_calls=10, period=60)  # 每分钟最多10次

def send_sms_with_limit(phone: str, message: str) -> dict:
    """发送短信（带限流）"""
    if not limiter.is_allowed(phone):
        return {"error": "发送太频繁，请稍后再试"}

    return send_sms(phone, message)

六、错误处理

错误类型与处理

class ToolError(Exception):
    """工具错误基类"""
    pass

class ValidationError(ToolError):
    """输入验证错误"""
    pass

class PermissionError(ToolError):
    """权限错误"""
    pass

class ExternalAPIError(ToolError):
    """外部API错误"""
    pass

def handle_tool_error(func):
    """错误处理装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except ValidationError as e:
            return {"error": f"输入错误: {str(e)}", "code": "INVALID_INPUT"}
        except PermissionError as e:
            return {"error": f"权限不足: {str(e)}", "code": "FORBIDDEN"}
        except ExternalAPIError as e:
            return {"error": "服务暂时不可用，请稍后重试", "code": "SERVICE_UNAVAILABLE"}
        except Exception as e:
            # 记录详细日志
            logger.error(f"Tool error: {e}", exc_info=True)
            return {"error": "系统错误，请稍后重试", "code": "INTERNAL_ERROR"}
    return wrapper

友好的错误信息

# ❌ 对用户不友好的错误
return {"error": "HTTPConnectionPool: Max retries exceeded"}

# ✅ 对用户友好的错误
return {"error": "订单系统暂时不可用，请稍后再试"}

重试机制

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def call_external_api(url: str, data: dict) -> dict:
    """调用外部API（带重试）"""
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=data) as response:
            if response.status >= 500:
                raise ExternalAPIError(f"Server error: {response.status}")
            return await response.json()

七、工具测试

单元测试

import pytest

class TestOrderTools:
    """订单工具测试"""

    def test_query_order_success(self, mocker):
        """测试正常查询"""
        # Mock API响应
        mocker.patch(
            "requests.get",
            return_value=MockResponse({
                "order_id": "ORD12345",
                "status": "shipped"
            })
        )

        result = OrderTools.query_order("ORD12345")

        assert result["order_id"] == "ORD12345"
        assert result["status"] == "shipped"

    def test_query_order_not_found(self, mocker):
        """测试订单不存在"""
        mocker.patch(
            "requests.get",
            return_value=MockResponse({"error": "Order not found"}, status=404)
        )

        result = OrderTools.query_order("ORD99999")

        assert "error" in result

    def test_query_order_invalid_id(self):
        """测试无效订单号"""
        with pytest.raises(ValidationError):
            OrderTools.query_order("invalid")

集成测试

@pytest.fixture
def test_agent():
    """测试用Agent"""
    return Agent(
        tools=[OrderTools.query_order, OrderTools.cancel_order],
        llm=MockLLM()
    )

class TestAgentWithTools:
    """Agent工具集成测试"""

    async def test_agent_can_query_order(self, test_agent):
        """测试Agent能正确调用查询工具"""
        response = await test_agent.run("查一下订单ORD12345")

        assert "ORD12345" in response
        assert "已发货" in response  # 假设mock返回shipped

    async def test_agent_handles_tool_error(self, test_agent):
        """测试Agent能处理工具错误"""
        response = await test_agent.run("查一下订单INVALID")

        assert "无法查询" in response or "错误" in response

八、工具注册与管理

工具注册表

class ToolRegistry:
    """工具注册表"""

    def __init__(self):
        self.tools = {}
        self.categories = defaultdict(list)

    def register(self, name: str, func: callable, category: str = "general"):
        """注册工具"""
        tool_schema = self.generate_schema(name, func)
        self.tools[name] = {
            "function": func,
            "schema": tool_schema,
            "category": category
        }
        self.categories[category].append(name)

    def generate_schema(self, name: str, func: callable) -> dict:
        """自动生成工具schema"""
        sig = inspect.signature(func)
        doc = inspect.getdoc(func)

        schema = {
            "name": name,
            "description": doc.split("n")[0] if doc else name,
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }

        for param_name, param in sig.parameters.items():
            param_schema = self.infer_param_schema(param)
            schema["parameters"]["properties"][param_name] = param_schema

            if param.default == inspect.Parameter.empty:
                schema["parameters"]["required"].append(param_name)

        return schema

    def get_tools_for_agent(self, categories: list = None) -> list:
        """获取指定分类的工具"""
        if not categories:
            return [t["schema"] for t in self.tools.values()]

        tools = []
        for cat in categories:
            tools.extend([self.tools[name]["schema"] 
                         for name in self.categories[cat]])
        return tools

# 使用示例
registry = ToolRegistry()
registry.register("query_order", OrderTools.query_order, category="order")
registry.register("send_sms", NotificationTools.send_sms, category="notification")

九、实战案例：客服系统工具集

工具定义

class CustomerServiceTools:
    """客服系统工具集"""

    # 订单工具
    @staticmethod
    def query_order(order_id: str) -> dict:
        """查询订单状态

        Args:
            order_id: 订单号，格式ORD开头

        Returns:
            订单信息，包含状态、物流等
        """
        # 实现...
        pass

    @staticmethod
    def cancel_order(order_id: str, reason: str) -> dict:
        """取消订单

        Args:
            order_id: 订单号
            reason: 取消原因
        """
        # 实现...
        pass

    # 商品工具
    @staticmethod
    def search_product(keyword: str, category: str = None) -> list:
        """搜索商品

        Args:
            keyword: 搜索关键词
            category: 商品分类（可选）

        Returns:
            商品列表
        """
        # 实现...
        pass

    # 通知工具
    @staticmethod
    def send_notification(user_id: str, message: str) -> dict:
        """发送通知给用户"""
        # 实现...
        pass

    # 工单工具
    @staticmethod
    def create_ticket(user_id: str, issue: str, priority: str = "normal") -> dict:
        """创建工单

        Args:
            user_id: 用户ID
            issue: 问题描述
            priority: 优先级，可选low/normal/high/urgent
        """
        # 实现...
        pass

工具Schema生成

# 自动生成OpenAI Function Calling格式的schema
CUSTOMER_SERVICE_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "query_order",
            "description": "查询订单状态，当用户询问订单进度、物流信息时使用",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单号，格式为ORD开头"
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "cancel_order",
            "description": "取消订单，当用户明确要求取消订单时使用",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单号"
                    },
                    "reason": {
                        "type": "string",
                        "description": "取消原因"
                    }
                },
                "required": ["order_id", "reason"]
            }
        }
    },
    # ... 更多工具
]

十、总结与最佳实践

工具开发检查清单

阶段	检查项
设计	名称语义清晰、描述完整、参数最小化
开发	输入验证、错误处理、日志记录
安全	权限控制、敏感信息脱敏、速率限制
测试	单元测试、集成测试、边界测试
文档	使用示例、错误码说明、更新日志

工具设计原则

1. 单一职责：一个工具做一件事
2. 原子操作：工具应该可以独立执行
3. 幂等性：重复执行结果相同
4. 可观测：有日志、有监控
5. 可恢复：失败可以重试或回滚

下期预告

明天聊聊AI Agent并发控制——高并发场景下如何保证稳定性！

---

往期回顾

• AI Agent知识库构建：RAG系统最佳实践
• AI Agent成本控制指南：Token花得肉疼怎么办？
• AI Agent评估体系：如何量化”智能程度”？