从零构建AI Agent框架：深入解析ReAct循环、工具调用与长期记忆实现

1. 项目概述：为什么我们需要亲手搭建一个AI Agent框架？

最近几年，AI Agent这个概念火得不行，几乎成了每个技术讨论的焦点。但说实话，市面上关于Agent的讨论，要么是停留在“它很智能、能自主完成任务”这种概念层面，要么就是直接甩给你一个OpenAI的API调用示例，告诉你用几个函数调用（Function Calling）就能实现一个简单的Agent。这中间存在一个巨大的断层：从一个简单的API调用，到一个真正健壮、可扩展、能处理复杂任务的AI Agent系统，到底该怎么走？这其中的理论支撑是什么？工程上的坑又有哪些？

这就是我动手写这个系列的原因。我不满足于只是“使用”别人封装好的Agent框架，我更想搞清楚它的“骨架”和“肌肉”是怎么长出来的。从零开始实现一个AI Agent框架，不仅仅是为了炫技，其核心价值在于深度理解与掌控。当你亲手设计它的思考循环（ReAct, CoT），实现工具调用（Tool Calling）的调度，处理长期记忆（Memory）的存储与检索，并搭建一个可靠的任务规划（Planning）与执行（Execution）引擎时，你才会真正明白那些现成框架（比如LangChain、AutoGen）里每个设计抉择背后的深意。你会知道为什么有些任务它处理得好，有些却会陷入死循环；你会懂得如何为你的特定业务场景定制最合适的Agent架构，而不是被框架的抽象所限制。

这个项目适合谁？我认为有三类朋友会特别有收获：一是对AI应用开发有浓厚兴趣，不满足于仅仅调用API的开发者；二是希望将AI能力深度集成到自家产品中，需要高度定制化Agent系统的技术负责人或架构师；三是任何想要深入理解“智能体”究竟如何运作的研究者或学生。我们将从最基础的理论模型讲起，然后用代码一步步把它们构建出来，过程中遇到的每一个错误和解决方案，我都会毫无保留地分享出来。

2. 核心理论基石：构建AI Agent的四大支柱

在动手写代码之前，我们必须把地基打牢。一个功能完整的AI Agent框架，其理论核心可以归纳为四个相互关联的支柱：规划（Planning）、记忆（Memory）、工具使用（Tool Use）和行动（Action）。理解这四者的关系，是设计任何Agent系统的前提。

2.1 规划模块：从目标到步骤的拆解艺术

规划，本质上是一个问题分解与路径搜索的过程。对于AI Agent而言，给定一个模糊的用户指令（如“帮我分析一下上季度的销售数据，并给出下季度的增长建议”），它需要自动将其分解为一系列可执行的具体步骤。这里涉及几个关键理论：

• 链式思考（Chain-of-Thought, CoT）与ReAct范式：这是当前让大语言模型（LLM）进行规划最主流的方法。CoT要求模型展示其推理的中间步骤。而ReAct（Reasoning + Acting）则更进一步，它将推理（Reason）和行动（Act）交织在一起。Agent不是一次性规划所有步骤，而是采用“思考一步，执行一步，观察结果，再思考下一步”的循环。这种方式更贴近人类解决问题的方式，也更能应对执行过程中的不确定性。在我们的框架设计中，ReAct循环将是核心执行引擎。
• 任务分解（Task Decomposition）：如何让LLM可靠地将大任务拆成小任务？常见策略有：1）按子目标分解：让LLM识别出达成总目标所需的中间子目标。2）按工具或能力分解：根据可用的工具集（如“查询数据库”、“生成图表”、“发送邮件”）来反向推导步骤。3）按时间或流程顺序分解：适用于有严格先后顺序的任务。在实现时，我们通常通过精心设计的系统提示词（System Prompt）来引导LLM进行这种分解。
• 规划算法：对于复杂任务，简单的逐步分解可能不够。我们可以引入更经典的算法思想，如基于树的搜索（广度优先、深度优先）或基于图的规划。例如，Agent可以将每个可能的行动步骤及其预期结果作为一个节点，构建一个搜索空间，然后寻找从初始状态到目标状态的最优路径。虽然这需要更多的计算和与环境的交互，但对于需要多步深度规划的场景至关重要。

