共享 memory 被多 Agent 写覆盖

你的 AutoGen 或 CrewAI 多 Agent 系统用一个共享 Redis 存储当「团队 memory」。Agent A 读取研究摘要，追加了 3 条结论，再把合并后的列表写回。几乎同一瞬间，Agent B 也读到了同一份摘要（在 A 写入落地之前），追加了另外 2 条结论，写回自己的版本。Agent A 那 3 条结论被悄悄覆盖。共享存储里现在只剩 Agent B 的 2 条，所有下游 Agent 都在不完整的上下文上工作。这就是经典的「读-改-写」竞态，也叫「丢失更新」（Lost Update），只不过发生在 LLM Agent 的 memory 上。

最快的修法： 不要再对共享 key 做客户端侧的读-改-写。把 get + 改 + set 换成服务端原子操作——追加日志用 RPUSH，改单个字段用 HSET——或者把读和写包进 Redis 的 WATCH/MULTI/EXEC 重试循环里。如果每个 Agent 其实只需要写自己那一块，就给每个 Agent 一个私有 key（agent:{id}:...），最后再合并。本文剩下的部分就是教你判断自己属于哪种情况，以及怎么验证修好了。

你属于哪一类？

症状	最可能的根因	跳到
只追加的列表，最终长度比写入次数还短	客户端 get+追加+set（非原子）	Step 1
整条记录被覆盖，只剩最后那个写者的字段	对 JSON 整块做读-改-写，没有 CAS	Step 2
本该写不同 key 的两个 Agent 撞到了同一个 key	key 由共享属性推导出来	Step 3
LangGraph 并行节点更新同一个 state key，丢了一个	reducer 用错或没设	Step 4
只在高并发下才偶发丢失	版本冲突后用旧数据重试，或 write-behind 刷盘	Step 2 与 Step 5
进程内 dict/list 损坏（TypeError、状态残缺）	临界区没加 asyncio.Lock/threading.Lock	Step 6

常见原因

1. 读-改-写没有原子的「比较并交换」

最常见的模式。Agent 读出完整状态，在 Python 里修改，再整块写回，中间没有任何「状态是否被改过」的检查。任何一对重叠的读-改-写都会丢掉其中一个 Agent 的改动。

怎么判断：找形如 state = store.get(key); state.update(new_data); store.set(key, state) 的代码。没有 if-not-modified-since 守卫，也没有 WATCH/CAS，这种写法必然丢并发写。

2. append 操作在存储层不是原子的

两个 Agent 都调 store.append(key, item)，但底层实现是 store.set(key, store.get(key) + [item])——一个非原子的读-改-写。两者读到同一份列表，各追加一条，各写回一个只多了一条的列表。后写的那次覆盖了前一次。

怎么判断：确认你的存储里 append/add 到底是服务端原子操作（Redis RPUSH），还是客户端 get + 改 + set。客户端方式在并发 Agent 下不安全。

3. 没有写锁——多个 Agent 写同一个 key

并行 Agent 都对同一个命名空间有写权限，没有任何互斥。流水线本以为各 Agent 会写到不同 key，但 key 是由共享的任务属性（任务类别、模型名、日期）推导出来的，于是撞车了。

怎么判断：给每次写入打日志，记录 key、agent ID、时间戳。两个不同的 agent ID 在亚秒级窗口内写同一个 key，就是它的特征。

4. 高负载下乐观锁的版本检查被绕过

每条记录有一个 version 字段。Agent 读到版本 v，算出更新，带 WHERE version = v 条件写回。高负载下写入失败（另一个 Agent 赢了），但错误处理分支拿原始（已经过期的）数据重试，而不是重新读取——于是重试用旧数据覆盖了那个赢了的写入。

怎么判断：检查处理版本冲突 / WatchError 的重试路径。如果它重放的是原始 payload，而不是重新读取再重算，第二次尝试就会破坏状态。

5. LangGraph 的 state reducer 合并方式不对

LangGraph 里并行节点的输出由每个 key 各自的 reducer 来合并。如果两个节点更新同一个 key 却没设 reducer，默认就是「后写覆盖」（last-write-wins），会丢掉一个更新。如果这个 key 存的是消息，却用了 operator.add 而不是 add_messages，可能得到重复消息或 tool-call 配对错乱，而不是正确的去重合并。

