👁 99 次阅读

crewAI多智能体共享Memory如何防止跨Agent污染？

来源：https://github.com/crewAIInc/crewAI/issues/6043

现象描述

在多智能体 Crew 架构中，agent 之间通过共享 memory（short-term / long-term / entity 三类）实现上下文传递。当一个 agent 被 prompt injection 攻陷或输出被操纵时，其写入共享 memory 的内容会污染 crew 中所有其他 agent。单个 researcher agent 的产出污染，会让 writer 和 reviewer 永久受影响。污染具有跨会话持久性——long-term memory 中的恶意内容会在后续会话中持续影响所有 agent 的决策。

背景原因

单 agent 系统中，memory 污染的影响范围仅限于该 agent 自身。多 agent crew 中，一条污染记录会被所有 agent 读取，形成放大效应。crewAI 的共享 memory 机制设计初衷是提升协作效率，但这恰恰构成了攻击面：

short-term memory：会话内跨 agent 传递，污染影响当前会话所有 agent
long-term memory：跨会话持久化，污染影响后续所有会话
entity memory：结构化实体存储，污染可能影响实体关联的所有 agent

该风险对应 OWASP 安全框架中的 ASI-06（Memory Poisoning），属于 Agentic Security Initiative 项目定义的风险类别（https://owasp.org/www-project-agentic-security-initiative/

排查与量化方法

当前 crewAI 尚未内置 memory 写入校验机制。排查 memory 污染问题可从以下维度入手： 1. 审计写入路径：检查所有 agent 的 output 是否直接进入共享 memory 而未经过滤 2. 检查 long-term memory 持久化内容：查看已存储的 memory 条目中是否存在异常模式（如指令性文本、角色切换语句） 3. 监控 entity memory 关联关系：确认实体字段中是否包含非预期内容 4. 对比 agent 行为变化：观察污染写入后，其他 agent 的输出是否出现异常模式

完整解决方案

提案方案：Crew 级 memory_guard 参数

issue #6043 提案者建议在 Crew 上新增 memory_guard 参数，在所有 memory 写入持久化之前做校验。提案者声称已有可用原型并愿意提 PR，但截至当前 issue 仍处于 open 状态，该参数尚未被官方接受或合入主干。提案给出的代码示例如下：

from crewai import Crew, Agent, Task
from agent_memory_guard import scan_text

def memory_validator(content: str) -> bool:
    result = scan_text(content)
    return result.risk_level != "critical"

crew = Crew(agents=[researcher, writer, reviewer], tasks=[...],
            memory=True, memory_guard=memory_validator)

逐行解释：

from agent_memory_guard import scan_text：引入 OWASP 项目 agent-memory-guard 库的扫描函数，该库宣称 59µs 延迟、零依赖
def memory_validator(content: str) -> bool：定义校验函数，接收待写入的文本内容，返回布尔值
result = scan_text(content)：调用扫描函数，返回带 risk_level 属性的结果对象
return result.risk_level != "critical"：当风险等级为 critical 时返回 False，拒绝写入
memory_guard=memory_validator：将校验函数挂载到 Crew 实例，拦截所有 memory 类型（short-term / long-term / entity）的写入

该 guard 会阻断匹配 injection 模式的内容，防止其进入共享 memory。

当前可落地的替代做法

由于 memory_guard 尚未合入，以下方案可在当前版本中实施： 方案 A：自建 memory 层 用向量库或自定义 store 替换 crewAI 内置 memory，在 store 的 add 方法中自行做扫描。这需要对 crewAI 的 memory 实现有一定了解，但提供了最大的控制粒度。 方案 B：外部 prompt-injection detector 做写前钩子 在 agent 输出进入 memory 之前，通过外部检测器（如 agent-memory-guard 或其他 prompt injection 检测工具）进行拦截。 方案 C：embedding 距离异常检测 对写入内容做 embedding，与已有语料库计算相似度，超过阈值时告警或拒绝写入。这能捕获语义层面的异常，但需要维护参考语料库。 方案 D：缩小攻击面 只保留 short-term memory，关闭 long-term 和 entity 共享。这牺牲了跨会话记忆能力，但显著降低了持久化污染风险。

最小可运行 demo（monkeypatch 方式）

以下 demo 展示如何通过 monkeypatch 将 memory_validator 挂到写入路径上：

import crewai
from agent_memory_guard import scan_text

def memory_validator(content: str) -> bool:
    result = scan_text(content)
    return result.risk_level != "critical"

