详情

首页手游攻略 Agent Memory 多维分类:从认知科学到工程实现

Agent Memory 多维分类:从认知科学到工程实现

佚名 2026-07-15 19:11:00
 复制代码title: Agent Memory 多维分类——从认知科学到工程实现
tags: [agent, memory, ai, architecture]
date: 2026-07-13

概述

AI Agent 的长期记忆分类几乎全是从人类认知心理学搬运过来的。当前学界和工业界最主流的分类框架将 Agent Memory 分为 Perceptual、Working、Episodic、Semantic 四类,外加常被忽略但至关重要的 Procedural、Prospective、Collective 三类补充类型。

本文以认知科学为源头,结合 主流框架对比,用 Golang 给出完整参考实现,并补充一个常被忽略的维度:记忆的管理策略(遗忘、生命周期转化、读写操作)。


一、认知科学源头

AI Agent 的记忆分类根植于四个经典认知理论。这里只列要点,详细讨论见 认知科学到 AI Agent 记忆分类的完整映射。

1.1 Atkinson & Shiffrin 多存储模型(1968)

时间维度感觉记忆(0.5s) → 短期记忆(20-30s) → 长期记忆(永久)

这是 LangGraph 两层架构(Checkpointer / Store)的直接理论来源。

1.2 Tulving 的情景 vs 语义二分法(1972)

长期记忆 = 情景记忆("我记得我去过巴黎")+ 语义记忆("我知道巴黎是法国首都")

维度情景记忆语义记忆
内容个人经历的事件事实、概念、知识
时间有明确时间标签时间标签
组织时空关系概念关系
遗忘细节容易丢失相对稳定

Tulving 后来(1985)补充了 Perceptual Representation Systems——对应 AI 中的 embedding 层。

1.3 Baddeley 的工作记忆模型(1974)

短期记忆不是被动存储,而是主动加工系统中央执行器 → 语音回路 + 视空间画板 + 情景缓冲区

在 AI 中:Context Window = 工作记忆,不只是"存消息",而是模型在此窗口内进行注意、推理、整合。

1.4 Cohen & Squire 的陈述性 vs 程序性(1980)

类型含义AI 对应
陈述性/外显记忆"知道巴黎是法国首都"Episodic + Semantic
程序性/内隐记忆"知道怎么骑车"System Prompt、Tool Defs、Fine-tuning 权重

二、四种核心记忆分类

2.1 Perceptual Memory——感知记忆

认知原型:人类的图像记忆(Iconic Memory,~0.5秒)和回声记忆(Echoic Memory,~3-4秒)。

AI 对应Embedding 层——不是"存储系统",而是输入信号的编码器。

关键特征

  • 模态相关:文本用 text-embedding,图片用 CLIP,音频用 Whisper encoder
  • 是情景记忆和语义记忆的索引基础(所有检索都依赖 embedding 质量)
  • 不直接存储,输出向量供下游使用

2.2 Working Memory——工作记忆

认知原型:Baddeley 的工作记忆——容量约 7±2 个组块,持续约 20-30 秒。

AI 对应LLM 的 Context Window——Agent "此刻正在思考什么"。

 复制代码┌─── Working Memory (Context Window) ─────────────────────┐System Prompt (行为约束)                                   │
│ [Semantic Block] "用户偏好简洁回答" ← 从语义记忆注入        │
│ [Episodic Result] "上次对话摘要..."  ← 从情景记忆检索        │
│ Human: "帮我修改登录页面样式"         ← 当前输入            │
│ AI: "先看一下代码..."                ← 当前推理            │
│ Tool: read_file → login.go (234行)   ← 工具结果            │
│ [消息超Window时被 SummarizationMiddleware 压缩]            │
│ 容量上限: 模型上下文窗口 (128K / 200K / 1M tokens)         │
└──────────────────────────────────────────────────────────┘

三个关键操作:

操作认知对应AI 实现
编码 (Encoding)注意力选择信息@before_model 钩子选择性注入记忆
维持 (Maintenance)复述保持消息追加到 state["messages"]
操作 (Manipulation)推理、整合、决策LLM Attention 在上下文内计算

