AI Agent 记忆系统(Memory System)设计


1. 需求分析与澄清

1.1 用例与用户画像

本文档以通用智能助手 Agent为核心场景进行设计,兼顾多轮对话、跨会话知识积累、工具调用等主流用例。目标用户画像如下:

用户类型 核心诉求 记忆需求
个人用户 跨会话保留偏好、历史决策 长期用户画像 + 会话摘要
企业知识工作者 访问内部文档、复盘过往任务 企业知识库 + Episode 记忆
自动化流程 批量任务、多步骤执行 任务状态机 + 工具调用结果缓存
多 Agent 协作 共享上下文、防止信息孤岛 共享黑板 + 消息传递协议

假设:本设计假设 LLM 主模型上下文窗口为 128k tokens,推理延迟目标为 < 3 秒(P95),系统并发规模为 1k~10k QPS。

1.2 核心功能需求

  1. 多层记忆分层存储:区分工作记忆、短期记忆、长期记忆,各层独立管理生命周期。
  2. 跨会话记忆持久化:用户历史对话、偏好、经验可在会话结束后保留并在新会话中检索。
  3. 高质量记忆检索:支持语义检索(向量相似度)+ 关键词检索(BM25)的混合策略。
  4. 记忆写回与更新:Agent 执行结束后能将有价值的经验自动提炼并写入长期记忆。
  5. 上下文压缩:当工作记忆接近 token 上限时,自动压缩或滑动窗口截断。
  6. 隐私与权限控制:不同用户的记忆严格隔离,敏感信息加密存储。

1.3 非功能性需求

维度 指标 说明
延迟 记忆检索 P95 < 100ms 不影响 LLM 推理链路
并发 1k~10k QPS 短期记忆读多写少
任务完成率 > 95% 记忆缺失不应导致任务失败
可观测性 结构化日志 + 链路追踪 每次检索/写入可回溯
安全性 RBAC + 数据加密 用户级记忆严格隔离
成本 Token 消耗可控 避免无效记忆注入上下文

1.4 边界与假设

  • 假设:Agent 以单租户或多租户 SaaS 形式部署,用户 ID 作为记忆隔离的主键。
  • 假设:长期记忆写回是异步操作,不阻塞主推理链路。
  • 边界外:强化学习在线调优不在本期设计范围内,作为后续演进方向。
  • 边界外:多模态记忆(图像、音频的向量化)本期仅预留接口,不展开实现。

2. 整体架构设计

2.1 架构概览(文字描述)

┌─────────────────────────────────────────────┐
│              感知 / 输入层                    │
│  用户消息 | 工具结果 | 外部 API | 文件/图像    │
└────────────────────┬────────────────────────┘
                     │ 构建上下文
                     ▼
┌─────────────────────────────────────────────┐
│         工作记忆(Working Memory)            │
│  · 当前对话上下文(Chat History Buffer)      │
│  · 任务状态机(Task State Machine)          │
│  · 本轮工具调用结果(Tool Results Cache)     │
│  · 容量:≤128k tokens,生命周期:单次任务     │
└──────────┬──────────────────────┬───────────┘
           │ 读/写                │ 语义检索
           ▼                      ▼
┌──────────────────┐    ┌─────────────────────┐
│  短期记忆         │    │  长期记忆             │
│  (Short-term)    │◄───►(Long-term)           │
│  · 会话摘要       │ 提炼│  · 知识库             │
│  · Episode 缓冲  │固化 │  · 用户画像           │
│  · Redis/KV      │    │  · 过往经验 Episodic  │
│  · TTL: 会话级   │    │  · 向量库+关系库+图库  │
└──────────────────┘    └─────────────────────┘
                     │ 注入相关记忆
                     ▼
┌─────────────────────────────────────────────┐
│         规划模块(Planner)                   │
│  Plan-and-Execute | 任务分解 | 动态重规划      │
└──────────────────────┬──────────────────────┘
                       │ 执行计划
                       ▼
┌─────────────────────────────────────────────┐
│         工具执行层(Action)                  │
│  工具注册表 | API 网关 | 结果写回记忆           │
└──────────────────────┬──────────────────────┘
                       │ 评估结果
                       ▼
