16 开发自定义工具

第七层：开发自定义工具

📌 参考：08-工具系统架构.md

快速开始

1. 最简工具模板

from nanobot.agent.tools.base import Tool

class HelloTool(Tool):
    @property
    def name(self) -> str:
        return "hello"
    
    @property
    def description(self) -> str:
        return "Say hello to someone"
    
    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "name": {
                    "type": "string",
                    "description": "Name to greet"
                }
            },
            "required": ["name"]
        }
    
    async def execute(self, name: str) -> str:
        return f"Hello, {name}!"

# 注册
agent.tools.register(HelloTool())

2. 测试工具

import asyncio

# 单元测试
async def test_hello():
    tool = HelloTool()
    result = await tool.execute(name="Alice")
    assert result == "Hello, Alice!"

asyncio.run(test_hello())

进阶示例

示例 1：HTTP API 调用

import httpx
from typing import Any

class WeatherTool(Tool):
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    @property
    def name(self) -> str:
        return "get_weather"
    
    @property
    def description(self) -> str:
        return "Get weather for a city"
    
    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "City name"
                }
            },
            "required": ["city"]
        }
    
    async def execute(self, city: str) -> str:
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    "https://api.openweathermap.org/data/2.5/weather",
                    params={"q": city, "appid": self.api_key}
                )
                response.raise_for_status()
                
                data = response.json()
                temp = data["main"]["temp"] - 273.15  # K to C
                desc = data["weather"][0]["description"]
                
                return f"{city}: {temp:.1f}°C, {desc}"
        
        except httpx.HTTPError as e:
            return f"Error fetching weather: {e}"

示例 2：数据库查询

import sqlite3

class DBQueryTool(Tool):
    def __init__(self, db_path: str):
        self.db_path = db_path
    
    @property
    def name(self) -> str:
        return "query_db"
    
    @property
    def description(self) -> str:
        return "Execute SQL query on database"
    
    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "SQL query (SELECT only)"
                }
            },
            "required": ["query"]
        }
    
    async def execute(self, query: str) -> str:
        # 安全检查
        if not query.strip().upper().startswith("SELECT"):
            return "Error: Only SELECT queries allowed"
        
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.execute(query)
            rows = cursor.fetchall()
            conn.close()
            
            if not rows:
                return "No results"
            
            # 格式化结果
            return "\n".join([str(row) for row in rows])
        
        except Exception as e:
            return f"Error: {e}"

示例 3：使用第三方库

from PIL import Image

class ImageInfoTool(Tool):
    @property
    def name(self) -> str:
        return "image_info"
    
    @property
    def description(self) -> str:
        return "Get image information"
    
    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Path to image"
                }
            },
            "required": ["path"]
        }
    
    async def execute(self, path: str) -> str:
        try:
            img = Image.open(path)
            return f"Size: {img.size}, Format: {img.format}, Mode: {img.mode}"
        except Exception as e:
            return f"Error: {e}"

最佳实践

1. 参数验证

@property
def parameters(self) -> dict:
    return {
        "type": "object",
        "properties": {
            "count": {
                "type": "integer",
                "description": "Number of items",
                "minimum": 1,
                "maximum": 100
            },
            "format": {
                "type": "string",
                "description": "Output format",
                "enum": ["json", "csv", "text"]
            }
        },
        "required": ["count"]
    }

2. 错误处理

async def execute(self, **kwargs) -> str:
    try:
        result = await self.do_work(**kwargs)
        return result
    
    except ValueError as e:
        return f"Invalid input: {e}"
    
    except httpx.HTTPError as e:
        return f"Network error: {e}"
    
    except Exception as e:
        logger.exception("Unexpected error in tool")
        return f"Error: {e}"

3. 添加日志

from loguru import logger

async def execute(self, url: str) -> str:
    logger.info(f"Fetching URL: {url}")
    
    try:
        result = await fetch(url)
        logger.debug(f"Got {len(result)} bytes")
        return result
    except Exception as e:
        logger.error(f"Failed to fetch {url}: {e}")
        return f"Error: {e}"

4. 超时控制

import asyncio

async def execute(self, url: str) -> str:
    try:
        result = await asyncio.wait_for(
            self.fetch_url(url),
            timeout=30.0
        )
        return result
    except asyncio.TimeoutError:
        return "Error: Request timeout"

工具注册

在 AgentLoop 中注册

修改 _register_default_tools：

def _register_default_tools(self):
    # ...现有工具...
    
    # 自定义工具
    self.tools.register(WeatherTool(api_key=self.weather_api_key))
    self.tools.register(DBQueryTool(db_path=str(self.workspace / "data.db")))

动态加载工具

def load_custom_tools(self, tools_dir: Path):
    """从目录加载自定义工具"""
    import importlib.util
    
    for tool_file in tools_dir.glob("*_tool.py"):
        # 动态导入
        spec = importlib.util.spec_from_file_location(
            tool_file.stem, tool_file
        )
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        
        # 查找 Tool 子类
        for name in dir(module):
            obj = getattr(module, name)
            if isinstance(obj, type) and issubclass(obj, Tool) and obj != Tool:
                self.tools.register(obj())

开发工作流

1. 创建工具文件：my_tool.py
2. 实现 Tool 接口
3. 本地测试：pytest 或手动测试
4. 注册到 Agent
5. 通过对话测试

小结

• ✅ 继承 Tool 基类并实现 4 个属性/方法
• ✅ 使用 JSON Schema 定义参数
• ✅ 返回字符串结果
• ✅ 完善的错误处理

参考：08-工具系统架构.md 了解更多

下一步：17-开发自定义技能.md