十、LangGraph能力详解：LangGraph 持久化（Persistence）

用户问："今天的天气怎么样？"
助手调用搜索工具，得到答案："今天晴天，25度。"（这个状态，包括对话历史和工具调用结果，被自动保存）
程序崩溃重启。
用户再问："那我需要带伞吗？"（与之前在同一会话下）
LangGraph 加载之前保存的状态，状态里记录了"今天晴天"。
助手看到上下文是晴天，于是直接回答："今天是晴天，您不需要带伞。"（无需再次调用搜索工具）

持久化能力包含两部分

在 LangGraph 中，持久化能力具体包含两部分能力：

线程级持久化：能够自动保存工作流执行过程中的状态快照，维持单次会话的完整上下文。
- 对应上述场景二
- 这里的【线程持久化】和【操作系统线程】概念完全独立区分。操作系统线程是进程内的执行单元，是操作系统能够进行调度的最小单位。而线程级持久化表示聊天过程中的单次会话的持久化信息，用来隔离不同的聊天会话。
- 这里的【状态快照】并非是之前学习过的 State 的快照。这个状态包含了所有必要的上下文信息，比如：已经调用过哪些工具、用户的输入、聊天历史、下一步要执行的节点等等。
跨会话持久化：通过存储（Store）保存用户信息、偏好设置等长期数据，实现不同对话间信息的持久化共享。
- 对应上述场景一
- 例如，可以将用户基本情况（有高血压病史）存储，后续无论在何时何地，或无论新开几个会话窗口，都可以基于用户基本信息（有高血压病史）生成结果。

2. 线程级持久化

2.1 线程级持久化是怎么工作的？

当我们开始执行工作流，过程中可能会发生崩溃或重启导致的中断等异常情况。

根据 LangGraph 的持久化机制，线程级持久化表示能够自动保存工作流执行过程中的状态快照，维持单次会话的完整上下文。当工作流执行到某一步时，它会自动保存当前步骤的状态快照。这个状态包含了所有必要的上下文信息，比如：已经调用过哪些工具、用户的输入、聊天历史、下一步要执行的节点等等。

因此，工作流在执行过程中会发生如下流程：

正常执行：紫色/蓝色节点和实线箭头
异常恢复：红色/绿色节点和虚线箭头
检查点回滚：橙色节点和虚线箭头

初始状态
   ↓
工作流启动 - 初始化上下文：用户输入、加载当前线程内的历史状态
   ↓
执行步骤1 - 状态1：+工具调用记录1、中间结果1、下一步要执行的节点
   ↓
Persistence 自动保存状态1快照到存储 ←── 异常中断（崩溃/重启）→ 状态丢失风险
   ↓    ↓
执行步骤2 - 状态2：+工具调用记录2、    从持久化存储恢复状态1快照
   中间结果2、更新聊天历史、    ↓
   下一步要执行的节点    回滚到状态1
   ↓
Persistence 自动保存状态2快照到存储 ←── 异常中断（崩溃/重启）→ 状态丢失风险
   ↓    ↓
执行步骤3 - 状态3：+最终计算结果、完整聊天历史、从持久化存储恢复状态2快照
   下一步要执行的节点    ↓
   ↓    回滚到状态2
Persistence 自动保存状态3快照到存储
   ↓
工作流终止
   ↓
从持久化存储选择状态1/2快照回滚 ←── 需要回滚

线程级持久化机制确保了：

状态不丢失：即使应用程序崩溃、重启，或者一个长时间的流程被中断，当它恢复时，可以从上次停止的地方继续执行，而不是从头开始。
支持长时间运行的任务：对于需要与用户进行多轮交互（如多步对话助手）或处理耗时极长的流程（如等待外部 API 回调），持久化是必不可少的。
检查点和回滚：我们可以将状态保存到某个时间点（检查点），并在需要时回滚到该状态。

2.2 核心概念

LangGraph 的持久化机制——线程级持久化是其核心功能，它通过【线程】和【检查点】这两个核心部分来实现。具体如下：

2.2.1 Threads（线程）

在 LangGraph 中，Thread 代表一个独立的工作流执行会话。可以把它想象成【与某个用户的一次完整对话历史】或【处理某个特定任务的一次完整执行过程】。例如在 DS 中的一次完整对话：

Thread 的关键特性如下：

隔离性：每个 Thread 都是完全独立的，它们的状态互不干扰
持久化单元：Thread 是状态持久化的基本单位
标识符：通过唯一的 thread_id 来识别

2.2.2 Checkpoints（检查点）

Checkpoint 是 Thread 在特定时刻的状态快照。它记录了工作流执行到某个节点时的完整状态。例如在会话中，每一次用户输入和对话结束后，都可以保存一个最新的状态快照。

Checkpoint 的关键特性：

状态快照（StateSnapshot）：保存了工作流在某个时间点的完整状态。包含【状态值】、【下一步要执行的节点】、【与此检查点关联的配置】和【与此检查点关联的元数据】等信息。

StateSnapshot 结构如下：

StateSnapshot(
    # 当前状态值（如：对话消息列表）
    values={'messages': [用户消息, AI回复, 用户消息...]},

    # 接下来要执行的节点
    next=('generate_response',),

    # 配置信息
    config={'configurable': {'thread_id': '123', 'checkpoint_id': 'abc'}},

    # 元数据（步骤号、来源、写入信息等）
    metadata={'step': 2, 'source': 'loop', 'writes': {...}},

    # 父检查点（形成链表）
    parent_config={'configurable': {'thread_id': '123', 'checkpoint_id': 'def...'}},

    # 创建时间
    created_at=''
)

版本历史与可恢复点：一个 Thread 可以有多个 Checkpoints，形成执行历史，使得同一个会话的历史状态可以从任意 Checkpoint 追溯和访问。

Threads 与 Checkpoints 关系如下：

2.3 线程级持久化使用姿势

2.3.1 步骤一：配置 checkpointer 持久化存储

在定义图时，我们需要指定 checkpointer。LangGraph 支持多种 checkpointer 的定义方式：

方式1：内存存储（MemorySaver）

最简单的方式，状态保存在程序内存中。适用于开发和测试，程序重启后状态会丢失。

这里使用快速上手——案例2的代码，在编译图时，直接添加编译参数 checkpointer

from langgraph.checkpoint.memory import InMemorySaver

# 定义存储方式
checkpointer = InMemorySaver()

# 用 checkpointer 编译图
agent = agent_builder.compile(checkpointer=checkpointer)

方式2：使用 Postgres 存储库

LangGraph 提供了几个检查点存储实现，适用于生产环境或需要状态持久化的场景：

SQLite 存储（langgraph-checkpoint-sqlite）：使用 SQLite 数据库（SqliteSaver / AsyncSqliteSaver），适合实验和本地工作流程
Postgres 存储（langgraph-checkpoint-postgres）：使用 Postgres 数据库（PostgresSaver / AsyncPostgresSaver），适合在生产中使用

为什么选择 PostgreSQL 作为存储库？

在众多持久化存储方案中，PostgreSQL 作为关系型数据库的佼佼者（PostgreSQL 和 MySQL 一样，都是最流行的开源关系型数据库），具备以下显著优势，使其成为持久化的理想选择：

• LangGraph 原生支持： LangGraph 提供了 PostgresSaver ，简化了与 PostgreSQL 的集成过程。

• 数据结构化与一致性：关系型数据库天生适合存储结构化数据。Graph 的状态，尤其是消息历史、用户档案、工具使用记录等，都可以很好地映射到表格结构中，确保数据的一致性和完整性。

• 可靠性与持久性： PostgreSQL 提供了事务支持、ACID 特性、数据备份与恢复机制，确保数据的持久性和高可用性，即使系统崩溃也能保证数据不丢失。

