代码仓库ChainReaction

找 LangChain 的记忆功能时，很多人会先去翻有没有一个叫 Memory 的类。v1 里没有这个东西。短期记忆就是 agent state 里的一份消息历史，加上你自己挂上去的字段，再交给 checkpointer 按 thread 存下来。把这三件事的分工弄清楚，后面大部分问题都能自己判断。

这篇先用最小例子跑通两轮对话，证明第二轮真的记得第一轮；然后解释 thread_id 和 checkpointer 的关系，以及换 thread 之后会发生什么；接着扩展自定义 state 字段，最后处理消息历史把上下文窗口撑爆的情况。文里的终端输出都是在 Windows 上连 DeepSeek 真跑出来的，脚本放在仓库的 ShortTermMemory/ 目录下。

顺带划一下短期和长期的分界。短期记忆活在一个 thread 里，thread 结束或者你换掉 thread_id，它就够不着了。跨会话记住用户偏好、历史结论，那是 long-term memory 的活，用的是另一套 Store 机制。这篇只谈 thread 内的部分。

短期记忆就是 state

create_agent 默认用 AgentState 管记忆，里面只有一个内置字段 messages，类型是 list[BaseMessage]，装的是当前 thread 的完整对话历史。官方文档把它描述成 append-only：新消息往后面加，已有的不会被替换。

这就是为什么翻遍文档也找不到一个 Memory 类。真要做成独立模块，它得自己维护消息列表、自己判断什么时候读、什么时候写、还要自己处理并发。LangChain 把这部分交给了图里的 state 和 checkpointer，agent 本身保持无状态。好处是 state 里除了消息还能放别的东西，工具和中间件读写的是同一份数据，不用在几个系统之间同步。

想让它跨轮记住东西，就要在创建 agent 时给一个 checkpointer：

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
import os
from langchain_openai import ChatOpenAI

# 全篇共用的模型配置，后面片段里出现的 model 都指这个
model = ChatOpenAI(
    api_key=os.getenv('DEEPSEEK_API_KEY'),
    base_url="https://api.deepseek.com/v1",
    model="deepseek-chat",
    temperature=0.1,
    max_tokens=1000,
)

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

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "thread-1"}}

r1 = agent.invoke(
    {"messages": [{"role": "user", "content": "你好，我叫小林，最近在学 LangChain。"}]},
    config,
)
print(r1["messages"][-1].content)

r2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？"}]},
    config,
)
print(r2["messages"][-1].content)
print("消息总条数:", len(r2["messages"]))

真实输出（长回复做了截断）：

1
2
3
4
5
6
7
8
9
[第一轮] 助手: 你好，小林！很高兴认识你
（回复较长，此处省略）

[第一轮] state 里的消息角色: ['human', 'ai']

[第二轮] 助手: 你叫小林
（回复较长，此处省略）
[第二轮] state 里的消息角色: ['human', 'ai', 'human', 'ai']
[第二轮] 消息总条数: 4

第二轮没有重新告诉模型名字，它照样答出来了，因为 thread-1 的 state 里躺着第一轮的 human 和 ai 两条消息。调用 agent 时传入的消息是一个增量更新，LangGraph 把它和 checkpointer 里取出的历史合并，再交给模型。

注意这里没有”把历史塞进 prompt”的显式步骤。state 在每个步骤开始时被读取，在 agent 被调用或某个步骤（比如工具调用）完成时写回。你看到的 r2["messages"] 是合并后的完整列表，不是这一轮新产生的两条。

一次多轮对话里，state 和 checkpointer 的配合大致是这样：

checkpointer 存下来的是一串快照，每份对应图执行的一次 super-step，super-step 可以理解成一次 tick。一次两轮的对话会产生好几份快照，每份都带着当时的完整 state。这个特性后面讲状态膨胀时还要用到。

thread_id 和 checkpointer

checkpointer 用 thread_id 作为主键来存取状态。挂了 checkpointer 却不传 thread_id，LangGraph 直接抛异常，不会”默认开一个”。我实测了一下：

1
[有 checkpointer，无 thread_id] ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id

这个报错信息挺直白，照着补 thread_id 就行。

