本文通过作者面试失败的经历,深入分析了传统Agent记忆系统的局限性,提出了一套完善的记忆架构。包括短期记忆的检查点机制、三层分级记忆架构(资源、条目、类别)、图谱与向量的混合结构,以及记忆的衰减与维护机制。文章强调了记忆是Agent的基础设施而非简单功能,解决了记忆冲突、衰减和检索效率等问题,为构建真正永不遗忘的Agent提供了实用方案。
关于Agent的Memory(记忆系统),我之前的认知很浅薄,就是Embedding,检索,如果分层的话,就是长期短期记忆,或者个人记忆,集体记忆。从来没有想过关于记忆还有哪些更深刻的见解,比如记忆冲突怎么解决?记忆衰减问题怎么处理?直到看到Rohit的这篇文章。文中描述他去面试,本来信心满满,结果因为对记忆了解不够深刻结果铩羽而归。文中的对记忆的深度描述很值得我借鉴,无论是实战,还是面试。现在分享给朋友们。
Rohit 的正文如下。
三个月前,我因为没能构建出一个永不遗忘的 Agent,在一次技术面试中被淘汰了。
我所知道的每一种方法都曾奏效……直到它失效。
我走进面试房间时充满自信。我做过聊天机器人,我理解 Embedding,我知道如何使用向量数据库。
但当面试官要求我设计一个 Agent,使其能够跨越数周而非仅在一个对话中记住用户的偏好时,我僵住了。
我的本能是标准的做法:将所有内容存储在向量数据库中,并在需要时检索相似的对话。
那些击垮我的问题很简单:规模如何处理?在一千次会话之后,你如何处理冲突的数据?你如何阻止它为了填补空白而伪造记忆?
我无言以对。
这次失败迫使我深入研究并找到一个解决方案:
大多数关于“有记忆的 Agent”的教程,教的都是如何用 RAG 来实现记忆。
问题不在于 Embedding。不在于 Token 限制。甚至不在于检索本身。
问题在于,记忆是基础设施,而不是一个功能。
以下是我为解决这个问题而构建的整个系统,以及我使用的代码。
1标准记忆的“死穴”
我曾认为的记忆是:保留对话历史,并将其塞入上下文窗口。
这对于大约 10 次交流是有效的。然后上下文窗口就会被填满。
于是你截断旧消息。现在你的 Agent 忘记了用户是素食主义者,并推荐了一家牛排馆。
你意识到对话历史不是记忆,它只是一个聊天记录。
“好吧,”我想。“我会嵌入每条消息,并使用相似性搜索来检索相关的消息。”
这在一段时间内效果更好。
但两周后,向量数据库有了 500 个条目。当用户问:“我跟你说过我的工作情况是什么?”检索系统返回了来自 12 个不同对话的片段。
Agent 看到的是:
- “我热爱我的工作”(第 1 周)
- “我在考虑辞职”(第 2 周)
- “我的经理很支持我”(第 1 周)
- “我的经理事无巨细地管理一切”(第 2 周)
哪一个是真的?
Agent 毫无头绪。它幻觉出一个综合结论:“你热爱你支持你的经理,但你正因为事无巨细的管理而考虑辞职。”
完全错误。用户在第 1 周和第 2 周之间换了工作。
这是关键的认知:Embedding 衡量的是相似性,而不是真相。
向量数据库有一个盲点:它们不理解时间、上下文或更新。它们只是吐出在数学上看起来与你所问内容接近的文本。这不是记忆;这是猜测。
修复需要一次思维上的转变。记忆不是一块硬盘。它是一个过程。你不能只存储数据;你必须赋予它生命周期,让它进化。
2短期记忆:CheckPoint 机制
在处理困难的部分(长期记忆)之前,我们需要处理短期连续性。
短期记忆是记住 30 秒前说了什么的能力。这实际上是一个已解决的问题。
解决方案是检查点(Checkpointing)。
每个 Agent 都作为一个状态机运行。它接收输入,更新内部状态,调用工具,生成输出,并再次更新状态。检查点是这个完整状态在特定时刻的快照。
这为你提供了三种能力:
确定性:重放任何对话。
可恢复性:如果 Agent 崩溃,可以从上次离开的地方精确恢复。
可调试性:回溯以检查 Agent 的“思考过程”。
在生产环境中,我使用 Postgres 支持的检查点。模式如下:
Python 中的检查点代码
这处理了“现在”。但检查点是短暂的。它们不会积累智慧。为此,我们需要长期架构。
3长期记忆核心:三层分级架构
经过数月的失败,我找到了两种真正有效的架构。
这模仿了人类对知识的分类方式。它最适用于助手、治疗师或伴侣型 Agent。
三层分级体系:
第 1 层:资源(原始数据)。真相的来源。未经处理的日志、上传、转录。不可变且带有时间戳。
第 2 层:条目(原子事实)。从资源中提取的离散事实(“用户偏好 Python”,“用户对贝类过敏”)。
第 3 层:类别(演进中的总结)。高层上下文。条目被分组到类似
work_preferences.md或personal_life.md的文件中。
写入路径:主动记忆
当新信息到达时,系统不会只是将其归档,而是会处理它。它会调出该类别的现有总结,并主动将新细节编织到叙事中。这会自动处理矛盾:如果用户提到他们已转向 Rust,系统不会只是将“Rust”添加到列表中;它会重写配置文件以替换旧的偏好。
import json class FileBasedMemory: def memorize(self, conversation_text, user_id): # Stage 1: Resource Ingestion (The Source of Truth) # Always save the raw input first. This allows for traceability. resource_id = self.save_resource(user_id, conversation_text) # Stage 2: Extraction # Extract atomic facts from the conversation. items = self.extract_items(conversation_text) # Stage 3: Batching (The Fix) # Group items by category to avoid opening/writing files multiple times. # Structure: { "work_life": ["User hates Java", "User loves Python"], ... } updates_by_category = {} for item in items: cat = self.classify_item(item) if cat not in updates_by_category: updates_by_category[cat] = [] updates_by_category[cat].append(item['content']) # Link item to the specific resource for traceability self.save_item(user_id, category=cat, item=item, source_resource_id=resource_id) # Stage 4: Evolve Summaries (One Write Per Category) for category, new_memories in updates_by_category.items(): existing_summary = self.load_category(user_id, category) # We pass the LIST of new memories, not just one updated_summary = self.evolve_summary( existing=existing_summary, new_memories=new_memories ) self.save_category(user_id, category, updated_summary) def extract_items(self, text): """Use LLM to extract atomic facts""" prompt = f"""Extract discrete facts from this conversation. Focus on preferences, behaviors, and important details. Conversation: {text} Return as JSON list of items.""" return llm.invoke(prompt) def evolve_summary(self, existing, new_memories): """ Update category summary with a BATCH of new information. """ # Convert list to bullet points for the prompt memory_list_text = "\n".join([f"- {m}" for m in new_memories]) prompt = f"""You are a Memory Synchronization Specialist. Topic Scope: User Profile ## Original Profile {existing if existing else "No existing profile."} ## New Memory Items to Integrate {memory_list_text} # Task 1. Update: If new items conflict with the Original Profile, overwrite the old facts. 2. Add: If items are new, append them logically. 3. Output: Return ONLY the updated markdown profile.""" return llm.invoke(prompt) # Helper stubs def save_resource(self, user_id, text): pass def save_item(self, user_id, category, item, source_resource_id): pass def save_category(self, user_id, category, content): pass def load_category(self, user_id, category): return "" def classify_item(self, item): return "general"读取路径(分层检索):
为了节省 Token,你不会拉取所有内容。
- 拉取类别总结。
- 询问 LLM:“这足够了吗?”
- 如果足够 -> 回应。
- 如果不够 -> 深入到特定条目。
class FileBasedRetrieval: def retrieve(self, query, user_id): # Stage 1: Category Selection (The Fix) # Instead of loading ALL content, we just list category NAMES and ask # the LLM which ones might contain the answer. all_categories = self.list_categories(user_id) relevant_categories = self.select_relevant_categories(query, all_categories) # Load only the relevant summaries summaries = {cat: self.load_category(user_id, cat) for cat in relevant_categories} # Stage 2: Sufficiency Check # Check if the high-level summaries answer the query if self.is_sufficient(query, summaries): return summaries # Stage 3: Hierarchical Search # If summaries are vague, generate a specific query to find atomic items # or raw resources. search_query = self.generate_search_query(query, summaries) # Search Level 1: Atomic Items (Extracted facts) items = self.search_items(user_id, search_query) if items: return items # Search Level 2: Raw Resources (Full text search fallback) resources = self.search_resources(user_id, search_query) return resources def select_relevant_categories(self, query, categories): """Filter to only the categories likely to hold the answer""" prompt = f"""Query: {query} Available Categories: {', '.join(categories)} Return a JSON list of the categories that are most relevant to this query.""" return llm.invoke(prompt) def is_sufficient(self, query, summaries): prompt = f"""Query: {query} Summaries: {summaries} Can you answer the query comprehensively with just these summaries? YES/NO""" return 'YES' in llm.invoke(prompt)4高级记忆:图谱与向量的混合结构
这对于叙事连贯性非常有效。但它难以处理复杂的关系。为此,你需要图谱。
文件式记忆难以处理复杂关系。对于精确的系统(CRM、研究),你需要一个图谱(Graph)。
混合结构
向量存储:用于发现,用于浮现相关或相似的文本。
知识图谱:用于精确性,将事实存储为主语-谓语-宾语关系。
基于图谱的记忆,带有冲突解决机制。
检索涉及并行运行搜索(向量 + 图谱遍历)并合并结果。
基于图谱的混合检索。
混合搜索
检索并行运行两次搜索:
向量搜索:查找语义相似的对话。
图谱遍历:查找与查询相关联的实体。
结果合并为一个统一的上下文。这防止了“记得一切但一无所知”的问题。
5Agent 记忆的“新陈代谢”:衰减与维护
这是没有人告诉你的:记忆必须衰减。
“永不遗忘”不意味着“记住每一个 Token”。它意味着“记住重要的事情”。
如果你不修剪数据库,你的 Agent 会变得困惑、缓慢且昂贵。
我运行后台 Cron 任务来保持系统健康:
夜间整合
每天凌晨 3 点,一个后台进程会回顾当天的对话。它寻找 Agent 在实时操作中遗漏的模式。它合并冗余的记忆。它将频繁访问的条目提升到更高优先级的存储。
每周总结
每周一次,系统会重新总结类别文件。它将旧条目压缩成更高层次的洞察。它修剪 90 天内未被访问的记忆。
每月重新索引
每月,我们会对记忆存储运行一次完整的重新索引。Embedding 会用最新的模型版本重建,图谱边缘会根据实际使用情况进行调整。任何一段时间未被触及的内容都会被归档。
# Memory maintenance cron job class MemoryMaintenance: def run_nightly_consolidation(self, user_id): """Run every night to consolidate memories""" # Get today's conversations recent_memories = self.get_memories_since(user_id, hours=24) # Identify redundant memories duplicates = self.find_duplicates(recent_memories) # Merge duplicates for group in duplicates: merged = self.merge_memories(group) self.replace_memories(group, merged) # Promote frequently accessed memories hot_memories = self.get_high_access_memories(user_id) for memory in hot_memories: self.increase_priority(memory) def run_weekly_summarization(self, user_id): """Run weekly to compress old memories""" # Get memories older than 30 days old_memories = self.get_memories_older_than(user_id, days=30) # Group by category categories = self.group_by_category(old_memories) # Summarize each category for category, memories in categories.items(): summary = self.create_summary(memories) self.archive_old_items(memories) self.save_summary(user_id, category, summary) # Prune rarely accessed memories stale = self.get_memories_not_accessed(user_id, days=90) self.archive_memories(stale) def run_monthly_reindex(self, user_id): """Run monthly to optimize the memory store""" all_memories = self.get_all_memories(user_id) # Regenerate embeddings for memory in all_memories: new_embedding = self.generate_embedding(memory.text) memory.embedding = new_embedding # Re-weight graph edges if self.using_graph: self.graph.reweight_edges_by_access() # Archive dead nodes dead_nodes = self.graph.find_unused_nodes(days=180) self.graph.archive_nodes(dead_nodes)这种维护能让记忆系统保持数月健康。
没有它,它们就会腐烂。
6生产级检索与注入逻辑
大多数检索系统失败是因为它们只依赖向量相似性。这是一个错误。一个健壮的记忆系统是反向工作的,它从上下文窗口的限制出发。它从使用合成查询进行广泛搜索开始,而不是原始用户输入。然后,它将这些搜索结果视为候选,而不是答案。我们通过一个“相关性评分器”和一个“时间衰减”函数来过滤这些候选。这确保了一个稍微不那么相关但时间上非常近的记忆,往往能击败六个月前的一个完美匹配。结果是一个提示,其中只包含 5-10 个真正有用的记忆 Token,而不是一堵相似文本的墙。
# Retrieval and injection logic class MemoryRetrieval: def retrieve_for_inference(self, user_message, user_id, max_tokens=2000): # Stage 1: Generate search query search_query = self.generate_query(user_message) # Stage 2: Semantic search candidates = self.vector_store.search( query=search_query, user_id=user_id, top_k=20 ) # Stage 3: Relevance filtering relevant = [] for candidate in candidates: score = self.calculate_relevance( candidate, user_message ) if score > 0.7: relevant.append((score, candidate)) # Stage 4: Temporal ranking ranked = [] for score, memory in relevant: age_days = (now() - memory.timestamp).days time_decay = 1.0 / (1.0 + (age_days / 30)) final_score = score * time_decay ranked.append((final_score, memory)) ranked.sort(reverse=True, key=lambda x: x[0]) # Stage 5: Context assembly selected_memories = [] token_count = 0 for score, memory in ranked: memory_tokens = self.count_tokens(memory.text) if token_count + memory_tokens > max_tokens: break selected_memories.append({ 'text': memory.text, 'timestamp': memory.timestamp, 'confidence': score }) token_count += memory_tokens return self.format_memory_context(selected_memories) def format_memory_context(self, memories): """Format memories for injection into prompt""" context = "=== RELEVANT MEMORIES ===\n\n" for mem in memories: context += f"[{mem['timestamp']}] (confidence: {mem['confidence']:.2f})\n" context += f"{mem['text']}\n\n" context += "=== END MEMORIES ===\n" return context这确保了 Agent 只看到它需要的东西。不多不少。
7总结:Agent 记忆的五个致命错误
在构建完这个系统后,我明白了为什么我在那次面试中失败了。大多数实现在生产环境中失败,是因为它们犯了五个关键错误:
错误 1:永远存储原始对话对话是嘈杂的。如果你存储每一个“嗯”和“像”,你的记忆就会被污染。提取事实,而不是转录本。
错误 2:盲目使用 EmbeddingEmbedding 找到的是相似性,而不是真相。“我热爱我的工作”和“我讨厌我的工作”嵌入后非常相似。你需要解决冲突的逻辑。
错误 3:没有记忆衰减没有衰减,你的 Agent 就会淹没在过去。它记住了你两年前的度假计划,却忘记了你当前的截止日期。
错误 4:没有写入规则如果 Agent 随时随地写入记忆,它就会写入垃圾。为值得记住的内容定义明确的规则。
错误 5:将记忆视为聊天历史这是致命的错误。聊天历史是短暂的。记忆是对所学知识的结构化表示。
真正的突破发生在我们不再将 Agent 视为简单的聊天机器人,而是开始将其视为操作系统。一个 Agent 需要完全相同的功能:
进程管理:跟踪多个并发任务。
内存管理:分配、更新和释放知识。
I/O 管理:与工具和用户交互。
最重要的是,它需要一个复杂的记忆架构。你需要“RAM”来处理当前对话的快速、易失性上下文,但你也需要一个“硬盘”——一种持久的、索引化的方式来存储在会话结束后仍然存在的知识。如果你不对该记忆进行定期维护,就像垃圾回收一样,系统最终会崩溃。
三个月前:
基本记忆
今天:
应用混合架构后
聊天机器人和伴侣之间的区别在于记忆。
记忆和好的记忆之间的区别在于架构。
如果你正在构建 Agent,这不再是可选项。用户期望持久性。他们期望学习。他们期望 Agent 记住他们是谁。
三个月前,我无法构建这个。现在,我已经交付了能够跨越数千次会话记住客户偏好的 Agent。
那次感觉像失败的面试淘汰,成为了理解生产系统真正需要什么的催化剂。
存储很便宜。结构很难。但结构是将一个无状态的语言模型转变为真正永不遗忘的东西。
未来的 Agent 不会只是拥有更多的参数或更好的训练数据。它们将拥有能够学习、进化并随着每一次交互而改进的记忆系统。
现在你知道如何构建它们了。
最后
我在一线科技企业深耕十二载,见证过太多因技术更迭而跃迁的案例。那些率先拥抱 AI 的同事,早已在效率与薪资上形成代际优势,我意识到有很多经验和知识值得分享给大家,也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。
我整理出这套 AI 大模型突围资料包:
- ✅AI大模型学习路线图
- ✅Agent行业报告
- ✅100集大模型视频教程
- ✅大模型书籍PDF
- ✅DeepSeek教程
- ✅AI产品经理入门资料
完整的大模型学习和面试资料已经上传带到CSDN的官方了,有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇
为什么说现在普通人就业/升职加薪的首选是AI大模型?
人工智能技术的爆发式增长,正以不可逆转之势重塑就业市场版图。从DeepSeek等国产大模型引发的科技圈热议,到全国两会关于AI产业发展的政策聚焦,再到招聘会上排起的长队,AI的热度已从技术领域渗透到就业市场的每一个角落。
智联招聘的最新数据给出了最直观的印证:2025年2月,AI领域求职人数同比增幅突破200%,远超其他行业平均水平;整个人工智能行业的求职增速达到33.4%,位居各行业榜首,其中人工智能工程师岗位的求职热度更是飙升69.6%。
AI产业的快速扩张,也让人才供需矛盾愈发突出。麦肯锡报告明确预测,到2030年中国AI专业人才需求将达600万人,人才缺口可能高达400万人,这一缺口不仅存在于核心技术领域,更蔓延至产业应用的各个环节。
资料包有什么?
①从入门到精通的全套视频教程⑤⑥
包含提示词工程、RAG、Agent等技术点
② AI大模型学习路线图(还有视频解说)
全过程AI大模型学习路线
③学习电子书籍和技术文档
市面上的大模型书籍确实太多了,这些是我精选出来的
④各大厂大模型面试题目详解
⑤ 这些资料真的有用吗?
这份资料由我和鲁为民博士共同整理,鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位,在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利,同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。
所有的视频教程由智泊AI老师录制,且资料与智泊AI共享,相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。
资料内容涵盖了从入门到进阶的各类视频教程和实战项目,无论你是小白还是有些技术基础的,这份资料都绝对能帮助你提升薪资待遇,转行大模型岗位。
智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念,通过动态追踪大模型开发、数据标注伦理等前沿技术趋势,构建起"前沿课程+智能实训+精准就业"的高效培养体系。
课堂上不光教理论,还带着学员做了十多个真实项目。学员要亲自上手搞数据清洗、模型调优这些硬核操作,把课本知识变成真本事!
如果说你是以下人群中的其中一类,都可以来智泊AI学习人工智能,找到高薪工作,一次小小的“投资”换来的是终身受益!
应届毕业生:无工作经验但想要系统学习AI大模型技术,期待通过实战项目掌握核心技术。
零基础转型:非技术背景但关注AI应用场景,计划通过低代码工具实现“AI+行业”跨界。
业务赋能 突破瓶颈:传统开发者(Java/前端等)学习Transformer架构与LangChain框架,向AI全栈工程师转型。
👉获取方式:
😝有需要的小伙伴,可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓**