企业官网建设流程全解析

• 人工智能
• AI Agent
• Agent 编排
• 低代码
• AI 技能
• RAG

本篇技术指南以 Nexent 工具开发规范 为骨架，结合 Nexent SDK 源码（sdk/nexent/core/tools/）与测试用例展开，系统讲解 Nexent 智能体工具体系的分类、共性架构、命名规范、代码模板与开发流程。读完本文，你将掌握在 Nexent 中从零开发一个风格一致、可维护、可扩展的 Agent 工具所需的完整规范与底层原理，并能直接参照仓库中的真实实现（如 ExaSearchTool、KnowledgeBaseSearchTool、StoreMemoryTool）落地自己的工具。

工具全景：Nexent SDK 内置工具分类

Nexent SDK 的工具体系覆盖了智能体运行所需的主要能力域。按功能划分，当前 SDK 包含以下九大类工具：

搜索工具

• ExaSearchTool：基于 EXA API 的网络搜索工具，适合实时信息更新、广泛话题与通用知识查询；
• TavilySearchTool：基于 Tavily API 的网络搜索工具；
• LinkupSearchTool：基于 Linkup API 的搜索工具；
• KnowledgeBaseSearchTool：本地知识库搜索工具，用于检索领域知识、文档等非公开信息；
• RAGFlowSearchTool：RAGFlow 知识检索；
• HaotianSearchTool / IndependentAidpSearchTool / AidpSearchTool / DifySearchTool / DataMateSearchTool / IdataSearchTool：企业级检索源接入，覆盖浩天、AIDP（含独立部署）、Dify、DataMate、iData 等外部知识平台。

以 ExaSearchTool 为例，其真实实现位于 exa_search_tool.py，通过exa_py客户端调用search_and_contents接口，并支持text.max_characters、livecrawl与图片链接抓取等高级参数。

文件管理工具

• CreateFileTool：创建包含内容的新文件（支持自定义编码，默认 utf-8，自动创建父目录，路径必须是工作区相对路径）；
• ReadFileTool：从文件系统读取文件内容；
• DeleteFileTool：从文件系统删除文件；
• MoveItemTool：移动或重命名文件和目录；
• CreateDirectoryTool：创建新目录；
• DeleteDirectoryTool：删除目录及其内容；
• ListDirectoryTool：列出目录内容和详细信息。

系统工具

• TerminalTool：执行 shell 命令和系统操作。其真实实现位于 terminal_tool.py，通过 SSH（paramiko）连接远程终端执行命令，支持session_name会话复用以跨命令保持 shell 状态，默认超时 30 秒。

通信工具

• GetEmailTool：通过 IMAP 的邮件获取工具；
• SendEmailTool：通过 SMTP 的邮件发送工具。

多模态工具

• AnalyzeTextFileTool：基于数据处理和大语言模型的文档问答工具；
• AnalyzeImageTool：基于视觉语言模型（VLM）的图片问答工具，支持 S3（s3://bucket/key、/bucket/key）、HTTP 与 HTTPS 三种图片来源；
• AnalyzeAudioTool：基于语音模型的音频理解工具；
• AnalyzeVideoTool：基于视觉语言模型的视频理解工具。

在 analyze_image_tool.py 中可以看到，多模态工具通过gateway.modality的VLMRequest与 Nexent 多模态网关交互，提示词模板则存放在sdk/nexent/core/prompts/analyze_image_zh.yaml等双语 YAML 中。

记忆工具

• StoreMemoryTool：将重要信息存入当前 Agent 短期记忆（MemoryLayer.AGENT+SHORT_TERM），每次运行最多存储 3 条（源码中max_stores_per_run = 3，见 store_memory_tool.py）。在新架构下，Agent 只能写入自己的 Agent 层短期记忆，子 Agent 与父 Agent 互不共享写入权限；
• SearchMemoryTool：按自然语言查询检索当前 Agent 记忆，top_k默认 5。当后端注入了MemoryContextService时，检索会路由到 Phase 4 检索管线（分数融合、时间衰减、MMR 去重、token 预算选择），见 search_memory_tool.py。

编排工具

• ParallelExecutorTool：并发执行多个子 Agent 或工具调用。该工具已注入 Python 环境，LLM 直接调用即可，禁止import；支持二元组/三元组任务格式、timeout（默认 120 秒/任务）与max_workers（默认 4），见 parallel_executor.py；
• CreatePlanTool / UpdatePlanStepTool：规划待办的创建与更新。这是 v1.4 规划特性：LLM 在代码块中调用工具完成计划注册与步骤状态更新，参数经 smolagents 的 Pydantic schema 结构化校验，见 plan_tools.py。

