超越 RAG： 构建具备“自我进化”能力的 Agentic Memory 系统

前言：检索增强生成（RAG）让 AI 拥有了“图书馆”，可以查阅资料；而 Agentic Memory（代理记忆） 则赋予了 AI “海马体”，让它拥有了长期记忆、行为习惯和自我进化的能力。

本文基于 Alex Spyropoulous 的文章以及当前 Python 生态的最佳实践，带你从零构建一个不仅能“回答问题”，更能“记住你”的智能 Agent。

第一部分：核心概念与架构设计

1. 什么是 Agentic Memory？它与 RAG 有何不同？

传统的 RAG (Retrieval-Augmented Generation) 主要是只读的：用户提问 -> 系统去数据库检索文档 -> 生成答案。系统本身并不随对话发生改变。

Agentic Memory 则是读写双向且具备进化能力的系统。它不仅检索信息，还在交互中动态生成新的记忆、提炼用户偏好规则，并定期对记忆进行整理和遗忘。

2. 核心架构：三个“时空平面”

为了在保证实时响应速度的同时维护庞大的记忆库，生产级的架构通常将任务划分为三个执行平面。

1. 实时平面 (Real-Time Plane)：

   • 目标：极速响应 (P95 < 4秒)。

   • 关键：利用语义缓存减少检索次数，利用双路检索同时获取知识与记忆。

2. 每日平面 (Daily Plane)：

   • 目标：知识整合与习惯养成。

   • 任务：将今天的对话与历史记忆建立语义关联，提炼用户偏好（如“用户喜欢 Python 代码附带类型注解”）。

3. 每周平面 (Weekly Plane)：

   • 目标：成本控制与存储健康。

   • 任务：记忆压缩（Summarization）与剪枝（Pruning），防止向量库无限膨胀。

第二部分：Python 实战落地

在 Python 中实现上述架构，目前主要有两种路径：

1. 使用专用库（如 Mem0）：开箱即用，适合快速落地。

2. 基于 LangChain 自建：适合深度定制复杂的检索与维护逻辑。

方案一：开箱即用的专用库 —— Mem0

Mem0 是目前最接近文章描述的“分层记忆”和“自动更新”的开源库。它自动处理了向量化、存储、检索，甚至支持“用户层级”的记忆隔离。

安装：

pip install mem0ai

代码示例：

from mem0 import Memory

# 1. 初始化 Memory (支持 Qdrant, Redis, Postgres 等)
config = {
    "vector_store": {
        "provider": "qdrant",
        "config": {"path": "./memory_db"} # 确保数据持久化保存
    },
    "llm": {
        "provider": "openai",
        "config": {"model": "gpt-4o"}
    }
}
m = Memory.from_config(config)

user_id = "developer_alex"

# 场景：用户第一次提问
query1 = "我的 React Native 列表滚动很卡，有 1000 多条数据。"
# 在实际业务中，add() 通常在 LLM 生成回复后异步调用
m.add(query1, user_id=user_id, metadata={"category": "tech_issue"})

# ... 假设过了几天 ...

# 场景：用户第二次提问（省略了上下文）
query2 = "上次那个列表优化的方案，推荐个代码库？"

# 2. 检索相关记忆
memories = m.search(query2, user_id=user_id)
print(f"检索到的上下文: {memories}")
# 输出会自动关联到 "React Native", "列表卡顿", "1000+数据" 等信息

方案二：基于 LangChain 的双路检索架构

如果你需要构建复杂的企业级 Agent，通常需要同时查阅“企业文档 (RAG)”和“用户记忆”。我们可以使用 LangChain 构建一个双路检索系统。

1. 架构逻辑

• 路 A (RAG)：检索静态文档（Wiki、手册）。

• 路 B (Memory)：检索动态记忆（用户偏好、历史故障）。

• 写回 (Write Back)：每一轮对话结束后，提取关键信息写回 Memory 库。

2. 代码实现

from operator import itemgetter
from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma

# --- 1. 准备两个向量库 ---

# A. 知识库 (RAG): 存放固定的文档
# persist_directory 确保数据落盘，不会随程序重启丢失
rag_vectorstore = Chroma(
    collection_name="knowledge_base", 
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./rag_db" 
)
rag_retriever = rag_vectorstore.as_retriever()

# B. 记忆库 (Agentic Memory): 存放用户画像和历史
# 注意：这是一个读写库，必须开启持久化
memory_vectorstore = Chroma(
    collection_name="user_memory", 
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./memory_db"
)

# --- 2. 定义双路 Prompt ---

