深圳网站设计是什么软件技术发展前景与趋势

深圳网站设计是什么,软件技术发展前景与趋势,wordpress wp_term_taxonomy,电子商务seo是什么意思文章详细介绍了如何从零开始构建名为HelloAgents的AI智能体框架#xff0c;分析了现有框架的局限性#xff0c;提出了轻量级教学友好、基于标准API、渐进式学习路径和统一工具抽象四大设计理念。通过核心框架层、Agent实现层和工具系统层的架构设计#xff0c;配合具体代码实…文章详细介绍了如何从零开始构建名为HelloAgents的AI智能体框架分析了现有框架的局限性提出了轻量级教学友好、基于标准API、渐进式学习路径和统一工具抽象四大设计理念。通过核心框架层、Agent实现层和工具系统层的架构设计配合具体代码实现和测试案例帮助开发者从使用者转变为构建者深度理解Agent工作原理掌握定制化开发能力。Datawhale干货作者陈思州Datawhale成员为了便于大家更系统的入门和学习AI智能体最近我们将为大家分享系列内容Agent合集在前面的文章中我们讲解了智能体的基础知识并体验了主流框架带来的开发便利。从本文开始我们将进入一个更具挑战也更有价值的阶段从零开始逐步构建一个智能体框架 —— HelloAgents。为什么需要自建Agent框架在智能体技术快速发展的今天市面上已经存在众多成熟的Agent框架。那么为什么我们还要从零开始构建一个新的框架呢1市面框架的快速迭代与局限性智能体领域是一个快速发展的领域随时会有新的概念产生对于智能体的设计每个框架都有自己的定位和理解不过智能体的核心知识点是一致的。过度抽象的复杂性许多框架为了追求通用性引入了大量抽象层和配置选项。以LangChain为例其链式调用机制虽然灵活但对初学者而言学习曲线陡峭往往需要理解大量概念才能完成简单任务。快速迭代带来的不稳定性商业化框架为了抢占市场API接口变更频繁。开发者经常面临版本升级后代码无法运行的困扰维护成本居高不下。黑盒化的实现逻辑许多框架将核心逻辑封装得过于严密开发者难以理解Agent的内部工作机制缺乏深度定制能力。遇到问题时只能依赖文档和社区支持尤其是如果社区不够活跃可能一个反馈意见会非常久也没有人推进影响后续的开发效率。依赖关系的复杂性成熟框架往往携带大量依赖包安装包体积庞大在需要与别的项目代码配合的下可能出现依赖冲突问题。2从使用者到构建者的能力跃迁构建自己的Agent框架实际上是一个从使用者向构建者转变的过程。这种转变带来的价值是长远的。深度理解Agent工作原理通过亲手实现每个组件开发者能够真正理解Agent的思考过程、工具调用机制、以及各种设计模式的好坏与区别。获得完全的控制权自建框架意味着对每一行代码都有完全的掌控可以根据具体需求进行精确调优而不受第三方框架设计理念的束缚。培养系统设计能力框架构建过程涉及模块化设计、接口抽象、错误处理等软件工程核心技能这些能力对开发者的长期成长具有重要价值。3定制化需求与深度掌握的必要性在实际应用中不同场景对智能体的需求差异巨大往往都需要在通用框架基础上做二次开发。特定领域的优化需求金融、医疗、教育等垂直领域往往需要针对性的提示词模板、特殊的工具集成、以及定制化的安全策略。性能与资源的精确控制生产环境中对响应时间、内存占用、并发处理能力都有严格要求通用框架的一刀切方案往往无法满足精细化需求。学习与教学的透明性要求在我们的教学场景中学习者更期待的是清晰地看到智能体的每一步构建过程理解不同范式的工作机制这要求框架具有高度的可观测性和可解释性。构建一个新框架需要怎么设计构建一个新的Agent框架关键不在于功能的多少而在于设计理念是否能真正解决现有框架的痛点。HelloAgents框架的设计围绕着一个核心问题展开如何让学习者既能快速上手又能深入理解Agent的工作原理当你初次接触任何成熟的框架时可能会被其丰富的功能所吸引但很快就会发现一个问题要完成一个简单的任务往往需要理解Chain、Agent、Tool、Memory、Retriever等十几个不同的概念。每个概念都有自己的抽象层学习曲线变得异常陡峭。这种复杂性虽然带来了强大的功能但也成为了初学者的障碍。HelloAgents框架试图在功能完整性和学习友好性之间找到平衡点形成了四个核心的设计理念。1轻量级与教学友好的平衡一个优秀的学习框架应该具备完整的可读性。HelloAgents将核心代码按照章节区分开这是基于一个简单的原则任何有一定编程基础的开发者都应该能够在合理的时间内完全理解框架的工作原理。在依赖管理方面框架采用了极简主义的策略。除了OpenAI的官方SDK和几个必要的基础库外不引入任何重型依赖。如果遇到问题时我们可以直接定位到框架本身的代码而不需要在复杂的依赖关系中寻找答案。2基于标准API的务实选择OpenAI的API已经成为了行业标准几乎所有主流的LLM提供商都在努力兼容这套接口。HelloAgents选择在这个标准之上构建而不是重新发明一套抽象接口。这个决定主要是出于几点动机。首先是兼容性的保证当你掌握了HelloAgents的使用方法后迁移到其他框架或将其集成到现有项目中时底层的API调用逻辑是完全一致的。其次是学习成本的降低。你不需要学习新的概念模型因为所有的操作都基于你已经熟悉的标准接口。3渐进式学习路径的精心设计HelloAgents提供了一条清晰的学习路径。我们将会把每一章的学习代码保存为一个可以pip下载的历史版本因此无需担心代码的使用成本因为每一个核心的功能都将会是你自己编写的。这种设计让你能够按照自己的需求和节奏前进。每一步的升级都是自然而然的不会产生概念上的跳跃或理解上的断层。值得一提的是我们这一章的内容也是基于前六章的内容来完善的。同样这一章也是为后续高级知识学习部分打下框架基础。4统一的“工具”抽象万物皆为工具为了彻底贯彻轻量级与教学友好的理念HelloAgents在架构上做出了一个关键的简化除了核心的Agent类一切皆为Tools。在许多其他框架中需要独立学习的Memory记忆、RAG检索增强生成、RL强化学习、MCP协议等模块在HelloAgents中都被统一抽象为一种“工具”。这种设计的初衷是消除不必要的抽象层让学习者可以回归到最直观的“智能体调用工具”这一核心逻辑上从而真正实现快速上手和深入理解的统一。构建一个新框架的全过程以我们的HelloAgents为例让我们先看看核心学习内容hello-agents/ ├── hello_agents/ │ │ │ ├── core/ # 核心框架层 │ │ ├── agent.py # Agent基类 │ │ ├── llm.py # HelloAgentsLLM统一接口 │ │ ├── message.py # 消息系统 │ │ ├── config.py # 配置管理 │ │ └── exceptions.py # 异常体系 │ │ │ ├── agents/ # Agent实现层 │ │ ├── simple_agent.py # SimpleAgent实现 │ │ ├── react_agent.py # ReActAgent实现 │ │ ├── reflection_agent.py # ReflectionAgent实现 │ │ └── plan_solve_agent.py # PlanAndSolveAgent实现 │ │ │ ├── tools/ # 工具系统层 │ │ ├── base.py # 工具基类 │ │ ├── registry.py # 工具注册机制 │ │ ├── chain.py # 工具链管理系统 │ │ ├── async_executor.py # 异步工具执行器 │ │ └── builtin/ # 内置工具集 │ │ ├── calculator.py # 计算工具 │ │ └── search.py # 搜索工具 └──在开始编写具体代码之前我们需要先建立一个清晰的架构蓝图。HelloAgents的架构设计遵循了分层解耦、职责单一、接口统一的核心原则这样既保持了代码的组织性也便于按照章节扩展内容。模块一核心框架层1、HelloAgentsLLM 扩展本节重点给大家介绍我们的升级方向代码不会展开。代码实践查看https://datawhalechina.github.io/hello-agents我们将把这个基础客户端改造为一个更具适应性的模型调用中枢。本次升级主要围绕以下三个目标展开多提供商支持实现对 OpenAI、ModelScope、智谱 AI 等多种主流 LLM 服务商的无缝切换避免框架与特定供应商绑定。本地模型集成引入 VLLM 和 Ollama 这两种高性能本地部署方案作为对第 3.2.3 节中 Hugging Face Transformers 方案的生产级补充满足数据隐私和成本控制的需求。自动检测机制建立一套自动识别机制使框架能根据环境信息智能推断所使用的 LLM 服务类型简化用户的配置过程。1️⃣ 支持多提供商我们之前定义的HelloAgentsLLM类已经能够通过api_key和base_url这两个核心参数连接任何兼容 OpenAI 接口的服务。这在理论上保证了通用性但在实际应用中不同的服务商在环境变量命名、默认 API 地址和推荐模型等方面都存在差异。如果每次切换服务商都需要用户手动查询并修改代码会极大影响开发效率。为了解决这一问题我们引入provider。其改进思路是让HelloAgentsLLM在内部处理不同服务商的配置细节从而为用户提供一个统一、简洁的调用体验。2️⃣ 本地模型调用为了在本地实现高性能、生产级的模型推理服务社区涌现出了 VLLM 和 Ollama 等优秀工具。它们通过连续批处理、PagedAttention 等技术显著提升了模型的吞吐量和运行效率并将模型封装为兼容 OpenAI 标准的 API 服务。这意味着我们可以将它们无缝地集成到HelloAgentsLLM中。VLLMVLLM 是一个为 LLM 推理设计的高性能 Python 库。它通过 PagedAttention 等先进技术可以实现比标准 Transformers 实现高出数倍的吞吐量。下面是在本地部署一个 VLLM 服务的完整步骤首先需要根据你的硬件环境特别是 CUDA 版本安装 VLLM。推荐遵循其官方文档进行安装以避免版本不匹配问题。安装完成后使用以下命令即可启动一个兼容 OpenAI 的 API 服务。VLLM 会自动从 Hugging Face Hub 下载指定的模型权重如果本地不存在。服务启动后便会在http://localhost:8000/v1地址上提供与 OpenAI 兼容的 API。OllamaOllama 进一步简化了本地模型的管理和部署它将模型下载、配置和服务启动等步骤封装到了一条命令中非常适合快速上手。访问 Ollama 官方网站下载并安装适用于你操作系统的客户端。安装后打开终端执行以下命令即可下载并运行一个模型以 Llama 3 为例。Ollama 会自动处理模型的下载、服务封装和硬件加速配置。当你在终端看到模型的交互提示时即表示服务已经成功在后台启动。Ollama 默认会在http://localhost:11434/v1地址上暴露 OpenAI 兼容的 API 接口。接入HelloAgentsLLM由于 VLLM 和 Ollama 都遵循了行业标准 API将它们接入HelloAgentsLLM的过程非常简单。我们只需在实例化客户端时将它们视为一个新的provider即可。通过这种统一的设计我们的智能体核心代码无需任何修改就可以在云端 API 和本地模型之间自由切换。这为后续应用的开发、部署、成本控制以及保护数据隐私提供了极大的灵活性。3️⃣ 自动检测机制为了尽可能减少用户的配置负担并遵循“约定优于配置”的原则HelloAgentsLLM内部设计了两个核心辅助方法_auto_detect_provider和_resolve_credentials。它们协同工作_auto_detect_provider负责根据环境信息推断服务商而_resolve_credentials则根据推断结果完成具体的参数配置。_auto_detect_provider方法负责根据环境信息按照下述优先级顺序尝试自动推断服务商最高优先级检查特定服务商的环境变量这是最直接、最可靠的判断依据。框架会依次检查MODELSCOPE_API_KEY,OPENAI_API_KEY,ZHIPU_API_KEY等环境变量是否存在。一旦发现任何一个就会立即确定对应的服务商。次高优先级根据base_url进行判断如果用户没有设置特定服务商的密钥但设置了通用的LLM_BASE_URL框架会转而解析这个 URL。域名匹配通过检查 URL 中是否包含api-inference.modelscope.cn,api.openai.com等特征字符串来识别云服务商。端口匹配通过检查 URL 中是否包含:11434(Ollama),:8000(VLLM) 等本地服务的标准端口来识别本地部署方案。辅助判断分析 API 密钥的格式在某些情况下如果上述两种方式都无法确定框架会尝试分析通用环境变量LLM_API_KEY的格式。例如某些服务商的 API 密钥有固定的前缀或独特的编码格式。不过由于这种方式可能存在模糊性例如多个服务商的密钥格式相似因此它的优先级较低仅作为辅助手段。一旦provider被确定无论是用户指定还是自动检测_resolve_credentials方法便会接手处理服务商的差异化配置。它会根据provider的值去主动查找对应的环境变量并为其设置默认的base_url。通过三种方法现在的HelloAgentsLLM具有以下显著优势表 1 HelloAgentLLM不同版本特性对比如上表所示这种演进体现了框架设计的重要原则从简单开始逐步完善。我们在保持接口简洁的同时增强了功能的完整性。2、框架接口实现在上节中我们构建了HelloAgentsLLM这一核心组件解决了与大语言模型通信的关键问题。不过它还需要一系列配套的接口和组件来处理数据流、管理配置、应对异常并为上层应用的构建提供一个清晰、统一的结构。本节将讲述以下三个核心文件message.py 定义了框架内统一的消息格式确保了智能体与模型之间信息传递的标准化。config.py 提供了一个中心化的配置管理方案使框架的行为易于调整和扩展。agent.py 定义了所有智能体的抽象基类Agent为后续实现不同类型的智能体提供了统一的接口和规范。1️⃣ Message 类在智能体与大语言模型的交互中对话历史是至关重要的上下文。为了规范地管理这些信息我们设计了一个简易Message类。在后续上下文工程章节中会对其进行扩展。消息系统 from typing import Optional, Dict, Any, Literal from datetime import datetime from pydantic import BaseModel # 定义消息角色的类型限制其取值 MessageRole Literal[user, assistant, system, tool] classMessage(BaseModel): 消息类 content: str role: MessageRole timestamp: datetime None metadata: Optional[Dict[str, Any]] None def__init__(self, content: str, role: MessageRole, **kwargs): super().__init__( contentcontent, rolerole, timestampkwargs.get(timestamp, datetime.now()), metadatakwargs.get(metadata, {}) ) defto_dict(self) - Dict[str, Any]: 转换为字典格式OpenAI API格式 return { role: self.role, content: self.content } def__str__(self) - str: returnf[{self.role}] {self.content}该类的设计有几个关键点。首先我们通过typing.Literal将role字段的取值严格限制为user,assistant,system,tool四种这直接对应 OpenAI API 的规范保证了类型安全。除了content和role这两个核心字段外我们还增加了timestamp和metadata为日志记录和未来功能扩展预留了空间。最后to_dict()方法是其核心功能之一负责将内部使用的Message对象转换为与 OpenAI API 兼容的字典格式体现了“对内丰富对外兼容”的设计原则。2️⃣ Config 类Config类的职责是将代码中硬编码配置参数集中起来并支持从环境变量中读取。配置管理 import os from typing import Optional, Dict, Any from pydantic import BaseModel classConfig(BaseModel): HelloAgents配置类 # LLM配置 default_model: str gpt-3.5-turbo default_provider: str openai temperature: float 0.7 max_tokens: Optional[int] None # 系统配置 debug: bool False log_level: str INFO # 其他配置 max_history_length: int 100 classmethod deffrom_env(cls) - Config: 从环境变量创建配置 return cls( debugos.getenv(DEBUG, false).lower() true, log_levelos.getenv(LOG_LEVEL, INFO), temperaturefloat(os.getenv(TEMPERATURE, 0.7)), max_tokensint(os.getenv(MAX_TOKENS)) if os.getenv(MAX_TOKENS) elseNone, ) defto_dict(self) - Dict[str, Any]: 转换为字典 return self.dict()首先我们将配置项按逻辑划分为LLM配置、系统配置等使结构一目了然。其次每个配置项都设有合理的默认值保证了框架在零配置下也能工作。最核心的是from_env()类方法它允许用户通过设置环境变量来覆盖默认配置无需修改代码这在部署到不同环境时尤其有用。3️⃣ Agent 抽象基类Agent类是整个框架的顶层抽象。它定义了一个智能体应该具备的通用行为和属性但并不关心具体的实现方式。我们通过 Python 的abc(Abstract Base Classes) 模块来实现它这强制所有具体的智能体实现如后续章节的SimpleAgent,ReActAgent等都必须遵循同一个“接口”。Agent基类 from abc import ABC, abstractmethod from typing import Optional, Any from .message import Message from .llm import HelloAgentsLLM from .config import Config classAgent(ABC): Agent基类 def__init__( self, name: str, llm: HelloAgentsLLM, system_prompt: Optional[str] None, config: Optional[Config] None ): self.name name self.llm llm self.system_prompt system_prompt self.config config or Config() self._history: list[Message] [] abstractmethod defrun(self, input_text: str, **kwargs) - str: 运行Agent pass defadd_message(self, message: Message): 添加消息到历史记录 self._history.append(message) defclear_history(self): 清空历史记录 self._history.clear() defget_history(self) - list[Message]: 获取历史记录 return self._history.copy() def__str__(self) - str: returnfAgent(name{self.name}, provider{self.llm.provider})该类的设计体现了面向对象中的抽象原则。首先它通过继承ABC被定义为一个不能直接实例化的抽象类。其构造函数__init__清晰地定义了 Agent 的核心依赖名称、LLM 实例、系统提示词和配置。最重要的部分是使用abstractmethod装饰的run方法它强制所有子类必须实现此方法从而保证了所有智能体都有统一的执行入口。此外基类还提供了通用的历史记录管理方法这些方法与Message类协同工作体现了组件间的联系。至此我们已经完成了HelloAgents框架核心基础组件的设计与实现。模块二Agent实现层本节内容将在三种经典Agent范式ReAct、Plan-and-Solve、Reflection基础上进行框架化重构并新增SimpleAgent作为基础对话范式。我们将把这些独立的Agent实现改造为基于统一架构的框架组件。本次重构主要围绕以下三个核心目标展开提示词工程的系统性提升对第四章中的提示词进行深度优化从特定任务导向转向通用化设计同时增强格式约束和角色定义。接口与格式的标准化统一建立统一的Agent基类和标准化的运行接口所有Agent都遵循相同的初始化参数、方法签名和历史管理机制。高度可配置的自定义能力支持用户自定义提示词模板、配置参数和执行策略。1️⃣ SimpleAgentSimpleAgent是最基础的Agent实现它展示了如何在框架基础上构建一个完整的对话智能体。我们将通过继承框架基类来重写SimpleAgent。首先在你的项目目录中创建一个my_simple_agent.py文件# my_simple_agent.py from typing import Optional, Iterator from hello_agents import SimpleAgent, HelloAgentsLLM, Config, Message classMySimpleAgent(SimpleAgent): 重写的简单对话Agent 展示如何基于框架基类构建自定义Agent def__init__( self, name: str, llm: HelloAgentsLLM, system_prompt: Optional[str] None, config: Optional[Config] None, tool_registry: Optional[ToolRegistry] None, enable_tool_calling: bool True ): super().__init__(name, llm, system_prompt, config) self.tool_registry tool_registry self.enable_tool_calling enable_tool_calling and tool_registry isnotNone print(f✅ {name} 初始化完成工具调用: {启用if self.enable_tool_calling else禁用})接下来我们需要重写Agent基类的抽象方法run。SimpleAgent支持可选的工具调用功能也方便后续章节的扩展# 继续在 my_simple_agent.py 中添加 import re classMySimpleAgent(SimpleAgent): # ... 前面的 __init__ 方法 defrun(self, input_text: str, max_tool_iterations: int 3, **kwargs) - str: 重写的运行方法 - 实现简单对话逻辑支持可选工具调用 print(f {self.name} 正在处理: {input_text}) # 构建消息列表 messages [] # 添加系统消息可能包含工具信息 enhanced_system_prompt self._get_enhanced_system_prompt() messages.append({role: system, content: enhanced_system_prompt}) # 添加历史消息 for msg in self._history: messages.append({role: msg.role, content: msg.content}) # 添加当前用户消息 messages.append({role: user, content: input_text}) # 如果没有启用工具调用使用简单对话逻辑 ifnot self.enable_tool_calling: response self.llm.invoke(messages, **kwargs) self.add_message(Message(input_text, user)) self.add_message(Message(response, assistant)) print(f✅ {self.name} 响应完成) return response # 支持多轮工具调用的逻辑 return self._run_with_tools(messages, input_text, max_tool_iterations, **kwargs) def_get_enhanced_system_prompt(self) - str: 构建增强的系统提示词包含工具信息 base_prompt self.system_prompt or你是一个有用的AI助手。 ifnot self.enable_tool_calling ornot self.tool_registry: return base_prompt # 获取工具描述 tools_description self.tool_registry.get_tools_description() ifnot tools_description or tools_description 暂无可用工具: return base_prompt tools_section \n\n## 可用工具\n tools_section 你可以使用以下工具来帮助回答问题:\n tools_section tools_description \n tools_section \n## 工具调用格式\n tools_section 当需要使用工具时请使用以下格式:\n tools_section [TOOL_CALL:{tool_name}:{parameters}]\n tools_section 例如:[TOOL_CALL:search:Python编程] 或 [TOOL_CALL:memory:recall用户信息]\n\n tools_section 工具调用结果会自动插入到对话中然后你可以基于结果继续回答。\n return base_prompt tools_section现在我们实现工具调用的核心逻辑# 继续在 my_simple_agent.py 中添加 classMySimpleAgent(SimpleAgent): # ... 前面的方法 def_run_with_tools(self, messages: list, input_text: str, max_tool_iterations: int, **kwargs) - str: 支持工具调用的运行逻辑 current_iteration 0 final_response while current_iteration max_tool_iterations: # 调用LLM response self.llm.invoke(messages, **kwargs) # 检查是否有工具调用 tool_calls self._parse_tool_calls(response) if tool_calls: print(f 检测到 {len(tool_calls)} 个工具调用) # 执行所有工具调用并收集结果 tool_results [] clean_response response for call in tool_calls: result self._execute_tool_call(call[tool_name], call[parameters]) tool_results.append(result) # 从响应中移除工具调用标记 clean_response clean_response.replace(call[original], ) # 构建包含工具结果的消息 messages.append({role: assistant, content: clean_response}) # 添加工具结果 tool_results_text \n\n.join(tool_results) messages.append({role: user, content: f工具执行结果:\n{tool_results_text}\n\n请基于这些结果给出完整的回答。}) current_iteration 1 continue # 没有工具调用这是最终回答 final_response response break # 如果超过最大迭代次数获取最后一次回答 if current_iteration max_tool_iterations andnot final_response: final_response self.llm.invoke(messages, **kwargs) # 保存到历史记录 self.add_message(Message(input_text, user)) self.add_message(Message(final_response, assistant)) print(f✅ {self.name} 响应完成) return final_response def_parse_tool_calls(self, text: str) - list: 解析文本中的工具调用 pattern r\[TOOL_CALL:([^:]):([^\]])\] matches re.findall(pattern, text) tool_calls [] for tool_name, parameters in matches: tool_calls.append({ tool_name: tool_name.strip(), parameters: parameters.strip(), original: f[TOOL_CALL:{tool_name}:{parameters}] }) return tool_calls def_execute_tool_call(self, tool_name: str, parameters: str) - str: 执行工具调用 ifnot self.tool_registry: returnf❌ 错误:未配置工具注册表 try: # 智能参数解析 if tool_name calculator: # 计算器工具直接传入表达式 result self.tool_registry.execute_tool(tool_name, parameters) else: # 其他工具使用智能参数解析 param_dict self._parse_tool_parameters(tool_name, parameters) tool self.tool_registry.get_tool(tool_name) ifnot tool: returnf❌ 错误:未找到工具 {tool_name} result tool.run(param_dict) returnf 工具 {tool_name} 执行结果:\n{result} except Exception as e: returnf❌ 工具调用失败:{str(e)} def_parse_tool_parameters(self, tool_name: str, parameters: str) - dict: 智能解析工具参数 param_dict {} ifin parameters: # 格式: keyvalue 或 actionsearch,queryPython if,in parameters: # 多个参数:actionsearch,queryPython,limit3 pairs parameters.split(,) for pair in pairs: ifin pair: key, value pair.split(, 1) param_dict[key.strip()] value.strip() else: # 单个参数:keyvalue key, value parameters.split(, 1) param_dict[key.strip()] value.strip() else: # 直接传入参数根据工具类型智能推断 if tool_name search: param_dict {query: parameters} elif tool_name memory: param_dict {action: search, query: parameters} else: param_dict {input: parameters} return param_dict我们还可以为自定义Agent添加流式响应功能和便利方法# 继续在 my_simple_agent.py 中添加 classMySimpleAgent(SimpleAgent): # ... 前面的方法 defstream_run(self, input_text: str, **kwargs) - Iterator[str]: 自定义的流式运行方法 print(f {self.name} 开始流式处理: {input_text}) messages [] if self.system_prompt: messages.append({role: system, content: self.system_prompt}) for msg in self._history: messages.append({role: msg.role, content: msg.content}) messages.append({role: user, content: input_text}) # 流式调用LLM full_response print( 实时响应: , end) for chunk in self.llm.stream_invoke(messages, **kwargs): full_response chunk print(chunk, end, flushTrue) yield chunk print() # 换行 # 保存完整对话到历史记录 self.add_message(Message(input_text, user)) self.add_message(Message(full_response, assistant)) print(f✅ {self.name} 流式响应完成) defadd_tool(self, tool) - None: 添加工具到Agent便利方法 ifnot self.tool_registry: from hello_agents import ToolRegistry self.tool_registry ToolRegistry() self.enable_tool_calling True self.tool_registry.register_tool(tool) print(f 工具 {tool.name} 已添加) defhas_tools(self) - bool: 检查是否有可用工具 return self.enable_tool_calling and self.tool_registry isnotNone defremove_tool(self, tool_name: str) - bool: 移除工具便利方法 if self.tool_registry: self.tool_registry.unregister(tool_name) returnTrue returnFalse deflist_tools(self) - list: 列出所有可用工具 if self.tool_registry: return self.tool_registry.list_tools() return []创建一个测试文件test_simple_agent.py# test_simple_agent.py from dotenv import load_dotenv from hello_agents import HelloAgentsLLM, ToolRegistry from hello_agents.tools import CalculatorTool from my_simple_agent import MySimpleAgent # 加载环境变量 load_dotenv() # 创建LLM实例 llm HelloAgentsLLM() # 测试1:基础对话Agent无工具 print( 测试1:基础对话 ) basic_agent MySimpleAgent( name基础助手, llmllm, system_prompt你是一个友好的AI助手请用简洁明了的方式回答问题。 ) response1 basic_agent.run(你好请介绍一下自己) print(f基础对话响应: {response1}\n) # 测试2:带工具的Agent print( 测试2:工具增强对话 ) tool_registry ToolRegistry() calculator CalculatorTool() tool_registry.register_tool(calculator) enhanced_agent MySimpleAgent( name增强助手, llmllm, system_prompt你是一个智能助手可以使用工具来帮助用户。, tool_registrytool_registry, enable_tool_callingTrue ) response2 enhanced_agent.run(请帮我计算 15 * 8 32) print(f工具增强响应: {response2}\n) # 测试3:流式响应 print( 测试3:流式响应 ) print(流式响应: , end) for chunk in basic_agent.stream_run(请解释什么是人工智能): pass# 内容已在stream_run中实时打印 # 测试4:动态添加工具 print(\n 测试4:动态工具管理 ) print(f添加工具前: {basic_agent.has_tools()}) basic_agent.add_tool(calculator) print(f添加工具后: {basic_agent.has_tools()}) print(f可用工具: {basic_agent.list_tools()}) # 查看对话历史 print(f\n对话历史: {len(basic_agent.get_history())} 条消息)在本节中我们通过继承Agent基类成功构建了一个功能完备且遵循框架规范的基础对话智能体MySimpleAgent。它不仅支持基础对话还具备可选的工具调用能力、流式响应和便利的工具管理方法。2️⃣ ReActAgent框架化的 ReActAgent 在保持核心逻辑不变的同时提升了代码的组织性和可维护性主要是通过提示词优化和与框架工具系统的集成。1提示词模板的改进保持了原有的格式要求强调每次只能执行一个步骤避免混乱并明确了两种Action的使用场景。MY_REACT_PROMPT 你是一个具备推理和行动能力的AI助手。你可以通过思考分析问题然后调用合适的工具来获取信息最终给出准确的答案。 ## 可用工具 {tools} ## 工作流程 请严格按照以下格式进行回应每次只能执行一个步骤: Thought: 分析当前问题思考需要什么信息或采取什么行动。 Action: 选择一个行动格式必须是以下之一: - {{tool_name}}[{{tool_input}}] - 调用指定工具 - Finish[最终答案] - 当你有足够信息给出最终答案时 ## 重要提醒 1. 每次回应必须包含Thought和Action两部分 2. 工具调用的格式必须严格遵循:工具名[参数] 3. 只有当你确信有足够信息回答问题时才使用Finish 4. 如果工具返回的信息不够继续使用其他工具或相同工具的不同参数 ## 当前任务 **Question:** {question} ## 执行历史 {history} 现在开始你的推理和行动: 2重写ReActAgent的完整实现创建my_react_agent.py文件来重写ReActAgent# my_react_agent.py import re from typing import Optional, List, Tuple from hello_agents import ReActAgent, HelloAgentsLLM, Config, Message, ToolRegistry classMyReActAgent(ReActAgent): 重写的ReAct Agent - 推理与行动结合的智能体 def__init__( self, name: str, llm: HelloAgentsLLM, tool_registry: ToolRegistry, system_prompt: Optional[str] None, config: Optional[Config] None, max_steps: int 5, custom_prompt: Optional[str] None ): super().__init__(name, llm, system_prompt, config) self.tool_registry tool_registry self.max_steps max_steps self.current_history: List[str] [] self.prompt_template custom_prompt if custom_prompt else MY_REACT_PROMPT print(f✅ {name} 初始化完成最大步数: {max_steps})其初始化参数的含义如下name Agent的名称。llmHelloAgentsLLM的实例负责与大语言模型通信。tool_registryToolRegistry的实例用于管理和执行Agent可用的工具。system_prompt 系统提示词用于设定Agent的角色和行为准则。config 配置对象用于传递框架级的设置。max_steps ReAct循环的最大执行步数防止无限循环。custom_prompt 自定义的提示词模板用于替换默认的ReAct提示词。框架化的ReActAgent将执行流程分解为清晰的步骤defrun(self, input_text: str, **kwargs) - str: 运行ReAct Agent self.current_history [] current_step 0 print(f\n {self.name} 开始处理问题: {input_text}) while current_step self.max_steps: current_step 1 print(f\n--- 第 {current_step} 步 ---) # 1. 构建提示词 tools_desc self.tool_registry.get_tools_description() history_str \n.join(self.current_history) prompt self.prompt_template.format( toolstools_desc, questioninput_text, historyhistory_str ) # 2. 调用LLM messages [{role: user, content: prompt}] response_text self.llm.invoke(messages, **kwargs) # 3. 解析输出 thought, action self._parse_output(response_text) # 4. 检查完成条件 if action and action.startswith(Finish): final_answer self._parse_action_input(action) self.add_message(Message(input_text, user)) self.add_message(Message(final_answer, assistant)) return final_answer # 5. 执行工具调用 if action: tool_name, tool_input self._parse_action(action) observation self.tool_registry.execute_tool(tool_name, tool_input) self.current_history.append(fAction: {action}) self.current_history.append(fObservation: {observation}) # 达到最大步数 final_answer 抱歉我无法在限定步数内完成这个任务。 self.add_message(Message(input_text, user)) self.add_message(Message(final_answer, assistant)) return final_answer通过以上重构我们将 ReAct 范式成功地集成到了框架中。核心改进在于利用了统一的ToolRegistry接口并通过一个可配置、格式更严谨的提示词模板提升了智能体执行思考-行动循环的稳定性。对于ReAct的测试案例由于需要调用工具所以统一放在文末提供测试代码。3️⃣ ReflectionAgent由于这几类Agent已经实现过核心逻辑所以这里只给出对应的Prompt。与之前专门针对代码生成的提示词不同框架化的版本采用了通用化设计使其适用于文本生成、分析、创作等多种场景并通过custom_prompts参数支持用户深度定制。DEFAULT_PROMPTS { initial: 请根据以下要求完成任务: 任务: {task} 请提供一个完整、准确的回答。 , reflect: 请仔细审查以下回答并找出可能的问题或改进空间: # 原始任务: {task} # 当前回答: {content} 请分析这个回答的质量指出不足之处并提出具体的改进建议。 如果回答已经很好请回答无需改进。 , refine: 请根据反馈意见改进你的回答: # 原始任务: {task} # 上一轮回答: {last_attempt} # 反馈意见: {feedback} 请提供一个改进后的回答。 }你可以尝试根据第四章的代码以及上文ReAct的实现构建出自己的MyReflectionAgent。下面提供一个测试代码供验证想法。# test_reflection_agent.py from dotenv import load_dotenv from hello_agents import HelloAgentsLLM from my_reflection_agent import MyReflectionAgent load_dotenv() llm HelloAgentsLLM() # 使用默认通用提示词 general_agent MyReflectionAgent(name我的反思助手, llmllm) # 使用自定义代码生成提示词类似第四章 code_prompts { initial: 你是Python专家请编写函数:{task}, reflect: 请审查代码的算法效率:\n任务:{task}\n代码:{content}, refine: 请根据反馈优化代码:\n任务:{task}\n反馈:{feedback} } code_agent MyReflectionAgent( name我的代码生成助手, llmllm, custom_promptscode_prompts ) # 测试使用 result general_agent.run(写一篇关于人工智能发展历程的简短文章) print(f最终结果: {result})4️⃣ PlanAndSolveAgent与第四章自由文本的计划输出不同框架化版本强制要求Planner以Python列表的格式输出计划并提供了完整的异常处理机制确保了后续步骤能够稳定执行。框架化的Plan-and-Solve提示词# 默认规划器提示词模板 DEFAULT_PLANNER_PROMPT 你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。 请确保计划中的每个步骤都是一个独立的、可执行的子任务并且严格按照逻辑顺序排列。 你的输出必须是一个Python列表其中每个元素都是一个描述子任务的字符串。 问题: {question} 请严格按照以下格式输出你的计划: python [步骤1, 步骤2, 步骤3, ...]“”默认执行器提示词模板DEFAULT_EXECUTOR_PROMPT “”你是一位顶级的AI执行专家。你的任务是严格按照给定的计划一步步地解决问题。你将收到原始问题、完整的计划、以及到目前为止已经完成的步骤和结果。请你专注于解决当前步骤并仅输出该步骤的最终答案不要输出任何额外的解释或对话。原始问题:{question}完整计划:{plan}历史步骤与结果:{history}当前步骤:{current_step}请仅输出针对当前步骤的回答:“”这一节仍然给出一个综合测试文件test_plan_solve_agent.py可以自行设计实现。 plaintext # test_plan_solve_agent.py from dotenv import load_dotenv from hello_agents.core.llm import HelloAgentsLLM from my_plan_solve_agent import MyPlanAndSolveAgent # 加载环境变量 load_dotenv() # 创建LLM实例 llm HelloAgentsLLM() # 创建自定义PlanAndSolveAgent agent MyPlanAndSolveAgent( name我的规划执行助手, llmllm ) # 测试复杂问题 question 一个水果店周一卖出了15个苹果。周二卖出的苹果数量是周一的两倍。周三卖出的数量比周二少了5个。请问这三天总共卖出了多少个苹果 result agent.run(question) print(f\n最终结果: {result}) # 查看对话历史 print(f对话历史: {len(agent.get_history())} 条消息)在最后可以补充一款新的提示词可以尝试实现custom_prompt载入自定义提示词。# 创建专门用于数学问题的自定义提示词 math_prompts { planner: 你是数学问题规划专家。请将数学问题分解为计算步骤: 问题: {question} 输出格式: python [计算步骤1, 计算步骤2, 求总和] , executor: 你是数学计算专家。请计算当前步骤: 问题: {question} 计划: {plan} 历史: {history} 当前步骤: {current_step} 请只输出数值结果: } # 使用自定义提示词创建数学专用Agent math_agent MyPlanAndSolveAgent( name数学计算助手, llmllm, custom_promptsmath_prompts ) # 测试数学问题 math_result math_agent.run(question) print(f数学专用Agent结果: {math_result})如表2所示通过这种框架化的重构我们不仅保持了第四章中各种Agent范式的核心功能还大幅提升了代码的组织性、可维护性和扩展性。所有Agent现在都共享统一的基础架构同时保持了各自的特色和优势。表 2 Agent不同章节实现对比5️⃣ FunctionCallAgentFunctionCallAgent是hello-agents在0.2.8之后引入的Agent它基于OpenAI原生函数调用机制的Agent展示了如何使用OpenAI的函数调用机制来构建Agent。 它支持以下功能 _build_tool_schemas:通过工具的description构建OpenAI的function calling schema _extract_message_content:从OpenAI的响应中提取文本 _parse_function_call_arguments:解析模型返回的JSON字符串参数 _convert_parameter_types:转换参数类型这些功能可以使其具备原生的OpenAI Functioncall的能力对比使用prompt约束的方式具备更强的鲁棒性。def_invoke_with_tools(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]], tool_choice: Union[str, dict], **kwargs): 调用底层OpenAI客户端执行函数调用 client getattr(self.llm, _client, None) if client isNone: raise RuntimeError(HelloAgentsLLM 未正确初始化客户端无法执行函数调用。) client_kwargs dict(kwargs) client_kwargs.setdefault(temperature, self.llm.temperature) if self.llm.max_tokens isnotNone: client_kwargs.setdefault(max_tokens, self.llm.max_tokens) return client.chat.completions.create( modelself.llm.model, messagesmessages, toolstools, tool_choicetool_choice, **client_kwargs, ) #内部逻辑是对Openai 原生的functioncall作再封装 #OpenAI 原生functioncall示例 from openai import OpenAI client OpenAI() tools [ { type: function, function: { name: get_current_weather, description: Get the current weather in a given location, parameters: { type: object, properties: { location: { type: string, description: The city and state, e.g. San Francisco, CA, }, unit: {type: string, enum: [celsius, fahrenheit]}, }, required: [location], }, } } ] messages [{role: user, content: Whats the weather like in Boston today?}] completion client.chat.completions.create( modelgpt-5, messagesmessages, toolstools, tool_choiceauto ) print(completion)模块三工具系统层本节内容将在前面构建的Agent基础架构上深入探讨工具系统的设计与实现。我们将从基础设施建设开始逐步深入到自定义开发设计。本节的学习目标围绕以下三个核心方面展开统一的工具抽象与管理建立标准化的Tool基类和ToolRegistry注册机制为工具的开发、注册、发现和执行提供统一的基础设施。实战驱动的工具开发以数学计算工具为案例展示如何设计和实现自定义工具让读者掌握工具开发的完整流程。高级整合与优化策略通过多源搜索工具的设计展示如何整合多个外部服务实现智能后端选择、结果合并和容错处理体现工具系统在复杂场景下的设计思维。1️⃣ 工具基类与注册机制设计在构建可扩展的工具系统时我们需要首先建立一套标准化的基础设施。这套基础设施包括Tool基类、ToolRegistry注册表以及工具管理机制。1Tool基类的抽象设计Tool基类是整个工具系统的核心抽象它定义了所有工具必须遵循的接口规范classTool(ABC): 工具基类 def__init__(self, name: str, description: str): self.name name self.description description abstractmethod defrun(self, parameters: Dict[str, Any]) - str: 执行工具 pass abstractmethod defget_parameters(self) - List[ToolParameter]: 获取工具参数定义 pass这个设计体现了面向对象设计的核心思想通过统一的run方法接口所有工具都能以一致的方式执行接受字典参数并返回字符串结果确保了框架的一致性。同时工具具备了自描述能力通过get_parameters方法能够清晰地告诉调用者自己需要什么参数这种内省机制为自动化文档生成和参数验证提供了基础。而name和description等元数据的设计则让工具系统具备了良好的可发现性和可理解性。2ToolParameter参数定义系统为了支持复杂的参数验证和文档生成我们设计了ToolParameter类classToolParameter(BaseModel): 工具参数定义 name: str type: str description: str required: bool True default: Any None这种设计让工具能够精确描述自己的参数需求支持类型检查、默认值设置和文档自动生成。3ToolRegistry注册表的实现ToolRegistry是工具系统的管理中枢它提供了工具的注册、发现、执行等核心功能在这一节我们主要用到以下功能classToolRegistry: HelloAgents工具注册表 def__init__(self): self._tools: dict[str, Tool] {} self._functions: dict[str, dict[str, Any]] {} defregister_tool(self, tool: Tool): 注册Tool对象 if tool.name in self._tools: print(f⚠️ 警告:工具 {tool.name} 已存在将被覆盖。) self._tools[tool.name] tool print(f✅ 工具 {tool.name} 已注册。) defregister_function(self, name: str, description: str, func: Callable[[str], str]): 直接注册函数作为工具简便方式 Args: name: 工具名称 description: 工具描述 func: 工具函数接受字符串参数返回字符串结果 if name in self._functions: print(f⚠️ 警告:工具 {name} 已存在将被覆盖。) self._functions[name] { description: description, func: func } print(f✅ 工具 {name} 已注册。)ToolRegistry支持两种注册方式Tool对象注册适合复杂工具支持完整的参数定义和验证函数直接注册适合简单工具快速集成现有函数4工具发现与管理机制注册表提供了丰富的工具管理功能defget_tools_description(self) - str: 获取所有可用工具的格式化描述字符串 descriptions [] # Tool对象描述 for tool in self._tools.values(): descriptions.append(f- {tool.name}: {tool.description}) # 函数工具描述 for name, info in self._functions.items(): descriptions.append(f- {name}: {info[description]}) return\n.join(descriptions) if descriptions else暂无可用工具这个方法生成的描述字符串可以直接用于构建Agent的提示词让Agent了解可用的工具。defto_openai_schema(self) - Dict[str, Any]: 转换为 OpenAI function calling schema 格式 用于 FunctionCallAgent使工具能够被 OpenAI 原生 function calling 使用 Returns: 符合 OpenAI function calling 标准的 schema parameters self.get_parameters() # 构建 properties properties {} required [] for param in parameters: # 基础属性定义 prop { type: param.type, description: param.description } # 如果有默认值添加到描述中OpenAI schema 不支持 default 字段 if param.default isnotNone: prop[description] f{param.description} (默认: {param.default}) # 如果是数组类型添加 items 定义 if param.type array: prop[items] {type: string} # 默认字符串数组 properties[param.name] prop # 收集必需参数 if param.required: required.append(param.name) return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: properties, required: required } } }这个方法生成的schema可以直接用于原生的OpenAI SDK的工具调用。2️⃣ 自定义工具开发有了基础设施后我们来看看如何开发一个完整的自定义工具。数学计算工具是一个很好的例子因为它简单直观最直接的方式是使用ToolRegistry的函数注册功能。让我们创建一个自定义的数学计算工具。首先在你的项目目录中创建my_calculator_tool.py# my_calculator_tool.py import ast import operator import math from hello_agents import ToolRegistry defmy_calculate(expression: str) - str: 简单的数学计算函数 ifnot expression.strip(): return计算表达式不能为空 # 支持的基本运算 operators { ast.Add: operator.add, # ast.Sub: operator.sub, # - ast.Mult: operator.mul, # * ast.Div: operator.truediv, # / } # 支持的基本函数 functions { sqrt: math.sqrt, pi: math.pi, } try: node ast.parse(expression, modeeval) result _eval_node(node.body, operators, functions) return str(result) except: return计算失败请检查表达式格式 def_eval_node(node, operators, functions): 简化的表达式求值 if isinstance(node, ast.Constant): return node.value elif isinstance(node, ast.BinOp): left _eval_node(node.left, operators, functions) right _eval_node(node.right, operators, functions) op operators.get(type(node.op)) return op(left, right) elif isinstance(node, ast.Call): func_name node.func.id if func_name in functions: args [_eval_node(arg, operators, functions) for arg in node.args] return functions[func_name](*args) elif isinstance(node, ast.Name): if node.id in functions: return functions[node.id] defcreate_calculator_registry(): 创建包含计算器的工具注册表 registry ToolRegistry() # 注册计算器函数 registry.register_function( namemy_calculator, description简单的数学计算工具支持基本运算(,-,*,/)和sqrt函数, funcmy_calculate ) return registry工具不仅支持基本的四则运算还涵盖了常用的数学函数和常数满足了大多数计算场景的需求。你也可以自己扩展这个文件制作一个更加完备的计算函数。我们提供一个测试文件test_my_calculator.py帮助你验证功能实现# test_my_calculator.py from dotenv import load_dotenv from my_calculator_tool import create_calculator_registry # 加载环境变量 load_dotenv() deftest_calculator_tool(): 测试自定义计算器工具 # 创建包含计算器的注册表 registry create_calculator_registry() print( 测试自定义计算器工具\n) # 简单测试用例 test_cases [ 2 3, # 基本加法 10 - 4, # 基本减法 5 * 6, # 基本乘法 15 / 3, # 基本除法 sqrt(16), # 平方根 ] for i, expression in enumerate(test_cases, 1): print(f测试 {i}: {expression}) result registry.execute_tool(my_calculator, expression) print(f结果: {result}\n) deftest_with_simple_agent(): 测试与SimpleAgent的集成 from hello_agents import HelloAgentsLLM # 创建LLM客户端 llm HelloAgentsLLM() # 创建包含计算器的注册表 registry create_calculator_registry() print( 与SimpleAgent集成测试:) # 模拟SimpleAgent使用工具的场景 user_question 请帮我计算 sqrt(16) 2 * 3 print(f用户问题: {user_question}) # 使用工具计算 calc_result registry.execute_tool(my_calculator, sqrt(16) 2 * 3) print(f计算结果: {calc_result}) # 构建最终回答 final_messages [ {role: user, content: f计算结果是 {calc_result}请用自然语言回答用户的问题:{user_question}} ] print(\n SimpleAgent的回答:) response llm.think(final_messages) for chunk in response: print(chunk, end, flushTrue) print(\n) if __name__ __main__: test_calculator_tool() test_with_simple_agent()通过这个简化的数学计算工具案例我们学会了如何快速开发自定义工具编写一个简单的计算函数通过ToolRegistry注册然后与SimpleAgent集成使用。为了更直观的观察这里提供了图1可以清晰理解代码的运行逻辑。图 1 基于Helloagents的SimpleAgent运行工作流3️⃣ 多源搜索工具在实际应用中我们经常需要整合多个外部服务来提供更强大的功能。搜索工具就是一个典型的例子它整合多个搜索引擎能提供更加完备的真实信息。在第一章我们使用过Tavily的搜索API在第四章我们使用过SerpApi的搜索API。因此这次我们使用这两个API来实现多源搜索功能。如果没安装对应的python依赖可以运行下面这条脚本pip install hello-agents[search]0.1.11搜索工具的统一接口设计HelloAgents框架内置的SearchTool展示了如何设计一个高级的多源搜索工具classSearchTool(Tool): 智能混合搜索工具 支持多种搜索引擎后端智能选择最佳搜索源: 1. 混合模式 (hybrid) - 智能选择TAVILY或SERPAPI 2. Tavily API (tavily) - 专业AI搜索 3. SerpApi (serpapi) - 传统Google搜索 def__init__(self, backend: str hybrid, tavily_key: Optional[str] None, serpapi_key: Optional[str] None): super().__init__( namesearch, description一个智能网页搜索引擎。支持混合搜索模式自动选择最佳搜索源。 ) self.backend backend self.tavily_key tavily_key or os.getenv(TAVILY_API_KEY) self.serpapi_key serpapi_key or os.getenv(SERPAPI_API_KEY) self.available_backends [] self._setup_backends()这个设计的核心思想是根据可用的API密钥和依赖库自动选择最佳的搜索后端。2TAVILY与SERPAPI搜索源的整合策略框架实现了智能的后端选择逻辑def_search_hybrid(self, query: str) - str: 混合搜索 - 智能选择最佳搜索源 # 优先使用TavilyAI优化的搜索 iftavilyin self.available_backends: try: return self._search_tavily(query) except Exception as e: print(f⚠️ Tavily搜索失败: {e}) # 如果Tavily失败尝试SerpApi ifserpapiin self.available_backends: print( 切换到SerpApi搜索) return self._search_serpapi(query) # 如果Tavily不可用使用SerpApi elifserpapiin self.available_backends: try: return self._search_serpapi(query) except Exception as e: print(f⚠️ SerpApi搜索失败: {e}) # 如果都不可用提示用户配置API return❌ 没有可用的搜索源请配置TAVILY_API_KEY或SERPAPI_API_KEY环境变量这种设计体现了高可用系统的核心理念通过降级机制系统能够从最优的搜索源逐步降级到可用的备选方案。当所有搜索源都不可用时明确提示用户配置正确的API密钥。3搜索结果的统一格式化不同搜索引擎返回的结果格式不同框架通过统一的格式化方法来处理def_search_tavily(self, query: str) - str: 使用Tavily搜索 response self.tavily_client.search( queryquery, search_depthbasic, include_answerTrue, max_results3 ) result f Tavily AI搜索结果:{response.get(answer, 未找到直接答案)}\n\n for i, item in enumerate(response.get(results, [])[:3], 1): result f[{i}] {item.get(title, )}\n result f {item.get(content, )[:200]}...\n result f 来源: {item.get(url, )}\n\n return result基于框架的设计思想我们可以创建自己的高级搜索工具。这次我们使用类的方式来展示不同的实现方法创建my_advanced_search.py# my_advanced_search.py import os from typing import Optional, List, Dict, Any from hello_agents import ToolRegistry classMyAdvancedSearchTool: 自定义高级搜索工具类 展示多源整合和智能选择的设计模式 def__init__(self): self.name my_advanced_search self.description 智能搜索工具支持多个搜索源自动选择最佳结果 self.search_sources [] self._setup_search_sources() def_setup_search_sources(self): 设置可用的搜索源 # 检查Tavily可用性 if os.getenv(TAVILY_API_KEY): try: from tavily import TavilyClient self.tavily_client TavilyClient(api_keyos.getenv(TAVILY_API_KEY)) self.search_sources.append(tavily) print(✅ Tavily搜索源已启用) except ImportError: print(⚠️ Tavily库未安装) # 检查SerpApi可用性 if os.getenv(SERPAPI_API_KEY): try: import serpapi self.search_sources.append(serpapi) print(✅ SerpApi搜索源已启用) except ImportError: print(⚠️ SerpApi库未安装) if self.search_sources: print(f 可用搜索源: {, .join(self.search_sources)}) else: print(⚠️ 没有可用的搜索源请配置API密钥) defsearch(self, query: str) - str: 执行智能搜索 ifnot query.strip(): return❌ 错误:搜索查询不能为空 # 检查是否有可用的搜索源 ifnot self.search_sources: return❌ 没有可用的搜索源请配置以下API密钥之一: 1. Tavily API: 设置环境变量 TAVILY_API_KEY 获取地址: https://tavily.com/ 2. SerpAPI: 设置环境变量 SERPAPI_API_KEY 获取地址: https://serpapi.com/ 配置后重新运行程序。 print(f 开始智能搜索: {query}) # 尝试多个搜索源返回最佳结果 for source in self.search_sources: try: if source tavily: result self._search_with_tavily(query) if result and未找到notin result: returnf Tavily AI搜索结果:\n\n{result} elif source serpapi: result self._search_with_serpapi(query) if result and未找到notin result: returnf SerpApi Google搜索结果:\n\n{result} except Exception as e: print(f⚠️ {source} 搜索失败: {e}) continue return❌ 所有搜索源都失败了请检查网络连接和API密钥配置 def_search_with_tavily(self, query: str) - str: 使用Tavily搜索 response self.tavily_client.search(queryquery, max_results3) if response.get(answer): result f AI直接答案:{response[answer]}\n\n else: result result 相关结果:\n for i, item in enumerate(response.get(results, [])[:3], 1): result f[{i}] {item.get(title, )}\n result f {item.get(content, )[:150]}...\n\n return result def_search_with_serpapi(self, query: str) - str: 使用SerpApi搜索 import serpapi search serpapi.GoogleSearch({ q: query, api_key: os.getenv(SERPAPI_API_KEY), num: 3 }) results search.get_dict() result Google搜索结果:\n iforganic_resultsin results: for i, res in enumerate(results[organic_results][:3], 1): result f[{i}] {res.get(title, )}\n result f {res.get(snippet, )}\n\n return result defcreate_advanced_search_registry(): 创建包含高级搜索工具的注册表 registry ToolRegistry() # 创建搜索工具实例 search_tool MyAdvancedSearchTool() # 注册搜索工具的方法作为函数 registry.register_function( nameadvanced_search, description高级搜索工具整合Tavily和SerpAPI多个搜索源提供更全面的搜索结果, funcsearch_tool.search ) return registry接下来可以测试我们自己编写的工具创建test_advanced_search.py# test_advanced_search.py from dotenv import load_dotenv from my_advanced_search import create_advanced_search_registry, MyAdvancedSearchTool # 加载环境变量 load_dotenv() deftest_advanced_search(): 测试高级搜索工具 # 创建包含高级搜索工具的注册表 registry create_advanced_search_registry() print( 测试高级搜索工具\n) # 测试查询 test_queries [ Python编程语言的历史, 人工智能的最新发展, 2024年科技趋势 ] for i, query in enumerate(test_queries, 1): print(f测试 {i}: {query}) result registry.execute_tool(advanced_search, query) print(f结果: {result}\n) print(- * 60 \n) deftest_api_configuration(): 测试API配置检查 print( 测试API配置检查:) # 直接创建搜索工具实例 search_tool MyAdvancedSearchTool() # 如果没有配置API会显示配置提示 result search_tool.search(机器学习算法) print(f搜索结果: {result}) deftest_with_agent(): 测试与Agent的集成 print(\n 与Agent集成测试:) print(高级搜索工具已准备就绪可以与Agent集成使用) # 显示工具描述 registry create_advanced_search_registry() tools_desc registry.get_tools_description() print(f工具描述:\n{tools_desc}) if __name__ __main__: test_advanced_search() test_api_configuration() test_with_agent()通过这个高级搜索工具的设计实践我们学会了如何使用类的方式来构建复杂的工具系统。相比函数方式类方式更适合需要维护状态如API客户端、配置信息的工具。4️⃣ 工具系统的高级特性在掌握了基础的工具开发和多源整合后我们来探讨工具系统的高级特性。这些特性能够让工具系统在复杂的生产环境中稳定运行并为Agent提供更强大的能力。1工具链式调用机制在实际应用中Agent经常需要组合使用多个工具来完成复杂任务。我们可以设计一个工具链管理器来支持这种场景这里借鉴了提到的图的概念# tool_chain_manager.py from typing import List, Dict, Any, Optional from hello_agents import ToolRegistry classToolChain: 工具链 - 支持多个工具的顺序执行 def__init__(self, name: str, description: str): self.name name self.description description self.steps: List[Dict[str, Any]] [] defadd_step(self, tool_name: str, input_template: str, output_key: str None): 添加工具执行步骤 Args: tool_name: 工具名称 input_template: 输入模板支持变量替换 output_key: 输出结果的键名用于后续步骤引用 self.steps.append({ tool_name: tool_name, input_template: input_template, output_key: output_key orfstep_{len(self.steps)}_result }) defexecute(self, registry: ToolRegistry, initial_input: str, context: Dict[str, Any] None) - str: 执行工具链 context context or {} context[input] initial_input print(f 开始执行工具链: {self.name}) for i, step in enumerate(self.steps, 1): tool_name step[tool_name] input_template step[input_template] output_key step[output_key] # 替换模板中的变量 try: tool_input input_template.format(**context) except KeyError as e: returnf❌ 工具链执行失败:模板变量 {e} 未找到 print(f 步骤 {i}: 使用 {tool_name} 处理 {tool_input[:50]}...) # 执行工具 result registry.execute_tool(tool_name, tool_input) context[output_key] result print(f ✅ 步骤 {i} 完成结果长度: {len(result)} 字符) # 返回最后一步的结果 final_result context[self.steps[-1][output_key]] print(f 工具链 {self.name} 执行完成) return final_result classToolChainManager: 工具链管理器 def__init__(self, registry: ToolRegistry): self.registry registry self.chains: Dict[str, ToolChain] {} defregister_chain(self, chain: ToolChain): 注册工具链 self.chains[chain.name] chain print(f✅ 工具链 {chain.name} 已注册) defexecute_chain(self, chain_name: str, input_data: str, context: Dict[str, Any] None) - str: 执行指定的工具链 if chain_name notin self.chains: returnf❌ 工具链 {chain_name} 不存在 chain self.chains[chain_name] return chain.execute(self.registry, input_data, context) deflist_chains(self) - List[str]: 列出所有工具链 return list(self.chains.keys()) # 使用示例 defcreate_research_chain() - ToolChain: 创建一个研究工具链:搜索 - 计算 - 总结 chain ToolChain( nameresearch_and_calculate, description搜索信息并进行相关计算 ) # 步骤1:搜索信息 chain.add_step( tool_namesearch, input_template{input}, output_keysearch_result ) # 步骤2:基于搜索结果进行计算如果需要 chain.add_step( tool_namemy_calculator, input_template根据以下信息计算相关数值:{search_result}, output_keycalculation_result ) return chain2异步工具执行支持对于耗时的工具操作我们可以提供异步执行支持# async_tool_executor.py import asyncio import concurrent.futures from typing import Dict, Any, List, Callable from hello_agents import ToolRegistry classAsyncToolExecutor: 异步工具执行器 def__init__(self, registry: ToolRegistry, max_workers: int 4): self.registry registry self.executor concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) asyncdefexecute_tool_async(self, tool_name: str, input_data: str) - str: 异步执行单个工具 loop asyncio.get_event_loop() def_execute(): return self.registry.execute_tool(tool_name, input_data) result await loop.run_in_executor(self.executor, _execute) return result asyncdefexecute_tools_parallel(self, tasks: List[Dict[str, str]]) - List[str]: 并行执行多个工具 print(f 开始并行执行 {len(tasks)} 个工具任务) # 创建异步任务 async_tasks [] for task in tasks: tool_name task[tool_name] input_data task[input_data] async_task self.execute_tool_async(tool_name, input_data) async_tasks.append(async_task) # 等待所有任务完成 results await asyncio.gather(*async_tasks) print(f✅ 所有工具任务执行完成) return results def__del__(self): 清理资源 if hasattr(self, executor): self.executor.shutdown(waitTrue) # 使用示例 asyncdeftest_parallel_execution(): 测试并行工具执行 from hello_agents import ToolRegistry registry ToolRegistry() # 假设已经注册了搜索和计算工具 executor AsyncToolExecutor(registry) # 定义并行任务 tasks [ {tool_name: search, input_data: Python编程}, {tool_name: search, input_data: 机器学习}, {tool_name: my_calculator, input_data: 2 2}, {tool_name: my_calculator, input_data: sqrt(16)}, ] # 并行执行 results await executor.execute_tools_parallel(tasks) for i, result in enumerate(results): print(f任务 {i1} 结果: {result[:100]}...)基于以上的设计和实现经验我们可以总结出工具系统开发的核心理念在设计层面每个工具都应该遵循单一职责原则专注于特定功能的同时保持接口的统一性并将完善的异常处理和安全优先的输入验证作为基本要求。在性能优化方面利用异步执行提高并发处理能力同时合理管理外部连接和系统资源。本文小结回顾本文我们完成了一项富有挑战的任务一步步构建了一个基础的智能体框架——HelloAgents。这个过程始终遵循着“分层解耦、职责单一、接口统一”的核心原则。在框架的具体实现中我们再次实现了四种经典的Agent范式。从SimpleAgent的基础对话模式到ReActAgent的推理与行动结合从ReflectionAgent的自我反思与迭代优化到PlanAndSolveAgent的分解规划与逐步执行。而工具系统作为Agent能力延伸的核心其构建过程则是一次完整的工程实践。ecute_tool(tool_name, input_data)result await loop.run_in_executor(self.executor, _execute)return resultasyncdefexecute_tools_parallel(self, tasks: List[Dict[str, str]]) - List[str]:“”“并行执行多个工具”“”print(f 开始并行执行 {len(tasks)} 个工具任务)创建异步任务async_tasks []for task in tasks:tool_name task[“tool_name”]input_data task[“input_data”]async_task self.execute_tool_async(tool_name, input_data)async_tasks.append(async_task)等待所有任务完成results await asyncio.gather(*async_tasks) print(f✅ 所有工具任务执行完成)return resultsdef__del__(self):“”“清理资源”“”if hasattr(self, ‘executor’):self.executor.shutdown(waitTrue)使用示例asyncdeftest_parallel_execution():“”“测试并行工具执行”“”from hello_agents import ToolRegistryregistry ToolRegistry()假设已经注册了搜索和计算工具executor AsyncToolExecutor(registry)定义并行任务tasks [ {tool_name: search, input_data: Python编程}, {tool_name: search, input_data: 机器学习}, {tool_name: my_calculator, input_data: 2 2}, {tool_name: my_calculator, input_data: sqrt(16)}, ]并行执行results await executor.execute_tools_parallel(tasks)for i, result in enumerate(results):print(f任务 {i1} 结果: {result[:100]}…)基于以上的设计和实现经验我们可以总结出工具系统开发的核心理念在设计层面每个工具都应该遵循单一职责原则专注于特定功能的同时保持接口的统一性并将完善的异常处理和安全优先的输入验证作为基本要求。在性能优化方面利用异步执行提高并发处理能力同时合理管理外部连接和系统资源。本文小结回顾本文我们完成了一项富有挑战的任务一步步构建了一个基础的智能体框架——HelloAgents。这个过程始终遵循着“分层解耦、职责单一、接口统一”的核心原则。在框架的具体实现中我们再次实现了四种经典的Agent范式。从SimpleAgent的基础对话模式到ReActAgent的推理与行动结合从ReflectionAgent的自我反思与迭代优化到PlanAndSolveAgent的分解规划与逐步执行。而工具系统作为Agent能力延伸的核心其构建过程则是一次完整的工程实践。在大模型时代我们如何有效的去学习大模型现如今大模型岗位需求越来越大但是相关岗位人才难求薪资持续走高AI运营薪资平均值约18457元AI工程师薪资平均值约37336元大模型算法薪资平均值约39607元。掌握大模型技术你还能拥有更多可能性• 成为一名全栈大模型工程师包括PromptLangChainLoRA等技术开发、运营、产品等方向全栈工程• 能够拥有模型二次训练和微调能力带领大家完成智能对话、文生图等热门应用• 薪资上浮10%-20%覆盖更多高薪岗位这是一个高需求、高待遇的热门方向和领域• 更优质的项目可以为未来创新创业提供基石。《AI大模型从0到精通全套学习包》如果你想要提升自己的能力却又没有方向想学大模型技术去帮助就业和转行又不知道怎么开始那么这一套**《AI大模型零基础入门到实战全套学习大礼包》以及《大模型应用开发视频教程》**一定可以帮助到你限免0元1全套AI大模型应用开发视频教程包含深度学习、提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点2大模型入门到实战全套学习大礼包01大模型系统化学习路线作为学习AI大模型技术的新手方向至关重要。 正确的学习路线可以为你节省时间少走弯路方向不对努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划带你从零基础入门到精通02大模型学习书籍文档学习AI大模型离不开书籍文档我精选了一系列大模型技术的书籍和学习文档电子版它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。03AI大模型最新行业报告2025最新行业报告针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。04大模型项目实战配套源码学以致用在项目实战中检验和巩固你所学到的知识同时为你找工作就业和职业发展打下坚实的基础。05大模型大厂面试真题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我精心整理了一份大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。*这些资料真的有用吗*这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。06以上全套大模型资料如何领取用微信加上就会给你发无偿分享遇到扫码问题可以私信或评论区找我