反过来，不挂 checkpointer 的时候，传 thread_id 没有任何作用。同一个 thread_id 连续调两轮，第二轮的状态里只有自己这一轮的消息：

1
2
3
[无 checkpointer] 第二轮 state 消息条数: 2
[无 checkpointer] 第二轮 state 消息角色: ['human', 'ai']
[无 checkpointer] 助手回答: 我不知道你的名字哦！我们还没有正式介绍过呢。

记忆不在 agent 对象里，而在 checkpointer 里。agent 每次都是无状态的，是 checkpointer 把上一轮的状态递了回来。

换一个 thread_id，就是换一个会话。下面这个脚本用同一个 agent，先在 thread-A 里告诉它喜欢的颜色，然后去 thread-B 问同一件事：

1
2
3
4
5
6
7
8
9
10
11
12
[thread-A] 用户: 记住：我最喜欢的颜色是蓝色。
[thread-A] 助手: 好的，我记住了！你最喜欢的颜色是蓝色。
[thread-A] 消息条数: 2

[thread-B] 用户: 我最喜欢什么颜色？
[thread-B] 助手: 哈哈，这个问题可难倒我啦
（回复较长，此处省略）
[thread-B] 消息条数: 2

[thread-A 再次提问] 用户: 我最喜欢什么颜色？
[thread-A 再次提问] 助手: 你最喜欢蓝色呀！这是你之前告诉我的，我记住啦。
[thread-A 再次提问] 消息条数: 4

thread-B 完全不知道蓝色的事，而回到 thread-A 之后记忆还在。消息条数也印证了这一点：thread-B 是 2 条，thread-A 累积到 4 条。

实际项目里，thread_id 通常按会话生成，不按用户生成。一个用户可能同时开好几个对话，每个对话一个 thread_id，聊天记录才不会串在一起。用户级的东西放长期记忆，会话级的放 thread。

create_agent 返回的是编译好的图，agent.get_state(config) 能直接读当前状态，agent.get_state_history(config) 能拿到这个 thread 的全部快照，调试和回放时很省事。我后面数 checkpoint 用的就是它。

选哪种 saver，差别比想象中大：

InMemorySaver SqliteSaver / PostgresSaver
数据放哪 进程内存 SQLite 文件 / Postgres 库
进程重启 状态全丢 保留
多进程或多副本 各存各的，互相看不见 共享同一份
依赖 langgraph-checkpoint 自带 另装 langgraph-checkpoint-sqlite 或 langgraph-checkpoint-postgres
适合 本地开发和测试 生产

官方把 InMemorySaver 定位成实验用，SQLite 适合本地工作流，Postgres 和 MongoDB 才是生产选项。Postgres 的用法长这样，setup() 会自动建表：

1
2
3
4
5
6
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    agent = create_agent(model=model, tools=[], checkpointer=checkpointer)

自定义 state 字段

AgentState 可以继承，加自己需要的字段，然后通过 state_schema 传给 create_agent：

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
from typing import Any

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import after_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from typing_extensions import NotRequired


class CustomState(AgentState):
    user_id: NotRequired[str]
    preferences: NotRequired[dict]
    model_calls: NotRequired[int]


@after_model(state_schema=CustomState)
def count_model_calls(state: CustomState, runtime: Runtime) -> dict[str, Any] | None:
    return {"model_calls": state.get("model_calls", 0) + 1}


agent = create_agent(
    model=model,
    tools=[],
    state_schema=CustomState,
    middleware=[count_model_calls],
    checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "thread-state"}}

r1 = agent.invoke(
    {
        "messages": [{"role": "user", "content": "你好，我是小林。"}],
        "user_id": "user_123",
        "preferences": {"theme": "dark", "lang": "zh"},
    },
    config,
)
print(r1.get("user_id"), r1.get("preferences"), r1.get("model_calls"))

r2 = agent.invoke(
    {"messages": [{"role": "user", "content": "帮我记一下，我住在杭州。"}]},
    config,
)
print(r2.get("user_id"), r2.get("preferences"), r2.get("model_calls"))

r3 = agent.invoke(
    {"messages": [{"role": "user", "content": "我住哪？我的 user_id 是多少？"}]},
    config,
)
print(r3.get("user_id"), r3.get("preferences"), r3.get("model_calls"))

