LangChain 1.x Agent 对话记忆

发表于 分类于
基于 create_agent 与 checkpointer，讲解短时记忆、thread_id、SQLite/Postgres 持久化、裁剪与摘要，示例对接火山方舟 Coding Plan。

前言

默认每次 invoke 互不共享历史，Agent 记不住上一轮用户说过什么。
多轮聊天、会话恢复、人工审批中断续跑，都依赖 短时记忆（thread 级状态）。
LangChain 1.x 把对话历史放在 Agent 的 messages 状态里，再用 checkpointer 按 thread_id 持久化快照。
本文只讲对话记忆：如何挂 checkpointer、如何隔离会话、如何在长对话下裁剪或摘要。
Agent 入门与工具见《LangChain 1.x Agent 教程》；摘要等中间件细节见《LangChain 中间件教程》。
示例继续对接 火山方舟 Coding Plan，模型用 ark-code-latest。
下文每个 Python 示例都是完整可运行脚本：复制到项目根目录（与 .env 同级）后执行 uv run python xxx.py 即可。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。
用 uv 初始化工程（示例项目名 langchain-agent-memory）、创建虚拟环境并声明基础依赖（会写入 pyproject.toml 与 uv.lock）。
SQLite / Postgres 等按节所需的额外包，写在对应示例前的安装代码块中。

1
2
3
4
uv init langchain-agent-memory
cd langchain-agent-memory
uv venv --python 3.12
uv add "langchain>=1.0,<2.0" langchain-openai langgraph python-dotenv rich

若已在空目录内初始化，也可写 uv init --name langchain-agent-memory，再执行 uv venv 与 uv add。

后续运行示例脚本时，可直接 uv run python demo.py，一般不必手动 activate 虚拟环境。
若习惯激活，Windows 使用 .venv\Scripts\activate，macOS / Linux 使用 source .venv/bin/activate。

rich 用于在终端里更清楚地查看 messages 与整次 invoke 返回值。
示例里常用 from rich import print as rprint。

在项目根目录创建 .env，写入 Coding Plan 的 API Key 与专用 Base URL。
不要把 .env 提交进 Git。

1
2
OPENAI_API_KEY=你的火山方舟 API Key
OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

示例脚本开头用 load_dotenv() 加载该文件。
请勿把 Base URL 写成普通方舟 .../api/v3，以免无法抵扣 Coding Plan 额度。

实现

工作原理

短时记忆是 线程（thread）级 的：同一 thread_id 共用一份状态，不同 ID 互不影响。
状态默认含 messages（对话历史），也可扩展自定义字段。
每次 invoke 或工具步骤结束后，checkpointer 写入快照；下次同线程调用先读出再继续。
这与跨会话的 长期记忆（Store） 不同：后者跨多个 thread_id 共享用户偏好等，本文末尾仅作边界说明。

启用记忆只需两步。

  1. 创建 Agent 时传入 checkpointer=...。
  2. 每次调用传入 config={"configurable": {"thread_id": "..."}}。

缺少任一步，都会变成「无记忆」或「每次新会话」。

内存续聊

开发与演示用 InMemorySaver 即可：进程内保存，重启即清空。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[],
    system_prompt="你是友好的中文助手，记得用户说过的话。",
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "demo-user-1"}}

r1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫小明，喜欢骑行。"}]},
    config,
)
rprint(r1["messages"][-1])

r2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？有什么爱好？"}]},
    config,
)
rprint(r2["messages"][-1])

第二次提问应能答出「小明」与骑行相关爱好。
注意两轮调用共用同一个 config（同一 thread_id）。

隔离会话

换一个 thread_id 等于新开聊天窗口：历史不会串到另一条线程。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[],
    system_prompt="你是中文助手。不知道的信息请直接说不知道。",
    checkpointer=InMemorySaver(),
)

cfg_a = {"configurable": {"thread_id": "user-a"}}
cfg_b = {"configurable": {"thread_id": "user-b"}}

agent.invoke(
    {"messages": [{"role": "user", "content": "我叫小明。"}]},
    cfg_a,
)

r_b = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    cfg_b,
)
rprint(r_b["messages"][-1])

r_a = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    cfg_a,
)
rprint(r_a["messages"][-1])

user-b 不应知道「小明」；user-a 仍应答出名字。
线上建议用稳定业务键生成 ID，例如 f"chat:{user_id}:{session_id}"，避免随意字符串冲突。

查看状态

