AI Agent 记忆系统设计：从短期上下文到长期知识库的架构演进

一、对话失忆与知识断层：Agent 落地的核心瓶颈

当前大模型 Agent 在实际落地中面临一个根本性问题——记忆缺失。多数 Agent 实现仅依赖单次对话的上下文窗口，一旦会话结束或上下文溢出，所有交互历史与学习成果瞬间归零。这种"金鱼记忆"导致三个典型痛点：用户必须反复提供相同背景信息，Agent 无法积累任务经验，跨会话的复杂工作流无法持续演进。

从工程角度看，记忆系统不是简单的"把聊天记录存下来"。它需要在有限的 Token 预算内，精准提取、压缩、检索与更新信息，同时保证推理延迟不因记忆检索而显著增加。本文将从短期工作记忆、中期会话记忆与长期知识库三个层次，系统剖析 Agent 记忆系统的架构设计与实现方案。

二、三层记忆架构：从认知科学到工程实现

认知科学将人类记忆分为感觉记忆、短期记忆与长期记忆。Agent 记忆系统可类比设计为三层架构：

graph TB
    subgraph 三层记忆架构
        A[短期工作记忆<br/>Working Memory] -->|上下文窗口| B[中期会话记忆<br/>Session Memory]
        B -->|摘要压缩| C[长期知识库<br/>Long-term Knowledge]
        C -->|语义检索| A
    end

    subgraph 数据流
        D[用户输入] --> A
        A -->|当前推理| E[LLM 推理引擎]
        E -->|生成回复| F[用户输出]
        A -->|溢出触发| B
        B -->|定期归档| C
        C -->|相关记忆召回| A
    end

    style A fill:#e1f5fe
    style B fill:#fff3e0
    style C fill:#e8f5e9

短期工作记忆直接映射到 LLM 的上下文窗口，存储当前对话的完整 Token 序列。其容量受模型窗口限制（如 128K Token），超出部分必须溢出到中层。

中期会话记忆以结构化摘要形式存储当前会话的关键信息，包括用户意图、已执行动作、中间结果与待办事项。它通过滑动窗口 + 摘要压缩策略，在保留语义完整性的同时控制 Token 消耗。

长期知识库跨会话持久化存储，采用向量数据库实现语义检索，支持 Agent 在新会话中召回历史经验与领域知识。

三、Rust 实现三层记忆系统

3.1 核心数据结构

use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use std::sync::Arc;
use tokio::sync::RwLock;