2.3 Episodic Memory——情景记忆

认知原型:Tulving——记得(remembering) 而非知道(knowing) 。包含 what、when、where 三要素。

AI 对应过往交互的记录序列——每次对话、每个工具调用、每个决策,以事件为单位存储。

 复制代码// 情景记忆的数据结构(参考 Generative Agents 的 Memory Stream)
type EpisodicRecord struct {
    ID          string      `json:"id"`
    Timestamp   time.Time   `json:"timestamp"`
    Content     string      `json:"content"`
    Importance  float64     `json:"importance"`  // LLM 评分 1-10
    Embedding   []float32   `json:"embedding"`
    ThreadID    string      `json:"thread_id"`
}// 示例数据
examples := []EpisodicRecord{
    {
        Timestamp:  time.Now().Add(-time.Hour),
        Content:   "用户询问了登录页面样式修改,Agent 读取了 login.go",
        Importance: 6,
        ThreadID:  "conv-042",
    },
    {
        Timestamp:  time.Now().Add(-24 * time.Hour),
        Content:   "用户要求用 Fyne 替代终端 UI,不接受 CLI 方案",
        Importance: 9,
        ThreadID:  "conv-041",
    },
}

检索策略:情景记忆的检索混合三个权重——这和语义记忆有本质区别。

 复制代码检索得分 = α × sematic_similarity + β × recency_score + γ × importance_scoreGenerative Agents 权重: 1.0 × recency + 1.0 × importance + 1.0 × relevance
CrewAI v1.15 权重:   0.5 × semantic + 0.3 × recency + 0.2 × importance

冷启动问题:新用户只有少量 episodic 记录,相似度全部很低时,不注入比注入不相关的更好。应该设置置信度门控。

框架情景记忆实现
Generative AgentsMemory Stream——recency × importance × relevance 检索
LettaArchival Memory——向量DB + archival_memory_search,Out-of-context
LangGraphStore (namespace="conversations")——带时间戳的 JSON 文档
EM-LLM事件分割 + 两阶段检索(语义相似 → 时间邻近扩展)
CrewAIUnified Memory 的 composite scoring

2.4 Semantic Memory——语义记忆

认知原型:Tulving——知道(knowing) 。没有时间标签——你知道"巴黎是法国首都"但你不记得"什么时候学会的"。

AI 对应结构化的、精心维护的长期知识——关于用户是谁、Agent 自己是谁、领域知识是什么。

 复制代码// 语义记忆 = 精炼的、无时间性的知识
type SemanticStore struct {
    UserProfile    UserProfile     `json:"user_profile"`
    Persona        AgentPersona    `json:"persona"`
    DomainKnowledge map[string]any `json:"domain_knowledge"`
}type UserProfile struct {
    Name        string            `json:"name"`
    Role        string            `json:"role"`
    TechStack   []string          `json:"tech_stack"`
    Preferences map[string]string `json:"preferences"`
    Constraints map[string]string `json:"constraints"`
}type AgentPersona struct {
    Name         string   `json:"name"`
    Capabilities []string `json:"capabilities"`
    Limitations  []string `json:"limitations"`
    Rules        []string `json:"behavior_rules"`
}

语义记忆常驻 vs 按需的关键分歧

策略代表框架做法
常驻上下文Letta Memory Blocks语义记忆固定在上下文中,Agent 用工具自行修改
按需检索LangGraph Store语义记忆存在外部,每次对话开始时检索注入

Letta 的做法更接近人类——你不需要"检索"自己的名字,它一直在那里。


三、超越四分类——常被忽略的记忆类型

3.1 Procedural Memory——程序性记忆

认知原型:Cohen & Squire——知道"怎么做"而非"知道是什么"。你可以骑自行车但很难用语言描述。

AI 对应固化在模型行为中,不需要检索:

 复制代码// 程序性记忆在 AI Agent 中的四种体现
