AI Agent 记忆持久化架构：从对话缓存到长期存储实战指南

# 最简单的"记忆"——但这在生产中完全不够
conversation_history = []

def chat(user_message: str) -> str:
    conversation_history.append({"role": "user", "content": user_message})
    response = llm.invoke(conversation_history)
    conversation_history.append({"role": "assistant", "content": response})
    return response

import redis
import json
from datetime import timedelta
from typing import Optional

class AgentCacheLayer:
    """Agent 记忆缓存层 - 管理热数据"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.client = redis.from_url(redis_url, decode_responses=True)
        self.default_ttl = timedelta(hours=2)
    
    def save_conversation_context(
        self, session_id: str, messages: list[dict], ttl: Optional[timedelta] = None
    ) -> None:
        """保存当前对话上下文，带自动过期"""
        key = f"agent:session:{session_id}:context"
        self.client.setex(
            key,
            ttl or self.default_ttl,
            json.dumps(messages, ensure_ascii=False)
        )
    
    def get_conversation_context(self, session_id: str) -> list[dict]:
        """获取对话上下文，未命中返回空列表"""
        key = f"agent:session:{session_id}:context"
        data = self.client.get(key)
        if data is None:
            return []
        return json.loads(data)
    
    def save_tool_state(self, session_id: str, tool_name: str, state: dict) -> None:
        """保存工具执行的中间状态（用于长时间工具调用的恢复）"""
        key = f"agent:session:{session_id}:tool:{tool_name}"
        self.client.setex(key, timedelta(minutes=30), json.dumps(state))
    
    def extend_ttl(self, session_id: str) -> None:
        """用户活跃时延长 TTL"""
        key = f"agent:session:{session_id}:context"
        self.client.expire(key, self.default_ttl)

from sqlalchemy import Column, String, JSON, DateTime, Integer, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime

Base = declarative_base()

class AgentSession(Base):
    __tablename__ = "agent_sessions"
    
    id = Column(String, primary_key=True)
    user_id = Column(String, nullable=False, index=True)
    status = Column(String, default="active")  # active, paused, completed, failed
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    metadata = Column(JSON, default=dict)

class AgentCheckpoint(Base):
    __tablename__ = "agent_checkpoints"
    
    id = Column(String, primary_key=True)
    session_id = Column(String, nullable=False, index=True)
    node_id = Column(String, nullable=False)
    sequence_num = Column(Integer, nullable=False)
    state_snapshot = Column(JSON, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

class UserProfile(Base):
    __tablename__ = "user_profiles"
    
    user_id = Column(String, primary_key=True)
    preferences = Column(JSON, default=dict)
    learned_facts = Column(JSON, default=list)
    interaction_count = Column(Integer, default=0)
    last_active_at = Column(DateTime)

import { QdrantClient } from '@qdrant/js-client-rest';
import { OpenAIEmbeddings } from '@langchain/openai';

interface MemoryEntry {
  id: string;
  content: string;
  sessionId: string;
  userId: string;
  timestamp: number;
  accessCount: number;
}

class SemanticMemoryStore {
  private qdrant: QdrantClient;
  private embeddings: OpenAIEmbeddings;
  private collectionName = 'agent_long_term_memory';

  constructor(qdrantUrl: string) {
    this.qdrant = new QdrantClient({ url: qdrantUrl });
    this.embeddings = new OpenAIEmbeddings({ modelName: 'text-embedding-3-small' });
  }

  async storeMemory(entry: MemoryEntry): Promise<void> {
    const vector = await this.embeddings.embedQuery(entry.content);
    
    await this.qdrant.upsert(this.collectionName, {
      points: [{
        id: entry.id,
        vector,
        payload: {
          content: entry.content,
          session_id: entry.sessionId,
          user_id: entry.userId,
          timestamp: entry.timestamp,
          access_count: entry.accessCount,
        },
      }],
    });
  }

  async retrieveRelevant(
    query: string, userId: string, topK: number = 5
  ): Promise<MemoryEntry[]> {
    const queryVector = await this.embeddings.embedQuery(query);
    
    const results = await this.qdrant.search(this.collectionName, {
      vector: queryVector,
      limit: topK,
      filter: {
        must: [{ key: 'user_id', match: { value: userId } }],
      },
    });

    return results.map(r => ({
      id: String(r.id),
      content: r.payload?.content as string,
      sessionId: r.payload?.session_id as string,
      userId: r.payload?.user_id as string,
      timestamp: r.payload?.timestamp as number,
      accessCount: r.payload?.access_count as number,
    }));
  }
}

from langgraph.graph import StateGraph, MessagesState
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

# 配置 PostgreSQL 连接池作为 Checkpoint 后端
pool = ConnectionPool(
    conninfo="postgresql://user:pass@localhost:5432/agent_db",
    min_size=5,
    max_size=20,
)
checkpointer = PostgresSaver(pool)
checkpointer.setup()  # 创建所需的表结构

# 定义 Agent 图
def reasoning_node(state: MessagesState) -> dict:
    """推理节点 - 每次执行后状态会被自动持久化"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: MessagesState) -> dict:
    """工具执行节点"""
    # 执行工具调用...
    return {"messages": [tool_result]}

