5.5. 记忆清理与遗忘是合规功能的另一面

记忆不应是 Agent 的“永生诅咒”。当用户明确说“忘掉刚才那段对话”，或者法律要求彻底擦除个人数据时，你的系统如果只是假装看不见，那它就不是工具，而是一个定时炸弹。本章我们从工程角度实现一套完整的记忆删除通路：自动过期、用户驱动的精确删除，以及让人放心的删除审计。

你需要什么

• Python 3.10+ 环境
• pip install langchain langchain-core langchain-community chromadb（截至当前教材版本，LangChain 0.2.x 可用）
• 一个允许你对文件系统/数据库执行删除操作的测试目录或临时数据库
• 预计用时：30 分钟

最终成果
你会得到一个带分层遗忘能力的 Agent 记忆模块：

• 支持按时间窗口自动过期，旧消息不再占用上下文；
• 支持用户自然语言指令“忘掉上次的对话”，联动存储层精确删除；
• 提供一份遗忘审计脚本，验证向量库、数据库和备份中是否还有残留数据。

这不仅是工程优化，更是将来你应对 GDPR“被遗忘权”合规审查时最直接的证据。

步骤一：理解 LangChain 的记忆存储基座

LangChain 的记忆模块围绕 BaseChatMessageHistory 构建。所有历史消息都存放在这个接口背后，它的 messages 属性返回消息列表，而 add_message 和 clear 方法则是我们操作遗忘的抓手。

我们先快速创建一条临时记忆，看看它的行为：

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

store = InMemoryChatMessageHistory()
store.add_message(HumanMessage(content="你好，我叫小明，我的工号是 8234。"))
store.add_message(AIMessage(content="你好小明，工号 8234 已记录。请问有什么可以帮您？"))
store.add_message(HumanMessage(content="帮我查一下我的公积金。"))
store.add_message(AIMessage(content="好的，正在查询公积金账户..."))

# 查看当前消息
print(f"当前消息数：{len(store.messages)}")  # 输出：4

默认的 InMemoryChatMessageHistory 没有过期机制，也不会响应“忘掉”指令。但它给我们提供了一个统一的接口，这正是实现遗忘治理的锚点。

踩坑经验
在生产系统中千万别只用 InMemoryChatMessageHistory。进程重启后记忆全部丢失不说，遗忘审计也会变得毫无意义。一切删除都必须落库，做到持久化可追溯。接下来的步骤我们都会基于自定义的可持久化历史记录实现。

步骤二：基于时间窗口的自动过期

很多合规场景并不需要记住用户三年前的聊天记录。我们可以为不同敏感度的记忆设定不同的生存时间（TTL），到期自动让旧消息进入“遗忘状态”。

下面的实现用一个模拟的数据库字典保存会话消息，每条消息都带上时间戳，并提供 purge_expired 方法清理过期消息。

import time
from datetime import datetime, timedelta
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

class TTLMessageHistory(BaseChatMessageHistory):
    """带时间窗口的记忆存储，超时消息会被自动清理。"""

    def __init__(self, session_id: str, ttl_seconds: int = 3600):
        self.session_id = session_id
        self.ttl_seconds = ttl_seconds
        # 模拟持久化存储：{ session_id: [ (timestamp, message), ... ] }
        # 在生产中请替换为数据库表
        self._storage = {}

    @property
    def messages(self) -> list[BaseMessage]:
        """返回当前未过期的消息列表。"""
        self.purge_expired()  # 每次读取前自动清理过期消息
        return [msg for ts, msg in self._storage.get(self.session_id, [])]

    def add_message(self, message: BaseMessage) -> None:
        """添加消息，同时记录当前时间戳。"""
        if self.session_id not in self._storage:
            self._storage[self.session_id] = []
        self._storage[self.session_id].append((time.time(), message))

    def purge_expired(self) -> int:
        """清除所有超过 TTL 的消息，返回被删除的数量。"""
        if self.session_id not in self._storage:
            return 0
        cutoff = time.time() - self.ttl_seconds
        original_len = len(self._storage[self.session_id])
        self._storage[self.session_id] = [
            (ts, msg) for ts, msg in self._storage[self.session_id] if ts > cutoff
        ]
        return original_len - len(self._storage[self.session_id])

    def clear(self) -> None:
        """立即清空当前会话的所有消息。"""
        self._storage.pop(self.session_id, None)

