LangChain 09 - 上下文与记忆学习导航与实战

模块 1：你只需要记住这些

---

1. 为什么需要记忆

大模型本身是无状态的——每次调用 agent.invoke() 都是全新的开始，不记得之前的对话。

举例：

• 情况1：第一轮告诉 Agent"我叫小明"，第二轮问"我叫什么"→ Agent 不知道
• 情况2：如果把前一轮的对话历史也传入 messages 列表 → Agent 能答出来

结论：Agent 不会自动记住上下文，需要额外的记忆模块保存历史交互信息。

2. 上下文工程

记忆（Memory）：专门负责"存储历史交互信息"的组件，核心作用是「保存上下文」和「提供上下文」。

上下文工程（Context Engineering）：负责"合理组织"这些记忆和任务信息，让 LLM 的响应更连贯、更贴合需求。

LangGraph 提供了三种管理上下文的方法，结合了可变性和生命周期维度：

上下文类型	描述	可变性	生命周期	访问方法
动态运行时上下文	在单次运行中会演变的可变数据	动态	单次运行	LangGraph state 对象
动态跨会话上下文	在对话间共享的持久数据（用户偏好、历史洞察、知识条目）	动态	跨对话	LangGraph store 对象
静态运行时上下文	在启动时传入的用户元数据、工具、数据库连接	静态	单次运行	LangGraph context 对象

3. 记忆的分类

LangChain 的记忆分为短期记忆和长期记忆：

短期记忆（Short-term memory / 会话级记忆 / thread-scoped memory）：

• 作用范围：单个对话线程（Thread）内
• 一旦开启新对话（更换 thread_id），记忆即消失
• 对应：state 对象

长期记忆（Long-term memory / 跨会话级记忆）：

• 在会话间存储用户特定或应用级数据
• 在会话线程间共享
• 记忆范围是任意自定义命名空间，不仅仅是单一线程 ID
• 对应：store 对象

版本演进：

• LangChain v0.x：通过专用的 xxxMemory 类管理记忆
• LangChain v1.x：Agent 构建在 LangGraph 之上，通过 state 和 store 构建记忆系统，使用更简单、功能更统一

4. 短期记忆三要素

LangChain 1.x 的短期记忆是三者的组合：

短期记忆 = State（会话内部状态）+ Checkpointer（持久化机制）+ Thread ID（会话作用域）

要素	作用
State	默认存储历史消息列表 messages，通过 State 管理历史消息
Checkpointer	负责将 State 作为检查点持久化保存，检查点是某个时刻的 State 快照
Thread ID	用于唯一标识 State，LangChain 运行时按照 thread_id 读写 State 快照

类比：就像玩 RPG 游戏时的"自动存档"——你不需要手动保存，系统在关键节点自动记录，下次进入游戏随时可以从上次的存档点继续。

5. InMemorySaver — 基于内存的持久化器

最便捷的使用方式，适合快速测试或调试。

使用步骤：

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.messages import HumanMessage

# 第1步：初始化记忆引擎
checkpointer = InMemorySaver()

# 第2步：绑定 Agent
agent = create_agent(
    model=model,
    tools=[],
    checkpointer=checkpointer  # 添加内存管理
)

# 第3步：设定会话 ID
config = {
    "configurable": {
        "thread_id": "1"
    }
}

# 第1轮对话
response1 = agent.invoke(
    {"messages": [HumanMessage("我叫张三")]},
    config=config
)

# 第2轮对话（同一个 thread_id，记得！）
response2 = agent.invoke(
    {"messages": [HumanMessage("我叫什么？")]},
    config=config
)
# Agent 能回答："你叫张三"

关键三步：

1. 初始化记忆引擎：checkpointer = InMemorySaver()
2. 绑定 Agent：在 create_agent 时传入 checkpointer
3. 设定会话 ID：通过 config = {"configurable": {"thread_id": "1"}} 指定线程标识

6. thread_id 的作用 — 会话隔离

同一个 thread_id 共享记忆，不同 thread_id 完全隔离。

# 会话 1
config1 = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": [HumanMessage("我叫张三")]}, config=config1)
agent.invoke({"messages": [HumanMessage("我叫什么？")]}, config=config1)  # 记得！

# 会话 2（完全独立）
config2 = {"configurable": {"thread_id": "2"}}
agent.invoke({"messages": [HumanMessage("你还记得我叫什么名字么？")]}, config=config2)
# Agent 不知道，因为会话 2 和会话 1 记忆空间隔离

生产场景：

• 多用户聊天：不同 thread_id = 不同用户会话
• 同一用户的不同任务：不同 thread_id = 不同任务上下文

7. 查看和更新线程状态

查看当前状态：

from rich import print as rprint
latest_state = agent.get_state(config)
rprint(latest_state)

State 快照包含完整的信息：消息列表、下一步节点（next）、配置（config）、元数据（metadata）、检查点 ID 等。

更新线程 ID：使用新的 thread_id 会重新开启对话，不会加载之前的记忆。

8. Checkpointer 工作原理

当你调用 agent.invoke() 时，checkpointer 自动执行四步：