需要调试或做 UI「历史回放」时，可用已编译图上的 get_state 读当前快照。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[],
    system_prompt="你是中文助手。",
    checkpointer=InMemorySaver(),
)

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

agent.invoke(
    {"messages": [{"role": "user", "content": "用一句话介绍对话记忆。"}]},
    config,
)

snapshot = agent.get_state(config)
rprint(snapshot.values.get("messages"))

snapshot.values 即该线程当前状态；也可配合 get_state_history 做时间旅行调试（按需查阅官方 Persistence 文档）。

SQLite 持久化

进程重启后仍要续聊时，把 checkpointer 换成文件型 SqliteSaver。
适合单机实验与本地工具；多实例高并发请用 Postgres。
先安装 SQLite checkpointer 包（sqlite3 为标准库，无需再装）。

1
uv add langgraph-checkpoint-sqlite

下面用文件型连接创建 SqliteSaver，同一 thread_id 下连续两轮对话验证续聊。

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
import os
import sqlite3

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.sqlite import SqliteSaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

# check_same_thread=False：实现内部有锁，便于在部分异步场景复用连接
conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
checkpointer = SqliteSaver(conn)

agent = create_agent(
    model=model,
    tools=[],
    system_prompt="你是友好的中文助手，记得用户说过的话。",
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "sqlite-demo-1"}}

r1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫小红，喜欢摄影。"}]},
    config,
)
rprint(r1["messages"][-1])

r2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？爱好是什么？"}]},
    config,
)
rprint(r2["messages"][-1])

也可用上下文管理器写法：with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer:。
把脚本跑两遍、第二次只问名字时，只要 thread_id 不变，仍应答出「小红」。

Postgres 持久化

多机部署或正式服务可用 Postgres 做 checkpointer；跨线程档案则用 PostgresStore。
本节示例演示长期记忆 Store。
先安装 Postgres 相关包与类型扩展。

1
uv add "langgraph-checkpoint-postgres" "psycopg[binary,pool]" typing_extensions langchain-core

按连接串初始化表结构前，请先配置下文 DB_URL。

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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
import os

from dotenv import load_dotenv
from typing_extensions import NotRequired

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import ToolRuntime
from langgraph.store.postgres import PostgresStore

load_dotenv(override=True)

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


class CustomState(AgentState):
    user_id: NotRequired[str]


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


@tool(parse_docstring=True)
def save_user_info(name: str, runtime: ToolRuntime) -> str:
    """
    将客户信息保存在长期记忆中

    Args:
        name : 用户名
        runtime : 工具的运行时

    Returns:
        str : 保存状态
    """
    namespace = ("users",)
    key = runtime.state["user_id"]
    value = {"name": name}
    runtime.store.put(namespace, key, value)
    return "saved"


@tool(parse_docstring=True)
def get_user_info(runtime: ToolRuntime) -> str:
    """
    从长期记忆中读取客户的信息

    Args:
        runtime : 工具的运行时

    Returns:
        str : 用户信息
    """
    namespace = ("users",)
    key = runtime.state["user_id"]
    item = runtime.store.get(namespace, key)
    return str(item.value) if item else "unknown"


DB_URL = os.environ["DB_URL"]

with PostgresStore.from_conn_string(DB_URL) as store:
    store.setup()

    agent = create_agent(
        model=model,
        tools=[save_user_info, get_user_info],
        store=store,
        middleware=[CustomStateMiddleware()],
        system_prompt=(
            "用户提及个人信息时，可以使用工具保存用户信息。"
            "如果用户询问个人信息时，可以尝试使用工具读取用户信息"
        ),
    )

    print("=" * 30, "-> 第一个会话（线程） <-", "=" * 30)
    response1 = agent.invoke(
        {
            "messages": [HumanMessage("你好，很高兴认识你，我是小花")],
            "user_id": "user-1",
        }
    )
    for msg in response1["messages"]:
        msg.pretty_print()

    print("=" * 30, "-> 第二个会话（线程） <-", "=" * 30)
    response2 = agent.invoke(
        {
            "messages": [HumanMessage("我是谁")],
            "user_id": "user-1",
        }
    )
    for msg in response2["messages"]:
        msg.pretty_print()

异步图请改用 AsyncPostgresSaver / AsyncSqliteSaver，并走 ainvoke。
连接串与权限按运维规范管理，勿把密码写进仓库。

环境变量

1
2
3
4
OPENAI_API_KEY="ark-183a42a0-d7d2-4b80-xxxx-39311e89ac22-xxxx"
OPENAI_BASE_URL="https://ark.cn-beijing.volces.com/api/coding/v3"

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

