LangChain 中的 Memory，本质上是为了让 Agent 保留上一轮执行留下的状态，并在后续调用中继续使用。

这些状态最终会影响下一轮 Agent 能看到什么上下文。这里的内容不只是聊天记录，还可能包括工具调用、工具结果、自定义数据、中断位置，以及后续待执行的节点。

如果这些信息在一次 invoke() 结束后就全部丢失，那么下一次调用只能重新开始。Agent 不仅无法延续之前的任务，模型拿到的上下文也会变得不完整。

LangChain 的 短期记忆（Short-term Memory） 负责的就是这件事：在同一段会话中保存 Agent 的执行状态，并在下一次调用时恢复，使后续执行能够基于已有状态继续进行。

要理解 LangChain 的短期记忆机制，需要先理清四个核心概念：

代码片段Text
Thread
  ↓
State
  ↓
Checkpoint
  ↓
Checkpointer

Thread 用来划分会话，State 保存当前状态，Checkpoint 记录某个执行时刻的状态快照，Checkpointer 则负责这些快照的保存和读取。

后面的短期记忆机制，基本都围绕这四个概念展开。

一、Memory 保存的是什么

在 LangChain Agent 中，短期记忆并不是一个独立的 Memory 对象。

它本质上是：

保存在当前 Thread 中，并能够跨多次 Agent 调用持续存在的 State。

这里首先要区分两个概念：

代码片段Text
Message
只是 Agent State 中的一部分

State
才是短期记忆保存的完整状态

一次简单对话的 State 可能主要由 messages 构成。

但真实 Agent 往往还需要维护更多数据，例如：

代码片段Text
messages
user_id
preferences
uploaded_files
current_step
authentication_status
...

因此，短期记忆的作用不只是让模型“记得上一句话”。

它保存的是 当前会话中 Agent 后续执行仍然需要的数据。

这些状态通过 Thread 进行隔离，在执行过程中形成 Checkpoint，并由 Checkpointer 持久化。

四者之间的关系如下：

代码片段Text
Thread
确定状态属于哪段会话

State
保存 Agent 当前的数据

Checkpoint
记录某个执行时刻的 State

Checkpointer
负责保存和读取 Checkpoint

二、短期记忆的边界

短期记忆最重要的边界是 Thread Scope，也就是 Thread 级别。

一个 Thread 可以理解成一段独立会话。

例如，一个客服 Agent 同时存在两段会话：

代码片段Text
Thread A
用户：我叫 Bob
Agent：你好 Bob

Thread B
用户：我叫 Alice
Agent：你好 Alice

随后在 Thread A 中继续提问：

代码片段Text
我叫什么名字？

Agent 应该能够从当前 Thread 的状态中获得 Bob 这个信息。

而在 Thread B 中提出同样的问题，则不应该读取到 Bob。

因此，短期记忆解决的是：

在同一个 Thread 的多次执行之间持续保存状态。

如果需要跨 Thread 保存用户画像、长期偏好或历史事实，则属于 长期记忆（Long-term Memory） 的范围，通常由 Store 负责。

LangGraph 提供了 InMemorySaver，用于在内存中保存 Agent 的检查点（checkpoint），从而记录每一步的执行状态。

我们来看一个最简单的短期记忆示例。

创建 Agent，并配置 InMemorySaver：

代码片段Python
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
 
checkpointer = InMemorySaver()
 
agent = create_agent(
    model="openai:gpt-5.5",
    tools=[],
    checkpointer=checkpointer,
)
 
config = {
    "configurable": {
        "thread_id": "user-1"
    }
}
 
result1 = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "我叫 Bob。"}
        ]
    },
    config=config,
)
 
result2 = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "我叫什么名字？"}
        ]
    },
    config=config,
)
 
print(result2["messages"][-1].content)

运行结果大体如下：

代码片段Text
你叫 Bob。

第二次调用只提交了“我叫什么名字”，并没有手动再次传入第一轮 Message。

