核心思想

记忆是把无状态的 LLM 变成能演化的智能体的关键——它同时承担连续性(你是谁)、上下文(手头任务)、学习(越做越好)三种职能。业界已收敛出四种记忆类型:上下文内记忆(工作台,即时但会话结束即清空)、外部记忆(数据库 / 向量库,跨会话持久)、情景记忆(记录过往行动的结果,支撑「从个人历史中少样本学习」)、语义 / 参数记忆(训练时固化进权重)。文章给出基于 Python + ChromaDB 的可落地实现,并强调好的记忆架构「两成在存储、八成在检索设计」,还需配上时间衰减、重要性评分、周期性整合三种遗忘策略。

设想有一天,你雇了一位很厉害的自由职业者。第一天她表现惊艳:抓出每一个 bug,文档写得干净利落,甚至还提出了一些你压根没想到的改进点。你印象深刻。

第二天,你走进办公室,跟她说:“嘿,还记得我们昨天讨论的那个问题吗?”

她顿了一下。看着你。微微一笑。

“抱歉……什么问题?”

没有记忆。没有上下文。一干二净。我写到这儿,大概和你一样觉得离谱。

大多数 LLM 就是这副样子。每一次新对话都是一个全新的开始。模型不知道你是谁、不知道你们一起做过什么,甚至不记得你几分钟前在另一个聊天窗口里说过什么。

对一个简单的聊天机器人来说,这没什么问题。但对一个智能体——一个要执行任务、做出决策、并随时间不断进步的系统——来说,这种健忘就是致命缺陷。

因为真正的智能,不只是把话回答好。它在于记得住、学得会,并能在过往的基础上继续生长。

正是记忆,把一个无状态的系统变成了某种真正会演化、会成长的东西

智能体记忆到底是什么?

智能体记忆并不是单一的某样东西。它更像是一套在幕后运转的系统——不同类型的存储、各式各样的信息检索方式,以及把这一切管理起来的聪明策略——好让智能体真正能把上下文延续下去,跨越一次又一次会话。

核心思想很简单:记忆不是在做一件事;它同时在做三件非常不同的事。

连续性(Continuity) 关乎身份。它是智能体得以知道你是谁、你偏好什么、你们已经一起做出过什么的依据。没有它,每一次互动都像是从零开始。

上下文(Context) 关乎手头的任务。刚刚发生了什么、用了哪个工具、返回了什么结果、接下来该做什么。正是它让多步骤的工作流不至于分崩离析。

学习(Learning) 关乎变得更好。搞清楚什么管用、什么不管用,并随时间慢慢改进决策,而不是把同样的错误一犯再犯。

把这三者合到一起,智能体就会显得稳定、可靠,并且每一次互动都更聪明一点。

图片由作者 @techwith_ram 设计

一个设计良好的智能体记忆系统会把这三件事全部照顾到,并为每一件事使用不同的存储后端。

四种记忆类型

这个领域已经收敛到了四种各不相同的记忆类型。可以把它们想象成大脑的四个不同部分,每一个都是为了某项特定工作而演化出来的。

图片由作者 @techwith_ram 设计

1. 上下文内记忆

上下文窗口就是你的智能体的工作台。摆在台面上的一切都可以即时取用。模型在一次前向传播里就能对它做推理,不需要任何检索步骤。

但这张工作台有尺寸上限。每一个 token 都要花钱、花时间。而且一旦会话结束,工作台就会被擦得干干净净。

上下文里都装着什么?

  • 系统提示词: 智能体的人设、规则、能力,以及当前日期 / 用户信息
  • 对话历史: 本次会话到目前为止的一来一回
  • 工具调用结果: 智能体刚刚调用的工具所返回的输出
  • 检索到的记忆: 从外部存储里拉进来的内容片段
  • 草稿区: 中间推理过程(也就是“一步步思考”产生的输出)

图片由作者 @techwith_ram 设计

滑动窗口问题

在长对话里,历史会不断累积,最终溢出上下文上限。“截掉最早的消息”这种粗暴做法会丢掉重要的早期上下文。更好的策略有:

  • 摘要:周期性地把陈旧的对话轮次压缩成一段简短摘要,并用这段摘要替换掉它们
  • 选择性保留:保留那些包含关键事实、决策或工具结果的对话轮次,丢弃闲聊。
  • 卸载到外部记忆:把重要事实抽取出来存进向量数据库,然后按需检索回来

