Agent 长期持久化记忆搭建实操 · devcfg

Agent 长期持久化记忆搭建实操 - 内部知识库文档

一、引言

在Agent实际落地中，无状态设计是导致多轮交互质量下降的首要原因。用户重复提问、Agent遗忘先前指令、跨会话无法调用历史经验——这些问题本质上是记忆系统缺失所致。本文围绕记忆分层理论，以LangGraph为代码主线，讲解短期记忆与长期记忆的完整搭建流程，并引入Postgres数据库作为结构化持久化方案。

读者完成后将掌握：记忆系统的架构分层设计、短期记忆的上下文窗口管理与状态维护、长期记忆的向量化存储与语义检索，以及记忆数据的生命周期管理策略。

二、记忆分层模型：从短期到长期的架构设计

2.1 三层次记忆模型

Agent记忆系统的设计参考了认知科学中的记忆分层理论。实践中我们一般划分为三个层次：

短期记忆（Short-term Memory） 负责存储当前会话中的原始输入输出，通常是LLM上下文窗口内的历史消息。其特点是读写频繁、容量有限（受模型token上限约束），主要用于维持单次对话的连贯性。

工作记忆（Working Memory） 跟踪Agent正在执行的任务状态。例如用户要求“先查天气，再订酒店”，工作记忆会维护一个任务栈，记录当前执行到第几步、哪些子任务已完成。当Agent需要临时切换到子任务（如查天气时再查核酸政策）时，工作记忆保证能够恢复主任务。

长期记忆（Long-term Memory） 存储跨会话的重要知识，包括用户偏好、业务规则、历史对话关键摘要等。长期记忆的载体通常是外部存储系统（向量数据库、关系型数据库），需要支持持久化、高效检索和过期清理。

三个层次在Agent运行时的协作关系如下：Agent收到用户输入后，首先从长期记忆中检索相关上下文，与短期记忆拼接后送入LLM，同时工作记忆负责管理当前任务栈的执行流转。任何有价值的交互结果，经过重要性评估后回写入长期记忆。

2.2 长期记忆的企业级必要性

在企业Agent场景下，长期记忆不仅是功能增强，更是系统可用性的前提。以下为典型的跨场景需求：

• 用户偏好学习：用户A偏好“详细的技术文档”，用户B偏好“简洁的摘要说明”，长期记忆需记录这些偏好并在后续交互中自动应用。
• 知识复用：某部门在上月的客户沟通中已确认过特定的技术方案，下次交流时Agent应能够参考历史纪要，避免重复讨论。
• 上下文推理：用户在周一咨询“登录失败问题”，周二再问“那还需要更新证书吗”——Agent需从长期记忆中关联到前日的登录失败场景，才能给出正确建议。

没有长期记忆的Agent等同于每次对话都重新初始化的空壳模型，无法沉淀经验，也无法建立用户信任。

三、短期记忆实现：上下文窗口管理与状态维护

3.1 滑动窗口截断策略

短期记忆最基础的管理策略是滑动窗口——只保留最近N轮对话。假设业务要求Agent记住最近5轮用户输入和助手回复，则窗口大小为10条消息（5入5出）。超出范围的消息直接丢弃。

以下为基于LangGraph的BufferWindowMemory实现示例：

from typing import List, Dict

class BufferWindowMemory:
    """滑动窗口式短期记忆，保留最近 N 轮对话"""
    def __init__(self, max_turns: int = 5):
        self.messages: List[Dict] = []
        self.max_turns = max_turns
    
    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        # 修剪窗口：超出 max_turns * 2（用户+助手）则丢弃最早
        if len(self.messages) > self.max_turns * 2:
            self.messages = self.messages[-(self.max_turns * 2):]
    
    def get_context(self) -> List[Dict]:
        return self.messages

