在多租户场景下保证 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：选择持久化后端

场景	推荐方案
开发 / 单租户原型	MemorySaver / InMemoryStore
生产 - 逻辑隔离	AsyncPostgresSaver + PostgresStore
生产 - 高并发低延迟	AsyncRedisSaver + RedisStore
生产 - 高安全需求	物理隔离：每租户独立数据库实例

📌 生产环境绝对不要用 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 关系在这里管
from sqlalchemy import Column, String, DateTime, UUID
from sqlalchemy.ext.declarative import declarative_base
import uuid

Base = declarative_base()

class Conversation(Base):
    __tablename__ = "conversations"
    id = Column(UUID, primary_key=True, default=uuid.uuid4)  # 这个就是 thread_id
    tenant_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 被绕过，也无法跨租户读数据
ALTER TABLE conversations ENABLE ROW LEVEL SECURITY;
CREATE POLICY conversations_rls ON conversations
  USING (tenant_id = current_setting('app.tenant_id')::uuid 
     AND user_id = current_setting('app.user_id')::uuid);

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

# agent.py
import uuid
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.store.postgres import PostgresStore
from langchain_openai import ChatOpenAI

# 全局共享同一个 Graph 定义（无状态），状态由 checkpointer 按 thread_id 隔离
async def build_agent():
    # 1. 短期记忆：Checkpointer，按 thread_id 物理分区
    checkpointer = AsyncPostgresSaver.from_conn_string(
        "postgres://user:pass@localhost:5432/langgraph"
    )
    await checkpointer.setup()

    # 2. 长期记忆：Store，按 namespace 隔离
    store = PostgresStore.from_conn_string(
        "postgres://user:pass@localhost:5432/langgraph"
    )
    await store.setup()

    # 3. 编译图（checkpointer + store 一起挂）
    llm = ChatOpenAI(model="gpt-4o-mini")
    builder = StateGraph(MessagesState)

    async def assistant_node(state: MessagesState, *, store, config):
        # 长期记忆读写示例：从 store 取用户偏好
        user_id = config["configurable"]["langgraph_auth_user_id"]
        # 显式作用域方案（方案B）：namespace 里带上 user_id
        pref = await store.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 和 store
    return builder.compile(checkpointer=checkpointer, store=store)

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

# thread_utils.py
import re
from fastapi import HTTPException

def build_thread_id(tenant_id: str, user_id: str, session_id: str) -> str:
    """生成全局唯一的复合 thread_id"""
    return f"org:{tenant_id}:user:{user_id}:session:{session_id}"

def parse_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)
    if not m:
        raise HTTPException(status_code=400, detail="Invalid thread_id format")
    if m.group("t") != tenant_id or m.group("u") != user_id:
        raise HTTPException(status_code=403, detail="Access denied to this thread")

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

# main.py
from fastapi import FastAPI, Header, Depends, HTTPException
from pydantic import BaseModel
from typing import Optional
import uuid

from agent import build_agent
from thread_utils import build_thread_id, parse_and_verify_thread_id

app = FastAPI()
agent = None  # 启动时初始化

@app.on_event("startup")
async def startup():
    global agent
    agent = await build_agent()

class ChatRequest(BaseModel):
    message: str
    session_id: Optional[str] = None  # 客户端可选传，不传则新建

def get_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 并校验签名、过期时间
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"tenant_id": x_tenant_id, "user_id": x_user_id}

@app.post("/chat")
async def chat(
    req: ChatRequest,
    auth: dict = Depends(get_current_tenant_user),
):
    tenant_id = auth["tenant_id"]
    user_id = auth["user_id"]

    # 1. 生成或复用 session_id
    session_id = req.session_id or str(uuid.uuid4())

    # 2. 构造复合 thread_id（服务端拼接，不信任客户端直接传 thread_id）
    thread_id = build_thread_id(tenant_id, user_id, session_id)

    # 3. 如果客户端传了 session_id，校验其归属（防越权）
    if req.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 = await agent.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.py
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import ConfigurableFieldSpec
from langchain_community.chat_message_histories import RedisChatMessageHistory
import redis

redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

def get_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}"
    return RedisChatMessageHistory(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,
        ),
    ],
)

# 调用时注入双键
await chain_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.py
import asyncio
from langgraph_sdk import get_client

async def main():
    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 写长期记忆
    await alice.store.put_item(["memories"], key="note", 
                                value={"text": "Alice private note"})
    
    # Bob 读不到 Alice 的数据
    bob_item = await bob.store.get_item(["memories"], key="note")
    assert bob_item is None, "❌ Bob 不应看到 Alice 的 store 数据"
    
    # 各自写自己的
    await bob.store.put_item(["memories"], key="note", 
                             value={"text": "Bob private note"})
    alice_item = await alice.store.get_item(["memories"], key="note")
    assert alice_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 双键工厂

隔离级别选型：

级别	实现方式	适用场景
逻辑隔离	复合 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

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