6.2. 按执行上下文隔离是第一条安全原则

电商客服机器人在凌晨三点突然报警：用户A看到了用户B的地址和上一单退款备注。根因不是算法，不是数据库，而是一个全局的 ChatMessageHistory 对象——八百个并发的客服协程，都在往同一个列表里塞消息。这个教训让我们反复验证了一件事：记忆一致性的基石不是锁，而是避免共享的可变状态。这一章我们就沿着这个方向深入，用三十分钟给出一套请求域的上下文隔离实现——这是高并发记忆治理的第一道防线。

你需要什么

• Python 3.10+（asyncio 标准库）
• langchain >= 0.2.7，langchain-core >= 0.2.10（当前调研资料记录的最新版本）
• FastAPI + uvicorn（仅用于模拟并发场景，实际可替换任何异步框架）
• 预计时间：30 分钟

最终成果

你会得到一个生产可用的请求级记忆工厂。它在每次 API 调用内提供完全隔离的对话记忆对象，无论并发多高、协程怎么被调度，用户 A 和用户 B 永远看不到对方的上下文。这个工厂可以直接接到 FastAPI、LangServe 或任何 asyncio 应用里。

第一步：复现“记忆泄露”事故，看清全局状态的危害

我们先人工触发一次事故，用体验记住教训。

创建一个文件 leak_demo.py，模拟一个简易客服对话：

import asyncio
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage

# 共享的全局记忆——这是万恶之源
global_history = ChatMessageHistory()

async def customer_session(session_id: int):
    """模拟一个客服会话"""
    # 加入当前用户的消息
    global_history.add_message(HumanMessage(content=f"我是用户{session_id}，我要查订单"))
    # 模拟AI回复
    global_history.add_message(AIMessage(content=f"用户{session_id}的订单已找到"))
    # 故意yield出让事件循环，让其他协程有机会写全局记忆
    await asyncio.sleep(0.01)
    # 假设后续对话继续追加
    global_history.add_message(HumanMessage(content=f"用户{session_id}再问一句"))
    # 返回当前记忆内容
    return [msg.content for msg in global_history.messages]

async def main():
    tasks = [customer_session(i) for i in range(3)]
    results = await asyncio.gather(*tasks)
    for i, msgs in enumerate(results):
        print(f"会话{i}看到的历史: {msgs}")

if __name__ == "__main__":
    asyncio.run(main())

运行后，你会发现每个会话返回的消息列表里都混杂了其他用户的内容。预期结果就是一段输出示例：

会话0看到的历史: ['我是用户0，我要查订单', "用户0的订单已找到", '我是用户1，我要查订单',
                   "用户1的订单已找到", '我是用户2，我要查订单', "用户2的订单已找到",
                   '用户0再问一句', '用户1再问一句', '用户2再问一句']

三个用户的对话像绞肉机一样搅在一起。这还是在没有真正的网络延迟、没有重试乱序的情况下。生产环境中，问题会更加隐蔽和随机。

⚠️ 踩坑经验
ChatMessageHistory 内部使用普通的 list，在多协程并发写入时既非线程安全，也非任务安全。即使你给 add_message 加锁，也无法阻止两个请求读到对方的脏数据——因为 Python 的异步模型里，锁只保护临界区，不保护你的业务逻辑边界。

第二步：用 ContextVar 把记忆绑定到当前异步任务

Python 3.7 引入的 ContextVar 专门解决这个问题。它允许我们为每个异步任务维护一个独立的“上下文副本”，自动在协程创建、切换时携带过去。

我们写一个基础示例 demo_contextvar.py：

import asyncio
from contextvars import ContextVar

# 声明一个上下文变量，默认值为空
user_context: ContextVar[str] = ContextVar("user_context", default="")

async def task(name: str):
    # 设置当前协程的值
    user_context.set(name)
    await asyncio.sleep(0.01)
    print(f"{name} 看到的值: {user_context.get()}")

async def main():
    await asyncio.gather(task("A"), task("B"))

if __name__ == "__main__":
    asyncio.run(main())

预期结果：每个任务只输出自己的名字，互不干扰。因为是单线程事件循环，但 ContextVar 内部通过协程上下文切换时自动保存和恢复实现了隔离。

现在我们把 ChatMessageHistory 实例绑定到 ContextVar 上，替代全局变量。

第三步：构建请求级记忆工厂

工厂的核心思路很简单：每个请求到达时，用 ContextVar 保存一个属于该请求的 ChatMessageHistory 实例，后续所有记忆读写都操作这个实例。当请求结束时，我们可以选择清理以释放内存。

文件 request_scoped_memory.py：

import uuid
from contextvars import ContextVar
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory

# 声明上下文变量，存储当前请求的记忆对象
_memory_var: ContextVar[BaseChatMessageHistory] = ContextVar(
    "request_memory", default=None
)

class MemoryFactory:
    """为每个请求创建并管理独立的 ChatMessageHistory 实例。

    工厂方法通常由中间件在请求开始时调用，结束时清理。
    """
    @staticmethod
    def initialize_session(session_id: str = None):
        """在请求生命周期开始时调用，创建空记忆副本"""
        if session_id is None:
            session_id = str(uuid.uuid4())
        history = ChatMessageHistory()
        # 可以将 session_id 绑定到 history 的 metadata 中（如有需要）
        _memory_var.set(history)
        return session_id

    @staticmethod
    def get_history() -> BaseChatMessageHistory:
        """获取当前请求的记忆实例。若尚未初始化，抛出异常。"""
        history = _memory_var.get()
        if history is None:
            raise RuntimeError("No request-scoped memory found. "
                               "Did you call `MemoryFactory.initialize_session()` in your middleware?")
        return history

    @staticmethod
    def clear():
        """请求结束时清理，防止内存堆积"""
        _memory_var.set(None)

