在多租户场景下保证 LangChain / LangGraph Agent 的上下文不串，核心是建立"租户 / 用户 / 会话"三层 ID 体系，让每一次invoke都携带全局唯一且经过鉴权校验的会话标识，并配合持久化后端做物理分区。

一、核心思路

LangChain 官方把 Agent 的"记忆"拆成了两个互补的持久化系统，多租户隔离也必须分层解决：

• Checkpointer（短期记忆）：持久化单个 thread 的图状态快照，负责对话连续性、人机协作、时间旅行、容错。多租户隔离靠config["configurable"]["thread_id"]做线程级隔离。
• Store（长期记忆）：持久化跨 thread 的应用数据，负责用户偏好、事实、共享知识。多租户隔离靠namespace 重写或应用内显式用户作用域。

💡 关键认知：线程隔离和存储隔离解决的是不同维度的问题——前者管"这一轮对话的历史"，后者管"跨对话的长期记忆"。要做严谨的多租户，两层都要配置。

隔离粒度上，业界主流做法是用复合 thread_id：

org:{org_id}:user:{user_id}:session:{session_id}

这样做的好处：

1. 一个用户可开多个独立会话，互不干扰
2. 便于按前缀批量清理（GDPR 合规删除时很关键）
3. 物理层消除状态交叉污染——Checkpointer 在底层存储按 thread_id 分区

⚠️ 注意：LangGraph 论坛官方专家明确提醒，不要滥用checkpoint_ns来做多租户——那是内部用于子图/分支层级标识的，应让运行时自动管理。正确做法是把thread_id当 conversation ID，确保其全局唯一（如 UUID/ULID），在应用层建一张conversations表记录thread_id ↔ tenant_id / user_id的映射并加 RLS（Row-Level Security）。

二、详细步骤

步骤 1：建立三层 ID 体系与鉴权网关

在 API 边界（FastAPI / Flask）做强校验——绝不能无条件信任客户端传入的thread_id，否则恶意用户猜测他人 thread_id 就能越权读取数据。

租户 ID (tenant_id) ──┐ 用户 ID (user_id) ──┼──→ 复合 thread_id ──→ 注入 config["configurable"] 会话 ID (session_id) ─┘

鉴权流程：

1. 从 JWT / API Key 解析出tenant_id和user_id
2. 客户端传入session_id（或服务端生成）
3. 服务端拼接thread_id = f"org:{tenant_id}:user:{user_id}:session:{session_id}"
4. 校验当前用户是否有权访问该session_id（查conversations表）
5. 通过后才往下游 Agent 注入

步骤 2：选择持久化后端

📌 生产环境绝对不要用InMemoryChatMessageHistory做持久化——进程重启所有对话丢失，且多 worker 之间不共享。

步骤 3：Checkpointer 线程隔离（短期记忆）

编译 Graph 时挂 Checkpointer，每次ainvoke时传入带thread_id的 config。

步骤 4：Store 命名空间隔离（长期记忆）

两种互斥方案二选一（不要叠加，否则会 double-scoping）：

• 方案 A：Auth 层自动前缀重写（推荐）。Graph 代码里用逻辑命名空间("memories", "preferences")，认证中间件自动在前面加上user-a前缀，最终存到("user-a", "memories", "preferences")。
• 方案 B：应用代码显式作用域。在 Graph 节点里从config["configurable"]["langgraph_auth_user_id"]取出 user_id，显式拼到 namespace 里。

步骤 5：请求上下文注入（FastAPI 实战）

通过请求头 / JWT 拿到租户和用户身份，注入到config里往下传。

步骤 6：自动化测试验证隔离性

写并发测试：两个线程用不同 session_id 同时打同一个 Agent，断言彼此看不到对方上下文。

步骤 7：生命周期管理与可观测

• 过期清理：记录thread_id最后活跃时间，定时扫描（如 48h 无交互）归档到冷存储后清 Redis
• LangSmith 追踪：给每条 Trace 打thread_id标签，便于按租户/会话筛选日志、统计 token、定位异常会话

三、代码详解

1. 数据模型与鉴权（应用层）

# models.py —— 应用级会话表，owner 关系在这里管fromsqlalchemyimportColumn,String,DateTime,UUIDfromsqlalchemy.ext.declarativeimportdeclarative_baseimportuuid Base=declarative_base()classConversation(Base):__tablename__="conversations"id=Column(UUID,primary_key=True,default=uuid.uuid4)# 这个就是 thread_idtenant_id=Column(String,nullable=False,index=True)user_id=Column(String,nullable=False,index=True)title=Column(String)created_at=Column(DateTime,default=__import__("datetime").datetime.now)updated_at=Column(DateTime,default=__import__("datetime").datetime.now)

-- 启用 RLS，确保即使 Checkpointer 被绕过，也无法跨租户读数据ALTERTABLEconversationsENABLEROWLEVELSECURITY;CREATEPOLICY conversations_rlsONconversationsUSING(tenant_id=current_setting('app.tenant_id')::uuidANDuser_id=current_setting('app.user_id')::uuid);

