Agent 记忆污染防护:当长期记忆被悄悄写入假知识
Agent 记忆污染防护:当长期记忆被悄悄写入假知识
一、当记忆变成后门:长期记忆为何是 Agent 的软肋
长期记忆让 Agent 跨会话保留经验。它本应提升效率,却也打开一扇隐蔽后门。攻击者不需要每次都注入,只需在记忆里写一句假知识,便能长期生效。
污染常发生在"无感的写入环节"。Agent 自动把对话结论存进记忆库,很少校验内容真伪。一句"管理员授权免验证"被存下,后续所有会话都默认信任它。这种持久化投毒,比单次 prompt 注入更难察觉。
更麻烦的是,记忆会被反复读取并参与推理。一旦假知识进入记忆,它会像种子一样不断影响后续判断。即使当次对话没有攻击,下一次调用仍会取出被污染的条目,造成延迟触发的错误决策。
跨会话污染还能放大影响面。一个被污染的偏好条目,可能同时作用于多个下游任务。攻击者用一次写入,撬动的是 Agent 在整个生命周期里的判断基准。这种"一次写入、长期生效"的特性,让记忆库成为高价值攻击目标。
防御的难点在于:记忆写入频率高、来源杂。用户、工具返回、检索结果都可能写记忆。若对每一条都做严格校验,会严重拖慢 Agent 响应。因此必须在"可写性"与"可信性"之间找到工程化的折中。
还有一层是隐式污染。攻击不直说假知识,而是让 Agent 自己"总结出"错误结论并存入记忆。比如诱导模型把一次偶然成功当成通用规则。这种由模型自发生成的污染,比外部直写更难识别,因为它披着"经验"的外衣。
二、记忆的写入链路与污染传播模型
把记忆系统看成"写入—存储—读取—推理"的完整回路会更清晰。污染可发生在任意环节,并在读取时激活。下图展示典型链路与防护插入点:
flowchart LR A[对话/工具产出] --> B{写入前校验} B -->|可疑| H[隔离待审] B -->|可信| C[记忆存储] C --> D{读取时再校验} D -->|冲突| H D -->|一致| E[注入上下文] E --> F[模型推理] F --> G[决策/动作] G -.-> A H -.-> I[人工复核/自动衰减]写入前校验拦截明显假知识;读取时再校验防止"绕过写入期"的滞后污染。双重闸门降低了单点失效风险。
三、生产级记忆污染检测实现
下面是一段可落地的记忆写入防护中间件。它对写入内容做来源可信度与自洽性校验,并内置超时与重试:
import asyncio import hashlib from dataclasses import dataclass TRUSTED_SOURCES = {"system", "verified_tool"} SENSITIVE_MARKERS = ["授权", "免验证", "管理员", "root", "bypass", "禁用检测"] @dataclass class MemoryItem: content: str source: str sig: str = "" def __post_init__(self): # 用内容+来源生成指纹,便于去重与溯源 self.sig = hashlib.sha256( (self.content + self.source).encode() ).hexdigest()[:16] async def _self_check(item: MemoryItem, timeout: float = 1.0) -> bool: # 调用事实一致性校验模型,判断内容与既有可信记忆是否冲突 try: conflict = await asyncio.wait_for( _consistency_model(item.content), timeout=timeout ) return bool(conflict) except asyncio.TimeoutError: # 超时按"不可信"处理,宁可进待审也不直接入库 return True except Exception: return True async def guard_write(item: MemoryItem, store, retries: int = 2) -> dict: # 来源不可信直接隔离,不进入主记忆 if item.source not in TRUSTED_SOURCES: for attempt in range(retries + 1): try: suspicious = await _self_check(item) break except Exception: if attempt == retries: suspicious = True continue if suspicious: await store.quarantine(item) # 隔离待审 return {"status": "quarantined", "sig": item.sig} await store.put(item) # 可信来源直接入库 return {"status": "stored", "sig": item.sig} async def guard_read(store, query: str, timeout: float = 1.0) -> list: # 读取时二次校验,防止写入期漏检的滞后污染 items = await store.get(query) clean = [] for it in items: if await _self_check(it, timeout=timeout): await store.quarantine(it) # 命中冲突即隔离 else: clean.append(it) return clean关键点:写入前按来源分级,不可信内容做一致性校验;校验超时按不可信降级,进隔离区;读取时二次校验,兜住滞后污染。重试保证校验服务短暂抖动时不丢写。
这段实现的精髓是"双向闸门"。写入期把明显可疑的挡在门外,读取期再扫一遍那些可能绕过写入校验的滞后污染。两者任一生效,都能阻断假知识进入推理上下文。配合指纹去重,同一污染条目不会反复入库占用空间。
四、记忆防护的边界:信任锚、误删与冷启动
这套防护并非无代价,落地时要先想清三件事。
信任锚会成单点。系统把"verified_tool"列为可信来源,若该工具本身被攻破,污染会直接入库。因此可信来源清单必须随供应链安全一起审计,不能静态写死。更稳妥的做法是给来源做签名校验,只有持钥来源写入才免检。
误删会伤体验。一致性校验可能把"新但正确"的知识误判为冲突。解决办法是隔离而非删除:可疑条目进待审区,人工确认后再决定入库或清除。核心权衡是:放宽校验,漏报上升;收紧校验,误删上升,只能按业务风险选默认值。
冷启动缺基准。新建 Agent 没有可信记忆库,自洽性校验缺乏对照,几乎所有写入都可能是"首条"。此时应退化为"来源白名单 + 人工审核"模式,待可信记忆积累到阈值再启用自动校验。否则早期记忆要么全隔离、要么全放行,都不可用。
还有一点:记忆有生命周期。过期记忆长期保留,会累积噪声甚至旧污染。应设置 TTL 与定期再校验,让可疑条目在期限到达时被重新评估,而不是永久沉睡在库里。把记忆当成有时效的资产来运营,才能既保留经验,又清掉毒。
五、总结
Agent 记忆污染的本质,是"持久化写入"绕过了单次对话的边界,让假知识长期参与推理。应对它的不是禁用记忆,而是建立写入前来源分级与读取时自洽校验的双重闸门。工程落地时,必须把校验超时降级、隔离而非误删、信任锚审计与冷启动策略一起考虑,才能让长期记忆既可用,又不被悄悄改写判断基准。