注意：规划模块最怕陷入“空转”。即Agent不停地进行推理，却无法产生任何有效的行动步骤。为了防止这一点，我们需要在提示词中明确约束（如“必须从可用工具列表中选择行动”），并在代码层面设置最大推理步数或超时机制。

2.2 记忆模块：赋予Agent持续性的关键

一个没有记忆的Agent，每次交互都是全新的开始，这显然不是“智能体”该有的样子。记忆模块负责存储、组织和检索历史信息，是Agent实现持续对话、学习用户偏好、积累领域知识的基础。我们可以从三个时间维度来设计记忆：

• 短期记忆/工作记忆：这通常指当前会话的上下文。由于LLM有上下文窗口长度限制，我们需要一个高效的上下文窗口管理策略。简单的做法是使用“滑动窗口”，只保留最近的N轮对话。更高级的策略则涉及对历史对话进行摘要（Summarization），将冗长的历史压缩成精炼的要点，再放入上下文，从而在有限的窗口内承载更长时间跨度的信息。
• 长期记忆：这是Agent的“知识库”或“经验库”。所有超越当前会话的有用信息都应存储在这里。实现长期记忆的核心技术是向量数据库（Vector Database）。我们将对话、执行结果、学到的知识等文本转换成向量（Embedding），存入向量库。当需要回忆时，将当前问题也转换成向量，进行相似度搜索，召回最相关的记忆片段，再注入到当前上下文中。这解决了传统数据库基于关键词匹配的局限性，实现了基于语义的关联回忆。
• 记忆的层次与结构：记忆不是扁平的。我们可以设计不同的“记忆体”，例如：核心事实记忆（用户提供的明确信息）、对话历史记忆、自我反思记忆（Agent对自己过去行动成功或失败的分析）、技能记忆（如何成功使用某个工具的经验）。在检索时，可以根据当前任务类型，决定优先从哪种记忆体中搜索。

2.3 工具使用模块：Agent与世界的接口

Agent再聪明，如果无法操作外部系统，也只能是个“思想家”。工具使用模块是Agent能力的放大器。其核心理论是将外部API、函数、甚至其他软件的能力，抽象成统一的、LLM可理解和调用的“工具”描述。

• 工具的描述与发现：每个工具都需要一个清晰的“说明书”，通常包括：工具名称、功能描述、必需的输入参数（及其类型、格式、说明）、可能的输出。这个说明书必须以结构化数据（如JSON Schema）的形式存在，并能被动态地加载到Agent的上下文中。这就是所谓的工具发现（Tool Discovery）机制。
• 工具的选择与调用：当Agent决定要采取行动时，它需要：1）从当前可用的工具列表中，根据任务描述选择最合适的一个或多个工具。2）根据工具说明书，生成符合要求的调用参数。3）以安全、可控的方式执行该工具对应的代码或API请求。这里的关键是参数验证与错误处理。LLM生成的参数可能是错误的、不完整的，框架必须在调用前进行严格的校验，并在调用失败时提供清晰的错误信息反馈给Agent，以便其进行修正（ReAct中的“观察”阶段）。
• 工具的扩展性：框架必须支持轻松地注册新工具。理想情况下，开发者只需要定义一个Python函数，并用装饰器或配置文件描述其元信息，框架就能自动将其纳入Agent的工具库。这涉及到灵活的插件化架构设计。

2.4 行动与执行循环：将计划付诸实践

这是将前三个模块串联起来的“总控制器”，通常体现为ReAct循环。其标准流程如下：

1. 观察（Observe）：Agent接收当前的用户输入和来自环境的最新反馈（如上一步工具执行的结果或错误）。
2. 思考（Think）：基于观察、历史记忆和可用工具，LLM进行推理，决定下一步该做什么。输出应被严格格式化，例如，必须明确指示是“继续思考”还是“调用某个工具”。
3. 行动（Act）：如果决定调用工具，则框架解析LLM的输出，提取工具名和参数，执行工具调用。
4. 循环：将行动的结果作为新的“观察”，进入下一轮循环，直到LLM输出“最终答案”或达到终止条件（如任务完成、步数超限）。

