第3章 记忆系统：构建Agent的长短期记忆

人类之所以能进行复杂的对话和决策，是因为我们有记忆。同样，一个实用的Agent也需要记忆能力——它需要记住之前的对话、积累的知识、以及用户偏好。本章将深入探讨如何为Agent构建完整的长短期记忆系统。

3.1 上下文窗口管理：短期记忆的滑动窗口策略

上下文窗口的本质

LLM的上下文窗口（Context Window）是Agent的"工作记忆"。GPT-4o支持128K Token，Claude 3.5支持200K，但这些都不是无限的。当对话历史超出窗口限制时，早期的内容就会被截断。

关键认知：上下文窗口不是越满越好。研究表明，当上下文接近窗口上限时，模型的注意力分配会下降，导致关键信息被忽略。这就是所谓的"Lost in the Middle"现象。

滑动窗口策略

最基础的短期记忆管理是滑动窗口（Sliding Window）：

class SlidingWindowMemory:def __init__(self, max_tokens: int = 8000):self.max_tokens = max_tokensself.messages: list[dict] = []self.approx_tokens_per_msg = 4# 粗略估算：1 token约0.25个英文单词def add_message(self, role: str, content: str):msg_tokens = len(content) // self.approx_tokens_per_msgself.messages.append({"role": role, "content": content})self._trim_to_fit()def _trim_to_fit(self):total_tokens = sum(len(m["content"]) // self.approx_tokens_per_msgfor m in self.messages)while total_tokens > self.max_tokens and len(self.messages) > 2:# 保留系统消息和最新消息，删除中间旧消息removed = self.messages.pop(1)total_tokens -= len(removed["content"]) // self.approx_tokens_per_msgdef get_messages(self) -> list[dict]:return self.messages

LangChain的内存管理

LangChain内置了多种内存管理策略：

from langchain.memory import ConversationBufferWindowMemory# 保留最近K轮对话memory = ConversationBufferWindowMemory(k=5, return_messages=True)memory.save_context({"input": "我叫张三"}, {"output": "你好张三！"})memory.save_context({"input": "我喜欢Python"}, {"output": "Python是很好的选择！"})memory.save_context({"input": "我在北京上班"}, {"output": "北京是个好地方！"})# 只保留最近5轮（10条消息），更早的自动丢弃print(memory.load_memory_variables({}))

3.2 向量数据库基础：Embedding原理与选型（Chroma, Milvus）

Embedding：文本的数值化表示

Embedding是将文本转换为高维向量（通常256-3072维）的技术。语义相近的文本，其向量在空间中的距离也更近。

from langchain_openai import OpenAIEmbeddingsembeddings = OpenAIEmbeddings(model="text-embedding-3-small")# 将文本转换为向量vector1 = embeddings.embed_query("机器学习是一门AI技术")vector2 = embeddings.embed_query("深度学习属于机器学习")vector3 = embeddings.embed_query("今天天气不错")print(f"相似文本距离: {cosine_similarity(vector1, vector2)}")# 距离近print(f"不相关文本距离: {cosine_similarity(vector1, vector3)}")# 距离远

向量数据库选型

Chroma快速上手

Chroma是最易上手的选择，适合开发和测试：

import chromadbfrom chromadb.config import Settings# 创建持久化客户端client = chromadb.PersistentClient(path="./chroma_db")# 创建集合（collection）collection = client.get_or_create_collection(name="agent_knowledge",metadata={"description": "Agent知识库"})# 添加文档collection.add(documents=["LangChain是一个用于构建LLM应用的框架","ReAct是一种结合推理和行动的Agent框架","RAG通过检索增强生成来减少幻觉"],ids=["doc1", "doc2", "doc3"],metadatas=[{"source": "docs", "category": "framework"},{"source": "docs", "category": "methodology"},{"source": "docs", "category": "methodology"}])# 查询相似文档results = collection.query(query_texts=["什么是RAG技术"],n_results=2)print(results["documents"])

3.3 长期记忆实现：基于向量检索的知识存储与召回

Agent长期记忆的架构

长期记忆系统由三个组件构成：

对话内容 -> 分块 -> Embedding -> 向量数据库|用户提问 -> Embedding -> 相似度搜索 -> 召回相关记忆|拼入提示词 -> LLM生成

基于LangChain的完整实现

from langchain_openai import OpenAIEmbeddings, ChatOpenAIfrom langchain_community.vectorstores import Chromafrom langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain.chains import create_history_aware_retriever, create_retrieval_chainfrom langchain.chains.combine_documents import create_stuff_documents_chain# 1. 准备文本分块器text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000,chunk_overlap=200,separators=["nn", "n", "。", " "])# 2. 分块并写入向量库documents = ["第一段对话记录...", "第二段对话记录..."]chunks = text_splitter.create_documents(documents)vectorstore = Chroma.from_documents(documents=chunks,embedding=OpenAIEmbeddings(model="text-embedding-3-small"),persist_directory="./agent_memory")# 3. 创建检索器retriever = vectorstore.as_retriever(search_type="similarity",search_kwargs={"k": 5}# 检索Top-5相关片段)# 4. 构建历史感知检索链llm = ChatOpenAI(model="gpt-4o")retrieval_chain = create_history_aware_retriever(llm, retriever, prompt)# 5. 组合成完整QA链qa_chain = create_retrieval_chain(retrieval_chain, llm)# 6. 使用result = qa_chain.invoke({"input": "我上次提到的那个项目进展如何？","chat_history": []# 可传入之前的对话历史})

记忆召回的精度优化

元数据过滤：为每条记忆添加元数据标签，查询时按标签过滤：