第二轮只传了 messages，不再传 user_id 和 preferences。真实输出：

1
2
3
4
5
6
[第一轮] user_id='user_123' preferences={'theme': 'dark', 'lang': 'zh'} model_calls=1

[第二轮] 只传 messages，不再传 user_id / preferences
[第二轮] user_id='user_123' preferences={'theme': 'dark', 'lang': 'zh'} model_calls=2

[第三轮] user_id='user_123' preferences={'theme': 'dark', 'lang': 'zh'} model_calls=3

自定义字段和 messages 一样被 checkpointer 保存，所以只需要在第一轮给一次。model_calls 每轮加一，靠的是 after_model 钩子返回的字典，它通过 state 的 reducer 合并回去。

字段声明用了 NotRequired，因为第一轮 invoke 时你未必会传。它属于类型标注，运行时不会因为少传就拦你：我试过把字段标成必填却不传，invoke 照样跑通，返回的 state 里连这个键都没有，state.get("user_id") 拿到的是 None。所以自定义字段要么第一轮就给全，要么在代码里自己兜住缺省值。合并规则按字段走：messages 配的是 add_messages，做追加；普通字段默认是覆盖。

还有个容易和 state 混在一起的概念是 context。context_schema 定义每次运行传入的只读上下文，比如 user_id、数据库连接、当前请求的元数据，工具里用 runtime.context 读。它不进 checkpoint，也不跨轮累积。要跨轮记住的放 state，只在单次运行里传递的放 context，分清楚能省不少事。

有个地方容易误会。第三轮我问模型”我的 user_id 是多少”，它答不上来。user_id 存在 state 里，但 state 字段不会自动进入 prompt，模型能看到只有 messages。想让模型知道，得靠中间件把字段拼进 system prompt，或者用工具读 runtime.state 再以工具结果的形式送回对话。

用中间件定义状态

上面把 state_schema=CustomState 同时给了 create_agent 和装饰器。如果这个字段只服务于某一个中间件，更干净的做法是只声明在中间件上：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, AgentState
from typing_extensions import NotRequired


class CustomState(AgentState):
    model_call_count: NotRequired[int]


class CallCounterMiddleware(AgentMiddleware[CustomState]):
    state_schema = CustomState

    def before_model(self, state, runtime):
        count = state.get("model_call_count", 0)
        if count > 10:
            return {"jump_to": "end"}
        return None

    def after_model(self, state, runtime):
        return {"model_call_count": state.get("model_call_count", 0) + 1}


agent = create_agent(model=model, tools=[], middleware=[CallCounterMiddleware()])

装饰器写法也支持 @before_model(state_schema=CustomState)。文档里 AgentMiddleware 的三个编译期属性，state_schema、tools、transformers，都是这个思路：能力跟着中间件走，工厂在编译时把它们合并进 agent 的 state。

我倾向中间件持有自己的字段，原因是可复用。计数器、调用次数、限流标志这类字段是中间件的实现细节，写进 create_agent(state_schema=...) 之后，换个 agent 就得再抄一遍，字段名还可能撞。反过来，user_id、preferences 这种业务层到处要读的字段，放 create_agent 的 state_schema 更合适，因为它属于 agent 本身，不归某个中间件管。

读和改消息历史

读有几种入口。调用结束后从返回值的 ["messages"] 里取，中间件里从 state["messages"] 取，工具里通过 ToolRuntime 取：

1
2
3
4
5
6
7
8
from langchain.tools import tool, ToolRuntime


@tool
def get_user_info(runtime: ToolRuntime) -> str:
    """Look up user info."""
    user_id = runtime.state["user_id"]
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

runtime 参数对模型不可见，模型看不到它，但工具能拿到完整 state。

返回值里的 messages 包含 tool 消息，工具调用的中间结果全在里面。只想看最终回答就取最后一条，想检查这一轮到底发生了什么，就顺着 tool_calls 和对应的 ToolMessage 往下看。别把它当成只有人机对话，agent 循环里每次工具往返都会留下痕迹。

改历史要用 RemoveMessage，而且得让 messages 走 add_messages 这个 reducer。AgentState 默认就是，所以直接写就行。删指定几条：

1
2
3
4
5
6
7
from langchain.messages import RemoveMessage