这个循环的实现质量，直接决定了Agent的可靠性和效率。我们需要处理很多细节，比如：如何解析LLM非结构化的输出？如何管理循环状态？如何优雅地处理超时和中断？

3. 框架设计与核心模块实现

有了理论指导，我们就可以开始设计框架的顶层架构了。我们的目标是构建一个轻量级、模块化、易于扩展的框架，我将其命名为“ThinkFlow”。整个框架的核心是一个中央调度器（Orchestrator），它协调着各个模块的运作。

3.1 整体架构与模块划分

ThinkFlow 的核心架构围绕一个主循环展开，主要包括以下组件：

• Agent Core：封装了与大语言模型（LLM）的交互，负责生成推理和决策。
• Planner：规划模块，负责将复杂任务分解为子任务序列。初期我们可以实现一个基于提示词的简单规划器。
• Memory Manager：记忆管理器，统一管理短期会话缓存和长期向量存储的读写。
• Tool Registry：工具注册表，一个全局的单例，负责所有工具的注册、描述和查找。
• Executor：执行器，负责安全地调用Tool Registry中的工具，并处理执行结果和异常。
• Orchestrator：调度器，实现上述的ReAct循环，串联所有模块。

它们之间的关系如下图所示（概念图）：用户请求首先到达Orchestrator，Orchestrator调用Planner进行任务分解（对于复杂任务），然后进入ReAct循环。在循环中，Orchestrator将当前状态（用户输入、历史、可用工具）交给Agent Core生成思考，根据思考决定是调用Executor执行工具，还是直接给出答案。整个过程中，Memory Manager负责提供相关记忆和保存新的经验。

3.2 核心模块代码实现拆解

接下来，我们深入到几个最关键模块的代码层面。我会用Python来演示，因为其生态丰富，易于理解。

首先，是工具系统的实现。这是Agent能力的边界。我们设计一个工具装饰器，让开发者能轻松地将任何函数变成Agent可用的工具。

# tool.py import inspect import json from typing import Callable, Dict, Any, get_type_hints class Tool: """工具类，封装一个可调用函数及其元数据""" def __init__(self, func: Callable, name: str, description: str): self.func = func self.name = name self.description = description self.parameters = self._extract_parameters(func) def _extract_parameters(self, func: Callable) -> Dict[str, Any]: """从函数签名中提取参数信息，生成JSON Schema格式""" sig = inspect.signature(func) type_hints = get_type_hints(func) parameters = {} for param_name, param in sig.parameters.items(): if param_name == 'self': continue param_info = {"type": "string"} # 默认类型 if param_name in type_hints: type_str = str(type_hints[param_name]) # 简化类型映射，实际项目需要更完善的映射 if 'int' in type_str: param_info["type"] = "integer" elif 'float' in type_str: param_info["type"] = "number" elif 'bool' in type_str: param_info["type"] = "boolean" elif 'list' in type_str or 'List' in type_str: param_info["type"] = "array" if param.default != inspect.Parameter.empty: param_info["default"] = param.default param_info["description"] = f"参数 {param_name}" # 可通过docstring获取更佳描述 parameters[param_name] = param_info return parameters def to_schema(self) -> Dict[str, Any]: """生成OpenAI Function Calling兼容的工具模式""" return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": { "type": "object", "properties": self.parameters, "required": [k for k, v in self.parameters.items() if 'default' not in v] } } } def execute(self, **kwargs): """执行工具调用""" # 这里可以加入参数验证、权限检查、调用日志等 try: result = self.func(**kwargs) return {"status": "success", "data": result} except Exception as e: return {"status": "error", "message": str(e)} class ToolRegistry: """全局工具注册表""" _instance = None _tools: Dict[str, Tool] = {} def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance @classmethod def register(cls, name: str = None, description: str = ""): """装饰器，用于注册工具函数""" def decorator(func: Callable): tool_name = name or func.__name__ tool = Tool(func, tool_name, description) cls._tools[tool_name] = tool return func return decorator @classmethod def get_tool(cls, name: str) -> Tool: return cls._tools.get(name) @classmethod def get_all_schemas(cls): return [tool.to_schema() for tool in cls._tools.values()] # 使用示例 @ToolRegistry.register(description="计算两个数字的和") def add(a: int, b: int) -> int: """返回a与b的和""" return a + b @ToolRegistry.register(name="search_web", description="在互联网上搜索信息") def search(query: str) -> str: # 模拟搜索，实际应调用搜索引擎API return f"关于'{query}'的搜索结果：..."