• 强大的查询能力： SQL 语言提供了灵活且强大的数据查询能力，方便我们对历史行为、用户数据进行分析、统计和审计。结合 pgvector 等扩展，甚至可以直接在数据库中进行向量相似度搜索，实现更高级的知识管理。

• 可扩展性：通过读写分离、分区、集群等技术，PostgreSQL 可以支持大规模的并发访问和数据存储，满足 Agent 在生产环境中的性能需求。

• 成熟的生态系统：拥有庞大的社区支持、丰富的工具和成熟的运维经验，降低了开发和维护成本。

使用 Docker 快速安装并启动 postgres：

# 1. 拉取 PostgreSQL 镜像
# 2. 运行 PostgreSQL 容器
# -p 5432:5432: 将容器的5432端口映射到宿主机的5432端口
# -e POSTGRES_PASSWORD=bit: 设置PostgreSQL的postgres用户密码
# --name postgres-sql: 给容器命名
# -d: 后台运行
docker pull postgres:latest
docker run --name postgres-sql -e POSTGRES_PASSWORD=bit -p 5432:5432 -d
postgres

可以使用 Navicat 测试链接：注意默认连接 postgres 初始数据库，如果有其他数据库可以配置连接。

安装 langgraph-checkpoint-postgres 包：

pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres

设置 PostgreSQL 连接字符串 (URI)：

PostgreSQL URI 通常遵循以下格式：postgresql://<user>:<password>@<host>:<port>/<database_name>

使用 Postgres 存储库作为检查点：

使用 PostgresSaver.from_conn_string() 方法从连接字符串创建一个新的 PostgresSaver 实例。注意：第一次使用 Postgres 检查点时需要调用 checkpointer.setup()

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres:bit@192.168.100.233:5432/postgres"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 第一次使用 Postgres 检查点时需要调用 checkpointer.setup()
    checkpointer.setup()

    # 编译图
    agent = agent_builder.compile(checkpointer=checkpointer)

    # ...后续调用...

2.3.2 步骤二：使用 Thread 进行执行

当我们编译好图并准备运行时，我们需要通过一个 Thread ID 来标识这次执行：

如果 Thread ID 不存在：LangGraph 会创建一个新的 Thread，并从初始状态开始执行
如果 Thread ID 已存在：LangGraph 会从 Checkpointer 中加载该 Thread 的最后一次保存的状态，并从这个状态继续执行

第一次执行：这里依旧使用postgres存储，进行第一次执行：创建一个新的 Thread (thread_id="1")

from langchain.messages import HumanMessage

DB_URI = "postgresql://postgres:bit@192.168.100.233:5432/postgres"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 第一次使用 Postgres 检查点时需要调用 checkpointer.setup()
    checkpointer.setup()

    # 编译图
    agent = agent_builder.compile(checkpointer=checkpointer)

    # 第一次执行，创建一个新的 Thread (thread_id="1")
    config = {"configurable": {"thread_id": "1"}}
    result1 = agent.invoke(
        {"messages": [HumanMessage(content="今天西安的天气如何？")]},
        config
    )

    print(f"调用 LLM 总次数: {result1['llm_calls']}次")
    for m in result1["messages"]:
        m.pretty_print()

运行后，可以看到postgres库中，已经存储了检查点信息：

一段时间后，再次使用相同的 thread_id 调用：

from langchain.messages import HumanMessage

DB_URI = "postgresql://postgres:bit@192.168.100.233:5432/postgres"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 编译图
    agent = agent_builder.compile(checkpointer=checkpointer)

    # ............. 一段时间后，程序可能重启了（使用postgres存储） .............

    # 再次使用相同的 thread_id 调用
    # LangGraph 会从上次的状态继续，而不是重新开始
    # 此时，result2 的上下文会包含之前的对话历史
    config = {"configurable": {"thread_id": "1"}}
    result2 = agent.invoke(
        {"messages": [HumanMessage(content="我们刚才聊到哪了？")]},
        config
    )

    print(f"调用 LLM 总次数: {result2['llm_calls']}次")
    for m in result2["messages"]:
        m.pretty_print()

第二次调用结果：

调用 LLM 总次数: 3次
============================== Human Message ==============================
今天西安的天气如何?
============================== Ai Message ==============================
Tool Calls:
  tavily_search (call_8ALCCwF3xudzjavTFN16rA0m)
Call ID: call_8ALCCwF3xudzjavTFN16rA0m
  Args:
    query: 西安天气
============================== Tool Message ==============================
(省略...)
============================== Ai Message ==============================
今天西安的天气情况如下：
- **天气**: 晴
- **气温**: 最高温度约为 13°C，最低温度约为 2°C
- **风速**: 东北风，约 6 英里/小时
- **空气质量**: 不健康
============================== Human Message ==============================
我们刚才聊到哪了?
============================== Ai Message ==============================
我们刚才聊到西安的天气情况，包括今天的气温和天气状况。如果你有其他问题或者想讨论的内容，请告诉我！

从结果看来，result2 的上下文会包含之前的对话历史。LangGraph 会从上次的状态继续，而不是重新开始。

2.3.3 其他基本用法

2.3.3.1 获取状态快照

当使用 checkpointer 编译图时，执行时就会在每个步骤处保存状态快照。在 LangGraph 中状态快照就是 StateSnapshot 对象，其具有以下关键属性：

StateSnapshot(
    # 当前状态值（如：对话消息列表）
    values={'messages': [用户消息, AI回复, 用户消息...]},

    # 接下来要执行的节点
    next=('generate_response',),

    # 配置信息
    config={'configurable': {'thread_id': '123', 'checkpoint_id': 'abc'}},

    # 元数据（步骤号、来源、写入信息等）
    metadata={'step': 2, 'source': 'loop', 'writes': {...}},

    # 父检查点（形成链表）
    parent_config={'configurable': {'thread_id': '123', 'checkpoint_id': 'def...'}},

    # 创建时间
    created_at=''
)

我们可以使用 get_state(config) 方法，获取编译后的图的最新状态快照：

from langchain.messages import HumanMessage
config = {"configurable": {"thread_id": "1"}}

# 调用前的状态快照
snapshot = agent.get_state(config)
print(snapshot)

result1 = agent.invoke(
    {"messages": [HumanMessage(content="你好")]},
    config
)

# 调用后的状态快照
snapshot = agent.get_state(config)
print(snapshot)

打印结果：

StateSnapshot(
    values={},
    next=(),
    config={'configurable': {'thread_id': '1'}},
    metadata=None,
    created_at=None,
    parent_config=None,
    tasks=(),
    interrupts=()
)

StateSnapshot(
    values={
        'messages': [
            HumanMessage(
                content='你好',
                additional_kwargs={},
                response_metadata={}
            ),
            AIMessage(
                content='你好！有什么我可以帮助你的吗？',
                additional_kwargs={'refusal': None},
                response_metadata={
                    'token_usage': {
                        'completion_tokens': 11,
                        'prompt_tokens': 130,
                        'total_tokens': 141,
                        'completion_tokens_details': {
                            'accepted_prediction_tokens': None,
                            'audio_tokens': None,
                            'reasoning_tokens': None,
                            'rejected_prediction_tokens': None
                        },
                        'prompt_tokens_details': {
                            'audio_tokens': None,
                            'cached_tokens': 0
                        }
                    },
                    'model_provider': 'openai',
                    'model_name': 'gpt-4o-mini-2024-07-18',
                    'system_fingerprint': 'fp_efad92c60b',
                    'id': 'chatcmpl-CiGBdkNSGgkZ3ccpa6NEAyUHrzqls',
                    'finish_reason': 'stop',
                    'logprobs': None
                },
                id='lc_run--f76d30e4-0e06-4f18-ab44-0e43e8ff70b8-0',
                usage_metadata={
                    'input_tokens': 130,
                    'output_tokens': 11,
                    'total_tokens': 141,
                    'input_token_details': {'cache_read': 0},
                    'output_token_details': {}
                }
            )
        ],
        'llm_calls': 1
    },
    next=(),
    config={
        'configurable': {
            'thread_id': '1',
            'checkpoint_ns': '',
            'checkpoint_id': '1f0cf5a3-b6ff-6b2e-8001-dc5cd083ad82'
        }
    },
    metadata={'source': 'loop', 'step': 1, 'parents': {}},
    created_at='2025-12-02T08:38:09.968610+00:00',
    parent_config={
        'configurable': {
            'thread_id': '1',
            'checkpoint_ns': '',
            'checkpoint_id': '1f0cf5a3-a7a5-6cf8-8000-ec666fd226fe'
        }
    },
    tasks=(),
    interrupts=()
)