# 构建图并绑定 checkpointer
graph = StateGraph(MessagesState)
graph.add_node("reason", reasoning_node)
graph.add_node("tools", tool_node)
graph.add_edge("reason", "tools")
graph.add_edge("tools", "reason")

app = graph.compile(checkpointer=checkpointer)

# 使用 thread_id 标识会话，状态自动持久化
config = {"configurable": {"thread_id": "session-abc-123"}}
result = app.invoke({"messages": [("user", "帮我分析这份数据")]}, config=config)

# 崩溃恢复：使用相同的 thread_id 即可从最近检查点继续
result = app.invoke({"messages": [("user", "继续上次的分析")]}, config=config)

import hashlib
import pickle
from dataclasses import dataclass, field
from typing import Any

@dataclass
class CheckpointData:
    session_id: str
    node_id: str
    sequence: int
    state: dict[str, Any]
    parent_checkpoint_id: str | None = None
    checksum: str = field(init=False)
    
    def __post_init__(self):
        self.checksum = self._compute_checksum()
    
    def _compute_checksum(self) -> str:
        """计算状态校验和，用于验证数据完整性"""
        content = pickle.dumps((self.session_id, self.node_id, self.sequence, self.state))
        return hashlib.sha256(content).hexdigest()
    
    def verify_integrity(self) -> bool:
        """验证检查点数据是否被篡改"""
        return self.checksum == self._compute_checksum()

class CustomCheckpointer:
    def __init__(self, db_session):
        self.db = db_session
    
    async def save(self, checkpoint: CheckpointData) -> str:
        """保存检查点，返回检查点 ID"""
        checkpoint_id = f"{checkpoint.session_id}:{checkpoint.sequence}"
        
        await self.db.execute(
            """
            INSERT INTO agent_checkpoints 
            (id, session_id, node_id, sequence_num, state_snapshot, checksum, parent_id)
            VALUES ($1, $2, $3, $4, $5, $6, $7)
            ON CONFLICT (id) DO UPDATE SET state_snapshot = $5, checksum = $6
            """,
            checkpoint_id,
            checkpoint.session_id,
            checkpoint.node_id,
            checkpoint.sequence,
            checkpoint.state,
            checkpoint.checksum,
            checkpoint.parent_checkpoint_id,
        )
        return checkpoint_id
    
    async def load_latest(self, session_id: str) -> CheckpointData | None:
        """加载最近的检查点"""
        row = await self.db.fetchrow(
            """
            SELECT * FROM agent_checkpoints 
            WHERE session_id = $1 
            ORDER BY sequence_num DESC LIMIT 1
            """,
            session_id,
        )
        if row is None:
            return None
        
        checkpoint = CheckpointData(
            session_id=row["session_id"],
            node_id=row["node_id"],
            sequence=row["sequence_num"],
            state=row["state_snapshot"],
            parent_checkpoint_id=row["parent_id"],
        )
        
        if not checkpoint.verify_integrity():
            raise RuntimeError(f"Checkpoint integrity check failed: {session_id}")
        
        return checkpoint