这个工具系统实现了几个关键点：1）利用装饰器实现无侵入式的工具注册。2）自动从函数签名和类型注解中提取参数信息，生成标准化的模式描述。3）提供了统一的执行和错误处理接口。

其次，是实现一个基于向量数据库的长期记忆模块。这里我们以ChromaDB为例，它是一个轻量级的向量数据库。

# memory.py from typing import List, Dict, Any import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import uuid class LongTermMemory: """基于向量数据库的长期记忆""" def __init__(self, persist_directory: str = "./memory_db"): # 初始化嵌入模型 self.embedder = SentenceTransformer('all-MiniLM-L6-v2') # 轻量级句子嵌入模型 # 初始化Chroma客户端 self.client = chromadb.PersistentClient( path=persist_directory, settings=Settings(anonymized_telemetry=False) ) # 获取或创建集合（类似于表） self.collection = self.client.get_or_create_collection(name="agent_memories") def _generate_embedding(self, text: str) -> List[float]: """生成文本的向量嵌入""" return self.embedder.encode(text).tolist() def store(self, content: str, metadata: Dict[str, Any] = None): """存储一段记忆""" if metadata is None: metadata = {} embedding = self._generate_embedding(content) id = str(uuid.uuid4()) self.collection.add( embeddings=[embedding], documents=[content], metadatas=[metadata], ids=[id] ) return id def retrieve(self, query: str, n_results: int = 5) -> List[Dict[str, Any]]: """根据查询检索最相关的记忆""" query_embedding = self._generate_embedding(query) results = self.collection.query( query_embeddings=[query_embedding], n_results=n_results ) memories = [] # 解析返回结果 if results['documents']: for i in range(len(results['documents'][0])): memories.append({ "content": results['documents'][0][i], "metadata": results['metadatas'][0][i], "distance": results['distances'][0][i] }) return memories def clear(self): """清空所有记忆（谨慎使用）""" self.client.delete_collection(name="agent_memories") self.collection = self.client.create_collection(name="agent_memories")

这个记忆模块封装了向量的生成、存储和检索。在实际使用中，我们可以在Agent完成一个重要任务后，将任务总结和关键结果store起来。当遇到新任务时，先用retrieve查找相关历史经验，并将其作为上下文提供给LLM，从而实现“经验复用”。

4. 核心执行引擎：ReAct循环的工程实现

有了工具和记忆，现在我们需要一个“大脑”来驱动一切。这个大脑就是ReAct循环执行引擎。它的职责是维持Agent的思考-行动周期，并处理所有状态流转。

4.1 状态管理与循环控制

首先，我们需要定义Agent在循环中的状态。一个典型的状态对象应包含：

# orchestrator.py from dataclasses import dataclass, field from typing import List, Dict, Any, Optional @dataclass class AgentState: """Agent在一次运行中的完整状态""" user_input: str # 原始用户输入 task_description: str # 当前要处理的任务描述（可能被分解过） full_history: List[Dict[str, str]] # 完整的对话历史，格式：[{"role": "user/assistant/system", "content": "..."}] recent_observations: List[str] # 最近几步的观察（工具执行结果等） available_tools_schema: List[Dict] # 当前可用的工具模式列表 # 从长期记忆中检索到的相关上下文 retrieved_memories: List[str] = field(default_factory=list) # 当前循环的步数 step_count: int = 0 # 最终答案（如果已得出） final_answer: Optional[str] = None

接下来，我们实现Orchestrator的核心循环逻辑。这里的关键是设计一个强引导性的系统提示词，让LLM严格按照我们期望的格式（如JSON）输出，以便程序化解析。