这里没有使用任何全局锁，也不需要关心并发协程数量。

第四步：集成进 LangChain 的 RunnableWithMessageHistory

LangChain 生态中，RunnableWithMessageHistory 期望你提供一个 get_session_history 回调，它接收一个会话 ID 并返回对应的 BaseChatMessageHistory 对象。我们可以把这个回调植入我们的工厂。

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 回调函数：根据 session_id 返回历史记录。这里我们直接返回当前请求的记忆。
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    # 注意：session_id 由 RunnableWithMessageHistory 自动传入，
    # 但我们已在中间件中根据 session_id 初始化了对应的 ContextVar。
    # 因此这里忽略参数，直接返回当前请求的记忆。
    return MemoryFactory.get_history()

# 组装链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个客服助手，回答用户问题。"),
    ("placeholder", "{history}"),
    ("human", "{input}"),
])
llm = ChatOpenAI(model="gpt-3.5-turbo")  # 可替换为其他模型

chain = prompt | llm

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

对于每个请求，只要在进入链调用前通过中间件初始化Session，便万事大吉。

FastAPI 集成片段：

from fastapi import FastAPI, Request
import asyncio

app = FastAPI()

@app.middleware("http")
async def request_memory_middleware(request: Request, call_next):
    # 请求到来时初始化记忆，使用 URL 参数或 header 携带的 session_id
    session_id = request.headers.get("X-Session-ID", str(uuid.uuid4()))
    MemoryFactory.initialize_session(session_id)
    response = await call_next(request)
    # 清理，避免内存泄露
    MemoryFactory.clear()
    return response

@app.post("/chat")
async def chat(request: Request):
    body = await request.json()
    user_input = body.get("message")
    session_id = request.headers.get("X-Session-ID")
    # 配置异步调用
    config = {"configurable": {"session_id": session_id}}
    response = await chain_with_history.ainvoke(
        {"input": user_input},
        config=config,
    )
    return {"reply": response.content}

第五步：验证隔离效果

我们用 httpx 和 asyncio 并发发送多个请求，然后检查各自回复中是否含有其它用户的痕迹。

测试脚本 test_isolation.py：

import asyncio
import httpx

async def send_request(client, session_id, message):
    headers = {"X-Session-ID": session_id}
    resp = await client.post(
        "http://localhost:8000/chat",
        json={"message": message},
        headers=headers,
    )
    return resp.json()["reply"]

async def main():
    async with httpx.AsyncClient() as client:
        # 同时发起两个完全不同的会话
        replies = await asyncio.gather(
            send_request(client, "user1", "我的订单号是123，查一下物流"),
            send_request(client, "user2", "我想退货，商品ID是456"),
        )
        print("用户1回复:", replies[0])
        print("用户2回复:", replies[1])
        # 用户1的回复中不应出现“用户2”或“456”等字样
        assert "456" not in replies[0], "隔离失败，用户1看到了用户2的上下文"

if __name__ == "__main__":
    asyncio.run(main())

运行前确保 FastAPI 服务已启动，然后执行测试。预期结果：断言通过，两个会话完全隔离。

回顾

我们做了什么

• 用一个真实的电商事故场景复现了全局 ChatMessageHistory 造成的记忆泄露；
• 理解了 ContextVar 的异步任务隔离机制；
• 构建了生产级的 MemoryFactory，自动为每个请求创建独立的 ChatMessageHistory；
• 将该工厂注入 RunnableWithMessageHistory 并集成到 FastAPI 服务中；
• 编写并发测试，验证隔离效果。

整个流程耗时约 30 分钟（不含学习基础概念）。

行动清单

1. 替换所有全局 ChatMessageHistory：搜索项目中每个 ChatMessageHistory() 实例，改用 MemoryFactory.get_history() 获取。
2. 在 ASGI 中间件中设置/清理上下文：确保每个请求开头调用 MemoryFactory.initialize_session()，结尾调用 clear()。
3. 使用 session_id 打通全链路：通过 RunnableWithMessageHistory 的 configurable 传递 session_id，使链内记忆自动路由。
4. 加入监控报警：在 get_history 中如遇未初始化，主动抛出明确异常并接入日志，方便排查中间件遗漏。
5. 压测验证：使用 Locust 或 k6 模拟 200+ 并发长会话，确认无上下文泄漏。

现在，我们的单实例请求级隔离已经牢靠了。但这套方案隐含着另一个前提：所有同一个 session 的请求都必须落到同一个进程里。当我们将服务扩展到多台机器，或者多进程部署时，用户 A 的后续请求可能被路由到另一台实例，而那个实例的 ContextVar 里空空如也。下一章《分布式智能体需要全局一致的上下文视图》将讨论多服务实例共享一个智能体时的上下文同步与一致性协议，让记忆跨越进程边界。

写在边上：你刚刚实现的 MemoryFactory 其实已经为分布式扩展埋下了钩子——get_history 方法内部可以在 ContextVar 为空时，转而从外部存储（Redis/Database）加载历史并挂载到当前上下文，实现“冷启动”后的上下文恢复。这个模式我们会在下一章具体展开。