type ProceduralMemory struct {
    // 1. System Prompt——始终在上下文中的行为规则
    SystemPrompt string    // 2. Tool Definitions——Agent 的能力边界
    Tools []ToolDef    // 3. Few-shot Examples——验证过的行为模式
    Examples []InteractionPattern    // 4. 最终形态:Fine-tuned Weights——模型本能
}// 示例:Voyager 的 Skill Library 是程序性记忆的代码化实现
type SkillLibrary struct {
    Skills map[string]Skill // 可执行、可组合的代码技能
}type Skill struct {
    Name        string   `json:"name"`
    Code        string   `json:"code"`         // 可执行代码
    Description string   `json:"description"`
    DependsOn   []string `json:"depends_on"`   // 可组合:技能之间可以调用
}

Voyager 的核心贡献:技能以可执行代码存储,可以在新环境中直接复用,无需重新生成。这避免了 fine-tuning 的灾难性遗忘。

3.2 Prospective Memory——前瞻记忆

认知原型:记住"下午 3 点有会议"——意图和计划。

 复制代码// 前瞻记忆——Agent 的意图和待办系统
type ProspectiveMemory struct {
    ActiveGoals     []Goal        `json:"active_goals"`
    ScheduledTasks  []ScheduledTask `json:"scheduled_tasks"`
    InterruptedTasks []InterruptedTask `json:"interrupted_tasks"`
}type Goal struct {
    ID        string    `json:"id"`
    Content   string    `json:"content"`
    Priority  int       `json:"priority"`
    CreatedAt time.Time `json:"created_at"`
    Status    string    `json:"status"` // active | blocked | completed
    ParentID  string    `json:"parent_id"` // 子目标层级
}type InterruptedTask struct {
    OriginalGoal  Goal   `json:"original_goal"`
    InterruptedAt time.Time `json:"interrupted_at"`
    Context       string `json:"context"` // 恢复任务需要的上下文
    ResumeHint    string `json:"resume_hint"`
}

为什么重要:AutoGPT 最大的失败模式就是 goal drift——Agent 在复杂任务中忘记了最初的目标。前瞻记忆就是为解决这个问题而存在的。

3.3 Collective Memory——集体记忆

认知原型:社会共享知识——团队 wiki、组织文化。

AI 对应多 Agent 系统的共享知识池

 复制代码// 集体记忆:多 Agent 共享的知识池
type CollectiveMemory struct {
    SharedEpisodic  *VectorStore  // 所有 Agent 可见的事件日志
    SharedSemantic  *GraphStore   // 共享知识图谱
    AgentPrivate    map[string]*MemoryStore // 每个 Agent 的私有记忆
}// ChatDev 的例子:
// CEO、CTO、Programmer 三个 Agent 共享一个 project memory
// 但各自有自己的 working memory


四、记忆的生命周期与转化

四类记忆不是孤立的——它们之间存在动态的转化关系。这一点在原始认知模型中没有对应,是 AI Agent 独有的工程挑战。

4.1 转化路径

 复制代码// 记忆生命周期管理器
type MemoryLifecycleManager struct {
    working     *WorkingMemory
    episodic    *EpisodicStore
    semantic    *SemanticStore
    procedural  *ProceduralConfig
}// 路径 1: Working → Episodic(自动)
// 每次对话结束时,工作记忆自动沉淀为情景记忆事件
func (m *MemoryLifecycleManager) CommitSession(session *Session) error {
    record := EpisodicRecord{
        Timestamp:  time.Now(),
        Content:    summarizeSession(session),
        Importance: llmScoreImportance(session),  // LLM 评分 1-10
        Embedding:  embed(session.Summary),
        ThreadID:   session.ID,
    }
    return m.episodic.Store(record)
}// 路径 2: Episodic → Semantic(需要 Reflection)
// 从多条情景记忆中提炼语义知识——定时批量处理,不阻塞用户交互
func (m *MemoryLifecycleManager) Reflect(ctx context.Context, userID string) error {
    recent := m.episodic.Query(userID, sinceLastReflection(userID), limit=20)
    if len(recent) < 3 {
        return nil // 不够多,不值得反思
    }    insights := m.llmGenerate(ctx, buildReflectionPrompt(recent))
    for _, ins := range insights {
        if ins.Confidence > 0.7 && !m.semantic.AlreadyExists(ins) {
            m.semantic.Upsert(userID, ins.Key, ins.Value)
        }
    }
    return nil
}// 路径 3: Semantic → Working(注入)
// 每次新对话开始时注入用户画像
func (m *MemoryLifecycleManager) InjectIntoWorking(userID string) string {
    profile := m.semantic.GetProfile(userID)
    prefs   := m.semantic.GetPreferences(userID)
    return fmt.Sprintf("[用户: %s, 偏好: %v]", profile, prefs)
}// 路径 4: Semantic → Procedural(固化,最慢)
// 经过充分验证的行为模式固化为 System Prompt 或 Fine-tune 样本
func (m *MemoryLifecycleManager) Solidify(rule Rule, evidenceCount int) {
    if evidenceCount >= SOLIDIFICATION_THRESHOLD {
        m.procedural.AddSystemRule(rule.Content)
        logSolidification(rule)
    }
}

