三层记忆架构设计

核心设计思想：

拆分记忆层级 + 结构化存储 + 动态压缩 + 按需召回 + 遗忘 / 淘汰 + 安全隔离
不依赖大模型原生上下文，通用适配单 Agent / 多 Agent、长对话、业务系统。

短时记忆（Working Memory）

• 作用：存储当前会话实时交互数据
• 存储方式：内存或会话State（如LangGraph的State对象）
• 特点：
  • 高时效性，随会话结束销毁
  • 支持实时增删改和权限控制
  • 内容：用户输入、Agent输出、临时槽位、流程状态

中长期记忆（Session Memory）

• 作用：沉淀单次会话核心信息
• 存储方式：Redis或临时数据库（绑定session_id）
• 处理逻辑：
  • 自动摘要和去重
  • 保留结构化数据（用户需求、偏好、关键结论）

长期记忆（Long-term Memory）

• 作用：跨会话持久化核心数据
• 存储方式：关系型数据库+向量库（如Milvus/FAISS）
• 内容：
  • 用户画像、历史业务记录、高频偏好
• 能力：支持语义检索和定时归档

---

结构化存储规范

字段分类设计

# 消息类基础字段
msg_fields = ["role", "content", "timestamp", "msg_type"]

# 业务结构化字段
biz_fields = ["slots", "intent", "user_profile", "biz_data"]

# 标记与索引字段
meta_fields = [
    "sensitivity",  # 敏感等级
    "valid",        # 有效性标记
    "agent_permission",
    "session_id",
    "user_id",
    "vector_embedding"
]

---

五大核心能力实现

1. 记忆写入与抽取

• 实时轻量级NLP处理：
  • 意图识别（例如BERT-small）
  • 槽位填充（正则/规则兜底）
• 分级存储策略：
  • 闲聊内容直接丢弃
  • 敏感数据写入隔离区

2. 动态压缩算法

• 触发条件：
  • 上下文长度≥4K tokens
  • 或对话轮次>10轮
• 技术实现：
  • 增量式摘要（T5-base微调）
  • 冗余对话合并（Levenshtein距离去重）

3. 按需召回机制

def memory_recall(current_intent, user_id):
    # 向量检索长期记忆
    long_term = vector_search(current_intent, user_id)  
    # 结合权限过滤
    return filter_by_permission(long_term, agent_role)

4. 遗忘策略

类型	实现方式	示例阈值
时间遗忘	Redis TTL自动过期	会话数据保留7天
容量遗忘	LRU淘汰机制	上限1000条记录
有效性遗忘	标记valid=False	冲突数据自动覆盖

5. 安全隔离方案

• 多租户隔离：user_id+session_id双校验
• 敏感数据加密：AES-256加密存储
• 注入检测：正则匹配恶意模式（如SQL注入特征）

---

技术选型建议

组件	推荐方案	适用场景
短时记忆	内存字典	单机部署
会话记忆	Redis Cluster	分布式Agent
向量检索	FAISS-IVF	千万级数据
摘要模型	T5-small+规则后处理	低成本场景

---

关键流程伪代码

def process_message(input_msg):
    # 安全检测
    if detect_malicious(input_msg):
        raise SecurityError
    
    # 记忆召回
    recalled_mem = memory_recall(
        intent=parse_intent(input_msg),
        user_id=input_msg.user_id
    )
    
    # 生成Prompt
    prompt = build_prompt(
        working_mem=get_working_memory(),
        recalled_mem=recalled_mem
    )
    
    # 推理与记忆更新
    response = llm.generate(prompt)
    update_memory_layers(response)
    
    # 定时维护任务
    if needs_compression():
        run_memory_compression()

---

性能优化技巧

• 冷热分离：高频访问的短期记忆驻留内存
• 批量写入：中长期记忆采用异步批量提交
• 索引优化：对user_id+timestamp建立复合索引
• 分级缓存：Redis→Memcached→DB三级存储