AI Agent 记忆管理实战：让你的智能体真正记住用户上下文

为什么 AI Agent 需要记忆管理？

大语言模型本质上是无状态的——每次调用都是全新的开始，它不记得上一轮对话说了什么，也不知道用户的偏好和历史操作。这对于简单的问答场景还好，但当我们构建真正的 AI Agent 时，这个问题就变得致命了。

想象一下：你告诉 Agent "我是一名 Python 开发者，不喜欢冗长的说明，直接给代码"，下一轮对话它又开始啰嗦地解释基础概念——这种体验会让用户彻底失去耐心。

记忆管理解决的核心问题包括：

• 上下文连续性：跨会话记住用户信息和偏好

• 知识积累：Agent 能从历史交互中学习

• 个性化响应：根据用户画像调整回复风格

• 任务追踪：记住长期任务的进度和状态

记忆的三种类型：短期、长期与情节记忆

借鉴人类认知科学的分类，AI Agent 的记忆通常分为三种类型，每种类型对应不同的存储策略和使用场景。

短期记忆（Working Memory）

短期记忆对应当前会话的上下文窗口。最简单的实现就是把对话历史全部塞进 prompt，但这会很快撞上 token 限制。

from langchain.memory import ConversationBufferWindowMemory

# 只保留最近 10 轮对话
memory = ConversationBufferWindowMemory(
    k=10,
    return_messages=True,
    memory_key="chat_history"
)

# 或者使用摘要记忆，自动压缩历史
from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
summary_memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # 超过则自动摘要
    return_messages=True
)

ConversationSummaryBufferMemory 是处理长对话的利器：它保留最近的原始消息，同时把更早的历史压缩成摘要，兼顾了细节保留和 token 节省。

长期记忆（Long-term Memory）

长期记忆需要持久化存储，跨会话保留。常见方案是使用向量数据库存储记忆片段，查询时通过语义相似度检索相关记忆。

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.memory import VectorStoreRetrieverMemory

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    collection_name="agent_memory",
    embedding_function=embeddings,
    persist_directory="./memory_db"
)

retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
vector_memory = VectorStoreRetrieverMemory(
    retriever=retriever,
    memory_key="relevant_history"
)

# 保存记忆
vector_memory.save_context(
    {"input": "我在做一个电商推荐系统"},
    {"output": "好的，我来帮你设计推荐算法"}
)

# 查询相关记忆
relevant = vector_memory.load_memory_variables(
    {"prompt": "推荐系统进展如何了？"}
)
print(relevant["relevant_history"])

情节记忆（Episodic Memory）

情节记忆记录具体的事件和经历，比用户的操作日志、重要决策节点等。可以用结构化数据库（如 PostgreSQL）存储，支持按时间、类型等维度检索。

实战：用 Redis + LangChain 实现跨会话记忆

下面我们来构建一个完整的跨会话记忆方案，使用 Redis 存储会话历史，支持多用户隔离。

环境准备

pip install langchain langchain-openai langchain-community redis
docker run -d -p 6379:6379 redis:7-alpine

实现代码

import redis
import json
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory
from langchain.schema import HumanMessage, AIMessage

class PersistentAgentMemory:
    """支持跨会话持久化的 Agent 记忆管理器"""
    
    def __init__(self, user_id: str, redis_url: str = "redis://localhost:6379"):
        self.user_id = user_id
        self.redis_client = redis.from_url(redis_url)
        self.session_key = f"agent:memory:{user_id}"
        self.profile_key = f"agent:profile:{user_id}"
        
    def load_history(self, limit: int = 20) -> list:
        """加载最近的对话历史"""
        messages = self.redis_client.lrange(self.session_key, -limit*2, -1)
        history = []
        for msg in messages:
            data = json.loads(msg)
            if data["role"] == "human":
                history.append(HumanMessage(content=data["content"]))
            else:
                history.append(AIMessage(content=data["content"]))
        return history
    
    def save_message(self, role: str, content: str):
        """保存单条消息"""
        message = {
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        }
        self.redis_client.rpush(self.session_key, json.dumps(message, ensure_ascii=False))
        # 只保留最近 200 条
        self.redis_client.ltrim(self.session_key, -200, -1)
        
    def update_user_profile(self, key: str, value: str):
        """更新用户画像"""
        self.redis_client.hset(self.profile_key, key, value)
        
    def get_user_profile(self) -> dict:
        """获取用户画像"""
        profile = self.redis_client.hgetall(self.profile_key)
        return {k.decode(): v.decode() for k, v in profile.items()}


def create_agent_with_memory(user_id: str):
    """创建带记忆的 Agent"""
    memory_manager = PersistentAgentMemory(user_id)
    llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
    
    def chat(user_input: str) -> str:
        # 加载历史
        history = memory_manager.load_history(limit=10)
        profile = memory_manager.get_user_profile()
        
        # 构建系统提示
        profile_str = "\n".join([f"- {k}: {v}" for k, v in profile.items()])
        system_prompt = f"""你是一个有记忆的 AI 助手。
        
用户画像：
{profile_str if profile_str else "暂无"}

请根据对话历史和用户画像提供个性化的回复。"""
        
        # 组装消息
        messages = [{"role": "system", "content": system_prompt}]
        for msg in history:
            if isinstance(msg, HumanMessage):
                messages.append({"role": "user", "content": msg.content})
            else:
                messages.append({"role": "assistant", "content": msg.content})
        messages.append({"role": "user", "content": user_input})
        
        # 调用 LLM
        response = llm.invoke(messages)
        answer = response.content
        
        # 持久化
        memory_manager.save_message("human", user_input)
        memory_manager.save_message("assistant", answer)
        
        return answer
    
    return chat