自定义状态

除 messages 外，可扩展 AgentState，把业务字段一并纳入短时记忆并由 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
36
37
38
39
40
41
42
43
44
45
46
import os

from dotenv import load_dotenv
from langchain.agents import AgentState, create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]


class CustomAgentState(AgentState):
    user_id: str
    preferences: dict


model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[],
    state_schema=CustomAgentState,
    system_prompt="你是中文助手。可参考用户偏好作答。",
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "custom-state-1"}}

result = agent.invoke(
    {
        "messages": [{"role": "user", "content": "根据我的偏好，推荐一种沟通语气。"}],
        "user_id": "u_1001",
        "preferences": {"tone": "简洁", "lang": "zh"},
    },
    config,
)
rprint(result["messages"][-1])
rprint({"user_id": result.get("user_id"), "preferences": result.get("preferences")})

自定义字段会随 checkpoint 一起保存；工具里也可通过 ToolRuntime 读写这些状态（见官方 Short-term memory 文档）。

裁剪历史

上下文窗口有限时，可在 before_model 里只保留系统首条与最近若干条，避免撑爆 token。
裁剪会 永久改写 该线程状态中的消息列表（配合 RemoveMessage）。

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
51
52
53
54
55
56
57
58
59
60
61
62
import os
from typing import Any

from dotenv import load_dotenv
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import before_model
from langchain.chat_models import init_chat_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
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]


@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """只保留首条与最近几条，控制上下文长度。"""
    messages = state["messages"]
    if len(messages) <= 4:
        return None

    first_msg = messages[0]
    recent = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),
            first_msg,
            *recent,
        ]
    }


model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[],
    middleware=[trim_messages],
    system_prompt="你是简洁的中文助手。",
    checkpointer=InMemorySaver(),
)

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

agent.invoke({"messages": [{"role": "user", "content": "我叫 Bob。"}]}, config)
agent.invoke({"messages": [{"role": "user", "content": "写一句关于猫的短诗。"}]}, config)
agent.invoke({"messages": [{"role": "user", "content": "再写一句关于狗的。"}]}, config)
final = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    config,
)
rprint(final["messages"][-1])

删除或裁剪后，务必保证历史仍合法：例如 tool call 与对应 ToolMessage 成对出现，部分厂商要求以 user 消息开头。
裁剪可能丢掉早期事实；需要保留要点时改用摘要。

摘要压缩

SummarizationMiddleware 在触发条件满足后，把较早消息压成摘要再继续对话，比硬删更不易丢关键信息。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from rich import print as rprint

load_dotenv()

api_key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=api_key,
    base_url=base_url,
)

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=InMemorySaver(),
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger=("tokens", 4000),
            keep=("messages", 20),
        ),
    ],
    system_prompt="你是中文助手。",
)

config = {"configurable": {"thread_id": "summary-1"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "用一句话说明为什么 Agent 需要对话记忆。"}]},
    config,
)
rprint(result["messages"][-1])

trigger / keep 可按模型窗口与成本调整；完整参数见中间件教程与官方文档。
短对话几乎不会触发摘要，验证时可用更低阈值或更长多轮输入。

扩展

跨多个聊天线程仍要记住「用户偏好、档案」时，应使用 长期记忆（Store），而不是把一切塞进单个 thread_id。
短时记忆负责「这一通会话说了什么」；长期记忆负责「跨会话仍成立的事实」。
二者可同时启用：checkpointer 管线程状态，Store 管跨线程键值。
更多写法见官方 Long-term memory 文档，本文不展开。

验证

按下面顺序确认记忆链路可用。

  1. 配置 Coding Plan 的 .env，跑「内存续聊」脚本，第二问应答出姓名与爱好。
  2. 跑「隔离会话」脚本，确认不同 thread_id 不串话。
  3. 跑「SQLite 持久化」后结束进程再启动，用同一 thread_id 追问，应仍记得上轮信息。
  4. （可选）接入 Postgres 后执行 setup()，用 CHECKPOINT_DB_URI 跑通一轮读写。

总结

  1. 多轮对话 = checkpointer + 稳定的 thread_id；缺一不可。
  2. 演示用 InMemorySaver；单机文件用 SqliteSaver；多机生产用 PostgresSaver。
  3. 用不同 thread_id 隔离用户/会话；用 get_state 排查状态。
  4. 上下文过长时优先摘要，其次裁剪；删除后保持消息结构合法。
  5. 跨会话偏好走长期记忆 Store，不要与短时对话历史混为一谈。