Try Live
Add Docs
Rankings
Pricing
Docs
Install
Theme
Install
Docs
Pricing
More...
More...
Try Live
Rankings
Enterprise
Create API Key
Add Docs
Hello-Agents
https://github.com/datawhalechina/hello-agents
Admin
Hello-Agents is a comprehensive tutorial that teaches how to build AI-native intelligent agent
...
Tokens:
722,298
Snippets:
3,627
Trust Score:
9.3
Update:
1 month ago
Context
Skills
Chat
Benchmark
83.2
Suggestions
Latest
Show doc for...
Code
Info
Show Results
Context Summary (auto-generated)
Raw
Copy
Link
# Hello-Agents Hello-Agents 是一个全面系统的 AI 智能体开发教程,由 Datawhale 社区开源项目团队打造。本教程从基础理论到实战应用,系统性地教授如何从零构建 AI 原生智能体系统,涵盖核心原理、架构模式、框架开发以及多智能体应用的实践实现。 教程共分 16 章,内容包括:智能体基础概念与发展历史、大语言模型原理、经典智能体范式(ReAct、Plan-and-Solve、Reflection)、低代码平台搭建、主流框架实践(LangChain、LlamaIndex、AgentScope)、自建框架 HelloAgents、记忆与检索增强(RAG)、上下文工程、通信协议(MCP/A2A/ANP)、强化学习训练、性能评估,以及三个完整案例项目:智能旅行助手、深度研究智能体和赛博小镇模拟。 --- ## HelloAgentsLLM 统一 LLM 客户端 HelloAgentsLLM 是框架的核心组件,提供统一的大语言模型调用接口,支持多供应商(OpenAI、ModelScope、智谱AI)和本地模型(VLLM、Ollama)的无缝切换,具备自动检测机制。 ```python import os from openai import OpenAI from dotenv import load_dotenv from typing import List, Dict load_dotenv() class HelloAgentsLLM: """ 为 Hello Agents 定制的 LLM 客户端。 支持任何兼容 OpenAI 接口的服务,默认使用流式响应。 """ def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None): self.model = model or os.getenv("LLM_MODEL_ID") apiKey = apiKey or os.getenv("LLM_API_KEY") baseUrl = baseUrl or os.getenv("LLM_BASE_URL") timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60)) if not all([self.model, apiKey, baseUrl]): raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。") self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout) def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str: """调用大语言模型进行思考,返回流式响应""" response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=temperature, stream=True, ) collected_content = [] for chunk in response: content = chunk.choices[0].delta.content or "" print(content, end="", flush=True) collected_content.append(content) print() return "".join(collected_content) # 使用示例 llmClient = HelloAgentsLLM() messages = [ {"role": "system", "content": "你是一个有用的AI助手。"}, {"role": "user", "content": "写一个快速排序算法"} ] response = llmClient.think(messages) ``` --- ## ReActAgent 推理与行动智能体 ReActAgent 实现了经典的 ReAct(Reasoning + Acting)范式,通过思考-行动-观察的循环来解决复杂任务,支持工具调用和多步推理。 ```python import re from llm_client import HelloAgentsLLM from tools import ToolExecutor, search REACT_PROMPT_TEMPLATE = """ 请注意,你是一个有能力调用外部工具的智能助手。 可用工具如下: {tools} 请严格按照以下格式进行回应: Thought: 你的思考过程,用于分析问题、拆解任务和规划下一步行动。 Action: 你决定采取的行动,必须是以下格式之一: - `{{tool_name}}[{{tool_input}}]`:调用一个可用工具。 - `Finish[最终答案]`:当你认为已经获得最终答案时。 现在,请开始解决以下问题: Question: {question} History: {history} """ class ReActAgent: def __init__(self, llm_client: HelloAgentsLLM, tool_executor: ToolExecutor, max_steps: int = 5): self.llm_client = llm_client self.tool_executor = tool_executor self.max_steps = max_steps self.history = [] def run(self, question: str): self.history = [] current_step = 0 while current_step < self.max_steps: current_step += 1 print(f"\n--- 第 {current_step} 步 ---") tools_desc = self.tool_executor.getAvailableTools() history_str = "\n".join(self.history) prompt = REACT_PROMPT_TEMPLATE.format(tools=tools_desc, question=question, history=history_str) messages = [{"role": "user", "content": prompt}] response_text = self.llm_client.think(messages=messages) thought, action = self._parse_output(response_text) if thought: print(f"🤔 思考: {thought}") if action.startswith("Finish"): final_answer = self._parse_action_input(action) print(f"🎉 最终答案: {final_answer}") return final_answer tool_name, tool_input = self._parse_action(action) print(f"🎬 行动: {tool_name}[{tool_input}]") tool_function = self.tool_executor.getTool(tool_name) observation = tool_function(tool_input) if tool_function else f"错误:未找到工具" print(f"👀 观察: {observation}") self.history.append(f"Action: {action}") self.history.append(f"Observation: {observation}") return None # 使用示例 llm = HelloAgentsLLM() tool_executor = ToolExecutor() tool_executor.registerTool("Search", "网页搜索引擎", search) agent = ReActAgent(llm_client=llm, tool_executor=tool_executor) agent.run("华为最新的手机是哪一款?它的主要卖点是什么?") ``` --- ## SimpleAgent 基础对话智能体 SimpleAgent 是最基础的智能体实现,支持基础对话、可选工具调用和流式响应,是构建更复杂智能体的基础。 ```python from hello_agents import SimpleAgent, HelloAgentsLLM, ToolRegistry from hello_agents.tools import CalculatorTool from dotenv import load_dotenv load_dotenv() # 创建 LLM 实例 - 框架自动检测 provider llm = HelloAgentsLLM() # 创建 SimpleAgent agent = SimpleAgent( name="AI助手", llm=llm, system_prompt="你是一个有用的AI助手" ) # 基础对话 response = agent.run("你好!请介绍一下自己") print(response) # 添加工具功能 tool_registry = ToolRegistry() calculator = CalculatorTool() tool_registry.register_tool(calculator) agent.tool_registry = tool_registry # 使用工具进行计算 response = agent.run("请帮我计算 2 + 3 * 4") print(response) # 查看对话历史 print(f"历史消息数: {len(agent.get_history())}") ``` --- ## ToolRegistry 工具注册系统 ToolRegistry 提供统一的工具管理机制,支持工具注册、发现和执行,是智能体工具系统的核心组件。 ```python from hello_agents import ToolRegistry # 创建工具注册表 registry = ToolRegistry() # 方式1:注册 Tool 对象 from hello_agents.tools import CalculatorTool, SearchTool calculator = CalculatorTool() registry.register_tool(calculator) # 方式2:直接注册函数作为工具 def my_calculate(expression: str) -> str: """简单的数学计算函数""" try: result = eval(expression) return str(result) except: return "计算失败" registry.register_function( name="my_calculator", description="简单的数学计算工具,支持基本运算(+,-,*,/)", func=my_calculate ) # 获取工具描述(用于构建提示词) tools_desc = registry.get_tools_description() print(f"可用工具:\n{tools_desc}") # 执行工具 result = registry.execute_tool("my_calculator", "15 * 8 + 32") print(f"计算结果: {result}") # 列出所有工具 print(f"已注册工具: {registry.list_tools()}") ``` --- ## MemoryTool 记忆工具 MemoryTool 为智能体提供记忆能力,支持工作记忆、情景记忆、语义记忆和感知记忆四种类型,模拟人类认知系统的记忆机制。 ```python from hello_agents import SimpleAgent, HelloAgentsLLM, ToolRegistry from hello_agents.tools import MemoryTool # 创建具有记忆能力的 Agent llm = HelloAgentsLLM() agent = SimpleAgent(name="记忆助手", llm=llm) # 创建记忆工具 memory_tool = MemoryTool(user_id="user123") tool_registry = ToolRegistry() tool_registry.register_tool(memory_tool) agent.tool_registry = tool_registry # 添加语义记忆(抽象知识) memory_tool.execute("add", content="用户张三是一名Python开发者,专注于机器学习", memory_type="semantic", importance=0.8 ) # 添加情景记忆(具体事件) memory_tool.execute("add", content="2024年3月15日,用户完成了第一个AI项目", memory_type="episodic", importance=0.7 ) # 添加工作记忆(临时信息) memory_tool.execute("add", content="用户正在询问关于深度学习的问题", memory_type="working", importance=0.6 ) # 搜索记忆 result = memory_tool.execute("search", query="Python开发", limit=5) print(f"搜索结果: {result}") # 获取记忆摘要 summary = memory_tool.execute("summary") print(f"记忆摘要: {summary}") # 记忆整合(短期转长期) memory_tool.execute("consolidate") ``` --- ## RAGTool 检索增强生成工具 RAGTool 实现检索增强生成(RAG)功能,从外部知识库检索相关信息作为上下文,提升智能体回答的准确性和时效性。 ```python from hello_agents import SimpleAgent, HelloAgentsLLM, ToolRegistry from hello_agents.tools import RAGTool # 创建 RAG 工具 rag_tool = RAGTool(knowledge_base_path="./knowledge_base") tool_registry = ToolRegistry() tool_registry.register_tool(rag_tool) # 索引文档 rag_tool.run({ "action": "index", "file_path": "./docs/technical_guide.pdf" }) # 搜索相关知识 result = rag_tool.run({ "action": "search", "query": "如何配置智能体的记忆系统?", "limit": 5, "min_score": 0.3 }) print(f"检索结果: {result}") # 与 Agent 集成使用 llm = HelloAgentsLLM() agent = SimpleAgent( name="知识助手", llm=llm, system_prompt="你是一个有知识检索能力的AI助手" ) agent.tool_registry = tool_registry response = agent.run("请解释智能体的记忆系统是如何工作的?") print(response) ``` --- ## MCPTool MCP 协议工具 MCPTool 实现 Model Context Protocol(MCP)协议,提供智能体与外部工具/资源的标准化通信方式,支持多种 MCP 服务器连接。 ```python from hello_agents import SimpleAgent, HelloAgentsLLM from hello_agents.tools import MCPTool agent = SimpleAgent(name="助手", llm=HelloAgentsLLM()) # 方式1:使用内置演示服务器 mcp_tool = MCPTool() # 默认提供 add, subtract, multiply, divide 等工具 agent.add_tool(mcp_tool) # 智能体可以使用内置工具 response = agent.run("计算 123 + 456") print(response) # 方式2:连接外部 MCP 服务器 # 连接文件系统服务器 fs_tool = MCPTool( name="filesystem", description="访问本地文件系统", server_command=["npx", "-y", "@modelcontextprotocol/server-filesystem", "."] ) agent.add_tool(fs_tool) # 连接 GitHub 服务器 github_tool = MCPTool( name="github", description="访问GitHub仓库", server_command=["npx", "-y", "@modelcontextprotocol/server-github"] ) agent.add_tool(github_tool) # 直接调用 MCP 工具 result = mcp_tool.run({ "action": "call_tool", "tool_name": "add", "arguments": {"a": 10, "b": 20} }) print(f"MCP计算结果: {result}") # 输出: 30.0 # Agent 现在可以自动使用这些工具 response = agent.run("请读取 README.md 文件的内容") print(response) ``` --- ## ContextBuilder 上下文构建器 ContextBuilder 实现 GSSC(Gather-Select-Structure-Compress)流水线,用于在推理阶段策划和维护最优的信息集合,解决上下文窗口有限的问题。 ```python from hello_agents.context import ContextBuilder, ContextConfig, ContextPacket from hello_agents.tools import MemoryTool, RAGTool from datetime import datetime # 配置上下文构建器 config = ContextConfig( max_tokens=3000, # 最大 token 数量 reserve_ratio=0.2, # 为系统指令预留的比例 min_relevance=0.1, # 最低相关性阈值 enable_compression=True, # 启用压缩 recency_weight=0.3, # 新近性权重 relevance_weight=0.7 # 相关性权重 ) # 创建上下文构建器 memory_tool = MemoryTool(user_id="user123") rag_tool = RAGTool(knowledge_base_path="./knowledge") builder = ContextBuilder( config=config, memory_tool=memory_tool, rag_tool=rag_tool ) # 构建上下文 context = builder.build( user_query="如何实现智能体的记忆系统?", system_instructions="你是一个AI开发专家助手", conversation_history=[] ) print(f"构建的上下文:\n{context}") # 添加自定义信息包 custom_packet = ContextPacket( content="用户偏好:简洁的代码示例", timestamp=datetime.now(), token_count=10, relevance_score=0.8, metadata={"type": "user_preference"} ) context_with_custom = builder.build( user_query="写一个记忆存储的示例", custom_packets=[custom_packet] ) ``` --- ## Plan-and-Solve 规划求解智能体 PlanAndSolveAgent 实现规划与求解分离的范式,先将复杂问题分解为子任务计划,再逐步执行每个子任务。 ```python from hello_agents import HelloAgentsLLM DEFAULT_PLANNER_PROMPT = """ 你是一个顶级的AI规划专家。你的任务是将用户提出的复杂问题分解成一个由多个简单步骤组成的行动计划。 请确保计划中的每个步骤都是一个独立的、可执行的子任务,并且严格按照逻辑顺序排列。 问题: {question} 请严格按照以下格式输出你的计划: ```python ["步骤1", "步骤2", "步骤3", ...] ``` """ DEFAULT_EXECUTOR_PROMPT = """ 你是一位顶级的AI执行专家。请严格按照给定的计划,一步步地解决问题。 # 原始问题: {question} # 完整计划: {plan} # 历史步骤与结果: {history} # 当前步骤: {current_step} 请仅输出针对"当前步骤"的回答: """ class PlanAndSolveAgent: def __init__(self, llm: HelloAgentsLLM): self.llm = llm def run(self, question: str) -> str: # 步骤1:生成计划 planner_prompt = DEFAULT_PLANNER_PROMPT.format(question=question) plan_response = self.llm.think([{"role": "user", "content": planner_prompt}]) plan = eval(plan_response) # 解析 Python 列表 print(f"📋 生成计划: {plan}") # 步骤2:逐步执行 history = [] for i, step in enumerate(plan): print(f"\n▶️ 执行步骤 {i+1}: {step}") executor_prompt = DEFAULT_EXECUTOR_PROMPT.format( question=question, plan=plan, history="\n".join(history), current_step=step ) result = self.llm.think([{"role": "user", "content": executor_prompt}]) history.append(f"步骤{i+1}: {step}\n结果: {result}") print(f"✅ 结果: {result}") return history[-1] # 使用示例 llm = HelloAgentsLLM() agent = PlanAndSolveAgent(llm=llm) result = agent.run("一个水果店周一卖出了15个苹果。周二卖出的苹果数量是周一的两倍。周三卖出的数量比周二少了5个。请问这三天总共卖出了多少个苹果?") print(f"\n最终结果: {result}") ``` --- ## Reflection 自我反思智能体 ReflectionAgent 实现执行-反思-优化的循环,通过自我评估和迭代改进来提升输出质量。 ```python from hello_agents import HelloAgentsLLM DEFAULT_PROMPTS = { "initial": """请根据以下要求完成任务: 任务: {task} 请提供一个完整、准确的回答。 """, "reflect": """请仔细审查以下回答,并找出可能的问题或改进空间: # 原始任务: {task} # 当前回答: {content} 请分析这个回答的质量,指出不足之处,并提出具体的改进建议。 如果回答已经很好,请回答"无需改进"。 """, "refine": """请根据反馈意见改进你的回答: # 原始任务: {task} # 上一轮回答: {last_attempt} # 反馈意见: {feedback} 请提供一个改进后的回答。 """ } class ReflectionAgent: def __init__(self, llm: HelloAgentsLLM, max_iterations: int = 3): self.llm = llm self.max_iterations = max_iterations def run(self, task: str) -> str: # 初始生成 initial_prompt = DEFAULT_PROMPTS["initial"].format(task=task) current_content = self.llm.think([{"role": "user", "content": initial_prompt}]) print(f"📝 初始回答:\n{current_content}") for i in range(self.max_iterations): print(f"\n🔄 反思迭代 {i+1}") # 反思阶段 reflect_prompt = DEFAULT_PROMPTS["reflect"].format( task=task, content=current_content ) feedback = self.llm.think([{"role": "user", "content": reflect_prompt}]) print(f"💭 反馈: {feedback[:200]}...") if "无需改进" in feedback: print("✅ 回答已达到满意质量") break # 优化阶段 refine_prompt = DEFAULT_PROMPTS["refine"].format( task=task, last_attempt=current_content, feedback=feedback ) current_content = self.llm.think([{"role": "user", "content": refine_prompt}]) print(f"📝 改进后回答:\n{current_content[:300]}...") return current_content # 使用示例 llm = HelloAgentsLLM() agent = ReflectionAgent(llm=llm, max_iterations=3) result = agent.run("写一篇关于人工智能发展历程的简短文章") print(f"\n最终结果:\n{result}") ``` --- ## Message 消息系统 Message 类定义了框架内统一的消息格式,确保智能体与模型之间信息传递的标准化,支持用户、助手、系统和工具四种角色。 ```python from typing import Optional, Dict, Any, Literal from datetime import datetime from pydantic import BaseModel MessageRole = Literal["user", "assistant", "system", "tool"] class Message(BaseModel): """消息类""" content: str role: MessageRole timestamp: datetime = None metadata: Optional[Dict[str, Any]] = None def __init__(self, content: str, role: MessageRole, **kwargs): super().__init__( content=content, role=role, timestamp=kwargs.get('timestamp', datetime.now()), metadata=kwargs.get('metadata', {}) ) def to_dict(self) -> Dict[str, Any]: """转换为字典格式(OpenAI API格式)""" return {"role": self.role, "content": self.content} def __str__(self) -> str: return f"[{self.role}] {self.content}" # 使用示例 user_msg = Message("你好,请介绍一下自己", "user") assistant_msg = Message("我是一个AI助手", "assistant") system_msg = Message("你是一个有用的助手", "system") print(user_msg.to_dict()) # 输出: {'role': 'user', 'content': '你好,请介绍一下自己'} # 构建对话历史 history = [user_msg, assistant_msg] messages = [msg.to_dict() for msg in history] ``` --- ## Agent 抽象基类 Agent 基类定义了所有智能体必须遵循的接口规范,提供统一的初始化参数、方法签名和历史管理机制。 ```python from abc import ABC, abstractmethod from typing import Optional, Any, List from hello_agents.core.message import Message from hello_agents.core.llm import HelloAgentsLLM from hello_agents.core.config import Config class Agent(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 def run(self, input_text: str, **kwargs) -> str: """运行 Agent - 子类必须实现""" pass def add_message(self, message: Message): """添加消息到历史记录""" self._history.append(message) def clear_history(self): """清空历史记录""" self._history.clear() def get_history(self) -> List[Message]: """获取历史记录""" return self._history.copy() def __str__(self) -> str: return f"Agent(name={self.name}, provider={self.llm.provider})" # 创建自定义 Agent class MyCustomAgent(Agent): def run(self, input_text: str, **kwargs) -> str: messages = [] if self.system_prompt: messages.append({"role": "system", "content": self.system_prompt}) for msg in self._history: messages.append(msg.to_dict()) messages.append({"role": "user", "content": input_text}) response = self.llm.invoke(messages, **kwargs) self.add_message(Message(input_text, "user")) self.add_message(Message(response, "assistant")) return response ``` --- ## FunctionCallAgent 原生函数调用智能体 FunctionCallAgent 基于 OpenAI 原生函数调用机制,提供更强鲁棒性的工具调用能力,自动构建函数调用 schema。 ```python from hello_agents import FunctionCallAgent, HelloAgentsLLM, ToolRegistry from hello_agents.tools import CalculatorTool, SearchTool # 创建工具注册表 tool_registry = ToolRegistry() tool_registry.register_tool(CalculatorTool()) tool_registry.register_tool(SearchTool()) # 创建 FunctionCallAgent llm = HelloAgentsLLM() agent = FunctionCallAgent( name="函数调用助手", llm=llm, tool_registry=tool_registry, system_prompt="你是一个能够调用工具的智能助手" ) # 运行智能体 response = agent.run("请计算 15 * 8 + 32 的结果") print(response) # 工具的 OpenAI Schema 会自动生成 for tool in tool_registry.list_tools(): schema = tool_registry.get_tool(tool).to_openai_schema() print(f"工具 {tool} 的 Schema: {schema}") ``` --- ## 异步工具执行器 AsyncToolExecutor 提供异步工具执行支持,支持并行执行多个工具任务,提高耗时操作的处理效率。 ```python import asyncio import concurrent.futures from typing import Dict, List from hello_agents import ToolRegistry class AsyncToolExecutor: """异步工具执行器""" def __init__(self, registry: ToolRegistry, max_workers: int = 4): self.registry = registry self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) async def execute_tool_async(self, tool_name: str, input_data: str) -> str: """异步执行单个工具""" loop = asyncio.get_event_loop() result = await loop.run_in_executor( self.executor, lambda: self.registry.execute_tool(tool_name, input_data) ) return result async def execute_tools_parallel(self, tasks: List[Dict[str, str]]) -> List[str]: """并行执行多个工具""" print(f"🚀 开始并行执行 {len(tasks)} 个工具任务") async_tasks = [ self.execute_tool_async(task["tool_name"], task["input_data"]) for task in tasks ] results = await asyncio.gather(*async_tasks) print(f"✅ 所有工具任务执行完成") return results # 使用示例 async def main(): registry = ToolRegistry() # 注册工具... executor = AsyncToolExecutor(registry) tasks = [ {"tool_name": "search", "input_data": "Python编程"}, {"tool_name": "search", "input_data": "机器学习"}, {"tool_name": "calculator", "input_data": "2 + 2"}, ] results = await executor.execute_tools_parallel(tasks) for i, result in enumerate(results): print(f"任务 {i+1} 结果: {result[:100]}...") asyncio.run(main()) ``` --- ## 工具链管理器 ToolChain 支持多个工具的顺序执行,实现复杂任务的工具编排和结果传递。 ```python from typing import List, Dict, Any from hello_agents import ToolRegistry class ToolChain: """工具链 - 支持多个工具的顺序执行""" def __init__(self, name: str, description: str): self.name = name self.description = description self.steps: List[Dict[str, Any]] = [] def add_step(self, tool_name: str, input_template: str, output_key: str = None): """添加工具执行步骤""" self.steps.append({ "tool_name": tool_name, "input_template": input_template, "output_key": output_key or f"step_{len(self.steps)}_result" }) def execute(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_input = step["input_template"].format(**context) print(f" 步骤 {i}: 使用 {step['tool_name']} 处理") result = registry.execute_tool(step["tool_name"], tool_input) context[step["output_key"]] = result return context[self.steps[-1]["output_key"]] # 使用示例:创建研究工具链 chain = ToolChain( name="research_chain", description="搜索信息并进行分析" ) chain.add_step( tool_name="search", input_template="{input}", output_key="search_result" ) chain.add_step( tool_name="summarizer", input_template="请总结以下信息: {search_result}", output_key="summary" ) # 执行工具链 result = chain.execute(registry, "人工智能最新发展趋势") print(f"工具链结果: {result}") ``` --- Hello-Agents 教程涵盖了构建现代 AI 智能体系统的完整知识体系。从基础的 LLM 调用、经典智能体范式(ReAct、Plan-and-Solve、Reflection),到高级的记忆系统、RAG 检索增强、上下文工程和多智能体通信协议,提供了系统化的学习路径。 该框架的核心设计理念是"万物皆为工具"——将记忆、RAG、MCP 等模块都统一抽象为工具,简化了学习曲线。通过分层架构(核心框架层、Agent 实现层、工具系统层)实现了关注点分离,便于扩展和维护。教程还提供了三个完整的实战项目:智能旅行助手、深度研究智能体和赛博小镇模拟,让学习者能够将理论知识应用到实际场景中。