智能Agent记忆管理实战：从短期记忆到长期记忆的架构演进

1次阅读

共计 2354 个字符，预计需要花费 6 分钟才能阅读完成。

在开发智能 Agent 时，记忆管理是个让人头疼的问题。想象一下，当用户问 ” 还记得我上周提过的需求吗？” 时，Agent 要么一脸懵，要么翻出几十条无关对话。这种尴尬背后是两大核心问题：

短期记忆 (Short-term Memory) 局限性：当前主流 LLM 的上下文窗口有限（比如 GPT- 4 的 32k tokens），长对话会被无情截断
长期记忆 (Long-term Memory) 检索低效：用户画像、历史行为等数据散落在不同系统，传统 SQL 查询根本不适合语义搜索

更糟的是，当你想把两者结合时，会发现：

Redis 缓存的历史对话很快过期
向量数据库虽然能语义搜索，但实时更新成本高
分布式部署时，记忆同步可能引发状态冲突

技术方案	写入延迟(ms)	读取延迟(ms)	向量检索 QPS
Redis	1.2	0.8	不支持
Memcached	0.9	0.6	不支持
Pinecone	15.4	8.7	1200
FAISS(本地)	N/A	N/A	8500

热数据：Redis/Memcached 适合存储短期对话上下文
冷数据：FAISS+Pinecone 组合实现低成本长期记忆
混合架构：用 Redis 做缓存层减轻向量库压力

import redis
from datetime import timedelta

class ShortTermMemory:
    def __init__(self, host='localhost', port=6379):
        self.client = redis.StrictRedis(
            host=host, 
            port=port,
            decode_responses=True
        )

    def add_context(self, user_id: str, message: str, ttl: int = 3600) -> bool:
        """存储对话上下文，默认 1 小时过期"""
        key = f"st_mem:{user_id}"
        return self.client.append(key, message + "\n") and \
               self.client.expire(key, ttl)

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class LongTermMemory:
    def __init__(self, dim=384):
        self.index = faiss.IndexFlatL2(dim)
        self.encoder = SentenceTransformer('paraphrase-MiniLM-L6-v2')
        self.redis = redis.StrictRedis()  # 用于存储原始文本

    def add_memory(self, user_id: str, text: str) -> int:
        """将文本编码后存入向量库"""
        embedding = self.encoder.encode(text)
        vec_id = self.index.ntotal
        self.index.add(np.array([embedding]))

        # 原始文本存 Redis，7 天过期
        self.redis.set(f"lt_mem:{user_id}:{vec_id}", 
            text,
            ex=timedelta(days=7)
        )
        return vec_id

当记忆条目超过阈值时，用 LLM 提取关键信息：

def compress_memory(texts: List[str]) -> str:
    prompt = """ 请从以下对话中提取关键信息，保持时间、人物、需求等核心要素：{texts}
    ---
    摘要："""return llm.generate(prompt.format(texts="\n".join(texts)))

分布式环境下推荐使用 Redis 乐观锁：

with self.redis.pipeline() as pipe:
    try:
        pipe.watch(user_id)
        # 读取当前版本
        current_ver = pipe.get(f"{user_id}:version")
        pipe.multi()
        pipe.set(f"{user_id}:memory", new_memory)
        pipe.incr(f"{user_id}:version")
        pipe.execute()
    except WatchError:
        # 重试或合并冲突
        handle_conflict()

预热方案：启动时加载最近 3 天活跃用户的记忆索引
懒加载：首次查询时异步构建用户专属向量子集

实现自动清理接口：

@app.delete("/memories/{user_id}")
def forget_user(user_id: str):
    # 删除所有相关键
    keys = redis.keys(f"*:{user_id}:*")
    if keys:
        redis.delete(*keys)
    # 从向量库移除（需维护映射关系）...

使用 Locust 模拟的基准测试结果（单节点 8 核 16G）：