4.2 关键设计决策

转化路径触发时机成本可靠性要求
Working → Episodic每次会话结束低(一次 embed + insert)
Episodic → Semantic定时离线(夜间/批量)中(LLM 调用反思)高(需去重+冲突检测)
Semantic → Working每次会话开始低(一次精确查询)极高
Semantic → Procedural经过充分验证后低(追加配置)极高(需人工审核)

4.3 EM-LLM 的事件分割贡献

EM-LLM(ICLR 2025)提出了一个关键创新:事件分割(Event Segmentation) 。不是按轮次或 token 数切割情景记忆,而是用 Bayesian Surprise 检测事件边界——当上下文发生显著变化时自动切分为新事件。

 复制代码原始 token 序列
    │
    ▼ Bayesian Surprise 检测边界
    ├── Event 1: "用户询问天气"       [tokens 0-145]
    ├── Event 2: "Agent 调用 API"     [tokens 146-312]
    ├── Event 3: "返回结果并解释"     [tokens 313-489]
    └── Event 4: "用户追问细节"       [tokens 490-...]
    │
    ▼ 两阶段检索
  Stage 1: 语义相似度 → 找到 Event 1
  Stage 2: 时间邻近扩展 → 带出 Event 2, 3, 4(保持事件完整性)

这比固定长度的 sliding window 更接近人类处理情景记忆的方式——人类记忆也是按"事件"而非"token 数"组织的。


五、遗忘策略——被忽视的记忆管理维度

5.1 策略对比

策略实现方式代表系统优点缺陷
FIFO 覆盖固定 buffer,最旧的出队RET-LLM简单,内存可控可能丢弃重要旧信息
手动删除用户/Agent 显式删除ChatDB精确需主动决策,容易堆积
合并压缩N 条同类 → 1 条总结GITM保留核心,减少存储丢失细节
重要性衰减low-importance 沉到底部Generative Agents软遗忘,不丢失数据仍然占用存储
硬遗忘定期清理低于阈值的老记录(学术盲区)释放存储,提升检索质量需要调阈值

5.2 Golang 实现

 复制代码// 遗忘策略接口
