分布式智能体需要全局一致的上下文视图

时间线锚点：2026年，LangGraph、ScyllaDB等框架已普遍支持将智能体状态外部化至分布式存储。但当多实例共享同一智能体时，一个根本问题浮现——如何让所有副本同时看到“同一个过去”？

让我们先回到一个真实的故障现场。

某客服智能体部署了 3 个 Kubernetes Pod 以应对高并发。用户 A 的首次请求落在 Pod-1，询问“我的订单 8823 到哪里了？”；第二次请求因负载均衡落到 Pod-2，继续追问“那它能改地址吗？”。Pod-2 的内存中没有前一条消息的历史，智能体重新问了一遍订单号，用户感到困惑和愤怒。这不是模型能力问题，而是上下文视图在分布式边界上发生了断裂。

核心结论就一句话：当多个服务实例共享一个逻辑上的智能体时，它们的上下文视图必须全局一致，否则智能体将表现出不可预测的“失忆症”——这不是模型的幻觉，是架构的幻觉。

你上一章刚实现的 MemoryFactory 利用 ContextVar 做请求级隔离，但隔离不等于同步。本章我们将这条隔离边界扩展为一条分布式一致性协议，让历史消息成为一个全局可访问、可拼合、可互斥更新的共享事实。

问题拆解：为什么上下文在分布式架构中天然不一致？

把单机 Agent 拆成多实例后，消息历史本质上变成了一个 分布式共享变量。根据 CAP 定理，我们必须在一致性、可用性和分区容忍性之间做取舍。在智能体的场景里，这意味着：

作者的结论：在没有外部协调层的多实例部署中，智能体的“记忆”只是一个局部幻想。要获得全局一致的上下文视图，必须让消息历史具备三个特性：持久化、可版本化和可并发控制。

使用 Redis 作为共享记忆后端

解决不一致的第一步，是把消息历史从进程内存中搬到一个所有实例都能访问的位置。Redis 的 Hash 数据结构天然适合保存会话历史：以 session_id 为 Key，内部字段存储序号化消息。

基础实现：读写 RedisHash 的历史同步

import json
import redis.asyncio as aioredis
from typing import List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

class RedisMemoryBackend:
    """基于 RedisHash 的分布式消息历史存储"""

    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = aioredis.from_url(redis_url)

    async def add_message(self, session_id: str, msg: BaseMessage) -> None:
        key = f"session:{session_id}:history"
        seq = await self.redis.hlen(key)  # 字段数即序号
        payload = json.dumps({"type": msg.type, "content": msg.content})
        await self.redis.hset(key, str(seq), payload)

    async def get_history(self, session_id: str) -> List[BaseMessage]:
        key = f"session:{session_id}:history"
        raw = await self.redis.hgetall(key)
        # 按序号排序后反序列化
        items = sorted(raw.items(), key=lambda x: int(x[0]))
        msgs = []
        for _, v in items:
            data = json.loads(v)
            if data["type"] == "human":
                msgs.append(HumanMessage(content=data["content"]))
            elif data["type"] == "ai":
                msgs.append(AIMessage(content=data["content"]))
        return msgs

吞吐量简评（基于本地 Redis Stack 7.2 单实例测试）：

截至当前调研资料，这种基于 Redis 的共享历史模式在轻量级多实例场景（< 10 Pod）中足够稳定。但当并发写密集时，会暴露一个新问题：最终一致性与智能体幻觉。

最终一致性与智能体幻觉

假设 Pod-1 刚刚写入用户消息“我改地址为上海市静安区”，而 Pod-2 在同一时刻读取这条会话的历史来回答“配送时间是多少？”。由于 Redis 主从复制的异步特性，或者仅仅是因为 Pod-2 尚未刷新缓存，Pod-2 读到的历史没有那条新地址。

结果：智能体可能用旧地址继续计算配送时间，给出一个错误答案——这就是由上下文不一致引发的智能体幻觉。

冲突的场景推演

用版本号解决读己之写

解决问题的关键在于让读取方知道“我看到的是不是最新的”。我们在每个 session 的 Hash 中增加一个 __version 字段：

class VersionedRedisMemoryBackend(RedisMemoryBackend):
    async def add_message(self, session_id: str, msg: BaseMessage) -> int:
        key = f"session:{session_id}:history"
        # 原子递增版本号
        new_version = await self.redis.hincrby(key, "__version", 1)
        seq = await self.redis.hlen(key) - 1  # 减掉 __version 自身
        payload = json.dumps({"type": msg.type, "content": msg.content, "v": new_version})
        await self.redis.hset(key, str(seq), payload)
        return new_version

    async def get_history_with_version(self, session_id: str):
        key = f"session:{session_id}:history"
        raw = await self.redis.hgetall(key)
        version = int(raw.pop(b"__version", 0))
        # ... 反序列化同前，返回 (messages, version)
        return msgs, version