2. LangGraph Agent（Checkpointer + Store 双层隔离）

# agent.pyimportuuidfromlanggraph.graphimportStateGraph,START,END,MessagesStatefromlanggraph.checkpoint.postgres.aioimportAsyncPostgresSaverfromlanggraph.store.postgresimportPostgresStorefromlangchain_openaiimportChatOpenAI# 全局共享同一个 Graph 定义（无状态），状态由 checkpointer 按 thread_id 隔离asyncdefbuild_agent():# 1. 短期记忆：Checkpointer，按 thread_id 物理分区checkpointer=AsyncPostgresSaver.from_conn_string("postgres://user:pass@localhost:5432/langgraph")awaitcheckpointer.setup()# 2. 长期记忆：Store，按 namespace 隔离store=PostgresStore.from_conn_string("postgres://user:pass@localhost:5432/langgraph")awaitstore.setup()# 3. 编译图（checkpointer + store 一起挂）llm=ChatOpenAI(model="gpt-4o-mini")builder=StateGraph(MessagesState)asyncdefassistant_node(state:MessagesState,*,store,config):# 长期记忆读写示例：从 store 取用户偏好user_id=config["configurable"]["langgraph_auth_user_id"]# 显式作用域方案（方案B）：namespace 里带上 user_idpref=awaitstore.aget((user_id,"memories","preferences"),"settings")system_hint=f"用户偏好：{pref.value if pref else '{}'}"messages=[("system",system_hint)]+state["messages"]return{"messages":[llm.invoke(messages)]}builder.add_node("assistant",assistant_node)builder.add_edge(START,"assistant")builder.add_edge("assistant",END)# 编译时同时挂 checkpointer 和 storereturnbuilder.compile(checkpointer=checkpointer,store=store)

3. 复合 thread_id 生成与校验（核心防串台逻辑）

# thread_utils.pyimportrefromfastapiimportHTTPExceptiondefbuild_thread_id(tenant_id:str,user_id:str,session_id:str)->str:"""生成全局唯一的复合 thread_id"""returnf"org:{tenant_id}:user:{user_id}:session:{session_id}"defparse_and_verify_thread_id(thread_id:str,tenant_id:str,user_id:str)->None:""" 校验客户端传入的 thread_id 是否真的属于该租户/用户。 防止越权：恶意用户猜别人的 session_id 会被这里拦住。 """pattern=re.compile(r"^org:(?P<t>.+):user:(?P<u>.+):session:(?P<s>.+)$")m=pattern.match(thread_id)ifnotm:raiseHTTPException(status_code=400,detail="Invalid thread_id format")ifm.group("t")!=tenant_idorm.group("u")!=user_id:raiseHTTPException(status_code=403,detail="Access denied to this thread")

4. FastAPI 接口层（请求上下文注入）

# main.pyfromfastapiimportFastAPI,Header,Depends,HTTPExceptionfrompydanticimportBaseModelfromtypingimportOptionalimportuuidfromagentimportbuild_agentfromthread_utilsimportbuild_thread_id,parse_and_verify_thread_id app=FastAPI()agent=None# 启动时初始化@app.on_event("startup")asyncdefstartup():globalagent agent=awaitbuild_agent()classChatRequest(BaseModel):message:strsession_id:Optional[str]=None# 客户端可选传，不传则新建defget_current_tenant_user(x_tenant_id:str=Header(...,alias="X-Tenant-Id"),x_user_id:str=Header(...,alias="X-User-Id"),authorization:str=Header(...,alias="Authorization"),):"""模拟 JWT 鉴权，真实场景用 jwt.decode() 校验签名"""# TODO: 真实项目中这里解码 JWT 并校验签名、过期时间ifnotauthorization.startswith("Bearer "):raiseHTTPException(status_code=401,detail="Invalid token")return{"tenant_id":x_tenant_id,"user_id":x_user_id}@app.post("/chat")asyncdefchat(req:ChatRequest,auth:dict=Depends(get_current_tenant_user),):tenant_id=auth["tenant_id"]user_id=auth["user_id"]# 1. 生成或复用 session_idsession_id=req.session_idorstr(uuid.uuid4())# 2. 构造复合 thread_id（服务端拼接，不信任客户端直接传 thread_id）thread_id=build_thread_id(tenant_id,user_id,session_id)# 3. 如果客户端传了 session_id，校验其归属（防越权）ifreq.session_id:parse_and_verify_thread_id(thread_id,tenant_id,user_id)# 4. 注入 config —— 这是隔离的核心！config={"configurable":{"thread_id":thread_id,# Checkpointer 用它做物理分区"langgraph_auth_user_id":user_id,# Store 用它做 namespace 作用域"tenant_id":tenant_id,# 业务层透传}}# 5. 调用 Agent —— 不同租户/用户/会话的 thread_id 不同，上下文完全隔离result=awaitagent.ainvoke({"messages":[{"role":"user","content":req.message}]},config=config,)return{"session_id":session_id,"thread_id":thread_id,"reply":result["messages"][-1].content,}