import json
import time
from pathlib import Path
from typing import Callable

class WriteAheadLog:
    """简化的 WAL 实现 - 生产中建议使用 PostgreSQL 原生 WAL"""
    
    def __init__(self, log_dir: str = "./wal_logs"):
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(parents=True, exist_ok=True)
    
    def log_operation(self, session_id: str, operation: dict) -> str:
        """记录操作到 WAL，返回 LSN (Log Sequence Number)"""
        lsn = f"{session_id}:{int(time.time() * 1000)}"
        entry = {
            "lsn": lsn,
            "session_id": session_id,
            "operation": operation,
            "status": "pending",
            "timestamp": time.time(),
        }
        
        log_file = self.log_dir / f"{session_id}.wal"
        with open(log_file, "a") as f:
            f.write(json.dumps(entry) + "\n")
        
        return lsn
    
    def mark_completed(self, session_id: str, lsn: str) -> None:
        """标记操作已完成"""
        log_file = self.log_dir / f"{session_id}.wal"
        lines = log_file.read_text().strip().split("\n")
        
        updated = []
        for line in lines:
            entry = json.loads(line)
            if entry["lsn"] == lsn:
                entry["status"] = "completed"
            updated.append(json.dumps(entry))
        
        log_file.write_text("\n".join(updated) + "\n")
    
    def recover_pending(self, session_id: str) -> list[dict]:
        """恢复所有未完成的操作"""
        log_file = self.log_dir / f"{session_id}.wal"
        if not log_file.exists():
            return []
        
        pending = []
        for line in log_file.read_text().strip().split("\n"):
            entry = json.loads(line)
            if entry["status"] == "pending":
                pending.append(entry["operation"])
        
        return pending

interface AgentEvent {
  eventId: string;
  sessionId: string;
  type: string;
  payload: Record<string, unknown>;
  timestamp: number;
  version: number;
}

interface AgentState {
  messages: Array<{ role: string; content: string }>;
  toolResults: Record<string, unknown>;
  currentNode: string;
  metadata: Record<string, unknown>;
}

class AgentEventStore {
  private events: Map<string, AgentEvent[]> = new Map();
  
  append(event: AgentEvent): void {
    const sessionEvents = this.events.get(event.sessionId) || [];
    sessionEvents.push(event);
    this.events.set(event.sessionId, sessionEvents);
    
    // 同时持久化到数据库
    this.persistEvent(event);
  }

  rebuildState(sessionId: string, upToVersion?: number): AgentState {
    const sessionEvents = this.events.get(sessionId) || [];
    const relevantEvents = upToVersion
      ? sessionEvents.filter(e => e.version <= upToVersion)
      : sessionEvents;
    
    // 从空状态开始，按顺序应用每个事件
    let state: AgentState = {
      messages: [],
      toolResults: {},
      currentNode: 'start',
      metadata: {},
    };

    for (const event of relevantEvents) {
      state = this.applyEvent(state, event);
    }
    
    return state;
  }

  private applyEvent(state: AgentState, event: AgentEvent): AgentState {
    switch (event.type) {
      case 'MESSAGE_ADDED':
        return {
          ...state,
          messages: [...state.messages, event.payload as any],
        };
      case 'TOOL_EXECUTED':
        return {
          ...state,
          toolResults: {
            ...state.toolResults,
            [event.payload.toolName as string]: event.payload.result,
          },
        };
      case 'NODE_TRANSITION':
        return { ...state, currentNode: event.payload.targetNode as string };
      default:
        return state;
    }
  }