注意：该方法简单高效，但面临两个局限：（1）如果某轮对话非常长（如用户粘贴了整份日志），窗口可能迅速被单一消息占满；（2）模型无法访问被丢弃的早期上下文，可能导致“遗忘”问题。

3.2 摘要压缩策略

为解决滑动窗口的“一刀切”问题，可以引入ConversationSummaryMemory——定期对早期对话进行摘要压缩，取代原始消息放入窗口。

from langchain.memory import ConversationSummaryMemory
from langchain.llms import OpenAI

summary_memory = ConversationSummaryMemory(
    llm=OpenAI(model="gpt-4"),
    max_token_limit=500  # 摘要不超过500token
)

# 每轮对话后，自动生成/更新摘要
summary_memory.save_context(
    {"input": "北京的天气如何？"}, 
    {"output": "今天是晴天，气温20°C。"}
)
# 检索时返回：摘要（早期）+ 原始消息（近期）
context = summary_memory.buffer

该方案在长对话中表现更稳定，但增加了LLM调用开销（每次保存都需要摘要）。实践中建议：仅在窗口触发截断时执行一次摘要，而非每轮都做。

3.3 状态管理的边界

短期记忆虽然能维持对话连贯性，但无法解决跨会话的“遗忘”问题。例如用户第二天回来问“我昨天说那个方案，你觉得怎么样？”，短期记忆已经清空，必须依赖长期记忆来提供前日讨论内容。这就是短期记忆的职责边界——仅限当前会话内，不应把跨会话知识放在短期记忆里管理。

四、长期记忆存储方案：向量数据库 + Postgres 融合设计

4.1 双载体存储架构

长期记忆需要同时支持两种检索模式：语义检索（模糊搜索“关于数据库优化的经验”）和结构化检索（精确查询“用户A的配置偏好”）。单一存储方案往往难以两全。实践中采用向量数据库（FAISS/Chroma）与关系型数据库（Postgres）融合的方案：

存储载体	适用场景	优势	劣势
向量数据库	语义相似度搜索	检索速度快，支持embedding	结构化查询弱
Postgres	精确查询、事务管理、报表	强事务，ACID	全文搜索不如向量检索

4.2 双写模型设计

当Agent抽取到一条待存储的记忆时，同时写入两个系统：

1. 将记忆内容转换为向量（通过Embedding模型如text-embedding-ada-002），存入向量数据库的索引。
2. 将记忆的原始文本、实体标签、时间戳、重要性评分等结构化字段写入Postgres表。

检索时，先通过向量数据库获取Top K相似记忆列表，再通过Postgres精准回查这些记忆的完整字段。这样可以兼顾检索速度与数据完整性。

Postgres核心表结构示例：

CREATE TABLE agent_memories (
    id SERIAL PRIMARY KEY,
    user_id VARCHAR(64) NOT NULL,         -- 用户/会话标识
    memory_key VARCHAR(256) NOT NULL,      -- 记忆实体（如“用户偏好_办公时间”）
    memory_value TEXT NOT NULL,            -- 记忆内容
    importance FLOAT DEFAULT 0.5,          -- 重要性评分 [0,1]
    vector_id VARCHAR(64),                 -- 对应向量数据库中的索引ID
    created_at TIMESTAMP DEFAULT NOW(),
    expires_at TIMESTAMP,                  -- TTL过期时间
    is_active BOOLEAN DEFAULT TRUE
);

CREATE INDEX idx_memories_user ON agent_memories(user_id);
CREATE INDEX idx_memories_key ON agent_memories(memory_key);
CREATE INDEX idx_memories_expires ON agent_memories(expires_at) WHERE is_active = TRUE;

工程提示：写入Postgres时应使用连接池（如psycopg2的ThreadedConnectionPool），避免每次存储都新建连接的开销。向量数据库写入建议使用异步API，不阻塞主流程。

五、记忆生命周期管理：重要性评估与自动清理

5.1 为什么需要生命周期管理

