LangGraph源码学习路线：第 14 篇 Memory / Store 与上下文工程源码解剖#

1
Checkpoint / Streaming
2
  ↓
3
Memory / Store / Context Engineering
4
  ↓
5
Middleware / Guardrails
6
  ↓
7
工程化 Agent 能力

1
from langgraph.graph import StateGraph, MessagesState
2
from langgraph.checkpoint.memory import InMemorySaver
3
from langgraph.store.memory import InMemoryStore
4
from langchain_core.messages import HumanMessage, SystemMessage
5

6
checkpointer = InMemorySaver()
7
store = InMemoryStore()
8

9
def call_model(state: MessagesState, config, *, store: InMemoryStore):
10
    user_id = config["configurable"]["user_id"]
11
    namespace = ("travel_assistant", user_id, "preferences")
12

13
    memories = store.search(namespace, query="travel preferences")
14
    preference_text = format_preferences(memories)
15
    visible_messages = trim_messages(state["messages"], max_tokens=3000)
16

17
    model_messages = [
18
        SystemMessage(content=f"已知长期偏好：{preference_text}"),
19
        *visible_messages,
20
    ]
21
    response = model.invoke(model_messages)
22
    return {"messages": [response]}
23

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

26
result = graph.invoke(
27
    {"messages": [HumanMessage(content="帮我规划大阪 4 天行程，少走路。")]},
28
    config={"configurable": {"thread_id": "trip-2026-osaka", "user_id": "u_123"}},
29
)

1
本轮 input: {"messages": [HumanMessage(...)]}
2
  ↓
3
RunnableConfig.configurable: {thread_id, user_id}
4
  ↓
5
checkpointer.get_tuple(thread_id)
6
  ↓
7
恢复上一轮 StateSnapshot / channel values
8
  ↓
9
messages reducer 合并本轮 HumanMessage
10
  ↓
11
节点通过 store.search(namespace, query) 读取长期记忆
12
  ↓
13
trim_messages(state["messages"]) + memory injection
14
  ↓
15
model.invoke(list[BaseMessage])
16
  ↓
17
节点返回 {"messages": [AIMessage(...)]}
18
  ↓
19
checkpointer.put(...) 保存新的 thread state
20
  ↓
21
可选：store.put(namespace, key, value) 保存跨线程长期记忆

1
def load_travel_preferences(store, user_id: str, current_request: str) -> list[str]:
2
    namespace = ("travel_assistant", user_id, "preferences")
3
    items = store.search(namespace, query=current_request, limit=5)
4
    return [item.value["text"] for item in items if item.value.get("confidence", 0) >= 0.7]
5

6

7
def build_model_messages(state, store, config):
8
    user_id = config["configurable"]["user_id"]
9
    last_user_text = state["messages"][-1].content
10
    preferences = load_travel_preferences(store, user_id, last_user_text)
11

12
    injected_memory = SystemMessage(
13
        content="长期偏好：" + "；".join(preferences) if preferences else "暂无可用长期偏好"
14
    )
15
    recent_messages = trim_messages(state["messages"], max_tokens=3000)
16
    return [injected_memory, *recent_messages]

1
langgraph/
2
├── graph/
3
├── pregel/
4
├── types.py
5
└── func/
6

7
langchain_core/
8
├── runnables/
9
├── tools/
10
├── retrievers.py
11
└── messages/

1
1. 先读官方概念文档，确认公共契约。
2
2. 再读公开 API 或装饰器入口。
3
3. 顺着构建期对象进入 runtime。
4
4. 追踪一次 invoke / stream / tool call / task call。
5
5. 单独检查异常、重试、恢复和扩展点。
6
6. 最后回到旅行规划助手做工程判断。

1
Public API / Decorator / Tool Protocol
2
  ↓
3
Runtime wrapper
4
  ↓
5
State / Config / Context
6
  ↓
7
Storage / Model / Tool / Graph Runtime

1
公共 API
2
  负责：给业务代码稳定入口。
3
  不负责：暴露所有内部调度细节。
4

5
运行时协议
6
  负责：让状态、配置、工具、模型和持久化协作。
7
  不负责：替业务判断所有策略。
8

9
扩展接口
10
  负责：允许业务插入可维护的定制逻辑。
11
  不负责：保证错误扩展仍然安全。

1
先找公开入口
2
  ↓
3
确认输入输出类型
4
  ↓
5
沿调用方追到核心实现
6
  ↓
7
记录状态与对象变化
8
  ↓
9
检查分支、异常和结束条件
10
  ↓
11
回到设计目的

1
def compile_travel_graph(builder, checkpointer, store):
2
    compiled = builder.compile(
3
        checkpointer=checkpointer,
4
        store=store,
5
    )
6
    return compiled

1
def state_graph_compile(builder, checkpointer=None, *, store=None):
2
    builder.validate_graph()
