在多租户场景下保证 LangChain / LangGraph Agent 的上下文不串核心是建立租户 / 用户 / 会话三层 ID 体系让每一次invoke都携带全局唯一且经过鉴权校验的会话标识并配合持久化后端做物理分区。一、核心思路LangChain 官方把 Agent 的记忆拆成了两个互补的持久化系统多租户隔离也必须分层解决Checkpointer短期记忆持久化单个 thread 的图状态快照负责对话连续性、人机协作、时间旅行、容错。多租户隔离靠config[configurable][thread_id]做线程级隔离。Store长期记忆持久化跨 thread 的应用数据负责用户偏好、事实、共享知识。多租户隔离靠namespace 重写或应用内显式用户作用域。 关键认知线程隔离和存储隔离解决的是不同维度的问题——前者管这一轮对话的历史后者管跨对话的长期记忆。要做严谨的多租户两层都要配置。隔离粒度上业界主流做法是用复合 thread_idorg:{org_id}:user:{user_id}:session:{session_id}这样做的好处一个用户可开多个独立会话互不干扰便于按前缀批量清理GDPR 合规删除时很关键物理层消除状态交叉污染——Checkpointer 在底层存储按 thread_id 分区⚠️ 注意LangGraph 论坛官方专家明确提醒不要滥用checkpoint_ns来做多租户——那是内部用于子图/分支层级标识的应让运行时自动管理。正确做法是把thread_id当 conversation ID确保其全局唯一如 UUID/ULID在应用层建一张conversations表记录thread_id ↔ tenant_id / user_id的映射并加 RLSRow-Level Security。二、详细步骤步骤 1建立三层 ID 体系与鉴权网关在 API 边界FastAPI / Flask做强校验——绝不能无条件信任客户端传入的thread_id否则恶意用户猜测他人 thread_id 就能越权读取数据。租户 ID (tenant_id) ──┐ 用户 ID (user_id) ──┼──→ 复合 thread_id ──→ 注入 config[configurable] 会话 ID (session_id) ─┘鉴权流程从 JWT / API Key 解析出tenant_id和user_id客户端传入session_id或服务端生成服务端拼接thread_id forg:{tenant_id}:user:{user_id}:session:{session_id}校验当前用户是否有权访问该session_id查conversations表通过后才往下游 Agent 注入步骤 2选择持久化后端场景推荐方案开发 / 单租户原型MemorySaver/InMemoryStore生产 - 逻辑隔离AsyncPostgresSaverPostgresStore生产 - 高并发低延迟AsyncRedisSaverRedisStore生产 - 高安全需求物理隔离每租户独立数据库实例 生产环境绝对不要用InMemoryChatMessageHistory做持久化——进程重启所有对话丢失且多 worker 之间不共享。步骤 3Checkpointer 线程隔离短期记忆编译 Graph 时挂 Checkpointer每次ainvoke时传入带thread_id的 config。步骤 4Store 命名空间隔离长期记忆两种互斥方案二选一不要叠加否则会 double-scoping方案 AAuth 层自动前缀重写推荐。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 无交互归档到冷存储后清 RedisLangSmith 追踪给每条 Trace 打thread_id标签便于按租户/会话筛选日志、统计 token、定位异常会话三、代码详解1. 数据模型与鉴权应用层# models.py —— 应用级会话表owner 关系在这里管fromsqlalchemyimportColumn,String,DateTime,UUIDfromsqlalchemy.ext.declarativeimportdeclarative_baseimportuuid Basedeclarative_base()classConversation(Base):__tablename__conversationsidColumn(UUID,primary_keyTrue,defaultuuid.uuid4)# 这个就是 thread_idtenant_idColumn(String,nullableFalse,indexTrue)user_idColumn(String,nullableFalse,indexTrue)titleColumn(String)created_atColumn(DateTime,default__import__(datetime).datetime.now)updated_atColumn(DateTime,default__import__(datetime).datetime.now)-- 启用 RLS确保即使 Checkpointer 被绕过也无法跨租户读数据ALTERTABLEconversationsENABLEROWLEVELSECURITY;CREATEPOLICY conversations_rlsONconversationsUSING(tenant_idcurrent_setting(app.tenant_id)::uuidANDuser_idcurrent_setting(app.user_id)::uuid);2. LangGraph AgentCheckpointer Store 双层隔离# agent.pyimportuuidfromlanggraph.graphimportStateGraph,START,END,MessagesStatefromlanggraph.checkpoint.postgres.aioimportAsyncPostgresSaverfromlanggraph.store.postgresimportPostgresStorefromlangchain_openaiimportChatOpenAI# 全局共享同一个 Graph 定义无状态状态由 checkpointer 按 thread_id 隔离asyncdefbuild_agent():# 1. 短期记忆Checkpointer按 thread_id 物理分区checkpointerAsyncPostgresSaver.from_conn_string(postgres://user:passlocalhost:5432/langgraph)awaitcheckpointer.setup()# 2. 长期记忆Store按 namespace 隔离storePostgresStore.from_conn_string(postgres://user:passlocalhost:5432/langgraph)awaitstore.setup()# 3. 编译图checkpointer store 一起挂llmChatOpenAI(modelgpt-4o-mini)builderStateGraph(MessagesState)asyncdefassistant_node(state:MessagesState,*,store,config):# 长期记忆读写示例从 store 取用户偏好user_idconfig[configurable][langgraph_auth_user_id]# 显式作用域方案方案Bnamespace 里带上 user_idprefawaitstore.aget((user_id,memories,preferences),settings)system_hintf用户偏好{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(checkpointercheckpointer,storestore)3. 复合 thread_id 生成与校验核心防串台逻辑# thread_utils.pyimportrefromfastapiimportHTTPExceptiondefbuild_thread_id(tenant_id:str,user_id:str,session_id:str)-str:生成全局唯一的复合 thread_idreturnforg:{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 会被这里拦住。 patternre.compile(r^org:(?Pt.):user:(?Pu.):session:(?Ps.)$)mpattern.match(thread_id)ifnotm:raiseHTTPException(status_code400,detailInvalid thread_id format)ifm.group(t)!tenant_idorm.group(u)!user_id:raiseHTTPException(status_code403,detailAccess denied to this thread)4. FastAPI 接口层请求上下文注入# main.pyfromfastapiimportFastAPI,Header,Depends,HTTPExceptionfrompydanticimportBaseModelfromtypingimportOptionalimportuuidfromagentimportbuild_agentfromthread_utilsimportbuild_thread_id,parse_and_verify_thread_id appFastAPI()agentNone# 启动时初始化app.on_event(startup)asyncdefstartup():globalagent agentawaitbuild_agent()classChatRequest(BaseModel):message:strsession_id:Optional[str]None# 客户端可选传不传则新建defget_current_tenant_user(x_tenant_id:strHeader(...,aliasX-Tenant-Id),x_user_id:strHeader(...,aliasX-User-Id),authorization:strHeader(...,aliasAuthorization),):模拟 JWT 鉴权真实场景用 jwt.decode() 校验签名# TODO: 真实项目中这里解码 JWT 并校验签名、过期时间ifnotauthorization.startswith(Bearer ):raiseHTTPException(status_code401,detailInvalid token)return{tenant_id:x_tenant_id,user_id:x_user_id}app.post(/chat)asyncdefchat(req:ChatRequest,auth:dictDepends(get_current_tenant_user),):tenant_idauth[tenant_id]user_idauth[user_id]# 1. 生成或复用 session_idsession_idreq.session_idorstr(uuid.uuid4())# 2. 构造复合 thread_id服务端拼接不信任客户端直接传 thread_idthread_idbuild_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 不同上下文完全隔离resultawaitagent.ainvoke({messages:[{role:user,content:req.message}]},configconfig,)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_clientredis.Redis(hostlocalhost,port6379,db0,decode_responsesTrue)defget_session_history(user_id:str,conversation_id:str)-RedisChatMessageHistory: 工厂函数session_id 直接拼成 user:{user_id}:conv:{conversation_id} Redis 里按这个 key 做物理分区不同用户/会话天然隔离 composite_keyfuser:{user_id}:conv:{conversation_id}returnRedisChatMessageHistory(session_idcomposite_key,redis_clientredis_client)# 用 user_id conversation_id 双键做历史工厂chain_with_historyRunnableWithMessageHistory(chain,get_session_history,input_messages_keyinput,history_messages_keyhistory,history_factory_config[ConfigurableFieldSpec(iduser_id,annotationstr,nameUser ID,descriptionUnique identifier for the user.,is_sharedTrue,),ConfigurableFieldSpec(idconversation_id,annotationstr,nameConversation ID,descriptionUnique identifier for the conversation.,is_sharedTrue,),],)# 调用时注入双键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():aliceget_client(urlhttp://localhost:60058,headers{Authorization:Bearer user1-token})bobget_client(urlhttp://localhost:60058,headers{Authorization:Bearer user2-token})# Alice 写长期记忆awaitalice.store.put_item([memories],keynote,value{text:Alice private note})# Bob 读不到 Alice 的数据bob_itemawaitbob.store.get_item([memories],keynote)assertbob_itemisNone,❌ Bob 不应看到 Alice 的 store 数据# 各自写自己的awaitbob.store.put_item([memories],keynote,value{text:Bob private note})alice_itemawaitalice.store.get_item([memories],keynote)assertalice_item[value][text]Alice private noteprint(✅ 长期记忆隔离验证通过)asyncio.run(main())并发测试建议用threading起 10 个线程5 个模拟用户 A、5 个模拟用户 B 同时发消息断言彼此回复中不出现对方上下文——这能抓出单用户测试发现不了的竞态 bug。四、总结防串台的本质让每一次 Agent 调用都携带服务端生成的、经过鉴权校验的、全局唯一的会话标识并通过持久化后端的物理分区Postgres / Redis 的 key 隔离落地。三层防御体系身份层JWT / API Key 解析tenant_iduser_id绝不信任客户端传入的thread_id会话层复合thread_id org:{tenant_id}:user:{user_id}:session:{session_id}Checkpointer 按此物理分区记忆层Store 的 namespace 要么由 Auth 层自动加用户前缀要么在应用代码里显式拼user_idLangGraph vs LangChain 传统链LangGraph用AsyncPostgresSaver/AsyncRedisSaver做 Checkpointer PostgresStore做 Storeconfig[configurable][thread_id]是隔离主键LangChain LCEL用RunnableWithMessageHistoryRedisChatMessageHistorysession_id做隔离主键支持user_idconversation_id双键工厂隔离级别选型级别实现方式适用场景逻辑隔离复合 thread_id 命名空间中小规模、互信租户物理隔离每租户独立数据库实例大规模、高安全/合规需求混合隔离按租户重要性分级多级别安全需求⚠️ 三个最容易踩的坑① 直接用user_id当thread_id→ 同一用户多窗口并发会状态覆盖② 用InMemoryChatMessageHistory上生产 → 重启丢数据、多 worker 不共享③ 滥用checkpoint_ns做多租户 → 那是 LangGraph 内部子图层级标识会让时间旅行/调试元数据乱掉生产加固清单✅ 应用层conversations表 RLS 双重保险✅thread_id用 UUID/ULID 保证全局唯一✅ 长期记忆 namespace 与 thread_id 双重隔离✅ 过期会话定时归档清理如 48h 无活跃✅ LangSmith 按thread_id打标追踪✅ 并发隔离测试纳入 CI按这套方案落地数百租户同时在线、每租户多会话并发都能做到上下文零串台。
在多租户场景下保证 LangChain / LangGraph Agent 的上下文不串核心是建立租户 / 用户 / 会话三层 ID 体系让每一次invoke都携带全局唯一且经过鉴权校验的会话标识并配合持久化后端做物理分区。一、核心思路LangChain 官方把 Agent 的记忆拆成了两个互补的持久化系统多租户隔离也必须分层解决Checkpointer短期记忆持久化单个 thread 的图状态快照负责对话连续性、人机协作、时间旅行、容错。多租户隔离靠config[configurable][thread_id]做线程级隔离。Store长期记忆持久化跨 thread 的应用数据负责用户偏好、事实、共享知识。多租户隔离靠namespace 重写或应用内显式用户作用域。 关键认知线程隔离和存储隔离解决的是不同维度的问题——前者管这一轮对话的历史后者管跨对话的长期记忆。要做严谨的多租户两层都要配置。隔离粒度上业界主流做法是用复合 thread_idorg:{org_id}:user:{user_id}:session:{session_id}这样做的好处一个用户可开多个独立会话互不干扰便于按前缀批量清理GDPR 合规删除时很关键物理层消除状态交叉污染——Checkpointer 在底层存储按 thread_id 分区⚠️ 注意LangGraph 论坛官方专家明确提醒不要滥用checkpoint_ns来做多租户——那是内部用于子图/分支层级标识的应让运行时自动管理。正确做法是把thread_id当 conversation ID确保其全局唯一如 UUID/ULID在应用层建一张conversations表记录thread_id ↔ tenant_id / user_id的映射并加 RLSRow-Level Security。二、详细步骤步骤 1建立三层 ID 体系与鉴权网关在 API 边界FastAPI / Flask做强校验——绝不能无条件信任客户端传入的thread_id否则恶意用户猜测他人 thread_id 就能越权读取数据。租户 ID (tenant_id) ──┐ 用户 ID (user_id) ──┼──→ 复合 thread_id ──→ 注入 config[configurable] 会话 ID (session_id) ─┘鉴权流程从 JWT / API Key 解析出tenant_id和user_id客户端传入session_id或服务端生成服务端拼接thread_id forg:{tenant_id}:user:{user_id}:session:{session_id}校验当前用户是否有权访问该session_id查conversations表通过后才往下游 Agent 注入步骤 2选择持久化后端场景推荐方案开发 / 单租户原型MemorySaver/InMemoryStore生产 - 逻辑隔离AsyncPostgresSaverPostgresStore生产 - 高并发低延迟AsyncRedisSaverRedisStore生产 - 高安全需求物理隔离每租户独立数据库实例 生产环境绝对不要用InMemoryChatMessageHistory做持久化——进程重启所有对话丢失且多 worker 之间不共享。步骤 3Checkpointer 线程隔离短期记忆编译 Graph 时挂 Checkpointer每次ainvoke时传入带thread_id的 config。步骤 4Store 命名空间隔离长期记忆两种互斥方案二选一不要叠加否则会 double-scoping方案 AAuth 层自动前缀重写推荐。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 无交互归档到冷存储后清 RedisLangSmith 追踪给每条 Trace 打thread_id标签便于按租户/会话筛选日志、统计 token、定位异常会话三、代码详解1. 数据模型与鉴权应用层# models.py —— 应用级会话表owner 关系在这里管fromsqlalchemyimportColumn,String,DateTime,UUIDfromsqlalchemy.ext.declarativeimportdeclarative_baseimportuuid Basedeclarative_base()classConversation(Base):__tablename__conversationsidColumn(UUID,primary_keyTrue,defaultuuid.uuid4)# 这个就是 thread_idtenant_idColumn(String,nullableFalse,indexTrue)user_idColumn(String,nullableFalse,indexTrue)titleColumn(String)created_atColumn(DateTime,default__import__(datetime).datetime.now)updated_atColumn(DateTime,default__import__(datetime).datetime.now)-- 启用 RLS确保即使 Checkpointer 被绕过也无法跨租户读数据ALTERTABLEconversationsENABLEROWLEVELSECURITY;CREATEPOLICY conversations_rlsONconversationsUSING(tenant_idcurrent_setting(app.tenant_id)::uuidANDuser_idcurrent_setting(app.user_id)::uuid);2. LangGraph AgentCheckpointer Store 双层隔离# agent.pyimportuuidfromlanggraph.graphimportStateGraph,START,END,MessagesStatefromlanggraph.checkpoint.postgres.aioimportAsyncPostgresSaverfromlanggraph.store.postgresimportPostgresStorefromlangchain_openaiimportChatOpenAI# 全局共享同一个 Graph 定义无状态状态由 checkpointer 按 thread_id 隔离asyncdefbuild_agent():# 1. 短期记忆Checkpointer按 thread_id 物理分区checkpointerAsyncPostgresSaver.from_conn_string(postgres://user:passlocalhost:5432/langgraph)awaitcheckpointer.setup()# 2. 长期记忆Store按 namespace 隔离storePostgresStore.from_conn_string(postgres://user:passlocalhost:5432/langgraph)awaitstore.setup()# 3. 编译图checkpointer store 一起挂llmChatOpenAI(modelgpt-4o-mini)builderStateGraph(MessagesState)asyncdefassistant_node(state:MessagesState,*,store,config):# 长期记忆读写示例从 store 取用户偏好user_idconfig[configurable][langgraph_auth_user_id]# 显式作用域方案方案Bnamespace 里带上 user_idprefawaitstore.aget((user_id,memories,preferences),settings)system_hintf用户偏好{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(checkpointercheckpointer,storestore)3. 复合 thread_id 生成与校验核心防串台逻辑# thread_utils.pyimportrefromfastapiimportHTTPExceptiondefbuild_thread_id(tenant_id:str,user_id:str,session_id:str)-str:生成全局唯一的复合 thread_idreturnforg:{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 会被这里拦住。 patternre.compile(r^org:(?Pt.):user:(?Pu.):session:(?Ps.)$)mpattern.match(thread_id)ifnotm:raiseHTTPException(status_code400,detailInvalid thread_id format)ifm.group(t)!tenant_idorm.group(u)!user_id:raiseHTTPException(status_code403,detailAccess denied to this thread)4. FastAPI 接口层请求上下文注入# main.pyfromfastapiimportFastAPI,Header,Depends,HTTPExceptionfrompydanticimportBaseModelfromtypingimportOptionalimportuuidfromagentimportbuild_agentfromthread_utilsimportbuild_thread_id,parse_and_verify_thread_id appFastAPI()agentNone# 启动时初始化app.on_event(startup)asyncdefstartup():globalagent agentawaitbuild_agent()classChatRequest(BaseModel):message:strsession_id:Optional[str]None# 客户端可选传不传则新建defget_current_tenant_user(x_tenant_id:strHeader(...,aliasX-Tenant-Id),x_user_id:strHeader(...,aliasX-User-Id),authorization:strHeader(...,aliasAuthorization),):模拟 JWT 鉴权真实场景用 jwt.decode() 校验签名# TODO: 真实项目中这里解码 JWT 并校验签名、过期时间ifnotauthorization.startswith(Bearer ):raiseHTTPException(status_code401,detailInvalid token)return{tenant_id:x_tenant_id,user_id:x_user_id}app.post(/chat)asyncdefchat(req:ChatRequest,auth:dictDepends(get_current_tenant_user),):tenant_idauth[tenant_id]user_idauth[user_id]# 1. 生成或复用 session_idsession_idreq.session_idorstr(uuid.uuid4())# 2. 构造复合 thread_id服务端拼接不信任客户端直接传 thread_idthread_idbuild_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 不同上下文完全隔离resultawaitagent.ainvoke({messages:[{role:user,content:req.message}]},configconfig,)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_clientredis.Redis(hostlocalhost,port6379,db0,decode_responsesTrue)defget_session_history(user_id:str,conversation_id:str)-RedisChatMessageHistory: 工厂函数session_id 直接拼成 user:{user_id}:conv:{conversation_id} Redis 里按这个 key 做物理分区不同用户/会话天然隔离 composite_keyfuser:{user_id}:conv:{conversation_id}returnRedisChatMessageHistory(session_idcomposite_key,redis_clientredis_client)# 用 user_id conversation_id 双键做历史工厂chain_with_historyRunnableWithMessageHistory(chain,get_session_history,input_messages_keyinput,history_messages_keyhistory,history_factory_config[ConfigurableFieldSpec(iduser_id,annotationstr,nameUser ID,descriptionUnique identifier for the user.,is_sharedTrue,),ConfigurableFieldSpec(idconversation_id,annotationstr,nameConversation ID,descriptionUnique identifier for the conversation.,is_sharedTrue,),],)# 调用时注入双键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():aliceget_client(urlhttp://localhost:60058,headers{Authorization:Bearer user1-token})bobget_client(urlhttp://localhost:60058,headers{Authorization:Bearer user2-token})# Alice 写长期记忆awaitalice.store.put_item([memories],keynote,value{text:Alice private note})# Bob 读不到 Alice 的数据bob_itemawaitbob.store.get_item([memories],keynote)assertbob_itemisNone,❌ Bob 不应看到 Alice 的 store 数据# 各自写自己的awaitbob.store.put_item([memories],keynote,value{text:Bob private note})alice_itemawaitalice.store.get_item([memories],keynote)assertalice_item[value][text]Alice private noteprint(✅ 长期记忆隔离验证通过)asyncio.run(main())并发测试建议用threading起 10 个线程5 个模拟用户 A、5 个模拟用户 B 同时发消息断言彼此回复中不出现对方上下文——这能抓出单用户测试发现不了的竞态 bug。四、总结防串台的本质让每一次 Agent 调用都携带服务端生成的、经过鉴权校验的、全局唯一的会话标识并通过持久化后端的物理分区Postgres / Redis 的 key 隔离落地。三层防御体系身份层JWT / API Key 解析tenant_iduser_id绝不信任客户端传入的thread_id会话层复合thread_id org:{tenant_id}:user:{user_id}:session:{session_id}Checkpointer 按此物理分区记忆层Store 的 namespace 要么由 Auth 层自动加用户前缀要么在应用代码里显式拼user_idLangGraph vs LangChain 传统链LangGraph用AsyncPostgresSaver/AsyncRedisSaver做 Checkpointer PostgresStore做 Storeconfig[configurable][thread_id]是隔离主键LangChain LCEL用RunnableWithMessageHistoryRedisChatMessageHistorysession_id做隔离主键支持user_idconversation_id双键工厂隔离级别选型级别实现方式适用场景逻辑隔离复合 thread_id 命名空间中小规模、互信租户物理隔离每租户独立数据库实例大规模、高安全/合规需求混合隔离按租户重要性分级多级别安全需求⚠️ 三个最容易踩的坑① 直接用user_id当thread_id→ 同一用户多窗口并发会状态覆盖② 用InMemoryChatMessageHistory上生产 → 重启丢数据、多 worker 不共享③ 滥用checkpoint_ns做多租户 → 那是 LangGraph 内部子图层级标识会让时间旅行/调试元数据乱掉生产加固清单✅ 应用层conversations表 RLS 双重保险✅thread_id用 UUID/ULID 保证全局唯一✅ 长期记忆 namespace 与 thread_id 双重隔离✅ 过期会话定时归档清理如 48h 无活跃✅ LangSmith 按thread_id打标追踪✅ 并发隔离测试纳入 CI按这套方案落地数百租户同时在线、每租户多会话并发都能做到上下文零串台。
