在异步 AI Agent 服务中，如何用 ContextVar 实现请求级别的上下文隔离，避免并发请求间的数据污染？

这个问题特别务实，是在真实项目中几乎一定会碰到的。面试官问这个，其实是在考察你有没有在并发环境下写过有状态的 AI Agent 服务，以及是否理解上下文隔离的工程细节。下面我按从问题到解决方案的逻辑链来组织，尽量不背概念，像在聊架构经验。

🧨 先还原事故现场：数据污染到底怎么发生的？¶

想象一个基于 FastAPI + asyncio 的 AI Agent 服务，它对外提供 /chat 接口。这个 Agent 在调用 LLM、执行工具时，都需要知道“当前请求的用户是谁、会话 ID 是什么、对话历史到哪了”。

新手可能写出这样的代码：

# 全局状态或类属性（反例）
class Agent:
    def __init__(self):
        self.current_user_id = None
        self.conversation_history = []

agent = Agent()

@app.post("/chat")
async def chat(request: ChatRequest):
    agent.current_user_id = request.user_id   # 直接赋值
    agent.conversation_history = load_history(request.session_id)
    reply = await agent.run(request.message)
    return reply

在异步并发下，请求 A 刚设置完 current_user_id = "alice"，await 让出控制权，请求 B 进来把它覆盖成 "bob"。等请求 A 后续调用 LLM 时，拿到的就是 "bob"，导致 Alice 收到了 Bob 的回复，或者更糟——把 Bob 的历史发给了 Alice。

这就是典型的请求级状态污染，根源是：用“线程不安全 + 协程不安全”的共享变量来承载请求级数据。

🎯 解决思路：给每个请求分配一个隔离的“上下文空间”¶

threading.local 解决不了这个问题，因为在一个线程里可以跑几十个协程，它们共享同一个 local 空间。我们需要的是 Task 级隔离——同一个协程 Task 内，所有子调用链自动共享同一份上下文，但不同 Task 之间绝对隔离。

这正是 contextvars.ContextVar 的设计初衷。把它用到 Agent 服务里，整个架构一下子就干净了。

🏗️ 实现四步走（带图标）¶

🔹 Step 1：定义请求级上下文变量

不要零散定义，最好集中在一个模块里，像这样：

import contextvars
from typing import Optional, List, Dict

# 请求级上下文：用户ID、会话ID、对话历史等
user_id_var: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar('user_id', default=None)
session_id_var: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar('session_id', default=None)
history_var: contextvars.ContextVar[List[Dict]] = contextvars.ContextVar('history', default=[])

这样全局定义是安全的，因为具体的“值”是绑定在 Task 上的，不是绑定在变量本身。

🔹 Step 2：在请求入口“种”上下文（中间件）

这是最关键的一步，必须保证在任何业务逻辑执行之前，当前 Task 的上下文已经初始化完成。在 FastAPI 里通常用中间件：

from fastapi import Request

@app.middleware("http")
async def context_middleware(request: Request, call_next):
    # 从请求头、JWT token 等获取当前用户信息
    user_id = request.headers.get("X-User-ID", "anonymous")
    session_id = request.cookies.get("session_id", "unknown")

    # 设置上下文，注意保存 token，用于请求结束后恢复或清理
    token_user = user_id_var.set(user_id)
    token_session = session_id_var.set(session_id)

    try:
        response = await call_next(request)
        return response
    finally:
        # 清理当前 Task 的上下文，避免残留（尤其重要）
        user_id_var.reset(token_user)
        session_id_var.reset(token_session)

set() 返回一个 token，用这个 token 调用 reset() 可以精确地把该变量恢复到设置之前的状态。这是彻底清理的推荐做法。

🔹 Step 3：在 Agent 内部无痛读取

现在，Agent 的任何函数、任何深度，都可以直接读取上下文，而无需改动函数签名：

async def call_llm(prompt: str):
    uid = user_id_var.get()       # 自动获取当前请求的 user_id
    logger.info(f"[{uid}] 正在调用 LLM")
    # ... 拼接 prompt，甚至可以把 uid 放到 metadata 传给模型

工具调用也一样：

async def search_knowledge_base(query: str):
    uid = user_id_var.get()
    logger.info(f"[{uid}] 搜索知识库: {query}")
    # 这里的 uid 一定是当前请求的，不会被其他协程覆盖

整个调用链（Agent → 多个 Tool → 多个 LLM 调用）中，上下文像隐形的参数一样自动传播，零侵入。

🔹 Step 4：处理同步边界（如在线程池执行工具）

Agent 有时会调用同步的数据库驱动或第三方库，为了不阻塞事件循环，我们会用 asyncio.to_thread 丢到线程池。但默认情况下线程不继承 ContextVar 上下文，直接执行会导致 get() 返回默认值。

解决办法是手动传递上下文：

import asyncio

def sync_db_query(query):
    # 此时 ContextVar 已经丢失，需要显式传入参数
    # 但更优雅的是用 copy_context().run() 把上下文带过去
    ...

async def run_sync_tool(ctx, query):
    # 把当前上下文复制，在线程里用 .run() 执行
    return await asyncio.to_thread(ctx.run, sync_db_query, query)

# 调用时
ctx = contextvars.copy_context()
result = await run_sync_tool(ctx, "SELECT ...")

这样即使在另一个线程里运行，ContextVar 的值也能被正确读取，隔离依然有效。

🧩 额外加分项：流式响应的上下文安全¶

现在的 Agent 往往返回流式数据（SSE）。在生成器里直接用 ContextVar 是安全的，因为生成器关联的是同一个 Task。例如：

async def stream_agent_response():
    uid = user_id_var.get()
    async for chunk in llm_astream(prompt):
        logger.debug(f"[{uid}] 发送片段: {chunk}")
        yield chunk

只要你的生成器生命周期在当前请求内，上下文就不会丢。这是 asyncio 与 ContextVar 结合的自然优势。

🧠 为什么要这样设计？三点思考¶

🛡️ 安全第一：在 AI Agent 中，上下文往往包含敏感信息（用户身份、历史对话）。用全局变量就是数据泄露的定时炸弹，而 ContextVar 从语言级别提供了隔离保障。
🧩 松耦合：Agent 的各个组件（LLM、Tool、Memory）只依赖上下文变量，不互相直接传参，代码模块化程度极高，方便单独测试。
🔧 可观测性天然增强：所有日志都可以自动带上 user_id，你可以在日志配置里用一个过滤器从 ContextVar 取值，不用每行日志都手写。

🚦 最终总结（面试时可直接用的结论）¶

在异步 AI Agent 服务中，用 ContextVar 实现请求级上下文隔离，核心就是四步：定义变量 → 中间件种入 → 内部无痛读取 → 跨线程手动传递。它从根本上消除了协程并发下的数据串扰，比 threading.local 安全，比函数传参优雅，特别适合链路长、组件多的 Agent 系统。再配合 copy_context 处理同步边界，以及请求结束时的 token 重置，就能构建一个既轻量又坚固的请求级上下文体系。

这种回答，既有清晰的逻辑步骤，又有具体的代码轮廓，还覆盖了线程池、流式响应这些工程细节，会让面试官觉得你真的在生产环境里踩过坑、做过设计，而不是背完八股文就来答题。