长期记忆不加节制地增长会导致两个问题：存储膨胀（增加成本）和检索噪声（大量无关记忆干扰LLM推理）。必须对记忆进行分级与过期清理。

5.2 基于LLM的重要性评估

使用LLM评估用户输入中的信息是否值得存入长期记忆。评估Prompt示例如下：

你是一个记忆评分系统。判断以下用户消息中是否含有“值得长期记住”的信息。
评估标准（0-1）：
- 0.0~0.3：闲聊、问候、临时指令，无需记录
- 0.4~0.6：临时偏好或一次性话术，无需长期保存
- 0.7~0.9：明确的个人偏好、业务规则、项目信息，值得存储
- 1.0：关键决策、配置修改、安全策略，必须永久保留

用户消息: {user_input}

仅返回一个0-1之间的浮点数。

根据评分设定TTL（Time To Live）：

def calculate_ttl(importance: float) -> int:
    if importance >= 0.9:
        return -1  # 永久保留
    elif importance >= 0.7:
        return 90 * 24 * 3600  # 90天
    elif importance >= 0.4:
        return 30 * 24 * 3600  # 30天
    else:
        return 7 * 24 * 3600   # 1周

注意：不要对所有消息都评估，只针对明显带有陈述性信息的内容（如“我住在北京”），避免不必要的LLM调用。

5.3 定时清理脚本

每天凌晨定时执行清理任务：

import psycopg2
from datetime import datetime, timedelta

def clean_expired_memories():
    conn = psycopg2.connect("dbname=agent_memory")
    cur = conn.cursor()
    # 硬删除：清理过期且重要性不高的记忆
    cur.execute("""
        DELETE FROM agent_memories
        WHERE expires_at IS NOT NULL
          AND expires_at < NOW()
          AND importance < 0.7
    """)
    # 软标记：对于重要性高的记忆，延长TTL而非直接删除
    cur.execute("""
        UPDATE agent_memories
        SET expires_at = NOW() + INTERVAL '90 days'
        WHERE expires_at < NOW()
          AND importance >= 0.7
          AND is_active = TRUE
    """)
    conn.commit()
    cur.close()
    conn.close()

策略思想：高价值记忆只延长不删除，低价值记忆到点即淘汰。同时保留一条审计日志记录删除操作，以便问题回溯。

六、给Agent添加长期记忆：LangGraph中的Memory模块实操

6.1 在Graph中嵌入Memory节点

LangGraph允许将记忆的写入和检索封装为独立的图节点，在Agent执行流转中自动触发。以下为典型设计：

import langgraph as lg
from langgraph.executor import AgentExecutor
from langgraph.prompt import SystemMessage
from typing import Dict, Any

# 1. 定义提取记忆节点
def extract_memory(state: Dict[str, Any]) -> Dict[str, Any]:
    """从本轮用户输入中识别可存储信息并写入Postgres"""
    user_input = state["current_input"]
    importance = llm_evaluate_importance(user_input)
    if importance >= 0.4:
        # 提取实体与偏好（可用NER或简单规则）
        memory_key = extract_entity(user_input)  # 如 "user_location"
        memory_value = user_input
        write_to_postgres(state["user_id"], memory_key, memory_value, importance)
        state["memory_extracted"] = True
    return state

# 2. 定义检索记忆节点
def retrieve_memory(state: Dict[str, Any]) -> Dict[str, Any]:
    """在图执行前回查相关记忆，注入System Prompt"""
    user_input = state["current_input"]
    # 向量检索
    similar_memories = vector_store.search(user_input, top_k=3)
    # 结构化检索
    exact_memories = query_postgres(state["user_id"], keys=["user_preference"])
    # 合并记忆
    merged = format_memories(similar_memories + exact_memories)
    # 注入到System Prompt
    state["system_prompt_suffix"] = f"以下为基于用户历史的记忆：{merged}"
    return state

