LangChain 进阶：Runtime 三大核心——Context / State / Store 详解

对象	本质	是否可变	生命周期	典型数据	存储能力
Context	运行时静态上下文（入参配置）	❌ 不可变	单次 invoke/run 内，一次调用全程不变	user_id、tenant_id、权限标识、环境标记	不持久，调用传入，执行结束销毁
State	Graph 状态（短时会话内存）	✅ 可变	同一个 thread_id 会话内，Graph 流转过程	messages、中间结果、计数器、当前任务参数	靠 Checkpointer 做会话断点持久（线程内）
Store	跨线程 KV 存储（长时记忆）	✅ 可变	跨 thread、跨会话、长期存在	用户偏好、用户档案、长期记忆、知识库事实	BaseStore 实现：内存/PG/SQLite/Mongo，独立于 checkpoint

关键区分： Checkpointer 存的是 State 的快照（会话断点）；Store 存的是业务自定义的全局/用户维度数据，不属于 Graph State 本身 Runtime 是载体：节点函数签名增加 runtime: Runtime[Context] 参数，LangGraph 自动注入，不需要手动传

二、逐个拆解：概念、为什么需要、怎么用

1. Context（静态运行上下文）

概念

Context 是调用 Graph 时一次性传入、在本次执行链路全程只读不变的强类型配置对象，不属于 State，不会被 reducer 修改，不会被 checkpointer 保存。

注意：Context 不是 prompt 上下文，是程序运行身份/环境上下文

为什么需要

很多业务信息不适合塞到 State、也不适合硬编码、也不适合放到 prompt：

user_id / tenant_id / app_id：身份鉴权、数据隔离；
权限等级、功能开关、模型版本标记；
链路 trace_id、请求来源标记；如果把 user_id 放进 State，容易被 LLM/工具意外篡改；Context 只读，保证身份在整个执行链路稳定可信。

怎么使用

dataclass 定义类型；
invoke 的 config 中传入 context={...}；
节点函数声明 runtime: Runtime[MyContext]；
读取：runtime.context.user_id，只可读，不能修改。

2. State（Graph 会话状态，短时内存）

概念

State 是 LangGraph 图的核心，代表当前对话线程内动态变化的数据，由 Channel + Reducer 定义更新规则；每走完一个节点，State 按 reducer 规则合并更新；搭配 MemorySaver/PostgresCheckpointer 可以断点续跑、回溯。

就是我们最熟悉的 MessagesState，本质就是内置 State：消息历史、中间输出、工具返回值都放这里

为什么需要

Agent 是多步流转（LLM→工具→LLM），单轮 LLM 调用无法完成任务：

保存对话历史；
保存工具返回结果、中间计算值；
分支判断依赖临时变量；
支持 human-in-the-loop、断点重试、时间旅行（time travel）调试； State 定义了 Graph 的“状态机”，没有 State 就没有 LangGraph 的循环 Agent。

怎么使用

继承 TypedDict / dataclass 定义 State，指定 channel 和 reducer；
StateGraph(MyState) 构建图；
节点第一个参数永远是 state: MyState；
return 字典，按 reducer 规则合并回 state；
compile 时传入 checkpointer 开启会话持久化（同一个 thread_id 恢复状态）。

重点：State 是线程（thread）隔离，换一个 thread_id 就是全新会话。

3. Store（跨线程持久存储，长时记忆）

概念

Store 是 LangGraph 提供的独立 KV 存储抽象 BaseStore，数据按 namespace + key 组织，支持 put / get / delete / search；不属于 Graph State，不受 thread 限制，同一个 user_id，换不同 thread_id 都能读到同一份记忆/档案。

典型实现：InMemoryStore、PostgresStore、SQLiteStore、MongoStore

为什么需要

State + Checkpointer 只能记住当前这一次聊天会话；业务需要跨会话记忆：

用户偏好：“我喜欢简洁回答”；
用户个人信息：姓名、职业、联系方式；
长期事实记忆：用户告诉过 Agent 的个人信息；
租户知识库、共享业务字典；如果全部塞进 State，每次会话都要重复说；Store 解决跨会话记忆。

怎么使用

实例化 Store；
compile(graph, store=store)；
节点通过 runtime.store 访问；
读写 API：
- await runtime.store.put(namespace, key, value)
- await runtime.store.get(namespace, key)
- await runtime.store.search(namespace, query=...)（带向量检索）
  
  namespace 一般用元组，用来做数据隔离：(user_id, "memories")、(tenant_id, "profile")

三、完整可运行代码演示（Python）

依赖安装

pip install langgraph langchain-openai python-dotenv

from dataclasses import dataclass
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage

# ====================== 1. 定义 Context（只读静态上下文） ======================
@dataclass
class AgentContext:
    user_id: str
    tenant_id: str
    permission: str