  private async persistEvent(event: AgentEvent): Promise<void> {
    // 写入 PostgreSQL 事件表
    // INSERT INTO agent_events (event_id, session_id, type, payload, timestamp, version) ...
  }
}

import hashlib
from functools import wraps

class IdempotencyStore:
    """幂等性存储 - 防止重复执行"""
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def compute_key(self, session_id: str, operation: str, params: dict) -> str:
        """计算幂等性键"""
        content = f"{session_id}:{operation}:{json.dumps(params, sort_keys=True)}"
        return f"idempotency:{hashlib.md5(content.encode()).hexdigest()}"
    
    def check_and_set(self, key: str, ttl_seconds: int = 3600) -> bool:
        """检查操作是否已执行，如果未执行则标记"""
        return bool(self.redis.set(key, "1", nx=True, ex=ttl_seconds))
    
    def get_cached_result(self, key: str) -> dict | None:
        """获取已缓存的执行结果"""
        result = self.redis.get(f"{key}:result")
        return json.loads(result) if result else None
    
    def cache_result(self, key: str, result: dict, ttl_seconds: int = 3600) -> None:
        """缓存执行结果"""
        self.redis.setex(f"{key}:result", ttl_seconds, json.dumps(result))


def idempotent(operation_name: str):
    """幂等性装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(self, *args, **kwargs):
            store = self.idempotency_store
            key = store.compute_key(
                self.session_id, operation_name, {"args": args, "kwargs": kwargs}
            )
            
            # 检查是否已有缓存结果
            cached = store.get_cached_result(key)
            if cached is not None:
                return cached
            
            # 尝试获取执行锁
            if not store.check_and_set(key):
                # 另一个实例正在执行，等待结果
                return await self._wait_for_result(key)
            
            # 执行操作并缓存结果
            result = await func(self, *args, **kwargs)
            store.cache_result(key, result)
            return result
        
        return wrapper
    return decorator

from dataclasses import dataclass
from enum import Enum

class MemoryScope(Enum):
    SESSION = "session"      # 会话级 - 严格隔离
    USER = "user"            # 用户级 - 跨会话共享
    GLOBAL = "global"        # 全局级 - 所有用户共享（如系统知识）

@dataclass
class ScopedMemoryManager:
    """多层级记忆管理器"""
    
    cache: AgentCacheLayer      # Redis
    db: Any                     # PostgreSQL session
    vector_store: Any           # 向量数据库
    
    async def read(self, session_id: str, user_id: str, query: str) -> dict:
        """按优先级合并多层级记忆"""
        
        # 1. 会话级：当前对话上下文（最高优先级）
        session_context = self.cache.get_conversation_context(session_id)
        
        # 2. 用户级：跨会话的用户知识
        user_memories = await self.vector_store.retrieveRelevant(
            query=query, user_id=user_id, topK=5
        )
        
        # 3. 全局级：系统通用知识
        global_knowledge = await self.vector_store.retrieveRelevant(
            query=query, user_id="__global__", topK=3
        )
        
        return {
            "session_context": session_context,
            "user_memories": user_memories,
            "global_knowledge": global_knowledge,
        }
    
    async def promote_to_user_scope(
        self, session_id: str, user_id: str, knowledge: str
    ) -> None:
        """将会话中学到的知识提升到用户级（异步，不阻塞主流程）"""
        await self.vector_store.storeMemory({
            "id": f"{user_id}:{hash(knowledge)}",
            "content": knowledge,
            "sessionId": session_id,
            "userId": user_id,
            "timestamp": time.time(),
            "accessCount": 0,
        })

interface VersionedMemory {
  id: string;
  content: string;
  version: number;
  updatedAt: number;
}

class OptimisticLockingStore {
  async updateSharedMemory(
    userId: string,
    memoryId: string,
    newContent: string,
    expectedVersion: number
  ): Promise<boolean> {
    // 使用乐观锁：只有版本号匹配时才更新
    const result = await this.db.query(
      `UPDATE user_memories 
       SET content = $1, version = version + 1, updated_at = NOW()
       WHERE user_id = $2 AND id = $3 AND version = $4`,
      [newContent, userId, memoryId, expectedVersion]
    );

    if (result.rowCount === 0) {
      // 版本冲突 - 需要合并策略
      return false;
    }
    return true;
  }

  async mergeConflict(
    existing: VersionedMemory, incoming: string
  ): Promise<string> {
    // 使用 LLM 合并冲突的记忆内容
    const merged = await this.llm.invoke(
      `Merge these two memory entries into one coherent entry:
       Existing: ${existing.content}
       New: ${incoming}
       Output only the merged content.`
    );
    return merged;
  }
}

import math
from datetime import datetime, timedelta

class MemoryDecayPolicy:
    """基于访问频率的记忆衰减策略"""
    
    HALF_LIFE_DAYS = 30  # 30 天不访问，权重衰减一半
    MIN_SCORE = 0.1      # 最低分数阈值，低于此值则淘汰
    
    def compute_score(self, last_accessed: datetime, access_count: int) -> float:
        """计算记忆权重分数"""
        days_since_access = (datetime.utcnow() - last_accessed).days
        decay = math.exp(-0.693 * days_since_access / self.HALF_LIFE_DAYS)
        frequency_boost = math.log(1 + access_count)
        return decay * frequency_boost
    
    def should_evict(self, last_accessed: datetime, access_count: int) -> bool:
        """判断是否应淘汰"""
        return self.compute_score(last_accessed, access_count) < self.MIN_SCORE

import structlog
from prometheus_client import Histogram, Counter, Gauge

logger = structlog.get_logger()

# 关键指标
CHECKPOINT_LATENCY = Histogram(
    "agent_checkpoint_save_seconds", 
    "Checkpoint 保存延迟",
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0]
)
RECOVERY_COUNTER = Counter(
    "agent_recovery_total", 
    "故障恢复次数",
    ["recovery_type"]  # checkpoint, wal, event_replay
)
MEMORY_SIZE_GAUGE = Gauge(
    "agent_memory_entries_total", 
    "记忆条目总数",
    ["scope", "layer"]
)

class MonitoredCheckpointer:
    async def save_checkpoint(self, checkpoint: CheckpointData) -> str:
        with CHECKPOINT_LATENCY.time():
            result = await self.inner.save(checkpoint)
        
        logger.info(
            "checkpoint_saved",
            session_id=checkpoint.session_id,
            node_id=checkpoint.node_id,
            sequence=checkpoint.sequence,
            state_size_bytes=len(json.dumps(checkpoint.state)),
        )
        return result
    
    async def recover(self, session_id: str) -> CheckpointData | None:
        checkpoint = await self.inner.load_latest(session_id)
        if checkpoint:
            RECOVERY_COUNTER.labels(recovery_type="checkpoint").inc()
            logger.info(
                "checkpoint_recovered",
                session_id=session_id,
                recovered_at_sequence=checkpoint.sequence,
            )
        return checkpoint

# 使用 Debezium 或 pg_logical 捕获变更
# 以下为 CDC 消费者示例

async def handle_cdc_event(event: dict) -> None:
    """处理来自 PostgreSQL 的 CDC 事件"""
    table = event["source"]["table"]
    operation = event["op"]  # c=create, u=update, d=delete
    
    if table == "user_profiles" and operation in ("c", "u"):
        after = event["after"]
        # 将新的用户偏好同步到向量数据库
        for fact in after.get("learned_facts", []):
            await vector_store.storeMemory({
                "id": f"profile:{after['user_id']}:{hash(fact)}",
                "content": fact,
                "userId": after["user_id"],
                "sessionId": "__profile_sync__",
                "timestamp": time.time(),
                "accessCount": 1,
            })