怎么判断：逐个检查 state TypedDict 里的 key。凡是被多个并行节点写入的 key 都需要显式 reducer——普通列表用 Annotated[list, operator.add]，聊天消息用 Annotated[list, add_messages]，字典用自定义合并函数。

6. 进程内的数据结构不是并发安全的

跨线程或跨 asyncio task 共享的内存 dict 或 list。在 CPython 里，GIL 让单次 d[k] = v 是原子的，但复合的「读-判断-写」（if k not in d: d[k] = v）不是；而在 asyncio 下，每个 await 都是一个让出点，另一个协程可能在那里插进来。

怎么判断：确认数据结构的类型，找出对它的复合操作是否没有被 threading.Lock（线程）或 asyncio.Lock（协程）包住。

7. write-behind 缓存刷盘时丢掉在途写入

write-behind 缓存每隔几秒刷一次盘。两个 Agent 在同一个刷盘窗口内写入；刷盘只持久化最后那份内存快照，里面可能只有两次写入中的一次。

怎么判断：把缓存的刷盘间隔和并发 Agent 写入的典型间隔对比。如果刷盘间隔更长，写入就可能在刷盘时被丢掉。共享 Agent memory 优先用 write-through。

最短修复路径

Step 1：共享 memory 的写入用服务端原子操作

只追加的数据，往 Redis 列表里 push，别做读-改-写：

import json
import redis

r = redis.Redis()

# 错误——客户端读-改-写，会丢并发追加
def append_finding_unsafe(key: str, finding: str):
    findings = json.loads(r.get(key) or "[]")
    findings.append(finding)
    r.set(key, json.dumps(findings))

# 正确——服务端原子追加，并发下安全
def append_finding_safe(key: str, finding: str):
    r.rpush(key, finding)  # RPUSH 是原子的，没有读-改-写

def get_findings(key: str) -> list:
    return [item.decode() for item in r.lrange(key, 0, -1)]

结构化记录就更新单个 hash 字段，只动被改的那个字段：

# 字段级原子写——其余字段原封不动
r.hset(
    "agent_memory",
    f"finding:{agent_id}:{timestamp}",
    json.dumps(finding_data),
)

RPUSH、HSET、ZADD、SADD 以及 SQL 的 INSERT ... ON CONFLICT 都是单次往返的原子操作。先考虑用它们，再考虑加锁。

Step 2：乐观锁要配正确的重试

确实需要「读出一个值、变换它、再写回」时，把整个过程包进 WATCH/MULTI/EXEC。在 redis-py 里，WATCH 之后命令是立即执行的（所以 get 能拿到值），而 EXEC 会在 key 自 WATCH 后被改过时抛 WatchError：

import logging
import json

logger = logging.getLogger(__name__)

class ConcurrencyError(RuntimeError):
    pass

def update_with_optimistic_lock(key: str, update_fn, max_retries: int = 5):
    for attempt in range(max_retries):
        with r.pipeline() as pipe:
            try:
                pipe.watch(key)                       # 监视并发改动
                current = json.loads(pipe.get(key) or "{}")
                new_state = update_fn(current)        # 基于「当前」状态重算
                pipe.multi()                          # 开启事务
                pipe.set(key, json.dumps(new_state))
                pipe.execute()                        # key 自 WATCH 后被改过则失败
                return new_state
            except redis.WatchError:
                logger.debug("CAS 冲突 %s，第 %d 次重试", key, attempt + 1)
                continue                              # 循环会重新读取——绝不重放旧数据
    raise ConcurrencyError(f"重试 {max_retries} 次后仍无法更新 {key}")

原因 4 的 bug 就是冲突时重放原始 payload。修法是上面的 continue：它回到 watch + 重新读取，让 update_fn 始终基于最新状态运行。redis-py 还内置了 r.transaction(接受 pipe 的函数, key)，能帮你封装好这套 watch-重试的样板代码。

Step 3：按 Agent ID 分区

最稳的修法是直接消除争用。给每个 Agent 一个私有的写命名空间，等 fan-out 完成后由单个 coordinator 统一合并：

def write_agent_memory(agent_id: str, key: str, value: dict):
    # 私有 key——别的 Agent 永远不写这里，所以不需要锁
    r.set(f"agent:{agent_id}:{key}", json.dumps(value))

def read_shared_memory(key: str) -> dict:
    # 共享命名空间对 worker Agent 只读
    return json.loads(r.get(f"shared:{key}") or "{}")