调用方在回答生成前后比对版本号：若发现版本号在读取后发生了变化，可以重试或向用户确认。这是对“最终一致性”的一种务实补偿——不追求强一致，但保证可检测冲突。

作者的结论：在分布式智能体场景中，绝对的一致性成本过高。更可行的方案是“写时递增版本号 + 读时携带版本号 + 回答前冲突检测”，将幻觉风险降低到可处理水平。

跨实例记忆锁与乐观并发控制

版本号解决了“读到旧数据”的问题，但没有解决“两个 Pod 同时写入引起的状态覆盖”。当用户在 App 端和 Web 端几乎同时说两件不同的事，两个请求可能落入不同 Pod，产生并发写。

场景：并发写导致记忆丢失

Pod-1 写入消息 M1：“我的订单号是 8823”
Pod-2 写入消息 M2：“我要改地址为上海市”

如果两方都先执行 HLEN 获取当前长度 N，然后 HSET 到 N+1，最终可能只有一个消息被保留，另一个被覆盖。

乐观并发控制（Optimistic Concurrency Control）

我们不再信任“先读后写”的序列，而是要求每次写操作附带预期版本号，用 Redis 的 Lua 脚本做原子检查：

-- lua: atomic_append.lua
local key = KEYS[1]
local expected_version = tonumber(ARGV[1])
local seq = ARGV[2]
local payload = ARGV[3]

local current_version = tonumber(redis.call('HGET', key, '__version') or 0)
if current_version == expected_version then
    redis.call('HSET', key, seq, payload)
    redis.call('HINCRBY', key, '__version', 1)
    return 1  -- 成功
else
    return 0  -- 冲突，需重试
end

Python 侧的调用封装：

class OptimisticLockMemoryBackend(VersionedRedisMemoryBackend):
    def __init__(self, *args, max_retries=3, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_retries = max_retries
        # 注册 Lua 脚本
        self._script = None

    async def _get_script(self):
        if self._script is None:
            with open("atomic_append.lua", "r") as f:
                self._script = self.redis.register_script(f.read())
        return self._script

    async def add_message_safe(self, session_id: str, msg: BaseMessage) -> bool:
        key = f"session:{session_id}:history"
        script = await self._get_script()
        for attempt in range(1, self.max_retries + 1):
            current_version = int(await self.redis.hget(key, "__version") or 0)
            seq = await self.redis.hlen(key) - 1  # 减掉 __version
            payload = json.dumps({"type": msg.type, "content": msg.content})
            success = await script(keys=[key], args=[current_version, seq, payload])
            if success:
                return True
            # 冲突，指数退避后重试
            await asyncio.sleep(0.1 * (2 ** (attempt - 1)))
        return False  # 重试耗尽

降级策略：当并发控制失败时

不是所有场景都需要严格的写冲突保护。如果你的智能体仅做信息查询（读多写少），可以采用悲观锁替代方案——用 Redlock 对 session 加分布式锁：

from redis.asyncio.lock import Lock

async def locked_add_message(self, session_id, msg):
    lock_key = f"lock:session:{session_id}:history"
    lock = Lock(self.redis, lock_key, timeout=5)
    async with lock:
        # 串行化写操作
        await self.add_message(session_id, msg)

对比：乐观 vs 悲观并发控制

从当前调研资料看，LangGraph 的 Checkpointer 接口已支持将每一步的状态写入外部存储，内部使用的就是类似 OCC 的机制（基于 step 序号和 checkpoint_id 做冲突检测）。这印证了“版本号 + 乐观重试”是社区验证过的可行路径。

全局一致视图的落地清单

当你将单实例智能体扩展为多实例部署时，以下是必须完成的四项检查：

1. 共享存储：消息历史不再存在于进程内 ContextVar，而是迁移至 Redis（或等价的 KV 存储）。
2. 版本号机制：每条历史都有一个单调递增的版本号，用于冲突检测和“读己之写”保护。
3. 原子写入：并发写必须通过 Lua 脚本的 CAS 操作或分布式锁进行保护。
4. 降级路径：当锁竞争超时或乐观重试耗尽时，系统有明确的失败处理（向用户诚实反馈“系统繁忙，请重试”）。

将 MemoryFactory.get_history 改造为优先从 Redis 加载历史、并挂载到当前 ContextVar 的逻辑，正是你在上一章末尾预留的“冷启动上下文恢复”钩子。现在，这个钩子有了完整的一致性协议作为底座。

从分布式一致性到流式场景

上下文同步的问题在分布式多实例中暴露得最为彻底，但它并不是唯一的挑战。当智能体通过（Server-Sent Events）或 WebSocket 向用户流式推送回答时，上下文管理的维度就变了：不再是“跨进程同步”，而是“如何在生成过程中动态扩展、组装和截断上下文，以保持流式的实时性和连贯性”。

下一章将进入这个新战场——流式场景中的上下文管理。 我们将解决长期运行 Agent 在推送模式下如何管理不断膨胀的提示词，以及当上下文窗口临界溢出时，是“遗忘”还是“压缩”？