智能体核心技术七大模块之（三）

工具使用与API调用——让智能体拥有执行能力

本文是七大模块的第三部分，深入探讨智能体如何定义、注册、调用和管理外部工具与API。工具模块是智能体从“对话”走向“执行”的关键桥梁。我们将通过UML建模、详细设计和完整代码实现，构建一个健壮、可扩展的工具使用框架，支持参数验证、错误处理、异步调用，并与LLM的function calling机制无缝集成。

---

1. 核心概念与设计目标

1.1 什么是工具使用

工具使用是指智能体通过调用预定义的函数或API来获取信息、执行操作的能力。这些工具可以包括：

• 本地函数：如计算器、代码执行器。
• 远程API：如天气查询、数据库查询、邮件发送。
• 知识库：如向量数据库检索。
• 其他智能体：子智能体协作。

1.2 设计目标

• 通用性：支持任意类型的工具，统一调用接口。
• 安全性：对工具参数进行验证，防止注入攻击。
• 可扩展性：轻松添加新工具，无需修改核心代码。
• 容错性：妥善处理超时、错误、重试。
• 与LLM集成：适配OpenAI function calling等标准。
• 可观测性：记录工具调用日志、耗时、结果。

---

2. 系统架构与UML建模

2.1 核心类设计

2.2 工具调用时序图

---

3. 详细设计

3.1 工具定义规范

每个工具需要提供：

• name：唯一标识符。
• description：描述工具功能，供LLM理解。
• parameters：JSON Schema格式的参数描述，用于验证和LLM生成参数。

例如，天气工具的parameters：

{
  "type": "object",
  "properties": {
    "city": {
      "type": "string",
      "description": "城市名称，如北京、上海"
    },
    "unit": {
      "type": "string",
      "enum": ["celsius", "fahrenheit"],
      "default": "celsius"
    }
  },
  "required": ["city"]
}

3.2 工具注册与发现

ToolRegistry负责管理所有工具，并提供以下功能：

• 注册工具（register）
• 按名称获取工具（get）
• 列出所有工具（list_tools）
• 生成用于OpenAI function calling的格式（get_function_definitions）

3.3 参数验证

每个工具可重写validate方法，默认实现基于JSON Schema验证（使用jsonschema库）。也可自定义复杂验证逻辑。

3.4 工具执行与结果封装

ToolInvoker负责：

• 获取工具实例。
• 参数验证。
• 执行工具（捕获异常、记录耗时）。
• 返回标准化的ToolResult对象。

ToolResult包含：

• success：布尔值，是否成功。
• result：成功时的返回值。
• error：失败时的错误信息。
• execution_time：执行耗时（秒）。
• metadata：附加信息（如API调用ID）。

3.5 与LLM Function Calling集成

当使用OpenAI等支持function calling的模型时，我们需要将工具列表转换为模型所需的格式。ToolRegistry.get_function_definitions()返回如下列表：

[
    {
        "type": "function",
        "function": {
            "name": tool.name,
            "description": tool.description,
            "parameters": tool.parameters
        }
    }
    for tool in self.list_tools()
]

在推理模块中，可以将此列表传递给LLM，并处理返回的function_call。

3.6 错误处理与重试

ToolInvoker可实现简单的重试机制（针对网络错误等）。我们将在invoke_with_safety方法中实现最多3次重试，指数退避。

3.7 异步支持

对于耗时工具（如调用外部API），可提供异步版本async_execute，但为简化，本文暂不实现，但框架留有扩展接口。

---

4. 项目文件结构

在原有项目基础上，新增/修改以下文件：

agent_core/
├── agent/
│   ├── core/
│   │   ├── __init__.py
│   │   ├── tool.py              # 增强：Tool基类增加parameters、validate
│   │   ├── tool_registry.py     # 新增：工具注册表
│   │   ├── tool_invoker.py      # 新增：工具调用器
│   │   ├── executor.py          # 修改：使用ToolInvoker
│   │   ├── models.py            # 新增ToolResult
│   │   └── ... (其他不变)
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── calculator.py        # 修改：继承新Tool，定义parameters
│   │   ├── weather.py           # 修改：同上，并实现真实API调用
│   │   ├── email.py             # 修改
│   │   └── database.py          # 新增示例：数据库查询工具
│   ├── llm/... (不变)
│   ├── nlu/... (不变)
│   ├── planners/... (不变)
│   └── utils/... (不变)
├── examples/
│   └── tool_demo.py             # 新示例，展示工具使用
└── requirements.txt             # 添加依赖：requests, jsonschema

---

5. 源代码完整实现

5.1 依赖安装

pip install requests jsonschema

5.2 工具结果模型 (agent/core/models.py)

添加ToolResult类：

from dataclasses import dataclass
from typing import Any, Optional