def delete_messages(state):
    messages = state["messages"]
    if len(messages) > 2:
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}

清空全部用 REMOVE_ALL_MESSAGES：

1
2
3
4
5
from langgraph.graph.message import REMOVE_ALL_MESSAGES


def clear_messages(state):
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}

工具里也能改 state，返回一个 Command 即可：

1
2
3
4
5
6
7
from langgraph.types import Command


@tool
def update_user_info(runtime: ToolRuntime) -> Command:
    """Look up and update user info."""
    return Command(update={"user_name": "John Smith"})

删消息时注意结果历史是否合法。有些 provider 要求历史以 user 消息开头；带 tool call 的 assistant 消息后面必须跟着对应的 tool 结果消息。删中间那段很容易踩这个坑，报错通常来自 provider 而不是 LangChain。

state 在 agent 一次运行里的读写位置：

上下文窗口压力

聊得越久，messages 越长，迟早撞上模型的上下文窗口。就算窗口够大，长历史也会拖慢响应、抬高成本，模型还容易被无关的旧内容带偏。官方给了两类常见做法：修剪和摘要。

判断该不该动手可以先看用量。AIMessage 上带 usage_metadata，里面有 input_tokens 和 output_tokens，把最近几轮加起来就能估出增长趋势。不过这是事后观察，真正拦在调用前的是中间件。

修剪

修剪就是丢消息。写一个 before_model 中间件，在调用模型前把历史砍短：

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
from typing import Any

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime


@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """消息超过 4 条时，只留第一条和最近两条。"""
    messages = state["messages"]
    if len(messages) <= 4:
        return None
    first_msg = messages[0]
    recent_messages = messages[-2:]
    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),
            first_msg,
            *recent_messages,
        ]
    }


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

跑四轮，每轮打印 state 里的消息角色：

1
2
3
4
5
6
7
8
9
10
11
12
13
第 1 轮 state 消息角色: ['human', 'ai']
第 1 轮 state 消息条数: 2

第 2 轮 state 消息角色: ['human', 'ai', 'human', 'ai']
第 2 轮 state 消息条数: 4

第 3 轮 state 消息角色: ['human', 'ai', 'human', 'ai']
第 3 轮 state 消息条数: 4

第 4 轮 用户: 我叫什么名字？
第 4 轮 助手: 你叫小林。
第 4 轮 state 消息角色: ['human', 'ai', 'human', 'ai']
第 4 轮 state 消息条数: 4

从第 3 轮开始条数一直卡在 4，因为中间那两条被丢掉了。第 4 轮还能答出名字，是因为我特意保留了第一条。如果不保留第一条，模型就会失忆。修剪策略的核心是决定留哪条，而不是砍掉多少条。

修剪全程不需要模型参与，执行快，结果也可预测。代价是丢掉的对话永久消失，哪怕后面用户又提起那个话题，模型也无从查证。要不要保留第一条，取决于你的 system prompt 里有没有把关键信息固化下来。如果关键事实都在 system prompt 里，第一条 human 消息其实也可以丢。

摘要

摘要用模型把旧消息压成一段文字，再用内置的 SummarizationMiddleware：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger=("messages", 6),
            keep=("messages", 2),
        )
    ],
    checkpointer=InMemorySaver(),
)

trigger 是触发条件，keep 是摘要之后保留多少最近的消息。两个参数都支持 tokens、messages、fraction 三种单位。trigger 传一个列表就是 OR 逻辑，任意一条满足即触发；传字典是 AND 逻辑，里面所有阈值都要满足。fraction 依赖模型的 profile 数据，拿不到就换成 tokens 或 messages，或者自己给模型配 profile。跑四轮的真实输出：

1
2
3
4
5
6
7
8
第 3 轮 state 消息角色: ['human', 'ai', 'human', 'ai', 'human', 'ai']
第 3 轮 state 消息条数: 6

第 4 轮 用户: 我叫什么名字？我住哪？
第 4 轮 助手: 你叫小林，住在杭州。
第 4 轮 state 消息角色: ['human', 'ai', 'human', 'ai']
第 4 轮 state 消息条数: 4
第 4 轮 首条消息(human): Here is a summary of the conversation to date:  ## SESSION INTENT  The user (小林, living in Hangzhou) is engaging in a casual ...

