一、为什么大模型需要「记忆」

大多数人以为，跟 ChatGPT 这类应用多轮对话时，模型是「记得」前面说过什么的。其实大模型本身是无状态的——它不记住任何上下文。每次调用 agent.invoke() 都是一次全新的开始。

举个例子：先告诉它「我叫小明」，再问「我叫什么名字」，如果两次调用之间不传递任何历史，模型会老实回答「我不知道你的名字」。

我们真正想要的，是下面这种感觉：

用户：我叫小明
Agent：很高兴认识你，小明
...
用户：我叫什么名字？
Agent：你叫小明。  ✅

要实现这一点，就需要一个额外的模块去保存历史交互信息，在下一次请求时把历史一并塞给模型。在 LangChain 里，这个模块就叫 Memory（记忆），核心职责是「保存上下文」和「提供上下文」。

而 上下文工程（Context Engineering） 负责「合理组织」这些记忆和任务信息，让 LLM 每次响应时都能「看到」之前的对话，从而产生连贯、贴合需求的答案。这是 Agent 实现复杂多轮交互的核心基础。

二、上下文的两大维度：可变性 × 生命周期

LangChain 的上下文工程构建在 LangGraph 之上。LangGraph 提供了三种管理上下文的方法，从「可变性」和「生命周期」两个维度划分：

上下文类型

描述

可变性

生命周期

访问方法

动态运行时上下文

单次运行中会演变的可变数据

动态

单次运行

state 对象

动态跨会话上下文

对话间共享的持久数据（偏好、洞察、知识条目）

动态

跨对话

store 对象

静态运行时上下文

启动时传入的用户元数据、工具、数据库连接

静态

单次运行

context 对象

记住这三兄弟：state（短期）、store（长期）、context（静态），全文都围绕它们展开。

从记忆长短的角度，又分为两类：

  • 短期记忆（Short-term / 会话级 / thread-scoped）：作用范围是单个线程（Thread），换一个 thread_id 记忆即消失。

  • 长期记忆（Long-term / 跨会话级）：在会话间存储用户级或应用级数据，任何线程都能随时调用。

三、短期记忆：State + Checkpointer + Thread ID

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

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

  • State：默认存储历史消息列表 messages，通过 State 管理历史消息。

  • Checkpointer：把 State 作为「检查点」持久化保存（某个时刻的 State 快照）。

  • Thread ID：唯一标识 State，运行时按 thread_id 读写快照。

这就像玩 RPG 游戏的「自动存档」：不用手动保存，系统在关键节点自动记录，下次随时从存档点继续。

3.1 三步让 Agent 拥有记忆

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.messages import HumanMessage
​
# ① 初始化记忆引擎
checkpointer = InMemorySaver()
​
# ② 创建 Agent 时绑定 checkpointer
agent = create_agent(model=model, checkpointer=checkpointer)
​
# ③ 调用时指定 thread_id
config = {"configurable": {"thread_id": "1"}}
​
agent.invoke({"messages": [HumanMessage("我叫张三")]}, config=config)
agent.invoke({"messages": [HumanMessage("我叫什么？")]}, config=config)
# => 你叫张三。 ✅

关键就三步：建 checkpointer → 绑 Agent → 传 thread_id。同一个 thread_id 共享记忆，不同 thread_id 完全隔离。

3.2 thread_id 是隔离会话的核心开关

  • 多用户聊天：user_alice / user_bob 各自独立，Agent 不会串味。

  • 同一用户不同任务：task_coding / task_docs 分开，互不干扰。

3.3 checkpointer 在背后做了什么

每次 invoke 后，checkpointer 自动执行四件事：读取历史 → 追加新消息 → 调模型（传完整历史）→ 保存新历史。你只需要传新消息，历史维护完全自动化。

3.4 常见坑（Agent 不记得了怎么办）

  • ✅ 是否加了 checkpointer？

  • ✅ 是否传入了 config？

  • ✅ 两次调用的 thread_id 是否相同？