① 读取之前的历史
② 追加新消息
③ 调用模型（传入完整历史）
④ 保存新的历史

关键：你只需要传新消息，checkpointer 会自动管理历史。

# 你只需要传新消息
agent.invoke(
    {"messages": [{"role": "user", "content": "新问题"}]},
    config
)
# checkpointer 自动：
# 1. 读取 thread_id 对应的历史消息
# 2. 追加新的 HumanMessage
# 3. 调用模型（传入完整历史）
# 4. 保存 AIMessage 到历史

9. 常见问题排查

问题1：为什么 Agent 不记得？

检查三个条件：

• 是否添加了 checkpointer=InMemorySaver()？
• 是否传入了 config 参数？
• 两次调用的 thread_id 是否相同？

# ❌ 错误：没有 checkpointer
agent = create_agent(model=model, tools=[])

# ❌ 错误：没有 config
agent = create_agent(model=model, tools=[], checkpointer=InMemorySaver())
agent.invoke({...})  # 不会记住

# ❌ 错误：thread_id 不同
agent.invoke({...}, config={"configurable": {"thread_id": "1"}})
agent.invoke({...}, config={"configurable": {"thread_id": "2"}})  # 不同会话

# ✅ 正确
agent = create_agent(model=model, tools=[], checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "1"}}
agent.invoke({...}, config)
agent.invoke({...}, config)  # 记得！

问题2：InMemorySaver 会丢失数据吗？

会！InMemorySaver 只保存在内存中：

• 同一进程内有效（不支持跨进程共享）
• 程序重启后丢失
• 不同进程无法共享

解决方案：使用持久化存储（SQLite、PostgreSQL）

问题3：内存会无限增长吗？

会！默认情况下，InMemorySaver 会保存所有消息。问题包括：

• 消息越来越多，token 消耗增加，可能超过模型限制
• 响应速度变慢、成本增加

解决方案：上下文管理（消息裁剪、消息删除、消息摘要）

问题4：如何清空某个会话的历史？

InMemorySaver 没有提供删除 API。临时方案：使用新的 thread_id 或重新创建 Agent。

10. PostgresSaver — 基于外部存储介质的持久化器

生产环境要用持久化的外部存储介质，如 PostgreSQL。

使用步骤：

from langgraph.checkpoint.postgres import PostgresSaver

DB_URL = "postgresql://langchain_user:abcd1234@118.195.128.47:5432/langchain_db?sslmode=disable"

with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    # 初始化 PostgreSQL 数据库（首次运行创建表）
    checkpointer.setup()
    
    agent = create_agent(
        model=model,
        checkpointer=checkpointer
    )
    
    config = {"configurable": {"thread_id": "1"}}
    response1 = agent.invoke(
        {"messages": [HumanMessage("你好，我是老王")]},
        config=config
    )
    response2 = agent.invoke(
        {"messages": [HumanMessage("你好，我是谁？")]},
        config=config
    )
    # Agent 能回答："你是老王"

关键点：

• checkpointer.setup() 首次运行创建必要的表，重复执行不会重新建表（底层是 CREATE IF NOT EXISTS）
• 必须在 with 语句块中使用，确保连接正确管理
• 需要额外依赖：pip install langgraph-checkpoint-postgres

PostgreSQL 表结构（setup() 创建四张表）：

表名	说明
checkpoints	主表，存每个 thread 在某个时刻的 checkpoint 快照
checkpoint_blobs	存不适合直接内联进 checkpoints 表的较复杂 channel 值
checkpoint_writes	存中间写入 / pending writes，不是最终完整 checkpoint
checkpoint_migrations	迁移版本表，非业务数据表

11. InMemorySaver vs PostgresSaver 对比

核心区别：进程结束后状态是否保留。

特性	InMemorySaver	PostgresSaver
存储介质	内存	PostgreSQL 数据库
进程结束后	数据丢失	数据保留
重建 Saver() 后	历史状态丢失	历史状态可通过 thread_id 恢复
适用场景	测试、调试	生产环境
跨进程共享	不支持	支持

验证实验：

# 举例1：基于内存存储
# 每次运行创建新的 InMemorySaver()，历史 State 被丢弃
# → 三次 invoke 的输出完全相同（每次从头开始）

# 举例2：基于外部存储器存储
# 每次运行创建新的 PostgresSaver()，但连接同一个数据库
# → 只要 thread_id 一致，历史状态可以和当前调用串联起来（状态是累积的）

结论：即便重新创建 Saver()，只要 thread_id 一致，PostgresSaver 的历史状态就可以通过数据库加载恢复。

12. 记忆治理策略（上下文管理）

随着对话进行，历史消息不断累积，带来三大挑战：

1. LLM 的上下文窗口有限，完整历史可能无法装入
2. 即便窗口够大，模型会被陈旧或离题的内容"分散注意力"
3. 高昂的 token 花费

需要对上下文进行管理：消息裁剪、消息删除、消息摘要。

12.1 消息裁剪（before_model）

目标：调用模型前裁剪上下文，控制 token 用量。
策略：保留系统初始消息和最近若干消息。
时机：在 before_model 钩子中执行。