Agent 仍然能够得到 Bob，是因为两次执行使用了相同的 thread_id。

第一次执行产生的 State 被 Checkpointer 保存；第二次执行开始时，Runtime 根据 thread_id 重新加载此前保存的状态。这里的 thread_id 由调用方通过 config 传入，用于标识对话线程的唯一性，Checkpointer 正是依赖它来定位并恢复对应的历史快照。

因此，尽管两次 invoke() 传入的 messages 列表彼此独立，第二次调用也并没有去手工追加第一次的消息；但 Checkpointer 会根据相同 thread_id 恢复第一次执行后的 State，再将第二次传入的消息合并到已有 messages 中。

这就是短期记忆最基本的工作方式。

三、Thread 如何划分会话

thread_id 是理解短期记忆的第一个关键点。

它既不是某条 Message 的 ID，也不是某一次 invoke() 的执行 ID，而是：

一组连续 Agent 执行共享状态的标识符。

可以把它看成一段会话对应的主键：

代码片段Text
thread_id = "user-1"

        invoke 1
           ↓
        invoke 2
           ↓
        invoke 3
           ↓

     同一份 Thread State

只要继续使用同一个 thread_id，Checkpointer 就能够定位到这段会话对应的 Checkpoint，并恢复已有 State。

更换 thread_id，则进入另一段独立会话。

例如，把前面的第二次调用修改为：

代码片段Python
another_config = {
    "configurable": {
        "thread_id": "user-2"
    }
}
 
result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "我叫什么名字？"}
        ]
    },
    config=another_config,
)
 
print(result["messages"][-1].content)

运行结果大致会是：

代码片段Text
你还没有告诉我你的名字。

这里并不是 Agent 丢失了原来的状态，而是本次调用使用了新的 thread_id。

对于 Checkpointer 来说：

代码片段Text
thread_id = user-1

和：

代码片段Text
thread_id = user-2

对应的是两套独立的状态历史。

因此，在实际项目中，thread_id 更适合对应明确的业务会话，例如：

代码片段Text
chat-20260820-001
customer-service-session-9281
order-assistant-10086

需要注意，不宜长期直接使用 user_id 作为 thread_id。

同一个用户可能同时存在多段独立会话。如果所有对话始终共用一个 Thread，State 会持续增长，不相关的话题也会进入同一个上下文。

user_id 描述的是“谁”，而 thread_id 描述的是“哪一段会话”。

两者承担的职责并不相同。

四、State 保存 Agent 当前状态

Thread 定义了状态的边界，真正被保存的数据则是 State（状态）。

State 是 Agent 在执行过程中持续读取和更新的一组数据。

LangChain Agent 默认使用 AgentState，其中最常见的字段是：

代码片段Text
messages

平时看到的“Agent 记得上一轮对话”，实际上就是 messages 作为 State 的一部分，被保存到了当前 Thread 中。

但 State 并不限于 Message。

真实应用还可能需要保存：

代码片段Text
messages
user_id
preferences
uploaded_files
current_step
authentication_status
...

例如，可以扩展默认的 AgentState：

代码片段Python
from langchain.agents import AgentState, create_agent
from langgraph.checkpoint.memory import InMemorySaver
 
 
class CustomAgentState(AgentState):
    user_id: str
    preferences: dict
 
 
agent = create_agent(
    model="openai:gpt-5.5",
    tools=[],
    state_schema=CustomAgentState,
    checkpointer=InMemorySaver(),
)
 
config = {
    "configurable": {
        "thread_id": "thread-1001"
    }
}
 
result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "以后回答尽量简洁。"}
        ],
        "user_id": "u1001",
        "preferences": {
            "answer_style": "concise"
        },
    },
    config=config,
)
 
snapshot = agent.get_state(config)
 
print(snapshot.values["user_id"])
print(snapshot.values["preferences"])

运行结果

代码片段Text
u1001
{'answer_style': 'concise'}

