AI Agent开发实战（三）：工具调用让Agent动起来

一、开场：只会聊天的Agent是不够的

大家好，我是老金。

上一篇我们实现了基本的对话Agent，但只会聊天的Agent用处有限。

真正的Agent需要能够执行动作：

• 查询天气
• 搜索网络
• 操作数据库
• 调用API

今天我们给Agent添加工具调用能力。

二、工具调用原理

2.1 OpenAI Function Calling

OpenAI的Function Calling机制：

用户输入 → Agent思考 → 决定调用工具 → 执行工具 → 返回结果 → 继续思考 → 最终回答

2.2 工具定义格式

# OpenAI工具定义格式
tool_definition = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "获取指定城市的天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，如：北京"
                }
            },
            "required": ["city"]
        }
    }
}

三、工具抽象设计

3.1 工具基类

# src/tools/base.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from pydantic import BaseModel
import json

class ToolResult(BaseModel):
    """工具执行结果"""
    success: bool
    result: Any
    error: Optional[str] = None

    def to_string(self) -> str:
        """转换为字符串（返回给LLM）"""
        if self.success:
            if isinstance(self.result, str):
                return self.result
            return json.dumps(self.result, ensure_ascii=False)
        else:
            return f"错误: {self.error}"

class BaseTool(ABC):
    """工具基类"""

    name: str = "base_tool"
    description: str = "基础工具"
    parameters_schema: Dict[str, Any] = {}

    @abstractmethod
    async def execute(self, **kwargs) -> ToolResult:
        """执行工具"""
        pass

    def get_definition(self) -> Dict[str, Any]:
        """获取工具定义（OpenAI格式）"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters_schema
            }
        }

    def validate_parameters(self, **kwargs) -> bool:
        """验证参数"""
        required = self.parameters_schema.get("required", [])
        for param in required:
            if param not in kwargs:
                return False
        return True

3.2 工具注册表

# src/tools/registry.py
from typing import Dict, List, Type
from .base import BaseTool

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

    def __init__(self):
        self._tools: Dict[str, BaseTool] = {}

    def register(self, tool: BaseTool):
        """注册工具"""
        self._tools[tool.name] = tool

    def get(self, name: str) -> BaseTool:
        """获取工具"""
        return self._tools.get(name)

    def get_all(self) -> List[BaseTool]:
        """获取所有工具"""
        return list(self._tools.values())

    def get_definitions(self) -> List[Dict[str, Any]]:
        """获取所有工具定义"""
        return [tool.get_definition() for tool in self._tools.values()]

    def list_tools(self) -> List[str]:
        """列出所有工具名"""
        return list(self._tools.keys())

# 全局工具注册表
tool_registry = ToolRegistry()

def register_tool(tool_class: Type[BaseTool]):
    """工具注册装饰器"""
    tool = tool_class()
    tool_registry.register(tool)
    return tool_class

四、实现具体工具

4.1 天气查询工具

# src/tools/weather.py
from .base import BaseTool, ToolResult
from .registry import register_tool
import httpx

@register_tool
class WeatherTool(BaseTool):
    """天气查询工具"""

    name = "get_weather"
    description = "获取指定城市的天气信息"
    parameters_schema = {
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "城市名称，如：北京、上海"
            }
        },
        "required": ["city"]
    }

    async def execute(self, city: str, **kwargs) -> ToolResult:
        """执行天气查询"""
        try:
            # 这里使用模拟数据，实际应调用天气API
            # 例如：和风天气、心知天气等

            # 模拟天气数据
            weather_data = {
                "北京": {"temp": 15, "weather": "晴", "humidity": 45},
                "上海": {"temp": 18, "weather": "多云", "humidity": 60},
                "广州": {"temp": 25, "weather": "晴", "humidity": 70},
                "深圳": {"temp": 26, "weather": "多云", "humidity": 75},
            }

            if city in weather_data:
                data = weather_data[city]
                return ToolResult(
                    success=True,
                    result=f"{city}当前天气：{data['weather']}，温度{data['temp']}℃，湿度{data['humidity']}%"
                )
            else:
                return ToolResult(
                    success=True,
                    result=f"未找到{city}的天气数据，请尝试其他城市"
                )
        except Exception as e:
            return ToolResult(success=False, result=None, error=str(e))

4.2 网络搜索工具

# src/tools/search.py
from .base import BaseTool, ToolResult
from .registry import register_tool
import httpx
import json

@register_tool
class WebSearchTool(BaseTool):
    """网络搜索工具"""

    name = "web_search"
    description = "搜索网络获取信息"
    parameters_schema = {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词"
            },
            "num_results": {
                "type": "integer",
                "description": "返回结果数量，默认3",
                "default": 3
            }
        },
        "required": ["query"]
    }

    async def execute(self, query: str, num_results: int = 3, **kwargs) -> ToolResult:
        """执行搜索"""
        try:
            # 模拟搜索结果
            # 实际应调用搜索API（如Bing、Google Custom Search等）

            mock_results = [
                {
                    "title": f"关于{query}的介绍",
                    "snippet": f"这是关于{query}的详细信息和说明...",
                    "url": f"https://example.com/{query}"
                },
                {
                    "title": f"{query}最新动态",
                    "snippet": f"{query}相关的最新新闻和进展...",
                    "url": f"https://news.example.com/{query}"
                },
                {
                    "title": f"如何学习{query}",
                    "snippet": f"{query}入门教程和学习资源...",
                    "url": f"https://tutorial.example.com/{query}"
                }
            ]

            results = mock_results[:num_results]
            return ToolResult(
                success=True,
                result={
                    "query": query,
                    "results": results
                }
            )
        except Exception as e:
            return ToolResult(success=False, result=None, error=str(e))

4.3 计算器工具

# src/tools/calculator.py
from .base import BaseTool, ToolResult
from .registry import register_tool
import math

@register_tool
class CalculatorTool(BaseTool):
    """计算器工具"""

    name = "calculate"
    description = "执行数学计算"
    parameters_schema = {
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "数学表达式，如：2+2, sqrt(16), sin(3.14)"
            }
        },
        "required": ["expression"]
    }

    # 允许的数学函数
    SAFE_FUNCTIONS = {
        'abs': abs, 'round': round, 'min': min, 'max': max,
        'sqrt': math.sqrt, 'sin': math.sin, 'cos': math.cos,
        'tan': math.tan, 'log': math.log, 'log10': math.log10,
        'exp': math.exp, 'pow': pow, 'pi': math.pi, 'e': math.e
    }

    async def execute(self, expression: str, **kwargs) -> ToolResult:
        """执行计算"""
        try:
            # 安全执行数学表达式
            # 只允许数字和指定的数学函数
            result = eval(expression, {"__builtins__": {}}, self.SAFE_FUNCTIONS)
            return ToolResult(
                success=True,
                result=f"计算结果：{expression} = {result}"
            )
        except Exception as e:
            return ToolResult(
                success=False,
                result=None,
                error=f"计算错误：{str(e)}"
            )

4.4 时间工具

# src/tools/datetime_tool.py
from .base import BaseTool, ToolResult
from .registry import register_tool
from datetime import datetime
import pytz

@register_tool
class DateTimeTool(BaseTool):
    """日期时间工具"""

    name = "get_datetime"
    description = "获取当前日期时间"
    parameters_schema = {
        "type": "object",
        "properties": {
            "timezone": {
                "type": "string",
                "description": "时区，如：Asia/Shanghai，默认本地时区",
                "default": "local"
            },
            "format": {
                "type": "string",
                "description": "日期格式，如：%Y-%m-%d %H:%M:%S",
                "default": "%Y-%m-%d %H:%M:%S"
            }
        },
        "required": []
    }

    async def execute(self, timezone: str = "local", format: str = "%Y-%m-%d %H:%M:%S", **kwargs) -> ToolResult:
        """获取时间"""
        try:
            if timezone == "local":
                now = datetime.now()
            else:
                tz = pytz.timezone(timezone)
                now = datetime.now(tz)

            formatted = now.strftime(format)
            return ToolResult(
                success=True,
                result=f"当前时间：{formatted}（时区：{timezone}）"
            )
        except Exception as e:
            return ToolResult(success=False, result=None, error=str(e))

五、工具调用Agent

5.1 ToolAgent实现

# src/agents/tool_agent.py
from .base import BaseAgent, Message
from ..tools.registry import tool_registry
from ..utils.llm_client import LLMClient
from ..utils.logger import logger
from typing import List, Dict, Any, Optional
import json

class ToolAgent(BaseAgent):
    """工具调用Agent"""

    def __init__(
        self,
        llm_client: LLMClient,
        tools: List[str] = None,  # 指定可用的工具名
        **kwargs
    ):
        super().__init__(llm_client, **kwargs)

        # 注册工具
        self.available_tools = {}
        if tools:
            for tool_name in tools:
                tool = tool_registry.get(tool_name)
                if tool:
                    self.available_tools[tool_name] = tool
        else:
            # 使用所有注册的工具
            self.available_tools = {t.name: t for t in tool_registry.get_all()}

    def _get_tool_definitions(self) -> List[Dict[str, Any]]:
        """获取工具定义"""
        return [tool.get_definition() for tool in self.available_tools.values()]

    async def think(self) -> str:
        """思考（可能调用工具）"""
        messages = self.get_history()
        tools = self._get_tool_definitions()

        # 调用LLM（带工具）
        response = await self.llm.chat_with_tools(
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )

        # 检查是否需要调用工具
        if response.get("tool_calls"):
            return await self._handle_tool_calls(response["tool_calls"])
        else:
            # 直接返回内容
            content = response.get("content", "")
            if content:
                self._add_message("assistant", content)
            return content

    async def _handle_tool_calls(self, tool_calls: List[Any]) -> str:
        """处理工具调用"""
        results = []

        for tool_call in tool_calls:
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)

            logger.info(f"调用工具: {tool_name}({tool_args})")

            # 执行工具
            result = await self._execute_tool(tool_name, tool_args)
            results.append(result)

            # 添加工具调用到历史
            self._add_message("assistant", None)  # 会在下一轮被使用
            self.state.messages.append(Message(
                role="tool",
                content=result.to_string(),
                name=tool_name
            ))

        # 继续思考
        return await self.think()

    async def _execute_tool(self, tool_name: str, args: Dict[str, Any]) -> ToolResult:
        """执行工具"""
        tool = self.available_tools.get(tool_name)

        if not tool:
            return ToolResult(
                success=False,
                result=None,
                error=f"未知工具：{tool_name}"
            )

        # 验证参数
        if not tool.validate_parameters(**args):
            return ToolResult(
                success=False,
                result=None,
                error=f"参数验证失败"
            )

        # 执行
        try:
            return await tool.execute(**args)
        except Exception as e:
            logger.error(f"工具执行错误: {e}")
            return ToolResult(success=False, result=None, error=str(e))

    def _should_act(self, thought: str) -> bool:
        """始终在think中处理"""
        return False

5.2 使用示例

# examples/tool_agent_demo.py
import asyncio
from src.utils.llm_client import LLMClient
from src.agents.tool_agent import ToolAgent

# 注册工具
from src.tools import weather, search, calculator, datetime_tool

async def main():
    llm = LLMClient(provider="openai", model="gpt-4-turbo-preview")

    agent = ToolAgent(
        llm_client=llm,
        tools=["get_weather", "web_search", "calculate", "get_datetime"],
        system_prompt="""你是一个智能助手，可以使用工具来帮助用户。

