本文深入浅出地解析2026年Agent记忆系统的四层核心架构,涵盖上下文记忆、外部记忆、情景记忆和语义/参数记忆,搭配可直接运行的Python实例代码,详细讲解记忆层构建、向量数据库应用及记忆管理策略,兼顾小白入门理解与程序员实操需求,是2026年学习大模型Agent开发不可多得的收藏级指南。

花10分钟,彻底搞懂Agent记忆系统的四层架构,轻松入门大模型Agent开发,收藏起来避免后续找不到!

图像

想象一下:你开发了一个AI Agent,第一天它惊艳全场——精准定位代码bug、生成规范文档,甚至提出超出预期的优化方案。可到了第二天,你问它“还记得昨天我们讨论的那个优化思路吗?”,它却一脸“茫然”:“抱歉,我没印象了。”

这不是Agent不够智能,而是它缺少“记忆”。对于普通LLM聊天机器人,这种“失忆”或许无关紧要,但对于需要持续跑任务、做决策、迭代优化的Agent来说,没有记忆就等于每次交互都要“从零开始”,根本无法实现真正的智能演进。

2026年,大模型Agent的核心竞争力早已不是“响应速度”,而是“记忆能力”——记忆让无状态的LLM,变成能跨时间、懂上下文、会自主学习的智能体。今天,我们就从基础到实操,把Agent记忆系统讲透,小白能看懂,程序员能上手。

1. 什么是 Agentic Memory?

Agentic memory 不是单一组件,它更像一套幕后系统:不同类型的存储、不同检索方式、以及智能管理策略,让 Agent 能真正跨时间携带上下文。

说白了,记忆同时在做三件有区别的事。

连续性(Continuity)关乎身份。Agent 如何知道你是谁、你偏好什么、以及你们已经一起建立了什么。没有它,每次交互都像从零开始。

上下文(Context)关乎当前任务。刚发生了什么、用了什么工具、返回了什么结果、以及下一步需要做什么。它让多步骤工作流不至于崩掉。

学习(Learning)关乎变得更好。理解什么有效、什么无效、慢慢改进决策而不是重复同样的错误。

三者合一,让 Agent 每次交互都更一致、更可靠、甚至更智能一点。

Image 1: Agent Memory System

一个设计良好的 Agent 记忆系统同时处理这三件事,每种用不同的存储后端。

2. 四种记忆类型


业界已经沉淀出四种有区别的记忆类型。可以把它们理解成大脑的四个区域,每个各司其职。

Image 2: Four Memory Types

2.1 上下文记忆(In-context Memory)

上下文窗口是 Agent 的工作台。上面的东西都能即时访问,模型可以在单次前向传播中推理。不需要检索步骤。

但工作台有尺寸限制。每个 token 都要花钱和时间。会话结束时,工作台会被清空。

上下文窗口里有什么?

  • System prompt:Agent 人设、规则、能力、当前日期/用户信息
  • 对话历史:本次会话的来回交互
  • 工具调用结果:Agent 刚调用的工具输出
  • 检索到的记忆:从外部存储拉取的片段
  • Scratchpad:中间推理(逐步思考输出)

滑动窗口问题

长对话中,历史积累最终会溢出上下文限制。简单截断最旧消息会丢失重要的早期上下文。更好的策略:

  • 摘要(Summarization):定期把旧的对话压缩成简短摘要,用摘要替换
  • 选择性保留(Selective retention):保留包含关键事实、决策或工具结果的对话;丢弃闲聊
  • 卸载到外部记忆:把重要事实提取到向量存储,需要时检索

2.2 外部记忆(External Memory)

外部记忆是任何持久化在模型之外的东西——数据库、向量存储、键值存储、文件。它跨越会话边界存活。如果存储得当,Agent 能记住六个月前的事。

外部存储有两种形式:

结构化存储(精确查询):PostgreSQL、Redis、SQLite。按 key、ID 或 SQL 查询。快速、可预测,适合用户画像、偏好和结构化数据。

向量存储(语义搜索):Pinecone、Chroma、pgvector。按语义查询,“找与这个概念相似的记忆”。对非结构化笔记和情景回忆至关重要。

Image 3: External Memory Types

检索步骤是瓶颈。 检索不到正确的记忆,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,而不是从手工数据集聚类。

反思循环:

Image 4: Reflection Loop

2.4 语义/参数记忆(Semantic/Parametric Memory)

这是模型与生俱来的记忆。所有东西在训练时编码进权重:世界事实、语言模式、推理策略、编码规范、文化知识。

它一直都在。Agent 从不需要检索它。但它有硬限制:

  • 训练时冻结:模型不知道 cutoff 日期之后发生了什么
  • 运行时无法更新:不重训练或微调就无法注入新的永久事实
  • 不透明:无法检查模型到底"知道"什么或不知道什么
  • 容易幻觉:模型用似是而非的错误内容填补空白

对于时间敏感、领域特定或私有的内容,不要依赖参数记忆。用外部检索。参数记忆是你在没有更好来源时的通用世界知识 fallback。

正确的思维模型:参数记忆是 Agent 的通识教育。外部、情景和上下文记忆是 Agent 的在职经验。最好的 Agent 两者结合。

3. 记忆如何在 Agent Loop 中流动?


让我们整合起来。以下是 Agent 每次处理请求时发生的事——展示每个记忆系统都在运作。

Image 5: Full Memory System Flow