# 3. 构建LangGraph
graph = lg.Graph()
graph.add_node("retrieve_memory", retrieve_memory)
graph.add_node("agent_decision", agent_llm_call)
graph.add_node("execution_loop", tool_executor)
graph.add_node("extract_memory", extract_memory)
graph.add_edge("retrieve_memory", "agent_decision")
graph.add_edge("agent_decision", "execution_loop")
graph.add_edge("execution_loop", "extract_memory")
graph.set_entry_node("retrieve_memory")
graph.set_finish_node("extract_memory")
executor = AgentExecutor(graph)

6.2 关键工程细节

• 检索时机：retrieve_memory节点放在图执行的最前面，确保LLM在决策前已获取历史上下文。写入节点extract_memory放在图执行结束后，避免干扰当前轮逻辑。

• 去重策略：同一用户对同一实体多次产生记忆时，应更新而非追加。Postgres中的ON CONFLICT语句可用于实现upsert。

• 错误处理：如果Postgres或向量数据库写入失败，不应导致整个Agent执行失败。在extract_memory节点内捕获异常，使用graph.set_error_handler定义降级策略（如仅写入日志，继续执行）。

七、检索增强技巧：多模态检索与记忆融合策略

7.1 混合记忆输入

LLM的输入包含两部分记忆：当前会话的短期记忆（原始消息）和长期记忆检索结果。如何融合直接影响推理质量。推荐策略：

最终输入 = 短期记忆（原始消息，保留最近N轮）
         + 长期记忆（Top K相关，按重要性排序）
         + 工作记忆（当前任务栈状态）

在LangChain中可通过自定义Memory类实现：

from langchain.memory import CombinedMemory
from langchain.schema import BaseMemory

class HybridMemory(BaseMemory):
    short_memory: BufferWindowMemory
    long_retriever: Any  # 向量检索器
    
    @property
    def memory_variables(self) -> List[str]:
        return ["short_context", "long_context"]
    
    def load_memory_variables(self, inputs: Dict) -> Dict:
        short_context = self.short_memory.get_context()
        query = inputs.get("input", "")
        long_context = self.long_retriever.get_relevant_documents(query)
        return {
            "short_context": short_context,
            "long_context": [d.page_content for d in long_context]
        }

7.2 排序融合与重排序

当同时使用语义检索和关键词检索时，需要对结果进行融合排序。RRF（Reciprocal Rank Fusion）是轻量的融合方法：

def rrf_fusion(results: List[List[Any]], k: int = 60) -> List[Any]:
    scores = {}
    for rank, doc in enumerate(results):
        scores[doc.id] = scores.get(doc.id, 0) + 1 / (rank + k)
    return sorted(scores.keys(), key=lambda x: scores[x], reverse=True)

重排序：在融合后，使用一个轻量级模型（如cross-encoder）对Top 20结果重新打分排序，进一步压缩噪声。该步骤在延时敏感场景下可省略，在离线批量场景下推荐使用。

7.3 时间衰减因子

记忆的时效性也很重要。1年前的“用户偏好”与1天前的“用户偏好”可信度不同。在检索阶段加入时间衰减：

def time_decay_score(original_score: float, age_hours: float) -> float:
    decay_factor = 0.98  # 每小时衰减2%
    return original_score * (decay_factor ** age_hours)

对于业务规则类记忆，衰减因子通常设为1（不衰减）；对于临时偏好的记忆，衰减可以更快。

八、踩坑记录与性能优化实践

8.1 常见问题清单

1. 记忆写入/读取瓶颈

现象：Agent每次执行都要先检索长期记忆，在高并发下Postgres查询延迟从1ms飙升至200ms，整条链路耗时翻倍。

原因：未使用连接池、未对记忆表做分区索引。

优化方案：

• 强制使用连接池（如psycopg2.pool.ThreadedConnectionPool），最小连接数=并发数×2。
• 按user_id做哈希分区，每个分区独立索引。