def publish_to_shared(agent_id: str, contribution_key: str, value: dict):
    # 追加到 stream；由单个 coordinator 顺序消费并合并
    r.xadd("shared_memory_stream", {
        "agent_id": agent_id,
        "key": contribution_key,
        "value": json.dumps(value),
    })

XADD 原子地把消息追加进 Redis Stream 并保序，所以 coordinator 能看到每一条贡献、按既定策略解决冲突，而不是靠写入时机碰运气。

Step 4：修好 LangGraph 并行节点的 reducer

给每个被多个并行节点写入的 key 都设显式 reducer。聊天消息用 add_messages（按消息 ID 去重合并，保持 tool-call 配对完整），普通列表用 operator.add：

from typing import Annotated
from typing_extensions import TypedDict
import operator
from langgraph.graph.message import add_messages

def deep_merge(a: dict, b: dict) -> dict:
    result = dict(a)
    for k, v in b.items():
        if k in result and isinstance(result[k], dict) and isinstance(v, dict):
            result[k] = deep_merge(result[k], v)
        else:
            result[k] = v
    return result

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]          # 聊天消息按 ID 去重合并
    findings: Annotated[list[str], operator.add]     # 两个列表拼接
    artifacts: Annotated[dict, deep_merge]           # 字典自定义深合并

让每个并行节点只返回自己的增量（return {"findings": ["finding 3"]}），永远不要返回完整 state——合并交给 reducer。然后显式测它：

def test_parallel_findings_merge():
    a = {"findings": ["finding 1", "finding 2"]}
    b = {"findings": ["finding 3"]}
    merged = operator.add(a["findings"], b["findings"])
    assert len(merged) == 3   # fan-in 后三条都在

Step 5：记录写入并检测冲突

把碰撞变得可观测，这样既能证明修好了，也能抓回归：

import time

def monitored_write(key: str, value, agent_id: str):
    r.set(key, json.dumps(value))
    r.lpush(f"write_log:{key}", json.dumps({
        "agent": agent_id, "ts": time.time(), "size": len(str(value)),
    }))
    recent = r.lrange(f"write_log:{key}", 0, 5)
    agents = {json.loads(e)["agent"] for e in recent}
    if len(agents) > 1:
        logger.warning("检测到 %s 上有并发写入，涉及 agent %s", key, agents)

Step 6：给进程内共享结构加锁

如果共享 memory 是进程内对象而不是 Redis，就把每个复合操作包进锁里，让读-判断-写作为一个临界区整体执行：

import asyncio

class SharedMemory:
    def __init__(self):
        self._data: dict = {}
        self._lock = asyncio.Lock()

    async def append_to_list(self, key: str, item) -> None:
        async with self._lock:               # 没有别的协程能插进来
            self._data.setdefault(key, []).append(item)

    async def atomic_update(self, key: str, updater):
        async with self._lock:
            self._data[key] = updater(self._data.get(key))
            return self._data[key]

如果是 OS 线程并发，用 threading.Lock，写法同样是 with self._lock:。

如何确认修好了

写一个能复现原来丢失的并发测试，然后看着它通过：

import threading

def test_no_lost_appends():
    key = "test:findings"
    r.delete(key)
    writers = [threading.Thread(target=append_finding_safe, args=(key, f"f{i}"))
               for i in range(50)]
    for t in writers: t.start()
    for t in writers: t.join()
    assert len(get_findings(key)) == 50   # 每次写入都在，一条不丢

跑 10 到 20 遍（并发 bug 是概率性的）。不安全的版本会偶尔丢条目；原子版本每次都返回正好 50。Redis 这边还可以 tail write_log:{key}（Step 5），或开启键空间通知（CONFIG SET notify-keyspace-events KEA），在生产里确认你的碰撞窗口内没有两个 agent ID 撞同一个 key。

预防建议

• 所有共享写入优先用服务端原子操作（Redis RPUSH、HSET、ZADD；SQL INSERT ON CONFLICT），而不是客户端读-改-写。
• 把 memory 按 Agent 拆成私有命名空间，再经由 coordinator 合并；只有在无法避免共享某个 key 时才上锁。
• 需要先读后写时用 WATCH/CAS，重试时一定重新读取——绝不重放过期 payload。
• 在 LangGraph 里，给每个被多个并行节点写入的 key 设显式 reducer（消息用 add_messages，列表用 operator.add，字典用自定义合并），并让节点只返回增量。
• 进程内 dict/list 的复合操作用 asyncio.Lock 或 threading.Lock 包住。
• 共享 Agent memory 不要用 write-behind 缓存，用 write-through 保证持久性。
• 明确标注哪些 key 是单写者（私有）、哪些是多写者（共享）；共享 key 必须有显式并发控制。
• 把上面的并发测试放进 CI，确保以后重构不会悄悄把竞态带回来。