type ForgettingStrategy interface {
    ShouldForget(record EpisodicRecord) bool
}// 策略 1: 软遗忘——重要性衰减
type ImportanceDecay struct {
    DecayRate float64 // 如 0.1 => 每天衰减 10%
}func (s *ImportanceDecay) Apply(records []EpisodicRecord) []EpisodicRecord {
    for i := range records {
        days := time.Since(records[i].Timestamp).Hours() / 24
        records[i].Importance *= math.Exp(-s.DecayRate * days)
        if records[i].Importance < 0.1 {
            records[i].Importance = 0.1 // floor——不完全消失
        }
    }
    return records
}// 策略 2: 硬遗忘——存储清理(通常离线执行)
type HardForget struct {
    ImportanceThreshold float64
    AgeDaysThreshold    int
}func (s *HardForget) Cleanup(store EpisodicStore, userID string) (int, error) {
    candidates, err := store.Query(userID, QueryFilter{
        ImportanceMax: s.ImportanceThreshold,
        OlderThanDays: s.AgeDaysThreshold,
    })
    if err != nil {
        return 0, err
    }
    for _, record := range candidates {
        store.Delete(record.ID)
    }
    return len(candidates), nil
}// 策略 3: 合并压缩——同类信息聚合
type MergeCompress struct {
    MergeThreshold int // 同类记录达到 N 条时触发合并
}func (s *MergeCompress) Compact(store EpisodicStore, userID, topic string) error {
    related := store.FindByTopic(userID, topic)
    if len(related) < s.MergeThreshold {
        return nil
    }
    summary := llmSummarize(related) // LLM 总结 N 条 → 1 条
    for _, old := range related {
        store.Delete(old.ID)
    }
    return store.Store(EpisodicRecord{
        Content:    summary,
        Importance: maxImportance(related),
        Timestamp:  time.Now(),
    })
}// 遗忘调度器:定时执行
type ForgettingScheduler struct {
    strategies []ForgettingStrategy
}func (s *ForgettingScheduler) RunDaily(ctx context.Context) {
    // 在用户不活跃的时间段(如凌晨 3 点)运行
    for _, strat := range s.strategies {
        strat.Apply(...)
    }
}

六、框架实现对比

维度Generative AgentsLetta (MemGPT)LangGraphCrewAIEM-LLM
WorkingContext WindowContext WindowCheckpointer (state)当前对话Context Window
EpisodicMemory Stream (recency×importance×relevance)Archival Memory (向量DB)Store ("episodic" ns)Unified Memory (composite score)Event-based (Bayesian Surprise 分割)
SemanticReflections 写回 StreamMemory Blocks (常驻、自编辑)Store ("semantic" ns)Unified Memory (同一 DB)(不单独区分)
ProceduralPlanning 机制System PromptTool Defs + PromptAgent Role Config
核心创新ReflectionAgent 自己管理记忆(tool calling)namespace 隔离的通用 Store统一向量库 + 复合评分认知式事件分割
遗忘策略重要性衰减手动 (memory_replace)开发者自行实现
分类哲学学术完整性优先Agent 自主管理优先灵活性与自由度优先简单统一优先认知逼真度优先

关键分歧:Letta 让 Agent 自己管理语义记忆(用 tool calling),比其他框架更"认知真实"但也更激进;LangGraph 把分类决策完全交给开发者,是最灵活也最需要设计能力的方案。


七、Golang 参考实现