关于 InMemorySaver：数据只存在内存里，进程结束就丢，不同进程无法共享。所以它只适合测试和调试，生产环境要换持久化后端。

3.5 生产环境：PostgreSQL 持久化

from langgraph.checkpoint.postgres import PostgresSaver
​
DB_URL = "postgresql://user:pass@host:5432/db?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    checkpointer.setup()   # 首次运行建表（CREATE IF NOT EXISTS）
    agent = create_agent(model=model, checkpointer=checkpointer)

setup() 会创建四张表：checkpoints（主表，存每个 thread 的快照）、checkpoint_blobs（存复杂 channel 值）、checkpoint_writes（中间写入）、checkpoint_migrations（迁移版本）。

关键结论：InMemorySaver 重建即丢历史；而 PostgresSaver 只要 thread_id 一致，即使重建 Saver 也能串联起历史状态。

四、记忆治理：上下文不能无限膨胀

对话一长，state 就会带来三个问题：上下文窗口装不下、模型被陈旧内容「分心」、token 花费飙升。于是需要对历史做压缩、清理、重组。主要有四种策略：

4.1 消息裁剪（before_model）

在调用模型前裁剪消息列表，控制模型的可见范围。通常保留系统初始消息 + 最近若干条，适合成本敏感、对旧上下文依赖不强的场景。

from langchain.agents.middleware import before_model
from langgraph.graph.message import REMOVE_ALL_MESSAGES
​
@before_model
def trim_messages(state, runtime):
    messages = state["messages"]
    if len(messages) <= 3:
        return None
    first_msg = messages[0]
    recent = messages[-4:] if len(messages) % 2 else messages[-3:]
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), first_msg, *recent]}

4.2 消息删除（after_model）

与裁剪相反，它在模型调用完成后把某些消息从列表里移除，永久改变状态，适合明确「遗忘/清理/重置」的场景。

from langchain.agents.middleware import after_model
​
@after_model
def delete_old_messages(state, runtime):
    messages = state["messages"]
    if len(messages) > 5:
        to_delete = len(messages) - 5
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:to_delete]]}
    return None

一个有趣的细节：RemoveMessage 并不是真的删数组元素，而是追加一条「墓碑」标记，运行时由内置的 Reducer 计算「原始消息 + 墓碑」后，在丢给模型前过滤掉被标记的消息。

4.3 摘要（SummarizationMiddleware）

裁剪和删除都会丢信息。摘要是更折中的方案——保语义，不保原文。官方内置 SummarizationMiddleware：

from langchain.agents.middleware import SummarizationMiddleware
​
agent = create_agent(
    model=model_out,
    checkpointer=InMemorySaver(),
    middleware=[
        SummarizationMiddleware(
            model=model_in,               # 可以用便宜的模型做摘要
            trigger=[("tokens", 100)],    # 超过 100 token 就摘要
            keep=("messages", 2),         # 保留最近 2 条原文
        )
    ]
)

超过阈值时才触发，用便宜模型（如 gpt-4o-mini）做摘要，通常比传输全部历史更省。

4.4 自定义过滤策略

通过中间件可以任意改动消息列表，实现任何过滤策略，灵活性拉满。

五、state 是什么

state 是 Agent 底层有状态运行图的状态信息，是 AgentState（TypedDict 子类）的实例，可像字典一样读写。它有三大字段：

  • messages：截至当前节点的历史消息（Required）。

  • jump_to：跳转到运行图的指定节点（NotRequired，中间件章节讲过）。

  • structured_response：启用结构化输出时，结构化后的内容记录在这里。

六、长期记忆：store → namespace → key → value

短期记忆是会话级、会话间不共享；长期记忆是用户级/应用级，任何会话都能随时访问。比如「你喜欢简短回答」「某个用户是 VIP」「某个流程过去怎么做效果好」。

6.1 三类长期记忆（参考 CoALA 论文）

类型

存什么

例子

Semantic（语义记忆）

事实

用户喜欢简洁回答、常用中文

Episodic（情景记忆）

经验

过去某任务怎么成功、few-shot 示例

Procedural（程序性记忆）

规则/做事方法