2. 向量维度过高导致检索延迟

现象：使用1536维的embedding模型，在100万条记录上检索耗时超过300ms，无法满足实时对话要求。

优化方案：

• 降维：将embedding从1536维压缩至256维（使用PCA或自编码器），牺牲少量精度换取5倍以上速度提升。
• 索引分片：使用IVF（Inverted File）索引，设置nprobe=20（检索时仅搜索20个最相关的聚类中心）。

3. Postgres连接池过小

现象：Agent实例从5扩到20后，Postgres连接数仍然只有10，大量请求排队等待。

优化方案：动态扩展连接池，与Agent实例数成正比。上限不宜超过数据库可接受的连接数（通常500以内）。

8.2 性能优化实测数据

以下为内部压测数据（8核32G机器，10万条记忆，并发30）：

优化措施	平均响应耗时	内存占用	说明
无优化（直接查全表）	1200ms	600MB	不可接受
加索引+连接池	230ms	650MB	可接受
异步写入+缓存层	85ms	700MB	推荐方案
加缓存层+向量索引分片	45ms	800MB	高吞吐场景

缓存层使用Redis，存储最近1000条高频记忆，避免对Postgres的重复查询。写入时先写Redis（1ms），异步批量写Postgres（每5秒或积压20条写一次）。

九、总结与拓展方向

9.1 核心技术点回顾

本文完整覆盖了Agent长期持久化记忆的搭建流程，要点如下：

1. 分层记忆架构：短期记忆（对话上下文）、工作记忆（任务状态栈）、长期记忆（知识库），三层各有职责与生命周期。

2. 短期记忆窗口管理：滑动窗口截断与摘要压缩两种策略，根据业务场景选择或组合。

3. 长期记忆多模态存储：向量数据库负责语义检索，Postgres负责结构化事务存储，双写保证数据完整性与检索效率。

4. 生命周期管理：LLM评估重要性→设定TTL→定时清理策略，避免存储膨胀与检索噪声。

5. 检索增强融合：短期+长期记忆混合输入，RRF排序融合，时间衰减因子提升相关度。

6. 性能优化：异步写入、连接池、缓存层、索引分片，确保高并发下的响应稳定性。

9.2 落地建议

• 优先实现短期记忆：大多数Agent应用在短期内（单会话内）表现更好，短期记忆是基础，两周内可完成集成。

• 长期记忆逐步叠加：先从用户偏好、业务规则等结构化数据开始，再扩展至自由文本的记忆抽取。不要一次性上线所有功能。

• 监控记忆命中率：在日志中记录每次检索是否返回有效记忆，命中率低于30%说明检索策略或记忆填充策略需要优化。

• 预留扩展接口：后续可能接入多模态记忆（图像/音频），在存储设计时字段类型建议使用JSONB（Postgres）等可扩展格式。

9.3 拓展方向

• 多模态记忆：Agent不仅能记住文本，还能记住对话中的图片（如产品示意图）、音频（如客户语气），这对工业售后、设计审核等场景有实际价值。
• 基于记忆的主动推理：Agent在空闲时主动检索长期记忆，发现用户之前提到过的问题未解决，主动触发回访流程。
• 联邦记忆：多个Agent实例共享长期记忆库，但需要解决数据安全与冲突合并问题。

例如客服Agent A和客服Agent B遇到同一位客户，如何避免记忆矛盾。

9.4 参考资料

• LangGraph官方Memory文档：[链接待内部wiki补充]
• PostgreSQL AI扩展：pgvector、pg_later
• 向量数据库选型对比：FAISS vs Chroma vs Milvus 性能基准表
• LangChain Memory模块源码解析

文档撰写完毕，如有疑问或需要补充细节，欢迎联系本文作者或所在小组。

总结

通过本文的学习，相信你已经对「Agent长期记忆搭建」有了更深入的理解。建议结合实际项目多加练习。如有疑问，欢迎交流！