7.1 核心接口定义

 复制代码package memoryimport (
    "context"
    "time"
)// ============================================================
// 核心类型
// ============================================================// MemoryVector 记忆项的向量表示(感知层输出)
type MemoryVector []float32// MemoryRecord 通用记忆记录
type MemoryRecord struct {
    ID         string        `json:"id"`
    Content    string        `json:"content"`
    Embedding  MemoryVector  `json:"embedding"`
    Importance float64       `json:"importance"`
    CreatedAt  time.Time     `json:"created_at"`
    UpdatedAt  time.Time     `json:"updated_at"`
    Metadata   map[string]any `json:"metadata"`
}// RetrievalScore 检索评分
type RetrievalScore struct {
    SemanticSimilarity float64
    Recency            float64
    Importance         float64
    Composite          float64
}// ============================================================
// 感知层接口 (Perceptual Memory)
// ============================================================type Embedder interface {
    Embed(ctx context.Context, text string) (MemoryVector, error)
    EmbedBatch(ctx context.Context, texts []string) ([]MemoryVector, error)
    Dimension() int
}// ============================================================
// 工作记忆接口 (Working Memory)
// ============================================================type WorkingMemory interface {
    // 管理上下文窗口内的信息
    Append(msg Message)
    GetWindow(maxTokens int) []Message
    Summarize(ctx context.Context, model LLM) (string, error)
    // 注入外部记忆
    Inject(block string) // 语义记忆常驻块
    InjectEpisodic(results []MemoryRecord) // 情景检索结果
}type Message struct {
    Role      string    `json:"role"`     // system / user / assistant / tool
    Content   string    `json:"content"`
    ToolName  string    `json:"tool,omitempty"`
    Timestamp time.Time `json:"timestamp"`
}// ============================================================
// 情景记忆接口 (Episodic Memory)
// ============================================================type EpisodicStore interface {
    // 写入
    Store(record MemoryRecord) error
    StoreBatch(records []MemoryRecord) error    // 检索(时间+语义混合)
    Search(ctx context.Context, query string, opts SearchOptions) ([]MemoryRecord, error)
    SearchByTime(userID string, since, until time.Time) ([]MemoryRecord, error)    // 管理
    UpdateImportance(id string, importance float64) error
    Delete(id string) error    // 遗忘
    ApplyForgetting(strategy ForgettingStrategy, userID string) (int, error)
}type SearchOptions struct {
    UserID    string
    Limit     int
    // 三要素权重
    Alpha     float64 // semantic weight
    Beta      float64 // recency weight
    Gamma     float64 // importance weight
    // 门控
    MinScore  float64 // 低于此分数的结果不返回(解决冷启动问题)
}// ============================================================
// 语义记忆接口 (Semantic Memory)
// ============================================================type SemanticStore interface {
    // 精确读写(语义记忆应该是结构化的)
    Get(namespace, key string) (string, error)
    Put(namespace, key string, value any) error
    Delete(namespace, key string) error
    List(namespace string) (map[string]any, error)    // 用户画像(最常用的语义记忆子集)
    GetProfile(userID string) (*UserProfile, error)
    UpdatePreference(userID, key, value string) error
}// ============================================================
// 反思机制 (Reflection)
// ============================================================type ReflectionEngine interface {
    // 从情景记忆中提炼语义知识
    Reflect(ctx context.Context, userID string, recent []MemoryRecord) ([]Insight, error)
}type Insight struct {
    Key        string  `json:"key"`
    Value      string  `json:"value"`
    Confidence float64 `json:"confidence"`
    Evidence   []string `json:"evidence"` // 支撑这条 insight 的 episodic record IDs
}// ============================================================
// 前瞻记忆接口 (Prospective Memory)
// ============================================================type ProspectiveStore interface {
    // 目标追踪
    SetGoal(goal Goal) error
    GetActiveGoals(userID string) ([]Goal, error)
    CompleteGoal(goalID string) error    // 中断恢复
    SaveCheckpoint(taskID string, context string) error
    ResumeTask(taskID string) (string, error) // 返回恢复上下文
}

7.2 核心实现:混合检索

 复制代码// 情景记忆的检索实现——这是整个系统最核心的逻辑
func (s *episodicStore) Search(
    ctx context.Context, query string, opts SearchOptions,
) ([]MemoryRecord, error) {
    if opts.Limit == 0 {
        opts.Limit = 5
    }
    if opts.Alpha == 0 && opts.Beta == 0 && opts.Gamma == 0 {
        // 默认权重(Generative Agents 风格)
        opts.Alpha, opts.Beta, opts.Gamma = 1.0, 1.0, 1.0
    }    // Step 1: 语义相似度检索(粗筛)
    queryVec, err := s.embedder.Embed(ctx, query)
    if err != nil {
        return nil, fmt.Errorf("embed query: %w", err)
    }    candidates, err := s.vectorDB.Search(ctx, queryVec, opts.Limit*3) // 过采样
    if err != nil {
        return nil, fmt.Errorf("vector search: %w", err)
    }    // Step 2: 复合评分
    now := time.Now()
    type scored struct {
        record MemoryRecord
        score  RetrievalScore
    }
    var scored []scored    for _, c := range candidates {
        // 语义相似度(从向量搜索中获取)
        semanticScore := c.Score        // 时间衰减:指数衰减,越近越高
        hoursAgo := now.Sub(c.Record.CreatedAt).Hours()
        recencyScore := math.Exp(-0.01 * hoursAgo) // λ=0.01        // 重要性(LLM 评分或默认)
        importanceScore := c.Record.Importance
        if importanceScore == 0 {
            importanceScore = 5.0 // 默认中等重要
        }        composite := opts.Alpha*semanticScore +
            opts.Beta*recencyScore +
            opts.Gamma*importanceScore/10.0        // 门控:低于阈值的直接丢弃
        if composite < opts.MinScore {
            continue
        }        scored = append(scored, scored{c.Record, RetrievalScore{
            SemanticSimilarity: semanticScore,
            Recency:            recencyScore,
            Importance:         importanceScore,
            Composite:          composite,
        }})
    }    // Step 3: 按复合分数排序
    sort.Slice(scored, func(i, j int) bool {
        return scored[i].score.Composite > scored[j].score.Composite
    })    // Step 4: 截断
    if len(scored) > opts.Limit {
        scored = scored[:opts.Limit]
    }    records := make([]MemoryRecord, len(scored))
    for i, s := range scored {
        records[i] = s.record
    }    return records, nil
}

