Tool/Function 设计规范

Tool 和 Function 是 Agent 与外部世界交互的桥梁，良好的设计能提高 Agent 的可靠性和效率。

一、核心概念

1.1 Tool vs Function

Function（函数）:
- 单一功能
- 确定性输出
- 无副作用

Tool（工具）:
- 可包含多个 Function
- 可能有副作用
- 支持状态管理

1.2 使用场景

# Function 调用
- 天气查询
- 计算器
- 翻译服务
- 数据库查询

# Tool 调用
- 搜索引擎
- 代码执行
- API 调用
- 文件操作

二、定义规范

2.1 JSON Schema 定义

TOOL_SCHEMA = {
    "name": "search",
    "description": "搜索互联网获取信息",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词"
            },
            "num_results": {
                "type": "integer",
                "description": "返回结果数量",
                "default": 5
            }
        },
        "required": ["query"]
    }
}

2.2 Python 实现

from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(..., description="搜索关键词")
    num_results: int = Field(5, description="返回结果数量", ge=1, le=20)

class SearchResult(BaseModel):
    title: str
    url: str
    snippet: str

async def search_tool(input: SearchInput) -> list[SearchResult]:
    """搜索互联网获取信息"""
    results = await search_api(input.query, input.num_results)
    return [SearchResult(**r) for r in results]

2.3 注册机制

class ToolRegistry:
    def __init__(self):
        self.tools = {}
    
    def register(self, func: callable, schema: dict):
        name = schema["name"]
        self.tools[name] = {
            "func": func,
            "schema": schema
        }
    
    def get(self, name: str) -> dict:
        return self.tools.get(name)
    
    def list_tools(self) -> list[dict]:
        return [
            {"name": name, "schema": t["schema"]}
            for name, t in self.tools.items()
        ]

三、参数验证

3.1 类型检查

def validate_params(params: dict, schema: dict) -> tuple[bool, str]:
    # 检查必填参数
    required = schema.get("required", [])
    for param in required:
        if param not in params:
            return False, f"Missing required parameter: {param}"
    
    # 检查类型
    properties = schema.get("properties", {})
    for name, value in params.items():
        if name not in properties:
            continue
        
        expected_type = properties[name].get("type")
        if not check_type(value, expected_type):
            return False, f"Invalid type for {name}"
    
    return True, ""

3.2 范围验证

def validate_range(value: any, constraints: dict) -> bool:
    if "minimum" in constraints and value < constraints["minimum"]:
        return False
    if "maximum" in constraints and value > constraints["maximum"]:
        return False
    if "minLength" in constraints and len(value) < constraints["minLength"]:
        return False
    if "maxLength" in constraints and len(value) > constraints["maxLength"]:
        return False
    return True

3.3 自定义验证

def validate_email(value: str) -> bool:
    import re
    pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
    return bool(re.match(pattern, value))

def validate_date(value: str) -> bool:
    try:
        datetime.strptime(value, "%Y-%m-%d")
        return True
    except ValueError:
        return False

四、错误处理

4.1 错误分类

class ToolError(Exception):
    """基础错误"""
    pass

class ValidationError(ToolError):
    """参数验证错误"""
    pass

class ExecutionError(ToolError):
    """执行错误"""
    pass

class TimeoutError(ToolError):
    """超时错误"""
    pass

4.2 错误响应

def execute_tool(name: str, params: dict) -> dict:
    try:
        tool = registry.get(name)
        if not tool:
            return {
                "success": False,
                "error": f"Tool not found: {name}"
            }
        
        # 验证参数
        valid, msg = validate_params(params, tool["schema"])
        if not valid:
            return {
                "success": False,
                "error": msg
            }
        
        # 执行
        result = await tool["func"](**params)
        return {
            "success": True,
            "result": result
        }
        
    except ValidationError as e:
        return {"success": False, "error": str(e)}
    except TimeoutError as e:
        return {"success": False, "error": "Timeout"}
    except Exception as e:
        return {"success": False, "error": f"Internal error: {e}"}

4.3 重试机制

async def execute_with_retry(
    name: str, 
    params: dict,
    max_retries: int = 3
) -> dict:
    for i in range(max_retries):
        result = await execute_tool(name, params)
        
        if result["success"]:
            return result
        
        # 可重试的错误
        if is_retryable_error(result["error"]):
            await asyncio.sleep(2 ** i)  # 指数退避
            continue
        
        # 不可重试的错误
        return result
    
    return result

五、最佳实践

5.1 设计原则

1. 单一职责：一个 Tool 只做一件事
2. 明确定义：清晰的输入输出
3. 幂等性：多次调用结果相同
4. 无状态：不依赖外部状态
5. 可测试：易于单元测试

5.2 文档规范

TOOL_DOCUMENTATION = """
# Tool 名称：search

## 描述
搜索互联网获取相关信息

## 参数
- query (string, required): 搜索关键词
- num_results (integer, optional): 返回结果数量，默认 5，范围 1-20

## 返回值
列表，每个元素包含：
- title: 标题
- url: 链接
- snippet: 摘要

## 错误
- INVALID_QUERY: 查询无效
- TIMEOUT: 搜索超时
- RATE_LIMIT: 请求频率超限

## 示例
输入：{"query": "AI news", "num_results": 3}
输出：[{"title": "...", "url": "...", "snippet": "..."}]
"""

5.3 性能优化

# 连接池
class SearchTool:
    def __init__(self):
        self.session = aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(limit=100)
        )
    
    async def close(self):
        await self.session.close()

# 缓存
from functools import lru_cache

class CachedTool:
    @lru_cache(maxsize=1000)
    async def search(self, query: str) -> list:
        return await self._search_impl(query)

六、总结

Tool/Function 设计核心要点：

良好的 Tool/Function 设计是构建可靠 Agent 系统的基础。