┌─────────────────────────────────────────────┐
│         反思 / 经验沉淀(Reflection)          │
│  评估质量 | 生成摘要 | 异步写回长期记忆         │
└─────────────────────────────────────────────┘

核心数据流

  • 读路径:输入 → 触发记忆检索 → 从短期 + 长期记忆中取回相关片段 → 注入工作记忆 → LLM 推理。
  • 写路径:任务结束 → Reflection 模块评估 → 压缩摘要 → 异步写入短期/长期记忆。
  • 提炼路径:短期记忆 TTL 到期前,高价值 Episode 通过蒸馏固化到长期记忆。

2.2 核心设计模式

  • ReAct(Reasoning + Acting):用于工具调用循环,每步推理结合当前工作记忆。
  • Plan-and-Execute:复杂任务先生成全局计划,再逐步执行,支持动态重规划。
  • Reflection / Self-Critique:任务完成后自我评估,驱动记忆写回质量。
  • RAG(Retrieval-Augmented Generation):长期记忆的核心检索模式,按需注入上下文。

3. 详细模块设计

3.1 感知与输入处理

输入类型

输入类型 格式 处理策略
用户文本消息 plaintext / markdown 直接追加到对话历史
工具调用结果 JSON 序列化后追加,超长则摘要
文件/文档 PDF/DOCX/CSV 分块提取,存入临时向量索引
图像(预留) Base64 / URL 多模态模型生成文本描述

上下文构建策略

当前上下文 = 系统提示(System Prompt)
           + 相关长期记忆片段(语义检索 Top-K)
           + 会话摘要(短期记忆)
           + 近 N 轮对话(滑动窗口)
           + 当前用户输入

Token 预算分配(以 128k 为例):

  • 系统提示:~2k
  • 长期记忆注入:~8k(可动态调整)
  • 会话摘要:~4k
  • 近期对话:~40k
  • 工具定义 + 预留:~16k
  • 当前输入 + 输出:~58k

3.2 记忆系统(核心章节)

3.2.1 分层架构总览
┌─────────────────────────────────────────┐
│  Layer 1: Working Memory(工作记忆)       │  ← 最快,最小,单任务
├─────────────────────────────────────────┤
│  Layer 2: Short-term Memory(短期记忆)   │  ← 会话级,跨轮次
├─────────────────────────────────────────┤
│  Layer 3: Long-term Memory(长期记忆)    │  ← 持久化,跨会话
└─────────────────────────────────────────┘

在这里插入图片描述

3.2.2 工作记忆(Working Memory)

定义:Agent 当前推理循环中可直接访问的活跃信息,即 LLM 的上下文窗口(Context Window)本身。

核心组件

  • Chat History Buffer:最近 N 轮对话(动态滑动窗口)。
  • Task State Machine:当前子任务状态(pending / running / done / failed)。
  • Tool Results Cache:本轮工具调用的返回值,避免重复调用。
  • Scratch Pad:LLM 的中间推理草稿(Chain-of-Thought)。

容量管理策略

def manage_working_memory(context: list, token_budget: int) -> list:
    """
    滑动窗口 + 摘要压缩策略
    """
    current_tokens = count_tokens(context)
    
    if current_tokens < token_budget * 0.8:
        return context  # 未超 80%,直接返回
    
    # 超出阈值:对最早的 K 轮对话做摘要压缩
    old_turns = context[:COMPRESS_WINDOW]
    summary = llm_summarize(old_turns)  # 调用快速小模型
    return [summary] + context[COMPRESS_WINDOW:]

生命周期:随单次任务/会话创建和销毁,不持久化。


3.2.3 短期记忆(Short-term Memory)

定义:跨多轮对话但限于当前会话的记忆,存储会话摘要、中间结果、Episode Buffer。

数据模型

{
  "session_id": "sess_abc123",
  "user_id": "user_xyz",
  "created_at": "2026-06-07T10:00:00Z",
  "expires_at": "2026-06-07T22:00:00Z",
  "summary": "用户正在规划一次东京旅行,已确认5天行程,预算8000元。",
  "episodes": [
    {
      "turn": 12,
      "role": "tool_result",
      "content": "航班查询结果:MU123 6月20日 08:00 上海→东京,¥3200",
      "tokens": 48
    }
  ],
  "entity_slots": {
    "destination": "东京",
    "budget": 8000,
    "duration": 5
  }
}