3
    state_channels = builder.build_state_channels()
4
    node_specs = builder.attach_nodes_and_edges()
5

6
    compiled = CompiledStateGraph(
7
        builder=builder,
8
        channels=state_channels,
9
        nodes=node_specs,
10
        checkpointer=checkpointer,
11
        store=store,
12
    )
13
    return compiled.validate()

1
def preference_namespace(config) -> tuple[str, ...]:
2
    user_id = config["configurable"]["user_id"]
3
    return ("travel_assistant", user_id, "preferences")
4

5

6
def remember_preference(store, config, fact: dict):
7
    namespace = preference_namespace(config)
8
    key = fact["id"]
9
    value = {
10
        "text": fact["text"],
11
        "source_thread_id": config["configurable"]["thread_id"],
12
        "confidence": fact["confidence"],
13
    }
14
    store.put(namespace, key, value)

1
def make_context_builder(max_prompt_tokens: int):
2
    def build_context(state, store, config):
3
        user_id = config["configurable"]["user_id"]
4
        namespace = ("travel_assistant", user_id, "preferences")
5
        memories = store.search(namespace, query=state["messages"][-1].content, limit=5)
6

7
        memory_message = SystemMessage(
8
            content=format_memory_for_prompt(memories)
9
        )
10
        recent_messages = trim_messages(
11
            state["messages"],
12
            max_tokens=max_prompt_tokens,
13
            strategy="last",
14
            include_system=False,
15
        )
16
        return [memory_message, *recent_messages]
17

18
    return build_context

1
graph.invoke(input, config)
2
  ↓
3
解析 configurable.thread_id / user_id
4
  ↓
5
checkpointer 按 thread_id 读取上一轮 checkpoint
6
  ↓
7
合并本轮 input 到 state channels
8
  ↓
9
节点读取 store(namespace, query)
10
  ↓
11
裁剪 state["messages"] 并注入长期记忆
12
  ↓
13
model.invoke(list[BaseMessage])
14
  ↓
15
节点返回 state update
16
  ↓
17
checkpointer 保存新 thread state
18
  ↓
19
可选 store.put(...) 写长期记忆

1
def invoke_with_checkpoint(compiled, input, config):
2
    thread_id = config["configurable"].get("thread_id")
3
    if thread_id is None:
4
        raise ValueError("checkpointed graph requires configurable.thread_id")
5

6
    checkpoint_tuple = compiled.checkpointer.get_tuple(config)
7
    previous_channels = checkpoint_tuple.checkpoint["channel_values"] if checkpoint_tuple else {}
8

9
    next_channels = merge_input_with_channels(
10
        previous_channels,
11
        input,
12
        reducers=compiled.channels,
13
    )
14
    return run_pregel_loop(compiled, next_channels, config)

1
def call_model_with_memory(state, config, store):
2
    user_id = config["configurable"]["user_id"]
3
    namespace = ("travel_assistant", user_id, "preferences")
4
    query = state["messages"][-1].content
5

6
    items = store.search(namespace, query=query, limit=5)
7
    preferences = [item.value["text"] for item in items]
8

9
    prompt_messages = [
10
        SystemMessage(content="用户长期偏好：" + "；".join(preferences)),
11
        *trim_messages(state["messages"], max_tokens=3000),
12
    ]
13
    ai_message = model.invoke(prompt_messages)
14
    return {"messages": [ai_message]}

1
def trim_and_inject_messages(state, injected_memory):
2
    raw_messages = state["messages"]
3
    visible_history = trim_messages(
4
        raw_messages,
5
        max_tokens=3000,
6
        strategy="last",
7
        include_system=False,
8
    )
9
    return [injected_memory, *visible_history]

1
def finish_turn(compiled, config, state_update, extracted_facts):
2
    new_state = apply_reducers(compiled.channels, state_update)
3
    compiled.checkpointer.put(config, checkpoint_from(new_state))
4

5
    if compiled.store is not None:
6
        for fact in extracted_facts:
7
            if fact["confidence"] >= 0.7 and fact["kind"] == "stable_preference":
8
                namespace = ("travel_assistant", fact["user_id"], "preferences")
9
                compiled.store.put(namespace, fact["id"], fact)
10

11
    return new_state

1
Input + config
2
  ↓
3
thread_id 恢复 checkpoint
4
  ↓
5
user_id 构造 namespace 并读取 store
6
  ↓
7
messages 裁剪 + memory 注入
8
  ↓
9
AIMessage state update
10
  ↓
11
checkpoint 保存短期状态 / store 保存长期记忆

1
def maybe_load_memory(store, config, state):
2
    if store is None:
3
        return SystemMessage(content="暂无长期记忆。")
4

5
    user_id = config["configurable"].get("user_id")
6
    if user_id is None:
7
        return SystemMessage(content="未启用用户级长期记忆。")
8

9
    namespace = ("travel_assistant", user_id, "preferences")
