工具设计原则

一个好的 Agent 工具应遵循以下原则：

  • 单一职责：每个工具只做一件事，做好一件事
  • 清晰的接口：输入输出明确，易于理解和使用
  • 健壮的错误处理：优雅地处理各种异常情况
  • 可观测性：提供日志和指标，便于调试和监控
  • 安全性：对敏感操作有权限控制和审计

工具接口设计

from pydantic import BaseModel, Field
from typing import Optional, List, Any
from enum import Enum

class ToolCategory(str, Enum):
    SEARCH = "search"
    DATA = "data"
    ACTION = "action"
    UTILITY = "utility"

class ToolParameter(BaseModel):
    name: str
    type: str
    description: str
    required: bool = True
    enum: Optional[List[str]] = None

class ToolDefinition(BaseModel):
    name: str = Field(description="工具名称")
    description: str = Field(description="详细描述，帮助Agent理解何时使用")
    category: ToolCategory
    parameters: List[ToolParameter]
    risk_level: str = Field(default="low", pattern="^(low|medium|high|critical)$")

    def to_openai_function(self) -> dict:
        """转换为 OpenAI Function Calling 格式"""
        properties = {}
        required = []

        for param in self.parameters:
            prop = {"type": param.type, "description": param.description}
            if param.enum:
                prop["enum"] = param.enum
            properties[param.name] = prop
            if param.required:
                required.append(param.name)

        return {
            "name": self.name,
            "description": self.description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required
            }
        }

实现一个完整的工具

import logging
import time
from functools import wraps

logger = logging.getLogger(__name__)

def tool_logger(func):
    """工具装饰器：自动记录调用日志和耗时"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            logger.info(f"Tool {func.__name__} succeeded in {elapsed:.2f}s")
            return result
        except Exception as e:
            elapsed = time.time() - start
            logger.error(f"Tool {func.__name__} failed in {elapsed:.2f}s: {e}")
            raise
    return wrapper

class DatabaseQueryTool:
    """数据库查询工具"""

    definition = ToolDefinition(
        name="query_database",
        description="查询数据库，支持SQL查询。用于获取用户数据、订单信息、统计报表等",
        category=ToolCategory.DATA,
        parameters=[
            ToolParameter(
                name="query_type",
                type="string",
                description="查询类型",
                enum=["user", "order", "stats"]
            ),
            ToolParameter(
                name="filters",
                type="object",
                description="查询过滤条件",
                required=False
            )
        ],
        risk_level="medium"
    )

    def __init__(self, db_connection):
        self.db = db_connection
        self.allowed_tables = ["users", "orders", "products"]

    @tool_logger
    def execute(self, query_type: str, filters: dict = None):
        # 安全校验
        if query_type not in ["user", "order", "stats"]:
            return {"error": f"不支持的查询类型: {query_type}"}

        # 构建安全的查询
        query = self._build_safe_query(query_type, filters or {})

        # 执行查询
        result = self.db.execute(query)

        return {
            "success": True,
            "data": result,
            "count": len(result)
        }

    def _build_safe_query(self, query_type: str, filters: dict):
        # 使用参数化查询，防止 SQL 注入
        # 只允许查询预定义的表
        pass

工具注册与发现

class ToolRegistry:
    """工具注册中心"""

    def __init__(self):
        self._tools: dict[str, Any] = {}

    def register(self, tool):
        """注册工具"""
        name = tool.definition.name
        if name in self._tools:
            raise ValueError(f"工具 {name} 已注册")
        self._tools[name] = tool
        logger.info(f"注册工具: {name}")

    def get_all_definitions(self):
        """获取所有工具定义（用于 Function Calling）"""
        return [
            tool.definition.to_openai_function()
            for tool in self._tools.values()
        ]

    def execute(self, name: str, params: dict):
        """执行工具"""
        if name not in self._tools:
            return {"error": f"工具不存在: {name}"}

        tool = self._tools[name]

        # 高风险操作需要确认
        if tool.definition.risk_level in ["high", "critical"]:
            if not self._confirm_execution(name, params):
                return {"error": "操作被用户取消"}

        return tool.execute(**params)

工具测试

import pytest

class TestDatabaseQueryTool:
    def setup_method(self):
        self.tool = DatabaseQueryTool(MockDB())

    def test_valid_query(self):
        result = self.tool.execute("user", {"name": "张三"})
        assert result["success"] is True
        assert "data" in result

    def test_invalid_query_type(self):
        result = self.tool.execute("delete_all")
        assert "error" in result

    def test_sql_injection_prevention(self):
        result = self.tool.execute("user", {"name": "'; DROP TABLE users; --"})
        assert result["success"] is True  # 不应该执行恶意SQL

工具性能优化

  • 结果缓存：对相同参数的查询结果进行缓存
  • 超时控制：为每个工具设置执行超时时间
  • 并行执行：独立的工具调用可以并行执行
  • 结果裁剪：返回给 LLM 的结果应精简，只包含必要信息

总结

工具是 Agent 能力的延伸。好的工具设计让 Agent 如虎添翼，差的工具设计让 Agent 寸步难行。投入时间做好工具设计和测试，是构建高质量 Agent 系统的关键。