AI Agent短期记忆系统:基于ReAct框架与OpenAI API的工程实践

📅 2026/8/8 3:03:10
AI Agent短期记忆系统:基于ReAct框架与OpenAI API的工程实践
1. 项目概述为什么AI需要“短期记忆”如果你玩过早期的聊天机器人或者用过一些基础的大模型接口肯定遇到过这样的尴尬你告诉它“我叫小明”聊了五句之后你再问“我叫什么名字”它很可能已经忘得一干二净或者开始胡编乱造。这种“金鱼脑”式的体验让AI显得非常不智能也极大地限制了它的应用场景比如客服、个性化助手或者游戏NPC。这就是我们今天要解决的核心问题为AI Agent赋予“短期记忆”能力。简单来说“短期记忆”就是让AI能够记住在当前一次对话或任务执行过程中产生的关键信息。它不同于需要长期存储和检索的“长期记忆”比如知识库短期记忆更关注会话的上下文连贯性。想象一下你和朋友聊天你们会自然地记住刚才讨论的话题、提到的名字和达成的共识这就是短期记忆在起作用。对于AI Agent而言实现短期记忆就是让它能像人一样在单次交互的“工作内存”中暂存和调用必要的信息从而做出更连贯、更合理的决策和回应。这个需求在基于大语言模型LLM构建的Agent中尤为突出。无论是使用OpenAI API、Claude还是国内的各种大模型模型本身对长上下文的处理能力虽然越来越强但默认情况下它并不会主动地、结构化地记住你告诉它的“事实”。每次调用你都需要把所有的历史对话作为上下文喂给它这不仅浪费宝贵的Token直接关系到API成本而且在复杂、多轮的任务中信息很容易被淹没或扭曲。因此实现短期记忆本质上是在模型外部构建一个轻量级、高效率的“信息暂存区”。这个暂存区会动态地记录本次会话中的关键实体如人名、地点、用户偏好如“我喜欢喝美式咖啡”以及任务状态如“已经完成了步骤A和B”。当Agent需要进行下一轮推理或回复时它会优先从这个暂存区提取信息再结合当前的问题和长期知识库生成最终响应。这就像给Agent配备了一个随身便签本随时记录随时翻阅。2. 短期记忆系统的核心设计思路为Agent设计短期记忆不是一个简单的“存储-读取”问题。你需要考虑记忆的粒度、存储格式、触发机制以及最重要的——如何让记忆有效地影响Agent的决策过程。一个健壮的短期记忆系统通常包含以下几个核心组件。2.1 记忆的粒度与分类不是所有对话中的信息都值得被记住。一股脑地存储所有对话历史那叫“聊天记录”不是“记忆”。我们需要对信息进行筛选和分类。通常短期记忆可以分为几个层次实体记忆这是最基础的一层。记录在对话中明确提及的、具有特定指代的对象。例如用户说“我住在北京朝阳区”那么“用户居住地北京朝阳区”就应该作为一个实体被记录。这通常包括人名、地点、时间、数字、特定物品等。偏好记忆记录用户表现出的倾向性或习惯。例如用户说“把背景调成暗色模式”或者“报告请用PDF格式”。这类信息对于提供个性化服务至关重要。会话状态记忆记录多轮对话的进程和上下文。例如在一个订餐Agent中需要记住用户已经选择了“披萨”正在询问“尺寸”。或者在一个故障排查Agent中记住已经尝试过的步骤“重启了路由器”。意图/目标记忆记录用户在当前会话中的核心目标。例如用户一开始说“我想规划一个周末旅行”那么这个“规划周末旅行”就是最高级别的目标后续的所有对话都应围绕这个目标展开。设计时你需要根据你的Agent的具体任务来决定侧重哪些记忆类型。一个客服Agent可能更关注实体和会话状态而一个创意写作助手则可能更关注风格偏好。2.2 记忆的存储与数据结构短期记忆需要快速读写因此通常使用内存中的数据结构如Python的字典dict或列表list。对于更复杂或需要持久化如下文会提到的“记忆沉淀”的场景可能会用到轻量级数据库如SQLite或向量数据库用于基于语义的检索。一个典型的内存中的短期记忆结构可能如下所示short_term_memory { “session_id”: “abc123”, “entities”: { “user_name”: “小明”, “user_city”: “北京”, “current_project”: “短期记忆系统开发” }, “preferences”: { “output_format”: “markdown”, “verbosity_level”: “detailed” }, “conversation_state”: { “current_task”: “解释记忆分类”, “completed_steps”: [“介绍实体记忆” “介绍偏好记忆”], “next_expected_action”: “询问用户是否理解” }, “dialogue_history”: [ # 可选保留最近N轮原始对话用于上下文 {“role”: “user”, “content”: “短期记忆有哪些类型”}, {“role”: “assistant”, “content”: “主要分为实体记忆、偏好记忆...”} ] }注意dialogue_history的保留需要谨慎。虽然保留原始对话能提供最丰富的上下文但它会迅速消耗Token。一个常见的策略是只保留最近3-5轮对话或者通过摘要Summarization的方式将更早的对话压缩成一段简短的文本描述后再存储。2.3 记忆的读写机制何时记如何用这是短期记忆系统的“大脑”决定了系统的智能程度。记忆的写入记主动抽取在Agent生成回复后或解析用户输入后调用一个专门的“记忆抽取”模块。这个模块可以是一个提示词工程Prompt Engineering任务让LLM从文本中提取结构化信息也可以基于规则如正则表达式匹配特定模式。示例Prompt用于LLM抽取“请从以下对话中提取关键信息并以JSON格式输出。需要提取的信息包括1. 出现的人名、地名、组织名等实体2. 用户明确表达的偏好如格式、风格等3. 当前正在进行的任务或目标。对话内容[用户输入和AI回复]”被动触发当用户输入中包含明显的声明语句如“我是...”、“我喜欢...”、“我的目标是...”或通过意图识别模块判断为需要记录的信息时触发写入。记忆的读取用上下文注入在每次调用LLM生成回复前将当前相关的短期记忆以自然语言或结构化的提示插入到请求的上下文Prompt中。这是最直接的方式。示例“已知以下关于当前用户和会话的信息用户叫小明他喜欢详细的Markdown格式回复。我们正在讨论‘短期记忆分类’已经介绍了实体记忆和偏好记忆。现在请回答用户的下一个问题[用户新问题]”决策依据在基于ReActReasoning-Acting等框架的Agent中短期记忆可以作为“观察”Observation的一部分直接影响智能体的“思考”Thought和下一步“行动”Action。例如记忆中有“用户是VIP客户”那么决策时可能选择优先处理其请求。3. 基于ReAct框架与OpenAI API的实战实现理论讲完了我们动手搭建一个最简单的、具备短期记忆的对话Agent。我们将使用经典的ReAct框架和OpenAI的GPT模型API。3.1 环境准备与基础架构首先确保你已安装必要的库并准备好OpenAI API Key。pip install openai我们的Agent核心循环将遵循ReAct模式思考Thought- 行动Action- 观察Observation并在此过程中融入短期记忆。import openai import json from typing import Dict, List, Any, Optional # 设置你的OpenAI API Key openai.api_key “你的API_KEY” class ShortTermMemory: 短期记忆体 def __init__(self, session_id: str): self.session_id session_id self.entities: Dict[str, Any] {} self.preferences: Dict[str, Any] {} self.conversation_state: Dict[str, Any] {“current_goal”: None, “steps”: []} self.recent_dialogue: List[Dict] [] # 保留最近3轮对话 def update_from_dialogue(self, user_input: str, agent_response: str): 从一轮对话中更新记忆简化版实际应用需要更复杂的NLP解析 # 1. 保存对话历史 self.recent_dialogue.append({“role”: “user”, “content”: user_input}) self.revious_dialogue.append({“role”: “assistant”, “content”: agent_response}) if len(self.recent_dialogue) 6: # 只保留3轮userassistant为一轮 self.recent_dialogue self.recent_dialogue[-6:] # 2. 非常简单的规则抽取实际项目应用LLM或更复杂的NLP工具 if “我叫” in user_input or “我是” in user_input: # 简陋的提取名字例如“我叫张三” - “张三” name_part user_input.replace(“我叫”, “”).replace(“我是”, “”).strip().split(“ ”)[0] if name_part: self.entities[“user_name”] name_part print(f”[记忆更新] 记录用户姓名: {name_part}”) if “喜欢” in user_input and (“格式” in user_input or “风格” in user_input): # 简陋的提取偏好 self.preferences[“output_style”] “用户表达了偏好” print(f”[记忆更新] 记录用户偏好”) def get_context_prompt(self) - str: 将记忆转化为自然语言上下文提示 context_lines [] if self.entities.get(“user_name”): context_lines.append(f“用户的名字是{self.entities[‘user_name’]}。”) if self.conversation_state.get(“current_goal”): context_lines.append(f“当前对话的目标是{self.conversation_state[‘current_goal’]}。”) if self.conversation_state.get(“steps”): context_lines.append(f“已经完成的步骤有{‘ ’.join(self.conversation_state[‘steps’])}。”) return “ ”.join(context_lines) if context_lines else “”3.2 实现ReAct循环与记忆整合接下来我们构建一个主循环将记忆体整合到ReAct的每一步中。class MemoryEnhancedAgent: def __init__(self): self.memory ShortTermMemory(session_id“test_session”) self.actions { # 定义Agent可以执行的动作 “answer”: self._act_answer, “ask_clarification”: self._act_ask_clarification, “update_goal”: self._act_update_goal, “finalize”: self._act_finalize } def run(self, initial_input: str): print(f“用户: {initial_input}”) self.memory.conversation_state[“current_goal”] “回答用户问题” max_turns 10 user_input initial_input for turn in range(max_turns): # 步骤1: 思考 (Thought) - 结合记忆进行推理 thought_prompt f 你是我的助手。以下是当前已知信息 {self.memory.get_context_prompt()} 最近的对话 {self._format_recent_dialogue()} 用户最新消息{user_input} 请思考基于已知信息和对话历史我应该做什么请从可用动作中选择{list(self.actions.keys())}。 你的思考过程 thought self._call_llm(thought_prompt, max_tokens150) print(f“思考: {thought}”) # 步骤2: 决定行动 (Action) - 从思考中解析出动作 action, action_input self._parse_action(thought, user_input) if not action: action “answer” # 默认动作 # 步骤3: 执行行动 (Act) 并获取观察 (Observation) print(f“执行动作: {action}”) observation self.actions[action](action_input, user_input) # 步骤4: 更新记忆 (Memory Update) - 基于本轮交互 # 这里我们先生成一个临时回复用于更新记忆实际回复可能由_action生成 temp_agent_response observation if action “answer” else f“执行了{action}动作。” self.memory.update_from_dialogue(user_input, temp_agent_response) # 步骤5: 检查是否结束或获取下一轮用户输入 if action “finalize”: print(“对话结束。”) break # 模拟下一轮用户输入在实际应用中这里会等待真实用户输入 # 为了演示我们简单地将观察作为下一轮输入或结束。 if “?” in observation and turn max_turns - 1: user_input “是的请继续。” # 模拟用户确认 print(f“用户 (模拟): {user_input}”) else: print(f“助手: {observation}”) break # 假设一轮回答结束 def _call_llm(self, prompt: str, max_tokens500) - str: 调用OpenAI API简化版需添加错误处理 try: response openai.ChatCompletion.create( model“gpt-3.5-turbo”, # 或 “gpt-4” messages[{“role”: “user”, “content”: prompt}], max_tokensmax_tokens, temperature0.7, ) return response.choices[0].message.content.strip() except Exception as e: return f“调用API时出错: {e}” def _parse_action(self, thought: str, user_input: str) - (str, str): 从思考文本中解析出动作和输入非常简单的实现 thought_lower thought.lower() if “ask” in thought_lower or “clarif” in thought_lower: return “ask_clarification”, user_input elif “update goal” in thought_lower: return “update_goal”, user_input elif “final” in thought_lower or “end” in thought_lower: return “finalize”, “” else: return “answer”, user_input # 默认回答 def _act_answer(self, action_input: str, user_input: str) - str: 执行‘回答’动作 answer_prompt f 基于以下已知信息和对话历史回答问题。 已知信息{self.memory.get_context_prompt()} 对话历史{self._format_recent_dialogue()} 问题{user_input} 请给出专业、清晰的回答。 return self._call_llm(answer_prompt) def _act_ask_clarification(self, action_input: str, user_input: str) - str: 执行‘请求澄清’动作 return “为了更好的帮助你可以请你再详细说明一下你的问题吗” def _act_update_goal(self, action_input: str, user_input: str) - str: 执行‘更新目标’动作 # 这里可以调用LLM来从用户输入中提取新目标 self.memory.conversation_state[“current_goal”] f“更新后的目标基于: {user_input[:50]}...” self.memory.conversation_state[“steps”].append(“更新了对话目标”) return “好的我已经更新了我们对话的目标。” def _act_finalize(self, action_input: str, user_input: str) - str: 执行‘结束’动作 return “本次对话到此结束感谢你的交流” def _format_recent_dialogue(self) - str: return “\n”.join([f“{msg[‘role’]}: {msg[‘content’]}” for msg in self.memory.recent_dialogue[-4:]]) # 返回最近2轮 # 运行Agent if __name__ “__main__”: agent MemoryEnhancedAgent() agent.run(“你好我叫李雷。我想了解一下如何学习Python编程。”)这个示例虽然简化但清晰地展示了短期记忆如何与ReAct循环结合记忆注入思考在_call_llm生成思考前thought_prompt中包含了self.memory.get_context_prompt()。记忆更新在每轮动作执行后通过self.memory.update_from_dialogue更新记忆。记忆辅助回答在执行_act_answer时回答的Prompt也包含了记忆上下文。3.3 从短期到长期记忆的沉淀与清理短期记忆不能无限增长。我们需要一个机制来决定哪些记忆值得被转化为长期记忆存入数据库或向量库哪些应该被遗忘。记忆沉淀当一次会话结束或者某个记忆片段被反复调用、且被判定为重要时例如用户多次强调自己的偏好可以触发沉淀过程。这通常需要另一个LLM调用或规则判断“总结本次对话中关于用户的核心信息”或“判断‘用户是素食主义者’这条信息是否具有长期价值”。记忆清理最简单的清理策略是基于会话。当会话结束时清空整个短期记忆体。更精细的策略可以基于时间衰减或重要性评分在会话过程中就逐步淘汰陈旧或次要的信息。# 一个简单的记忆沉淀示例函数 def condense_memory_to_long_term(short_term_memory: ShortTermMemory) - Dict: 将短期记忆压缩、提炼准备存入长期存储 summary_prompt f 请总结以下对话片段中的关键个人信息和用户稳定偏好这些信息值得长期记住以便在未来提供个性化服务。 对话片段{short_term_memory.recent_dialogue} 已知已记录的实体{short_term_memory.entities} 请用JSON格式输出包含‘user_profile’和‘key_preferences’两个字段。 # 调用LLM生成总结 # long_term_info call_llm(summary_prompt) # return json.loads(long_term_info) # 此处为演示返回模拟数据 return { “user_profile”: {“name”: short_term_memory.entities.get(“user_name”, “未知”)}, “key_preferences”: short_term_memory.preferences }4. 常见问题、优化策略与避坑指南在实际开发中你会遇到各种各样的问题。下面是我从项目实践中总结的一些关键点和避坑技巧。4.1 典型问题与排查问题现象可能原因排查与解决思路Agent“忘记”刚告诉它的信息1. 记忆更新逻辑未触发或出错。2. 记忆上下文未正确注入到后续的Prompt中。3. 记忆键Key冲突或覆盖。1.加日志在记忆的update和get_context方法中加入详细日志打印出每次操作的内容。2.检查Prompt将实际发送给LLM的完整Prompt打印出来确认记忆文本是否在其中。3.简化测试先用一个固定的记忆如{“name”: “TestUser”}测试看Agent是否能正确引用排除抽取逻辑的问题。记忆内容混乱或错误1. 信息抽取的Prompt或规则不准确抽到了无关内容。2. 多轮对话后记忆未及时清理新旧信息矛盾。1.优化抽取器为LLM设计更精确的抽取Prompt提供明确范例Few-shot。或使用更专业的NLP模型进行实体识别。2.引入置信度与冲突解决为新抽取的信息添加置信度分数。当新信息与旧信息冲突时根据置信度、信息新鲜度或来源优先级进行裁决。Token使用量激增1. 存储了完整的对话历史且轮次过多。2. 记忆上下文描述过于冗长。1.摘要化不要存储原始长文本。定期如每5轮用LLM对之前的对话历史进行摘要只存储摘要。2.结构化与压缩坚持使用结构化的记忆如JSON而非自然语言描述。只存储关键事实而不是完整的句子。3.设置上限为recent_dialogue和每个记忆字段的条目数设置硬性上限。Agent过度依赖记忆显得刻板记忆的权重过高或者记忆信息过时后未更新导致Agent无法根据新情况灵活调整。1.动态上下文构建不要总是把全部记忆塞进Prompt。设计一个“记忆检索”模块只选取与当前用户问题最相关的几条记忆放入上下文。2.为记忆添加元数据如时间戳、使用频率。在构建上下文时优先使用新鲜、高频的记忆。4.2 高级优化策略向量化记忆检索当记忆条目很多时线性查找效率低。可以将每条记忆如“用户喜欢咖啡”转换为向量嵌入Embedding。当用户提问时将问题也转换为向量然后通过向量相似度检索最相关的几条记忆。这非常适合“偏好记忆”这类非精确匹配的场景。记忆重要性评分不是所有信息都平等重要。“用户叫小明”可能比“用户今天喝了水”更重要。可以在信息被记录时由LLM或规则系统为其打上一个重要性分数如1-10。在需要节省Token时优先保留高分记忆。分层记忆系统将记忆分为“工作记忆”极短期存储正在处理的任务细节、“情景记忆”本次会话和“长期记忆”。不同层级的读写策略和存储介质不同形成更接近人类记忆的体系。测试与评估建立测试用例。例如“第一轮告诉Agent‘我讨厌苹果’第二轮问‘我喜欢什么水果’”。一个合格的Agent应该回答“你不喜欢苹果”或进行澄清而不是胡乱推荐苹果。通过自动化测试来持续验证记忆系统的有效性。4.3 实操心得与避坑指南起步宜简不宜繁不要一开始就设计复杂的记忆结构。从一个简单的字典开始只记录一两个关键字段如username确保读写链路打通再逐步增加复杂性。LLM既是执行器也是记忆的管理者我们不仅用LLM来生成回复也用它来理解和提炼记忆抽取、摘要、重要性评分。这能极大提升记忆系统的智能度但也会增加成本和延迟。需要权衡。记忆的更新是“修正”而非“替换”当用户说“其实我叫李华不叫李明”时你的记忆更新逻辑应该能处理这种更正而不是简单地新增一条“user_name: 李华”导致存在两个矛盾的名字。可以考虑为关键实体设置唯一键或引入版本管理。成本监控至关重要记忆系统尤其是依赖LLM进行摘要和抽取的系统会显著增加API调用次数和Token消耗。在开发早期就要加入成本计量并设定预算警报。用户可控性考虑提供让用户查看和修正Agent记忆的途径。例如用户可以问“你记得我的哪些信息”并且可以说“你记错了我其实住在上海”。这能增加系统的透明度和可信度。实现短期记忆是构建实用AI Agent的关键一步。它让对话从“一问一答”的孤立模式升级为有上下文、有状态的连续协作过程。虽然挑战不少但从一个最小可行产品MVP开始逐步迭代你会发现你的Agent正变得越来越“善解人意”和“靠谱”。