这里保存的已经不只是对话消息。

user_id 和 preferences 同样属于当前 Agent State，并且可以随着 Thread 一起保存和恢复。

因此：

代码片段Text
Message
只是 State 中的一部分

State
才是短期记忆的实际载体

上传文件、任务阶段、认证状态以及工具执行过程中产生的数据，只要后续仍需要在当前 Thread 中使用，都可以设计为 State 的一部分。

五、Checkpoint 如何记录状态

有了 State，还需要解决另一个问题：

State 应该在什么时候保存？

如果只在整个 Agent 执行完成之后保存一次，那么执行过程中一旦发生异常、中断，或者进入人工审批阶段，此前产生的状态仍然可能丢失。

LangGraph 通过 Checkpoint（检查点） 解决这个问题。

Checkpoint 是：

某一个执行时刻的 State 快照。

LangGraph 会在执行过程中的 super-step 边界保存 Checkpoint。

例如一个简单执行流程：

代码片段Text
START
  ↓
Model
  ↓
Tool
  ↓
Model
  ↓
END

系统并不是只在 END 时记录最终状态，而会随着执行推进形成多个状态快照。

一个 Thread 的状态历史可能类似：

代码片段Text
Thread 1001

Checkpoint 1
messages = [...]

      ↓

Checkpoint 2
messages = [...]
tool result = ...

      ↓

Checkpoint 3
messages = [...]
final answer = ...

因此：

代码片段Text
Thread
表示一段连续的执行历史

Checkpoint
表示这段历史中某个时刻的状态快照

一个 Thread 可以包含多个 Checkpoint。

这也是 LangGraph 能够支持状态历史、Human-in-the-loop、故障恢复和 Time Travel 的基础。

系统不仅保存“当前状态”，还保留执行过程中形成的状态节点。

可以通过 get_state() 查看当前 Thread 的最新状态：

代码片段Python
snapshot = agent.get_state(config)
 
print(snapshot.values.keys())
print(snapshot.next)
print(snapshot.config["configurable"]["thread_id"])

运行结果

对于已经正常完成的 Agent，一次典型结果如下：

代码片段Text
dict_keys(['messages', 'user_id', 'preferences'])
()
thread-1001

其中，values 保存当前 Checkpoint 对应的 State。

next 表示接下来等待执行的节点。

这里返回空元组：

代码片段Text
()

说明当前执行已经结束。

如果 Agent 停留在某个中断点，next 中则可能包含后续等待执行的节点。

此外，StateSnapshot 还可以包含 Checkpoint ID、父 Checkpoint、执行元数据以及待执行任务等信息。

Checkpoint 因此不仅用于保存聊天上下文，也记录了 Agent 当前处于什么执行位置。

六、Checkpointer 负责持久化

真正负责保存 Checkpoint 的组件叫 Checkpointer。

前面的示例使用了：

代码片段Python
InMemorySaver()

它会将 Checkpoint 保存在当前 Python 进程的内存中。

到这里，Thread、State、Checkpoint 和 Checkpointer 的关系可以完整表示为：

代码片段Text
Thread
  │
  └── 包含多个 Checkpoint
          │
          └── 保存不同阶段的 State
                  │
                  └── 由 Checkpointer 持久化

Checkpointer 位于 LangGraph Runtime 和底层存储之间。

Agent 执行过程中，Runtime 负责产生 State 的变化；Checkpointer 则负责将 Checkpoint 和相关中间写入保存下来。

下一次执行时，再根据 thread_id 读取对应状态。

InMemorySaver 的边界

InMemorySaver 适合：

  • 学习
  • 本地开发
  • 单元测试
  • Demo

因为数据只存在于当前 Python 进程中。

程序退出之后：

代码片段Text
Python Process
      ↓
内存释放
      ↓
Checkpoint 消失

因此，它并不适合作为生产环境中的持久化方案。

如果需要在应用重启之后继续恢复 Thread，就需要使用数据库支持的 Checkpointer。