# 使用示例
agent = create_agent_with_memory("user_001")
print(agent("你好，我叫小明，是一个 Python 后端开发者"))
print(agent("给我推荐一个适合我的异步框架"))  # 会记住"Python后端开发者"

进阶：用向量检索实现语义记忆检索

对于需要从大量历史记忆中精准召回的场景，向量检索是必不可少的。以下是一个完整的语义记忆系统：

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.schema import Document
import numpy as np

class SemanticMemoryStore:
    """基于向量检索的语义记忆存储"""
    
    def __init__(self, user_id: str):
        self.user_id = user_id
        self.embeddings = OpenAIEmbeddings()
        self.index_path = f"./vector_store/{user_id}"
        
        try:
            self.vectorstore = FAISS.load_local(
                self.index_path, 
                self.embeddings,
                allow_dangerous_deserialization=True
            )
            print(f"加载已有记忆索引，用户：{user_id}")
        except:
            # 首次使用，创建空索引
            self.vectorstore = None
            
    def add_memory(self, content: str, metadata: dict = None):
        """添加新记忆"""
        doc = Document(
            page_content=content,
            metadata=metadata or {"timestamp": datetime.now().isoformat()}
        )
        if self.vectorstore is None:
            self.vectorstore = FAISS.from_documents([doc], self.embeddings)
        else:
            self.vectorstore.add_documents([doc])
        self.vectorstore.save_local(self.index_path)
        
    def search_memory(self, query: str, top_k: int = 5) -> list[str]:
        """语义检索相关记忆"""
        if self.vectorstore is None:
            return []
        results = self.vectorstore.similarity_search_with_score(query, k=top_k)
        # 过滤相似度太低的结果（FAISS使用L2距离，越小越相似）
        return [doc.page_content for doc, score in results if score < 1.5]


# 使用示例
store = SemanticMemoryStore("user_001")
store.add_memory("用户喜欢简洁的代码风格，不喜欢过度封装")
store.add_memory("用户正在开发一个基于 FastAPI 的微服务项目")
store.add_memory("用户使用 PostgreSQL 作为主数据库")

# 语义检索
memories = store.search_memory("数据库选型建议")
print("相关记忆：", memories)
# 输出：["用户使用 PostgreSQL 作为主数据库", ...]

记忆压缩与遗忘机制

无限积累的记忆会带来存储和检索性能问题，同时也可能引入过时信息的干扰。借鉴人类记忆的遗忘曲线，我们需要为 Agent 设计记忆的老化和压缩机制。

• 时间衰减：按时间戳给记忆打分，越久远的记忆权重越低

• 访问频率加权：被检索到的次数越多，说明越重要，应该保留

• 定期摘要压缩：用 LLM 把多条细粒度记忆合并成高层次摘要

• 矛盾检测：新记忆与旧记忆冲突时，更新而非简单叠加

async def compress_old_memories(user_id: str, days_threshold: int = 30):
    """压缩 30 天前的旧记忆"""
    store = SemanticMemoryStore(user_id)
    llm = ChatOpenAI(model="gpt-4o-mini")
    
    # 获取旧记忆（实际需要按时间过滤）
    old_memories = store.search_memory("", top_k=50)  # 获取全部
    
    if len(old_memories) < 10:
        return  # 记忆太少不需要压缩
    
    # 用 LLM 提炼关键信息
    prompt = f"""以下是用户的历史记忆片段，请提炼出最关键的用户特征和偏好，输出5条精华摘要：

{chr(10).join(old_memories[:20])}

请输出5条精华，每条不超过50字："""
    
    result = await llm.ainvoke(prompt)
    summaries = result.content.strip().split("\n")
    
    # 删除旧记忆，保存压缩后的摘要
    for summary in summaries:
        store.add_memory(
            summary.strip(), 
            metadata={"type": "compressed_summary", "timestamp": datetime.now().isoformat()}
        )
    
    print(f"记忆压缩完成，{len(old_memories)} 条 → {len(summaries)} 条")

部署建议与踩坑总结

在生产环境中落地 AI Agent 记忆系统，有几个关键问题需要特别注意：

• 隐私合规：记忆中可能含有用户的敏感信息，必须加密存储，并支持用户的"遗忘权"（一键清除所有记忆）

• 记忆污染：如果 Agent 记住了错误信息（用户说错了、或数据异常），需要提供纠错机制，而不是把错误越记越深

• 多端同步：用户可能在 PC、手机多端使用，记忆需要实时同步，用 Redis Pub/Sub 或消息队列解决

• 冷启动问题：新用户没有任何历史记忆，要设计合理的引导对话，快速建立初始画像

• Embedding 费用控制：向量化每条记忆都需要调用 Embedding API，建议批量处理或使用本地模型（如 text2vec-chinese）降低成本

记忆管理是 AI Agent 从"好用"到"好用且懂我"的关键一跃。随着 MemGPT、Mem0 等专门面向 Agent 记忆的框架日趋成熟，这个领域还有很大的探索空间。欢迎在评论区分享你的实践经验！