template = """
你是一个智能助手。请基于以下信息回答用户问题。

【外部知识 (RAG)】:
{rag_context}

【用户记忆 (Memory)】:
{memory_context}

用户问题: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI(model="gpt-4o")

# --- 3. 构建并行检索链 ---

def format_docs(docs):
    return "\n\n".join([d.page_content for d in docs])

# 自定义检索函数：实现基于 user_id 的过滤
# 这一点非常关键，避免检索到其他用户的记忆！
def retrieve_user_memory(inputs):
    user_id = inputs["user_id"]
    question = inputs["question"]
    # 使用 vectorstore 的 search 方法，并传入 filter 参数
    # filter 语法取决于具体的向量数据库 (Chroma/Pinecone/Qdrant)
    docs = memory_vectorstore.similarity_search(
        question,
        filter={"user_id": user_id} 
    )
    return format_docs(docs)

# 并行执行两路检索
# 输入结构变为字典: {"question": "...", "user_id": "..."}
retrieval_chain = RunnableParallel(
    # RAG 不需要 user_id，只取 question 进行检索
    rag_context=itemgetter("question") | rag_retriever | format_docs,
    # Memory 使用我们自定义的带 Filter 的函数
    memory_context=retrieve_user_memory,
    question=itemgetter("question")
)

# 完整的生成链
chain = retrieval_chain | prompt | model | StrOutputParser()

# --- 4. 运行与“写回” (Agentic 的关键) ---

def chat_loop(user_input, user_id="u1"):
    # Step 1: 生成回答
    # 注意：invoke 时传入的是一个字典，包含 question 和 user_id
    response = chain.invoke({"question": user_input, "user_id": user_id})
    print(f"AI: {response}")
    
    # Step 2: 【Agentic 核心】提取并写回记忆
    # 这一步让 Agent 记住了这次交互的关键点
    # 在生产中，强烈建议使用 LLM 总结出结构化的 Fact 再存入，而非存储原始对话
    # 例如：llm.invoke("Summarize this interaction into a memory fact...")
    memory_vectorstore.add_texts(
        [f"User Query: {user_input}\nSummary: React Native Performance Issue"], 
        metadatas=[{"user_id": user_id, "timestamp": "2025-11-28"}]
    )

# chat_loop("FlatList 性能很差怎么办？", user_id="user_123")

第三部分：维护平面的实现 (Daily/Weekly Jobs)

“记忆”如果不整理，很快就会变成垃圾场。这就是架构中 Daily/Weekly Plane 的作用。这部分逻辑通常不在主对话链路中，而是通过后台定时任务（Cron Jobs）实现。

Python 实现“每周记忆压缩”

我们可以使用 schedule 库配合 LLM 来实现记忆的定期压缩。

import schedule
import time
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini") # 使用更便宜的模型做整理工作

def compress_weekly_memories():
    print("开始执行每周记忆压缩任务...")
    
    # 1. 模拟：从向量库获取 7 天前的零散记忆
    # 在实际代码中，需要使用 memory_vectorstore.get(where={...})
    # old_memories = vector_store.get(filter={"timestamp": {"$lt": seven_days_ago}})
    old_memories_text = """
    - 2025-11-01: 用户询问 React Native 列表卡顿
    - 2025-11-01: 建议使用 VirtualizedList
    - 2025-11-03: 用户询问 VirtualizedList 的参数配置
    - 2025-11-03: 提供了 getItemLayout 示例代码
    """
    
    # 2. 使用 LLM 进行语义压缩 (Summarization)
    summary_prompt = f"""
    将以下零散的对话记忆压缩为一条简练的长期记忆，保留关键技术细节，去除寒暄：
    {old_memories_text}
    """
    compressed_memory = llm.invoke(summary_prompt).content
    
    # 3. 更新向量库：删除旧的零散条目，插入新的合并条目
    print(f"压缩后的长期记忆: {compressed_memory}")
    # 输出示例: "用户在 React Native 项目中遇到长列表卡顿，通过实施 VirtualizedList 和 getItemLayout 优化解决了问题。"
    
    # 伪代码：
    # vector_store.delete(old_ids)
    # vector_store.add(compressed_memory, type="long_term_consolidated")

# 定义定时任务：每周一凌晨 2 点执行
schedule.every().monday.at("02:00").do(compress_weekly_memories)

# 实际运行循环
# while True:
#     schedule.run_pending()
#     time.sleep(1)

第四部分：生产环境优化策略

除了代码实现，要让 Agent 真正好用，还需要引入以下优化策略：

1. 预测性记忆 (Predictive Memory)

   • 原理：在用户发送完整查询前，或者对于高频问题，利用 Redis 缓存层直接返回结果。

   • 收益：可将常见问题的响应延迟从 200ms 降至 20ms。

2. 加权插入 (Weighted Insertion)

   • 原理：新近的、带有用户正向反馈（点赞/采纳）的记忆权重更高。

   • 公式：检索分数 = 向量相似度 * 时间衰减系数 * 重要性权重。

3. 记忆分层存储

   • 0-7天：存储在高性能向量库（如 Pinecone/Qdrant），FP32 精度。

   • 7-30天：FP16 精度，节省空间。

   • 30+天：压缩为摘要文本，存入低成本存储或归档。

总结

构建生产级的 Agentic Memory 系统，核心在于闭环。

1. 实时层：利用 LangChain/Mem0 实现“读写双向”的交互。

2. 维护层：利用 Python Scripts 实现记忆的“压缩与遗忘”。

3. 优化层：利用 缓存与加权 提升系统的响应速度和准确度。

通过这套架构，你的 AI Agent 将不再是一个用完即走的工具，而是一个越用越懂你的智能伙伴。

参考文献

https://alexspyropoulos.com/posts/demystifying-agentic-memory/