2.3.3.2 获取状态历史记录

我们可以通过调用 get_state_history(config) 来获取给定线程的图执行的完整历史记录。这将返回与配置中提供的线程 ID 关联的 StateSnapshot 对象列表。

from langchain.messages import HumanMessage
config = {"configurable": {"thread_id": "1"}}
result1 = agent.invoke(
    {"messages": [HumanMessage(content="你好")]},
    config
)

# 查看状态历史记录
history = list(agent.get_state_history(config))
print(history)

返回结果将按时间顺序排序，列表中的第一个检查点（StateSnapshot）是最新的。内容如下（原输出很长很乱，丢给ai叫他给我调整了缩进）

[
    StateSnapshot(
        values={
            'messages': [
                HumanMessage(
                    content='你好',
                    additional_kwargs={},
                    response_metadata={}
                ),
                AIMessage(
                    content='你好！有什么我可以帮助你的吗？',
                    additional_kwargs={'refusal': None},
                    response_metadata={
                        'token_usage': {
                            'completion_tokens': 11,
                            'prompt_tokens': 130,
                            'total_tokens': 141,
                            'completion_tokens_details': {
                                'accepted_prediction_tokens': None,
                                'audio_tokens': None,
                                'reasoning_tokens': None,
                                'rejected_prediction_tokens': None
                            },
                            'prompt_tokens_details': {
                                'audio_tokens': None,
                                'cached_tokens': 0
                            }
                        },
                        'model_provider': 'openai',
                        'model_name': 'gpt-4o-mini-2024-07-18',
                        'system_fingerprint': 'fp_efad92c60b',
                        'id': 'chatcmpl-CiGUUZCWPLKOuRv6sdzolSVAzwtMW',
                        'finish_reason': 'stop',
                        'logprobs': None
                    },
                    id='lc_run--7d1e5db6-e8b2-42c1-9fd8-ba21c42705ac-0',
                    usage_metadata={
                        'input_tokens': 130,
                        'output_tokens': 11,
                        'total_tokens': 141,
                        'input_token_details': {'cache_read': 0},
                        'output_token_details': {}
                    }
                )
            ],
            'llm_calls': 1
        },
        next=(),
        config={
            'configurable': {
                'thread_id': '1',
                'checkpoint_ns': '',
                'checkpoint_id': '1f0cf5cf-48e3-6bbb-8001-c351576393a6'
            }
        },
        metadata={'source': 'loop', 'step': 1, 'parents': {}},
        created_at='2025-12-02T08:57:39.538826+00:00',
        parent_config={
            'configurable': {
                'thread_id': '1',
                'checkpoint_ns': '',
                'checkpoint_id': '1f0cf5cf-38d7-61a8-8000-2dacafb54f11'
            }
        },
        tasks=(),
        interrupts=()
    ),
    StateSnapshot(
        values={
            'messages': [
                HumanMessage(
                    content='你好',
                    additional_kwargs={},
                    response_metadata={}
                )
            ]
        },
        next=('llm_call',),
        config={
            'configurable': {
                'thread_id': '1',
                'checkpoint_ns': '',
                'checkpoint_id': '1f0cf5cf-38d7-61a8-8000-2dacafb54f11'
            }
        },
        metadata={'source': 'loop', 'step': 0, 'parents': {}},
        created_at='2025-12-02T08:57:37.855930+00:00',
        parent_config={
            'configurable': {
                'thread_id': '1',
                'checkpoint_ns': '',
                'checkpoint_id': '1f0cf5cf-38d4-6e0c-bfff-6ce6b2a2dd01'
            }
        },
        tasks=(
            PregelTask(
                id='44c184e5-8661-3f3c-eeee-7a2f569f7e88',
                name='llm_call',
                path=('__pregel_pull', 'llm_call'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'messages': [
                        AIMessage(
                            content='你好！有什么我可以帮助你的吗？',
                            additional_kwargs={'refusal': None},
                            response_metadata={
                                'token_usage': {
                                    'completion_tokens': 11,
                                    'prompt_tokens': 130,
                                    'total_tokens': 141,
                                    'completion_tokens_details': {
                                        'accepted_prediction_tokens': None,
                                        'audio_tokens': None,
                                        'reasoning_tokens': None,
                                        'rejected_prediction_tokens': None
                                    },
                                    'prompt_tokens_details': {
                                        'audio_tokens': None,
                                        'cached_tokens': 0
                                    }
                                },
                                'model_provider': 'openai',
                                'model_name': 'gpt-4o-mini-2024-07-18',
                                'system_fingerprint': 'fp_efad92c60b',
                                'id': 'chatcmpl-CiGUUZCWPLKOuRv6sdzolSVAzwtMW',
                                'finish_reason': 'stop',
                                'logprobs': None
                            },
                            id='lc_run--7d1e5db6-e8b2-42c1-9fd8-ba21c42705ac-0',
                            usage_metadata={
                                'input_tokens': 130,
                                'output_tokens': 11,
                                'total_tokens': 141,
                                'input_token_details': {'cache_read': 0},
                                'output_token_details': {}
                            }
                        )
                    ],
                    'llm_calls': 1
                }
            ),
        ),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': []},
        next=('__start__',),
        config={
            'configurable': {
                'thread_id': '1',
                'checkpoint_ns': '',
                'checkpoint_id': '1f0cf5cf-38d4-6e0c-bfff-6ce6b2a2dd01'
            }
        },
        metadata={'source': 'input', 'step': -1, 'parents': {}},
        created_at='2025-12-02T08:57:37.855022+00:00',
        parent_config=None,
        tasks=(
            PregelTask(
                id='56c07f44-a6a6-a8b8-899a-d5c24b2ff671',
                name='__start__',
                path=('__pregel_pull', '__start__'),
                error=None,
                interrupts=(),
                state=None,
                result={
                    'messages': [
                        HumanMessage(
                            content='你好',
                            additional_kwargs={},
                            response_metadata={}
                        )
                    ]
                }
            ),
        ),
        interrupts=()
    )
]

2.3.3.3 重放

如果我们用一个 thread_id 和一个 checkpoint_id（表示检查点标识符，用于指代线程内的特定检查点）来调用一个图，那么我们将重新执行对应于 checkpoint_id 之后的步骤。

步骤：

先执行一次完整的流程，获取一次完整历史记录
保存中间过程某一次快照，并重新执行快照后的步骤
获取第二次调用后的完整历史记录，验证是否重放成功

from langchain.messages import HumanMessage
config = {"configurable": {"thread_id": "1"}}

# 第一次执行
result1 = agent.invoke(
    {"messages": [HumanMessage(content="今天西安的天气如何？")]},
    config
)