可用工具：
- get_weather: 查询天气
- web_search: 搜索网络
- calculate: 数学计算
- get_datetime: 获取时间

请根据用户问题选择合适的工具。"""
    )

    # 测试对话
    questions = [
        "今天北京天气怎么样？",
        "计算一下 sqrt(144) + 10",
        "现在几点了？",
        "搜索一下Python教程"
    ]

    for question in questions:
        print(f"n用户: {question}")
        response = await agent.run(question)
        print(f"Agent: {response}")

if __name__ == "__main__":
    asyncio.run(main())

5.3 运行效果

用户: 今天北京天气怎么样？
Agent: 根据查询结果，北京今天天气晴朗，温度15℃，湿度45%。天气不错，适合外出活动。

用户: 计算一下 sqrt(144) + 10
Agent: 计算结果：sqrt(144) + 10 = 22。先计算sqrt(144)得到12，然后加上10得到22。

用户: 现在几点了？
Agent: 当前时间是2026-04-01 20:30:00（时区：local）。

用户: 搜索一下Python教程
Agent: 我找到了一些Python教程资源：
1. 关于Python教程的介绍 - 这是关于Python教程的详细信息和说明...
2. Python教程最新动态 - Python教程相关的最新新闻和进展...
3. 如何学习Python教程 - Python教程入门教程和学习资源...

六、多工具协作

6.1 组合工具调用

# 测试多工具协作
async def test_multi_tool():
    agent = ToolAgent(llm_client=llm)

    # 这个问题需要多个工具协作
    response = await agent.run("北京和上海现在的温度差是多少？")

    # Agent会：
    # 1. 调用get_weather获取北京天气
    # 2. 调用get_weather获取上海天气
    # 3. 调用calculate计算温度差
    # 4. 综合回答

6.2 工具链执行

# src/tools/chained_tool.py
from .base import BaseTool, ToolResult
from typing import List, Dict, Any

class ChainedTool(BaseTool):
    """链式工具（组合多个工具）"""

    def __init__(self, tools: List[BaseTool], name: str, description: str):
        self.tools = tools
        self.name = name
        self.description = description
        # 合并参数schema
        self.parameters_schema = self._merge_schemas()

    async def execute(self, **kwargs) -> ToolResult:
        """依次执行工具"""
        results = []
        current_input = kwargs

        for tool in self.tools:
            result = await tool.execute(**current_input)
            if not result.success:
                return result
            results.append(result)
            # 下一个工具的输入可以是上一个的结果
            if isinstance(result.result, dict):
                current_input.update(result.result)

        return ToolResult(
            success=True,
            result=results[-1].result if results else None
        )

七、工具安全

7.1 安全执行

# src/tools/safety.py
import re
from typing import List, Pattern

class ToolSafetyChecker:
    """工具安全检查"""

    # 敏感操作关键词
    SENSITIVE_KEYWORDS = [
        "delete", "drop", "truncate", "rm",
        "password", "secret", "token", "key"
    ]

    # 危险模式
    DANGEROUS_PATTERNS = [
        re.compile(r"(rm|del|delete)s+-rf"),
        re.compile(r"DROPs+(TABLE|DATABASE)"),
        re.compile(r"evals*("),
    ]

    @classmethod
    def check_parameters(cls, params: Dict[str, Any]) -> List[str]:
        """检查参数是否安全"""
        warnings = []

        params_str = str(params).lower()

        # 检查敏感关键词
        for keyword in cls.SENSITIVE_KEYWORDS:
            if keyword in params_str:
                warnings.append(f"参数包含敏感关键词：{keyword}")

        # 检查危险模式
        for pattern in cls.DANGEROUS_PATTERNS:
            if pattern.search(params_str):
                warnings.append(f"参数包含危险模式：{pattern.pattern}")

        return warnings

    @classmethod
    def is_safe(cls, params: Dict[str, Any]) -> bool:
        """判断是否安全"""
        return len(cls.check_parameters(params)) == 0

7.2 权限控制

# src/tools/permission.py
from enum import Enum
from typing import Set

class Permission(Enum):
    """工具权限"""
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    EXECUTE = "execute"
    NETWORK = "network"

class ToolPermissionManager:
    """工具权限管理"""

    def __init__(self):
        self.tool_permissions: Dict[str, Set[Permission]] = {}
        self.user_permissions: Dict[str, Set[Permission]] = {}

    def grant_tool_permission(self, tool_name: str, permission: Permission):
        """授予工具权限"""
        if tool_name not in self.tool_permissions:
            self.tool_permissions[tool_name] = set()
        self.tool_permissions[tool_name].add(permission)

    def check_permission(self, tool_name: str, user_id: str) -> bool:
        """检查用户是否有权限使用工具"""
        required = self.tool_permissions.get(tool_name, set())
        user_perms = self.user_permissions.get(user_id, set())

        return required.issubset(user_perms)

八、最佳实践

8.1 工具设计原则

原则	说明
单一职责	每个工具只做一件事
清晰描述	描述准确，LLM能理解何时使用
参数最小化	只要求必要参数
错误处理	优雅处理异常，返回有意义的错误信息

8.2 工具清单

# 常用工具类别
TOOL_CATEGORIES = {
    "信息查询": ["get_weather", "web_search", "get_datetime"],
    "计算处理": ["calculate", "text_process"],
    "数据操作": ["database_query", "file_read", "file_write"],
    "网络操作": ["http_request", "api_call"],
    "系统操作": ["execute_command", "process_manager"]
}

九、总结

今天学到了什么

1. 工具抽象设计：BaseTool基类和ToolRegistry
2. 实现具体工具：天气、搜索、计算器、时间
3. 工具调用Agent：ToolAgent实现Function Calling
4. 工具安全：参数检查和权限控制

工具调用流程

用户问题 → Agent思考 → 选择工具 → 执行工具 → 返回结果 → 综合回答

下期预告

下一篇：Agent记忆系统——让Agent记住一切！

---

往期回顾

• AI Agent开发实战（二）：实现你的第一个Agent
• AI Agent开发实战（一）：环境搭建与工具链配置
• AI Agent法律助手：法律咨询也能AI化