# orchestrator.py (续) import json import re class ReactOrchestrator: def __init__(self, llm_client, memory: LongTermMemory, max_steps=20): self.llm = llm_client self.memory = memory self.max_steps = max_steps self.tool_registry = ToolRegistry() def _build_system_prompt(self, state: AgentState) -> str: """构建引导ReAct循环的系统提示词""" prompt = f"""你是一个智能助手，可以通过使用工具来解决问题。请遵循以下步骤： 1. 分析当前任务和可用工具。 2. 如果需要使用工具，请严格按照以下JSON格式回应： {{"thought": "你的推理过程", "action": "tool_name", "action_input": {{"arg1": "value1", ...}}}} 3. 如果任务已完成，可以给出最终答案，请使用格式： {{"thought": "总结推理", "action": "final_answer", "action_input": "你的最终答案"}} 当前任务：{state.task_description} 可用工具： {json.dumps(state.available_tools_schema, indent=2, ensure_ascii=False)} 历史观察（最近几步）： {chr(10).join(state.recent_observations[-3:]) if state.recent_observations else "无"} 相关历史经验： {chr(10).join(state.retrieved_memories) if state.retrieved_memories else "无"} 请开始你的思考。""" return prompt def _parse_llm_output(self, text: str) -> Dict[str, Any]: """解析LLM的输出，提取结构化的动作指令""" # 尝试从文本中提取JSON块 json_pattern = r'```json\s*(.*?)\s*```|```\s*(.*?)\s*```|\{.*\}' matches = re.findall(json_pattern, text, re.DOTALL) json_str = None for match in matches: json_str = match[0] or match[1] or match[2] if json_str: break if not json_str: # 如果没有找到代码块，尝试直接查找最外层的花括号 try: start = text.find('{') end = text.rfind('}') + 1 if start != -1 and end != 0: json_str = text[start:end] except: json_str = None if json_str: try: return json.loads(json_str) except json.JSONDecodeError: # 如果解析失败，回退到启发式解析或报错 pass # 如果无法解析为JSON，则假设模型直接给出了最终答案 return {"thought": text, "action": "final_answer", "action_input": text} def run(self, user_input: str) -> str: """主运行循环""" # 1. 初始化状态 state = AgentState( user_input=user_input, task_description=user_input, full_history=[{"role": "user", "content": user_input}], recent_observations=[], available_tools_schema=self.tool_registry.get_all_schemas(), retrieved_memories=[] ) # 2. 从长期记忆中检索相关上下文 if self.memory: memories = self.memory.retrieve(user_input, n_results=3) state.retrieved_memories = [m["content"] for m in memories] # 3. ReAct 循环 for step in range(self.max_steps): state.step_count = step + 1 print(f"\n=== 步骤 {step + 1} ===") # 3.1 构建当前轮次的对话上下文 messages = [] # 添加系统提示词 system_prompt = self._build_system_prompt(state) messages.append({"role": "system", "content": system_prompt}) # 添加完整对话历史（可选，可只添加摘要） messages.extend(state.full_history[-6:]) # 保留最近几轮 # 3.2 调用LLM进行思考 llm_response = self.llm.chat_completion(messages) assistant_reply = llm_response["choices"][0]["message"]["content"] state.full_history.append({"role": "assistant", "content": assistant_reply}) print(f"助理回复：{assistant_reply}") # 3.3 解析LLM输出，得到动作指令 action_dict = self._parse_llm_output(assistant_reply) thought = action_dict.get("thought", "") action = action_dict.get("action", "") action_input = action_dict.get("action_input", {}) print(f"解析结果：思考 - {thought}， 动作 - {action}") # 3.4 执行动作 if action == "final_answer": state.final_answer = str(action_input) # 可选：将成功的任务经验存入长期记忆 if self.memory and state.final_answer: summary = f"任务：{state.task_description}\n结果：{state.final_answer[:200]}..." self.memory.store(summary, metadata={"type": "successful_task"}) break elif action in [tool.name for tool in self.tool_registry._tools.values()]: # 执行工具调用 tool = self.tool_registry.get_tool(action) if tool: try: # 确保action_input是字典 if isinstance(action_input, str): try: action_input = json.loads(action_input) except: action_input = {"input": action_input} result = tool.execute(**action_input) observation = f"工具 '{action}' 执行结果：{result}" except Exception as e: observation = f"工具 '{action}' 执行出错：{str(e)}" else: observation = f"错误：未知工具 '{action}'" # 将观察结果加入状态 state.recent_observations.append(observation) state.full_history.append({"role": "user", "content": observation}) print(f"观察：{observation}") else: # 无法识别的动作，作为观察反馈给LLM observation = f"无法理解的动作指令：{action}。请使用指定的JSON格式，并选择可用的工具或给出最终答案。" state.recent_observations.append(observation) state.full_history.append({"role": "user", "content": observation}) print(f"观察：{observation}") # 3.5 检查步数限制 if step == self.max_steps - 1: state.final_answer = f"达到最大步数限制（{self.max_steps}），任务未完成。最后状态：{state.recent_observations[-1] if state.recent_observations else '无进展'}" break return state.final_answer or "任务执行未产生明确结果。"