# 存储时添加元数据collection.add(documents=["用户偏好深色主题"],ids=["mem_001"],metadatas={"type": "preference", "user": "zhangsan", "timestamp": "2024-01-15"})# 检索时按元数据过滤results = collection.query(query_texts=["用户喜欢什么主题"],where={"type": "preference", "user": "zhangsan"},n_results=3)

时间衰减权重：近期记忆应比远期记忆有更高权重：

import mathfrom datetime import datetimedef time_decay_score(base_score: float, timestamp: str, half_life_days: int = 30) -> float:"""计算考虑时间衰减的记忆分数"""days_ago = (datetime.now() - datetime.fromisoformat(timestamp)).daysdecay_factor = math.exp(-0.693 * days_ago / half_life_days)# ln(2) ≈ 0.693return base_score * decay_factor

3.4 记忆压缩与总结：处理超长对话历史的算法

为什么需要记忆压缩？

当对话持续进行，原始消息会不断累积。即使用滑动窗口，也会丢失重要上下文。记忆压缩通过总结旧对话来保留核心信息。

滚动总结策略

from langchain_openai import ChatOpenAIfrom langchain_core.messages import SystemMessage, HumanMessage, AIMessageclass RollingSummaryMemory:def __init__(self, summary_interval: int = 10):self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)self.summary_interval = summary_intervalself.messages: list = []self.summary: str = ""self.message_count: int = 0def add_exchange(self, user_msg: str, assistant_msg: str):self.messages.append(HumanMessage(content=user_msg))self.messages.append(AIMessage(content=assistant_msg))self.message_count += 1if self.message_count >= self.summary_interval:self._compress()def _compress(self):"""将当前消息窗口压缩为摘要，追加到总摘要"""messages_text = "n".join([f"{'用户' if isinstance(m, HumanMessage) else '助手'}：{m.content}"for m in self.messages])compress_prompt = f"""请将以下对话压缩为简洁的摘要，保留关键信息（事实、决策、偏好）：已有摘要：{self.summary}新对话：{messages_text}请输出更新后的完整摘要（不是增量部分）。"""response = self.llm.invoke([SystemMessage(content=compress_prompt)])self.summary = response.contentself.messages = []# 清空已压缩的消息self.message_count = 0def get_context(self, current_input: str) -> list:"""获取当前上下文（摘要 + 最近消息）"""context = []if self.summary:context.append(SystemMessage(content=f"历史对话摘要：{self.summary}"))context.extend(self.messages)context.append(HumanMessage(content=current_input))return context

关键信息提取

除了总结全文，还可以提取结构化的关键信息：

EXTRACTION_PROMPT = """从以下对话中提取关键信息，以JSON格式输出：1. facts: 对话中提到的事实信息2. decisions: 做出的决策或选择3. preferences: 用户表达的偏好4. tasks: 提到的待办事项对话：{conversation}"""

这种方式生成的是结构化记忆，更适合Agent后续的精确检索。

3.5 记忆遗忘机制：模拟人类记忆的衰减与更新策略

为什么要让Agent"遗忘"？

人类的记忆系统天然具有遗忘机制——不重要的信息随时间淡去，重要的信息通过反复激活得以强化。Agent同样需要遗忘：

1. 信息过时：用户三个月前的地址可能已经变更
2. 存储成本：无限累积的向量数据会拖慢检索速度
3. 噪音干扰：太多无关记忆会干扰检索精度

实现记忆遗忘

方案一：访问计数 + 时间衰减

class AgentMemory:def __init__(self, decay_half_life_days: int = 60):self.memories: dict = {}self.decay_half_life = decay_half_life_daysdef store(self, key: str, content: str, metadata: dict = None):self.memories[key] = {"content": content,"metadata": metadata or {},"access_count": 1,"last_accessed": datetime.now().isoformat(),"created": datetime.now().isoformat(),}def recall(self, key: str):if key not in self.memories:return Nonemem = self.memories[key]mem["access_count"] += 1mem["last_accessed"] = datetime.now().isoformat()return mem["content"]def get_relevance_score(self, key: str) -> float:"""计算记忆的相关性分数（综合访问频率和时间衰减）"""mem = self.memories[key]frequency_boost = min(mem["access_count"] / 10, 1.0)# 频率归一化days_ago = (datetime.now() - datetime.fromisoformat(mem["last_accessed"])).daystime_decay = math.exp(-0.693 * days_ago / self.decay_half_life)return frequency_boost * time_decaydef cleanup(self, threshold: float = 0.1):"""清理低相关性记忆"""to_delete = [k for k, v in self.memories.items()if self.get_relevance_score(k) < threshold]for k in to_delete:del self.memories[k]

方案二：重要度标注

在存储时就标注记忆的重要程度：

def store_memory_with_importance(content: str,importance: str,# "high", "medium", "low"half_life: dict = {"high": 180, "medium": 60, "low": 14}):decay_days = half_life.get(importance, 60)memory_entry = {"content": content,"importance": importance,"decay_half_life": decay_days,"created": datetime.now().isoformat()}# 写入向量库，检索时用importance作为权重

记忆更新策略

当信息发生变化时，旧记忆需要被更新而非简单覆盖：

def update_memory(old_key: str, new_content: str, collection):"""更新记忆：保留旧版本作为历史，写入新版本"""old_memory = collection.get(ids=[old_key])if old_memory and old_memory["documents"]:# 将旧版本存入历史集合history_key = f"{old_key}_v{datetime.now().strftime('%Y%m%d%H%M%S')}"collection.add(documents=old_memory["documents"],ids=[history_key],metadatas=[{**old_memory["metadatas"][0], "status": "superseded"}])# 更新当前版本collection.update(documents=[new_content],ids=[old_key],metadatas=[{"status": "current", "updated": datetime.now().isoformat()}])

本章小结