from langchain_core.messages import HumanMessage
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig
from typing import Any

@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    messages = state["messages"]
    if len(messages) <= 3:
        return None
    first_msg = messages[0]
    # 奇数取后4条，偶数取后3条，确保截取的是完整的对话对
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    new_messages = [first_msg] + recent_messages
    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),  # 先清空所有消息
            *new_messages                            # 再放回裁剪后的消息
        ]
    }

agent = create_agent(
    model=model,
    middleware=[trim_messages],
    checkpointer=InMemorySaver(),
)

关键点：

• RemoveMessage(id=REMOVE_ALL_MESSAGES) 是一个特殊指令，表示清空所有消息
• 然后放回裁剪后的消息列表（第一条 + 最近几条）
• 裁剪发生在模型调用前，只影响模型看到的上下文
• 奇偶判断确保截取的是完整的 Human-AI 对话对

裁剪效果分析（4 次 invoke）：

步骤	触发前消息数	是否触发裁剪	传给 LLM 的消息
1	1	不触发	[H1]
2	3	不触发	[H1, A1, H2]
3	5	触发！取后4条	[H1(第一条), A1, H2, A2, H3(后4条)]
4	7	触发！取后4条	[H1(第一条), A2, H3, A3, H4(后4条)]

12.2 消息删除（after_model）

目标：模型调用完成后将某些消息从列表中永久移除。
策略：保持固定数量的消息，超出阈值的最早消息被删除。
时机：在 after_model 钩子中执行。

from langchain.messages import RemoveMessage
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig

@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    messages = state["messages"]
    # 保持最近的 5 条消息
    if len(messages) > 5:
        to_delete = len(messages) - 5
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:to_delete]]}
    return None

agent = create_agent(
    model=model,
    middleware=[delete_old_messages],
    checkpointer=InMemorySaver()
)

关键点：

• 消息裁剪强调"模型调用前裁剪"，消息删除强调"模型调用后永久更改状态"
• RemoveMessage(id=m.id) 按 ID 精准删除特定消息
• 删除的是最老的消息，保持最新的固定条数

消息列表数量变化分析：

第1轮：2条（不触发）→ 记住"你是老王"
第2轮：4条（不触发）→ 记住"你是老王、我是小王"
第3轮：6条（触发！删1条）→ 删除最早的"你好，我是老王"
第4轮：8条（触发！删3条）→ 保持最后5条

12.3 RemoveMessage 的底层机制

当你在中间件里返回 RemoveMessage(id="1") 时，框架不会直接从数组里删掉消息。它的底层处理是：

1. 追加"墓碑"标记：框架把 RemoveMessage 作为一条新记录追加到状态历史中
2. 下次读取时过滤：当 LangGraph 下次从 checkpointer 读取状态时，会扫描所有墓碑标记，被标记的消息不会被加载到消息列表中

这就像 Git 的版本控制——不是物理删除，而是标记删除。历史记录仍然存在于 checkpointer 中，但逻辑上被"隐藏"了。

13. 长期记忆概述

长期记忆（Long-term memory）在会话间存储用户特定或应用级数据，基于 store 对象持久化数据。

核心概念：

• store：长期记忆对象，跨会话持久化的数据
• namespace（命名空间）：用元组（tuple）表示的层级路径，组织数据

Namespace 层级结构示例：

store
 ├─ namespace = ("users", "Alice", "memories")
 │  ├─ key = "profile"
 │  │  value = {"name": "李四", "city": "上海", "preferences": ["简洁风格"]}
 │  ├─ key = "travel_preference"
 │  │  value = {"favorite_cities": ["北京", "杭州"], "budget_level": "medium"}
 │  └─ key = "writing_style"
 │     value = {"tone": "professional", "language": "zh-CN"}
 │
 ├─ namespace = ("users", "user_2", "memories")
 │  ├─ key = "profile"
 │  │  value = {"name": "李四", "city": "深圳"}
 │  └─ key = "product_interest"
 │     value = {"topics": ["AI Agent", "RAG", "Workflow"]}
 │
 └─ namespace = ("users", "user_1", "thread_3", "artifacts")
    ├─ key = "draft_v1"
    │  value = {"type": "html", "content": "..."}
    └─ key = "final_scheme"
       value = {"palette": ["#F5F1E8", "#1F3A5F"]}

与短期记忆的区别：

• 短期记忆：同一 thread_id 内共享，不同 thread_id 隔离
• 长期记忆：全局共享，不受 thread_id 限制，通过 namespace 组织

14. Store 基础 API

LangChain 1.2.x 的长期记忆基于 store 持久化数据，相关 API：

API	作用
put()	写入数据
get()	读取数据
search()	检索数据

14.1 put() — 写入 API

def put(
    self,
    namespace: tuple[str, ...],  # 文档所在的层级路径
    key: str,                     # 该路径下的唯一键
    value: dict[str, Any],        # 要保存的 JSON-like 字典
    index: Literal[False] | list[str] | None = None,  # 控制语义检索索引
    *,
    ttl: float | None = NOT_PROVIDED,  # 可选，过期时间
) -> None