@dataclass
class ToolResult:
    """工具执行结果"""
    success: bool
    result: Any = None
    error: Optional[str] = None
    execution_time: float = 0.0
    metadata: dict = None

5.3 增强的工具基类 (agent/core/tool.py)

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
import jsonschema
from .models import ToolResult

class Tool(ABC):
    @property
    @abstractmethod
    def name(self) -> str:
        pass

    @property
    @abstractmethod
    def description(self) -> str:
        pass

    @property
    def parameters(self) -> dict:
        """返回JSON Schema格式的参数描述，默认空对象"""
        return {
            "type": "object",
            "properties": {},
            "required": []
        }

    def validate(self, params: Dict[str, Any]) -> bool:
        """使用JSON Schema验证参数，可被子类重写"""
        try:
            jsonschema.validate(instance=params, schema=self.parameters)
            return True
        except jsonschema.ValidationError:
            return False

    @abstractmethod
    def execute(self, params: Dict[str, Any]) -> ToolResult:
        """执行工具，返回ToolResult"""
        pass

    # 可选异步版本
    # async def async_execute(self, params): ...

5.4 工具注册表 (agent/core/tool_registry.py)

from typing import Dict, List, Optional
from .tool import Tool

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

    def register(self, tool: Tool) -> None:
        if tool.name in self._tools:
            raise ValueError(f"Tool '{tool.name}' already registered")
        self._tools[tool.name] = tool

    def get(self, name: str) -> Optional[Tool]:
        return self._tools.get(name)

    def list_tools(self) -> List[Tool]:
        return list(self._tools.values())

    def get_function_definitions(self) -> List[dict]:
        """返回符合OpenAI function calling格式的工具定义列表"""
        definitions = []
        for tool in self.list_tools():
            definitions.append({
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters
                }
            })
        return definitions

5.5 工具调用器 (agent/core/tool_invoker.py)

import time
import logging
from typing import Dict, Any
from .tool_registry import ToolRegistry
from .models import ToolResult

logger = logging.getLogger(__name__)

class ToolInvoker:
    def __init__(self, registry: ToolRegistry, max_retries: int = 3):
        self.registry = registry
        self.max_retries = max_retries

    def invoke(self, tool_name: str, params: Dict[str, Any]) -> ToolResult:
        """调用工具，不重试"""
        tool = self.registry.get(tool_name)
        if not tool:
            return ToolResult(success=False, error=f"Tool '{tool_name}' not found")

        # 参数验证
        if not tool.validate(params):
            return ToolResult(success=False, error=f"Parameter validation failed for tool '{tool_name}'")

        # 执行并计时
        start = time.time()
        try:
            result = tool.execute(params)
            elapsed = time.time() - start
            result.execution_time = elapsed
            return result
        except Exception as e:
            elapsed = time.time() - start
            logger.exception(f"Tool '{tool_name}' execution failed")
            return ToolResult(success=False, error=str(e), execution_time=elapsed)

    def invoke_with_safety(self, tool_name: str, params: Dict[str, Any]) -> ToolResult:
        """带重试的调用"""
        for attempt in range(self.max_retries):
            result = self.invoke(tool_name, params)
            if result.success:
                return result
            # 如果是可重试的错误（如网络超时），则等待后重试
            if attempt < self.max_retries - 1:
                wait = 2 ** attempt  # 指数退避
                time.sleep(wait)
        return result  # 最后一次结果

5.6 修改执行器 (agent/core/executor.py)

之前的Executor直接调用工具，现在改为使用ToolInvoker。

from typing import Dict
from .models import Action, Observation
from .tool_invoker import ToolInvoker

class Executor:
    def __init__(self, invoker: ToolInvoker):
        self.invoker = invoker

    def execute(self, action: Action) -> Observation:
        if action.type != 'tool' or not action.tool:
            return Observation(tool="", result="", error="无工具调用")
        result = self.invoker.invoke_with_safety(action.tool, action.tool_params or {})
        if result.success:
            return Observation(tool=action.tool, result=str(result.result))
        else:
            return Observation(tool=action.tool, result="", error=result.error)

注意：Observation目前只有result字段，我们暂时将工具结果转换为字符串。更好的做法是让Observation支持结构化数据，但为保持兼容，先这样。

5.7 修改Agent以使用新组件

agent/core/agent.py需要调整：创建ToolRegistry和ToolInvoker，并传递给Executor。同时保留register_tool方法方便注册。

from .tool_registry import ToolRegistry
from .tool_invoker import ToolInvoker
from .executor import Executor
# ... 其他导入

class Agent:
    def __init__(self, nlu_engine, planner, llm):
        self.perception = Perception(nlu_engine)
        self.planner = planner
        self.reasoning = Reasoning(llm)
        self.tool_registry = ToolRegistry()
        self.tool_invoker = ToolInvoker(self.tool_registry)
        self.executor = Executor(self.tool_invoker)
        self.state = State()
        self.reflector = Reflector()
        self.max_iterations = 20

    def register_tool(self, tool: Tool):
        self.tool_registry.register(tool)

    # ... run方法保持不变（因为它使用self.executor）

