当前位置：首页 > news >正文

Heurist Agent Framework工具系统详解：如何扩展自定义工具

news 2026/7/20 16:37:44

Heurist Agent Framework工具系统详解：如何扩展自定义工具

【免费下载链接】heurist-agent-frameworkA flexible multi-interface AI agent framework for building agents with reasoning, tool use, memory, deep research, blockchain interaction, MCP, and agents-as-a-service.项目地址: https://gitcode.com/gh_mirrors/he/heurist-agent-framework

Heurist Agent Framework是一个强大的多接口AI智能体框架，它提供了灵活的工具系统，让开发者能够轻松构建具有推理、工具使用、记忆、深度研究和区块链交互能力的AI智能体。本文将深入探讨Heurist Agent Framework的工具系统，并为您展示如何扩展自定义工具，让您的智能体更加强大和实用。

什么是Heurist Agent Framework工具系统？

Heurist Agent Framework的工具系统是一个模块化的架构，允许开发者将各种功能封装为可重用的工具。这些工具可以被AI智能体调用，从而扩展其能力范围。工具系统采用了装饰器模式，让工具定义变得简单直观。

工具系统的核心优势在于其灵活性——您可以轻松添加新的工具，而无需修改框架的核心代码。每个工具都是一个独立的函数，通过@tool装饰器进行标记，框架会自动处理工具的注册、参数验证和执行。

工具系统核心组件

1. 工具装饰器（Tool Decorator）

工具装饰器是Heurist Agent Framework工具系统的核心组件，位于core/tools/tool_decorator.py。这个装饰器为普通函数添加了必要的元数据，使其能够被框架识别为可调用的工具。

@tool("工具描述") def 工具函数名(参数1: 类型, 参数2: 类型) -> 返回类型: """工具详细说明""" # 工具逻辑 return {"result": 结果}

装饰器会自动生成OpenAI兼容的函数模式，包括参数模式、类型验证和描述信息。这意味着您定义的任何工具都可以直接被大型语言模型理解和调用。

2. 工具箱（ToolBox）

工具箱是工具的容器和管理器，位于core/tools/tools.py。它负责：

工具的注册和管理
工具模式的生成
工具的执行调度
错误处理和日志记录

每个智能体都可以拥有自己的工具箱实例，您可以根据需要定制工具集合。

3. 工具处理器（Tool Handlers）

工具处理器负责实际执行工具调用。当AI智能体决定使用某个工具时，框架会自动调用相应的处理器函数，传递参数并返回结果。

如何创建自定义工具：完整指南

步骤1：创建工具函数

创建一个新的Python文件，定义您的工具函数。以下是一个简单的计算器工具示例：

from core.tools.tool_decorator import tool @tool("计算两个数字的和") def add(a: int, b: int) -> int: """将两个整数相加并返回结果""" result = a + b return {"result": result} @tool("计算两个数字的乘积") def multiply(a: int, b: int) -> int: """将两个整数相乘并返回结果""" result = a * b return {"result": result}

步骤2：注册工具到工具箱

创建一个自定义工具箱类，继承自ToolBox基类：

from core.tools.tools import ToolBox class MyCustomToolBox(ToolBox): def __init__(self): super().__init__() # 注册装饰的工具 self.decorated_tools = [ add, multiply, # 添加更多工具... ]

步骤3：集成到智能体

在您的智能体类中使用自定义工具箱：

from mesh.mesh_agent import MeshAgent from core.tools.tools import Tools class MyCustomAgent(MeshAgent): def __init__(self): super().__init__() # 初始化工具系统 self.tools = Tools(MyCustomToolBox) def get_tool_schemas(self): """返回工具模式列表""" return self.tools.get_tools_config() async def _handle_tool_logic(self, tool_name, args): """处理工具调用""" return await self.tools.execute_tool(tool_name, args, self)

步骤4：添加高级功能

您可以为工具添加更复杂的功能，比如访问外部API、处理文件、或者与其他系统集成：

import requests from core.tools.tool_decorator import tool @tool("获取天气信息") def get_weather(city: str, country: str = "CN") -> dict: """获取指定城市的天气信息 Args: city: 城市名称 country: 国家代码（默认：CN） Returns: 包含天气信息的字典 """ # 调用天气API response = requests.get( f"https://api.weatherapi.com/v1/current.json", params={"q": f"{city},{country}", "key": os.getenv("WEATHER_API_KEY")} ) if response.status_code == 200: data = response.json() return { "result": f"{city}的天气：{data['current']['condition']['text']}，温度：{data['current']['temp_c']}°C", "data": data } else: return {"result": f"无法获取{city}的天气信息", "error": response.text}

实战案例：创建区块链数据查询工具

让我们创建一个实际的区块链数据查询工具，展示Heurist Agent Framework在区块链领域的强大能力：

