多智能体共享记忆的架构选择与冲突处理
文章区分上下文窗口、消息传递与共享持久状态,分析多智能体因信息遗漏和旧数据覆盖导致的协作失败。内容涵盖五类架构模式、对比表,以及带作用域和冲突检测的最小实现。
AI 深度解读
两个 Agent 单独工作都很出色,放在一起却可能互相拆台:一个记住客户偏好 Telegram,另一个没读到,第三个又用旧信息覆盖了它。原文的核心判断是,多 Agent 协作失败,问题未必出在模型,也可能出在共享记忆的设计。
作者把共享记忆拆成五种架构,并用 Python 示例说明作用域、版本检查和过期时间如何配合。它值得读的地方,是把「让 Agent 记住东西」还原成程序员熟悉的状态管理问题。不过,示例离生产环境仍有距离,尤其不能把键名前缀当成权限控制。
共享记忆为什么不能靠加长上下文解决
共享记忆,是多个 Agent 在直接消息传递之外,都能访问的持久状态。客户偏好、任务决策、可复用产物,都可以属于这层状态。
上下文窗口解决的是一个 Agent 一次能看到多少内容;共享记忆解决的是多个 Agent 能否基于同一份事实行动。窗口再长,也不会自动把另一个 Agent 刚更新的客户偏好同步过来,更不会替你处理覆盖冲突。
共享记忆应该作为独立的状态系统设计。
这里有两个容易混淆的目标:一是让信息能被找到,二是让信息值得相信。把对话塞进向量库,可能改善前者,却没有自动解决来源、有效期和写入权限。
原文也提醒,不要因为 Agent 经常遗忘,就把全部记忆注入每次提示。无关内容越多,读取成本和延迟越高,模型也更难找到需要的事实。作者建议选择性检索、压缩旧记忆,并给注入内容设置严格的 token 预算,但没有提供准确率或成本的对照数据。
五种方案怎么选,为什么通常需要组合
作者列出的五种模式,分别解决不同问题,不能简单理解为五个互相替代的数据库选项。
| 模式 | 适合保存什么 | 主要代价或限制 |
|---|---|---|
| 共享草稿板(blackboard) | 少量 Agent 围绕同一任务的工作信息 | 内容增长后,每个 Agent 都要承担读取成本 |
| 键值状态存储 | 有明确键名和归属的结构化事实 | 模糊知识难以归入固定键 |
| 向量记忆 | 文档、历史对话、非结构化知识 | 相似结果未必是正确答案 |
| 事件日志(event sourcing) | 状态变化、审计记录、失败回放 | 没有物化视图时,读取当前状态需要额外计算 |
| 分层记忆 | Agent 私有、团队共享、组织全局的数据 | 作用域规则不清会让信息难以查找 |
例如,customer:42:preferred_channel 适合放进 Redis、DynamoDB 或 Postgres 表,通过明确的键读取。它需要的是当前准确值,语义相似度帮不上多少忙。
向量记忆适合回答「以前有没有讨论过类似问题」。但「找到相关讨论」和「确认当前事实」是两回事。作者建议把精确事实留在结构化存储中,需要模糊召回时再加入向量检索。
事件日志则保存 TaskCompleted、PreferenceUpdated 这样的不可变事件。当前状态可以通过重放事件推导出来。所谓物化视图,就是提前整理并保存好当前结果,避免每次读取都从头重算。
分层记忆又是另一条维度:它规定谁能看到什么,可以和前面几种存储组合。作者给出的常见搭配是,键值存储管事实,向量记忆管召回,事件日志管审计。
工具方面,原文提到 LangGraph 的检查点状态、CrewAI 的共享 crew 记忆,以及 Letta(原 MemGPT)和 Mem0。但没有展开能力对比,选型时仍要检查它们对作用域、存储后端和控制权的假设。
版本号能拦住什么,示例又漏掉了什么
原文示例使用 Python 3.10+,把作用域键、乐观并发控制和 TTL 放在一起。它解决的是:记录属于哪个范围,写入是否基于旧版本,以及事实何时失效。
import time
from dataclasses import dataclass, field
from typing import Any, Optional
class VersionConflict(Exception):
pass
@dataclass
class MemoryRecord:
value: Any
version: int = 1
author: str = ""
expires_at: Optional[float] = None
updated_at: float = field(default_factory=time.time)
class SharedMemory:
def __init__(self):
self._store: dict[str, MemoryRecord] = {}
@staticmethod
def _key(scope: str, key: str) -> str:
# scope examples: "org", "team:support", "agent:triage-1"
return f"{scope}::{key}"
def read(self, scope: str, key: str) -> Optional[MemoryRecord]:
rec = self._store.get(self._key(scope, key))
if rec and rec.expires_at and rec.expires_at < time.time():
return None # expired: treat as missing
return rec
def write(
self,
scope: str,
key: str,
value: Any,
author: str,
expected_version: Optional[int] = None,
ttl_seconds: Optional[float] = None,
) -> MemoryRecord:
k = self._key(scope, key)
current = self._store.get(k)
# Optimistic concurrency: reject writes based on stale reads
if current and expected_version is not None:
if current.version != expected_version:
raise VersionConflict(
f"{k} is at v{current.version}, "
f"writer expected v{expected_version}"
)
rec = MemoryRecord(
value=value,
version=(current.version + 1) if current else 1,
author=author,
expires_at=(time.time() + ttl_seconds) if ttl_seconds else None,
)
self._store[k] = rec
return rec
要看的是 if current and expected_version is not None::只有调用方传入预期版本,已有记录的版本检查才会执行。省略 expected_version,仍然可以覆盖当前值。
使用时,triage-agent 先在 team:support 下,把 customer:42:channel 写成 telegram;billing-agent 读取后,再带着 expected_version=seen.version 尝试更新为 whatsapp。如果中间有人修改记录,就抛出 VersionConflict,调用方重新读取、协调并重试。
这只能发现「读完之后记录变了」,不能判断 Telegram 和 WhatsApp 哪个才符合客户意愿。冲突后的业务裁决,原文没有实现。
示例还有三个边界。首先,内存字典不能跨进程共享,也不能在重启后保留数据。作者明确要求生产环境使用 Postgres 行版本列,或 Redis WATCH/transactions;迁移时必须让版本比较和写入原子完成。
其次,scope 只是拼进键名。代码没有校验调用者是否有权访问 org、team:support 或 agent:triage-1。
最后,过期记录在读取时返回空,但没有从字典删除,写入时仍能取到旧记录。ttl_seconds=0 也会被当成不设置过期时间。这些都需要在实际接口中明确约定。
我的判断
我认同作者把问题重点放在一致性、权限和可观测性上。相比一开始就引入向量数据库,先把明确事实存好,通常更容易解释系统为什么做出某个决定。但原文没有基准测试、生产规模或故障统计,这是一份设计经验指南,不能据此认定某种组合已经被验证最优。
作者归纳的六类故障也很实用:记忆污染、过时读取、写入冲突、上下文膨胀、跨边界泄漏,以及无法调试。它们需要不同措施,不能靠一个「记忆模块」统一兜底。
错误事实要记录作者和来源,区分「观察到的」与「推断出的」,从私有范围晋升到共享范围前进行验证。易变事实可以增加 last_verified_at,配合 TTL 和版本检查。并发更新可以选择乐观锁、追加日志,或每个键只允许一个写入者。
但示例只保存了 author,没有来源、验证时间、访问审计或真正的权限检查。读写日志还需要记录 Agent ID、时间戳和版本,才能追溯错误从哪里进入系统。
提示词里的访问限制不能代替记忆层的权限校验。
对于只有两三个 Agent 的原型,共享草稿板足够合理;对于已经存在多个租户或团队的产品,作者建议从一开始设计作用域。这里也不必把所有存储模式一次配齐,应按实际的数据和审计需求逐步增加。
能拿来做什么
- 如果正在用多个 AI 编程 Agent 协作,先把当前任务决策和可复用产物放进明确的共享位置,并标出写入者,避免各自依据不同版本继续工作。
- 如果在 Node 服务中实现共享状态,可以先用现有关系表保存结构化事实,给更新接口增加版本条件;发生冲突后重新读取并协调,避免直接覆盖。
- 如果系统需要检索历史讨论或文档,再加入向量召回,并把检索结果与当前有效事实区分开,避免让相似内容直接决定状态更新。
- 如果产品服务多个用户或团队,应在记忆层校验访问范围,同时记录读写的 Agent ID、时间和版本,让跨边界访问与异常决策都有迹可查。