# 保存调用工具前的状态
print("-" * 80)
print(f"第一次执行历史：")
to_replay = None
for state in agent.get_state_history(config):
    print("checkpoint_id: ", state.config["configurable"]["checkpoint_id"],
          "消息数: ", len(state.values["messages"]),
          "下一节点: ", state.next)
    if len(state.values["messages"]) == 2:  # 保存调用工具前的状态
        to_replay = state

print("-" * 80)
print(f"从{to_replay.next}节点开始重新执行，重放配置: {to_replay.config}")

# 第二次执行：重放
result2 = agent.invoke(None, config=to_replay.config)
print("-" * 80)
print(f"第二次执行历史：重放后")
# 查看新的历史记录
for state in agent.get_state_history(config):
    print("checkpoint_id: ", state.config["configurable"]["checkpoint_id"],
          "消息数: ", len(state.values["messages"]),
          "下一节点: ", state.next)

result2['messages'][-1].pretty_print()

执行结果如下所示。可以看到历史记录中帮我们记录所有的执行过程，包括重放前和重放后的步骤！这同时也证明了重放成功。

执行结果：

第一次执行历史：
checkpoint_id:  1f0cf6bc-cb09-61e3-8003-1a74660e9324 消息数： 4 下一节点： ()
checkpoint_id:  1f0cf6bc-b148-624e-8002-f696045e95a8 消息数： 3 下一节点： ('llm_call',)
checkpoint_id:  1f0cf6bc-8a86-62a5-8001-38aa0c3d8d80 消息数： 2 下一节点： ('tool_node',)
checkpoint_id:  1f0cf6bc-7abe-6d9a-8000-ef816c492e36 消息数： 1 下一节点： ('llm_call',)
checkpoint_id:  1f0cf6bc-7abc-6ce5-bfff-5241076b5574 消息数： 0 下一节点： ('__start__',)