存储选型:Redis(主)+ 关系型数据库(备份)

  • Redis 提供 O(1) 读写,TTL 自动清理。
  • 关系库作为持久化备份,支持会话审计。

检索策略

  • 精确查询GET session:{session_id}:summary
  • 实体槽位:提取当前输入的命名实体,与 entity_slots 对比补全。

写入时机

  • 每隔 5 轮对话触发一次增量摘要更新(异步)。
  • 会话结束时触发完整摘要生成 + 高价值 Episode 蒸馏流程。

3.2.4 长期记忆(Long-term Memory)

定义:持久化、跨会话的知识存储,包含三个子系统。

子系统一:语义知识库(Semantic Memory)

  • 存储事实性知识、领域知识、工具使用手册。
  • 存储后端:向量数据库(Qdrant / Weaviate / Pinecone)。
  • 向量模型:text-embedding-3-large(OpenAI)或 bge-large-zh-v1.5(中文)。
  • 检索:HNSW 近似近邻(ANN),Top-K=10,相似度阈值 > 0.75。

子系统二:情景记忆(Episodic Memory)

  • 存储过往具体任务的执行记录、成功/失败经验。
  • 数据结构:
{
  "episode_id": "ep_20260607_001",
  "user_id": "user_xyz",
  "task_type": "travel_planning",
  "summary": "成功完成东京5日旅行规划,用户最终满意度高。",
  "key_decisions": ["选择MU航班", "住新宿区", "包含富士山一日游"],
  "outcome": "success",
  "lessons": "用户偏好性价比,不喜欢奢华酒店",
  "embedding": [0.023, -0.145, ...],  // 768维向量
  "timestamp": "2026-06-07T18:00:00Z"
}

子系统三:用户画像(User Profile)

  • 存储用户长期偏好、习惯、权限配置。
  • 存储后端:关系型数据库(PostgreSQL)+ 内存缓存(Redis)。
  • 更新策略:Reflection 模块异步触发,每次任务结束更新。

混合检索策略(Hybrid RAG)

def retrieve_long_term_memory(query: str, user_id: str, top_k: int = 5):
    # 1. 向量语义检索
    query_embedding = embed(query)
    vector_results = vector_db.search(
        collection=f"user_{user_id}",
        vector=query_embedding,
        top_k=top_k * 2,
        score_threshold=0.75
    )
    
    # 2. BM25 关键词检索
    bm25_results = bm25_index.search(query, top_k=top_k * 2)
    
    # 3. RRF 融合排序(Reciprocal Rank Fusion)
    fused = reciprocal_rank_fusion(vector_results, bm25_results)
    
    # 4. 按相关性 + 时效性重排(近期记忆加权)
    reranked = rerank_with_recency(fused, decay_factor=0.95)
    
    return reranked[:top_k]

记忆写入(Memory Consolidation)

新 Episode → Reflection 评估(价值评分)
           → 评分 > 阈值 → 生成摘要向量 → 写入向量库
           → 更新用户画像实体槽位
           → 低价值 → 丢弃或仅保留短期

3.2.5 记忆检索触发机制
触发条件 检索类型 优先级
用户输入包含历史相关实体 长期 + 短期
任务类型与历史任务相似 Episodic Memory
工具调用失败,需参考过往经验 Episodic Memory
每轮对话开始 短期摘要 高(必选)
用户提及偏好/习惯 用户画像

3.3 规划模块

设计模式:Plan-and-Execute + 动态重规划

class Planner:
    def plan(self, task: str, memory_context: MemoryContext) -> TaskPlan:
        """
        生成初始任务计划,注入相关记忆
        """
        prompt = build_plan_prompt(task, memory_context)
        raw_plan = llm.generate(prompt, model="claude-sonnet-4-6")
        return parse_task_graph(raw_plan)  # 返回 DAG 结构
    
    def replan(self, original_plan: TaskPlan, 
               failure_info: FailureInfo,
               memory_context: MemoryContext) -> TaskPlan:
        """
        工具失败或结果异常时触发重规划
        - 从 Episodic Memory 检索类似失败案例
        - 生成替代方案
        """
        similar_failures = memory_context.search_failures(failure_info.type)
        return self.plan(original_plan.remaining_tasks, 
                        memory_context.with_failures(similar_failures))

