10分钟搞懂大模型Agent记忆系统四层架构，附Python实现

花 10 分钟搞清楚 Agent 记忆系统的四层架构。目录什么是 Agentic Memory四种记忆类型2.1 上下文记忆In-context Memory2.2 外部记忆External Memory2.3 情景记忆Episodic Memory2.4 语义/参数记忆Semantic/Parametric Memory记忆如何在 Agent Loop 中流动构建记忆层4.1 MemoryStore 类4.2 EpisodicLogger 类4.3 整合记忆增强 Agent向量数据库5.1 相似性搜索原理记忆管理6.1 基于时间的衰减6.2 写入时的重要性打分6.3 定期整合总结想象某天你雇了一个天才员工。第一天她表现惊艳发现了所有 bug、写了清晰的文档、甚至提出了你没想到的改进。你很满意。第二天你走过去说“嘿还记得昨天我们讨论的那个问题吗”她停顿了一下看着你微微一笑。“抱歉……什么问题”没有记忆没有上下文。完全忘了。你在写这段的时候也会和我一样意外。对LLM来说这是常态 。每次新对话都是从零开始。模型不记得你是谁、不记得你们一起做过什么甚至不记得几分钟前在另一个窗口讨论了什么。对于简单聊天机器人这无所谓。但对于 Agent——那种跑任务、做决策、随时间改进的系统——这种失忆是致命的。真正的智能不只是响应得良好还关乎记忆、学习以及在既有基础上进化。记忆把一个无状态系统变成了能真正演进的智能体。什么是 Agentic MemoryAgentic memory 不是单一组件它更像一套幕后系统不同类型的存储、不同检索方式、以及智能管理策略让 Agent 能真正跨时间携带上下文。说白了记忆同时在做三件有区别的事。连续性Continuity关乎身份。Agent 如何知道你是谁、你偏好什么、以及你们已经一起建立了什么。没有它每次交互都像从零开始。上下文Context关乎当前任务。刚发生了什么、用了什么工具、返回了什么结果、以及下一步需要做什么。它让多步骤工作流不至于崩掉。学习Learning关乎变得更好。理解什么有效、什么无效、慢慢改进决策而不是重复同样的错误。三者合一让 Agent 每次交互都更一致、更可靠、甚至更智能一点。一个设计良好的 Agent 记忆系统同时处理这三件事每种用不同的存储后端。四种记忆类型业界已经沉淀出四种有区别的记忆类型。可以把它们理解成大脑的四个区域每个各司其职。2.1 上下文记忆In-context Memory上下文窗口是 Agent 的**工作台。**上面的东西都能即时访问模型可以在单次前向传播中推理。不需要检索步骤。但工作台有尺寸限制。每个 token 都要花钱和时间。会话结束时工作台会被清空。上下文窗口里有什么System promptAgent 人设、规则、能力、当前日期/用户信息对话历史本次会话的来回交互工具调用结果Agent 刚调用的工具输出检索到的记忆从外部存储拉取的片段Scratchpad中间推理逐步思考输出滑动窗口问题长对话中历史积累最终会溢出上下文限制。简单截断最旧消息会丢失重要的早期上下文。更好的策略**摘要Summarization**定期把旧的对话压缩成简短摘要用摘要替换**选择性保留Selective retention**保留包含关键事实、决策或工具结果的对话丢弃闲聊**卸载到外部记忆**把重要事实提取到向量存储需要时检索2.2 外部记忆External Memory外部记忆是任何持久化在模型之外的东西——数据库、向量存储、键值存储、文件。它跨越会话边界存活。如果存储得当Agent 能记住六个月前的事。外部存储有两种形式**结构化存储精确查询**PostgreSQL、Redis、SQLite。按 key、ID 或 SQL 查询。快速、可预测适合用户画像、偏好和结构化数据。**向量存储语义搜索**Pinecone、Chroma、pgvector。按语义查询“找与这个概念相似的记忆”。对非结构化笔记和情景回忆至关重要。检索步骤是瓶颈。检索不到正确的记忆Agent 就当那些记忆不存在。记忆架构搞得好不好80% 看检索设计。2.3 情景记忆Episodic Memory情景记忆是最被低估的类型。外部记忆存储事实情景记忆存储事件——具体来说是过去行为的结果。最简单形式是结构化日志每次 Agent 完成一个任务它记录发生了什么。随着时间这个日志成为丰富的自我知识来源Agent 可以在做决策前查阅。一个 episode 长这样{ episode_id: ep_20240315_003, timestamp: 2024-03-15T14:23:11Z, task: Summarize 50-page PDF into 3 bullet points, approach: Sequential chunking, 2000 tokens per chunk, outcome: success, duration_ms: 4820, token_cost: 12400, quality_score: 0.91, notes: Worked well. Hierarchical chunking would be faster., embedding: [0.023, -0.441, 0.182, /* ... 1536 dims */]}新任务进来时Agent 检索语义最相似的老 episode用它们来选择策略。说白了这是从个人历史做 few-shot learning而不是从手工数据集聚类。反思循环2.4 语义/参数记忆Semantic/Parametric Memory这是模型与生俱来的记忆。所有东西在训练时编码进权重世界事实、语言模式、推理策略、编码规范、文化知识。它一直都在。Agent 从不需要检索它。但它有硬限制**训练时冻结**模型不知道 cutoff 日期之后发生了什么**运行时无法更新**不重训练或微调就无法注入新的永久事实**不透明**无法检查模型到底知道什么或不知道什么**容易幻觉**模型用似是而非的错误内容填补空白对于时间敏感、领域特定或私有的内容不要依赖参数记忆。用外部检索。参数记忆是你在没有更好来源时的通用世界知识 fallback。正确的思维模型参数记忆是 Agent 的通识教育。外部、情景和上下文记忆是 Agent 的在职经验。最好的 Agent 两者结合。记忆如何在 Agent Loop 中流动让我们整合起来。以下是 Agent 每次处理请求时发生的事——展示每个记忆系统都在运作。注意记忆操作是包裹 LLM 调用的先检索、再写入。模型本身无状态是记忆系统给了无状态的 Agent 有状态的假象。构建记忆层上代码。用 Python OpenAI 做嵌入、ChromaDB 做本地向量存储。概念是通用的换库就行。pip install chromadb openai anthropic python-dotenv4.1 MemoryStore 类这是记忆层的核心写入带嵌入和语义检索。import chromadbfrom openai import OpenAIfrom datetime import datetimeimport json, uuidclass MemoryStore: Persistent vector memory for an AI agent. def __init__(self, agent_id: str, persist_dir: str ./memory_db): self.agent_id agent_id self.openai OpenAI() # ChromaDB stores vectors on disk, persists across restarts self.client chromadb.PersistentClient(pathpersist_dir) self.collection self.client.get_or_create_collection( namefagent_{agent_id}_memories, metadata{hnsw:space: cosine} # cosine similarity ) def _embed(self, text: str) - list[float]: Convert text to embedding vector using OpenAI. response self.openai.embeddings.create( modeltext-embedding-3-small, inputtext ) return response.data[0].embedding def remember( self, content: str, memory_type: str general, metadata: dict None ) - str: Store a memory. Returns the memory ID. memory_id str(uuid.uuid4()) embedding self._embed(content) meta { type: memory_type, timestamp: datetime.utcnow().isoformat(), agent_id: self.agent_id, **(metadata or {}) } self.collection.add( ids[memory_id], embeddings[embedding], documents[content], metadatas[meta] ) return memory_id def recall( self, query: str, k: int 5, memory_type: str None, min_relevance: float 0.6 ) - list[dict]: Retrieve the k most relevant memories for a query. query_embedding self._embed(query) where {type: memory_type} if memory_type else None results self.collection.query( query_embeddings[query_embedding], n_resultsk, wherewhere, include[documents, metadatas, distances] ) memories [] for doc, meta, dist in zip( results[documents][0], results[metadatas][0], results[distances][0] ): relevance 1 - dist # cosine distance → similarity if relevance min_relevance: memories.append({ content: doc, metadata: meta, relevance: round(relevance, 3) }) return sorted(memories, keylambda x: x[relevance], reverseTrue) def forget(self, memory_id: str): Delete a specific memory (GDPR compliance, stale data, etc.) self.collection.delete(ids[memory_id])4.2 EpisodicLogger 类在 MemoryStore 之上叠加坡情日志层。from .store import MemoryStorefrom dataclasses import dataclass, asdictfrom typing import Optionalimport timedataclassclass Episode: task: str approach: str outcome: str # success | partial | failure duration_ms: int token_cost: int quality_score: float # 0.0 – 1.0, set by evaluator or user notes: str error: Optional[str] Noneclass EpisodicLogger: def __init__(self, memory_store: MemoryStore): self.store memory_store def log(self, episode: Episode): Save an episode to memory as a searchable document. # Build a rich text representation for semantic search doc ( fTask: {episode.task}\n fApproach: {episode.approach}\n fOutcome: {episode.outcome}\n fNotes: {episode.notes} ) self.store.remember( contentdoc, memory_typeepisode, metadata{ outcome: episode.outcome, quality_score: episode.quality_score, duration_ms: episode.duration_ms, token_cost: episode.token_cost, } ) def recall_similar(self, task: str, k: int 3) - list[dict]: Find past episodes similar to the current task. return self.store.recall( querytask, kk, memory_typeepisode, min_relevance0.65 )4.3 整合记忆增强 Agentimport anthropicfrom memory.store import MemoryStorefrom memory.episodic import EpisodicLogger, Episodeimport timeclass MemoryAugmentedAgent: def __init__(self, agent_id: str): self.client anthropic.Anthropic() self.memory MemoryStore(agent_id) self.episodes EpisodicLogger(self.memory) def _build_memory_context(self, user_message: str) - str: Retrieve relevant memories and format them for injection. # Semantic search for related facts memories self.memory.recall(user_message, k4) # Similar past task approaches episodes self.episodes.recall_similar(user_message, k2) context_parts [] if memories: context_parts.append(## Relevant memories\n \n.join([ f- [{m[metadata][type]}] {m[content]} f (relevance: {m[relevance]}) for m in memories ]) ) if episodes: context_parts.append(## Past similar tasks\n \n.join([ f- {e[content][:200]}... for e in episodes ]) ) return \n\n.join(context_parts) if context_parts else def run(self, user_message: str) - str: start time.time() # 1. Retrieve relevant memory memory_context self._build_memory_context(user_message) # 2. Build system prompt with injected memory system You are a helpful agent with memory.You have access to relevant context from past interactions.Use this context to give better, more personalized responses. if memory_context: system f\n\n{memory_context} # 3. Call the model response self.client.messages.create( modelclaude-opus-4-6, max_tokens1024, systemsystem, messages[{role: user, content: user_message}] ) answer response.content[0].text duration int((time.time() - start) * 1000) # 4. Save useful info to memory for next time self.memory.remember( contentfUser asked: {user_message[:200]}, memory_typeinteraction ) # 5. Log the episode self.episodes.log(Episode( taskuser_message[:200], approachsingle-turn with memory retrieval, outcomesuccess, duration_msduration, token_costresponse.usage.input_tokens response.usage.output_tokens, quality_score1.0, # would come from evaluation in prod )) return answer向量数据库向量检索是正经记忆系统的核心。它不靠精确匹配不像 SQL而是在高维空间里找最近邻。这才实现了语义搜索——即便没有共同词汇也能找到语义相关的记忆。5.1 相似性搜索原理每个记忆被转换成向量用 OpenAI 嵌入模型是 1,536 个浮点数数组。概念上相似的文本产生相似的向量。查询时嵌入查询语句用余弦相似度找最接近的向量。import numpy as npdef cosine_similarity(a: list, b: list) - float: 1.0 identical meaning 0.0 unrelated -1.0 opposite meaning a, b np.array(a), np.array(b) return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))# Example: these two sentences will have high similarityembedding_a embed(The user prefers dark mode)embedding_b embed(They like their interface theme to be dark)score cosine_similarity(embedding_a, embedding_b)# → ~0.91 (very similar)本地开发用 ChromaDB。准备部署时如果已经在用 Postgres评估 pgvector零额外基础设施。需要大规模时用 Pinecone 或 Qdrant。记忆管理记忆系统光积累不够还得会忘。一个没有焦点的存储会随时间退化——检索结果越来越嘈杂、延迟上升、相互矛盾的记忆让 Agent 晕头转向。遗忘策略主要有三种6.1 基于时间的衰减老记忆通常相关性更低。按近因和语义相关性组合给记忆打分。研究中使用的公式import mathfrom datetime import datetimedef memory_score( relevance: float, # cosine similarity 0–1 importance: float, # stored at write time 0–1 created_at: datetime, # when memory was formed recency_weight: float 0.3, decay_factor: float 0.995) - float: Inspired by the Generative Agents paper (Park et al., 2023). Balances: how relevant, how important, how recent. hours_old (datetime.utcnow() - created_at).total_seconds() / 3600 recency math.pow(decay_factor, hours_old) return ( relevance * 0.4 importance * 0.3 recency * recency_weight )6.2 写入时的重要性打分存储记忆时让模型给自己输出打分。只存储高分的项。这在源头过滤噪音。import reasync def score_importance(client, content: str) - float: Ask the LLM if information is worth saving (0.0 to 1.0). prompt fRate the importance of saving this for future interactions. 0.0 trivial (greeting) 0.5 moderately useful 1.0 critical (preferences, errors, decisions) Information: {content} Reply with ONLY the number. try: response await client.messages.create( modelclaude-3-haiku-20240307, max_tokens10, messages[{role: user, content: prompt}] ) text response.content[0].text.strip() match re.search(r[-]?\d*\.\d|\d, text) if match: score float(match.group()) return max(0.0, min(1.0, score)) except Exception: pass return 0.5 # Default fallback6.3 定期整合每晚运行任务把重复或高度相似的记忆合并成单一规范摘要。这类似于人类睡眠巩固记忆的方式。async def consolidate_memories(store: MemoryStore, similarity_threshold: float 0.92): Efficiently merge near-duplicate memories using vector search. all_mems store.collection.get(include[documents, embeddings, ids]) if not all_mems[ids]: return visited set() consolidated_docs [] for i, (mem_id, doc, emb) in enumerate(zip( all_mems[ids], all_mems[documents], all_mems[embeddings] )): if mem_id in visited: continue results store.collection.query( query_embeddings[emb], n_results10, include[documents, distances] ) group [doc] visited.add(mem_id) for res_id, res_doc, dist in zip(results[ids][0], results[documents][0], results[distances][0]): sim 1.0 - dist if res_id ! mem_id and res_id not in visited and sim similarity_threshold: group.append(res_doc) visited.add(res_id) if len(group) 1: summary await summarize_group(group) consolidated_docs.append(summary) else: consolidated_docs.append(doc) store.collection.delete(where{}) for doc in consolidated_docs: await store.remember(doc)总结没有记忆Agent 每次交互都从零开始。有了记忆Agent 才能理解你、适应你、跟你一起进化。记忆层的设计才是关键记住什么、遗忘什么、怎么用这些信息——这才是拉开差距的地方。最后唠两句为什么AI大模型成为越来越多程序员转行就业、升职加薪的首选很简单这些岗位缺人且高薪智联招聘的最新数据给出了最直观的印证2025年2月AI领域求职人数同比增幅突破200% 远超其他行业平均水平整个人工智能行业的求职增速达到33.4%位居各行业榜首其中人工智能工程师岗位的求职热度更是飙升69.6%。AI产业的快速扩张也让人才供需矛盾愈发突出。麦肯锡报告明确预测到2030年中国AI专业人才需求将达600万人人才缺口可能高达400万人这一缺口不仅存在于核心技术领域更蔓延至产业应用的各个环节。那0基础普通人如何学习大模型 深耕科技一线十二载亲历技术浪潮变迁。我见证那些率先拥抱AI的同行如何建立起效率与薪资的代际优势。如今我将积累的大模型面试真题、独家资料、技术报告与实战路线系统整理分享于此为你扫清学习困惑共赴AI时代新程。我整理出这套 AI 大模型突围资料包【允许白嫖】✅从入门到精通的全套视频教程✅AI大模型学习路线图0基础到项目实战仅需90天✅大模型书籍与技术文档PDF✅各大厂大模型面试题目详解✅640套AI大模型报告合集✅大模型入门实战训练这份完整版的大模型 AI 学习和面试资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】①从入门到精通的全套视频教程包含提示词工程、RAG、Agent等技术点② AI大模型学习路线图0基础到项目实战仅需90天全过程AI大模型学习路线③学习电子书籍和技术文档市面上的大模型书籍确实太多了这些是我精选出来的④各大厂大模型面试题目详解⑤640套AI大模型报告合集⑥大模型入门实战训练如果说你是以下人群中的其中一类都可以来智泊AI学习人工智能找到高薪工作一次小小的“投资”换来的是终身受益应届毕业生‌无工作经验但想要系统学习AI大模型技术期待通过实战项目掌握核心技术。零基础转型‌非技术背景但关注AI应用场景计划通过低代码工具实现“AI行业”跨界‌。业务赋能 ‌突破瓶颈传统开发者Java/前端等学习Transformer架构与LangChain框架向AI全栈工程师转型‌。获取方式有需要的小伙伴可以保存图片到wx扫描二v码免费领取【保证100%免费】