import os from typing import Dict, Any from core.tools.tool_decorator import tool @tool("查询以太坊地址余额") def get_eth_balance(address: str) -> Dict[str, Any]: """查询指定以太坊地址的ETH余额 Args: address: 以太坊地址（0x开头） Returns: 包含余额信息的字典 """ from web3 import Web3 # 使用环境变量配置的RPC节点 w3 = Web3(Web3.HTTPProvider(os.getenv("ETH_RPC_URL"))) if not w3.is_connected(): return {"result": "无法连接到以太坊节点", "error": "连接失败"} # 验证地址格式 if not w3.is_address(address): return {"result": "无效的以太坊地址", "error": "地址格式错误"} # 查询余额 balance_wei = w3.eth.get_balance(address) balance_eth = w3.from_wei(balance_wei, 'ether') return { "result": f"地址 {address} 的ETH余额为：{balance_eth:.4f} ETH", "data": { "address": address, "balance_wei": str(balance_wei), "balance_eth": float(balance_eth), "unit": "ETH" } } @tool("查询代币持有量") def get_token_balance(address: str, token_address: str) -> Dict[str, Any]: """查询指定地址的代币持有量 Args: address: 用户地址 token_address: 代币合约地址 Returns: 包含代币余额信息的字典 """ from web3 import Web3 w3 = Web3(Web3.HTTPProvider(os.getenv("ETH_RPC_URL"))) # ERC20代币ABI（简化版） erc20_abi = [ { "constant": True, "inputs": [{"name": "_owner", "type": "address"}], "name": "balanceOf", "outputs": [{"name": "balance", "type": "uint256"}], "type": "function" }, { "constant": True, "inputs": [], "name": "decimals", "outputs": [{"name": "", "type": "uint8"}], "type": "function" }, { "constant": True, "inputs": [], "name": "symbol", "outputs": [{"name": "", "type": "string"}], "type": "function" } ] # 创建代币合约实例 token_contract = w3.eth.contract(address=token_address, abi=erc20_abi) # 查询代币信息 try: balance = token_contract.functions.balanceOf(address).call() decimals = token_contract.functions.decimals().call() symbol = token_contract.functions.symbol().call() # 计算实际代币数量 actual_balance = balance / (10 ** decimals) return { "result": f"地址 {address} 持有 {actual_balance:.4f} {symbol}", "data": { "address": address, "token_address": token_address, "balance_raw": str(balance), "balance_formatted": float(actual_balance), "decimals": decimals, "symbol": symbol } } except Exception as e: return {"result": f"查询代币余额失败：{str(e)}", "error": str(e)}

工具系统的高级特性

1. 工具链式调用

Heurist Agent Framework支持工具链式调用，一个工具的输出可以作为另一个工具的输入：

@tool("分析地址的资产组合") async def analyze_wallet_portfolio(address: str, agent_context) -> Dict[str, Any]: """分析以太坊地址的完整资产组合 Args: address: 以太坊地址 agent_context: 智能体上下文（自动注入） Returns: 包含资产分析结果的字典 """ # 获取ETH余额 eth_result = await get_eth_balance({"address": address}, agent_context) # 获取常见代币余额 common_tokens = { "USDT": "0xdAC17F958D2ee523a2206206994597C13D831ec7", "USDC": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", "DAI": "0x6B175474E89094C44Da98b954EedeAC495271d0F" } token_balances = {} for symbol, token_address in common_tokens.items(): token_result = await get_token_balance( {"address": address, "token_address": token_address}, agent_context ) if "data" in token_result: token_balances[symbol] = token_result["data"] # 生成分析报告 total_value = 0 if "data" in eth_result: total_value += eth_result["data"].get("balance_eth", 0) for symbol, data in token_balances.items(): total_value += data.get("balance_formatted", 0) return { "result": f"地址 {address} 的总资产价值约为 ${total_value:.2f}", "data": { "eth_balance": eth_result.get("data"), "token_balances": token_balances, "total_value_usd": total_value, "analysis_timestamp": datetime.now().isoformat() } }

2. 工具缓存机制

框架内置了缓存装饰器，可以显著提升工具性能：

from decorators import with_cache @tool("获取代币价格信息") @with_cache(ttl_seconds=300) # 缓存5分钟 async def get_token_price(token_symbol: str) -> Dict[str, Any]: """获取代币的当前价格信息 Args: token_symbol: 代币符号（如：ETH、BTC、USDT） Returns: 包含价格信息的字典 """ # 这里调用价格API # 由于有缓存，相同的请求在5分钟内不会重复调用API pass

3. 错误处理和重试

工具系统内置了错误处理和重试机制：

from decorators import with_retry @tool("调用外部API") @with_retry(max_retries=3, delay=1) async def call_external_api(api_url: str, params: dict) -> Dict[str, Any]: """调用外部API，自动重试失败请求 Args: api_url: API地址 params: 请求参数 Returns: API响应数据 """ import aiohttp async with aiohttp.ClientSession() as session: async with session.get(api_url, params=params) as response: if response.status == 200: data = await response.json() return {"result": "API调用成功", "data": data} else: raise Exception(f"API调用失败，状态码：{response.status}")

最佳实践和注意事项

1. 工具设计原则

单一职责：每个工具只做一件事，并且做好
清晰的接口：使用类型注解明确参数和返回值类型
完善的文档：为每个工具提供详细的文档字符串
错误处理：妥善处理异常情况，返回有意义的错误信息
性能优化：对于耗时的操作，考虑添加缓存机制

2. 安全性考虑

API密钥管理：永远不要硬编码API密钥，使用环境变量
输入验证：验证所有输入参数，防止注入攻击
访问控制：根据需要对工具访问进行权限控制
日志记录：记录重要的工具调用信息

3. 测试和验证

为您的工具编写测试用例：

import pytest from my_tools import get_eth_balance def test_get_eth_balance(): """测试以太坊余额查询工具""" result = get_eth_balance({"address": "0x742d35Cc6634C0532925a3b844Bc9e...省略"}) assert "result" in result assert "data" in result assert "balance_eth" in result["data"]