一种具有溯源功能的四层记忆架构，专为人工智能智能体设计
短期记忆与长期记忆的分离、四层结构（缓冲层、情景层、语义层、程序层）、存储选择、异步写入，以及防止内存污染的机制。
想象一下，在一个漫长的工作流程的第十四步出现故障的智能体。这并非模型的问题，只是该智能体不再记得自己之前做了什么。这种故障遵循一种常见的模式：工具调用和提示语都很正常，但一旦对话超过某个隐形的长度限制，智能体就会开始陷入循环，重复已经得到解答的问题，获取它早已拥有的数据，并且在第六步到第十二步之间自相矛盾，仿佛前五步从未发生过一样。
更换不同的前沿模型也很难改变这一状况。真正的瓶颈在于模型周围的基础设施，更确切地说，是缺乏真正的记忆架构。对于聊天机器人而言，将最近的N条消息插入提示语中就能解决问题，但当智能体需要执行二十步的工作流程，并且在第十七步时仍需记住第三步时的决策时，这种方法就不再有效了。
解决这一问题的方案是四层内存堆栈，每层背后都有相应的存储空间。下文将介绍每一层的功能、GitHub、Anthropic和Berkeley的团队如何描述他们的实现版本，以及每部分的Python代码示例。
上下文窗口并非内存
上下文窗口属于工作内存：即RAM，而非硬盘。其中的每个标记都有相应的成本。将整个对话历史塞进128K的窗口看似安全，但一旦使用需要大量工具的智能体，问题就会显现。包含30步操作、工具调用、观察结果和推理过程的流程会迅速消耗掉10万多个标记，而当窗口被填满后，最旧的内容会被移除，而这些内容通常正是用户最初的约束条件与任务描述所在。
一个更为隐蔽的问题情况更糟。斯坦福大学和加州大学伯克利分校的研究人员于2024年在ACL期刊上发表了一篇题为《迷失在中间：语言模型如何处理长上下文》的研究，发现大语言模型的性能在上下文窗口内呈现U形曲线。模型对位于上下文开头和结尾的信息处理得很好，而对埋在中间处的信息则表现明显较差。这一现象在多种架构、不同模型规模以及不同上下文长度下均存在，测试的窗口大小从4K到32K个标记不等。麻省理工学院在2025年进行的后续研究从数学角度对此进行了解释，认为因果注意力掩码和旋转位置编码都会降低序列两端远处标记的注意力权重。
对于模型代理而言，其后果显而易见。即便所有条件都满足，模型也无法均匀地处理这些信息。在60K令牌上下文限制下达到30K时，某些关键约束很可能会被完全忽略。
模型会遗忘信息，因为系统输入的信息量超出了它可靠处理的能力，而且没有任何机制来控制进入处理窗口的内容。
初始划分：短期记忆与长期记忆
在四层分类法出现之前，大多数工程师心中其实有一个更简单的划分方式。人工智能系统中的记忆可分为短期记忆和长期记忆，了解二者各自的局限性有助于设计更精细的架构。
短期记忆的效力仅限于当前会话，仅存在于单次交互过程中。它存储着智能体当下所需的信息：当前任务的状态、所处的步骤、上一个工具的输出以及正在构建的推理过程。实际上这就是上下文窗口，它具有高度动态性，会不断更新，每个标记的处理成本较高，并且在会话结束时会被清除。
LangChain的文档对此有很好的描述，将短期记忆定义为用于存储对话历史、工具输出以及临时推理状态。但未提及的是，当规模扩大时，这部分内容也会首先被挤占。在包含二十个智能体步骤的工作流中，中间输出、反思结果、搜索结果以及工具回复会不断累积，共同争夺有限的存储空间。这就好比在有人不断用新信息打断你的同时，还试图在脑海中记住一个复杂的计划。
短期记忆包含两个实际组成部分：
- 任务状态是进展的工作记录：当前步骤、待处理的子任务、工具执行状态以及各智能体之间的协作情况。例如，一个旅行规划智能体会记录目的地、预算限制、已查询的航班信息以及候选酒店，这样在进入第八步时就不必重复第二步的工作。
- 中间输出则是执行过程中的推理结果：工具的输出、生成的摘要、SQL查询结果、搜索片段以及临时笔记。Anthropic的工程团队发现，在各类生产系统中，那些将笔记保存在上下文窗口之外并在后续再调用的智能体，在处理长任务时表现优于那些试图将所有信息都保留在上下文中的智能体。记笔记本身也是一种短期记忆策略。
长期记忆能够超越会话的持续时间而存在。它使智能体能够记住用户喜欢简洁的回答、上周的部署因迁移问题失败，或是数据库架构在1月发生了变化。LangChain将其定义为可在不同线程和会话之间持续存在的记忆，实际上这类数据需要存储在数据库中，而非上下文窗口中。
它主要有两种类型：
- 用户偏好是随着时间逐渐积累的个性化信号，包括写作风格、语气、技术深度、工具选择以及格式习惯。一个能够记住你始终要求使用Markdown格式且需要高技术深度内容的写作助手，其实用性是任何无状态系统都无法比拟的，无论该系统的模型有多优秀。
许多团队也以同样的方式犯错。他们通过填满上下文窗口来实现短期记忆，又通过构建向量存储来管理长期记忆。这两种做法都有问题，或者至少不够完善。第一种由于上述原因无法在大规模环境中使用；第二种则行不通，因为相似度搜索并不适合作为获取会话日志、偏好设置或行为规则的手段，而且不同类型的长期记忆需要不同的后端支持。四层设计是对短期/长期记忆划分的进一步优化，它将每种记忆类型对应到所需的存储方式。
四种记忆类型，一种分类体系
那些正在构建生产级 AI 系统的团队都独立得出了相同的分类方式。名称虽有所不同，但结构却一致。
2023 年，加州大学伯克利分校的 Charles Packer 及其同事提出的 MemGPT 项目推广了这样一种理念：将大型语言模型的上下文窗口视作有限且宝贵的 RAM。操作系统通过分页和磁盘存储来扩展内存，而智能体框架也可以通过将较旧的上下文移至更大容量的存储中并在需要时调取，从而模拟长期记忆。这一理念后来成为了 Letta 的基础。
LangChain 的 LangMem 团队、谷歌 DeepMind 开发的 ReMem 框架以及 Mem0 的生产级架构，都描述了四种类型的变体：
- 工作记忆指的是当前上下文中的内容：当前的对话内容、调取的文档以及工具的输出结果。它的处理速度很快，容量有限，且每处理一个标记的成本较高。
- 情景记忆用于存储过去交互的记录，比如上周二智能体为某位用户做了什么以及最终的结果如何。这类记忆按时间顺序排列，仅限于当前会话范围，通常存储在快速的键值存储系统或SQL数据库中，并通过最近性或语义相似度来检索。
- 语义记忆则保存通用事实、领域知识以及在过程中学到的各种信息。它相当于RAG层，每当任务需要领域知识或之前的分析结果时，就会通过嵌入向量之间的相似度来调用这部分信息。
这些分类与认知科学理论相呼应，这也解释了为何各团队会采用这类分类方式。它们对应于人类神经科学中的情景记忆、语义记忆和程序性记忆系统，而将工作记忆视为上下文窗口也并非比喻，而是最接近的实际功能类比。
选择存储方式：SQL、向量还是图结构
应该用哪种数据库来存储记忆？在编写代码之前，诚实的答案是这取决于所处理的记忆类型。
向量数据库（如 Weaviate、Pinecone、Qdrant、Chroma）非常适合用于语义记忆存储，因为它们能够在大规模数据下快速实现基于嵌入向量的相似度搜索，而且目前大多数这类数据库都提供了将 BM25 与向量检索相结合的混合检索方式，其在文本处理方面的性能明显优于纯向量搜索。不过这类数据库缺乏时间推理和关系查询功能，此外，两个嵌入向量几乎完全相同但事实却相互矛盾的情况也可能会误导用户。因此应避免将事件记录和结构化的审计轨迹存储在向量数据库中。
SQL数据库，如PostgreSQL和SQLite，非常适合处理阶段性记忆类数据。查询是结构化的，可按用户ID、会话、任务类型或时间戳进行筛选，写入量较低，而且当智能体需要记录自己的操作内容及时间时，ACID特性至关重要。带有pgvector扩展的PostgreSQL能够在同一系统中同时实现关系型查询和向量搜索。在记录数量低于数千万条时，这是最实用的选择，因为它具备熟悉的操作方式、低成本的托管服务，还有便于调试的SQL语言。
图数据库，如Neo4j，仅在阶段性数据或语义数据具有复杂的关联结构、需要多跳查询时才适合用于内存系统，例如在文档创建时确定三个月前是谁向谁汇报了情况。简单的智能体任务记录则不适用此类数据库。很多团队最初选择图数据库，后来却发现其实更需要SQL。
在生产环境部署中，最常见的模式是使用关系型存储（SQL或PostgreSQL）来保存剧集和用户资料，用向量存储进行语义查询，并通过进程内状态管理工作上下文；而对于那些希望降低运营负担的团队，则会采用PostgreSQL结合pgvector的混合方案。应根据内存类型的功能需求来选择存储层级，不必对所有数据都进行向量化处理。
第一层：带汇总功能的缓冲内存
最基础的形式是一个容量固定的双端队列：
from collections import deque
from typing import List, Dict
class BufferMemory:
def __init__(self, max_turns: int = 10):
# Each turn = one user message + one assistant message
self.history: deque = deque(maxlen=max_turns * 2)
def add_message(self, role: str, content: str):
self.history.append({"role": role, "content": content})
def get_context(self) -> List[Dict]:
return list(self.history)
def token_estimate(self) -> int:
total_chars = sum(len(m["content"]) for m in self.history)
return total_chars // 4 # rough approximation
def clear(self):
self.history.clear()
这种方法适用于简短的对话。在较长的对话中，旧消息会逐渐出现在末尾，导致模型难以把握重点。LangMem将一种更好的方法称为“热路径”法，它会在丢弃旧消息之前将其汇总为摘要：
import os
from collections import deque
from typing import List, Dict
from openai import OpenAI
class SummarizingBufferMemory:
def __init__(self, max_turns: int = 8, summary_batch: int = 4):
self.client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
self.model = "deepseek/deepseek-v4-flash-0731"
self.recent: deque = deque(maxlen=max_turns * 2)
self.rolling_summary: str = ""
self.summary_batch = summary_batch
def add_message(self, role: str, content: str):
if len(self.recent) == self.recent.maxlen:
self._compress_oldest()
self.recent.append({"role": role, "content": content})
def _compress_oldest(self):
batch = [self.recent.popleft() for _ in range(min(self.summary_batch * 2, len(self.recent)))]
text = "\n".join(f"{m['role']}: {m['content']}" for m in batch)
response = self.client.chat.completions.create(
model=self.model,
max_tokens=300,
messages=[
{
"role": "system",
"content": "You compress conversation history. Reply with the summary only.",
},
{
"role": "user",
"content": (
"Summarize this conversation segment in 2-4 sentences. "
"Preserve any decisions made, constraints stated, and conclusions reached.\n\n"
f"{text}"
),
},
],
)
new_summary = response.choices[0].message.content.strip()
if self.rolling_summary:
self.rolling_summary = f"{self.rolling_summary} | {new_summary}"
else:
self.rolling_summary = new_summary
def get_context(self) -> List[Dict]:
context = []
if self.rolling_summary:
context.append({
"role": "system",
"content": f"[Prior conversation summary: {self.rolling_summary}]",
})
context.extend(list(self.recent))
return context
LangMem将实时运行的热路径摘要功能（会为用户带来延迟）与会话结束后才执行的后台摘要功能分开。对于面向用户的应用而言，几乎总是后台摘要更优：因为可以分析整个对话内容，所以摘要质量更高，且无需用户等待。而对于无人值守运行的自主智能体来说，热路径摘要也是可以接受的，反正也没有用户在等待。
第二层：能够自我校验引用的情景记忆
在这方面很多团队都存在不足：他们虽然构建了检索系统，却从不记录智能体实际执行了什么操作。
GitHub Copilot的工程博客介绍了一种值得借鉴的模式。每条记忆条目都包含引用信息，即文件路径、URL、工具名称和行号等具体位置，这样在检索时就能确认存储的事实是否仍然有效。GitHub将此称为即时验证。当智能体在任务开始时加载某条记忆时，它会检查所引用的来源是否仍然存在且仍能支持该说法。如果不存在或不再有效，这条记忆就会被修正或丢弃，从而无需离线整理即可让记忆库保持良好状态。
GitHub报告了可量化的效果：具备记忆功能的编码智能体的拉取请求合并率提升了7%，从83%上升至90%。
import json
import sqlite3
from datetime import datetime, timedelta
from dataclasses import dataclass, field
@dataclass
class Episode:
task_type: str
user_query: str
outcome: str
tools_used: list
citations: list # source references for validation
duration_seconds: float
success: bool
session_id: str = ""
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
class EpisodicMemory:
EXPIRY_DAYS = 28 # GitHub Copilot's production default
def __init__(self, db_path: str = "episodes.db"):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_schema()
def _init_schema(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS episodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_type TEXT NOT NULL,
user_query TEXT,
outcome TEXT,
tools_used TEXT,
citations TEXT,
duration_s REAL,
success INTEGER,
session_id TEXT,
created_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_task_type ON episodes (task_type, success, created_at);
""")
self.conn.commit()
def record(self, ep: Episode):
self.conn.execute(
"""
INSERT INTO episodes
(task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
ep.task_type, ep.user_query, ep.outcome,
json.dumps(ep.tools_used), json.dumps(ep.citations),
ep.duration_seconds, int(ep.success),
ep.session_id, ep.created_at,
),
)
self.conn.commit()
def recall_similar(self, task_type: str, limit: int = 3) -> list[Episode]:
cutoff = (datetime.utcnow() - timedelta(days=self.EXPIRY_DAYS)).isoformat()
cursor = self.conn.execute(
"""
SELECT task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at
FROM episodes
WHERE task_type = ? AND success = 1 AND created_at > ?
ORDER BY created_at DESC
LIMIT ?
""",
(task_type, cutoff, limit),
)
return [
Episode(
task_type=r[0], user_query=r[1], outcome=r[2],
tools_used=json.loads(r[3]), citations=json.loads(r[4]),
duration_seconds=r[5], success=bool(r[6]),
session_id=r[7], created_at=r[8],
)
for r in cursor.fetchall()
]
def validate_citations(self, episode: Episode, validator_fn) -> bool:
"""
validator_fn(citation: str) -> bool
Check if cited sources are still valid (file exists, URL responds, etc.)
Return False if any citation fails; the episode should be discarded.
"""
return all(validator_fn(c) for c in episode.citations)
28天的有效期遵循了GitHub的生产环境选择。旧版本内容不会被删除，但将无法被检索到；而对经过日期过滤的索引使用LIMIT查询速度很快，且无需后台整理任务。
第三层：基于Weaviate混合搜索的语义记忆
这就是RAG层。大多数指南仅介绍其设置与停止使用方法，而实际生产中的问题主要在于检索质量与数据来源的可靠性。
常见的第一个错误是将语义记忆设置为只读模式。长期运行的智能体需要将其设为读写模式：当智能体在执行任务时发现某个事实，经验证后认为值得保留，就应将其写入语义记忆中以供后续会话使用。第二个错误是仅依赖余弦相似度。Weaviate的混合搜索结合了BM25关键词匹配与向量相似度，其在智能体记忆检索方面的表现通常优于单独使用任一方法。
第三，这也是导致数据被污染的原因——这些条目往往没有明确的来源信息。每条记录都需要有来源、置信度评分以及会话标识符。缺少这些要素的话，就无法实现基于信任度的检索；那些过时或存疑的条目也无法被降权处理，而要找出错误的检索结果也只能靠猜测。
import uuid as uuid_lib
from datetime import datetime
import weaviate
from weaviate.classes.config import Configure, Property, DataType, VectorDistances
from weaviate.classes.query import MetadataQuery, HybridFusion, Filter
def embed(text: str) -> list[float]:
"""
Swap in any embedding source: OpenAI, Cohere, sentence-transformers, etc.
Returns a normalized float vector.
"""
from sentence_transformers import SentenceTransformer
_model = SentenceTransformer("all-MiniLM-L6-v2")
return _model.encode(text, normalize_embeddings=True).tolist()
class WeaviateSemanticMemory:
COLLECTION = "AgentMemory"
def __init__(self, host: str = "localhost", port: int = 8080):
self.client = weaviate.connect_to_local(host=host, port=port)
self._ensure_collection()
def _ensure_collection(self):
if self.client.collections.exists(self.COLLECTION):
return
self.client.collections.create(
name=self.COLLECTION,
# We provide our own vectors; no built-in vectorizer needed.
# Swap to Configure.Vectorizer.text2vec_openai() if you prefer managed embedding.
vectorizer_config=Configure.Vectorizer.none(),
vector_index_config=Configure.VectorIndex.hnsw(
distance_metric=VectorDistances.COSINE
),
properties=[
Property(name="content", data_type=DataType.TEXT),
Property(name="task_type", data_type=DataType.TEXT),
Property(name="source", data_type=DataType.TEXT),
Property(name="citations", data_type=DataType.TEXT),
Property(name="session_id", data_type=DataType.TEXT),
Property(name="confidence", data_type=DataType.NUMBER),
Property(name="created_at", data_type=DataType.TEXT),
],
)
def store(
self,
content: str,
task_type: str = "",
source: str = "agent",
citations: str = "",
session_id: str = "",
confidence: float = 1.0,
) -> str:
collection = self.client.collections.get(self.COLLECTION)
doc_id = str(uuid_lib.uuid4())
collection.data.insert(
properties={
"content": content,
"task_type": task_type,
"source": source,
"citations": citations,
"session_id": session_id,
"confidence": confidence,
"created_at": datetime.utcnow().isoformat(),
},
vector=embed(content),
uuid=doc_id,
)
return doc_id
def retrieve_hybrid(
self,
query: str,
n_results: int = 5,
min_confidence: float = 0.6,
task_type: str = None,
) -> list[dict]:
collection = self.client.collections.get(self.COLLECTION)
# Filter by confidence floor and optionally by task type
confidence_filter = Filter.by_property("confidence").greater_or_equal(min_confidence)
if task_type:
active_filter = (
Filter.by_property("task_type").equal(task_type) & confidence_filter
)
else:
active_filter = confidence_filter
results = collection.query.hybrid(
query=query,
vector=embed(query),
limit=n_results,
fusion_type=HybridFusion.RELATIVE_SCORE,
filters=active_filter,
return_metadata=MetadataQuery(score=True),
)
return [
{
"content": obj.properties["content"],
"score": obj.metadata.score,
"confidence": obj.properties.get("confidence", 1.0),
"source": obj.properties.get("source", ""),
"citations": obj.properties.get("citations", ""),
"uuid": str(obj.uuid),
}
for obj in results.objects
]
def update_confidence(self, doc_id: str, new_confidence: float):
collection = self.client.collections.get(self.COLLECTION)
collection.data.update(
uuid=doc_id,
properties={"confidence": new_confidence},
)
def close(self):
self.client.close()
在许多实现中，retrieve_hybrid函数缺少两个重要细节。其一，Weaviate的混合端点会同时接收原始查询文本（用于BM25搜索）及其嵌入向量（用于向量搜索）。精确的关键词匹配来自前者，而语义相似度匹配则来自后者，将两者结合使用的效果远优于单独使用任一方式。其二就是min_confidence过滤机制，该机制会将可信度较低的记忆排除在上下文之外，无论其在语义上与查询的相似程度如何。实际上这是一种安全控制措施，而不仅仅是相关度调整手段。
update_confidence方法负责提供反馈机制。如果在执行任务时智能体的推理结果与检索到的记忆相矛盾，则应降低该记忆的置信度；如果引用验证确认其正确性，则应提高其置信度。这样的机制能让记忆库不断通过实际使用进行优化。
第4层：程序性记忆与提示词演化
这一层最容易被忽视。最初的设计是使用一个包含规则的JSON文件，智能体读取这些规则并在出错后加以遵循或扩展。虽然这种方法可行，但存在局限，因为单独文件中的规则会争夺上下文空间，且需要手动检索。
LangChain的LangMem团队提出了一个值得借鉴的理念：在实际应用中，程序化记忆以系统提示语的形式存在。它既不存储在提示语之外，也不从中被检索出来，本身就是提示语。每当一次失败让智能体学会新的行为规则时，提示语本身就会被重写，他们将这种做法称为“程序化记忆的提示语优化”。由此学到的行为会变成模型在每次交互开始时都会读取的自然语言指令。
对于大多数团队而言，应从更简单的JSON文件版本开始：
import json
import os
import re
from datetime import datetime, timedelta, timezone
from openai import OpenAI
MODEL = "deepseek/deepseek-v4-flash-0731"
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
class ProceduralMemory:
def __init__(self, rules_path: str = "procedural_rules.json"):
self.rules_path = rules_path
self.rules: list[dict] = self._load()
def _load(self) -> list[dict]:
try:
with open(self.rules_path) as f:
return json.load(f)
except FileNotFoundError:
return []
def add_rule(self, situation: str, action: str, reason: str, confidence: float = 1.0):
self.rules.append({
"situation": situation,
"action": action,
"reason": reason,
"confidence": float(confidence),
"added_at": datetime.now(timezone.utc).isoformat(),
"trigger_count": 0,
})
self._save()
def get_applicable_rules(self, context: str, min_confidence: float = 0.7) -> list[dict]:
relevant = []
ctx_lower = context.lower()
dirty = False
for rule in self.rules:
if rule.get("confidence", 1.0) < min_confidence:
continue
keywords = rule["situation"].lower().split()
if not keywords:
continue
hits = sum(1 for kw in keywords if kw in ctx_lower)
if hits >= max(1, len(keywords) // 3):
rule["trigger_count"] = rule.get("trigger_count", 0) + 1
relevant.append(rule)
dirty = True
if dirty:
self._save()
relevant.sort(key=lambda r: r.get("confidence", 1.0), reverse=True)
return relevant[:5]
def prune_stale(self, max_age_days: int = 60, min_triggers: int = 2):
cutoff = (datetime.now(timezone.utc) - timedelta(days=max_age_days)).isoformat()
self.rules = [
r for r in self.rules
if r.get("added_at", "") > cutoff or r.get("trigger_count", 0) >= min_triggers
]
self._save()
def _save(self):
tmp = f"{self.rules_path}.tmp"
with open(tmp, "w") as f:
json.dump(self.rules, f, indent=2)
os.replace(tmp, self.rules_path)
def _parse_json(text: str) -> dict:
text = text.strip()
fenced = re.search(r"```(?:json)?\s*(.*?)```", text, re.S)
if fenced:
text = fenced.group(1).strip()
return json.loads(text)
def extract_rule_from_failure(failure_trace: str, memory: ProceduralMemory):
response = client.chat.completions.create(
model=MODEL,
max_tokens=250,
response_format={"type": "json_object"},
messages=[
{
"role": "system",
"content": "You return only a JSON object. No prose, no code fences.",
},
{
"role": "user",
"content": (
"A task failed. Extract one behavioral rule to prevent this failure.\n"
"Return ONLY valid JSON with keys: situation, action, reason, "
"confidence (0.0-1.0)\n"
f"Failure trace:\n{failure_trace}"
),
},
],
)
try:
rule = _parse_json(response.choices[0].message.content)
except (json.JSONDecodeError, AttributeError, TypeError):
return None
if not all(k in rule for k in ("situation", "action", "reason")):
return None
memory.add_rule(
situation=str(rule["situation"]),
action=str(rule["action"]),
reason=str(rule["reason"]),
confidence=float(rule.get("confidence", 1.0)),
)
return rule
prune_stale在流程层面解决遗忘问题。因一次异常故障而创建且此后再未被触发的规则应当被移除，而那些经常被调用的规则则无论年龄多大都会保留。这虽是一种粗略的方法，但在此层实现遗忘处理最为简单。
一旦智能体能够持续运行，接下来的自然步骤就是将那些既可靠又高频使用的规则纳入系统提示语中。LangMem通过提示语优化循环来实现这一功能，而对于规模较小的系统，手动筛选也同样可行。
各层之间的连接
下表展示了这四层在实际应用中的组合方式。有一个原则是不可违背的：对内存的写入绝不能影响响应速度。
import json
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
@dataclass
class TaskResult:
task_type: str
query: str
outcome: str
tools_used: list
citations: list
duration_seconds: float
success: bool
new_facts: list[dict] = field(default_factory=list)
session_id: str = ""
class MemoryManager:
def __init__(
self,
weaviate_host: str = "localhost",
episodic_db: str = "episodes.db",
rules_path: str = "procedural_rules.json",
):
self.buffer = SummarizingBufferMemory(max_turns=8)
self.episodic = EpisodicMemory(db_path=episodic_db)
self.semantic = WeaviateSemanticMemory(host=weaviate_host)
self.procedural = ProceduralMemory(rules_path=rules_path)
self._pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="memory_write")
def build_context(self, query: str, task_type: str) -> list[dict]:
context: list[dict] = []
# Procedural rules first: they constrain behavior throughout the task
rules = self.procedural.get_applicable_rules(query)
if rules:
rules_text = "\n".join(
f"- When '{r['situation']}': {r['action']} (reason: {r['reason']})"
for r in rules
)
context.append({"role": "user", "content": f"[Behavioral rules:\n{rules_text}]"})
# Semantic facts: domain knowledge and past discoveries
facts = self.semantic.retrieve_hybrid(query, n_results=5, min_confidence=0.6)
if facts:
facts_text = "\n".join(f"- {f['content']}" for f in facts)
context.append({"role": "user", "content": f"[Relevant knowledge:\n{facts_text}]"})
# Past episodes: outcome templates for similar tasks
episodes = self.episodic.recall_similar(task_type, limit=3)
if episodes:
ep_text = "\n".join(
f"- Outcome: {e.outcome} (tools: {', '.join(e.tools_used)})"
for e in episodes
)
context.append({"role": "user", "content": f"[Past similar tasks:\n{ep_text}]"})
# Current conversation last: the model reads this most carefully
context.extend(self.buffer.get_context())
return context
def record_turn(self, role: str, content: str):
self.buffer.add_message(role, content)
def persist(self, result: TaskResult):
# Submit to thread pool and return immediately; never block the caller
self._pool.submit(self._persist_worker, result)
def _persist_worker(self, result: TaskResult):
ep = Episode(
task_type=result.task_type,
user_query=result.query,
outcome=result.outcome,
tools_used=result.tools_used,
citations=result.citations,
duration_seconds=result.duration_seconds,
success=result.success,
session_id=result.session_id,
)
self.episodic.record(ep)
for fact in result.new_facts:
self.semantic.store(**fact)
if not result.success:
extract_rule_from_failure(result.outcome, self.procedural)
def shutdown(self):
self._pool.shutdown(wait=True)
self.semantic.close()
build_context中上下文的顺序是经过精心设计的。关于位置注意力偏差的研究表明，模型会更重视窗口的起始和结束部分。行为规则被放在最顶部，因为它们是应当影响后续所有内容的约束条件。当前对话则位于底部，因为这是模型接下来需要回答的内容。语义事实与过往对话记录则处于中间位置，在提供参考的同时不会干扰当前任务。
self._pool.submit(self._persist_worker, result)这行代码是该类中最重要的工程决策。由于每次写入操作都会立即返回，因此在任何数据库写入开始之前，大语言模型的响应就已经开始流向用户了。
决定架构的三个关键决策
在确定了四层架构之后，三项架构决策会影响到其他所有方面，而许多团队至少会犯一项错误。
热路径写入与后台写入。LangMem团队在这一点上投入了大量精力。热路径写入发生在对话进行期间，当智能体发现值得存储的事实时会立即执行写入操作。后台写入则在会话结束后进行，通过分析整个对话内容来确定哪些信息需要保留。热路径写入会增加延迟，因为响应循环中的每一个semantic.store()调用都意味着一次数据库提交，用户必须等待其完成。而后台写入则能提升信息检索的质量，因为它可以审视整个对话中的主题与决策，并存储更丰富的摘要。对于面向用户的实际应用而言，几乎总是选择后台写入方案。persist()中的线程池能够在避免完全异步处理流程的同时，带来大部分性能提升。
检索或摘要生成。这两种方法解决的是不同的问题，但团队常常将它们混为一谈。检索是指在庞大的语料库中找到相关段落。查询具有明确性，而语料库规模较大且相对稳定，比如文档、产品目录或代码。当查询较为复杂、相关信息分散在多个段落中，或是语料库更新速度超过嵌入索引的更新速度时，检索方法就会失效。摘要生成则是将对话或文档压缩以适配特定上下文。当输入为序列形式，如对话历史记录，并且可以接受一定程度的信息损失时，该方法能有效工作；但如果被舍弃的细节其实很重要，它就会失败。2025年，该行业开始用“上下文工程”来替代这两个术语：问题不再是在检索与摘要生成之间做选择，而是如何结合存储的知识、检索到的段落以及压缩后的历史记录，以获得最佳效果。
此次推理调用所处的上下文。内存管理器的build_context方法属于上下文构建技术，因为它决定了要从四个存储中读取哪些数据、从每个存储中读取多少内容，以及如何将这些数据组合起来。 从一开始就确立溯源机制。无法在已有的内存系统中事后添加溯源功能。从第一次写入数据开始，每个条目都应包含其来源、置信度得分、会话ID以及创建时间。没有这些信息，就无法实现基于信任度的检索、无法按最新程度进行过滤，也无法对错误的检索结果进行审计。那些在初期阶段就忽略溯源机制的团队，最终会花费数周时间试图将这一功能添加到原本已存储着数千条且未包含溯源信息的数据库中。
内存污染：攻击模式与防御措施
这是内存设计指南往往遗漏的部分，但实际上每份指南都应包含它。
2025年发表在arXiv上的一篇论文提出了MINJA攻击框架，这是目前针对大语言模型智能体内存投毒现象最全面的公开分析之一。该论文的作者指出，攻击者无需特殊权限，仅通过常规用户交互，即可利用三步流程将注入成功率提升至95%以上。第一步为桥接步骤，将无害的合法内容与攻击者目标主题关联起来；第二步为暗示步骤，提供微妙线索使有害内存看起来可信；第三步为逐步压缩步骤，将注入内容压缩到极小规模，从而能在内存提取后依然存活。
内存投毒与普通的提示注入不同之处在于，被污染的内存会跨越会话边界并持续累积效应。在第一个会话中的一次成功注入，可能会无限期地影响后续的所有会话。
2025年另一篇关于贝叶斯信任防御的论文提出了一种覆盖攻击面的五层设计：
- 写入时的输入审核。在每次内存写入之前，会根据多个独立信号计算综合信任分数：来源属性、指令模式的语义分析以及针对用户历史记录的异常检测。单靠某一信号是不够的，而MINJA注入技术正是为通过单一信号的检测而设计的。
- 写入前的指令过滤。如果某个待存储的内容看起来像命令（如“始终”、“从不”、“你必须”、“不要”），则应对其进行清理或直接丢弃，而非保存下来。真实的事实性内存无需向智能体下达指令。
min_confidence过滤功能就是一个例子。来源不可靠的记忆无论其相似度多高都不会被纳入上下文考虑。在更完善的实现中，信任度字段会根据验证结果进行更新，因此经过验证的记忆会获得更高的信任度，而与之矛盾的记忆则会失去信任度。对于大多数团队而言，合理的处理顺序是首先在每次写入时对数据来源进行标记，然后在写入时检测指令，最后再通过信任度加权的方式进行检索。行为监控则放在最后，因为它需要一个行为基准，而建立这样的基准需要数周的生产环境数据。
还有一点：针对已有合法条目的真实内存系统所进行的研究表明，与实验室环境相比，现实条件会大大降低攻击的效率，因为被篡改的条目会与合法数据竞争检索位置。但这并非可以忽略防御措施的理由，反而说明必须保持合法条目的高质量和高可信度，因为它们才是第一道防线。
应首先构建什么
并非立即就需要全部四层功能。对于大多数人工智能应用而言，其中两层就能解决约80%的内存相关问题：第一层（缓冲区加摘要功能）和第三层（具备混合搜索、置信度过滤及溯源功能的Weaviate）。这两层足以处理最常见的问题，即智能体在任务执行过程中遗忘部分内容，以及缺乏领域知识的情况。先构建这两层，对比任务完成情况，在此之后再判断还缺少哪些功能。
当发现自己会问智能体为何忘记了上一次任务的结果时，就该添加情景记忆，因为这个问题正是需要引入该功能的信号。不要过早添加。而当同一类型的错误在多个任务中反复出现且提示工程无法解决时，才需要添加程序记忆，因为行为模式本身就需要相应的行为记忆来支撑。
对于大多数团队而言，搭配 pgvector 的 PostgreSQL 是最合理的默认选择。通过执行 CREATE EXTENSION IF NOT EXISTS vector 即可创建一个数据库实例，它具备 ACID 遵从性、灵活的 SQL 语法、向量相似度搜索功能以及混合评分机制；据估算，10K 条向量的存储成本约为每月 45 美元。只有当语义向量数据量增长到足以让 PostgreSQL 的 HNSW 索引成为性能瓶颈时，独立的向量数据库才具有价值，而这种情况大多数团队是永远不会遇到的。
应将每次向量数据的写入操作设为异步处理。响应用户请求不应依赖于数据库的提交操作，因此可将该任务交由线程池处理，然后继续执行其他操作。
在生产环境中请避免使用内存中的向量存储方案。没有持久化存储的部署方式会在服务器首次重启时丢失所有已积累的数据，这种错误很少有人会犯两次。