# 注意：crewAI 内部 memory 实现细节可能随版本变化
# 以下代码为示意性演示，需根据实际安装的 crewAI 版本调整导入路径
# 核心思路：在 memory 写入方法前插入校验钩子

original_save = crewai.memory.Memory.save

def guarded_save(self, content, **kwargs):
    if not memory_validator(content):
        print(f"[BLOCKED] Memory write rejected: {content[:80]}...")
        return
    return original_save(self, content, **kwargs)

crewai.memory.Memory.save = guarded_save

上述代码通过替换 Memory 类的 save 方法实现写前拦截。实际部署时需确认 crewAI 当前版本的 memory 类路径和方法签名，因为内部实现可能在不同版本间发生变化。

方案对比

方案	拦截时机	覆盖范围	局限性
Crew 级 memory_guard（提案中）	写前拦截	所有 memory 类型	尚未合入，需等待 PR
每 agent 做输入校验（已考虑但不选）	输入阶段	仅 agent 输入	抓不住工具输出和 agent 间消息流入 memory 的情况
事后清理 memory（已考虑但不选）	写入后	全部	被动，发现问题时损害已经造成
禁用共享 memory（已考虑但不选）	架构层面	全部	违背多智能体协作的初衷
自建 memory 层 + 扫描	写前拦截	可定制	需要自行维护，耦合度高
embedding 异常检测	写前/写后	语义层面	需维护参考语料库，有假阳性

常见错误排查

1. 误以为 short-term memory 也持久化：short-term memory 通常仅在当前会话内有效，不会跨会话持久化。但 long-term memory 和 entity memory 是持久化的，污染具有跨会话影响。 2. 把 memory_guard 当成运行时拦截：memory_guard 是写前拦截机制，只能阻止新的恶意内容写入。已经写入的旧内容不会自动清理，需要额外的清理流程。 3. risk_level 阈值过严导致正常业务内容被拒：将 risk_level 阈值设为过低（如 medium）会导致正常业务文本被误判为风险内容。建议从 critical 开始，根据实际误判率逐步调整。 4. entity memory 中结构化实体字段无法用纯文本扫描：entity memory 存储的是结构化实体数据，纯文本扫描可能无法覆盖所有字段。需要对结构化字段做逐字段扫描或采用专门的实体校验逻辑。 5. 跨 crew 复用 memory 时 guard 未继承：如果多个 crew 实例共享同一 memory 存储，仅在其中一个 crew 上设置 memory_guard 不会保护其他 crew 的写入路径。需要在每个 crew 实例上独立设置 guard，或在底层 store 层面统一拦截。

实践建议

当前阶段，优先采用方案 A（自建 memory 层）或方案 D（缩小攻击面），在 memory_guard 合入前降低风险
部署 agent-memory-guard 作为写前钩子时，注意其 59µs 延迟对高吞吐场景的影响
定期审计 long-term memory 和 entity memory 中的已存储内容，清理可疑条目
关注 issue #6043 的进展，该提案者声称已有原型并愿意提 PR，但目前 issue 中未附 PR 链接
该 feature-request 当前状态为 open，标签为 feature-request，评论数 27，官方尚未接受 memory_guard 参数

FAQ

Q: memory_guard 参数现在可以用吗？ A: 不可以。memory_guard 是 issue #6043 中提出的 feature-request，当前状态为 open，官方尚未接受该参数或合入主干。提案者声称已有可用原型并愿意提 PR，但 issue 中未附 PR 链接。 Q: agent-memory-guard 库是什么？ A: agent-memory-guard 是 OWASP 项目下的第三方库，PyPI 地址为 https://pypi.org/project/agent-memory-guard/ ，GitHub 地址为 https://github.com/OWASP/www-project-agent-memory-guard 。该库宣称 59µs 延迟、零依赖，提供 scan_text 函数用于检测 prompt injection 模式。 Q: 不共享 memory 可以吗？ A: 可以缩小攻击面，但违背多智能体协作的初衷。issue 提案者将此列为已考虑但不选的替代方案。更推荐的做法是在共享 memory 的基础上增加写前校验。 Q: 如何检测已有的 memory 污染？ A: 当前 crewAI 未提供内置的 memory 审计工具。可通过以下方式排查：导出 memory 存储内容，用 agent-memory-guard 的 scan_text 逐条扫描；或对 memory 内容做 embedding 距离分析，识别与正常语料库差异过大的条目。 Q: memory_guard 能防止所有类型的攻击吗？ A: 不能。memory_guard 基于模式匹配检测 injection payload，对于语义层面的间接操纵（如通过正常表述传递误导性信息）可能无法识别。建议结合 embedding 异常检测和人工审计形成多层防护。