# 测试自动过期
store = TTLMessageHistory(session_id="user_42", ttl_seconds=2)
store.add_message(HumanMessage(content="我的交易密码是 123456。"))
time.sleep(3)
print(f"过期后消息数：{len(store.messages)}")  # 期望输出：0

预期结果：等待 3 秒后，store.messages 为空，表明超时消息已被静默清除。如果你查看 _storage，会发现记录还在，但读取时自动过滤，这是为了给审计留下“删除标记”的空间——后面我们会完善这一点。

设计 TTL 策略的小清单

• 金融/医疗等敏感数据：TTL 设为几分钟，甚至不允许长期记忆。
• 一般对话偏好：保存 30 天，辅助个性化体验。
• 为了调试保留的日志：到期自动归档而非物理删除（但归档同样需要能被审计遗忘）。

步骤三：用户驱动的选择性删除

用户说“忘掉上次的对话”，我们需要听懂、找到对应消息，然后从所有存储层中彻底擦除。

这里我们模拟一个简单的删除工作流，它解读自然语言指令，调用 history 的分层删除方法。

from typing import Optional

class AgentMemoryManager:
    """管理 Agent 的多层记忆：短期历史、向量摘要库、备份日志。"""

    def __init__(self, short_term: TTLMessageHistory):
        self.short_term = short_term
        # 长期记忆通常以向量形式存放在向量库中，这里用字典模拟
        self.vector_store = {"user_42": ["小明 工号8234", "偏好：只查看余额不交易"]}
        # 模拟备份表
        self.backup_log = []

    def forget_last_conversation(self) -> str:
        """执行‘忘掉上次对话’的指令。"""
        # 找到最近一轮完整的 Human-AI 对话对
        msgs = self.short_term.messages
        removed = []
        # 简单策略：从末尾向前找到最后一个 HumanMessage，然后删除它及其后面的所有消息
        for i in range(len(msgs)-1, -1, -1):
            if isinstance(msgs[i], HumanMessage):
                removed = msgs[i:]
                # 修改底层存储：重建存储列表，排除被删除的片段
                self._truncate_messages(i)
                break
        if not removed:
            return "没有找到可删除的对话。"
        # 联动向量库
        self._remove_from_vectors(removed)
        # 联动备份
        self.backup_log.append(f"DELETED at {datetime.now()}: {[m.content[:20] for m in removed]}")
        return f"已删除最近一轮对话，涉及 {len(removed)} 条消息。"

    def _truncate_messages(self, index: int):
        """保留索引之前的所有消息，删除 index 及之后的消息。"""
        session = self.short_term._storage.get(self.short_term.session_id, [])
        self.short_term._storage[self.short_term.session_id] = session[:index]

    def _remove_from_vectors(self, messages: list[BaseMessage]):
        """从向量库中删除相关记录。演示用简单内容匹配。"""
        # 实际需调用向量库的 delete 接口，根据 metadata 中的 message_id 删除
        user_key = self.short_term.session_id
        if user_key in self.vector_store:
            contents = [m.content for m in messages]
            self.vector_store[user_key] = [
                doc for doc in self.vector_store[user_key]
                if not any(content in doc for content in contents)
            ]

# 使用示例
history = TTLMessageHistory(session_id="user_42", ttl_seconds=3600)
history.add_message(HumanMessage(content="我换工作了，新公司是数据动力。"))
history.add_message(AIMessage(content="已更新您的公司信息为数据动力。"))
history.add_message(HumanMessage(content="忘掉刚才那段对话。"))

manager = AgentMemoryManager(history)
print(manager.forget_last_conversation())
# 预期输出：已删除最近一轮对话，涉及 2 条消息。
print(f"剩余消息数：{len(history.messages)}")  # 0
print(f"向量库残留：{manager.vector_store['user_42']}")  # []

