LangGraph 10：Store 与 Context

发表于 更新于 分类于
用 InMemoryStore 与 PostgresStore 做跨线程记忆，并用 Runtime 注入请求级上下文。

前言

上一篇的时光旅行仍走 Checkpoint：绑在某个 thread_id 的快照链上。
Store 是另一条记忆轴：跨线程的长期键值，不随某次对话的检查点链消失。
还有一类数据既不该进 State（会进检查点），也不适合放 Store（只对本次请求有效），用 Runtime context 按次注入。
本文对比三者，并给出内存 Store、PostgresStore 与 context_schema 示例。
内存版可直接跑；Postgres 需配置 DB_URL，无数据库时可跳过。
聊天模型对接 火山方舟 Coding Plan。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。
用 uv 初始化工程并声明依赖。

1
2
3
4
uv init langgraph-store
cd langgraph-store
uv venv --python 3.12
uv add "langchain>=1.0,<2.0" langchain-openai langchain-core langgraph python-dotenv rich

在项目根目录创建 .env：

1
2
OPENAI_API_KEY=你的火山方舟 API Key
OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

不要把 .env 提交进 Git。

示例

记忆轴对比

Checkpoint、Store、Context 都能「带点信息」，但作用域不同。

Checkpoint Store Context
作用域 单线程对话时间线 跨线程长期记忆 单次 invoke
典型内容 messages、图中间态 用户偏好、资料档案 租户 ID、会员等级
读取方式 get_state / history store.get / search runtime.context
注入图 checkpointer= store= + runtime.store context_schema + context=

同一用户换 thread_id，Checkpoint 是新会话，Store 里的档案还在。
context 每次 invoke 现给，默认不进检查点，也不进 Store。
检查点时间线的回放与分叉见上一篇 《LangGraph 09：时光旅行》。

InMemoryStore

用 InMemoryStore 写入偏好，节点里通过 Runtime 读取。
换 thread_id 仍能读到同一用户的档案。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import os
from typing import Final, TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.3,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

USERS_NS: Final = ("users",)
PREFERENCES_KEY: Final = "preferences"

store = InMemoryStore()
store.put(
    (*USERS_NS, "Alice"),
    PREFERENCES_KEY,
    {"course": "组成原理", "sports": "跑步", "food": "酸奶"},
)


class OverAllState(MessagesState):
    username: str
    user_input: str
    output: str
    preferences: dict[str, str]


def check_preference_node(state: OverAllState, runtime: Runtime) -> OverAllState:
    item = runtime.store.get((*USERS_NS, state["username"]), PREFERENCES_KEY)
    if item is None:
        return {}
    return {"preferences": item.value}


def llm_node(state: OverAllState) -> OverAllState:
    preference = state.get("preferences", {})
    prompt = f"用户偏好:{preference}\n用户需求:{state['user_input']}"
    resp = model.invoke(
        [
            SystemMessage("请根据用户偏好简洁回复。"),
            HumanMessage(prompt),
        ]
    )
    return {"messages": [resp], "output": resp.content}


builder = StateGraph(state_schema=OverAllState)
builder.add_node("check_preference_node", check_preference_node)
builder.add_node("llm_node", llm_node)
builder.add_edge(START, "check_preference_node")
builder.add_edge("check_preference_node", "llm_node")
builder.add_edge("llm_node", END)

graph = builder.compile(checkpointer=InMemorySaver(), store=store)
config = {"configurable": {"thread_id": "store-demo"}}
rprint(
    graph.invoke(
        {"username": "Alice", "user_input": "推荐一下吃的"},
        config=config,
    )
)

同一用户换 thread_id 仍可通过 Store 读到偏好；Checkpoint 则只服务当前线程。
无数据库时用 InMemoryStore 即可；API（put / get / search）与下一节 Postgres 版一致。

PostgresStore

跨进程、可重启时把 Store 换到 Postgres。
put / get 与内存版相同，只是用上下文管理器拿连接，并先 setup() 建表。

先安装可选依赖。

1
uv add "langgraph-checkpoint-postgres" "psycopg[binary,pool]"

在 .env 写入连接串，由 load_dotenv() 加载。

1
2
3
4
OPENAI_API_KEY=你的火山方舟 API Key
OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

DB_URL="postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable"