例如 PostgreSQL：

代码片段Bash
pip install -U langgraph-checkpoint-postgres "psycopg[binary]"

生产环境可以使用 PostgresSaver。

此外，也存在 SQLite、Azure Cosmos DB 等 Checkpointer 实现，可以根据应用的部署方式和持久化要求选择。

因此，判断短期记忆是否真正具备持久化能力，关键不在于“是否启用了 Memory”，而在于两个问题：

代码片段Text
使用了什么 Checkpointer？

Checkpoint 最终保存在哪里？

如果底层只是 InMemorySaver，进程结束之后状态仍然会消失。

如果使用数据库 Checkpointer，则可以跨进程、跨重启恢复 Thread 状态。

七、Checkpoint 如何支持恢复执行

Checkpoint 的价值不仅体现在多轮对话。

对于 Agent 系统而言，更重要的一项能力是：

执行可以暂停，并在之后从已有状态继续。

Human-in-the-loop 是一个典型场景。

假设 Agent 准备执行删除数据的操作：

代码片段Text
用户请求
   ↓
模型决定调用 delete_records
   ↓
等待人工审批
   ↓
审批通过
   ↓
继续执行工具

等待审批可能只有几秒，也可能持续数小时。

这种情况下，不能依赖原来的 Python 函数一直保持阻塞。

LangChain 的 Human-in-the-loop Middleware 会在需要审批的 Tool Call 前触发 interrupt。

此时 Checkpointer 保存当前 Agent State。

等审批结果到达之后，再使用相同的 thread_id 恢复执行。

例如：

代码片段Python
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
 
 
def delete_records(table: str) -> str:
    """Delete old records from a table."""
    return f"Deleted old records from {table}"
 
 
agent = create_agent(
    model="openai:gpt-5.5",
    tools=[delete_records],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "delete_records": True
            }
        )
    ],
    checkpointer=InMemorySaver(),
)
 
config = {
    "configurable": {
        "thread_id": "cleanup-task-1"
    }
}
 
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "删除 logs 表中的旧记录"
            }
        ]
    },
    config=config,
    version="v2",
)
 
if result.interrupts:
    result = agent.invoke(
        Command(
            resume={
                "decisions": [
                    {"type": "approve"}
                ]
            }
        ),
        config=config,
        version="v2",
    )
 
print(result.value["messages"][-1].content)

运行结果如下：

代码片段Text
已删除 logs 表中的旧记录。

第一次 invoke() 执行到审批节点后中断。

此时 Checkpointer 保存的并不只是用户的 Message，还包括当前 Agent State，以及恢复执行所需要的状态信息。

审批通过之后，第二次调用传入：

代码片段Python
Command(resume=...)

同时继续使用原来的：

代码片段Python
thread_id = "cleanup-task-1"

Runtime 根据这个 Thread 定位此前保存的 Checkpoint，从中断位置继续执行，而不是重新从入口运行整个 Agent。

因此，Memory 与 Resume 看起来解决的是两个不同的问题：

代码片段Text
Memory
让 Agent 保留之前的状态

Resume
让 Agent 从之前的状态继续执行

但两者底层依赖的是同一套机制：

State 的持久化与恢复。

这也是为什么 Checkpoint 对 Agent 的意义远大于保存聊天记录。

它同时构成了 Agent 执行可靠性的一部分。

八、实际项目中如何选择

把前面的机制放回实际项目，就比较容易判断什么时候需要短期记忆。

1. 单次模型调用

如果只是一次独立调用：

代码片段Text
Input
  ↓
Model
  ↓
Output

执行结束之后没有任何状态需要继续使用，就没有必要引入 Thread 级短期记忆。

2. 普通多轮聊天

如果应用需要：

代码片段Text
同一个 Thread
      ↓
多次 invoke
      ↓
保留历史 Message

配置 Checkpointer 就可以获得基本的短期记忆能力。

