Agent 记忆系统的架构设计：短期、长期与工作记忆的分层存储策略
Agent 记忆系统的架构设计：短期、长期与工作记忆的分层存储策略
一、从无状态 Agent 到有记忆智能体——为什么分层记忆是工程化的必答题
智能 Agent 从简单的"一问一答"演进到"持续协作"的过程中，记忆系统是第一道必须跨越的工程门槛。一个不具备记忆的 Agent 每次对话都从零开始，无法记住用户偏好、无法积累任务上下文、更无法在多轮交互中保持连贯的目标感。
在生产环境中，这个问题会被进一步放大。当 Agent 需要跨会话记忆用户的 API 调用习惯、项目代码风格、常用的数据库连接参数时，扁平化的记忆存储方案在三个维度上暴露短板。首先是检索效率，随着记忆条目增长到数万条，全量加载到上下文的做法会直接击穿 Token 预算。其次是信息衰减，不分层级的记忆存储会导致关键决策信息被大量历史噪音淹没。最后是跨会话一致性，单一存储方案难以同时满足高频工作记忆的写入性能和长期记忆的持久化要求。
分层记忆架构的核心思想借鉴了认知心理学中的人类记忆模型。短期记忆对应当前会话的有效上下文，生命周期以分钟计。长期记忆对应跨会话持久化的事实性知识，生命周期以月或年计。工作记忆则是短期记忆与长期记忆的交汇层，承载当前任务最相关的上下文子集。这一分层策略为大模型 Agent 的工程化落地提供了可扩展的存储蓝图。
二、分层记忆的存储模型与数据流转——三层缓存的协同机制
分层记忆架构的数据流转遵循"写入→检索→合并→裁剪"的循环链路。每次用户交互触发以下步骤：当前会话的新信息首先写入短期记忆缓冲，同时语义编码器将其转化为向量嵌入；检索层从长期记忆中召回与当前上下文相关的 Top-K 条目；工作记忆合并短期记忆和检索到的长期记忆，经裁剪策略过滤后注入 Prompt；会话结束时，关键的决策结果和模式总结经重要性评分筛选后持久化到长期记忆。
这一架构的关键决策点在于两个环节：记忆检索的相似度阈值与记忆裁剪的重要性评分算法。检索阈值过高会导致有用信息遗漏，过低则引入无关噪音。重要性评分的核心依据包括信息的新近性（最近访问时间）、频率（被检索次数）和关联性（与核心任务目标的语义相似度）。
三、基于 Redis 与向量数据库的混合存储实现——面向 Java 后端的生产级代码
在 Java 技术栈下，分层记忆的存储层可以采用 Redis + Milvus/PGVector 的混合方案。短期记忆使用 Redis 的 Hash 结构存储，利用 TTL 机制自动淘汰过期数据。长期记忆使用向量数据库存储嵌入向量和元数据。以下代码展示了一个基于 Spring Boot 3 + Jedis + Milvus Java SDK 的记忆管理器核心实现。
import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.SetParams;
import io.milvus.client.MilvusServiceClient;
import io.milvus.grpc.SearchResults;
import io.milvus.param.dml.SearchParam;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.*;
import java.util.stream.Collectors;
/**
* 分层记忆管理器——协调短期、长期与工作记忆的读写
*/
public class LayeredMemoryManager {
private static final String SHORT_TERM_PREFIX = "mem:short:";
private static final int SHORT_TERM_TTL_SECONDS = 1800; // 30分钟
private static final int WORKING_MEMORY_MAX_TOKENS = 4096;
private static final double IMPORTANCE_THRESHOLD = 0.65;
private final Jedis jedis;
private final MilvusServiceClient milvusClient;
private final EmbeddingService embeddingService;
public LayeredMemoryManager(Jedis jedis,
MilvusServiceClient milvusClient,
EmbeddingService embeddingService) {
this.jedis = jedis;
this.milvusClient = milvusClient;
this.embeddingService = embeddingService;
}
/**
* 写入短期记忆，设置自动过期
*/
public void writeShortTerm(String sessionId, String role, String content) {
if (sessionId == null || sessionId.isEmpty() || content == null) {
throw new IllegalArgumentException("sessionId 和 content 不能为空");
}
long timestamp = System.currentTimeMillis();
String field = role + ":" + timestamp;
String key = SHORT_TERM_PREFIX + sessionId;
try {
jedis.hset(key, field, content);
jedis.expire(key, SHORT_TERM_TTL_SECONDS);
} catch (Exception e) {
throw new MemoryOperationException("写入短期记忆失败，sessionId=" + sessionId, e);
}
}
/**
* 从长期记忆中检索与当前查询语义相似的 Top-K 条目
*/
public List<MemoryEntry> retrieveLongTerm(String queryEmbedding, int topK) {
List<String> searchOutputFields = Arrays.asList("content", "importance", "timestamp");
SearchParam searchParam = SearchParam.newBuilder()
.withCollectionName("agent_long_term_memory")
.withTopK(topK)
.withFloatVectors(Collections.singletonList(
parseEmbeddingString(queryEmbedding)))
.withOutFields(searchOutputFields)
.withParams("{\"nprobe\": 16}")
.build();
try {
SearchResults results = milvusClient.search(searchParam).getData();
if (results.getResults().getFieldsDataCount() == 0) {
return Collections.emptyList();
}
return results.getResults().getFieldsDataList().stream()
.map(data -> new MemoryEntry(
data.getScalars().getStringData().getData(0),
data.getScalars().getFloatData().getData(0),
data.getScalars().getLongData().getData(0)))
.collect(Collectors.toList());
} catch (Exception e) {
// 向量检索失败时降级返回空列表，不阻断主流程
System.err.println("长期记忆检索异常：" + e.getMessage());
return Collections.emptyList();
}
}
/**
* 合并短期与长期记忆，经裁剪策略生成工作记忆
* 裁剪依据：重要性评分 + 新近性 + Token 预算
*/
public List<MemoryEntry> buildWorkingMemory(String sessionId, String queryEmbedding) {
// 1. 读取当前会话的短期记忆
String key = SHORT_TERM_PREFIX + sessionId;
Map<String, String> shortTermRaw;
try {
shortTermRaw = jedis.hgetAll(key);
} catch (Exception e) {
shortTermRaw = Collections.emptyMap();
}
List<MemoryEntry> candidates = shortTermRaw.entrySet().stream()
.map(e -> new MemoryEntry(e.getValue(), 0.9f, parseTimestamp(e.getKey())))
.collect(Collectors.toList());
// 2. 检索长期记忆
List<MemoryEntry> longTermResults = retrieveLongTerm(queryEmbedding, 10);
candidates.addAll(longTermResults);
// 3. 按综合评分排序：0.4 * 重要性 + 0.3 * 新近度 + 0.3 * 语义相似度
long currentTime = System.currentTimeMillis();
candidates.sort((a, b) -> Double.compare(
compositeScore(b, currentTime), compositeScore(a, currentTime)));
// 4. 按 Token 预算裁剪
List<MemoryEntry> workingMemory = new ArrayList<>();
int tokenBudget = 0;
for (MemoryEntry entry : candidates) {
int estimatedTokens = estimateTokens(entry.getContent());
if (tokenBudget + estimatedTokens > WORKING_MEMORY_MAX_TOKENS) {
break;
}
// 过滤低重要性低相似度的噪音
if (entry.getImportance() < IMPORTANCE_THRESHOLD
&& entry.getSimilarity() < 0.5) {
continue;
}
workingMemory.add(entry);
tokenBudget += estimatedTokens;
}
return workingMemory;
}
/**
* 会话结束时，高重要性记忆持久化到长期记忆
*/
public void persistToLongTerm(String embedding, String content, float importance) {
if (importance < IMPORTANCE_THRESHOLD) {
return; // 低于阈值不持久化
}
try {
List<List<Float>> vectors = Collections.singletonList(
parseEmbeddingString(embedding));
// 省略实际的 insert 调用细节...
System.out.println("长期记忆持久化完成，importance=" + importance);
} catch (Exception e) {
throw new MemoryOperationException("长期记忆持久化失败", e);
}
}
private double compositeScore(MemoryEntry entry, long currentTime) {
double recency = 1.0 - Math.min(1.0,
(currentTime - entry.getTimestamp()) / (24.0 * 3600_000));
return 0.4 * entry.getImportance()
+ 0.3 * recency
+ 0.3 * entry.getSimilarity();
}
private int estimateTokens(String content) {
return content == null ? 0 : content.length() / 2;
}
private long parseTimestamp(String field) {
String[] parts = field.split(":");
return parts.length >= 2 ? Long.parseLong(parts[1]) : 0L;
}
private List<Float> parseEmbeddingString(String embedding) {
return Arrays.stream(embedding.split(","))
.map(Float::parseFloat).collect(Collectors.toList());
}
static class MemoryEntry {
private final String content;
private final float importance;
private final float similarity;
private final long timestamp;
public MemoryEntry(String content, float importance, long timestamp) {
this(content, importance, 0.0f, timestamp);
}
public MemoryEntry(String content, float importance, float similarity, long timestamp) {
this.content = content;
this.importance = importance;
this.similarity = similarity;
this.timestamp = timestamp;
}
public String getContent() { return content; }
public float getImportance() { return importance; }
public float getSimilarity() { return similarity; }
public long getTimestamp() { return timestamp; }
}
static class MemoryOperationException extends RuntimeException {
public MemoryOperationException(String message, Throwable cause) {
super(message, cause);
}
}
interface EmbeddingService {
String encode(String text);
}
}
以上实现有几个值得关注的工程决策。短期记忆使用 Redis Hash 而非 List，目的是支持按消息角色和目标时间戳进行精确检索。工作记忆的裁剪策略采用了加权评分而非简单的 Top-K 选取，融合重要性、新近度与语义相似度三个维度。长期记忆的持久化前置了重要性评分过滤，避免低质量数据污染向量库。
四、分层架构的边界条件与工程权衡——当记忆系统成为新的瓶颈
分层记忆架构并非银弹，其引入的复杂性需要在工程层面审慎评估。
在延迟维度，每次交互都需触发向量检索与评分排序，即使使用 Redis 缓存热门检索向量的余弦相似度索引，P99 延迟仍可能增加 30ms 至 80ms。在实时性要求严格的场景下，可以引入异步预加载机制——在用户输入前，基于上一次交互的上下文预取长期记忆候选集。
在存储成本维度，每条长期记忆需存储原文本和 768 至 1536 维的嵌入向量。日均万次交互的业务场景下，单月新增存储可达数百 GB。需要制定清晰的存储分层策略：近 7 天的向量存储在高性能 SSD 上，30 天以上归档到对象存储。
在准确性维度，向量检索的语义匹配天然存在召回偏差。对于需要精确事实记忆的场景（如用户说"上次我们使用的数据库连接端口是 3307"），纯粹的向量检索可能漏召回。此时应在长期记忆层引入结构化字段索引与向量索引的双通道检索——先用结构化条件过滤候选集，再在缩小后的范围内做语义排序。
这一架构的禁用场景包括：单次交互即结束的问答场景（记忆系统额外开销无收益）、毫秒级延迟要求的在线推理链路（应使用纯内存缓存替代）、以及记忆体量极小（百条以内）的轻量场景（全量加载到上下文更简单高效）。
五、总结
Agent 记忆系统的分层存储策略是构建有状态智能体的核心基础设施。短期记忆缓冲提供会话内的高速读写，长期记忆向量库支持跨会话的知识积累，工作记忆作为两者的合并层负责上下文裁剪与排序。
从落地节奏看，推荐分三个阶段推进。第一阶段实现基于 Redis 的短期记忆，验证会话内的上下文连续性。第二阶段引入向量数据库和语义检索，打通跨会话记忆链路。第三阶段加入重要性评分与裁剪策略，完成三层记忆的自动化协同。
技术选型上，Java 后端推荐 Redis Stack（内置向量检索能力）或 Milvus 作为长期记忆存储，短期记忆层使用 Jedis 客户端连接 Redis 集群。记忆质量的关键不只在架构设计，更在于持久化前的信息筛选策略——什么样的信息值得被"记住"，这个问题的工程化答案决定了记忆系统是成为智能资产还是噪音仓库。
更多推荐
27
0- 0
已为社区贡献24条内容
所有评论(0)