这个Orchestrator实现了一个完整的、可运行的ReAct循环。它包含了状态管理、提示词工程、LLM输出解析、工具执行调度以及循环终止判断。这是整个框架的“心脏”。

4.2 与LLM的集成策略

上面的代码中，llm_client是一个抽象接口。在实际项目中，我们需要适配不同的LLM提供商（如OpenAI、Anthropic、国内大模型等）。一个好的设计是定义一个统一的LLM客户端接口：

# llm_client.py from abc import ABC, abstractmethod import openai # 示例 class BaseLLMClient(ABC): @abstractmethod def chat_completion(self, messages: List[Dict], **kwargs) -> Dict[str, Any]: """发送聊天补全请求，返回统一格式的响应""" pass class OpenAIClient(BaseLLMClient): def __init__(self, model: str = "gpt-4", api_key: str = None): self.model = model self.client = openai.OpenAI(api_key=api_key) def chat_completion(self, messages: List[Dict], **kwargs) -> Dict[str, Any]: response = self.client.chat.completions.create( model=self.model, messages=messages, **kwargs ) # 将响应格式化为统一的字典格式 return { "id": response.id, "choices": [{ "message": { "role": choice.message.role, "content": choice.message.content } } for choice in response.choices] }

这样，Orchestrator就与具体的LLM实现解耦了，我们可以轻松切换后端模型。

5. 实战演练：构建一个数据分析助手Agent

理论和技术都讲完了，现在我们用一个完整的例子，把所有的模块串起来，构建一个能实际运行的“数据分析助手”Agent。这个Agent的目标是：用户用自然语言提出一个数据分析需求（比如“帮我分析一下销售数据，找出最畅销的产品”），Agent能自动调用相应的工具（读取文件、执行查询、绘制图表）来完成分析并给出结论。

5.1 定义领域专用工具

首先，我们需要为这个数据分析场景创建几个工具。假设我们有一个CSV格式的销售数据文件sales.csv。

# data_analysis_tools.py import pandas as pd import matplotlib.pyplot as plt import io import base64 from ToolRegistry import ToolRegistry @ToolRegistry.register(description="读取指定路径的CSV文件，返回数据预览") def read_csv(file_path: str, n_rows: int = 5) -> str: """读取CSV文件并返回前几行预览""" try: df = pd.read_csv(file_path) preview = df.head(n_rows).to_string() shape = df.shape return f"文件 '{file_path}' 读取成功。数据形状：{shape}。前{n_rows}行预览：\n{preview}" except Exception as e: return f"读取文件失败：{str(e)}" @ToolRegistry.register(description="对数据进行SQL-like的查询，例如按列分组求和") def query_data(file_path: str, group_by: str, aggregate: str = "sum") -> str: """对数据进行简单的分组聚合查询""" try: df = pd.read_csv(file_path) if group_by not in df.columns: return f"错误：数据中不存在列 '{group_by}'。" if aggregate == "sum": result = df.groupby(group_by).sum(numeric_only=True) elif aggregate == "mean": result = df.groupby(group_by).mean(numeric_only=True) elif aggregate == "count": result = df.groupby(group_by).size() else: return f"不支持的聚合操作：{aggregate}，请使用 sum, mean 或 count。" return f"按 '{group_by}' 分组，执行 '{aggregate}' 聚合的结果：\n{result.to_string()}" except Exception as e: return f"查询失败：{str(e)}" @ToolRegistry.register(name="plot_chart", description="根据数据生成折线图或柱状图，返回图片的base64编码") def plot_chart(file_path: str, x_column: str, y_column: str, chart_type: str = "line") -> str: """生成图表并返回base64字符串""" try: df = pd.read_csv(file_path) if x_column not in df.columns or y_column not in df.columns: return f"错误：指定的列不存在。" plt.figure(figsize=(10, 6)) if chart_type == "line": plt.plot(df[x_column], df[y_column]) elif chart_type == "bar": plt.bar(df[x_column], df[y_column]) else: return f"不支持的图表类型：{chart_type}，请使用 'line' 或 'bar'。" plt.xlabel(x_column) plt.ylabel(y_column) plt.title(f"{y_column} vs {x_column}") plt.tight_layout() # 将图表保存到内存缓冲区，并转换为base64 buf = io.BytesIO() plt.savefig(buf, format='png') plt.close() buf.seek(0) img_base64 = base64.b64encode(buf.read()).decode('utf-8') return f"data:image/png;base64,{img_base64}" except Exception as e: return f"生成图表失败：{str(e)}" finally: plt.close('all')