数据工具

• MySqlTool / PostgreSqlTool / MsSqlTool：对数据库执行受控 SQL 查询；
• UploadToS3Tool / DownloadFromS3Tool：S3 对象上传与下载。

技能与任务工具（系统内部管理）

• ReadSkillConfigTool / ReadSkillMdTool：读取技能目录下的config.yaml与SKILL.md；
• RunSkillScriptTool：执行技能脚本；
• WriteSkillFileTool：向租户隔离的技能目录写入文件；
• CreateScheduledTaskProposalTool：创建待用户确认的定时任务提案（仅保存提案，不执行业务任务）。

注：以上技能与任务类工具由系统内部管理，未在tools/init.py中统一导出。这一点可以从__init__.py的__all__清单直接验证——其中仅导出了搜索、文件、邮件、终端、多模态、记忆、并行、规划与数据库类工具，技能/定时任务类工具不在导出列表内。

工具共性特征

1. 基础架构

• 基类继承：所有工具必须继承自smolagents.tools.Tool（smolagents 是 Nexent 的底层 Agent 框架，Nexent 的CoreAgent也继承自其CodeAgent，见 core_agent.py）；
• 参数管理：使用pydantic.Field进行参数定义和验证；
• 流式输出：集成MessageObserver支持实时消息传递；
• 多语言支持：内置中英文双语提示信息（description/description_zh）。

2. 核心属性

每个工具类必须包含以下类属性：

class ToolExample(Tool): name = "tool_name" # 工具唯一标识符 description = "Tool function description" # 详细功能说明（英文） description_zh = "工具功能描述" # 详细功能说明（中文） inputs = { # 输入参数定义 "param": { "type": "string", "description": "Parameter description", "description_zh": "参数描述" } } init_param_descriptions = { # 构造参数说明（与 __init__ 参数对应） "config_param": {"description": "Config parameter description", "description_zh": "配置参数说明"} } output_type = "string" # 输出类型 category = "search" # 工具分类（ToolCategory 枚举值） tool_sign = "b" # 工具标识符（ToolSign 枚举值）

category与tool_sign均从 tools_common_message.py 中的ToolCategory/ToolSign枚举取值，而非直接硬编码字符串。ToolCategory的完整定义包括：SEARCH、FILE、EMAIL、TERMINAL、MULTIMODAL、DATABASE、MEMORY、SKILL、PLANNING。

3. 消息处理机制

• ProcessType 枚举：使用不同类型区分消息。在 observer.py 中定义的枚举非常丰富，与工具直接相关的主要有：TOOL（工具名）、CARD（前端卡片渲染内容）、SEARCH_CONTENT（工具搜索内容）、PICTURE_WEB（联网搜索后的图片）、MEMORY_SEARCH（记忆搜索状态）、PLAN/PLAN_STEP_UPDATE（规划 JSON 与步骤状态）等；
• Observer 模式：通过MessageObserver实现实时消息推送。MessageObserver.add_message()会经过与ProcessType一一对应的MessageTransformer转换后，以 JSON 格式（Message.to_json()）追加到统一输出队列，供 Agent 流式桥接层消费；
• JSON 格式：所有消息内容使用 JSON 格式确保一致性；
• 运行状态消息统一发送：ProcessType.TOOL运行状态消息由core_agent.py中的_wrap_tool_for_observer桥接器在每次调用工具时统一发送（并附带tool_call_id关联上下文）。该桥接器会包装工具的forward方法——调用前生成uuid4作为tool_call_id，发送 TOOL 事件与序列化后的tool_arguments，再通过observer.tool_call_context(tool_call_id)建立上下文关联后执行真实逻辑（见 core_agent.py）。因此工具内部不再自行发送 TOOL 消息，只需发送CARD、SEARCH_CONTENT、PICTURE_WEB等补充消息；
• 历史遗留：早期部分工具（如 SQL 工具基类 sql_database_base_tool.py）仍保留running_prompt_zh/en属性并在工具内发送 TOOL 消息，新工具不再采用该写法。

4. 异常处理策略

• 统一异常：使用Exception抛出错误信息；
• 错误日志：使用logging模块记录详细错误信息；
• 优雅降级：在可能的情况下提供备选方案。典型如 ExaSearchTool 的图片过滤失败时，会回退为发送未过滤的原始图片列表（见_filter_images的except分支）。

命名规范

文件命名

• 格式：{功能名}_tool.py
• 风格：小写字母，单词间用下划线连接
• 示例：exa_search_tool.py,knowledge_base_search_tool.py

类命名