5.8 重构现有工具

以WeatherTool为例，展示如何实现增强版。

agent/tools/weather.py

import requests
import random
from agent.core.tool import Tool
from agent.core.models import ToolResult

class WeatherTool(Tool):
    name = "weather"
    description = "查询指定城市的当前天气"

    @property
    def parameters(self):
        return {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，如北京、上海"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "default": "celsius",
                    "description": "温度单位"
                }
            },
            "required": ["city"]
        }

    def execute(self, params):
        city = params.get("city")
        unit = params.get("unit", "celsius")
        # 模拟API调用（实际可替换为真实天气API）
        # 这里用随机数据模拟
        weathers = ["晴", "多云", "阴", "小雨", "大雨"]
        temps_c = random.randint(15, 30)
        if unit == "fahrenheit":
            temps = temps_c * 9/5 + 32
            unit_str = "℉"
        else:
            temps = temps_c
            unit_str = "℃"
        result_str = f"{city}天气：{random.choice(weathers)}，温度：{temps}{unit_str}"
        return ToolResult(success=True, result=result_str)

agent/tools/calculator.py

import math
from agent.core.tool import Tool
from agent.core.models import ToolResult

class CalculatorTool(Tool):
    name = "calculator"
    description = "计算数学表达式，支持 + - * / 和函数 sqrt, pow"

    @property
    def parameters(self):
        return {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "数学表达式，例如 '2+2' 或 'sqrt(25)'"
                }
            },
            "required": ["expression"]
        }

    def execute(self, params):
        expr = params.get("expression", "")
        try:
            # 安全评估，允许使用math中的函数
            allowed_names = {k: v for k, v in math.__dict__.items() if not k.startswith("__")}
            allowed_names.update({"abs": abs, "round": round})
            result = eval(expr, {"__builtins__": {}}, allowed_names)
            return ToolResult(success=True, result=f"计算结果：{result}")
        except Exception as e:
            return ToolResult(success=False, error=f"计算错误：{str(e)}")

agent/tools/email.py（略作调整）

from agent.core.tool import Tool
from agent.core.models import ToolResult

class EmailTool(Tool):
    name = "email"
    description = "发送邮件或根据姓名获取邮箱地址"

    @property
    def parameters(self):
        return {
            "type": "object",
            "properties": {
                "action": {
                    "type": "string",
                    "enum": ["get_email", "send"],
                    "description": "操作类型：get_email（根据姓名获取邮箱）或 send（发送邮件）"
                },
                "name": {"type": "string", "description": "姓名，用于get_email"},
                "to": {"type": "string", "description": "收件人邮箱，用于send"},
                "content": {"type": "string", "description": "邮件内容，用于send"}
            },
            "required": ["action"]
        }

    contacts = {"张三": "zhangsan@example.com", "李四": "lisi@example.com"}

    def execute(self, params):
        action = params.get("action")
        if action == "get_email":
            name = params.get("name")
            if not name:
                return ToolResult(success=False, error="缺少name参数")
            email = self.contacts.get(name, f"{name}@unknown.com")
            return ToolResult(success=True, result=email)
        elif action == "send":
            to = params.get("to")
            content = params.get("content")
            if not to or not content:
                return ToolResult(success=False, error="缺少to或content参数")
            # 模拟发送
            return ToolResult(success=True, result=f"邮件已发送至 {to}，内容：{content}")
        else:
            return ToolResult(success=False, error=f"未知action: {action}")

新增工具示例：数据库查询 (agent/tools/database.py)

import sqlite3
from agent.core.tool import Tool
from agent.core.models import ToolResult

class DatabaseTool(Tool):
    name = "database"
    description = "执行SQL查询，返回结果"

    @property
    def parameters(self):
        return {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "SQL查询语句，如 'SELECT * FROM users'"
                }
            },
            "required": ["query"]
        }

    def __init__(self, db_path=":memory:"):
        self.db_path = db_path
        # 初始化示例数据库
        self._init_db()

    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)")
        cursor.execute("INSERT OR IGNORE INTO users (id, name, email) VALUES (1, '张三', 'zhangsan@example.com')")
        cursor.execute("INSERT OR IGNORE INTO users (id, name, email) VALUES (2, '李四', 'lisi@example.com')")
        conn.commit()
        conn.close()

    def execute(self, params):
        query = params.get("query")
        if not query:
            return ToolResult(success=False, error="缺少query参数")
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            cursor.execute(query)
            # 如果是SELECT，获取结果
            if query.strip().upper().startswith("SELECT"):
                rows = cursor.fetchall()
                columns = [description[0] for description in cursor.description]
                result = [dict(zip(columns, row)) for row in rows]
            else:
                conn.commit()
                result = f"受影响行数: {cursor.rowcount}"
            conn.close()
            return ToolResult(success=True, result=result)
        except Exception as e:
            return ToolResult(success=False, error=f"数据库错误: {str(e)}")