系统提示词、工作流程、工具调用规则

6.2 四层存储架构

store（记忆仓库） → namespace（命名空间，tuple） → key（唯一键，str） → value（值，dict）

  • Store：InMemoryStore（测试）/ PostgresStore（生产）。

  • Namespace：元组层级路径，像文件目录，用于分组隔离。

  • Key：命名空间下的唯一标识。

  • Value：JSON-like 字典。

store.put(("users", "alice", "memories"), "pref_food", {"category": "food", "text": "Alice likes sushi"})
item = store.get(("users", "alice", "memories"), "pref_food")

  • put()：写入。参数有 index（语义索引配置）、ttl（过期时间，可选）。

  • get()：按 namespace + key 精确查询，返回 Item 对象（含 created_at / updated_at）。

  • search()：两者检索方式——按 filter 做结构化过滤，或按 query 做语义相似度检索（返回带 score 的 SearchItem 列表，按 score 降序）。

语义检索需要配置 index：

index_config = {
    "embed": embedding_model,   # 自定义函数或嵌入模型对象
    "dims": 3072,               # 向量维度
    "fields": ["$", "course"],  # "$"=整体嵌入；也可指定字段
}
store = InMemoryStore(index=index_config)
for item in store.search(("users",), query="数电模电"):
    print(item)

6.4 在 Agent 运行图中访问长期记忆

  • 在工具中访问：runtime.store.put(...) / runtime.store.get(...)。

  • 在中间件中访问：Node-style 钩子里用 runtime.store，Wrap-style 钩子里用 request.runtime.store。

一个典型场景：第一个会话写入「我是小花」，第二个会话（独立线程）问「我是谁」，Agent 仍能答出「你是小花」——因为长期记忆跨会话共享。

6.5 何时写入记忆

两种方式：

  • 热路径写（hot path）：回答的同时决定要不要记。优点：立即生效、可感知；缺点：增加延迟、逻辑复杂。适合用户偏好、账号资料。

  • 后台写（background）：先回答，记忆整理异步做。优点：主流程快、逻辑独立、适合批量；缺点：不能立刻生效。适合对话摘要、经验沉淀、行为分析。

七、静态运行时上下文：context

context 表示不可变的数据（用户元数据、工具、数据库连接），在运行开始时通过 invoke / stream 的 context 参数传入，运行期间不变。

from dataclasses import dataclass
​
@dataclass
class UserContext:
    username: str
​
agent = create_agent(..., context_schema=UserContext)
​
agent.invoke({"messages": [...]}, context=UserContext(username="Ada Lovelace"))

在中间件/工具里通过 runtime.context（或 request.runtime.context）访问。

三个实战用法：

  1. 额度校验（Node-style before_model）：从长期记忆查额度，不足则 jump_to="end" 中断流程。

  2. 动态工具集（Wrap-style wrap_model_call）：按用户身份裁剪暴露给模型的工具，request.override(tools=tools) 仅本次生效。

  3. 动态提示词（@dynamic_prompt）：按用户偏好动态改系统提示词——Ada 要简洁，Blackwell 要长篇大论，同一个问题得到截然不同的回答。

八、一张图记住全部

                    ┌─────────────────────────────────┐
                    │          LangChain Agent         │
                    └─────────────────────────────────┘
                                       │
        ┌──────────────┬───────────────┴──────────────┬──────────────┐
        │              │                              │              │
     短期记忆         长期记忆                      静态上下文       治理策略
  State + Check-    store → namespace             context        裁剪 / 删除
  pointer + thread   → key → value              （run 不变）       / 摘要
  （会话级隔离）     （跨会话共享）                                  （控制膨胀）

一句话总结：

  • 想让 Agent 在一条对话里记住你 → 用 checkpointer + thread_id（短期记忆）。

  • 想让 Agent 跨所有对话记住你 → 用 store（长期记忆）。

  • 想给 Agent 传入不可变的用户信息 → 用 context（静态上下文）。

  • 怕记忆无限膨胀 → 用裁剪 / 删除 / 摘要治理。