# ====================== 2. 定义 State（会话短时状态） ======================
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    tool_tip: str  # 本次会话临时字段

# ====================== 3. 初始化：Store（长时存储）、Checkpointer（State断点） ======================
store = InMemoryStore()
checkpointer = MemorySaver()
llm = ChatOpenAI(model="gpt-4o-mini")

# ====================== 4. 定义节点：注入 Runtime[AgentContext] ======================
async def load_user_profile(state: AgentState, runtime: Runtime[AgentContext]) -> AgentState:
    """从 Store 读取用户长期档案，注入到 prompt"""
    ctx = runtime.context
    namespace = (ctx.user_id, "profile")
    user_profile = await runtime.store.get(namespace, "info")
    profile_text = user_profile.value if user_profile else "暂无用户档案"

    new_msg = AIMessage(content=f"[读取档案]你的档案：{profile_text}")
    return {"messages": [new_msg]}

async def save_user_profile(state: AgentState, runtime: Runtime[AgentContext]) -> AgentState:
    """把用户信息写入 Store，跨会话永久保存"""
    ctx = runtime.context
    namespace = (ctx.user_id, "profile")
    # 模拟：提取用户信息写入长时Store
    await runtime.store.put(namespace, "info", {"name": "张三", "hobby": "爬山"})
    return {"messages": [AIMessage(content="✅ 用户档案已保存到长期Store")]}

async def chat_node(state: AgentState, runtime: Runtime[AgentContext]) -> AgentState:
    """普通对话节点，同时读取context身份信息"""
    ctx = runtime.context
    prompt_prefix = f"当前用户ID:{ctx.user_id},租户:{ctx.tenant_id},权限:{ctx.permission}\n"
    resp = await llm.ainvoke([HumanMessage(content=prompt_prefix)] + state["messages"])
    return {"messages": [resp]}

# ====================== 5. 构建 & 编译图（绑定 checkpointer + store） ======================
builder = StateGraph(AgentState)
builder.add_node("load_profile", load_user_profile)
builder.add_node("save_profile", save_user_profile)
builder.add_node("chat", chat_node)

builder.add_edge(START, "load_profile")
builder.add_edge("load_profile", "chat")
builder.add_edge("chat", "save_profile")
builder.add_edge("save_profile", END)

graph = builder.compile(
    checkpointer=checkpointer,
    store=store
)

# ====================== 6. 调用演示 ======================
async def demo():
    # thread_id 代表会话；context 是本次运行静态上下文
    config = {
        "configurable": {
            "thread_id": "session_001",
        },
        "context": AgentContext(
            user_id="u_1001",
            tenant_id="t_001",
            permission="admin"
        )
    }

    # 第一轮会话
    res1 = await graph.ainvoke(
        {"messages": [HumanMessage("你好，我是张三")]},
        config=config
    )
    print("===== 第一轮输出 =====")
    for m in res1["messages"]:
        print(m.content)

    # 换一个全新 thread_id（新会话！），但同一个 user_id，Store 依然能读到档案
    config_new_thread = {
        "configurable": {"thread_id": "session_002"},
        "context": AgentContext(user_id="u_1001", tenant_id="t_001", permission="admin")
    }
    res2 = await graph.ainvoke(
        {"messages": [HumanMessage("还记得我是谁吗？")]},
        config=config_new_thread
    )
    print("\n===== 新会话（新thread_id）输出 =====")
    for m in res2["messages"]:
        print(m.content)

if __name__ == "__main__":
    import asyncio
    asyncio.run(demo())

四、常见踩坑 & 最佳实践

❌ 不要在节点里尝试修改 runtime.context：它是只读，修改不会生效；身份/权限放这里；
❌ 不要把长期记忆写 State：换 thread 就丢；长期信息进 Store；
❌ 不要把鉴权信息塞 State：LLM 输出、工具返回容易污染/篡改；
✅ State：只放本轮会话流转需要的数据（消息、临时中间变量）；
✅ Store 按 namespace 做隔离，优先 (user_id, type) 设计，方便检索；
✅ Checkpointer 只管 State 断点；Store 是独立存储，二者互补；
✅ 生产环境替换 InMemoryStore 为 PostgresStore，支持向量 search 做记忆检索。

五、一句话总结（面试口述版）

Runtime 是 LangGraph 新版的运行注入对象，上面挂载三个东西： Context 是调用时传入、全程只读的身份配置，用来做鉴权隔离； State 是图的会话状态，单 thread 内可变短时内存，靠 checkpointer 断点； Store 是跨线程的 KV 长时存储，用来存用户档案、跨会话记忆，三者配合，就可以实现可鉴权、会话记忆、长期记忆的生产级 Agent。