Agentic Memory: 详解 Agent 记忆系统的架构与实现

✍ Ramakrishna (techwith_ram)🕐 2026-05-17📦 21.9 KB 🟢 已读 𝕏 文章列表

文章通过生动的类比，指出缺乏记忆是当前 LLM Agent 的主要短板。作者详细拆解了 Agent 记忆系统的三个核心功能：连续性、上下文和学习。文章重点介绍了四种记忆类型：上下文记忆、外部记忆、情景记忆和参数记忆，并深入探讨了如何通过检索增强（RAG）和反思循环来构建持久的智能。

# Agentic Memory: A Detailed Breakdown **作者**: 𝗿𝗮𝗺𝗮𝗸𝗿𝘂𝘀𝗵𝗻𝗮— 𝗲/𝗮𝗰𝗰 **日期**: 2026-03-27T12:00:14.000Z **来源**: [https://x.com/techwith_ram/status/2037499938574110770](https://x.com/techwith_ram/status/2037499938574110770) --- ![Image](/doc/1992/image/img_bb26bab7a3a3.jpg) Imagine one day you hire a brilliant freelancer. On her first day, she’s incredible, catches every bug, writes clean docs, & even suggests improvements you hadn’t thought of. You’re impressed. On the second day, you walk in & say, “Hey, remember that issue we discussed yesterday?” She pauses. Looks at you. Slight smile. “Sorry… what issue?” No memory. No context. Completely gone. You will be shocked as I am while writing this here. That’s exactly how most LLMs behave. Every new conversation is a fresh start. The model doesn’t know who you are, what you’ve built together, or what you discussed even a few minutes ago in another chat window. For a simple chatbot, that’s fine. But for an agent, something that runs tasks, makes decisions, and improves over time, this kind of amnesia is a dealbreaker. Because real intelligence isn’t just about responding well. It’s about remembering, learning, and building on what came before. Memory is what turns a stateless system into something that can actually evolve. # What is agentic memory, really? Agentic memory isn’t just one thing. It’s more like a system working behind the scenes, different types of storage, ways to retrieve information, and smart strategies to manage it all so an agent can actually carry context over time. The key idea is simple: memory isn’t doing one job; it’s doing three very different ones at the same time. ➜ Continuity is about identity. It’s how the agent knows who you are, what you prefer, and what you’ve already built together. Without it, every interaction feels like starting from scratch. ➜ Context is about the task at hand. What just happened, which tool was used, what came back as a result, and what needs to happen next. It’s what keeps multi-step workflows from falling apart. ➜ Learning is about getting better. Understanding what worked, what didn’t, and slowly improving decisions over time instead of repeating the same mistakes. Put together, it makes agents feel consistent, reliable, and a little more intelligent with every interaction. ![Image](/doc/1992/image/img_777900817bc0.jpg) Designed by author @techwith_ram A well-designed agent memory system handles all three, using different storage backends for each. # The 4-types of memory The field has converged on four distinct memory types. Think of them as four different parts of the brain, each evolved for a specific job. ![Image](/doc/1992/image/img_816bc696d924.jpg) Designed by author @techwith_ram ## 1. In-context memory The context window is your agent's working desk. Everything on it is instantly accessible. The model can reason over it in a single forward pass. No retrieval step required. But the desk has a size limit. Every token costs money and time. And when the session ends, the desk gets wiped clean. What lives in context? - System prompt: agent persona, rules, capabilities, current date/user info - Conversation history: the back-and-forth so far this session - Tool call results: outputs from tools the agent just invoked - Retrieved memories: chunks pulled in from external storage - Scratchpad: intermediate reasoning (think-step-by-step outputs) ![Image](/doc/1992/image/img_9671781aaba9.jpg) Designed by author @techwith_ram The sliding window problem In long conversations, history accumulates and eventually overflows the context limit. The naive solution of truncating the oldest messages loses important early context. Better strategies: - Summarization: periodically compress old turns into a brief summary and replace them with the summary - Selective retention: keep turns that contain key facts, decisions, or tool results; discard small talk. - Offload to external memory: extract important facts into a vector store, then retrieve them as needed ## 2. External memory External memory is anything that persists outside the model—databases, vector stores, key-value stores, and files. It survives session boundaries. Your agent can remember something from six months ago if you store it right. There are two flavors of external storage: Structured Store (Exact Lookup): PostgreSQL, Redis, SQLite. You query by key, ID, or SQL. Fast, predictable, great for user profiles, preferences, and structured data. Vector Stores (Semantic Search): Pinecone, Chroma, pgvector. You query by meaning, "find memories similar to this concept." "Essential for unstructured notes and episodic recall. ![Image](/doc/1992/image/img_786f14a29d98.jpg) Designed by author @techwith_ram The retrieval step is a bottleneck. If you don't retrieve the right memories, the agent behaves as if they don't exist. Good memory architecture is 20% storage and 80% retrieval design. ## 3. Episodic memory Episodic memory is the most underappreciated type. While external memory stores facts, episodic memory stores events, specifically the outcomes of past actions. The simplest form is a structured log: every time the agent completes a task, it records what happened. Over time, this log becomes a rich source of self-knowledge the agent can consult before making decisions. What an episode looks like: ``` { "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 */] } ``` When a new task comes in, the agent retrieves the most semantically similar past episodes and uses them to pick a strategy. This is essentially few-shot learning from personal history rather than from a handcrafted dataset. The reflection loop👇 ![Image](/doc/1992/image/img_e0d49d11e25f.jpg) Designed by author @techwith_ram ## Semantic/parametric memory This is the memory the model was born with. Everything is encoded in weights during training, facts about the world, language patterns, reasoning strategies, coding conventions, and cultural knowledge. It's always there. The agent never has to retrieve it. But it comes with hard limitations: - Frozen at training time: the model doesn't know what happened after its cutoff date - Can't be updated at runtime: you can't inject new permanent facts without retraining or fine-tuning - Opaque: you can't inspect exactly what the model "knows" or doesn't - Hallucination-prone: the model fills gaps with plausible-but-wrong completions. For anything time-sensitive, domain-specific, or private, don't rely on parametric memory. Use external retrieval. Parametric memory is your fallback for general world knowledge when no better source exists. The right mental model: parametric memory is the agent's general education. External, episodic, and in-context memory are the agent's on-the-job experience. The best agents combine both. # How does memory flow through an agent loop? Let's put it all together. Here's what happens every single time an agent processes a request—showing every memory system in action. ![Image](/doc/1992/image/img_84b3e337190f.jpg) Designed by author @techwith_ram Notice that memory operations bookend the LLM call: retrieval before, writing after. The model itself is stateless; the memory system is what gives the illusion of a stateful, aware agent. # Building a memory layer Let's build this. We'll use Python with OpenAI for embeddings and ChromaDB as our local vector store. The same concepts apply to any other stack—swap the libraries. ``` pip install chromadb openai anthropic python-dotenv ``` ## The MemoryStore class This handles writing memories (with embeddings) and semantic retrieval. It's the foundation everything else sits on. ``` import chromadb from openai import OpenAI from datetime import datetime import json, uuid class 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(path=persist_dir) self.collection = self.client.get_or_create_collection( name=f"agent_{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( model="text-embedding-3-small", input=text ) 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_results=k, where=where, 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, key=lambda x: x["relevance"], reverse=True) def forget(self, memory_id: str): """Delete a specific memory (GDPR compliance, stale data, etc.)""" self.collection.delete(ids=[memory_id]) ``` ## The EpisodicLogger class Now let's add the episode logging layer on top. ``` from .store import MemoryStore from dataclasses import dataclass, asdict from typing import Optional import time @dataclass class 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] = None class 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 = ( f"Task: {episode.task}\n" f"Approach: {episode.approach}\n" f"Outcome: {episode.outcome}\n" f"Notes: {episode.notes}" ) self.store.remember( content=doc, memory_type="episode", 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( query=task, k=k, memory_type="episode", min_relevance=0.65 ) ``` ## Putting it together: a memory-augmented agent ``` import anthropic from memory.store import MemoryStore from memory.episodic import EpisodicLogger, Episode import time class 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, k=4) # Similar past task approaches episodes = self.episodes.recall_similar(user_message, k=2) 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( model="claude-opus-4-6", max_tokens=1024, system=system, 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( content=f"User asked: {user_message[:200]}", memory_type="interaction" ) # 5. Log the episode self.episodes.log(Episode( task=user_message[:200], approach="single-turn with memory retrieval", outcome="success", duration_ms=duration, token_cost=response.usage.input_tokens + response.usage.output_tokens, quality_score=1.0, # would come from evaluation in prod )) return answer ``` # Vector databases It's the heart of any serious memory system. Instead of querying by exact match (like SQL), it finds the nearest neighbors of a vector in high-dimensional space. This is what enables semantic search — finding memories that are conceptually related even if they share no words. ## How similarity search works Every memory is converted to a vector (an array of 1,536 floats with OpenAI's embedding model). Conceptually similar texts produce similar vectors. When you query, you embed the query and find the closest vectors using cosine similarity. ``` import numpy as np def 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 similarity embedding_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) ``` Start with ChromaDB for local development. When you're ready to deploy, evaluate pgvector if you're already on Postgres with zero extra infrastructure. Use Pinecone or Qdrant when you need serious scale. # Memory management Real memory systems don't just accumulate. They curate. An ever-growing, unfocused store degrades over time—retrieval gets noisier, latency climbs, and contradictory memories confuse the agent. You need a forgetting strategy. Here are the three main approaches: ## 1. Time-based decay Older memories are less relevant. Score memories by a combination of recency and semantic relevance. The formula used in research: ``` import math from datetime import datetime def 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 ) ``` ## 2. Importance scoring at write time When storing a memory, ask the model to score its own output for importance. Only store high-scoring items. This filters noise at the source. ``` import re async def score_importance(client, content: str) -> float: """Ask the LLM if information is worth saving (0.0 to 1.0).""" prompt = f"""Rate 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( model="claude-3-haiku-20240307", # Use the current available Haiku model max_tokens=10, messages=[{"role": "user", "content": prompt}] ) # Extract the first string that looks like a float/int 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 fallback ``` ## 3. Periodic consolidation Run a nightly job that merges duplicate or highly similar memories into a single canonical summary. This is analogous to how human sleep consolidates memories. ``` 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 # Use the vector store's built-in search to find neighbors # this is much faster than a manual nested loop results = store.collection.query( query_embeddings=[emb], n_results=10, # Adjust based on expected density include=["documents", "distances"] ) # Identify group members (1.0 - distance = cosine similarity) 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) # Process the group if len(group) > 1: summary = await summarize_group(group) # Likely needs to be async consolidated_docs.append(summary) else: consolidated_docs.append(doc) # Atomic-ish replacement: Clear and Re-populate store.collection.delete(where={}) for doc in consolidated_docs: await store.remember(doc) ``` # FInal Thoughts At the end of the day, memory is what makes an AI feel less like a tool & more like a partner. Without it, every interaction starts from zero. With it, an agent can understand, adapt, and improve over time. The real power isn’t just in the model; it’s in how you design what the model remembers, what it forgets, and how it uses that information. Build the memory layer right, and everything else becomes smarter. Codes are AI-generated. Follow @techwith_ram for more such posts. ## 相关链接 - [𝗿𝗮𝗺𝗮𝗸𝗿𝘂𝘀𝗵𝗻𝗮— 𝗲/𝗮𝗰𝗰](https://x.com/techwith_ram) - [@techwith_ram](https://x.com/techwith_ram) - [141K](https://x.com/techwith_ram/status/2037499938574110770/analytics) - [@techwith_ram](https://x.com/@techwith_ram) - [@techwith_ram](https://x.com/@techwith_ram) - [@techwith_ram](https://x.com/@techwith_ram) - [@techwith_ram](https://x.com/@techwith_ram) - [@techwith_ram](https://x.com/@techwith_ram) - [@techwith_ram](https://x.com/@techwith_ram) - [@techwith_ram](https://x.com/@techwith_ram) - [Upgrade to Premium](https://x.com/i/premium_sign_up) - [8:00 PM · Mar 27, 2026](https://x.com/techwith_ram/status/2037499938574110770) - [141.5K Views](https://x.com/techwith_ram/status/2037499938574110770/analytics) - [View quotes](https://x.com/techwith_ram/status/2037499938574110770/quotes) --- *导出时间: 2026/5/17 16:13:15* --- ## 中文翻译 # Agentic Memory: A Detailed Breakdown (代理记忆：详细解析) **作者**: 𝗿𝗮𝗺𝗮𝗸𝗿𝘂𝘀𝗵𝗻𝗮— 𝗲/𝗮𝗰𝗰 **日期**: 2026-03-27T12:00:14.000Z **来源**: [https://x.com/techwith_ram/status/2037499938574110770](https://x.com/techwith_ram/status/2037499938574110770) --- ![Image](/doc/1992/image/img_bb26bab7a3a3.jpg) 试想有一天，你聘请了一位才华横溢的自由职业者。入职第一天，她表现惊人，捕捉到了每一个 Bug，编写了清晰的文档，甚至提出了你未曾想到的改进建议。你印象深刻。到了第二天，你走进来说：“嘿，记得我们昨天讨论的那个问题吗？” 她停顿了一下。看着你。微微一笑。 “抱歉……什么问题？” 没有记忆。没有上下文。完全消失了。你会像我此刻写下这段文字时一样感到震惊。这正是大多数大语言模型（LLM）的表现方式。每一次新的对话都是全新的开始。模型不知道你是谁，你们共同构建了什么，或者甚至几分钟前在另一个聊天窗口里讨论了什么。对于一个简单的聊天机器人来说，这没问题。但对于一个代理——那种运行任务、做出决策并随时间不断改进的东西——这种健忘症是致命的。因为真正的智能不仅仅是反应良好。它是关于记忆、学习以及建立在过去基础之上的能力。记忆正是将无状态系统转变为能够实际演化的东西的关键。 # 什么是代理记忆，本质上？代理记忆不仅仅是一样东西。它更像是一个在后台运行的系统，包含不同类型的存储、检索信息的方式，以及管理这一切的智能策略，以便代理能够真正地随时间保持上下文。核心思想很简单：记忆不是在做一份工作；它同时在做三份截然不同的工作。 ➜ **连续性** 关乎身份。它是代理知道你是谁、你偏好什么以及你们已经共同构建了什么的方式。没有它，每一次互动都感觉像是从零开始。 ➜ **上下文** 关乎手头的任务。刚刚发生了什么，使用了哪个工具，结果返回了什么，以及接下来需要发生什么。这是防止多步骤工作流程崩溃的关键。 ➜ **学习** 关乎变得更好。了解什么有效，什么无效，并随着时间的推移慢慢改进决策，而不是重复同样的错误。综合起来，它使代理在每一次互动中感觉更加一致、可靠，并且更聪明一点。 ![Image](/doc/1992/image/img_777900817bc0.jpg) 由作者 @techwith_ram 设计一个设计良好的代理记忆系统利用不同的存储后端来处理所有这三个方面。 # 4种类型的记忆该领域已经趋向于四种独特的记忆类型。可以把它们想象成大脑的四个不同部分，每一部分都为了特定的工作而进化。 ![Image](/doc/1992/image/img_816bc696d924.jpg) 由作者 @techwith_ram 设计 ## 1. 上下文内记忆上下文窗口是你代理的工作台。上面的所有东西都可以即时访问。模型可以在单次前向传递中对其进行推理。不需要检索步骤。但是工作台有大小限制。每一个 Token 都要花钱和时间。当会话结束时，工作台会被擦拭干净。上下文中存储着什么？ - **系统提示词**：代理人格、规则、能力、当前日期/用户信息 - **对话历史**：本次会话目前的来回对话 - **工具调用结果**：代理刚刚调用的工具的输出 - **检索到的记忆**：从外部存储中提取的块 - **草稿本**：中间推理（思维链输出） ![Image](/doc/1992/image/img_9671781aaba9.jpg) 由作者 @techwith_ram 设计 **滑动窗口问题** 在长对话中，历史记录会累积并最终溢出上下文限制。截断最旧消息的幼稚方案会丢失重要的早期上下文。更好的策略： - **摘要**：定期将旧的轮次压缩为简短摘要，并用摘要替换它们 - **选择性保留**：保留包含关键事实、决策或工具结果的轮次；丢弃闲聊。 - **卸载到外部记忆**：将重要事实提取到向量存储中，然后按需检索 ## 2. 外部记忆外部记忆是模型外部持久存在的任何东西——数据库、向量存储、键值存储和文件。它跨越会话边界而存在。如果你存储得当，你的代理可以记住六个月前的事情。有两种外部存储的形式： **结构化存储（精确查找）**：PostgreSQL, Redis, SQLite。你通过键、ID 或 SQL 查询。快速、可预测，非常适合用户配置文件、偏好和结构化数据。 **向量存储（语义搜索）**：Pinecone, Chroma, pgvector。你通过意义查询，“找到与这个概念相似的记忆。”对于非结构化笔记和情节回忆至关重要。 ![Image](/doc/1992/image/img_786f14a29d98.jpg) 由作者 @techwith_ram 设计检索步骤是一个瓶颈。如果你没有检索到正确的记忆，代理的行为就好像它们不存在一样。良好的记忆架构是 20% 的存储和 80% 的检索设计。 ## 3. 情节记忆情节记忆是最被低估的类型。虽然外部记忆存储事实，但情节记忆存储事件，特别是过去行为的结果。最简单的形式是结构化日志：每当代理完成一项任务时，它都会记录发生了什么。随着时间的推移，这个日志变成了丰富的自我知识来源，代理可以在做出决定之前查阅。一个情节看起来像这样： ``` { "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 */] } ``` 当一个新任务进来时，代理检索语义上最相似的过去情节，并利用它们来选择策略。这本质上是从个人历史而不是从手工制作的数据集中进行少样本学习。 **反思循环👇** ![Image](/doc/1992/image/img_e0d49d11e25f.jpg) 由作者 @techwith_ram 设计 ## 语义/参数记忆这是模型与生俱来的记忆。所有内容都在训练期间被编码到权重中，关于世界的事实、语言模式、推理策略、编码惯例和文化知识。它总是存在。代理永远不需要检索它。但它伴随着硬性限制： - **在训练时冻结**：模型不知道其截止日期之后发生的事情 - **无法在运行时更新**：如果不进行重新训练或微调，就无法注入新的永久性事实 - **不透明**：你无法准确检查模型“知道”或不知道什么 - **容易产生幻觉**：模型用看似合理但错误的补全来填补空白。对于任何时效性强、特定领域或私有的内容，不要依赖参数记忆。使用外部检索。当不存在更好的来源时，参数记忆是你的一般世界知识的后备。正确的心智模型：参数记忆是代理的通识教育。外部、情节和上下文内记忆是代理的在职经验。最好的代理将两者结合使用。 # 记忆如何在代理循环中流动？让我们把所有这些整合起来。以下是代理每次处理请求时发生的事情——展示了每个记忆系统都在运行。 ![Image](/doc/1992/image/img_84b3e337190f.jpg) 由作者 @techwith_ram 设计注意，记忆操作围绕着 LLM 调用：之前是检索，之后是写入。模型本身是无状态的；记忆系统正是创造了有状态的、有感知的代理的错觉的东西。 # 构建记忆层让我们构建这个。我们将使用 Python 结合 OpenAI 进行嵌入，并使用 ChromaDB 作为我们的本地向量存储。相同的概念适用于任何其他技术栈——只需交换库。 ``` pip install chromadb openai anthropic python-dotenv ``` ## MemoryStore 类这处理写入记忆（带嵌入）和语义检索。它是其他所有事物的基础。 ``` import chromadb from openai import OpenAI from datetime import datetime import json, uuid class 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(path=persist_dir) self.collection = self.client.get_or_create_collection( name=f"agent_{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( model="text-embedding-3-small", input=text ) 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_results=k, where=where, 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, key=lambda x: x["relevance"], reverse=True) def forget(self, memory_id: str): """Delete a specific memory (GDPR compliance, stale data, etc.)""" self.collection.delete(ids=[memory_id]) ``` ## EpisodicLogger 类现在让我们在顶部添加情节日志记录层。 ``` from .store import MemoryStore from dataclasses import dataclass, asdict from typing import Optional import time @dataclass class 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] = None class 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 = ( f"Task: {episode.task}\n" f"Approach: {episode.approach}\n" f"Outcome: {episode.outcome}\n" f"Notes: {episode.notes}" ) self.store.remember( content=doc, memory_type="episode", 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( query=task, k=k, memory_type="episode", min_relevance=0.65 ) ``` ## 组合在一起：一个记忆增强的代理 ``` import anthropic from memory.store import MemoryStore from memory.episodic import EpisodicLogger, Episode import time class 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, k=4) # Similar past task approaches episodes = self.episodes.recall_similar(user_message, k=2) 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( model="claude-opus-4-6", max_tokens=1024, system=system, 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( content=f"User asked: {user_message[:200]}", memory_type="interaction" ) # 5. Log the episode self.episodes.log(Episode( task=user_message[:200], approach="single-turn with memory retrieval", outcome="success", duration_ms=duration, token_cost=response.usage.input_tokens + response.usage.output_tokens, quality_score=1.0, # would come from evaluation in prod )) return answer ``` # 向量数据库它是任何严肃记忆系统的核心。它不是通过精确匹配（像 SQL）来查询，而是在高维空间中找到向量的最近邻。这正是实现语义搜索的原因——即使在记忆中没有共用词汇，也能找到概念上相关的记忆。 ## 相似性搜索是如何工作的每一条记忆都被转换为一个向量（使用 OpenAI 的嵌入模型是一个包含 1,536 个浮点数的数组）。概念上相似的文本会产生相似的向量。当你查询时，你嵌入查询并使用余弦相似度找到最接近的向量。

🔗 相关文档