参数说明：

• namespace：文档所在的层级路径（元组）
• key：该路径下的唯一键
• value：要保存的 JSON-like 字典
• index：控制语义检索索引
  • None（默认）：使用 store 初始化时配置的索引配置
  • False：不为该 item 建立语义索引
  • list[str]：只对指定字段路径建索引
• ttl：可选，过期时间（是否支持取决于具体 store 实现）

示例：

store.put(
    ("users", "alice", "memories"),  # namespace
    "pref_food",                     # key
    {"category": "food", "text": "Alice likes sushi"}  # value
)

14.2 get() — 读取 API

def get(
    self,
    namespace: tuple[str, ...],
    key: str,
    *,
    refresh_ttl: bool | None = None,  # 是否刷新 TTL
) -> Item | None

返回值：不是单纯的 value，而是完整的 Item 对象。

item = store.get(("users", "alice", "memories"), "pref_food")
if item is not None:
    print(item.value)
    # {'category': 'food', 'text': 'Alice likes sushi'}

Item 对象包含字段：

• namespace：命名空间
• key：键
• value：值
• created_at：创建时间
• updated_at：更新时间

14.3 InMemoryStore vs PostgresStore 的差异

InMemoryStore：

• 每次 put 都会创建一个新的 Item 对象，无论 namespace 和 key 是否相同
• created_at 和 updated_at 始终一致（因为是新建而非更新）

# InMemoryStore
store.put(("users",), "user-1", {"name": "小蓝"})
# Item(namespace=['users'], key='user-1', value={'name': '小蓝'},
#      created_at='2026-06-11T15:20:52...', updated_at='2026-06-11T15:20:52...')

store.put(("users",), "user-1", {"name": "小红"})
# Item(namespace=['users'], key='user-1', value={'name': '小红'},
#      created_at='2026-06-11T15:23:11...', updated_at='2026-06-11T15:23:11...')
# created_at 变了！因为是新建而非更新

PostgresStore：

• 更改数据的逻辑是 update 而非覆盖
• created_at 固定为 Item 创建时间，updated_at 为更新时间，二者可能不同

# PostgresStore
with PostgresStore.from_conn_string(DB_URL) as store:
    store.setup()
    store.put(("users",), "user-11", {"name": "小蓝"})
    # created_at='2026-06-11T23:24:31...', updated_at='2026-06-11T23:24:31...'

    store.put(("users",), "user-11", {"name": "小红"})
    # created_at='2026-06-11T23:24:31...'（不变）
    # updated_at='2026-06-11T23:26:11...'（变了）

14.4 search() — 检索 API

def search(
    self,
    namespace_prefix: tuple[str, ...],  # 命名空间前缀
    /,
    *,
    query: str | None = None,           # 语义检索查询
    filter: dict[str, Any] | None = None,  # 结构化过滤条件
    limit: int = 10,                    # 最大返回条数
    offset: int = 0,                    # 跳过的条数
    refresh_ttl: bool | None = None,
) -> list[SearchItem]

三种检索方式：

方式1：按 namespace 前缀搜索

# 搜索 ("users",) 下所有数据
for item in store.search(("users",)):
    print(item)

# 搜索 ("users", "Alice") 下的数据（更精确的前缀）
for item in store.search(("users", "Alice")):
    print(item)

方式2：按 filter 结构化过滤

filter 是 value 中的键值对组合，用于精确匹配。

# 过滤 food 为 "紫光园奶皮子酸奶" 的记录
for item in store.search(("users",), filter={"food": "紫光园奶皮子酸奶"}):
    print(item)

# 过滤 sports 为 "跑步" 的记录
for item in store.search(("users",), filter={"sports": "跑步"}):
    print(item)

# 过滤 course 为 "数字电路与模拟电路" 的记录
for item in store.search(("users",), filter={"course": "数字电路与模拟电路"}):
    print(item)

方式3：按 query 语义检索

需要 store 初始化时配置 IndexConfig（包含嵌入函数）。

# 语义搜索：查找与 "数电模电" 最相似的数据
for item in store.search(("users",), query="数电模电"):
    print(item)
# 返回结果按 score（相似度分数）降序排列

15. IndexConfig — 语义检索配置

class IndexConfig(TypedDict, total=False):
    dims: int                                    # 输出向量维度
    embed: Embeddings | EmbeddingsFunc | str     # 嵌入函数或模型
    fields: list[str] | None                     # 用于计算向量的属性列表

fields 可取值：

• ["$"]：将 value 作为整体嵌入
• ["field1", "field2"]：单独指定某个一级字段
• ["parent.child"]：从嵌套 JSON 对象中获取子字段
• ["array[*].field"]：从 JSON 数组的每个对象中获取子字段
• 上述形式可同时出现，每个元素都会生成一个嵌入向量

举例1：自定义嵌入函数

from langgraph.store.memory import InMemoryStore

# 自定义嵌入函数（返回固定向量，仅用于演示）
def embed(text: list[str]) -> list[list[float]]:
    return [[1.0] * 6 for _ in range(len(text))]