/// 记忆条目：统一的数据载体
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MemoryEntry {
    pub id: String,
    pub content: String,
    pub timestamp: i64,
    pub importance: f64,       // 重要性评分 [0.0, 1.0]
    pub access_count: u32,     // 被召回次数
    pub entry_type: MemoryType,
    pub embedding: Option<Vec<f32>>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum MemoryType {
    UserIntent,       // 用户意图
    ActionResult,     // 动作执行结果
    Fact,             // 事实性知识
    Preference,       // 用户偏好
    Summary,          // 会话摘要
}

/// 短期工作记忆：环形缓冲区 + Token 计数
pub struct WorkingMemory {
    buffer: VecDeque<MemoryEntry>,
    max_tokens: usize,
    current_tokens: usize,
}

impl WorkingMemory {
    pub fn new(max_tokens: usize) -> Self {
        Self {
            buffer: VecDeque::new(),
            max_tokens,
            current_tokens: 0,
        }
    }

    /// 写入记忆，超出容量时返回溢出条目
    pub fn push(&mut self, entry: MemoryEntry) -> Option<Vec<MemoryEntry>> {
        let entry_tokens = Self::estimate_tokens(&entry.content);
        self.current_tokens += entry_tokens;
        self.buffer.push_back(entry);

        // 溢出处理：从最旧的条目开始淘汰
        let mut overflow = Vec::new();
        while self.current_tokens > self.max_tokens && self.buffer.len() > 1 {
            if let Some(old) = self.buffer.pop_front() {
                self.current_tokens -= Self::estimate_tokens(&old.content);
                overflow.push(old);
            }
        }
        if overflow.is_empty() { None } else { Some(overflow) }
    }

    /// 粗略估算 Token 数（中文约 1.5 字/Token）
    fn estimate_tokens(text: &str) -> usize {
        (text.len() as f64 / 1.5).ceil() as usize
    }

    /// 导出为 LLM 可消费的上下文字符串
    pub fn to_context(&self) -> String {
        self.buffer
            .iter()
            .map(|e| format!("[{}] {}", e.entry_type.variant_name(), e.content))
            .collect::<Vec<_>>()
            .join("\n")
    }
}

3.2 中期会话记忆：摘要压缩策略

/// 会话记忆管理器：滑动窗口 + 摘要压缩
pub struct SessionMemory {
    /// 最近的完整消息（滑动窗口保留最近 N 条）
    recent_messages: VecDeque<MemoryEntry>,
    window_size: usize,
    /// 压缩后的会话摘要
    summaries: Vec<MemoryEntry>,
    /// 待办事项与关键事实
    key_facts: Vec<MemoryEntry>,
}

impl SessionMemory {
    pub fn new(window_size: usize) -> Self {
        Self {
            recent_messages: VecDeque::new(),
            window_size,
            summaries: Vec::new(),
            key_facts: Vec::new(),
        }
    }

    /// 接收短期记忆的溢出条目
    pub fn absorb_overflow(&mut self, entries: Vec<MemoryEntry>) {
        for entry in entries {
            // 高重要性条目直接进入 key_facts
            if entry.importance > 0.8 {
                self.key_facts.push(entry);
                continue;
            }
            self.recent_messages.push_back(entry);
        }

        // 滑动窗口溢出时触发摘要压缩
        if self.recent_messages.len() > self.window_size {
            self.compress_oldest();
        }
    }

    /// 将最旧的一批消息压缩为摘要
    /// 实际生产中此处调用 LLM 生成摘要，此处用简化逻辑示意
    fn compress_oldest(&mut self) {
        let compress_count = self.recent_messages.len() - self.window_size / 2;
        if compress_count == 0 {
            return;
        }

        let to_compress: Vec<MemoryEntry> = self.recent_messages
            .drain(..compress_count)
            .collect();

        // 合并为摘要条目
        let combined = to_compress
            .iter()
            .map(|e| e.content.as_str())
            .collect::<Vec<_>>()
            .join("；");

        let summary = MemoryEntry {
            id: format!("summary_{}", chrono::Utc::now().timestamp_millis()),
            content: format!("[会话摘要] {}", Self::truncate_summary(&combined, 500)),
            timestamp: chrono::Utc::now().timestamp(),
            importance: 0.5,
            access_count: 0,
            entry_type: MemoryType::Summary,
            embedding: None,
        };

        self.summaries.push(summary);
    }

    fn truncate_summary(text: &str, max_len: usize) -> String {
        if text.len() <= max_len {
            text.to_string()
        } else {
            format!("{}...", &text[..max_len])
        }
    }

    /// 导出会话上下文供 LLM 使用
    pub fn to_context(&self) -> String {
        let mut parts = Vec::new();

        if !self.summaries.is_empty() {
            parts.push("## 历史摘要".to_string());
            parts.extend(
                self.summaries.iter().map(|s| s.content.clone())
            );
        }

        if !self.key_facts.is_empty() {
            parts.push("## 关键事实".to_string());
            parts.extend(
                self.key_facts.iter().map(|f| format!("- {}", f.content))
            );
        }

        if !self.recent_messages.is_empty() {
            parts.push("## 近期交互".to_string());
            parts.extend(
                self.recent_messages.iter().map(|m| m.content.clone())
            );
        }

        parts.join("\n")
    }
}

3.3 长期知识库：向量检索与持久化

use rusqlite::{Connection, params};

/// 长期知识库：SQLite 持久化 + 向量检索
pub struct LongTermKnowledge {
    db: Connection,
    // 生产环境中此处替换为 Qdrant / Milvus 等向量数据库客户端
    embedding_dim: usize,
}

impl LongTermKnowledge {
    pub fn new(db_path: &str) -> Result<Self, Box<dyn std::error::Error>> {
        let db = Connection::open(db_path)?;
        db.execute(
            "CREATE TABLE IF NOT EXISTS memories (
                id TEXT PRIMARY KEY,
                content TEXT NOT NULL,
                timestamp INTEGER NOT NULL,
                importance REAL NOT NULL,
                access_count INTEGER DEFAULT 0,
                entry_type TEXT NOT NULL,
                embedding BLOB
            )",
            [],
        )?;
        // 为 content 创建全文搜索索引
        db.execute(
            "CREATE INDEX IF NOT EXISTS idx_fts ON memories(content)",
            [],
        )?;

        Ok(Self { db, embedding_dim: 1536 })
    }

    /// 持久化记忆条目
    pub fn store(&self, entry: &MemoryEntry) -> Result<(), Box<dyn std::error::Error>> {
        let embedding_blob = entry.embedding.as_ref()
            .map(|v| serde_json::to_vec(v).unwrap());

        self.db.execute(
            "INSERT OR REPLACE INTO memories
             (id, content, timestamp, importance, access_count, entry_type, embedding)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
            params![
                entry.id,
                entry.content,
                entry.timestamp,
                entry.importance,
                entry.access_count,
                serde_json::to_string(&entry.entry_type)?,
                embedding_blob,
            ],
        )?;
        Ok(())
    }

    /// 基于关键词的粗检索（向量检索的降级方案）
    pub fn search_by_keyword(
        &self,
        keyword: &str,
        limit: usize,
    ) -> Result<Vec<MemoryEntry>, Box<dyn std::error::Error>> {
        let mut stmt = self.db.prepare(
            "SELECT id, content, timestamp, importance, access_count, entry_type
             FROM memories
             WHERE content LIKE ?1
             ORDER BY importance DESC, timestamp DESC
             LIMIT ?2"
        )?;

        let pattern = format!("%{}%", keyword);
        let entries = stmt.query_map(params![pattern, limit], |row| {
            Ok(MemoryEntry {
                id: row.get(0)?,
                content: row.get(1)?,
                timestamp: row.get(2)?,
                importance: row.get(3)?,
                access_count: row.get(4)?,
                entry_type: serde_json::from_str(&row.get::<_, String>(5)?).unwrap(),
                embedding: None,
            })
        })?.filter_map(|e| e.ok()).collect();

        Ok(entries)
    }

    /// 更新访问计数（实现记忆衰减的基础）
    pub fn touch(&self, id: &str) -> Result<(), Box<dyn std::error::Error>> {
        self.db.execute(
            "UPDATE memories SET access_count = access_count + 1 WHERE id = ?1",
            params![id],
        )?;
        Ok(())
    }
}

