1. 项目缘起为什么AI Agent需要“短期记忆”最近在折腾AI Agent开发的朋友估计都遇到过同一个让人挠头的问题你跟Agent聊得正嗨让它帮你分析一份文档或者规划一个项目结果你刚把需求说完它转头就忘了你上一句说了啥。你不得不像个复读机一样把上下文、背景信息、你的偏好一遍又一遍地喂给它。这感觉就像在跟一个患有严重瞬时失忆症的天才对话空有强大的推理能力却记不住三秒前的事。这就是我们今天要解决的痛点为Agent赋予“短期记忆”。这里的“短期记忆”在技术语境下通常指的是对话上下文Conversation Context的管理。它不像长期记忆那样需要向量数据库做持久化存储和复杂检索它的核心任务是在当前一次会话或一个任务链中让Agent能记住之前交互过的所有关键信息并基于这些信息做出连贯、合理的后续决策。为什么这如此重要我们来看一个经典的ReActReasoning Acting框架应用场景。假设你让一个Agent帮你订机票用户“帮我查一下下周五从北京飞往上海的航班。”Agent思考用户需要查询航班。我需要调用“航班搜索工具”。行动调用工具返回了航班A上午、航班B下午。用户“下午的航班有哪些”理想的Agent思考用户刚才问了下周五北京到上海的航班我返回了A和B。现在用户问“下午的航班”指的是对上次结果的筛选所以我应该从结果中筛选出下午起飞的航班也就是航班B。失忆的Agent思考用户问“下午的航班有哪些”。这是一个新的、独立的查询。我需要调用“航班搜索工具”但缺少出发地、目的地、日期等关键信息。我应该向用户询问这些信息。看到了吗没有短期记忆Agent就无法理解指代“下午的航班”指代上一步的结果无法进行多轮对话更无法执行复杂的、需要多步协作的任务链。它每一步都是孤立的智能程度大打折扣。因此实现短期记忆是让Agent从“单次问答机器”进化为“连贯任务执行者”的关键一步。这不仅仅是技术实现更是提升Agent实用性和用户体验的核心。2. 短期记忆的技术本质上下文管理与Token的博弈在深入代码之前我们必须先理解短期记忆在技术上的实现载体和核心约束。对于基于大语言模型LLM如OpenAI的GPT系列的Agent来说短期记忆几乎等价于如何有效地管理和利用有限的上下文窗口Context Window。2.1 上下文窗口记忆的物理边界LLM并非真正“记住”了内容它只是在处理你输入的文本序列。这个序列有一个最大长度限制就是上下文窗口通常用Token数来衡量。例如GPT-3.5-turbo的典型窗口是16K tokensGPT-4 Turbo可以达到128K tokens。每一次你调用API发送给模型的messages数组包含系统指令、用户消息、助手历史回复等其总Token数不能超过这个限制。这个messages数组就是Agent短期记忆的全部内容。你通过不断将新的对话回合追加到这个数组并再次发送给模型来实现“记忆”的传递。2.2 Token经济学记忆的成本与优化Token是计费单位也是资源单位。把整个对话历史原封不动地塞进上下文有两个致命问题成本高昂每次API调用输入的Token数会计费。冗长的历史会迅速推高成本。效率低下模型的有效注意力是有限的。过长的、包含大量无关细节的历史会稀释模型对当前任务关键信息的关注度可能导致输出质量下降甚至因为触及窗口上限而请求失败。因此短期记忆的实现本质上是一场Token经济学的博弈如何在有限的上下文窗口内以最低的成本保留对当前任务最有价值的信息。2.3 记忆的粒度从原始对话到摘要提炼根据优化程度短期记忆的管理可以分为几个层次原始对话流Naive Approach最简单粗暴的方式将整个messages历史全部传递。适用于对话轮次很少的简单场景但不可持续。滑动窗口Sliding Window只保留最近N轮对话例如最近10条消息。这解决了无限增长的问题但会无情地丢弃超出窗口的早期信息哪怕这些信息如用户设定的核心目标至关重要。摘要压缩Summarization这是更高级的策略。当对话历史增长到一定长度时主动调用LLM对早期的、非近期的历史进行摘要总结然后用一段简短的摘要文本替代原来的大段历史。这样既保留了早期信息的核心语义又极大地节省了Token。增量摘要每次摘要时是基于上一版的摘要和新的历史进行的像滚雪球一样记忆被不断压缩和提炼。关键信息提取Key Information Extraction不保存完整对话而是定义一个结构化的“记忆体”在每轮交互后主动提取并更新其中的关键字段如“用户偏好”、“当前任务目标”、“已执行步骤”、“临时结果”等。这需要更精细的设计但记忆效率最高。对于大多数入门和中级应用滑动窗口摘要压缩的组合策略是一个实用且有效的起点。接下来我们就基于OpenAI API和ReAct模式来具体实现它。3. 实战构建基于OpenAI API与ReAct的带记忆Agent我们将构建一个简单的命令行Agent它能够理解用户指令在必要时调用预设的工具比如一个计算器一个搜索引擎模拟器并且能记住对话历史。我们会实现一个具备固定长度滑动窗口和触发式摘要功能的记忆管理器。3.1 环境准备与基础架构首先确保你已安装Python和必要的库。你需要一个OpenAI API Key。pip install openai我们设计几个核心类Message: 封装单条消息角色、内容。MemoryManager: 记忆管理器的核心负责存储、修剪、摘要历史消息。Tool: 工具基类定义工具的执行接口。Agent: Agent本体整合LLM调用、记忆管理、工具使用ReAct逻辑。让我们从MemoryManager开始这是短期记忆的核心。# memory_manager.py import tiktoken from openai import OpenAI import json class MemoryManager: def __init__(self, client: OpenAI, model: str gpt-3.5-turbo, max_tokens: int 4000, summary_trigger_length: int 10): 初始化记忆管理器。 :param client: OpenAI客户端实例 :param model: 用于摘要和对话的模型 :param max_tokens: 记忆系统允许的最大Token数软限制 :param summary_trigger_length: 当消息条数达到此值时触发摘要压缩 self.client client self.model model self.max_tokens max_tokens self.summary_trigger_length summary_trigger_length self.encoder tiktoken.encoding_for_model(model) # 用于计算Token self.messages [] # 存储完整的消息历史 self.summary # 存储当前的对话摘要 def add_message(self, role: str, content: str): 添加一条新消息到历史记录 self.messages.append({role: role, content: content}) def get_context_messages(self) - list: 获取当前应该发送给LLM的上下文消息列表。 这是记忆管理的核心逻辑决定哪些历史信息被保留在上下文中。 context_messages [] # 1. 始终加入系统提示词如果有的话和当前摘要 if self.summary: # 将摘要作为一条系统消息加入让模型知道这是之前对话的浓缩 context_messages.append({role: system, content: f以下是之前对话的摘要\n{self.summary}\n请基于此摘要和后续对话进行回应。}) # 在实际应用中你的主系统提示词可能在这里加入 # 2. 加入原始消息历史滑动窗口 # 我们这里采用简单的“最近N条”滑动窗口。更复杂的策略可以基于Token数。 recent_messages self.messages[-self.summary_trigger_length:] if self.messages else [] context_messages.extend(recent_messages) # 3. 检查并执行摘要压缩 self._check_and_summarize() return context_messages def _check_and_summarize(self): 检查历史长度如果过长则触发摘要压缩 if len(self.messages) self.summary_trigger_length: print(f[Memory] 历史消息数({len(self.messages)})超过触发阈值({self.summary_trigger_length})开始摘要压缩...) # 我们摘要的对象是“被滑动窗口挤出去”的早期消息以及当前的旧摘要。 to_summarize_messages self.messages[:-self.summary_trigger_length] # 早期消息 if self.summary: # 如果已有摘要将其也作为摘要的输入实现增量摘要 summary_msg {role: system, content: f现有摘要{self.summary}} to_summarize_messages [summary_msg] to_summarize_messages if to_summarize_messages: # 构建摘要提示 prompt [ {role: system, content: 你是一个对话摘要助手。请将以下的对话历史浓缩成一个简洁、连贯的段落摘要保留所有关于用户目标、已达成共识、重要事实和决策的关键信息。摘要将用于后续对话的上下文。}, {role: user, content: json.dumps(to_summarize_messages, ensure_asciiFalse)} ] try: response self.client.chat.completions.create( modelself.model, messagesprompt, max_tokens300, # 控制摘要长度 temperature0.2, ) new_summary response.choices[0].message.content # 更新摘要并移除已被摘要的原始消息 self.summary new_summary self.messages self.messages[-self.summary_trigger_length:] # 只保留最近窗口内的消息 print(f[Memory] 摘要完成。新摘要{self.summary[:100]}...) except Exception as e: print(f[Memory] 摘要生成失败{e}) # 摘要失败降级为更激进的滑动窗口直接丢弃更早的消息 # 可以只保留最近的一半或者直接清空早期消息 keep_count self.summary_trigger_length // 2 self.messages self.messages[-keep_count:] print(f[Memory] 已执行降级策略保留最近{keep_count}条消息。) def count_tokens(self, messages: list) - int: 粗略计算一组消息的Token数OpenAI计算方式更复杂此方法为估算 text .join([f{m[role]}: {m[content]} for m in messages]) return len(self.encoder.encode(text))这个MemoryManager实现了几个关键点滑动窗口get_context_messages方法默认只返回最近summary_trigger_length条消息。触发式摘要当总消息数超过阈值时_check_and_summarize方法会被调用。它将窗口外的早期消息以及旧摘要发送给LLM生成一个新的浓缩摘要。摘要集成新生成的摘要会在下一次get_context_messages时作为一条系统消息插入到上下文的最前面。这样模型就知道了被“遗忘”的细节的精华。降级策略摘要可能失败如API错误我们有一个降级方案直接丢弃更多历史保证系统还能运行。3.2 定义工具与ReAct智能体接下来我们定义两个简单的工具并构建Agent的核心循环。# agent.py from typing import Dict, Any, Callable import re import json class Tool: 工具基类 def __init__(self, name: str, description: str, func: Callable): self.name name self.description description self.func func def run(self, **kwargs): return self.func(**kwargs) class CalculatorTool(Tool): def __init__(self): super().__init__( namecalculator, description用于执行数学计算。输入一个数学表达式字符串如 3 5 * 2。, funcself._calculate ) def _calculate(self, expression: str) - str: try: # 警告使用eval有安全风险仅用于演示。生产环境应用安全表达式求值库。 result eval(expression) return f计算结果{expression} {result} except Exception as e: return f计算错误{e} class SearchTool(Tool): 模拟搜索工具 def __init__(self): super().__init__( namesearch, description用于搜索信息。输入一个查询字符串。, funcself._search ) # 模拟一个简单的知识库 self.knowledge_base { openai: OpenAI是一家专注于人工智能研究的公司创造了GPT系列模型。, react: ReAct是一个让LLM结合推理Reasoning和行动Acting的框架。, agent: AI Agent是能感知环境、做出决策并执行动作以实现目标的智能体。 } def _search(self, query: str) - str: query_lower query.lower() for key, value in self.available_tools.items(): if key in query_lower: return f搜索到信息{value} return f未找到关于 {query} 的明确信息。 class Agent: def __init__(self, client, memory_manager: MemoryManager, tools: Dict[str, Tool], system_prompt: str None): self.client client self.memory memory_manager self.tools tools self.system_prompt system_prompt or 你是一个有帮助的AI助手可以调用工具。请遵循以下步骤 1. 思考用户的问题是否需要使用工具以及使用哪个工具。 2. 如果需要工具请严格按照以下JSON格式回应且只输出这个JSON json {action: tool_call, tool_name: 工具名, tool_input: 工具的输入参数}如果不需要工具或者工具返回结果后需要进一步回应请用自然语言直接回答用户。 请确保你的思考过程在内心完成最终输出只能是JSON或自然语言不要混合。def run(self, user_input: str): 处理一轮用户输入 # 1. 将用户输入存入记忆 self.memory.add_message(user, user_input)# 2. 构建本次请求的完整上下文 context_messages [{role: system, content: self.system_prompt}] context_messages.extend(self.memory.get_context_messages()) # 注意get_context_messages已经包含了摘要和最近的历史。 max_attempts 5 for attempt in range(max_attempts): # 3. 调用LLM response self.client.chat.completions.create( modelself.memory.model, messagescontext_messages, temperature0.1, # 低温度使输出更稳定更适合工具调用 max_tokens500, ) llm_output response.choices[0].message.content print(f\n[Agent 思考] (尝试{attempt1}): {llm_output[:200]}...) # 4. 解析LLM输出判断是工具调用还是直接回复 tool_call_match re.search(rjson\s*(.*?)\s*, llm_output, re.DOTALL) if not tool_call_match: # 尝试直接解析为JSON如果LLM没加代码块 try: data json.loads(llm_output) if isinstance(data, dict) and data.get(action) tool_call: tool_call_match True tool_name data[tool_name] tool_input data[tool_input] except json.JSONDecodeError: # 不是JSON视为自然语言回复 pass if tool_call_match: # 解析工具调用 try: if tool_call_match is not True: # 来自代码块匹配 data json.loads(tool_call_match.group(1)) # 执行工具 if data[tool_name] in self.tools: tool self.tools[data[tool_name]] tool_input data.get(tool_input, ) print(f[Agent 行动] 调用工具 {tool.name}输入: {tool_input}) tool_result tool.run(**{expression: tool_input} if tool.namecalculator else {query: tool_input}) print(f[工具结果] {tool_result}) # 将工具调用和结果作为对话历史的一部分存入记忆 # 注意这里我们把LLM的“工具调用决策”和“工具结果”都存进去帮助模型理解上下文。 # 一种更清晰的存法是存一个结构化的“工具调用记录”。 self.memory.add_message(assistant, f[决定调用工具 {tool.name}]) self.memory.add_message(user, f[工具 {tool.name} 返回结果] {tool_result}) # 重要工具调用后需要让LLM基于结果继续回应。我们通过循环下一次迭代来实现。 # 在下一次循环中context_messages会通过memory.get_context_messages()包含新的工具交互历史。 continue # 继续循环让LLM处理工具结果 else: error_msg f未知工具{data[tool_name]} self.memory.add_message(assistant, f[工具调用错误] {error_msg}) except (json.JSONDecodeError, KeyError) as e: error_msg f解析工具调用失败{e}原始输出{llm_output} self.memory.add_message(assistant, f[错误] {error_msg}) else: # 自然语言回复直接输出并存入记忆 print(f\n[Agent 回复]: {llm_output}) self.memory.add_message(assistant, llm_output) break # 本轮对话结束跳出循环 else: # 循环结束仍未跳出可能陷入工具调用循环 final_msg 对话似乎陷入循环或出现问题。让我们重新开始。 print(f\n[Agent 回复]: {final_msg}) self.memory.add_message(assistant, final_msg)### 3.3 运行与测试观察记忆的作用 现在让我们写一个主程序来测试这个带记忆的Agent。 python # main.py from openai import OpenAI from memory_manager import MemoryManager from agent import Agent, CalculatorTool, SearchTool def main(): # 初始化 client OpenAI(api_key你的OpenAI_API_KEY) # 请替换为你的Key model gpt-3.5-turbo # 创建记忆管理器设置当消息超过5条时触发摘要 memory MemoryManager(client, model, summary_trigger_length5) # 创建工具集 tools { calculator: CalculatorTool(), search: SearchTool(), } # 创建Agent agent Agent(client, memory, tools) print(*50) print(带短期记忆的AI Agent已启动。) print(系统提示我可以使用计算器(calculator)和搜索(search)工具。) print(输入 quit 或 exit 退出。) print(*50) # 模拟一个多轮对话展示记忆如何工作 test_dialogue [ 你好请介绍下你自己。, OpenAI这家公司是做什么的, 用计算器算一下 123 乘以 456 等于多少。, 刚才我们聊到的OpenAI它最著名的产品是什么, # 这里需要记忆 再算一下刚才那个乘积加上 1000 是多少。, # 这里需要记忆 什么是ReAct框架, 总结一下我们刚才关于AI公司都聊了些什么。 # 这里需要记忆和摘要 ] for query in test_dialogue: print(f\n[用户] {query}) input(按回车键继续...) # 方便观察每一步 agent.run(query) # 交互模式 while True: try: user_input input(\n[你] ) if user_input.lower() in [quit, exit, q]: break agent.run(user_input) except KeyboardInterrupt: break print(\n对话结束。) if __name__ __main__: main()运行这个程序你会观察到在前几轮记忆是完整的原始消息。当对话轮次超过5条我们的summary_trigger_length后控制台会打印[Memory] 历史消息数(...)超过触发阈值(...)开始摘要压缩...。之后Agent在回答“刚才我们聊到的OpenAI...”和“再算一下刚才那个乘积...”时之所以能正确理解指代就是因为摘要里包含了“OpenAI是一家AI公司”以及最近的对话历史滑动窗口里包含了“123*45656088”这个结果。如果没有记忆管理Agent根本无法回答这两个问题。4. 记忆策略的进阶思考与优化方向上面的实现是一个基础Demo。在实际项目中短期记忆的设计要复杂得多需要考虑更多维度。4.1 不同记忆策略的对比与选型策略优点缺点适用场景完整历史信息无损逻辑最连贯Token消耗增长快成本高可能超窗口对话轮次极少10的简单场景固定长度滑动窗口实现简单内存可控会丢失窗口外的所有信息可能遗忘核心目标任务目标明确且集中在最近对话的场景摘要压缩极大节省Token保留长期语义摘要过程有信息损耗和额外API成本摘要质量影响大中长篇幅对话需要兼顾长期和短期记忆关键信息提取记忆效率极高高度结构化设计复杂需要预定义schema提取可能不准目标导向强的任务型Agent如客服、工作流混合策略灵活平衡效果较好实现和维护复杂绝大多数生产环境推荐实操心得不要追求“最完美”的记忆策略。从最简单的滑动窗口开始根据实际观察到的Agent“失忆”情况比如用户频繁问“刚才说的XXX”再逐步引入摘要或关键信息提取。过早优化是万恶之源。4.2 摘要质量的挑战与提升技巧摘要环节是记忆压缩的瓶颈。一个糟糕的摘要会让Agent基于错误的理解进行后续对话。问题1摘要丢失关键细节。比如用户说“我对花生严重过敏”摘要可能只变成“用户有食物过敏”。优化在系统提示词中强调必须保留的具体信息类型如“精确的数字、日期、名称、否定词不、禁止、极端形容词严重、必须等”。问题2摘要引入幻觉。LLM可能总结出原文没有的意思。优化使用更低温度temperature0或0.1并要求摘要“严格基于提供的历史不要添加任何未提及的信息”。问题3摘要冗长。失去了压缩的意义。优化明确限制摘要的Token数或句子数量例如“用不超过3句话总结”。一个改进的摘要提示词示例你负责生成对话摘要。请严格遵循 1. 基于提供的对话历史提取事实、用户目标、达成的共识和重要决定。 2. 必须保留具体数字、时间、人名、地名、物品名、用户明确表达的喜好/厌恶/禁忌。 3. 禁止添加任何历史中未出现的信息或主观推测。 4. 用中文输出语言简洁连贯长度控制在100字以内。 对话历史{history}4.3 记忆与工具调用的深度集成在我们的Demo中工具调用记录也被简单存为对话消息。这可行但不够优雅。更专业的做法是设计一个Interaction或Turn对象结构化存储每一轮的用户输入、Agent的思考过程Reasoning、工具调用Action、观察结果Observation。这样记忆管理器可以更精细地决定哪些部分需要保留如思考过程和关键观察哪些部分可以压缩或丢弃如冗长的原始工具输出。4.4 Token计算的精确性与成本控制我们示例中的count_tokens方法是粗略估算。OpenAI的Token计算方式对不同模型、不同角色如name字段有细微差别。对于生产环境建议使用OpenAI官方提供的tiktoken库进行精确计算。为记忆系统设置一个硬性的Token上限如模型上限的70%并实现一个trim_messages函数当上下文即将超限时优先移除最旧的、非核心的消息如冗长的寒暄或强制触发摘要。实现短期记忆就像为Agent安装了一个高速缓存Cache。它不能替代长期记忆向量数据库但对于完成一个连贯的会话或任务流至关重要。从理解上下文窗口的限制开始选择适合你场景的记忆策略在实践中不断观察和调整你的Agent才会真正“记住”用户成为一个得力的助手。