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

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 记忆的粒度与分类

不是所有对话中的信息都值得被记住。一股脑地存储所有对话历史,那叫“聊天记录”,不是“记忆”。我们需要对信息进行筛选和分类。通常,短期记忆可以分为几个层次:

  1. 实体记忆:这是最基础的一层。记录在对话中明确提及的、具有特定指代的对象。例如,用户说“我住在北京朝阳区”,那么“用户居住地:北京朝阳区”就应该作为一个实体被记录。这通常包括人名、地点、时间、数字、特定物品等。
  2. 偏好记忆:记录用户表现出的倾向性或习惯。例如,用户说“把背景调成暗色模式”,或者“报告请用PDF格式”。这类信息对于提供个性化服务至关重要。
  3. 会话状态记忆:记录多轮对话的进程和上下文。例如,在一个订餐Agent中,需要记住用户已经选择了“披萨”,正在询问“尺寸”。或者在一个故障排查Agent中,记住已经尝试过的步骤“重启了路由器”。
  4. 意图/目标记忆:记录用户在当前会话中的核心目标。例如,用户一开始说“我想规划一个周末旅行”,那么这个“规划周末旅行”就是最高级别的目标,后续的所有对话都应围绕这个目标展开。

设计时,你需要根据你的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格式回复。我们正在讨论‘短期记忆分类’,已经介绍了实体记忆和偏好记忆。现在请回答用户的下一个问题:[用户新问题]”
    • 决策依据:在基于ReAct(Reasoning-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轮(user+assistant为一轮) 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_tokens=150) 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_tokens=500) -> str: """调用OpenAI API(简化版,需添加错误处理)""" try: response = openai.ChatCompletion.create( model=“gpt-3.5-turbo”, # 或 “gpt-4” messages=[{“role”: “user”, “content”: prompt}], max_tokens=max_tokens, temperature=0.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循环结合:

  1. 记忆注入思考:在_call_llm生成思考前,thought_prompt中包含了self.memory.get_context_prompt()
  2. 记忆更新:在每轮动作执行后,通过self.memory.update_from_dialogue更新记忆。
  3. 记忆辅助回答:在执行_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.加日志:在记忆的updateget_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 高级优化策略

  1. 向量化记忆检索:当记忆条目很多时,线性查找效率低。可以将每条记忆(如“用户喜欢咖啡”)转换为向量嵌入(Embedding)。当用户提问时,将问题也转换为向量,然后通过向量相似度检索最相关的几条记忆。这非常适合“偏好记忆”这类非精确匹配的场景。
  2. 记忆重要性评分:不是所有信息都平等重要。“用户叫小明”可能比“用户今天喝了水”更重要。可以在信息被记录时,由LLM或规则系统为其打上一个重要性分数(如1-10)。在需要节省Token时,优先保留高分记忆。
  3. 分层记忆系统:将记忆分为“工作记忆”(极短期,存储正在处理的任务细节)、“情景记忆”(本次会话)和“长期记忆”。不同层级的读写策略和存储介质不同,形成更接近人类记忆的体系。
  4. 测试与评估:建立测试用例。例如:“第一轮告诉Agent‘我讨厌苹果’,第二轮问‘我喜欢什么水果?’”。一个合格的Agent应该回答“你不喜欢苹果”或进行澄清,而不是胡乱推荐苹果。通过自动化测试来持续验证记忆系统的有效性。

4.3 实操心得与避坑指南

  • 起步宜简不宜繁:不要一开始就设计复杂的记忆结构。从一个简单的字典开始,只记录一两个关键字段(如username),确保读写链路打通,再逐步增加复杂性。
  • LLM既是执行器,也是记忆的管理者:我们不仅用LLM来生成回复,也用它来理解和提炼记忆(抽取、摘要、重要性评分)。这能极大提升记忆系统的智能度,但也会增加成本和延迟。需要权衡。
  • 记忆的更新是“修正”而非“替换”:当用户说“其实我叫李华,不叫李明”时,你的记忆更新逻辑应该能处理这种更正,而不是简单地新增一条“user_name: 李华”导致存在两个矛盾的名字。可以考虑为关键实体设置唯一键,或引入版本管理。
  • 成本监控至关重要:记忆系统,尤其是依赖LLM进行摘要和抽取的系统,会显著增加API调用次数和Token消耗。在开发早期就要加入成本计量,并设定预算警报。
  • 用户可控性:考虑提供让用户查看和修正Agent记忆的途径。例如,用户可以问“你记得我的哪些信息?”,并且可以说“你记错了,我其实住在上海”。这能增加系统的透明度和可信度。

实现短期记忆是构建实用AI Agent的关键一步。它让对话从“一问一答”的孤立模式,升级为有上下文、有状态的连续协作过程。虽然挑战不少,但从一个最小可行产品(MVP)开始,逐步迭代,你会发现你的Agent正变得越来越“善解人意”和“靠谱”。