连接串从环境变量读取，勿把密码写进脚本。
未启动数据库时请跳过本节，继续用上文的内存示例。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import os
from typing import Final

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.runtime import Runtime
from langgraph.store.postgres import PostgresStore
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.3,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

DB_URL = os.environ["DB_URL"]
USERS_NS: Final = ("users",)
PREFERENCES_KEY: Final = "preferences"


class OverAllState(MessagesState):
    username: str
    user_input: str
    output: str
    preferences: dict[str, str]


def check_preference_node(state: OverAllState, runtime: Runtime) -> OverAllState:
    item = runtime.store.get((*USERS_NS, state["username"]), PREFERENCES_KEY)
    if item is None:
        return {}
    return {"preferences": item.value}


def llm_node(state: OverAllState) -> OverAllState:
    preference = state.get("preferences", {})
    prompt = f"用户偏好:{preference}\n用户需求:{state['user_input']}"
    resp = model.invoke(
        [
            SystemMessage("请根据用户偏好简洁回复。"),
            HumanMessage(prompt),
        ]
    )
    return {"messages": [resp], "output": resp.content}


builder = StateGraph(state_schema=OverAllState)
builder.add_node("check_preference_node", check_preference_node)
builder.add_node("llm_node", llm_node)
builder.add_edge(START, "check_preference_node")
builder.add_edge("check_preference_node", "llm_node")
builder.add_edge("llm_node", END)

with PostgresStore.from_conn_string(DB_URL) as store:
    store.setup()  # 首次建表，幂等
    store.put(
        (*USERS_NS, "Alice"),
        PREFERENCES_KEY,
        {"course": "组成原理", "sports": "跑步", "food": "酸奶"},
    )
    graph = builder.compile(checkpointer=InMemorySaver(), store=store)
    config = {"configurable": {"thread_id": "store-pg-demo"}}
    rprint(
        graph.invoke(
            {"username": "Alice", "user_input": "推荐一下吃的"},
            config=config,
        )
    )

图逻辑与内存版相同，只是 Store 换了后端。
关掉进程再跑，只要连的是同一库，Alice 的偏好仍在。
会话检查点若也要落库，把 InMemorySaver 换成 PostgresSaver，写法见 《LangGraph 08：Checkpoint 持久化》。

Runtime 上下文

会员等级、租户 ID 这类请求级数据，用 context_schema + invoke(..., context=...)。
节点通过 runtime.context 读取，默认不进入 Checkpoint，也不进 Store。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import os
from dataclasses import dataclass

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.runtime import Runtime
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.2,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


@dataclass
class UserContext:
    username: str
    membership_level: str


class OverAllState(MessagesState):
    user_input: str
    output: str


def llm_node(state: OverAllState, runtime: Runtime[UserContext]) -> OverAllState:
    ctx = runtime.context
    if ctx and ctx.membership_level == "VIP":
        system = (
            f"你是高级助理，当前 VIP 用户是{ctx.username}，"
            "请用『您』称呼，末尾加『VIP服务』。"
        )
    elif ctx:
        system = f"你是助理，当前用户是{ctx.username}，请简洁友好回复。"
    else:
        system = "你是助理，请简洁友好回复。"

    text = model.invoke(
        [SystemMessage(system), HumanMessage(state["user_input"])]
    ).content
    return {"output": text}


builder = StateGraph(state_schema=OverAllState, context_schema=UserContext)
builder.add_node("llm_node", llm_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", END)
graph = builder.compile()

rprint(
    graph.invoke(
        {"user_input": "最近有什么优惠？"},
        context=UserContext(username="Alice", membership_level="VIP"),
    )
)
rprint(
    graph.invoke(
        {"user_input": "最近有什么优惠？"},
        context=UserContext(username="Bob", membership_level="普通用户"),
    )
)

context 按次注入。
需要审计时可自行把关键字段拷进 State。

验证

  1. Store：Alice 的回复应提到酸奶等已写入偏好；换 thread_id 仍能读到。
  2. PostgresStore：配置 DB_URL 后重跑脚本，偏好仍能读到；无库时跳过。
  3. Context：VIP 与普通用户的语气、结尾应明显不同。

总结

  1. Store vs Checkpoint：长期跨线程用 Store；会话时间线用 Checkpoint。
  2. 落库：演示用 InMemoryStore；生产用 PostgresStore，连接串从 DB_URL 读取。
  3. Runtime context：请求级元数据用 context_schema，避免污染状态快照。