5.9 修改推理模块以支持function calling（可选）

如果需要利用OpenAI的function calling，可以在reasoning.py中修改decide方法，将工具定义传递给LLM，并解析返回的function_call。这里给出一个简单示例（假设使用OpenAI SDK）：

def decide(self, state_description, memory, tools):
    # tools 是工具字典，需要转换成function calling格式
    # 假设我们有tool_registry
    functions = self.tool_registry.get_function_definitions()
    messages = [{"role": "system", "content": state_description}]
    for msg in memory[-5:]:
        messages.append({"role": msg.role, "content": msg.content})
    
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages,
        functions=functions,
        function_call="auto"
    )
    message = response.choices[0].message
    if message.get("function_call"):
        fc = message.function_call
        return Action(type='tool', tool=fc.name, tool_params=json.loads(fc.arguments))
    else:
        return Action(type='final', content=message.content)

由于我们使用的是模拟LLM，暂不实现此部分，但框架已准备好。

5.10 运行示例 (examples/tool_demo.py)

import sys
sys.path.append("..")

from agent.core.agent import Agent
from agent.nlu.rule_based import RuleBasedNLU
from agent.planners.template_planner import TemplatePlanner
from agent.llm.mock import MockLLM
from agent.tools.calculator import CalculatorTool
from agent.tools.weather import WeatherTool
from agent.tools.email import EmailTool
from agent.tools.database import DatabaseTool

def main():
    nlu = RuleBasedNLU()
    planner = TemplatePlanner()
    llm = MockLLM()
    agent = Agent(nlu, planner, llm)
    
    # 注册工具
    agent.register_tool(CalculatorTool())
    agent.register_tool(WeatherTool())
    agent.register_tool(EmailTool())
    agent.register_tool(DatabaseTool())  # 新增数据库工具

    print("增强工具模块的智能体已启动")
    while True:
        user_input = input("\n用户: ")
        if user_input.lower() == 'quit':
            break
        response = agent.run(user_input)
        print(f"助手: {response}")

if __name__ == "__main__":
    main()

运行示例：

用户: 计算 2的平方根
助手: 计算结果：1.4142135623730951

用户: 查询北京天气
助手: 北京天气：多云，温度：25℃

用户: 查询数据库用户表
助手: [{"id": 1, "name": "张三", "email": "zhangsan@example.com"}, {"id": 2, "name": "李四", "email": "lisi@example.com"}]

注意：由于模拟LLM可能无法处理数据库查询，实际效果取决于LLM的生成。这里仅展示框架能力。

---

6. 总结与扩展

通过本文，我们为智能体构建了一个强大、灵活的工具使用模块：

• 定义了标准化的工具接口，支持参数描述和验证。
• 实现了工具注册表，集中管理所有工具。
• 设计了工具调用器，负责参数验证、执行、计时和重试。
• 与现有智能体框架无缝集成，只需少量修改。
• 提供了多种工具示例，包括本地计算、API调用、数据库查询。

未来扩展方向：

• 异步工具调用：使用asyncio支持并发工具调用。
• 工具调用缓存：对幂等工具缓存结果，减少重复调用。
• 工具组合：支持工具链或组合工具。
• 工具权限控制：为不同用户/场景限制可用工具。
• 更丰富的错误恢复：根据错误类型自动重试或通知规划器调整计划。

现在，你的智能体已经具备了调用真实世界工具的能力，向着“执行专家”又迈出了坚实一步！

---

附录：更新后的项目文件清单

agent_core/
├── agent/
│   ├── __init__.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── agent.py
│   │   ├── executor.py
│   │   ├── models.py
│   │   ├── perception.py
│   │   ├── planner.py
│   │   ├── plan.py
│   │   ├── reasoning.py
│   │   ├── reflector.py
│   │   ├── state.py
│   │   ├── tool.py               # 增强
│   │   ├── tool_registry.py       # 新增
│   │   └── tool_invoker.py        # 新增
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── calculator.py
│   │   ├── weather.py
│   │   ├── email.py
│   │   └── database.py            # 新增
│   ├── nlu/...
│   ├── planners/...
│   ├── llm/...
│   └── utils/...
├── examples/
│   └── tool_demo.py
├── requirements.txt                # 新增依赖
└── README.md

requirements.txt 内容：

requests>=2.25.0
jsonschema>=3.2.0

现在，你可以基于此框架继续构建更复杂的智能体应用，让AI真正成为你的“执行专家”！