5.2 组装并运行Agent

现在，我们把所有部件组装起来，并运行一个完整的任务。

# main.py from llm_client import OpenAIClient from memory import LongTermMemory from orchestrator import ReactOrchestrator import data_analysis_tools # 导入工具模块以完成注册 def main(): # 1. 初始化组件 llm = OpenAIClient(model="gpt-4") # 请替换为你的API Key memory = LongTermMemory() orchestrator = ReactOrchestrator(llm, memory, max_steps=15) # 2. 用户输入一个复杂任务 user_query = """ 请分析项目根目录下的 'sales.csv' 文件。 首先，告诉我这个文件里有什么数据。 然后，找出哪个产品（product列）的总销售额（sales列）最高。 最后，请为我生成一个展示每个产品总销售额的柱状图。 """ print("用户请求：", user_query) print("="*50) # 3. 运行Agent final_answer = orchestrator.run(user_query) # 4. 输出最终结果 print("\n" + "="*50) print("任务完成！最终答案：") print(final_answer) # 处理可能包含图片的答案 if final_answer and final_answer.startswith("data:image/png;base64,"): # 如果是图片，可以保存到本地 import base64 img_data = final_answer.split(',')[1] with open('output_chart.png', 'wb') as f: f.write(base64.b64decode(img_data)) print("图表已保存为 'output_chart.png'") else: print(final_answer) if __name__ == "__main__": main()

当你运行这段代码时，你会看到控制台打印出Agent的思考过程：

1. 步骤1：LLM分析任务，发现需要先读取文件。它输出JSON，调用read_csv工具。
2. 步骤2：Orchestrator执行工具，得到数据预览，并将其作为“观察”反馈给LLM。
3. 步骤3：LLM根据预览，知道下一步需要按产品分组求和。它调用query_data工具，参数为group_by='product', aggregate='sum'。
4. 步骤4：得到查询结果后，LLM决定生成图表。它调用plot_chart工具，指定x_column='product', y_column='sales', chart_type='bar'。
5. 步骤5：收到图表数据后，LLM综合所有信息，给出最终答案，并可能附上分析结论。

这个过程完美地展示了ReAct循环：思考 -> 行动 -> 观察 -> 再思考。我们的框架成功地协调了LLM的推理能力和外部工具的执行能力。

6. 避坑指南与性能优化实战

在从零搭建和实际使用这个框架的过程中，我踩过不少坑，也总结出一些让Agent更稳定、更高效的技巧。这部分是你在官方文档里很难看到的“实战经验”。

6.1 提示词工程：让LLM乖乖听话

LLM不按格式输出，是ReAct循环失败的主要原因。除了在代码中做健壮的解析，提示词的设计至关重要。

• 结构化输出强制：在系统提示词中，明确要求LLM以指定格式（如JSON）回复，并给出多个正反示例。例如：

  你必须以JSON格式回应。例如，调用工具：{"thought": "...", "action": "tool_name", "action_input": {...}}。给出最终答案：{"thought": "...", "action": "final_answer", "action_input": "..."}。任何其他格式都将导致错误。

• 思维链（CoT）引导：鼓励LLM在thought字段中详细推理。例如：“在thought中，请逐步解释你为什么选择这个工具，以及你期望得到什么结果。” 这不仅能提高动作的准确性，也便于我们调试。