index_config = {
    "embed": embed,
    "dims": 6,
    "fields": ["$", "course"]  # 对整体和 course 字段分别建索引
}

store = InMemoryStore(index=index_config)

举例2：使用嵌入模型

from langgraph.store.memory import InMemoryStore
from langchain.embeddings import init_embeddings
import os

embedding_model = init_embeddings(
    model="openai:text-embedding-3-large",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)

index_config = {
    "embed": embedding_model,
    "dims": 3072,  # text-embedding-3-large 的嵌入维度
    "fields": ["$"]
}

store = InMemoryStore(index=index_config)

查看嵌入向量：

from pprint import pprint
pprint(store._vectors)
# 结构：{namespace: {key: {field: [向量]}}}

16. 在 Agent 中访问长期记忆

可以在工具或中间件中访问长期记忆。

16.1 在工具中访问

通过 ToolRuntime 的 store 属性访问长期记忆。

from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent, AgentState
from langgraph.store.memory import InMemoryStore
from typing import NotRequired

store = InMemoryStore()

# 扩展 AgentState，增加 user_id 字段
class CustomState(AgentState):
    user_id: NotRequired[str]

@tool(parse_docstring=True)
def save_user_info(name: str, runtime: ToolRuntime) -> str:
    """将用户信息保存在长期记忆中"""
    runtime.store.put(("users",), runtime.state["user_id"], {"name": name})
    return "saved"

@tool(parse_docstring=True)
def get_user_info(runtime: ToolRuntime) -> str:
    """从长期记忆中读取用户信息"""
    item = runtime.store.get(("users",), runtime.state["user_id"])
    return str(item.value) if item else "unknown"

agent = create_agent(
    model=model,
    tools=[save_user_info, get_user_info],
    store=store,
    system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
    state_schema=CustomState,
)

# 第一个会话（线程）：写入长期记忆
response1 = agent.invoke({
    "messages": [HumanMessage("你好，很高兴认识你，我是小花")],
    "user_id": "user-1"
})

# 第二个会话（线程）：读取长期记忆（不同 thread_id，但共享 store）
response2 = agent.invoke({
    "messages": [HumanMessage("我是谁")],
    "user_id": "user-1"
})
# Agent 能回答："你是小花"

关键点：

• 两次 invoke 没有通过 config 串联，是两个独立的会话（不同 thread_id）
• 但第二个会话可以访问第一个会话写入长期记忆的内容
• CustomState 扩展了 AgentState，增加 user_id 字段，让 Agent 运行时知道当前用户
• runtime.store 是运行时自动注入的，无需手动传递

16.2 在中间件中访问

Node-style hooks（如 before_model）：