预期结果：执行“忘掉”指令后，短期历史、向量库中的对应数据都被清除，备份日志里留下了一条删除记录（但不保存原始内容）以保证可审计。

踩坑经验
真实系统中，用户说“忘掉上次对话”可能要结合对话结构识别。如果一轮对话被打断又继续，你需要在消息的 additional_kwargs 里记录 conversation_id 或者使用 response_metadata 中的 thread_id，这样才能精准定位一段完整对话，而不是机械地按最近一条人类消息截断。
另外，向量库的删除非常依赖 metadata 过滤，务必在设计记忆摘要存储时，就给每条摘要打上对应的 session_id 和 message_id。

步骤四：遗忘审计——验证删除是否彻底

合规人员不会只看你的承诺，他们会抽查实际存储。遗忘审计脚本正是我们主动自证清白的工具。

下面我们创建一个审计函数，它会扫描向量库、数据库和备份，确认指定会话的敏感信息是否真的消失了。

def audit_deletion(session_id: str, vector_store: dict, db_history: dict, backup_log: list, sensitive_keyword: str):
    """
    审计遗忘是否彻底。
    返回 (残留存在, 详情) ，残留存在为 True 表示发现问题。
    """
    report = {"vector": [], "db": [], "backup": []}

    # 1. 审计向量库
    if session_id in vector_store:
        matches = [doc for doc in vector_store[session_id] if sensitive_keyword in doc]
        if matches:
            report["vector"] = matches

    # 2. 审计短期数据库
    if session_id in db_history:
        # db_history 即 TTLMessageHistory 内部的 _storage
        messages = [msg for ts, msg in db_history[session_id] if sensitive_keyword in str(msg.content)]
        if messages:
            report["db"] = [m.content for m in messages]

    # 3. 审计备份日志
    for entry in backup_log:
        if session_id in entry and sensitive_keyword in entry:
            report["backup"].append(entry)

    residual_found = any(report.values())
    return residual_found, report

# 执行审计
residual, details = audit_deletion(
    session_id="user_42",
    vector_store=manager.vector_store,
    db_history=history._storage,
    backup_log=manager.backup_log,
    sensitive_keyword="数据动力"  # 用户要求遗忘的公司名
)
if not residual:
    print("审计通过：所有存储层均未发现敏感信息残留。")
else:
    print(f"审计失败！残留信息如下：\n{details}")

预期结果：如果步骤三执行成功，residual 应为 False。若审计失败，details 会精确指出哪一层还有残留，帮你快速定位是向量删除未及时生效，还是备份日志记录了原文（合规备份只应记录操作类型和时间，而非内容）。

自动化遗忘审计流水线建议

回顾

你刚刚完成了一套从自动过期、用户指令删除到合规审计的完整记忆清理方案：

• 分钟级：用 TTLMessageHistory 实现了基于时间窗口的自动过期；
• 场景驱动：通过 AgentMemoryManager 将自然语言“忘掉”指令转化为对短期历史、向量库的精确删除；
• 可证明：用 audit_deletion 脚本逐层核查，给出审计报告。

总耗时约 30 分钟。这正是从“能记住”到“敢忘记”的跨越——只有敢忘记的系统，才值得用户把记忆交给你。

行动清单

1. 在你的记忆管理层中，为所有消息打上时间戳和 session_id，这是实现遗忘的基础设施。
2. 基于消息敏感度，为不同会话设置差异化的 TTL，并在每次读取时执行自动清理。
3. 为“忘掉上次对话”这类指令设计一条清晰的执行通路：解析意图 → 定位消息边界 → 调用存储层删除。
4. 向量库删除务必依赖 metadata 过滤，不要在应用层做“读取-对比-删除”的低效操作。
5. 立即写一个审计脚本，并入 CI/CD，确保每次删除请求后自动触发核查。

下一章我们将直面一个更棘手的工程难题：《异步架构下的记忆一致性是硬骨头》——在高并发智能体服务中，一个错误的锁顺序就能让记忆陷入永久的不一致。