智能体核心技术七大模块之（三）
工具使用与API调用——让智能体拥有执行能力
本文是七大模块的第三部分，深入探讨智能体如何定义、注册、调用和管理外部工具与API。工具模块是智能体从“对话”走向“执行”的关键桥梁。我们将通过UML建模、详细设计和完整代码实现，构建一个健壮、可扩展的工具使用框架，支持参数验证、错误处理、异步调用，并与LLM的function calling机制无缝集成。
1. 核心概念与设计目标
1.1 什么是工具使用
工具使用是指智能体通过调用预定义的函数或API来获取信息、执行操作的能力。这些工具可以包括：
- 本地函数：如计算器、代码执行器。
- 远程API：如天气查询、数据库查询、邮件发送。
- 知识库：如向量数据库检索。
- 其他智能体：子智能体协作。
1.2 设计目标
- 通用性：支持任意类型的工具，统一调用接口。
- 安全性：对工具参数进行验证，防止注入攻击。
- 可扩展性：轻松添加新工具，无需修改核心代码。
- 容错性：妥善处理超时、错误、重试。
- 与LLM集成：适配OpenAI function calling等标准。
- 可观测性：记录工具调用日志、耗时、结果。
2. 系统架构与UML建模
2.1 核心类设计
#mermaid-svg-iJY64itbXvcHI2Cx{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-iJY64itbXvcHI2Cx .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-iJY64itbXvcHI2Cx .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-iJY64itbXvcHI2Cx .error-icon{fill:#552222;}#mermaid-svg-iJY64itbXvcHI2Cx .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-iJY64itbXvcHI2Cx .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-iJY64itbXvcHI2Cx .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-iJY64itbXvcHI2Cx .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-iJY64itbXvcHI2Cx .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-iJY64itbXvcHI2Cx .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-iJY64itbXvcHI2Cx .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-iJY64itbXvcHI2Cx .marker{fill:#333333;stroke:#333333;}#mermaid-svg-iJY64itbXvcHI2Cx .marker.cross{stroke:#333333;}#mermaid-svg-iJY64itbXvcHI2Cx svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-iJY64itbXvcHI2Cx p{margin:0;}#mermaid-svg-iJY64itbXvcHI2Cx g.classGroup text{fill:#9370DB;stroke:none;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:10px;}#mermaid-svg-iJY64itbXvcHI2Cx g.classGroup text .title{font-weight:bolder;}#mermaid-svg-iJY64itbXvcHI2Cx .nodeLabel,#mermaid-svg-iJY64itbXvcHI2Cx .edgeLabel{color:#131300;}#mermaid-svg-iJY64itbXvcHI2Cx .edgeLabel .label rect{fill:#ECECFF;}#mermaid-svg-iJY64itbXvcHI2Cx .label text{fill:#131300;}#mermaid-svg-iJY64itbXvcHI2Cx .labelBkg{background:#ECECFF;}#mermaid-svg-iJY64itbXvcHI2Cx .edgeLabel .label span{background:#ECECFF;}#mermaid-svg-iJY64itbXvcHI2Cx .classTitle{font-weight:bolder;}#mermaid-svg-iJY64itbXvcHI2Cx .node rect,#mermaid-svg-iJY64itbXvcHI2Cx .node circle,#mermaid-svg-iJY64itbXvcHI2Cx .node ellipse,#mermaid-svg-iJY64itbXvcHI2Cx .node polygon,#mermaid-svg-iJY64itbXvcHI2Cx .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-iJY64itbXvcHI2Cx .divider{stroke:#9370DB;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx g.clickable{cursor:pointer;}#mermaid-svg-iJY64itbXvcHI2Cx g.classGroup rect{fill:#ECECFF;stroke:#9370DB;}#mermaid-svg-iJY64itbXvcHI2Cx g.classGroup line{stroke:#9370DB;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx .classLabel .box{stroke:none;stroke-width:0;fill:#ECECFF;opacity:0.5;}#mermaid-svg-iJY64itbXvcHI2Cx .classLabel .label{fill:#9370DB;font-size:10px;}#mermaid-svg-iJY64itbXvcHI2Cx .relation{stroke:#333333;stroke-width:1;fill:none;}#mermaid-svg-iJY64itbXvcHI2Cx .dashed-line{stroke-dasharray:3;}#mermaid-svg-iJY64itbXvcHI2Cx .dotted-line{stroke-dasharray:1 2;}#mermaid-svg-iJY64itbXvcHI2Cx #compositionStart,#mermaid-svg-iJY64itbXvcHI2Cx .composition{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx #compositionEnd,#mermaid-svg-iJY64itbXvcHI2Cx .composition{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx #dependencyStart,#mermaid-svg-iJY64itbXvcHI2Cx .dependency{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx #dependencyStart,#mermaid-svg-iJY64itbXvcHI2Cx .dependency{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx #extensionStart,#mermaid-svg-iJY64itbXvcHI2Cx .extension{fill:transparent!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx #extensionEnd,#mermaid-svg-iJY64itbXvcHI2Cx .extension{fill:transparent!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx #aggregationStart,#mermaid-svg-iJY64itbXvcHI2Cx .aggregation{fill:transparent!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx #aggregationEnd,#mermaid-svg-iJY64itbXvcHI2Cx .aggregation{fill:transparent!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx #lollipopStart,#mermaid-svg-iJY64itbXvcHI2Cx .lollipop{fill:#ECECFF!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx #lollipopEnd,#mermaid-svg-iJY64itbXvcHI2Cx .lollipop{fill:#ECECFF!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-iJY64itbXvcHI2Cx .edgeTerminals{font-size:11px;line-height:initial;}#mermaid-svg-iJY64itbXvcHI2Cx .classTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-iJY64itbXvcHI2Cx .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-iJY64itbXvcHI2Cx .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-iJY64itbXvcHI2Cx :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}
«abstract»
Tool
+name: str
+description: str
+parameters: dict # JSON Schema
+execute(params: dict) : -> ToolResult
+validate(params: dict) : -> bool
ToolResult
+success: bool
+result: any
+error: str
+execution_time: float
+metadata: dict
ToolRegistry
-tools: dict[str, Tool]
+register(tool: Tool)
+get(name: str) : -> Tool
+list_tools() : -> list[Tool]
+get_function_definitions() : -> list[dict] # 用于function calling
ToolInvoker
-registry: ToolRegistry
+invoke(tool_name: str, params: dict) : -> ToolResult
+invoke_with_safety(tool_name: str, params: dict) : -> ToolResult
Executor
-invoker: ToolInvoker
+execute(action: Action) : -> Observation
Agent
-tool_registry: ToolRegistry
-tool_invoker: ToolInvoker
-executor: Executor
+register_tool(tool)
+run(user_input)
CalculatorTool
WeatherTool
EmailTool
DatabaseTool
2.2 工具调用时序图
ExternalAPIToolToolRegistryToolInvokerExecutorReasoningExternalAPIToolToolRegistryToolInvokerExecutorReasoning#mermaid-svg-aD71vHVpfbW49nkt{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-aD71vHVpfbW49nkt .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-aD71vHVpfbW49nkt .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-aD71vHVpfbW49nkt .error-icon{fill:#552222;}#mermaid-svg-aD71vHVpfbW49nkt .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-aD71vHVpfbW49nkt .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-aD71vHVpfbW49nkt .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-aD71vHVpfbW49nkt .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-aD71vHVpfbW49nkt .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-aD71vHVpfbW49nkt .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-aD71vHVpfbW49nkt .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-aD71vHVpfbW49nkt .marker{fill:#333333;stroke:#333333;}#mermaid-svg-aD71vHVpfbW49nkt .marker.cross{stroke:#333333;}#mermaid-svg-aD71vHVpfbW49nkt svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-aD71vHVpfbW49nkt p{margin:0;}#mermaid-svg-aD71vHVpfbW49nkt .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-aD71vHVpfbW49nkt text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-aD71vHVpfbW49nkt .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-aD71vHVpfbW49nkt .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-aD71vHVpfbW49nkt .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-aD71vHVpfbW49nkt .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-aD71vHVpfbW49nkt #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-aD71vHVpfbW49nkt .sequenceNumber{fill:white;}#mermaid-svg-aD71vHVpfbW49nkt #sequencenumber{fill:#333;}#mermaid-svg-aD71vHVpfbW49nkt #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-aD71vHVpfbW49nkt .messageText{fill:#333;stroke:none;}#mermaid-svg-aD71vHVpfbW49nkt .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-aD71vHVpfbW49nkt .labelText,#mermaid-svg-aD71vHVpfbW49nkt .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-aD71vHVpfbW49nkt .loopText,#mermaid-svg-aD71vHVpfbW49nkt .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-aD71vHVpfbW49nkt .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-aD71vHVpfbW49nkt .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-aD71vHVpfbW49nkt .noteText,#mermaid-svg-aD71vHVpfbW49nkt .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-aD71vHVpfbW49nkt .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-aD71vHVpfbW49nkt .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-aD71vHVpfbW49nkt .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-aD71vHVpfbW49nkt .actorPopupMenu{position:absolute;}#mermaid-svg-aD71vHVpfbW49nkt .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-aD71vHVpfbW49nkt .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-aD71vHVpfbW49nkt .actor-man circle,#mermaid-svg-aD71vHVpfbW49nkt line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-aD71vHVpfbW49nkt :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}alt[验证失败][验证成功]Action(tool="weather", params={"city":"北京"})invoke("weather", {"city":"北京"})get("weather")WeatherTool实例参数验证ToolResult(success=False, error="参数错误")execute({"city":"北京"})调用真实API（如HTTP请求）返回数据处理后的结果ToolResult(success=True, result="北京天气…")Observation(工具返回结果)
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真正成为你的“执行专家”！