10
    items = store.search(namespace, query=state["messages"][-1].content, limit=5)
11
    return SystemMessage(content=format_memory_for_prompt(items))

1
def write_memory_if_stable(store, config, extracted_fact):
2
    if store is None:
3
        return "skipped:no_store"
4
    if extracted_fact["confidence"] < 0.7:
5
        return "skipped:low_confidence"
6
    if extracted_fact["kind"] not in {"stable_preference", "accessibility_need"}:
7
        return "skipped:not_long_term"
8

9
    namespace = ("travel_assistant", config["configurable"]["user_id"], "preferences")
10
    store.put(namespace, extracted_fact["id"], extracted_fact)
11
    return "written"

1
def build_budgeted_messages(state, memory_message, token_budget):
2
    memory_tokens = count_tokens([memory_message])
3
    remaining = token_budget - memory_tokens
4
    if remaining <= 0:
5
        return [SystemMessage(content="长期记忆过长，已省略。"), state["messages"][-1]]
6

7
    visible_history = trim_messages(state["messages"], max_tokens=remaining)
8
    if not visible_history:
9
        visible_history = [state["messages"][-1]]
10

11
    return [memory_message, *visible_history]

1
正常结束：模型返回 AIMessage，state update 被 reducer 合并，checkpoint 写入完成。
2
可降级结束：store 缺失或检索失败，但本轮可以基于短期 state 回答。
3
保护性结束：上下文预算不足、namespace 不可信或记忆可能污染时拒绝写 store。
4
异常失败：checkpoint 恢复失败、state schema 不匹配或 reducer 无法合并时向上抛出。

1
class TravelProfileStore(BaseStore):
2
    def put(self, namespace, key, value, *, index=None):
3
        record = serialize_memory(namespace, key, value)
4
        database.upsert(record)
5

6
    def search(self, namespace, *, query=None, filter=None, limit=10):
7
        rows = database.search(namespace=namespace, query=query, filter=filter, limit=limit)
8
        return [deserialize_item(row) for row in rows]

1
def extract_and_persist_preferences(store, config, messages):
2
    candidates = preference_extractor.invoke({"messages": messages})
3
    written = []
4

5
    for candidate in candidates:
6
        if candidate.confidence < 0.7:
7
            continue
8
        if candidate.scope != "long_term":
9
            continue
10

11
        namespace = ("travel_assistant", config["configurable"]["user_id"], "preferences")
12
        value = {
13
            "text": redact(candidate.text),
14
            "source": config["configurable"]["thread_id"],
15
            "confidence": candidate.confidence,
16
        }
17
        store.put(namespace, candidate.id, value)
18
        written.append(candidate.id)
19

20
    return written

1
def build_grounded_context(state, store, retriever, config):
2
    user_id = config["configurable"]["user_id"]
3
    namespace = ("travel_assistant", user_id, "preferences")
4

5
    preferences = store.search(namespace, query=state["messages"][-1].content, limit=3)
6
    docs = retriever.invoke(state["messages"][-1].content)
7

8
    memory_message = SystemMessage(content=format_preferences(preferences))
9
    evidence_messages = format_docs_as_messages(docs)
10
    recent_messages = trim_messages(state["messages"], max_tokens=2500)
11

12
    return [memory_message, *evidence_messages, *recent_messages]

1
性能：额外抽象会增加序列化、检索、trace 和存储成本。
2
可靠性：可恢复机制要求状态 schema、幂等工具和错误分类清楚。
3
安全：metadata、state、tool args、retrieved docs 都可能包含敏感信息。
4
可观测性：至少记录 thread_id、node/tool、策略命中、错误和耗时。

1
公开 API 通常只负责归一化和装配；真正的运行语义在 runtime、协议对象和扩展点之间流转。

1
state、store、Document、metadata、trace、prompt messages 是不同协议；只有经过明确注入的内容才进入模型上下文。

1
可恢复异常可以转成 retry、fallback、interrupt；不可恢复异常必须上抛并进入 trace。

1
Public Config
2
  ↓
3
Normalize
4
  ↓
5
Build Runtime Object
6
  ↓
7
Attach Protocol / Extension / Storage

1
Input
2
  ↓
3
Runtime Entry
4
  ↓
5
Context / State / Config
6
  ↓
7
Core Execute
8
  ↓
9
State / Event / Output

1
正常路径 → 返回协议对象并更新必要状态
2
分支路径 → 路由、降级、拒答、人工确认或恢复
3
可恢复异常 → retry / fallback / repair / resume
4
不可恢复异常 → trace + raise
5
保护上限 → 停止循环、预算或高风险动作

1
第 15 篇：Middleware / Guardrails 与 Agent 执行控制源码解剖

1
这项能力在旅行规划助手中应该放在哪一层？
2
它和前一篇能力如何协作？
3
哪些边界需要通过源码证据确认？