7.3 记忆管理器:聚合所有类型

 复制代码// MemoryManager 是所有记忆类型的聚合入口
type MemoryManager struct {
    embedder    Embedder
    working     WorkingMemory
    episodic    EpisodicStore
    semantic    SemanticStore
    prospective ProspectiveStore
    reflector   ReflectionEngine
    llm         LLM
}func NewMemoryManager(cfg Config) (*MemoryManager, error) {
    embedder, err := NewOpenAIEmbedder(cfg.EmbeddingModel)
    if err != nil {
        return nil, err
    }    return &MemoryManager{
        embedder:    embedder,
        working:     NewSlidingWindow(cfg.MaxWindowTokens),
        episodic:    NewPGVectorStore(cfg.DBConn, embedder),
        semantic:    NewRedisStore(cfg.RedisAddr), // 语义记忆用 Redis 追求低延迟
        prospective: NewInMemoryProspectiveStore(),
        reflector:   NewLLMReflector(cfg.ReflectModel),
        llm:         NewOpenAI(cfg.Model),
    }, nil
}// OnSessionStart 每次新会话开始时调用
func (m *MemoryManager) OnSessionStart(ctx context.Context, userID string) error {
    // 1. 注入语义记忆(常驻)
    profile, err := m.semantic.GetProfile(userID)
    if err != nil {
        return err
    }
    m.working.Inject(formatProfileBlock(profile))    // 2. 检索相关情景记忆(按需)
    recentEpisodes, err := m.episodic.Search(ctx, "recent interactions", SearchOptions{
        UserID:  userID,
        Limit:   3,
        Alpha:   0.5,
        Beta:    0.3,
        Gamma:   0.2,
        MinScore: 0.3, // 门控:太不相关的不要注入
    })
    if err != nil {
        return err
    }
    m.working.InjectEpisodic(recentEpisodes)    return nil
}// OnSessionEnd 每次会话结束时调用
func (m *MemoryManager) OnSessionEnd(session *Session) error {
    record := MemoryRecord{
        Content:    buildSessionSummary(session),
        Importance: m.llm.ScoreImportance(session),
        Embedding:  m.embedder.Embed(ctx, session.Summary),
        CreatedAt:  time.Now(),
        Metadata: map[string]any{
            "user_id":    session.UserID,
            "thread_id":  session.ThreadID,
            "msg_count":  len(session.Messages),
        },
    }
    return m.episodic.Store(record)
}// NightlyReflection 离线反思任务(每天凌晨执行)
func (m *MemoryManager) NightlyReflection(ctx context.Context, userID string) error {
    recent, err := m.episodic.SearchByTime(userID,
        time.Now().Add(-24*time.Hour), time.Now())
    if err != nil || len(recent) < 3 {
        return err
    }    insights, err := m.reflector.Reflect(ctx, userID, recent)
    if err != nil {
        return err
    }    for _, ins := range insights {
        if ins.Confidence < 0.7 {
            continue
        }
        if exists, _ := m.semantic.Get("insights", ins.Key); exists != "" {
            continue // 去重:已有相似 insight
        }
        m.semantic.Put("insights", ins.Key, ins.Value)
    }    // 同时执行遗忘清理
    cleaned, _ := m.episodic.ApplyForgetting(&HardForget{
        ImportanceThreshold: 1.0,
        AgeDaysThreshold:    30,
    }, userID)
    log.Printf("nightly reflection: %d insights, cleaned %d old records", len(insights), cleaned)    return nil
}