触发重规划的条件

  • 工具调用连续失败 ≥ 2 次。
  • 中间结果与预期偏差超过阈值。
  • 用户实时反馈要求调整。
  • 任务完成率指标低于 80%(批量场景)。

任务依赖管理:DAG(有向无环图)表示子任务依赖,支持并行执行无依赖分支。


3.4 工具调用与执行

工具注册表(Tool Registry)

TOOL_REGISTRY = {
    "search_web": {
        "description": "搜索实时网页信息",
        "permissions": ["public"],
        "cache_ttl": 300,  # 5分钟缓存
        "timeout": 10,
        "rate_limit": "100/min"
    },
    "write_file": {
        "description": "写入文件系统",
        "permissions": ["admin", "write"],
        "require_hitl": True,   # 高风险操作需人工确认
        "timeout": 30,
    }
}

工具执行结果写回记忆

def execute_tool_and_record(tool_name: str, args: dict, 
                             working_memory: WorkingMemory):
    result = tool_registry.call(tool_name, args)
    
    # 写回工作记忆(本轮可见)
    working_memory.add_tool_result(tool_name, args, result)
    
    # 重要结果写入短期记忆
    if result.importance_score > 0.7:
        short_term_memory.add_episode(
            type="tool_result",
            content=result.summary,
            metadata={"tool": tool_name, "args": args}
        )
    
    return result

错误处理策略

错误类型 处理策略
网络超时 指数退避重试,最多 3 次
参数错误 LLM 自纠正,重新生成参数
权限不足 降级到只读工具,或触发 HITL
工具不可用 检索 Episodic Memory 中替代方案
结果异常 触发重规划,记录到失败经验库

3.5 反思与自我优化

反思触发时机

  • 每个子任务完成后(轻量反思)。
  • 整个任务链结束后(深度反思 + 记忆写回)。
  • 用户显式反馈(点赞/差评)。

反思流程

任务结束
  │
  ├── 评估指标计算
  │     ├── 目标达成率(用户反馈 or 自动评估)
  │     ├── 工具调用效率(调用次数 vs 最优路径)
  │     └── 上下文利用率(检索的记忆是否被使用)
  │
  ├── 经验提炼
  │     ├── 成功经验 → 提炼为可复用 Pattern
  │     └── 失败经验 → 提炼为 Anti-Pattern
  │
  └── 记忆写回(异步)
        ├── 高价值 Episode → 向量库
        ├── 用户偏好更新 → 用户画像
        └── 工具调用最优路径 → Procedural Memory

经验价值评分

def score_episode(episode: Episode) -> float:
    """综合评分决定是否写入长期记忆"""
    score = 0.0
    score += episode.user_satisfaction * 0.4   # 用户满意度
    score += episode.task_novelty * 0.3         # 任务新颖性
    score += episode.tool_efficiency * 0.2      # 工具使用效率
    score += episode.error_recovery * 0.1       # 异常恢复质量
    return score  # > 0.6 则写入长期记忆

4. 多智能体协作

4.1 角色定义

角色 职责 记忆访问权限
Orchestrator 任务分发、全局规划 读写所有层
Memory Agent 专职记忆检索与写回 读写所有层
Worker Agent 具体子任务执行 读工作记忆,写短期
Critic Agent 结果质量评估 只读所有层

4.2 共享黑板(Shared Blackboard)

多 Agent 间通过共享短期记忆(Redis Pub/Sub)协调:

# Orchestrator 发布任务状态
blackboard.publish("task:travel_001:status", {
    "phase": "hotel_search",
    "worker_id": "worker_02",
    "partial_result": {...}
})

# Worker 订阅并更新
blackboard.subscribe("task:travel_001:*", callback=handle_update)

4.3 记忆冲突消解

当多个 Worker 同时更新同一记忆条目时:

  • 向量库:追加策略,不覆盖,由 Memory Agent 定期去重。
  • 用户画像:乐观锁 + CAS(Compare-And-Swap),最后写入者胜。
  • 短期摘要:Orchestrator 统一归并,不允许 Worker 直接写入。

5. 非功能性保障

5.1 可靠性