• 工具描述精细化：给工具的description和参数描述写得越详细、越自然，LLM调用得就越准。避免使用技术性过强的语言，用LLM能理解的自然语言描述功能。例如，description="计算给定列表中所有数字的总和"比description="sum function"要好得多。

6.2 错误处理与循环稳定性

一个生产级的Agent框架必须能优雅地处理各种异常。

• 工具执行异常：工具可能因为网络、参数错误、权限等问题失败。我们的Tool.execute()方法已经返回了包含状态的字典。Orchestrator需要将这些错误信息清晰地反馈给LLM，让它有机会调整策略。例如，将错误信息格式化为：“工具X执行失败，原因为：Y。请检查你的输入参数或尝试其他方法。”

• LLM输出解析失败：即使用户提示词，LLM也可能输出非结构化内容。我们的_parse_llm_output函数尝试了多种提取JSON的方法。如果全部失败，最后的兜底策略是将整个输出当作final_answer，或者将其作为错误观察反馈给LLM，要求它重试。可以设置一个重试计数器，避免无限循环。

• 循环停滞检测：Agent有时会陷入“死循环”，比如反复调用同一个工具且得到相同结果。解决方法有：1）在状态中记录最近N次的动作和观察，如果检测到重复模式，则中断循环并提示用户。2）设置最大步数，这是必须的保险丝。

6.3 记忆系统的优化技巧

记忆系统用不好，反而会成为负担。

• 存储策略：不要什么都存。只存储成功的任务总结、重要的用户信息（如偏好）以及从失败中学习的教训。存储时，可以生成一个更概括的摘要，而不是原始对话的罗列。
• 检索优化：简单基于向量的相似度搜索有时会召回不相关的记忆。可以尝试：1）元数据过滤：在存储时为记忆打上标签（如task_type: "data_analysis"），检索时结合语义相似度和标签过滤。2）重排序（Rerank）：先用向量数据库召回较多结果（如10个），再用一个小型的、更精确的交叉编码器模型对它们进行重排序，选出最相关的3-5个。
• 上下文窗口管理：对于长对话，将完整的记忆全部放入上下文是不现实的。可以采用“摘要+最近记录”的混合模式。定期（例如每10轮对话）让LLM对之前的对话历史生成一个简短摘要，然后用这个摘要替代之前的大段历史，腾出空间给新的对话。

6.4 性能与成本考量

频繁调用LLM和向量数据库，成本和延迟是必须考虑的问题。

• LLM调用优化：
  • 缓存：对相同的提示词（或提示词哈希）的LLM响应进行缓存。这在开发调试和重复性问题中能节省大量成本。
  • 使用更小的模型：对于简单的工具选择、参数提取等任务，可以尝试使用更小、更快的模型（如GPT-3.5-Turbo），而只在需要复杂推理时使用大模型（如GPT-4）。
  • 并行工具调用：如果多个工具调用之间没有依赖关系，可以探索让LLM一次性输出多个工具调用请求，然后并行执行，减少循环次数。
• 向量检索优化：
  • 分块（Chunking）：存储长文本记忆时，将其分成有重叠的小块（如256个token一块）。这样检索时能更精准地定位到相关片段，而不是返回一整篇不相关的文档。
  • 选择合适的索引：ChromaDB、Pinecone等向量数据库都支持多种索引算法（如HNSW）。根据数据规模和查询延迟要求进行选择和调参。

从零开始实现一个AI Agent框架，就像亲手组装一台精密的钟表。你不仅知道了指针如何走动，更理解了每一个齿轮的咬合与发条的张力。这个过程让我深刻体会到，现成的框架固然方便，但隐藏了太多关键的细节和设计权衡。当你自己实现一遍，你会对“智能体”的脆弱性、对提示词的敏感性、对错误处理的必要性有全新的认识。这套代码只是一个起点，你可以在此基础上，继续探索多Agent协作、更复杂的规划算法（如基于LLM的Tree of Thoughts）、动态工具学习等高级特性。最重要的是，你拥有了根据具体业务需求，定制和优化每一个环节的能力，这才是核心竞争力所在。