从('tool_node',)节点开始重新执行，重放配置: {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f0cf6bc-8a86-62a5-8001-38aa0c3d8d80'}}

第二次执行历史：重放后
checkpoint_id:  1f0cf6bd-0e30-6cf9-8003-f22545cde7b2 消息数： 4 下一节点： ()
checkpoint_id:  1f0cf6bc-e376-6e52-8002-9c66bdb5577a 消息数： 3 下一节点： ('llm_call',)
checkpoint_id:  1f0cf6bc-cb09-61e3-8003-1a74660e9324 消息数： 4 下一节点： ()
...（后续历史记录）

今天西安的天气情况如下：
- **天气**: 晴
- **气温**: 最高温度约为 13°C，最低温度约为 2°C
- **风速**: 东北风，约 6 英里/小时
- **空气质量**: 不健康

重放功能实际应用为时间旅行。

2.3.3.4 更新状态

我们还可以编辑图状态。我们使用 update_state() 方法来做到这一点。

让我们更新用户的输入，换成其他搜索内容：

先执行一次完整的流程，获取一次完整历史记录
保存第一次调用LLM前的步骤快照，修改用户输入来更新快照，并重新执行更新后快照步骤

from langchain.messages import HumanMessage
from langgraph.types import Overwrite

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

# 第一次执行
result1 = agent.invoke(
    {"messages": [HumanMessage(content="今天西安的天气如何？")]},
    config
)

# 找到调用LLM前的步骤
print("-" * 80)
print(f"第一次执行历史：")
selected_state = None
for state in agent.get_state_history(config):
    print("checkpoint_id: ", state.config["configurable"]["checkpoint_id"],
          "消息数: ", len(state.values["messages"]),
          "下一节点: ", state.next)
    if len(state.values["messages"]) == 1:  # 此时消息数为1；下一节点是'llm_call'
        selected_state = state

print("-" * 80)
print(f"更新前配置: {selected_state.config}")

# 根据指定的config，更新对于步骤的值
# 更新用户输入
new_config = agent.update_state(
    selected_state.config,
    {"messages": Overwrite([HumanMessage(content="今天北京的天气如何？")])}  # 清空消息，重新写入
)
print("-" * 80)
print(f"更新后配置: {new_config}")

# 第二次执行：重放更新后的配置
result2 = agent.invoke(None, config=new_config)
for message in result2['messages']:
    message.pretty_print()

执行结果：

第一次执行历史：
checkpoint_id:  1f0cf6c8-743c-6b8d-8003-d8a1130c15e0 消息数： 4 下一节点： ()
checkpoint_id:  1f0cf6c8-4244-682c-8002-bdc8b86d460d 消息数： 3 下一节点： ('llm_call',)
checkpoint_id:  1f0cf6c8-3304-6782-8001-7ebc967b7bb1 消息数： 2 下一节点： ('tool_node',)
checkpoint_id:  1f0cf6c8-1b1c-62a0-8000-42b9d34bb01d 消息数： 1 下一节点： ('llm_call',)
checkpoint_id:  1f0cf6c8-1b19-6adb-bfff-915f44142490 消息数： 0 下一节点： ('__start__',)

更新前配置: {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f0cf6c8-1b1c-62a0-8000-42b9d34bb01d'}}

更新后配置: {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f0cf6c8-7444-61e3-8001-d9eb91c95052'}}
============================== Human Message ==============================
今天北京的天气如何?
============================== Ai Message ==============================
Tool Calls:
  tavily_search (call_FfUFDwThpNTBHoiJgvNmwtMr)
Call ID: call_FfUFDwThpNTBHoiJgvNmwtMr
  Args:
    query: 北京天气
============================== Tool Message ==============================
(省略...)
============================== Ai Message ==============================
今天北京的天气情况如下：
- 当前气温：-4°C
- 湿度：23%
- 风速：25.2 km/h
- 天气状况：晴朗
- 最高气温：预计为16°C

3. 跨会话持久化

3.1 Checkpoint 的局限性

3.1.1 问题场景：跨会话信息丢失

LangGraph 的 Checkpoint 机制提供了强大的短期记忆能力，它能够：

自动保存工作流每个步骤的状态快照
维持单次对话的完整上下文
隔离不同线程（Thread）的执行状态

简单示例（仅调用了下 LLM）：

import operator
from typing import TypedDict, Annotated
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START, END
from langgraph.graph import StateGraph

# 定义状态
class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

# 定义模型节点
model = init_chat_model("gpt-4o-mini", temperature=0)
def llm_call(state: dict):
    """LLM调用"""
    return {
        "messages": [
            model.invoke([SystemMessage(content="你是一个乐于助人的助手。")]
                         + state["messages"])
        ]
    }

# 构件图
builder = StateGraph(MessagesState)
builder.add_node("llm_call", llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
graph = builder.compile(checkpointer=InMemorySaver())

检查点可以完美处理会话内记忆：

config1 = {"configurable": {"thread_id": "1"}}
# 第一次对话
result1 = graph.invoke({"messages": [HumanMessage(content="我爱吃汉堡，推荐一家餐厅")]}, config1)

# AI记得：你爱吃汉堡
result2 = graph.invoke({"messages": [HumanMessage(content="我爱吃什么？")]}, config1)
result2["messages"][-1].pretty_print()
# 输出：你提到你爱吃汉堡，所以可以推测你喜欢美味的快餐和丰富的口味组合...

想象一个多会话的 AI 助手场景：

• 星期一，用户首次对话

• 星期二，用户开启一个新对话

问题出现：跨会话信息丢失！

# 星期一，用户首次对话
config1 = {"configurable": {"thread_id": "day_1"}}
result1 = graph.invoke({"messages": [HumanMessage(content="我爱吃汉堡，推荐一家餐厅")]}, config1)

# 星期二，用户开启一个新对话
config2 = {"configurable": {"thread_id": "day_2"}}
result2 = graph.invoke({"messages": [HumanMessage(content="我爱吃什么？")]}, config2)
result2["messages"][-1].pretty_print()
# 输出：我不知道你具体喜欢吃什么，但可以根据一些常见的食物类型来猜测...

问题：AI不记得用户喜欢汉堡！每次对话都要"重新认识"。

3.1.2 现实世界的需求：从"单次对话"到"终身服务"

例如一个智能客服系统，具有以下实际业务需求：

识别 VIP 客户，优先服务
避免重复询问相同问题
基于历史投诉优化服务

仅是检查点无法满足这些需求：

graph = builder.compile(checkpointer=InMemorySaver())

config1 = {"configurable": {"thread_id": "query_1"}}
result1 = graph.invoke({"messages": [HumanMessage(content="我的账户被冻结了")]}, config1)
# 用户第一次投诉账户问题，已解决。智能客服了如下过程：
# - 搜集用户信息
# - 了解用户问题与需求
# - 处理问题

config2 = {"configurable": {"thread_id": "query_2"}}
result2 = graph.invoke({"messages": [HumanMessage(content="我的账户又被冻结了")]}, config2)
# 10天后，用户第二次投诉账户问题。
# 由于智能客服不知道用户历史，无法准确解决问题。还需再次了解前因后果

如何做到【共享状态】的需求模型，如精准识别客户、保留VIP客户关键历史记录，是智能客服系统的关键。

3.2 解决方案：引入 Store

Store 像是一个长期记忆仓库，支持在我们执行过程中保存用户信息、偏好设置等长期数据，以实现不同对话间信息的持久化共享。

3.2.1 存储 vs 检查点

检查点：保存状态变化历史（时间线）
存储：保存结构化知识（数据库）

实际上，使用 Checkpoint + Store 模式才能够真正实现理想的 AI 记忆系统。

3.2.2 引入 Store 后，AI 应用架构的范式转变

Store 的引入，不是简单的功能增加，而是 AI 应用架构的范式转变：

阶段	特点	示例
第1阶段：无状态AI（石器时代）	每次对话都是陌生人	"你好！""你好！"（无限循环）
第2阶段：检查点（工业革命）	单次对话有记忆	"我叫小明""你好小明！"但明天就忘记
第3阶段：检查点+Store（智能时代）	终身记忆，持续学习	"一年前你喜欢披萨，现在还喜欢吗？""根据你的阅读历史，推荐这本书..."

Store 的引入，真正能做到：

从关注单次交互 → 到关注用户生命周期
从处理当前请求 → 到利用历史数据
从通用回复 → 到深度个性化

这种转变让 AI 从"工具"进化为"伙伴"，真正实现智能服务的核心理念：在正确的时间，以正确的方式，为正确的人提供正确的价值。

题外话：使用 AI 不要随意泄露自己的隐私信息，很容易被保存下来！

3.3 跨会话持久化使用姿势

要想使用 store，我们需要创建一个存储实例，其也有【内存级存储】与相关【存储库存储】两种方式。例如内存级存储：

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()

接着只需像以前一样使用 Checkpoints 和 Store 变量编译图表即可：

graph = builder.compile(checkpointer=checkpointer, store=store)

3.3.1 方式1：内存存储

Store 基本用法

Store 本身是通过 Namespace 区分不同数据，如下所示：

在 LangGraph 中，其提供了一个简单的内存实现 InMemoryStore 。想要进行存储，需要：

• 先定义命名空间：为了区分不同用户的记忆，需要一个“命名空间”。这就像在数据库里为每个用户创建一个独立的文件夹。命名空间用于组织记忆，通常按业务逻辑划分。一般用元组来定义命名空间。如：

# 使用元组 - 层次清晰，易于扩展
namespace1 = ("user_123", "preferences", "food") # 用户食物偏好
namespace2 = ("user_123", "preferences", "music") # 用户音乐偏好
namespace3 = ("user_123", "conversations", "2025-05") # 用户某天的对话历史
# 使用字符串 - 扁平且易混淆
namespace4 = "user_123_preferences_food" # 需要解析，容易出错
namespace5 = "user_123_preferences_music"
namespace6 = "user_123_conversations_2024"

• 当在对话中获取到用户的重要信息时，使用 store.put() 方法将内存保存到存储中的命名空间。该方法参数包含：

◦ namespace ：决定这个记忆属于谁以及是什么类型。

◦ memory_id ：是这个记忆条目的唯一键。

◦ memory_content ：是记忆的具体内容，一个字典。

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
store.put(namespace, memory_id, memory_content)

完整代码

# 1. 导入并创建存储
from langgraph.store.memory import InMemoryStore
import uuid # 用于生成唯一ID

store = InMemoryStore()

# 2. 定义命名空间 (Namespace)
# 命名空间用于组织记忆，通常按业务逻辑划分，例如按用户。
# 这里我们用一个元组 (用户ID, 记忆类型)
user_id = "user_123"
namespace = (user_id, "preferences") # 用户 user_123 的偏好记忆

# 3. 存入一条记忆 (Memory)
# 每条记忆需要一个唯一的 memory_id 和 一个 value (通常是字典)
memory_id = str(uuid.uuid4()) # 生成唯一ID，如 "abc-123-def-456"
memory_value = {"favorite_food": "汉堡", "allergy": "花粉"}
store.put(namespace, memory_id, memory_value)
print("记忆已存入！")

# 4. 读取记忆
# 可以搜索某个命名空间下的所有记忆
all_memories = store.search(namespace)
for mem in all_memories:
print(mem.dict()) # 记忆对象转成字典查看

运行

{
    'namespace': [
        'user_123', 'preferences'
    ],
    'key': 'db826e33-c68c-4669-a79a-3579bff02ff1',
    'value': {
        'favorite_food': '汉堡',
        'allergy': '花粉'
    },
    'created_at': '2025-12-03T08:16:14.134568+00:00',
    'updated_at': '2025-12-03T08:16:14.134576+00:00',
    'score': None
}

因使用元组作为命名空间，故同样支持下面的搜索方式：

all_memories = store.search((user_id, ))

在 LangGraph 中使用 Store

内存存储适用于开发和测试，程序重启后存储的数据会丢失。这里依旧使用快速上手——案例2 的代码进行演示。

由于要加入 Store，需要在合适的地方加入与存储关键信息相关的代码。如我们可以在每次调用 LLM 前先进行信息收集，然后带着收集到的共享信息进行 LLM 调用。因此，流程变成了：

这样，两部分信息将会被收集：一是用户发的消息；二是通过工具调用返回的结果信息也会被采集。代码如下：

• 在编译图时，直接添加编译参数 store ，如下所示：

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# 用 checkpointer + store 编译图
agent = agent_builder.compile(checkpointer=checkpointer, store=store)

现在，在任何一个节点的函数中，都可以通过注入 store 参数来访问这个全局存储。

• 新增提取用户信息节点在这个节点中，我们需要根据【用户发的消息】和【工具调用返回的结果】来采集需要收集的信息。收集的信息需要使用 Store 进行存储。关键设计如下：

1. 任何节点函数，如果需要访问 Store，可以通过在参数中声明 store: BaseStore 和 config: RunnableConfig 来获取。

2. 在这里可以通过 LLM 提取用户信息，因此定义结构化返回是很有必要的

# 定义结构化输出
class Person(BaseModel):
    """一个人的信息。"""
    # 注意:
    # 1. 每个字段都是 Optional “可选的” —— 允许 LLM 在不知道答案时输出 None。
    # 2. 每个字段都有一个 description “描述” —— LLM使用这个描述。
    name: Optional[str] = Field(default=None, description="这个人的名字")
    height_in_meters: Optional[str] = Field(default=None,
                                             description="以米为单位的高度")
    favourite_food: Optional[list[str]] = Field(default=None,
                                                 description="最喜欢的食物列表")

model_with_structured = model.with_structured_output(Person)

# 提取用户信息节点
def get_person_by_llm(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    """通过 LLM 提取用户信息"""
    # 1. 先提取
    people_info = model_with_structured.invoke(
        [
            SystemMessage(
                content="你是一个提取信息的专家，只从文本中提取我的相关信息，不能提取别人的信息。如果你不知道要提取的属性的值，属性值返回null。"
            )
        ]
        + state["messages"][-3:]  # 只查看最近3条消息
    )

    # 2. 再保存
    user_id = config["configurable"]["user_id"]

    # 保存用户基本信息
    namespace1 = (user_id, "info")
    # 每次put前应判断是否存在，再更新。否则会有多条记录被记录。这里简写
    store.put(
        namespace1,
        str(uuid.uuid4()),
        {
            "name": people_info.name,
            "height": people_info.height_in_meters
        }
    )

    # 保存用户偏好
    namespace2 = (user_id, "preferences")
    store.put(
        namespace2,
        str(uuid.uuid4()),
        {"favourite_food": people_info.favourite_food}  # 省略追加逻辑：先搜再更新
    )

    return {
        "llm_calls": state.get('llm_calls', 0) + 1
    }

• 更新模型调用节点：添加共享信息到提示词调用 LLM 之前，我们便可以通过查询 Store 获取共享信息，然后将其加入到提示词中，完成调用。

def llm_call(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    """LLM决定是否调用工具"""
    # 搜索用户信息
    user_id = config["configurable"]["user_id"]
    namespace1 = (user_id, "info")
    namespace2 = (user_id, "preferences")
    info_result = store.search(namespace1)
    pref_result = store.search(namespace2)

    return {
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content=(
                            f"你是一个乐于助人的助手，支持调用工具进行搜索。"
                            f"查询 LLM 前可参考以下信息："
                            f"1. 用户基本情况：{info_result[0].value} "
                            f"2. 用户偏好情况：{pref_result[0].value}"
                        )
                    )
                ]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

• 构件图时，加入新节点与调整边

根据下图完成调整：

agent_builder = StateGraph(MessagesState)
agent_builder.add_node(llm_call)
agent_builder.add_node(tool_node)
# 新增节点
agent_builder.add_node(get_person_by_llm)

# 调整边
agent_builder.add_edge(START, "get_person_by_llm")
agent_builder.add_edge("get_person_by_llm", "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tool_node", "get_person_by_llm")

到此，代码已经改造完成，完整代码如下：

import uuid
from typing import Optional
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain_core.runnables import RunnableConfig
from langchain_tavily import TavilySearch
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from pydantic import BaseModel, Field

# 步骤 1: 定义工具和模型
search = TavilySearch(max_results=4)
tools = [search]
# 绑定工具
model = init_chat_model("gpt-4o-mini", temperature=0)
model_with_tools = model.bind_tools(tools)

# 步骤 2: 定义状态
from langchain.messages import AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

class MessagesState(TypedDict):
    # 类型: list[AnyMessage] - 任意消息对象的列表
    # 合并策略: operator.add - 使用加法操作符进行状态合并
    # 效果: 当状态更新时，新的消息会追加到现有列表中，而不是替换
    messages: Annotated[list[AnyMessage], operator.add]
    # 类型: int - 整数值
    # 用途: 跟踪LLM（大语言模型）的调用次数
    llm_calls: int

# 步骤 3：新增提取信息节点
# 定义结构化输出
class Person(BaseModel):
    """一个人的信息。"""
    # 注意:
    # 1. 每个字段都是 Optional “可选的” —— 允许 LLM 在不知道答案时输出 None。
    # 2. 每个字段都有一个 description “描述” —— LLM使用这个描述。
    name: Optional[str] = Field(default=None, description="这个人的名字")
    height_in_meters: Optional[str] = Field(default=None, description="以米为单位的高度")
    favourite_food: Optional[list[str]] = Field(default=None, description="最喜欢的食物列表")

model_with_structured = model.with_structured_output(Person)

def get_person_by_llm(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    """通过 LLM 提取用户信息"""
    # 1. 先提取
    people_info = model_with_structured.invoke(
        [
            SystemMessage(
                content="你是一个提取信息的专家，只从文本中提取我的相关信息，不能提取别人的信息。如果你不知道要提取的属性的值，属性值返回null。"
            )
        ]
        + state["messages"][-3:]  # 只查看最近3条消息
    )

    # 2. 再保存
    user_id = config["configurable"]["user_id"]

    # 保存用户基本信息
    namespace1 = (user_id, "info")
    # 每次put前应判断是否存在，再更新。否则会有多条记录被记录。这里简写
    store.put(
        namespace1,
        str(uuid.uuid4()),
        {
            "name": people_info.name,
            "height": people_info.height_in_meters
        }
    )

    # 保存用户偏好
    namespace2 = (user_id, "preferences")
    store.put(
        namespace2,
        str(uuid.uuid4()),
        {"favourite_food": people_info.favourite_food}  # 省略追加逻辑：先搜再更新
    )

    return {
        "llm_calls": state.get('llm_calls', 0) + 1
    }

# 步骤 4: 更新模型调用节点：添加共享用户信息到提示词
def llm_call(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    """LLM决定是否调用工具"""
    # 搜索用户信息
    user_id = config["configurable"]["user_id"]
    namespace1 = (user_id, "info")
    namespace2 = (user_id, "preferences")
    info_result = store.search(namespace1)
    pref_result = store.search(namespace2)

    return {
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content=(
                            f"你是一个乐于助人的助手，支持调用工具进行搜索。"
                            f"查询 LLM 前可参考以下信息："
                            f"1. 用户基本情况：{info_result[0].value} "
                            f"2. 用户偏好情况：{pref_result[0].value}"
                        )
                    )
                ]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

# 步骤 5: 定义工具节点
tools_by_name = {tool.name: tool for tool in tools}

def tool_node(state: dict):
    """执行工具调用"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

# 步骤 6: 构件图
from langgraph.graph import StateGraph, START, END

# 定义结束逻辑
def should_continue(state: MessagesState):
    """根据LLM是否调用工具来决定是应该继续循环（路由到工具节点）还是停止循环（END）"""
    messages = state["messages"]
    last_message = messages[-1]
    # 如果LLM调用工具，则执行操作
    if last_message.tool_calls:
        return "tool_node"
    return END

# 加入新节点并修改边
agent_builder = StateGraph(MessagesState)
agent_builder.add_node(llm_call)
agent_builder.add_node(tool_node)
agent_builder.add_node(get_person_by_llm)
agent_builder.add_edge(START, "get_person_by_llm")
agent_builder.add_edge("get_person_by_llm", "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tool_node", "get_person_by_llm")

checkpointer = InMemorySaver()
store = InMemoryStore()

# 编译图
agent = agent_builder.compile(checkpointer=checkpointer, store=store)

• 运行与验证：同一用户但不同会话的请求

# 第一次聊天
config1 = {"configurable": {"thread_id": "1", "user_id": "1"}}
result1 = agent.invoke(
    {"messages": [HumanMessage(content="我叫李华，我最爱吃汉堡。我的朋友叫小明，他爱吃披萨")]},
    config1
)
print(f"\n调用 LLM 总次数：{result1['llm_calls']}次")
for m in result1["messages"]:
    m.pretty_print()

# ---------- 过了几天 ---------------
# 同一个人，再次进行对话
config2 = {"configurable": {"thread_id": "2", "user_id": "1"}}
result2 = agent.invoke(
    {"messages": [HumanMessage(content="给我推荐下餐厅")]},
    config2
)
print(f"\n调用 LLM 总次数：{result2['llm_calls']}次")
for m in result2["messages"]:
    m.pretty_print()

执行结果：

调用 LLM 总次数：2次

================================ Human Message =================================
我叫李华，我最爱吃汉堡。我的朋友叫小明，他爱吃披萨

================================== Ai Message ==================================
你好，李华！很高兴认识你。汉堡和披萨都是很受欢迎的美食。你和小明有没有一起去过什么好吃的地方呢？或者你们有没有想尝试的新餐厅？

调用 LLM 总次数：4次

================================ Human Message =================================
给我推荐下餐厅

================================== Ai Message ==================================
Tool Calls:
  tavily_search (call_btsV05aeqldqjaZNvgdlYRe9)
    Call ID: call_btsV05aeqldqjaZNvgdlYRe9
    Args:
      query: 推荐汉堡餐厅

================================= Tool Message =================================
{省略.....}

================================== Ai Message ==================================
以下是一些推荐的汉堡餐厅：
1. **[Burger She Wrote](https://www.novacircle.com/zh-CN/spots/north-america/united-states/california/los-angeles-county/los-angeles/burger-she-wrote-9ea267)** - 位于洛杉矶，这是一家小而温馨的餐厅，以其美味的和牛汉堡而闻名。
2. **[Tripadvisor 上洛杉矶的最佳汉堡](https://cn.tripadvisor.com/Restaurants-g32655-zfd10907-zfn7231034-Los_Angeles_California-Hamburger.html)** - 包含多家受欢迎的汉堡餐厅，如Bottega Louie和Eggslut，后者以其鸡蛋汉堡而著称。

希望这些推荐能帮助你找到美味的汉堡！

语义搜索 Store 的强大之处在于它支持语义搜索，而不仅仅是精确匹配。这意味着我们可以用自然语言问题来查找相关记忆。首先，我们需要配置带嵌入模型的 Store，如下所示：

3.3.2 方式2：Postgres 存储库

Postgres 存储库适用于生产环境或需要状态持久化的场景。由于之前已经启动过 PostgresSQL，这里可以直接连接到数据库，作为 PostgresStore 使用。只需在编译时设置 store 即可。

修改【内存存储】部分的代码：将内存存储方式修改为 Postgres 存储库。

注意：第一次使用 Postgres store 时需要调用 store.setup()

DB_URI = "postgresql://postgres:bit@192.168.100.233:5432/postgres"

with (
    PostgresSaver.from_conn_string(DB_URI) as checkpointer,
    PostgresStore.from_conn_string(DB_URI) as store,
):
    # 第一次使用 Postgres 检查点时需要调用 checkpointer.setup()
    checkpointer.setup()
    # 第一次使用 Postgres store 时需要调用 store.setup()
    store.setup()

    # 编译图
    agent = agent_builder.compile(checkpointer=checkpointer, store=store)
    # ...后续调用...

模拟第一次聊天：

DB_URI = "postgresql://postgres:bit@192.168.100.233:5432/postgres"

with (
    PostgresSaver.from_conn_string(DB_URI) as checkpointer,
    PostgresStore.from_conn_string(DB_URI) as store,
):
    # 第一次使用 Postgres 检查点时需要调用 checkpointer.setup()
    checkpointer.setup()
    # 第一次使用 Postgres store 时需要调用 store.setup()
    store.setup()

    # 编译图
    agent = agent_builder.compile(checkpointer=checkpointer, store=store)

    # 第一次聊天
    config1 = {"configurable": {"thread_id": "1", "user_id": "1"}}
    result1 = agent.invoke(
        {"messages": [HumanMessage(content="我叫李华，我最爱吃汉堡。我的朋友叫小明，他爱吃披萨")]},
        config1
    )
    print(f"\n调用 LLM 总次数：{result1['llm_calls']}次")
    for m in result1["messages"]:
        m.pretty_print()

运行系统后可以看到，postgres 中新增 store 相关表，其中存放了用户基本的信息：

再次验证：同一用户但不同会话的请求

DB_URI = "postgresql://postgres:bit@192.168.100.233:5432/postgres"

with (
    PostgresSaver.from_conn_string(DB_URI) as checkpointer,
    PostgresStore.from_conn_string(DB_URI) as store,
):
    # 第一次使用 Postgres 检查点时需要调用 checkpointer.setup()
    # checkpointer.setup()
    # 第一次使用 Postgres store 时需要调用 store.setup()
    # store.setup()

    # 编译图
    agent = agent_builder.compile(checkpointer=checkpointer, store=store)

    # ---------- 过了几天 ---------------
    # 同一个人，再次进行对话
    config2 = {"configurable": {"thread_id": "2", "user_id": "1"}}
    result2 = agent.invoke(
        {"messages": [HumanMessage(content="给我推荐下餐厅")]},
        config2
    )
    print(f"\n调用 LLM 总次数：{result2['llm_calls']}次")
    for m in result2["messages"]:
        m.pretty_print()

注意执行前，将以下代码注掉，因为：在存入store前，并没有编写不存在存入，存在更新的代码逻辑（只是演示），因此会将空的用户信息误存，导致LLM调用前查出来空的。

def get_person_by_llm(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    """通过 LLM 提取用户信息（注：保存逻辑已注释，避免误存空信息）"""
    # 1. 提取信息
    people_info = model_with_structured.invoke(
        [
            SystemMessage(
                content="你是一个提取信息的专家，只从文本中提取我的相关信息，不能提取别人的信息。如果你不知道要提取的属性的值，属性值返回null。"
            )
        ]
        + state["messages"][-3:]  # 只查看最近3条消息
    )

    user_id = config["configurable"]["user_id"]

    # ========== 保存逻辑暂时注释，避免误存空值 ==========
    # 原因：当前没有实现“不存在则存入、存在则更新”的逻辑，若 LLM 提取出的字段全为 None，
    # 会覆盖掉已有的正确用户信息，导致后续查询到空数据。

    # 保存用户基本信息
    # namespace1 = (user_id, "info")
    # store.put(
    #     namespace1,
    #     str(uuid.uuid4()),
    #     {
    #         "name": people_info.name,
    #         "height": people_info.height_in_meters
    #     }
    # )

    # 保存用户偏好
    # namespace2 = (user_id, "preferences")
    # store.put(
    #     namespace2,
    #     str(uuid.uuid4()),
    #     {"favourite_food": people_info.favourite_food}
    # )
    # =================================================

    return {
        "llm_calls": state.get('llm_calls', 0) + 1
    }

最终执行结果如下：（这里同样也是丢给ai缩进了一下格式）

调用 LLM 总次数：3次

================================ Human Message =================================
给我推荐下餐厅

================================== Ai Message ==================================
Tool Calls:
  tavily_search (call_PkIzsQRuCS6KZM15g3noHgBU)
    Call ID: call_PkIzsQRuCS6KZM15g3noHgBU
    Args:
      query: 推荐汉堡餐厅

================================= Tool Message =================================
{
  'query': '推荐汉堡餐厅',
  'follow_up_questions': None,
  'answer': None,
  'images': [],
  'results': [
    {
      'url': 'https://www.reddit.com/r/AskNYC/comments/1470o9z/best_burger_spot_in_nyc/?tl=zh-hans',
      'title': '纽约最好吃的汉堡店是哪家？ : r/AskNYC',
      'content': 'The Thompson 的Burger Joint 和Minetta Tavern 等被推荐。汉堡餐厅推荐 ，来自1 个月前。Smashed 和Korzo 等被推荐。纽约/布鲁克林最好的汉堡？ 以及 ...Read more',
      'score': 0.7336813,
      'raw_content': None
    },
    {
      'url': 'https://www.cosmopolitan.com/tw/lifestyle/food-and-drink/g44382807/hamburger-20230629/',
      'title': '美國旅遊必吃7大人氣漢堡店！IN-N-OUT最強勁敵',
      'content': '1.The Habit Burger Grill · 2.Cheeseburger in Paradise · 3.Five Guys · 4.IN-N-OUT · 5.Shake Shack · 6.SmashburgerRead more',
      'score': 0.61972505,
      'raw_content': None
    },
    {
      'url': 'https://cn.tripadvisor.com/Restaurants-g60763-zfd10907-zfn7102345-New_York_City_New_York-Hamburger.html',
      'title': '纽约市最佳汉堡',
      'content': "Ellen's Stardust Diner · (23,569 条点评). 美式烹饪, 晚餐 ; Virgil's Real BBQ - NYC · (4,968 条点评). 美式烹饪, 烧烤 ; 1. S'MAC · (488 条点评). 快餐小吃, 美式烹饪 ; 2.Read more",
      'score': 0.56719416,
      'raw_content': None
    },
    {
      'url': 'https://mliesl.edu/contents/ch/%E7%BE%8E%E5%9B%BD%E6%9C%80%E4%BD%B3%E6%B1%89%E5%A0%A1%E8%BF%9E%E9%94%81%E6%8E%92%E5%90%8D/',
      'title': '美国最佳汉堡连锁排名',
      'content': '最受好评的汉堡连锁店之一– Five Guys – 被康涅狄格州、乔治亚州、蒙大拿州、内布拉斯加州、俄勒冈州、南卡罗来纳州、佛蒙特州和西弗吉尼亚州评为第一名。 那是8 个州同意，...Read more',
      'score': 0.56690645,
      'raw_content': None
    }
  ],
  'response_time': 0.92,
  'request_id': '0ee8ee23-bbd3-48e2-b4c2-ca6d1c1dfe7a'
}

================================== Ai Message ==================================
以下是一些推荐的汉堡餐厅：
1. **[纽约最好吃的汉堡店](https://www.reddit.com/r/AskNYC/comments/1470o9z/best_burger_spot_in_nyc/?tl=zh-hans)** - 推荐的汉堡店包括 The Thompson 的 Burger Joint 和 Minetta Tavern 等。
2. **[美国旅游必吃7大人氣漢堡店](https://www.cosmopolitan.com/tw/lifestyle/food-and-drink/g44382807/hamburger-20230629/)** - 包括 The Habit Burger Grill、Cheeseburger in Paradise、Five Guys、IN-N-OUT、Shake Shack 和 Smashburger。
3. **[纽约市最佳汉堡](https://cn.tripadvisor.com/Restaurants-g60763-zfd10907-zfn7102345-New_York_City_New_York-Hamburger.html)** - 推荐的餐厅有 Ellen's Stardust Diner 和 Virgil's Real BBQ - NYC。
4. **[美国最佳汉堡连锁排名](https://mliesl.edu/contents/ch/%E7%BE%8E%E5%9B%BD%E6%9C%80%E4%BD%B3%E6%B1%89%E5%A0%A1%E8%BF%9E%E9%94%81%E6%8E%92%E5%90%8D/)** - Five Guys 是最受好评的汉堡连锁店之一。

希望这些推荐能帮助到您！如果您有特定的城市或地区需求，请告诉我。

补充一些内容：

Store 基本使用示例

from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import MemorySaver

# 创建 Store 和 Checkpointer
store = InMemoryStore()
checkpointer = MemorySaver()

# 编译图时同时传入 store 和 checkpointer
graph = builder.compile(checkpointer=checkpointer, store=store)

# 使用时，通过 namespace 组织数据
# namespace 类似于文件夹，用于隔离不同用户或不同维度的数据
config = {
    "configurable": {
        "thread_id": "user_123",
        "user_id": "user_123"  # 用于 Store 的 namespace
    }
}

# 在节点中使用 store 读写数据
def personalized_node(state, config):
    """个性化节点 - 从 store 读取用户偏好"""
    store = config["store"]
    user_id = config["configurable"]["user_id"]

    # 读取用户偏好（如果不存在返回默认值）
    preference = store.get(("user", user_id), "preference")

    # 基于用户偏好生成个性化回复
    if preference:
        return {"messages": [f"根据您的偏好：{preference.value}"]}
    else:
        return {"messages": ["请告诉我您的偏好"]}

# 写入用户偏好
store.put(("user", "user_123"), "preference", {"food": "汉堡", "color": "蓝色"})

Store 数据组织方式

# Store 使用 namespace + key 的方式组织数据
# namespace 是一个元组，用于隔离不同维度的数据

# 示例：存储用户信息
store.put(("user", "user_123"), "profile", {"name": "张三", "age": 25})
store.put(("user", "user_123"), "preferences", {"theme": "dark"})

# 示例：存储全局配置
store.put(("system",), "config", {"version": "1.0"})

# 读取数据
profile = store.get(("user", "user_123"), "profile")
config = store.get(("system",), "config")

# 搜索数据
all_users = store.search(("user",))

Checkpoint + Store 完整示例

from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, Annotated
import operator

# 定义状态
class State(TypedDict):
    messages: Annotated[list, operator.add]

# 定义节点
def chatbot(state: State, config):
    """聊天机器人 - 从 store 读取用户历史偏好"""
    store = config["config"]["store"]
    user_id = config["configurable"]["user_id"]

    # 读取用户历史
    history = store.get(("user", user_id), "history")

    # 构建带历史上下文的回复
    if history:
        context = f"用户历史偏好：{history.value}"
    else:
        context = "新用户"

    return {"messages": [AIMessage(content=f"你好！{context}")]}

# 创建存储
store = InMemoryStore()
checkpointer = MemorySaver()

# 构建图
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

# 编译时同时传入 checkpointer 和 store
graph = builder.compile(checkpointer=checkpointer, store=store)

# 使用
config = {
    "configurable": {
        "thread_id": "session_1",
        "user_id": "user_123"
    },
    "store": store  # store 需要传入 config
}

# 第一次对话
result = graph.invoke({"messages": [HumanMessage(content="我喜欢汉堡")]}, config)

# 存储用户偏好到 store
store.put(("user", "user_123"), "history", {"preference": "汉堡"})

# 第二次对话（新会话，但能读取到 store 中的偏好）
config2 = {
    "configurable": {
        "thread_id": "session_2",  # 新的会话
        "user_id": "user_123"      # 同一个用户
    },
    "store": store
}
result2 = graph.invoke({"messages": [HumanMessage(content="推荐餐厅")]}, config2)
print(result2["messages"][-1].content)
# 输出包含用户偏好的个性化回复

总结

持久化类型	适用场景	存储位置	生命周期
线程级持久化（Checkpoint）	单次会话的状态恢复	线程内部	单次会话
跨会话持久化（Store）	用户画像、偏好设置等长期数据	全局共享	跨会话持久

通过 Checkpoint + Store 的组合，LangGraph 实现了完整的 AI 记忆系统：

Checkpoint 保证单次会话的完整性和可恢复性
Store 实现跨会话的用户数据共享和个性化服务

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

GitHub今日热榜 | 2026-08-06

今日榜单呈现大换血，10个项目中7个为新晋项目。Cloudflare的Agent虚拟环境项目computer空降榜首，提供隔离文件系统和多执行后端支持。腾讯的TencentDB-Agent-Memory虽排名下滑至第3，但日增Star上涨70%。其他亮点包括： loopx为AI Agent团队提供看板管理系统经典面试指南system-design-primer回归榜单 DeepSeek-Reas