死循环避免

  • 工作记忆中维护 iteration_count,超过 MAX_ITER=20 强制终止。
  • 检测目标漂移:连续 3 轮输出相似度 > 0.95 判定为循环,触发重规划。

状态持久化

# 每完成一个子任务,异步持久化任务状态
async def checkpoint(task_id: str, state: TaskState):
    await redis.setex(
        f"checkpoint:{task_id}",
        3600,
        state.serialize()
    )

故障恢复:Agent 崩溃后从最近 Checkpoint 恢复,续跑剩余子任务。

5.2 安全性

  • 用户记忆隔离:所有记忆 Key 以 user_id 为前缀,向量库使用独立 Collection 或 namespace 过滤。
  • HITL(人工介入):不可逆操作(删除文件、发送邮件、支付)强制等待用户确认。
  • 工具权限最小化:Worker Agent 只能调用 Orchestrator 授权的工具子集。
  • 记忆内容过滤:写入长期记忆前,通过 PII 检测器过滤身份证、银行卡等敏感信息。

5.3 可观测性

结构化日志(每次记忆操作必须记录):

{
  "trace_id": "tr_abc123",
  "event": "memory_retrieve",
  "layer": "long_term",
  "query": "东京旅行预算",
  "results_count": 5,
  "top_score": 0.89,
  "latency_ms": 45,
  "tokens_injected": 312,
  "user_id": "user_xyz",
  "session_id": "sess_abc",
  "timestamp": "2026-06-07T10:05:23Z"
}

核心监控指标

指标 告警阈值
记忆检索延迟 P95 > 200ms
记忆写回失败率 > 1%
工作记忆 Token 利用率 > 90%
长期记忆命中率 < 30%(可能记忆质量差)
Episode 写入速率 异常波动 > 5×

链路追踪:使用 OpenTelemetry,记忆检索作为独立 Span,与 LLM 推理 Span 组成完整 Trace。

5.4 成本与延迟优化

模型分流

操作 模型选择 理由
主推理 Claude Sonnet 4.6 平衡质量与速度
上下文压缩 Claude Haiku 4.5 低延迟,低成本
向量化 text-embedding-3-small 速度优先
深度反思 Claude Opus 4.6 质量优先,异步执行

上下文压缩:采用递归摘要(Recursive Summarization),将 N 轮对话压缩为 1 段摘要,压缩比约 10:1。

记忆检索缓存:对相同 query + user_id 的检索结果缓存 60 秒(Redis),避免重复向量计算。


6. 关键流程举例

6.1 正常流程:跨会话记忆召回

用户(新会话): "帮我继续上次的东京旅行规划"
     │
     ▼
[感知层] 解析意图 → task_type: travel_planning, continuation: true
     │
     ▼
[短期记忆] GET session_history → 无(新会话)
     │
     ▼
[长期记忆检索] 
  query = "东京旅行规划" + user_id = "user_xyz"
  → 命中 Episode ep_20260607_001:
    "用户已确认5天行程,预算8000元,偏好性价比"
  → 命中用户画像:偏好经济舱,不喜欢奢华酒店
     │
     ▼
[工作记忆] 注入:历史摘要(312 tokens)+ 用户画像(128 tokens)
     │
     ▼
[LLM 推理] "基于您上次的规划,您已确定5天行程和8000元预算..."
     │
     ▼
[输出] 续接上次内容,用户无需重新说明背景

6.2 异常处理流程:工具失败 + 记忆辅助重规划

[Planner] 生成计划:调用 flight_search_tool 查询 6月20日上海→东京
     │
     ▼
[Action] flight_search_tool → 超时(第1次)
     │ 指数退避 2s
     ▼
[Action] flight_search_tool → 超时(第2次)
     │ 触发重规划
     ▼
[Episodic Memory 检索] query: "flight_search 失败替代方案"
  → 命中历史经验:
    "flight_search 不可用时,改用 web_search('携程东京航班') 可获取结果"
     │
     ▼
[Replanner] 替换计划:使用 web_search 工具
     │
     ▼
[Action] web_search("携程 上海东京 6月20日 航班") → 成功
     │
     ▼
[Reflection] 记录成功路径:写入 Episodic Memory
  "flight_search 超时时,web_search 携程可作为替代方案(成功率高)"
     │
     ▼
[输出] 返回航班信息给用户,整体任务完成