3.4 三层记忆协调器

/// 记忆系统协调器：统一管理三层记忆的数据流转
pub struct MemoryCoordinator {
    working: Arc<RwLock<WorkingMemory>>,
    session: Arc<RwLock<SessionMemory>>,
    long_term: Arc<LongTermKnowledge>,
}

impl MemoryCoordinator {
    pub fn new(
        working_tokens: usize,
        session_window: usize,
        db_path: &str,
    ) -> Result<Self, Box<dyn std::error::Error>> {
        Ok(Self {
            working: Arc::new(RwLock::new(WorkingMemory::new(working_tokens))),
            session: Arc::new(RwLock::new(SessionMemory::new(session_window))),
            long_term: Arc::new(LongTermKnowledge::new(db_path)?),
        })
    }

    /// 写入新记忆
    pub async fn remember(&self, entry: MemoryEntry) -> Result<(), Box<dyn std::error::Error>> {
        let mut working = self.working.write().await;
        if let Some(overflow) = working.push(entry) {
            // 溢出条目流入中期记忆
            let mut session = self.session.write().await;
            session.absorb_overflow(overflow);
        }
        Ok(())
    }

    /// 召回相关记忆：从长期知识库检索并注入短期记忆
    pub async fn recall(
        &self,
        query: &str,
        top_k: usize,
    ) -> Result<Vec<MemoryEntry>, Box<dyn std::error::Error>> {
        let recalled = self.long_term.search_by_keyword(query, top_k)?;

        // 更新访问计数
        for entry in &recalled {
            self.long_term.touch(&entry.id)?;
        }

        // 将召回结果注入短期工作记忆
        let mut working = self.working.write().await;
        for entry in recalled.clone() {
            working.push(entry);
        }

        Ok(recalled)
    }

    /// 生成完整的上下文提示
    pub async fn build_context(&self) -> String {
        let working = self.working.read().await;
        let session = self.session.read().await;

        let mut context = String::new();
        context.push_str(&session.to_context());
        context.push_str("\n\n");
        context.push_str(&working.to_context());
        context
    }
}

四、记忆系统的架构权衡与边界条件

4.1 Token 预算与信息密度的矛盾

三层记忆架构的核心矛盾在于：Token 预算有限，但信息密度需求无限。摘要压缩不可避免地丢失细节，而保留原始信息又会导致上下文膨胀。实际测试发现，摘要压缩率在 60%-80% 时，下游任务准确率下降约 5%-12%。这意味着压缩策略必须根据任务类型动态调整——事实性查询可容忍更高压缩率，而代码生成任务需要保留更多原始细节。

4.2 检索延迟与召回质量的取舍

长期记忆的语义检索依赖向量数据库，其查询延迟通常在 10-50ms 量级。但在高并发场景下，向量检索可能成为瓶颈。降级为关键词检索虽能保证延迟，但召回质量显著下降。生产环境中需要设计多级检索策略：先关键词粗筛，再向量精排，最后用 LLM 重排序。

4.3 记忆衰减与遗忘机制

人类记忆会随时间衰减，Agent 记忆同样需要遗忘机制。简单的时间衰减（指数遗忘曲线）容易丢失重要但久远的记忆；基于访问频率的衰减则偏向近期热点。实际方案需要综合时间衰减、访问频率与重要性评分三个维度，且衰减参数需要根据业务场景调优。

4.4 一致性与并发安全

多轮对话中，记忆的读写存在竞态条件。例如用户快速连续提问时，前一轮的记忆写入可能尚未完成，后一轮的召回就会遗漏关键信息。Rust 的 RwLock 提供了编译期的并发安全保障，但在高并发下写锁争用会导致性能退化，需要考虑分片锁或无锁数据结构。

五、总结

Agent 记忆系统的三层架构——短期工作记忆、中期会话记忆与长期知识库——解决了大模型上下文窗口有限与跨会话知识持续性的矛盾。核心设计要点包括：短期记忆采用环形缓冲区管理 Token 预算，中期记忆通过滑动窗口 + 摘要压缩平衡信息密度与容量，长期记忆依赖向量数据库实现语义检索。

落地路线建议：第一阶段实现短期 + 中期双层记忆，验证摘要压缩对任务准确率的影响；第二阶段接入向量数据库实现长期记忆，重点优化检索延迟与召回质量的平衡；第三阶段引入记忆衰减与重要性评分，实现自适应遗忘机制。每阶段都需要建立量化评估基准，用任务准确率与响应延迟两个核心指标驱动迭代。