常见问答 (FAQ)

Q: Redis 支持完全串行化的多 key 事务吗？ A: 支持，用 MULTI/EXEC 加 WATCH 做乐观锁。单实例 Redis 上这能跨任意 key。但在 Redis Cluster 上，事务里每个 key 必须 hash 到同一个 slot——用 hash tag（{team}:findings、{team}:version）强制把它们放到一起，否则会报 CROSSSLOT 错误。Lua 脚本（EVAL）也有同样的约束：它在服务端原子执行，但所有 KEYS 必须落在同一个 slot。

Q: Agent 之间通信，用消息队列是不是比共享可变 memory 更安全？ A: 通常是。Stream 和队列（Redis Streams、Kafka、SQS）从设计上就串行化写入——每条消息原子追加，消费者读到的是有序日志。对于本质上顺序的 Agent 工作（每个 Agent 贡献一些结论），stream 比可变共享 dict 更安全，也更可审计。

Q: 版本冲突一直在报，直接加个重试行不行？ A: 只有重试时会重新读取才行。原因 4 的 bug 就是拿原始的过期 payload 重试，结果覆盖了赢家。要让重试循环回到 WATCH + 重新读取 + 重算（Step 2）。如果冲突一直不断，说明这个 key 太热了——改成按 Agent 分区（Step 3）或原子追加（Step 1），让写入彻底不再争用。

Q: 状态已经被写坏了，怎么恢复？ A: 用写入日志重建。找到最后一次产生正确状态的写入，识别它之后那些重叠的写入，手动把丢失的更新重新应用一遍。Redis 这边可以开启键空间通知，捕获后续写入的实时日志用于事后分析。

Q: 向量库有没有同样的问题？ A: 有。Pinecone、Qdrant、Chroma 都是 upsert 语义——用同一个向量 ID 并发 upsert 会覆盖掉之前的整条记录（Pinecone 在 ID 重复时替换整条记录）。给每次写入用唯一 ID（agent_id + 时间戳 + 内容 hash），别给一个主题固定一个 ID。这样就把「破坏性覆盖」变成了「追加 + 去重」，而后者完全在你掌控之中。

相关阅读

• 两个并行 Agent 同时改一份文件
• Agent 编排器互相等待出现死锁
• Agent 交接时上下文丢失了
• Checkpoint 恢复出来的状态是损坏的
• 重启后 Agent 状态对不上
• 多 Agent 冲突

标签: #AI 编程 #Agents #排查

相关文章

常见问题解决库

Agent 预算在任务中途被吃光

Agent 执行长任务时 token 或费用预算耗尽，任务半途中止、半成品代码留在磁盘上。本文教你定位预算去向、抢救中间结果，并从断点续跑。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Checkpoint 恢复出来的状态是损坏的

Agent 从 checkpoint 恢复后状态字段缺失、类型错乱或值异常。本文教你定位损坏、回滚到完好的历史版本，并写出真正抗崩溃的 checkpoint。

2026/05/25 #AI 编程 #Agents

常见问题解决库

成本统计漏算了子 Agent 用量

成本仪表盘显示 2 美元，账单却是 18 美元。子 Agent 的 token 消耗没有归因到父级运行。本文给出补齐缺口的方法，2026 年 6 月核实。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 调用图出现循环但没人发现

多 Agent 之间互相交接，形成永不终止的环，因为没有任何东西在检测循环。本文教你抓出这个环、配好框架自带的上限、并把循环彻底限定住。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 交接时上下文丢失了

下游 Agent 重复询问已答过的问题，或输出与前几步的决策相矛盾。本文帮你定位丢失上下文的交接边界，并在一小时内接上结构化、可持久的状态传递。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 编排器互相等待出现死锁

两个 Agent 互相等待对方的输出，整条流水线永远挂起。本文教你找到环、加超时、几分钟内打破死锁，并给出 LangGraph、Temporal、AutoGen 的具体做法。

2026/05/25 #AI 编程 #Agents