2. 外部记忆

外部记忆指的是一切持久存在于模型之外的东西——数据库、向量数据库、键值存储和文件。它能跨越一次次会话留存下来。只要存得对,你的智能体可以记得六个月前的某件事。

外部存储有两种风味:

结构化存储(精确查找): PostgreSQL、Redis、SQLite。你按键、ID 或 SQL 来查询。速度快、可预测,非常适合用户档案、偏好设置和结构化数据。

向量数据库(语义搜索): Pinecone、Chroma、pgvector。你按“含义”来查询——“找出与这个概念相似的记忆”。对于非结构化的笔记和情景式回忆来说不可或缺。

图片由作者 @techwith_ram 设计

检索这一步是一个瓶颈。如果你没有检索到正确的记忆,智能体就会表现得好像这些记忆根本不存在。好的记忆架构,两成在存储,八成在检索设计

3. 情景记忆

情景记忆是最被低估的一种类型。外部记忆存的是事实,而情景记忆存的是事件——具体来说,是过往行动的结果。

它最简单的形式是一份结构化日志:智能体每完成一次任务,就记录下当时发生了什么。随着时间推移,这份日志会变成一座内容丰富的自我认知宝库,智能体在做决策之前可以随时翻阅它。

一条情景(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 */]
}

当一个新任务到来时,智能体会检索出语义上最相似的若干条过往情景,并据此挑选一种策略。这本质上就是从个人历史中做少样本学习,而不是从一份手工打造的数据集里学。

反思循环 👇

图片由作者 @techwith_ram 设计

语义记忆 / 参数记忆

这是模型与生俱来的记忆。一切都在训练期间被编码进了权重——关于世界的事实、语言模式、推理策略、编码规范,以及文化常识。

它永远都在。智能体根本不必去检索它。但它带着一些硬性局限:

  • 冻结在训练那一刻: 模型不知道它的知识截止日期之后发生了什么
  • 运行时无法更新: 不经过重新训练或微调,你没法注入新的永久性事实
  • 不透明: 你没法精确查看模型到底“知道”什么、不知道什么
  • 容易产生幻觉: 模型会用一些貌似合理、其实错误的内容来填补空白。

对于任何时效性强、领域特定或涉及隐私的信息,都不要依赖参数记忆。要用外部检索。参数记忆是你在没有更好来源时,用来兜底通用世界知识的备选项。

正确的心智模型是这样的:参数记忆是智能体接受的通识教育。而外部记忆、情景记忆和上下文内记忆,是智能体的在岗实战经验。最好的智能体会把两者结合起来。

记忆是如何在智能体循环里流转的?

我们把所有部件拼到一起。下面就是智能体每一次处理请求时所发生的事——把每一个记忆系统的运转都展示出来。

图片由作者 @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:
    """AI 智能体的持久化向量记忆。"""
 
    def __init__(self, agent_id: str, persist_dir: str = "./memory_db"):
        self.agent_id = agent_id
        self.openai = OpenAI()
 
        # ChromaDB 把向量存到磁盘上,重启后依然保留
        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"}  # 余弦相似度
        )
 
    def _embed(self, text: str) -> list[float]:
        """用 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:
        """存储一条记忆。返回该记忆的 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]:
        """检索与某个查询最相关的 k 条记忆。"""
        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  # 余弦距离 → 相似度
            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):
        """删除某一条特定的记忆(用于 GDPR 合规、清理过期数据等)。"""
        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,由评估器或用户给出
    notes: str = ""
    error: Optional[str] = None
 