# before_model 的 runtime 参数包含 store
def before_model(self, state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None:
    # 通过 runtime.store 访问长期记忆
    item = runtime.store.get(("users",), user_id)
    ...

Runtime 定义（包含 store 字段）：

@dataclass
class Runtime(Generic[ContextT]):
    context: ContextT = field(default=None)       # 静态上下文
    store: BaseStore | None = field(default=None)  # 长期记忆存储
    stream_writer: StreamWriter = ...               # 自定义流写入器
    previous: Any = field(default=None)             # 前次返回值

Wrap-style hooks（如 wrap_model_call、wrap_tool_call）：

# wrap_model_call 通过 request.runtime.store 访问
def wrap_model_call(self, request: ModelRequest, handler):
    store = request.runtime.store  # 通过 request 访问 store
    ...

# wrap_tool_call 通过 request.runtime.store 访问
def wrap_tool_call(self, request: ToolCallRequest, handler):
    store = request.runtime.store  # 通过 request 访问 store
    ...

17. 何时写入记忆

官方介绍了两种写入记忆的方式：

方式1：在主流程里写（hot path）

用户发消息，AI 一边回答，一边决定要不要记下来。

优点	缺点
立即生效	增加延迟
下一轮马上能用	逻辑变复杂
用户可感知，透明

方式2：在后台写（background）

先回答用户，记忆整理放到后台异步做。

优点	缺点
主流程更快	不能立刻生效
记忆逻辑更独立	要决定多久整理一次
更适合批量整理	触发时机不好选

工程选择建议：

• 用户偏好、账号资料：可热路径写
• 对话摘要、经验沉淀、行为分析：更适合后台写

18. 静态运行时上下文（Static Runtime Context）

静态运行时上下文表示不可变的数据，如用户元数据、工具和传递给应用程序的数据库连接对象。

通常在运行开始时通过 invoke / stream 的 context 参数传入。

from dataclasses import dataclass

@dataclass
class UserContext:
    username: str

agent = create_agent(
    model="deepseek-chat",
    tools=[get_weather, get_news],
    store=store,
    context_schema=UserContext,  # 指定上下文类型
)

# 通过 context 参数传入静态上下文
response = agent.invoke(
    {"messages": ["你好啊，今天北京天气如何？"]},
    context=UserContext(username="Ada Lovelace")
)

关键点：

• context_schema 指定上下文对象的类型
• context 参数是运行时静态配置，每次 invoke 互相独立
• 在中间件和工具中通过 runtime.context / request.runtime.context 访问

19. 综合案例：中间件 + Store + Context

19.1 工具权限过滤 + 额度校验

@dataclass
class UserContext:
    username: str

class CheckCredit(AgentMiddleware):
    """在模型调用前，根据长期记忆中的用户配置过滤可用工具"""
    def wrap_model_call(self, request: ModelRequest, handler):
        context = request.runtime.context
        store = request.runtime.store
        value = store.get(namespace, context.username).value
        
        tools = []
        for tool in request.tools:
            if value[tool.name] == "yes":
                tools.append(tool)
            else:
                logger.warning(f"{context.username} 无权调用 {tool.name}")
        
        request = request.override(tools=tools)  # 仅本次生效
        return handler(request)

class ToolGuard(AgentMiddleware):
    """在工具调用前，校验额度并更新长期记忆"""
    def wrap_tool_call(self, request: ToolCallRequest, handler):
        store = request.runtime.store
        context = request.runtime.context
        username = context.username
        value = store.get(namespace, username).value
        
        tool_name = request.tool.name
        credits_cost = CREDIT_MAP.get(tool_name)
        credits_left = value["tokens_left"]
        
        if credits_left < credits_cost:
            logger.warning(f"{username} 额度不足")
            return Command(update={
                "messages": [ToolMessage(
                    content="额度不足，无法调用工具",
                    tool_call_id=request.runtime.tool_call_id
                )]
            })
        
        # 更新剩余额度（写入长期记忆）
        credits_left -= credits_cost
        value["tokens_left"] = credits_left
        store.put(namespace, username, value)
        
        return handler(request)

agent = create_agent(
    model="deepseek-chat",
    middleware=[CheckCredit(), ToolGuard()],
    tools=[get_weather, get_news],
    store=store,
    context_schema=UserContext
)

19.2 dynamic_prompt — 动态系统提示词

@dynamic_prompt 是一个便捷中间件，用于动态更改系统提示词。

from langchain.agents.middleware import dynamic_prompt, ModelRequest

@dynamic_prompt
def personalized_prompt(request: ModelRequest) -> str:
    username = request.runtime.context.username
    store = request.runtime.store
    preferences = store.get(namespace, username).value["chat_preferences"]
    custom_prompt = f"# 用户偏好\n{'\n'.join(preferences)}"
    return custom_prompt

agent = create_agent(
    model="deepseek-chat",
    middleware=[personalized_prompt],
    store=store,
    context_schema=UserContext
)

# 不同用户得到不同的系统提示词
response1 = agent.invoke(
    {"messages": ["为什么花儿这样红？"]},
    context=UserContext(username="Ada Lovelace")  # 偏好：简洁
)
# → 简短回答

response2 = agent.invoke(
    {"messages": ["为什么花儿这样红？"]},
    context=UserContext(username="Blackwell")  # 偏好：长篇大论
)
# → 详细回答

关键点：dynamic_prompt 动态更新本次调用的系统提示词，不会更改消息列表。

19.3 工具中访问长期记忆（完整案例）

结合 Pydantic 模型 + ToolRuntime + Context + Store：

from pydantic import BaseModel, Field
from typing import List, Any
from langgraph.prebuilt.tool_node import ToolRuntime

class UserInfo(BaseModel):
    username: str = Field(description="用户名", default="unknown")
    age: int = Field(description="年龄", default=0)
    hobbies: List[str] = Field(description="兴趣爱好", default_factory=list)

@tool(parse_docstring=True)
def read_user_info(runtime: ToolRuntime[UserContext, Any]) -> UserInfo | str:
    """读取用户信息"""
    user_id = runtime.context.user_id
    item = runtime.store.get(namespace, user_id)
    if item:
        return UserInfo(**item.value)
    return ''

@tool(parse_docstring=True)
def write_user_info(user_info: UserInfo, runtime: ToolRuntime[UserContext, Any]) -> bool:
    """将用户信息写入长期记忆"""
    user_id = runtime.context.user_id
    runtime.store.put(namespace, user_id, user_info.model_dump())
    return True

agent = create_agent(
    model="deepseek-chat",
    tools=[read_user_info, write_user_info],
    store=store,
    checkpointer=InMemorySaver(),
    context_schema=UserContext
)

# thread_1：第一轮对话（写入记忆）
agent.invoke(
    {"messages": [SystemMessage("..."), HumanMessage("你好，我是韩立，我喜欢修仙")]},
    context=UserContext(user_id='user_1'),
    config=config1
)

# thread_1：第二轮对话（更新记忆，先读后写）
agent.invoke(
    {"messages": ["我今年两百岁了，我喜欢跑步"]},
    context=UserContext(user_id='user_1'),
    config=config1
)

# thread_2：不同线程，但通过长期记忆共享信息
agent.invoke(
    {"messages": ["你还记得我吗？"]},
    context=UserContext(user_id='user_1'),
    config=config2
)
# Agent 回答："你是韩立，200岁，喜欢修仙和跑步"

ToolRuntime 泛型说明：

• ToolRuntime[UserContext, Any]：第一个泛型是 Context 类型，第二个是 State 类型
• 如果不显式指定 Context 泛型，底层认为上下文对象为 None，invoke 时传 context 会抛出警告

---

模块 2：带教式理解

---

1. 为什么大模型是"无状态"的

大模型的 API 设计是"请求-响应"模式：你发一个请求，它返回一个响应，然后这次交互就结束了。模型服务端不会为你的下一次请求保留任何上下文。

这就像打电话给一个失忆的人——每次打电话他都不记得上次聊了什么。你必须每次把前情提要重新讲一遍。

Agent 也不例外：agent.invoke() 每次都是独立调用，如果你不把历史消息放进 messages 列表，Agent 就"不记得"之前说过什么。

2. 短期记忆为什么是"三要素"的组合

单独看每一个：

• 没有 State：消息列表没地方存——历史消息无处安放
• 没有 Checkpointer：State 只在内存中存活一次调用——跨调用无法保留
• 没有 Thread ID：多个用户的对话混在一起——无法区分谁的对话

三者缺一不可：State 负责存什么，Checkpointer 负责怎么存，Thread ID 负责区分谁是谁。

3. 为什么用 InMemorySaver 而不是直接传 messages

你确实可以每次手动把历史消息拼进 messages 列表传入（情况3的做法），但这样有几个问题：

1. 手动维护：你需要自己记录哪些消息是历史的，哪些是新输入的
2. 无法中断恢复：如果 Agent 执行到一半被中断，你无法恢复之前的状态
3. 多用户管理：你需要自己维护每个用户的对话历史

Checkpointer 自动帮你做这些事——你只需要传新消息，它自动读取历史、追加新消息、调用模型、保存结果。就像 Git 的自动 commit——你只管写代码，它帮你管理版本。

4. InMemorySaver 重建后为什么状态丢失

InMemorySaver() 把数据存在 Python 进程的内存里。当你创建一个新的 InMemorySaver() 实例时，它是一个全新的空对象——和之前那个实例没有任何关系。

这就像你换了一个新笔记本——旧笔记本上的笔记不会自动转移过来。

PostgresSaver 不同：数据存在外部数据库里，无论你创建多少个 PostgresSaver 实例，只要连接同一个数据库，用同一个 thread_id，就能读到之前的数据。

5. 消息裁剪和消息删除有什么区别

对比项	消息裁剪（before_model）	消息删除（after_model）
执行时机	模型调用前	模型调用后
影响范围	只影响模型看到的上下文	永久更改状态中的消息列表
使用工具	RemoveMessage(id=REMOVE_ALL_MESSAGES) 清空后重建	RemoveMessage(id=m.id) 按 ID 精准删除
适用场景	控制成本，保留系统消息和最近消息	明确要遗忘、清理、重置某些历史

裁剪是"临时遮罩"——模型看不到，但消息还在。删除是"物理清除"——消息从状态中移除（通过墓碑标记）。

6. RemoveMessage 为什么用"墓碑"而不是直接删除

直接从数组里删除消息会带来一个问题：checkpointer 的快照完整性被破坏。

LangGraph 的 checkpointer 是基于快照的——每次状态变化都会记录一个 checkpoint。如果直接物理删除消息，那之前的 checkpoint 引用的消息就不存在了，回溯历史时会出现不一致。

墓碑标记的做法是：不删除原始消息，而是追加一条"删除指令"。下次读取时，框架扫描所有墓碑标记，被标记的消息不会被加载。这样既实现了逻辑删除，又保持了 checkpoint 历史的完整性。

7. Namespace 为什么用元组而不是字符串

字符串路径（如 "users/alice/memories"）有歧义：如果某个 key 里包含 /，解析就会出错。

元组 ("users", "alice", "memories") 是确定性的：每个元素就是一个层级，不会出现歧义。而且元组天然支持前缀匹配——("users", "alice") 是 ("users", "alice", "memories") 的前缀，所以 search(("users", "Alice")) 能找到该前缀下所有数据。

8. InMemoryStore 为什么 created_at 和 updated_at 一样

InMemoryStore 的更新逻辑是"覆盖"而非"更新"——每次 put 都创建一个全新的 Item 对象，旧对象被丢弃。所以 created_at 和 updated_at 始终是同一个时间。

PostgresStore 的更新逻辑是 SQL 的 UPDATE——如果 namespace+key 已存在，只更新 value 和 updated_at 字段，created_at 保持不变。这更符合直觉。

9. 为什么在工具中访问 Store 需要扩展 AgentState

runtime.store 是全局共享的，但 runtime.store.get(namespace, key) 需要知道当前用户的 key（如 user_id）。

如果不扩展 AgentState，Agent 就不知道"当前是谁在说话"。通过 CustomState(AgentState) 增加 user_id 字段，调用时传入 "user_id": "user-1"，工具就能通过 runtime.state["user_id"] 获取当前用户 ID，从而在 Store 中定位正确的数据。

10. ToolRuntime 的泛型为什么重要

ToolRuntime 的签名是 ToolRuntime[ContextT, StateT]。

• 第一个泛型 ContextT：静态上下文对象的类型
• 第二个泛型 StateT：State 对象的类型

如果你不显式指定 ToolRuntime[UserContext, Any]，LangChain 底层默认 ContextT=None。当你在 invoke 时传了 context=UserContext(...)，框架会发现类型不匹配并抛出警告。

显式指定泛型不仅是为了类型提示，更是告诉框架"我会在运行时传入一个 UserContext 类型的上下文对象"。

---

这一章结束后，你应该能自己回答

---

1、为什么 Agent 需要记忆？大模型的什么特性导致了这个问题？

大模型本身是"无状态"的——每次调用 agent.invoke() 都是全新的开始，不记得之前的对话。Agent 也是基于大模型构建的，如果不把历史消息传入 messages 列表，Agent 就不知道之前说过什么。记忆模块的作用就是保存历史交互信息，让 LLM 在每次响应时都能"看到"之前的对话内容。

2、LangGraph 提供了哪三种管理上下文的方法？分别对应什么对象？

1. 动态运行时上下文（state 对象）：单次运行中演变的可变数据，生命周期为单次运行
2. 动态跨会话上下文（store 对象）：对话间共享的持久数据，生命周期为跨对话
3. 静态运行时上下文（context 对象）：启动时传入的不可变元数据，生命周期为单次运行

3、短期记忆的三要素是什么？各自的作用是什么？

State（会话内部状态）：默认存储历史消息列表 messages。Checkpointer（持久化机制）：将 State 作为检查点持久化保存。Thread ID（会话作用域）：唯一标识 State，运行时按 thread_id 读写 State 快照。三者缺一不可——State 负责存什么，Checkpointer 负责怎么存，Thread ID 负责区分谁是谁。

4、InMemorySaver 和 PostgresSaver 的核心区别是什么？

InMemorySaver 将状态保存在内存中，进程结束或重建 Saver 时历史状态丢失。PostgresSaver 将状态保存在 PostgreSQL 数据库中，即便重新创建 Saver，只要 thread_id 一致，历史状态就可以通过数据库加载恢复。InMemorySaver 适合测试调试，PostgresSaver 适合生产环境。

5、消息裁剪和消息删除的区别是什么？分别用什么 Hook 实现？

消息裁剪在模型调用前（before_model）执行，只影响模型看到的上下文，使用 RemoveMessage(id=REMOVE_ALL_MESSAGES) 清空后重建消息列表。消息删除在模型调用后（after_model）执行，永久更改状态中的消息列表，使用 RemoveMessage(id=m.id) 按 ID 精准删除特定消息。

6、Store 的 search() 支持哪三种检索方式？

1. 按 namespace 前缀搜索：传入 namespace_prefix 元组，返回该前缀下所有数据
2. 按 filter 结构化过滤：传入 value 中的键值对，精确匹配符合条件的记录
3. 按 query 语义检索：传入自然语言查询，通过向量相似度计算返回最相似的结果（需要配置 IndexConfig）

7、在 Agent 中有哪两种方式访问长期记忆？分别通过什么属性？

1. 在工具中访问：通过 runtime.store 访问（ToolRuntime 对象的 store 属性）
2. 在中间件中访问：Node-style hooks 通过 runtime.store 访问，Wrap-style hooks 通过 request.runtime.store 访问

---

小测试

---

1. 短期记忆的 checkpointer 工作流程是什么？

调用 agent.invoke() 时，checkpointer 自动执行四步：① 读取之前的历史（通过 thread_id 从 checkpointer 加载）② 追加新消息 ③ 调用模型（传入完整历史）④ 保存新的历史（写入 checkpointer）。你只需要传新消息，checkpointer 自动管理历史。

2. RemoveMessage 的"墓碑"机制是什么？

当返回 RemoveMessage(id=“1”) 时，框架不会从数组里物理删除消息，而是把 RemoveMessage 作为一条新记录追加到状态历史中（墓碑标记）。下次读取状态时，框架扫描所有墓碑标记，被标记的消息不会被加载到消息列表中。这保持了 checkpoint 历史的完整性。

3. IndexConfig 的 fields 参数有哪些取值方式？

["$"]：将 value 作为整体嵌入。["field1", "field2"]：单独指定一级字段。["parent.child"]：从嵌套 JSON 中获取子字段。["array[*].field"]：从 JSON 数组每个对象中获取子字段。上述形式可同时出现，每个元素生成一个独立的嵌入向量。

4. InMemoryStore 和 PostgresStore 在更新数据时有什么差异？

InMemoryStore 每次 put 都创建新的 Item 对象（覆盖逻辑），created_at 和 updated_at 始终一致。PostgresStore 的更新逻辑是 UPDATE（更新逻辑），如果 namespace+key 已存在，只更新 value 和 updated_at，created_at 保持不变。

5. 何时写入记忆的"hot path"和"background"分别是什么？如何选择？

hot path（主流程写）：用户发消息时 AI 一边回答一边记录，立即生效但增加延迟。background（后台写）：先回答用户，记忆整理放到后台异步做，主流程更快但不能立刻生效。用户偏好、账号资料适合热路径写，对话摘要、经验沉淀、行为分析更适合后台写。