7. 技术选型与依赖

组件 选型 选择理由
主推理模型 Claude Sonnet 4.6 长上下文(200k)、工具调用能力强
快速压缩模型 Claude Haiku 4.5 低延迟(<500ms),用于摘要压缩
深度反思模型 Claude Opus 4.6 最强推理质量,异步执行不影响延迟
向量数据库 Qdrant 开源、高性能、支持 payload 过滤
关系型数据库 PostgreSQL + pgvector 用户画像 + 混合检索备选
图数据库 Neo4j(预留) 知识图谱,实体关系推理
KV 缓存 Redis 7 短期记忆 + 检索结果缓存
消息队列 Kafka 记忆写回异步解耦
Agent 框架 LangGraph 支持有状态图执行,记忆管理友好
向量模型 bge-large-zh-v1.5 中文语义效果最佳
链路追踪 OpenTelemetry + Jaeger 开源标准,记忆操作可追溯
对象存储 MinIO / S3 长期记忆原文备份

8. 风险、权衡与演进

8.1 当前设计局限

局限 描述
记忆噪声问题 长期记忆积累后,低质量 Episode 可能干扰检索结果
向量检索幻觉 相似度高但语义不匹配的记忆被误注入上下文
摘要失真 压缩摘要可能丢失细节,影响续写任务准确性
冷启动问题 新用户无历史记忆,需设计默认偏好模板
多模态记忆 图像/音频的向量化记忆本期未实现,存在能力缺口

8.2 潜在风险

  • 成本爆炸:若记忆检索结果过多注入上下文,Token 消耗显著增加,需严格控制 Top-K 和 Token 预算。
  • 无限循环:重规划循环若未设置最大次数限制,可能导致无限重试。已通过 MAX_ITER 和熔断机制应对。
  • 记忆泄漏:多租户场景下,若 user_id 过滤逻辑出现 Bug,可能导致跨用户记忆访问,需单元测试覆盖。
  • 幻觉扩散:LLM 生成的错误摘要写入长期记忆后,可能持续影响后续任务。需通过 Critic Agent 和置信度过滤控制写入质量。

8.3 后续演进方向

  1. 记忆重要性动态衰减:引入时间衰减函数,旧记忆权重逐渐降低,避免长期记忆膨胀。
  2. 强化学习优化检索策略:用任务完成率作为奖励信号,在线优化哪类记忆最值得注入。
  3. 长期记忆自动压缩:定期对 Episodic Memory 做层次聚类 + 摘要蒸馏,控制存储规模。
  4. 多模态记忆:集成图像 CLIP 向量化,支持图像记忆存储与检索。
  5. 记忆图谱化:将实体关系存入 Neo4j,支持多跳推理和复杂关联查询。
  6. 联邦记忆:多个 Agent 实例共享部分记忆(只读),支持大规模分布式协作。

9. 总结

9.1 方案亮点

亮点 说明
三层分层架构 工作记忆(速度)+ 短期记忆(会话)+ 长期记忆(持久),各层职责清晰
混合检索策略 向量语义 + BM25 关键词 + RRF 融合,兼顾召回率与精确率
异步写回设计 记忆写入不阻塞主推理链路,保证 < 3s 端到端延迟
Reflection 驱动 任务结束自动提炼经验,记忆质量持续提升
Episodic 失败库 将错误经验结构化存储,驱动重规划,实现自我纠错
可观测性完备 每次记忆操作均有结构化日志和 OpenTelemetry Span

9.2 核心 Trade-off 总结

决策 选择 放弃
记忆写回时机 异步(低延迟) 写后立即一致性
检索策略 混合 RAG(高质量) 简单向量检索(低成本)
长期记忆存储 向量库 + 关系库(灵活) 纯关系库(简单)
摘要压缩 快模型(低成本) 最高质量压缩
工具失败处理 从记忆检索替代方案(鲁棒) 直接报错终止(简单)

9.3 适用场景

本记忆系统设计适用于:

  • 需要跨会话记忆的个人/企业智能助手。
  • 需要经验积累的自动化 Agent(随时间越用越聪明)。
  • 需要高质量上下文管理的长任务执行 Agent。
  • 多 Agent 协作场景中的共享知识底座

如有纰漏,欢迎指正!

更多推荐