注意记忆操作是包裹 LLM 调用的:先检索、再写入。模型本身无状态,是记忆系统给了无状态的 Agent 有状态的假象。

4. 构建记忆层


上代码。用 Python + OpenAI 做嵌入、ChromaDB 做本地向量存储。概念是通用的,换库就行。

pip install chromadb openai anthropic python-dotenv

4.1 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])

4.2 EpisodicLogger 类

在 MemoryStore 之上叠加坡情日志层。

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
        )

4.3 整合:记忆增强 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

5. 向量数据库


向量检索是正经记忆系统的核心。它不靠精确匹配(不像 SQL),而是在高维空间里找最近邻。这才实现了语义搜索——即便没有共同词汇,也能找到语义相关的记忆。

5.1 相似性搜索原理

每个记忆被转换成向量(用 OpenAI 嵌入模型是 1,536 个浮点数数组)。概念上相似的文本产生相似的向量。查询时,嵌入查询语句,用余弦相似度找最接近的向量。

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)

本地开发用 ChromaDB。准备部署时,如果已经在用 Postgres,评估 pgvector(零额外基础设施)。需要大规模时用 Pinecone 或 Qdrant。

6. 记忆管理


记忆系统光积累不够,还得会忘。一个没有焦点的存储会随时间退化——检索结果越来越嘈杂、延迟上升、相互矛盾的记忆让 Agent 晕头转向。

遗忘策略主要有三种:

6.1 基于时间的衰减

老记忆通常相关性更低。按近因和语义相关性组合给记忆打分。研究中使用的公式:

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
    )

6.2 写入时的重要性打分

存储记忆时,让模型给自己输出打分。只存储高分的项。这在源头过滤噪音。

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",
            max_tokens=10,
            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 fallback

6.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_results=10,
            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)

7. 总结


没有记忆,Agent 每次交互都从零开始。有了记忆,Agent 才能理解你、适应你、跟你一起进化。

记忆层的设计才是关键:记住什么、遗忘什么、怎么用这些信息——这才是拉开差距的地方。

如何学习大模型 AI ?

由于新岗位的生产效率,要优于被取代岗位的生产效率,所以实际上整个社会的生产效率是提升的。

但是具体到个人,只能说是:

“最先掌握AI的人,将会比较晚掌握AI的人有竞争优势”。

这句话,放在计算机、互联网、移动互联网的开局时期,都是一样的道理。

我在一线科技企业深耕十二载,见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事,早已在效率与薪资上形成代际优势,我意识到有很多经验和知识值得分享给大家,也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套 AI 大模型突围资料包

  • ✅ 从零到一的 AI 学习路径图
  • ✅ 大模型调优实战手册(附医疗/金融等大厂真实案例)
  • ✅ 百度/阿里专家闭门录播课
  • ✅ 大模型当下最新行业报告
  • ✅ 真实大厂面试真题
  • ✅ 2026 最新岗位需求图谱

所有资料 ⚡️ ,朋友们如果有需要 《AI大模型入门+进阶学习资源包》下方扫码获取~
在这里插入图片描述

① 全套AI大模型应用开发视频教程

(包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点)
在这里插入图片描述

② 大模型系统化学习路线

作为学习AI大模型技术的新手,方向至关重要。 正确的学习路线可以为你节省时间,少走弯路;方向不对,努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划,带你从零基础入门到精通!
在这里插入图片描述

③ 大模型学习书籍&文档

学习AI大模型离不开书籍文档,我精选了一系列大模型技术的书籍和学习文档(电子版),它们由领域内的顶尖专家撰写,内容全面、深入、详尽,为你学习大模型提供坚实的理论基础。
在这里插入图片描述

④ AI大模型最新行业报告

2025最新行业报告,针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估,以了解哪些行业更适合引入大模型的技术和应用,以及在哪些方面可以发挥大模型的优势。
在这里插入图片描述

⑤ 大模型项目实战&配套源码

学以致用,在项目实战中检验和巩固你所学到的知识,同时为你找工作就业和职业发展打下坚实的基础。
在这里插入图片描述

⑥ 大模型大厂面试真题

面试不仅是技术的较量,更需要充分的准备。在你已经掌握了大模型技术之后,就需要开始准备面试,我精心整理了一份大模型面试题库,涵盖当前面试中可能遇到的各种技术问题,让你在面试中游刃有余

图片

以上资料如何领取?

在这里插入图片描述

为什么大家都在学大模型?

最近科技巨头英特尔宣布裁员2万人,传统岗位不断缩减,但AI相关技术岗疯狂扩招,有3-5年经验,大厂薪资就能给到50K*20薪!

图片

不出1年,“有AI项目经验”将成为投递简历的门槛。

风口之下,与其像“温水煮青蛙”一样坐等被行业淘汰,不如先人一步,掌握AI大模型原理+应用技术+项目实操经验,“顺风”翻盘!
在这里插入图片描述
在这里插入图片描述

这些资料真的有用吗?

这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理,现任上海殷泊信息科技CEO,其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证,服务航天科工、国家电网等1000+企业,以第一作者在IEEE Transactions发表论文50+篇,获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目,无论你是小白还是有些技术基础的技术人员,这份资料都绝对能帮助你提升薪资待遇,转行大模型岗位。
在这里插入图片描述
在这里插入图片描述

以上全套大模型资料如何领取?

在这里插入图片描述

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