7.4 最小可跑示例

 复制代码func main() {
    mgr, _ := memory.NewMemoryManager(memory.Config{
        EmbeddingModel: "text-embedding-3-small",
        Model:          "gpt-4o",
        ReflectModel:   "gpt-4o-mini", // 反思用便宜模型
        DBConn:         os.Getenv("PG_CONN"),
        RedisAddr:      os.Getenv("REDIS_ADDR"),
    })    ctx := context.Background()
    userID := "user-zhangsan"    // 1. 会话开始——注入记忆
    mgr.OnSessionStart(ctx, userID)    // 2. 模拟一轮对话
    session := &memory.Session{
        UserID:   userID,
        ThreadID: "conv-042",
        Messages: []memory.Message{
            {Role: "user", Content: "帮我把登录页的样式改成 Material Design"},
            {Role: "assistant", Content: "好的,我先看一下现在的 login.go..."},
        },
    }    // 3. 会话结束——自动沉淀
    mgr.OnSessionEnd(session)    // 4. 下次会话——检索到上次对话
    records, _ := mgr.episodic.Search(ctx, "登录页面样式修改", memory.SearchOptions{
        UserID: userID, Limit: 3,
    })
    for _, r := range records {
        fmt.Printf("[%s] %s (importance: %.1f)n",
            r.CreatedAt.Format("01-02 15:04"), r.Content, r.Importance)
    }
    // Output:
    // [07-10 15:23] 用户要求用 Material Design 修改登录页面样式 (importance: 7.0)
}

八、总结:双轴设计框架

综合全部分析,Agent 记忆系统的完整设计空间可以归纳为一个双轴框架

分类轴 (What) ——记忆的类型:

类型核心问题存储检索生命周期
Perceptual输入如何编码?—(编码层)实时
Working此刻在思考什么?Context WindowAttention秒级
Episodic过去发生了什么?向量DB + 时间索引时间+语义混合日/周级
Semantic知道什么事实?KV Store (快速读写)常驻或精确匹配长期
Procedural怎么做?System Prompt + Tool Defs始终生效版本级
Prospective将来要做什么?任务队列优先级 + 状态会话/跨会话

管理轴 (How) ——记忆的操作:

 复制代码写入策略              检索策略              遗忘策略           转化策略
    │                    │                    │                 │
自动 vs 主动         注入 vs 按需         软遗忘(衰减)        W→E (自动)
去重 vs 保留         时间 vs 语义         硬遗忘(删除)        E→S (Reflection)
实时 vs 批量         置信度门控           合并(压缩)          S→P (固化)

设计建议速查

你的需求最小方案完整方案
"记一个会话就行"Working(裸 Context Window)+ Checkpointer
"跨会话记住用户"+ Semantic Store+ 每次注入 Profile
"智能召回历史对话"+ Episodic Store+ 时间+语义混合检索 + 置信度门控
"从经验中自我改进"+ Reflection 定时任务+ Memory Blocks 自编辑
"全都要"四类完整实现 + 前瞻记忆+ 记忆生命周期管理 + 遗忘调度

这是一个渐进式的设计路径——不需要一开始就上全部组件。根据任务周期和复杂度,逐步叠加。


参考论文与框架

来源核心贡献
认知科学到 AI Agent 记忆分类的完整映射从认知科学到 AI 的完整理论映射
Generative Agents (Park et al. 2023)Memory Stream + Reflection 机制
MemGPT / Letta (Packer et al. 2023)Agent 自主管理记忆,Memory Blocks
Voyager (Wang et al. 2023)Skill Library——可执行代码作为程序性记忆
Reflexion (Shinn et al. 2023)极简 episodic buffer + verbal reinforcement
EM-LLM (Fountas et al. 2024)Bayesian Surprise 事件分割
LLM Agent Survey (Xi et al. 2023)Memory Operations (Read/Write/Reflect) 框架
相关资讯
点击查看更多
游戏推荐
推荐专题
热门阅读
推荐下载