第 4 轮进模型前消息到了 7 条，超过阈值，中间那段被替换成一条摘要消息，所以最后 state 只剩 4 条。名字和城市都还在，因为摘要把它们写进去了。

两种策略的取舍：

维度 修剪 摘要
做法 直接丢弃整条消息 用模型把旧消息压成一段摘要
代码 自己写 before_model 中间件 用 SummarizationMiddleware
额外成本 没有 每次触发多一次模型调用
信息损失 丢掉的内容永久消失 语义保留，细节会被压缩
可控性 高，条数完全由你决定 偏低，摘要内容取决于模型
多模态 原样丢弃 旧图片和音频只剩文字摘要
适合 会话短，旧消息可以不要 长会话，早期信息后面还要用

我自己的默认选择是先修剪。它零成本、行为确定，调试起来省心。等发现”早期信息确实还有用”再换摘要。摘要也不是没有边界，官方特别说明它是纯文本压缩，不会压缩图片和音视频载荷，做图像密集的应用要把媒体放到文件系统或对象存储，历史里只传 URL 或文件引用。

生产环境的几个坑

内存 saver 只适合开发。InMemorySaver 的数据在进程内存里，重启就没了，多开一个 worker 也拿不到另一个 worker 的会话。上线就换 PostgresSaver 或 MongoDBSaver，别用内存 saver 硬扛。

并发 thread 之间是隔离的，不同 thread_id 互不影响，这也是它能直接支撑多用户会话的原因。但同一个 thread_id 上的并发请求要小心，checkpointer 以 thread 为主键，同一 thread 上的多次写入会按顺序落到同一条状态链上，交叉的两轮对话可能互相覆盖。一个会话同一时刻只处理一个请求，或者自己加锁。

状态膨胀值得单独盯。前面提过 checkpointer 在每个 super-step 存一份快照，两轮对话就会留下好几份。我数了一下：

1
2
3
4
5
6
7
8
9
10
11
12
13
# 沿用前面带 checkpointer 的 agent，thread_id 换成 thread-cp
config = {"configurable": {"thread_id": "thread-cp"}}

agent.invoke({"messages": [{"role": "user", "content": "你好，我叫小林。"}]}, config)
agent.invoke({"messages": [{"role": "user", "content": "我在杭州。"}]}, config)

snapshot = agent.get_state(config)
print("当前消息条数:", len(snapshot.values["messages"]))

history = list(agent.get_state_history(config))
print("checkpoint 总数:", len(history))
for snap in reversed(history):
    print("step=%s 消息条数=%d" % (snap.metadata.get("step"), len(snap.values["messages"])))
1
2
3
4
5
6
7
8
9
10
11
12
get_state 拿到的当前消息条数: 4
下一步要执行的节点: ()

thread-cp 一共保存了 6 份 checkpoint

按时间从早到晚，每份 checkpoint 里的消息条数:
  step=-1 消息条数=0
  step=0 消息条数=1
  step=1 消息条数=2
  step=2 消息条数=2
  step=3 消息条数=3
  step=4 消息条数=4

两轮对话，6 份快照，最后 4 条消息被重复存了多次。消息越长、轮次越多，重复得越厉害。上修剪或摘要，给老 thread 设清理周期，自定义字段里别塞大对象。

还有一点，换 saver 的时候 messages 里存的是完整的消息对象，包含工具调用和元数据。如果你打算跨版本升级 LangChain，先在测试环境验证旧 checkpoint 还能不能反序列化出来。

小结

  • 短期记忆没有独立模块，它就是 thread 的 state：messages 加自定义字段，由 checkpointer 按 thread_id 存取。
  • 不挂 checkpointer 就没有记忆，挂了不传 thread_id 会直接抛 ValueError。
  • 换 thread_id 就是换会话，隔离是默认行为；InMemorySaver 只用于开发，生产换数据库 saver。
  • 中间件自己的字段就声明在中间件的 state_schema 上，业务层共用的字段放 create_agent 的 state_schema。
  • 上下文压力优先用修剪，零成本且可控；需要保留早期语义时再上摘要，记得它不压缩多模态内容。