每次执行时，Agent 都可以恢复当前 Thread 已经保存的 messages。

3. 带业务状态的 Agent

如果 Agent 除了 Message，还需要持续保存：

  • 当前任务阶段
  • 用户认证状态
  • 文件信息
  • 临时业务参数
  • 工具产生的数据

就可以扩展 AgentState。

这时短期记忆保存的已经不只是聊天上下文，而是当前 Agent 的完整会话状态。

4. 需要暂停与恢复的工作流

如果系统涉及：

  • Human-in-the-loop
  • 人工审批
  • Agent Interrupt
  • 故障恢复
  • 长时间任务
  • 恢复执行

那么 Checkpoint 就不再只是一个对话记忆能力。

它已经成为 Agent Runtime 的基础设施。

没有可靠的状态持久化，Agent 在中断之后就无法准确恢复到之前的执行位置。

5. 开发环境与生产环境

开发阶段通常可以使用：

代码片段Text
InMemorySaver

它简单、轻量，适合测试和本地调试。

生产环境如果要求：

代码片段Text
应用重启
服务器切换
长时间暂停
分布式执行

之后仍然能够恢复状态，就应该使用数据库支持的 Checkpointer。

例如：

代码片段Text
PostgreSQL
SQLite
Azure Cosmos DB
...

最终选择取决于部署方式、数据规模以及可靠性要求。

九、短期记忆不是保存得越多越好

短期记忆能够保存 Thread 状态，并不意味着所有数据都应该一直保留。

其中最明显的问题是 messages。

随着会话不断进行：

代码片段Text
messages
   ↓
越来越多
   ↓
模型输入越来越长

最终会影响：

  • 上下文窗口占用
  • Token 成本
  • 模型延迟
  • 推理质量

因此，状态持久化 和 模型上下文管理 是两个相关但并不相同的问题。

Checkpointer 可以完整保存 Thread 的状态历史，但并不意味着每次模型调用都应该把全部历史 Message 原样放入上下文。

对于较长的 Thread，通常还需要结合：

  • Message trimming
  • Message deletion
  • Summarization
  • Context Editing

等机制控制模型实际看到的上下文。

所以更准确地说：

代码片段Text
Memory
解决状态如何保存

Context Engineering
解决模型当前应该看到哪些状态

前者保证状态不会丢失，后者控制这些状态如何进入模型上下文。

这两个问题需要分开处理。

十、总结

LangChain 的短期记忆，本质上是一套 Thread 级 Agent State 持久化机制。

它的核心关系可以整理为：

代码片段Text
Thread
确定状态属于哪段会话

State
保存 Agent 当前的数据

Checkpoint
记录执行过程中的 State 快照

Checkpointer
负责保存和读取 Checkpoint

Resume
从已有 Checkpoint 继续执行

同一个 thread_id 能够延续上一轮对话，并不是因为模型本身产生了记忆。

真正发生的是：

代码片段Text
第一次执行
    ↓
产生 State
    ↓
形成 Checkpoint
    ↓
Checkpointer 保存

下一次执行
    ↓
根据 thread_id
    ↓
加载 Checkpoint
    ↓
恢复 State
    ↓
继续运行

因此，Message 只是短期记忆的一部分。

一旦进入 Agent 场景，需要保存的往往还包括任务状态、工具结果、中断位置以及后续执行信息。

理解这一点之后，Thread、State、Checkpoint 和 Checkpointer 就不再是几组零散概念，而是一条完整的状态持久化链路。

本文讨论的是 Thread 内部的短期记忆。

实际应用中还有另一类需求：即使用户结束当前 Thread、重新开启一段会话，Agent 仍然能够知道他的偏好、资料以及过去形成的重要信息。

这类跨 Thread 数据不再主要依赖 Checkpoint，而是进入 LangChain 的 长期记忆（Long-term Memory） 机制，并涉及 Store、Namespace、Key 以及跨会话数据搜索。

这也是下一篇要继续介绍的内容。