class EpisodicLogger:
    def __init__(self, memory_store: MemoryStore):
        self.store = memory_store
 
    def log(self, episode: Episode):
        """把一条情景作为可搜索的文档保存进记忆。"""
        # 构造一段信息丰富的文本表示,供语义搜索使用
        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]:
        """找出与当前任务相似的过往情景。"""
        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:
        """检索相关记忆,并把它们格式化好以便注入。"""
        # 语义搜索相关事实
        memories = self.memory.recall(user_message, k=4)
        # 相似的过往任务做法
        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. 检索相关记忆
        memory_context = self._build_memory_context(user_message)
 
        # 2. 构造系统提示词,把记忆注入进去
        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. 调用模型
        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. 把有用的信息存进记忆,留给下一次用
        self.memory.remember(
            content=f"User asked: {user_message[:200]}",
            memory_type="interaction"
        )
 
        # 5. 记录这条情景
        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,  # 生产环境中这个值会来自评估
        ))
 
        return answer

向量数据库

它是任何一套严肃的记忆系统的核心。它不靠精确匹配来查询(像 SQL 那样),而是在高维空间里寻找某个向量的最近邻。正是这一点,使语义搜索成为可能——找出那些在概念上相关、即便没有任何共同词语的记忆。

相似度搜索是怎么运作的

每一条记忆都会被转换成一个向量(用 OpenAI 的嵌入模型,就是一个包含 1536 个浮点数的数组)。概念上相似的文本会产生相似的向量。当你查询时,你把查询也嵌入成向量,然后用余弦相似度找出最接近的那些向量。

import numpy as np
 
def cosine_similarity(a: list, b: list) -> float:
    """
    1.0  = 含义完全相同
    0.0  = 毫不相关
    -1.0 = 含义完全相反
    """
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
 
# 示例:下面这两句话会有很高的相似度
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(非常相似)

本地开发就从 ChromaDB 起步。当你准备好部署时,如果你本来就在用 Postgres,可以评估一下 pgvector——零额外基础设施。当你需要应对真正的大规模场景时,再用 Pinecone 或 Qdrant。

记忆管理

真正的记忆系统不会只是一味地堆积。它们会精挑细选。一个不断膨胀、毫无重点的存储会随时间推移而退化——检索变得越来越嘈杂,延迟一路攀升,相互矛盾的记忆把智能体搞糊涂。

你需要一套遗忘策略。下面是三种主要的做法:

1. 基于时间的衰减

越旧的记忆,相关性越低。把记忆按“新近程度”和“语义相关性”的组合来打分。研究中用到的公式是这样的:

import math
from datetime import datetime
 
def memory_score(
    relevance: float,      # 余弦相似度 0–1
    importance: float,     # 写入时存下来的 0–1
    created_at: datetime,  # 记忆形成的时间
    recency_weight: float = 0.3,
    decay_factor: float = 0.995
) -> float:
    """
    灵感来自 Generative Agents 论文(Park 等人,2023)。
    在三者之间做权衡:有多相关、有多重要、有多新近。
    """
    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. 写入时的重要性评分

在存储一条记忆时,让模型为它自己的输出打一个重要性分。只存那些高分的条目。这样可以从源头上过滤掉噪声。

import re
 
async def score_importance(client, content: str) -> float:
    """问问 LLM 这条信息值不值得保存(0.0 到 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", # 使用当前可用的 Haiku 模型
            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  # 默认兜底值

3. 周期性整合

跑一个每晚执行的任务,把重复的或高度相似的记忆合并成一条规范化的摘要。这类似于人类的睡眠是如何整合记忆的。

async def consolidate_memories(store: MemoryStore, similarity_threshold: float = 0.92):
    """用向量搜索高效地合并近似重复的记忆。"""
    
    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"]
        )
 
        # 识别出同组成员(1.0 - 距离 = 余弦相似度)
        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) # 很可能需要是 async 的
            consolidated_docs.append(summary)
        else:
            consolidated_docs.append(doc)
 
    # 近乎原子的替换:清空,然后重新填充
    store.collection.delete(where={})
    for doc in consolidated_docs:
        await store.remember(doc)

结语

说到底,记忆才是让一个 AI 感觉起来不那么像工具、而更像伙伴的东西。没有记忆,每一次互动都从零开始。有了记忆,智能体就能理解、适应,并随时间不断进步。

真正的力量不只在模型本身;它在于你如何设计——模型记住什么、忘掉什么,以及它如何使用这些信息。

把记忆层搭对,其余的一切就都会变得更聪明。

代码由 AI 生成。

想看更多这样的内容,关注 @techwith_ram

相关笔记