• 格式：{功能名}Tool
• 风格：大驼峰命名法（PascalCase）
• 示例：ExaSearchTool,KnowledgeBaseSearchTool

属性和方法命名

• 格式：小写字母，单词间用下划线连接
• 私有方法：以单下划线开头（如_filter_images）
• 示例：max_results,description_zh,_decode_subject

工具标识符规范

• tool_sign：单字母标识符，用于区分工具来源（如在前端引用标注[b0]、[a1]中标识检索来源）；
• 取值方式：从nexent.core.utils.tools_common_message中的ToolSign枚举取值，例如tool_sign = ToolSign.EXA_SEARCH.value；
• 分配规则（摘自 ToolSign 枚举定义）：

该文件还提供了TOOL_SIGN_MAPPING反向映射表，用于兼容以工具名为 key 的旧式调用。tool_sign不仅用于前端引用标注，还会随SearchResultTextMessage.to_model_dict()一起进入模型侧——其生成的index为{tool_sign}{cite_index}（如b0、a1），reference_mark为[[b0]]形式，供大模型在总结中直接引用检索来源。

代码结构模板

基础模板

以下模板展示了 Nexent 工具的标准骨架，开发新工具时可直接复制改造：

import json import logging from typing import Optional from smolagents.tools import Tool from pydantic import Field from ..utils.observer import MessageObserver, ProcessType from ..utils.tools_common_message import ToolCategory, ToolSign logger = logging.getLogger("your_tool_name") class YourTool(Tool): name = "your_tool" description = "Detailed description of the tool function, including applicable scenarios and usage" description_zh = "工具功能的中文描述，说明适用场景和使用方法" inputs = { "param1": { "type": "string", "description": "Detailed description of parameter 1", "description_zh": "参数1的中文描述" }, "param2": { "type": "integer", "description": "Detailed description of parameter 2", "description_zh": "参数2的中文描述", "default": 10, "nullable": True } } init_param_descriptions = { # 构造参数说明（与 __init__ 参数对应） "config_param": {"description": "Config parameter description", "description_zh": "配置参数说明"}, "optional_param": {"description": "Optional parameter description", "description_zh": "可选参数说明"} } output_type = "string" category = ToolCategory.FILE.value # 按工具类型选择 ToolCategory 枚举成员 tool_sign = ToolSign.FILE_OPERATION.value # 按工具类型选择 ToolSign 枚举成员 def __init__( self, config_param: str = Field(description="Config parameter"), observer: MessageObserver = Field(description="Message observer", default=None, exclude=True), optional_param: int = Field(description="Optional parameter", default=5) ): super().__init__() self.config_param = config_param self.observer = observer self.optional_param = optional_param # 记录操作序号（如果需要） self.record_ops = 0 def forward(self, param1: str, param2: int = 10) -> str: """工具的主要执行方法 Args: param1: 参数1说明 param2: 参数2说明 Returns: JSON格式的字符串结果 Raises: Exception: 详细的错误信息 """ try: # 发送卡片信息（可选） if self.observer: card_content = [{"icon": "your_icon", "text": param1}] self.observer.add_message("", ProcessType.CARD, json.dumps(card_content, ensure_ascii=False)) # 主要业务逻辑 result = self._execute_main_logic(param1, param2) # 处理结果并返回 return self._format_result(result) except Exception as e: logger.error(f"Error in {self.name}: {str(e)}") raise Exception(f"执行{self.name}时发生错误: {str(e)}") def _execute_main_logic(self, param1: str, param2: int): """执行主要业务逻辑的私有方法""" # 实现具体的业务逻辑 pass def _format_result(self, result) -> str: """格式化返回结果""" formatted_result = { "status": "success", "data": result, "tool": self.name } return json.dumps(formatted_result, ensure_ascii=False)

模板要点解读：

• exclude=True的 observer 参数：MessageObserver等运行时注入依赖必须在Field(exclude=True)中声明，避免它们被当作 LLM 可见的工具参数；
• unwrap_field_info处理：对exclude=True的参数，smolagents 可能传入原始FieldInfo，真实实现中普遍用pydantic_utils.unwrap_field_info解包后再使用（参见 ExaSearchTool 的max_results处理）；
• 参数范围约束：可借助 pydantic Field 的ge/le直接限制取值范围，例如 ExaSearchTool 的max_results定义ge=1, le=100，KnowledgeBaseSearchTool 的top_k同样限定ge=1, le=100并在__init__中再次max(1, min(..., 100))兜底。

搜索工具模板

搜索类工具在基础模板之上增加了统一的检索结果格式化逻辑，通过SearchResultTextMessage数据类保证所有搜索工具向模型和前端输出一致的契约：

import json import logging from typing import List from smolagents.tools import Tool from pydantic import Field from ..utils.observer import MessageObserver, ProcessType from ..utils.tools_common_message import SearchResultTextMessage, ToolCategory, ToolSign logger = logging.getLogger("search_tool_name") class SearchTool(Tool): name = "search_tool" description = "Detailed description of the search tool, including search scope and applicable scenarios" description_zh = "搜索工具的中文描述，包括搜索范围和适用场景" inputs = { "query": {"type": "string", "description": "The search query to perform", "description_zh": "搜索查询词"}, "max_results": {"type": "integer", "description": "Maximum number of results", "description_zh": "最大结果数", "default": 5, "nullable": True} } init_param_descriptions = { # 构造参数说明（与 __init__ 参数对应） "api_key": {"description": "API key", "description_zh": "API密钥"} } output_type = "string" category = ToolCategory.SEARCH.value tool_sign = ToolSign.EXA_SEARCH.value # 按实际搜索源选择 ToolSign 枚举成员 def __init__( self, api_key: str = Field(description="API key"), observer: MessageObserver = Field(description="Message observer", default=None, exclude=True), max_results: int = Field(description="Maximum number of search results", default=5) ): super().__init__() self.api_key = api_key self.observer = observer self.max_results = max_results # 引用序号从 1 起始（供 SearchResultTextMessage 的 cite_index 使用） self.record_ops = 1 def forward(self, query: str, max_results: int = None) -> str: if max_results is None: max_results = self.max_results # 发送卡片信息（可选；TOOL 运行状态消息由 core_agent 桥接器统一发送） if self.observer: card_content = [{"icon": "search", "text": query}] self.observer.add_message("", ProcessType.CARD, json.dumps(card_content, ensure_ascii=False)) try: # 执行搜索 search_results = self._perform_search(query, max_results) if not search_results: raise Exception("未找到搜索结果！请尝试更短或更宽泛的查询。") # 格式化搜索结果 formatted_results = self._format_search_results(search_results) # 记录搜索内容 if self.observer: search_results_data = json.dumps(formatted_results["json"], ensure_ascii=False) self.observer.add_message("", ProcessType.SEARCH_CONTENT, search_results_data) return json.dumps(formatted_results["return"], ensure_ascii=False) except Exception as e: logger.error(f"搜索错误: {str(e)}") raise Exception(f"搜索失败: {str(e)}") def _perform_search(self, query: str, max_results: int): """执行实际的搜索操作""" # 实现具体的搜索逻辑 pass def _format_search_results(self, results): """格式化搜索结果为统一格式""" search_results_json = [] search_results_return = [] for index, result in enumerate(results): search_result_message = SearchResultTextMessage( title=result.get("title", ""), url=result.get("url", ""), text=result.get("content", ""), published_date=result.get("date", ""), source_type="url", filename="", score=result.get("score", ""), score_details=result.get("score_details", {}), cite_index=self.record_ops + index, search_type=self.name, tool_sign=self.tool_sign ) search_results_json.append(search_result_message.to_dict()) search_results_return.append(search_result_message.to_model_dict()) self.record_ops += len(search_results_return) return { "json": search_results_json, "return": search_results_return }

模板中的双路输出值得注意：to_dict()生成的完整字段（含 url、score、score_details、tool_sign 等）通过SEARCH_CONTENT消息推送给前端用于展示与引用溯源；to_model_dict()生成的轻量字段（title、text、index、reference_mark）作为返回字符串回传给大模型用于生成总结。真实实现可对照 exa_search_tool.py 的forward方法，其还额外演示了PICTURE_WEB图片结果推送与基于数据处理服务的异步图片过滤（asyncio.Semaphore(10)限流、2 秒超时、positive/negative prompt 判定）。

开发流程规范

1. 开发前准备

• 确定工具功能和适用场景；
• 选择合适的工具分类和标识符（对照上文ToolCategory/ToolSign分配表）；
• 检查是否与现有工具功能重复（可先浏览 tools 目录 与 导出清单）。

2. 实现步骤

1. 创建工具文件：按命名规范创建{name}_tool.py；
2. 定义类结构：继承Tool基类，定义必要属性（name、description、description_zh、inputs、init_param_descriptions、output_type、category、tool_sign）；
3. 实现构造函数：使用pydantic Field定义参数，运行依赖（observer、service 实例等）标记exclude=True；
4. 实现 forward 方法：核心功能逻辑，注意参数默认值兜底（如if max_results is None: max_results = self.max_results）；
5. 添加私有方法：将复杂逻辑拆分为私有方法（如_perform_search、_format_search_results、_filter_images）；
6. 集成消息观察者：支持流式输出和多语言；
7. 异常处理：完善的错误处理和日志记录。

3. 测试和集成

1. 单元测试：测试各种输入情况和边界条件。仓库中已有成熟范例，如 test_exa_search_tool.py 通过patch.dict('sys.modules', module_mocks)对exa_py、aiohttp等外部依赖做模块级 mock，覆盖正常搜索、空结果、图片过滤、observer 消息等场景；同目录下还有test_create_file_tool.py、test_parallel_executor.py、test_store_memory_tool.py等；
2. 集成测试：与 CoreAgent 集成测试（CoreAgent在 core_agent.py 中通过_wrap_tool_for_observer自动包装工具，集成时需验证 TOOL 事件与tool_call_id关联是否正确）；
3. 更新导出：在init.py 中添加工具导出（若属于对 LLM 暴露的常规工具）；
4. 文档更新：更新相关文档和示例。

最佳实践

1. 性能优化

• 异步处理：对于耗时操作使用异步处理。ExaSearchTool 的_filter_images用asyncio.gather并发处理图片 URL，并通过_run_coroutine工具（见 StoreMemoryTool）在线程管理器中安全执行协程，避免阻塞 Agent 主循环；
• 连接池：复用网络连接减少延迟；
• 缓存机制：适当使用缓存提升响应速度；
• 并发控制：使用Semaphore控制并发请求数（ExaSearchTool 图片过滤限流为 10 路并发）。

2. 安全性

• 输入验证：严格验证输入参数。典型如 SQL 工具基类 sql_database_base_tool.py 内置安全校验：通过HIGH_RISK_SQL_PATTERNS黑名单拦截DROP DATABASE、GRANT、REVOKE、CREATE USER、INTO OUTFILE、LOAD DATA INFILE、EXEC xp_等高风险操作；通过WHERE_REQUIRED_PATTERNS强制UPDATE/DELETE语句必须携带WHERE条件；
• 敏感信息：API 密钥等敏感信息不应出现在日志中；
• 错误信息：避免在错误信息中泄露敏感信息；
• 超时控制：设置合理的超时时间防止阻塞（TerminalTool 默认 30 秒、ParallelExecutorTool 单任务默认 120 秒、图片过滤请求 2 秒超时）。

3. 可维护性

• 模块化设计：将复杂功能拆分为多个方法；
• 清晰注释：为复杂逻辑添加详细注释；
• 类型注解：使用完整的类型注解；
• 文档字符串：为所有公共方法添加文档字符串（含 Args / Returns / Raises 说明）。

4. 用户体验

• 多语言支持：提供中英文双语提示（description+description_zh，且各输入参数同样提供双语描述）；
• 进度反馈：通过MessageObserver提供实时反馈（CARD 卡片、SEARCH_CONTENT 检索内容、PICTURE_WEB 图片结果等）；
• 错误提示：提供清晰的错误信息和解决建议（如"未找到搜索结果！请尝试更短或更宽泛的查询。"）；
• 参数验证：在执行前验证参数有效性。

注意事项

1. 版本兼容：确保工具与不同版本的依赖库兼容；
2. 资源清理：及时释放网络连接、文件句柄等资源；
3. 日志级别：合理设置日志级别，避免过多调试信息；
4. 配置管理：所有配置通过构造函数参数传入（SDK 内不直接读取环境变量，环境变量由服务层读取后传入）。这一点与 README 中的旧描述不同，是当前 SDK 的硬性约定——即使 ExaSearchTool 内部存在os.getenv("DATA_PROCESS_SERVICE")读取，也是面向平台部署形态的例外，工具自身的 API Key、模型等核心配置一律由服务层注入；
5. 错误恢复：在可能的情况下提供错误恢复机制（如图片过滤失败回退原始列表、记忆写入超限降级为对话结束自动保存）。

结语

通过本文可以看到，Nexent SDK 的工具体系并非简单的函数集合，而是一套有着统一基类、统一消息契约、统一命名与标识符规范、统一安全与降级策略的工程化体系。无论是参考 README.md 与 工具开发规范原文 快速上手，还是对照 exa_search_tool.py、knowledge_base_search_tool.py 等真实实现深入理解消息桥接、结果格式化与安全校验的底层细节，遵循本文的规范都能保证新工具与现有生态保持高度一致。工具层之上的能力——无论是网络检索、知识库问答、多模态理解、记忆沉淀还是并发编排——都将因此获得更强的可维护性与可扩展性。

• 人工智能
• AI Agent
• Agent 编排
• 低代码
• AI 技能
• RAG