5. LangChain 传统链方案（RunnableWithMessageHistory）

如果用的是经典 LCEL Chain 而非 LangGraph，用RunnableWithMessageHistory配RedisChatMessageHistory：

# langchain_chain.pyfromlangchain_core.runnables.historyimportRunnableWithMessageHistoryfromlangchain_core.runnablesimportConfigurableFieldSpecfromlangchain_community.chat_message_historiesimportRedisChatMessageHistoryimportredis redis_client=redis.Redis(host="localhost",port=6379,db=0,decode_responses=True)defget_session_history(user_id:str,conversation_id:str)->RedisChatMessageHistory:""" 工厂函数：session_id 直接拼成 'user:{user_id}:conv:{conversation_id}' Redis 里按这个 key 做物理分区，不同用户/会话天然隔离 """composite_key=f"user:{user_id}:conv:{conversation_id}"returnRedisChatMessageHistory(session_id=composite_key,redis_client=redis_client)# 用 user_id + conversation_id 双键做历史工厂chain_with_history=RunnableWithMessageHistory(chain,get_session_history,input_messages_key="input",history_messages_key="history",history_factory_config=[ConfigurableFieldSpec(id="user_id",annotation=str,name="User ID",description="Unique identifier for the user.",is_shared=True,),ConfigurableFieldSpec(id="conversation_id",annotation=str,name="Conversation ID",description="Unique identifier for the conversation.",is_shared=True,),],)# 调用时注入双键awaitchain_with_history.ainvoke({"input":"What does cosine mean?"},config={"configurable":{"user_id":"tenant_A_user_123","conversation_id":"sess_001"}},)

💡 注意：RunnableWithMessageHistory的invoke()是同步的，FastAPI 异步接口里要用.ainvoke()避免阻塞事件循环。

6. 隔离性验证测试

# test_isolation.pyimportasynciofromlanggraph_sdkimportget_clientasyncdefmain():alice=get_client(url="http://localhost:60058",headers={"Authorization":"Bearer user1-token"})bob=get_client(url="http://localhost:60058",headers={"Authorization":"Bearer user2-token"})# Alice 写长期记忆awaitalice.store.put_item(["memories"],key="note",value={"text":"Alice private note"})# Bob 读不到 Alice 的数据bob_item=awaitbob.store.get_item(["memories"],key="note")assertbob_itemisNone,"❌ Bob 不应看到 Alice 的 store 数据"# 各自写自己的awaitbob.store.put_item(["memories"],key="note",value={"text":"Bob private note"})alice_item=awaitalice.store.get_item(["memories"],key="note")assertalice_item["value"]["text"]=="Alice private note"print("✅ 长期记忆隔离验证通过")asyncio.run(main())

并发测试建议：用threading起 10 个线程，5 个模拟用户 A、5 个模拟用户 B 同时发消息，断言彼此回复中不出现对方上下文——这能抓出单用户测试发现不了的竞态 bug。

四、总结

防串台的本质：让每一次 Agent 调用都携带服务端生成的、经过鉴权校验的、全局唯一的会话标识，并通过持久化后端的物理分区（Postgres / Redis 的 key 隔离）落地。

三层防御体系：

1. 身份层：JWT / API Key 解析tenant_id+user_id，绝不信任客户端传入的thread_id
2. 会话层：复合thread_id = org:{tenant_id}:user:{user_id}:session:{session_id}，Checkpointer 按此物理分区
3. 记忆层：Store 的 namespace 要么由 Auth 层自动加用户前缀，要么在应用代码里显式拼user_id

LangGraph vs LangChain 传统链：

• LangGraph：用AsyncPostgresSaver/AsyncRedisSaver做 Checkpointer +PostgresStore做 Store，config["configurable"]["thread_id"]是隔离主键
• LangChain LCEL：用RunnableWithMessageHistory+RedisChatMessageHistory，session_id做隔离主键，支持user_id+conversation_id双键工厂

隔离级别选型：

⚠️ 三个最容易踩的坑：
① 直接用user_id当thread_id→ 同一用户多窗口并发会状态覆盖
② 用InMemoryChatMessageHistory上生产 → 重启丢数据、多 worker 不共享
③ 滥用checkpoint_ns做多租户 → 那是 LangGraph 内部子图层级标识，会让时间旅行/调试元数据乱掉

生产加固清单：

• ✅ 应用层conversations表 + RLS 双重保险
• ✅thread_id用 UUID/ULID 保证全局唯一
• ✅ 长期记忆 namespace 与 thread_id 双重隔离
• ✅ 过期会话定时归档清理（如 48h 无活跃）
• ✅ LangSmith 按thread_id打标追踪
• ✅ 并发隔离测试纳入 CI

按这套方案落地，数百租户同时在线、每租户多会话并发都能做到上下文零串台。