LangChain / 上下文与记忆3 小时读完 (大约 23511 个字)

09-上下文与记忆

背景

09-上下文与记忆

第09章：上下文与记忆

1、概述

1.1 为什么需要记忆（Memory）

记忆是一种记住之前互动信息的系统。随着 Agent 处理涉及大量用户交互的复杂任务，记忆变得至关重要！

大多数的大模型应用程序都会有一个会话接口，允许我们进行多轮的对话，并有一定的上下文记忆能力。比如：

（图示略）

但实际上，大模型本身是”无状态“的，不会记忆任何上下文的。即每次调用 agent.invoke() 都是全新的开始，不记得之前的对话。

举例：

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 从 .env 文件中加载环境变量
load_dotenv(override=True)

model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

情况 1：

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage

messages1 = [
    HumanMessage("你好，我叫小明"),
    AIMessage("很高兴认识你，小明"),
    HumanMessage("你用一句话介绍下你自己"),
]

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

response1 = agent.invoke({"messages": messages1})

for msg in response1["messages"]:
    msg.pretty_print()

输出：

================================ Human Message =================================

你好，我叫小明
================================== Ai Message ==================================

很高兴认识你，小明
================================ Human Message =================================

你用一句话介绍下你自己
================================== Ai Message ==================================

我是一个可以帮助你回答问题、写作、翻译和思考的 AI 助手。

messages2 = [HumanMessage("我叫什么名字？")]

response2 = agent.invoke({"messages": messages2})

for msg in response2["messages"]:
    msg.pretty_print()

输出：

================================ Human Message =================================

我叫什么名字？
================================== Ai Message ==================================

我不知道你的名字，除非你告诉我。
如果你愿意，可以直接告诉我，我就记住并称呼你。

情况 2：

messages3 = [
    HumanMessage("你好，我叫小明"),
    AIMessage("很高兴认识你，小明"),
    HumanMessage("你用一句话介绍下你自己"),
    AIMessage("我是一个由 OpenAI 训练的人工智能助手，可以帮你回答问题、写作和解决各种任务。"),
    HumanMessage("我叫什么名字？")
]

response3 = agent.invoke({"messages": messages3})

for msg in response3["messages"]:
    msg.pretty_print()

输出：

================================ Human Message =================================

你好，我叫小明
================================== Ai Message ==================================

很高兴认识你，小明
================================ Human Message =================================

你用一句话介绍下你自己
================================== Ai Message ==================================

我是一个由 OpenAI 训练的人工智能助手，可以帮你回答问题、写作和解决各种任务。
================================ Human Message =================================

我叫什么名字？
================================== Ai Message ==================================

你叫小明。

我们希望的智能体应该是右图的情况：

（图示略）

1.2 如何解决记忆问题

1.2.1 上下文工程

实现这个记忆功能，就需要额外的模块去保存我们和模型对话的上下文信息，然后在下一次请求时，把历史信息都输入给模型，让模型输出结果。

在 LangChain 中，记忆（Memory）就是专门负责”存储历史交互信息“的组件，核心作用是「保存上下文」和「提供上下文」，让 LLM 在每次响应时，都能”看到”之前的对话内容。

上下文工程（Context Engineering）负责”合理组织”这些记忆和任务信息，让 LLM 的响应更连贯、更贴合需求。这也是 Agent 能实现复杂多轮交互的核心基础。

类比：

（图示略）

举例：

（图示略）

1.2.2 上下文类型及相关的 API

LangChain 的上下文工程是基于 Agent 讨论的，而上下文工程是构建在 LangGraph 之上的。

LangGraph 提供了三种管理上下文的方法，这些方法结合了可变性和生命周期维度：

上下文类型	可变性	生命周期	描述	访问方法
动态运行时上下文	动态	单次运行	LangGraph 在单次运行中会演变的可变数据	`state` 对象
动态跨会话上下文	动态	跨对话	在对话间共享的持久数据。比如用户偏好、历史洞察、知识条目	`store` 对象
静态运行时上下文	静态	单次运行	在启动时传入的用户元数据、工具、数据库连接	`context` 对象

1.3 LangChain 的记忆

1.3.1 记忆的分类

官方说明：https://docs.langchain.com/oss/python/concepts/memory

记忆分为短期记忆和长期记忆，对应不同的使用场景：

短期记忆（Short-term memory、会话级记忆、thread-scoped memory）：作用范围是单个对话线程（Thread）内，一旦开启新对话（更换 thread_id），记忆即消失。
长期记忆（Long-term memory，跨会话级记忆）：在会话间存储用户特定或应用级数据，并在会话线程间共享。它可以随时在任何线程中被调用。记忆的范围是任意自定义命名空间，而不仅仅是单一线程 ID。

1.3.2 记忆的管理

在 LangChain v0.x 版本中，通过专用的 xxxMemory 类管理记忆。

在 LangChain v1.x 版本中，Agent 是构建在 LangGraph 图结构之上的，通过上文提到的 state 和 store 构建记忆系统。使用更简单、功能更统一。

state：短期记忆对象，以会话为单位组织，包含当前会话的所有消息记录以及自定义信息。
store：长期记忆对象，跨会话持久化的数据，通常需要结合向量数据库或外部存储实现。

2、短期记忆

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

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

State：默认存储历史消息列表 messages，通过 State 管理历史消息
Checkpointer：负责将 State 作为检查点持久化保存，检查点是某个时刻的 State 快照
Thread ID：用于唯一标识 State，LangChain 运行时会按照 thread_id 读写 State 快照

这就像玩 RPG 游戏时的”自动存档”：你不需要手动保存，系统在关键节点自动记录，下次进入游戏随时可以从上次的存档点继续。

2.1 基于内存的持久化器

这是最便捷的使用方式，适合快速测试或调试。

2.1.1 举例 1：没有记忆

from langchain.agents import create_agent
from langchain.messages import HumanMessage

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

print("\n第一轮对话：")
response1 = agent.invoke({
    "messages": [HumanMessage("我叫张三")]
})
print(f"Agent: {response1['messages'][-1].content}")

print("\n第二轮对话：")
response2 = agent.invoke({
    "messages": [HumanMessage("我叫什么？")]
})
print(f"Agent: {response2['messages'][-1].content}")

输出：

第一轮对话：
Agent: 你好，张三！很高兴认识你。有什么我可以帮你的吗？

第二轮对话：
Agent: 我不知道你的名字，除非你告诉我。
如果你愿意，可以直接发给我，我就能记住在这次对话里称呼你。

2.1.2 举例 2：拥有记忆

步骤 1：第一、二轮对话

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

# 1. 创建 Agent 时添加 checkpointer
agent = create_agent(
    model=model,
    checkpointer=checkpointer   # 添加内存管理
)

# 2. 调用时指定 thread_id
config = {
    "configurable": {
        "thread_id": "1"
    }
}

print("\n第一轮对话：")
response1 = agent.invoke({
    "messages": [HumanMessage("我叫张三")]},
    config=config   # 传入 config
)
print(f"Agent: {response1['messages'][-1].content}")

print("\n第二轮对话：")
response2 = agent.invoke({
    "messages": [HumanMessage("我叫什么？")]},
    config=config   # 使用相同的 thread_id
)
print(f"Agent: {response2['messages'][-1].content}")

输出：

第一轮对话：
Agent: 你好，张三！很高兴认识你。有什么我可以帮你的吗？

第二轮对话：
Agent: 你叫张三。

说明：你只需传入 checkpointer 和 config，Agent 就能自然具备连续对话能力。

from rich import print as rprint

latest_state = agent.get_state(config)
rprint(latest_state)

输出（已精简）：

StateSnapshot(
    values={
        'messages': [
            HumanMessage(content='我叫张三', ...),
            AIMessage(content='你好，张三！很高兴认识你。有什么我可以帮你的吗？', ...),
            HumanMessage(content='我叫什么？', ...),
            AIMessage(content='你叫张三。', ...)
        ]
    },
    next=(),
    config={
        'configurable': {
            'thread_id': '1',
            'checkpoint_ns': '',
            'checkpoint_id': '1f164d5b-625c-679d-8004-1ef8a3347091'
        }
    },
    ...
)

步骤 2：第三轮对话

当我们再次进行对话时，直接带入线程 ID，即可带入此前对话记忆：

print("\n第三轮对话：")
response3 = agent.invoke({
    "messages": [HumanMessage("我刚才问了什么问题？")]},
    config=config   # 使用相同的 thread_id
)
print(f"Agent: {response3['messages'][-1].content}")

输出：

1 2	第三轮对话： Agent: 你刚才问的是："我叫什么？"

说明：第三轮对话中的”我刚才问了什么问题？”就能在第二轮存储的消息历史中找到答案。所有消息历史都会自动追加到 AgentState 的 messages 字段中，无需手动维护。

步骤 3：更新线程 ID

而如果更新线程 ID，则会重新开启对话：

config2 = {
    "configurable": {
        "thread_id": "2"
    }
}

response4 = agent.invoke(
    {"messages": [HumanMessage("你还记得我叫什么名字么？")]},
    config=config2
)

print(response4['messages'][-1].content)

输出：

1 2	我目前不知道你的名字，因为你还没有告诉我。如果你愿意，可以直接发我你的名字，我就用这个称呼你。

说明：thread_id 隔离不同会话空间。

2.1.3 关键步骤说明

第 1 步：初始化记忆引擎

1	checkpointer = InMemorySaver()

创建一个内存级的记忆存储。

注意：InMemorySaver 内存中保存，进程结束就丢失数据，适合测试。生产环境可换成数据库持久化的 SqliteSaver、PostgresSaver 等。

第 2 步：绑定 Agent

agent = create_agent(
    model=model,
    checkpointer=checkpointer
)

在 create_agent 时传入 checkpointer，让 Agent 具备状态存储能力。

第 3 步：设定会话 ID

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

为每次调用指定线程标识。同一个 thread_id 共享记忆，不同 thread_id 完全隔离。

# 会话 1
config1 = {"configurable": {"thread_id": "1"}}
agent.invoke({...}, config=config1)

# 会话 2
config2 = {"configurable": {"thread_id": "2"}}
agent.invoke({...}, config=config2)

# 两个会话完全独立

thread_id 是记忆管理的核心开关：在会话 2 里询问会话 1 的会话信息，Agent 会表示不知道——因为双方记忆空间完全隔离。

生产环境中：

场景 1：多用户聊天

不同 thread_id = 不同会话，Agent 能正确记住每个会话的内容。

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

# 用户 Alice
config_alice = {"configurable": {"thread_id": "user_alice"}}
agent.invoke({"messages": [...]}, config_alice)
...

# 用户 Bob
config_bob = {"configurable": {"thread_id": "user_bob"}}
agent.invoke({"messages": [...]}, config_bob)
...

# 两个会话完全独立

场景 2：同一用户的不同任务

# 任务 1：写代码
config_task1 = {"configurable": {"thread_id": "task_coding"}}
agent.invoke({"messages": [...]}, config_task1)
...

# 任务 2：写文档
config_task2 = {"configurable": {"thread_id": "task_docs"}}
agent.invoke({"messages": [...]}, config_task2)
...

2.1.4 工作原理

内存保存了什么？

agent.invoke({"messages": [{"role": "user", "content": "你好"}]}, config)
# InMemorySaver 保存：
# {
#     "thread_id": "xxx",
#     "messages": [
#         HumanMessage("你好"),
#         AIMessage("你好！有什么可以帮助你的吗？")
#     ]
# }

agent.invoke({"messages": [{"role": "user", "content": "天气"}]}, config)
# InMemorySaver 更新：
# {
#     "thread_id": "xxx",
#     "messages": [
#         HumanMessage("你好"),
#         AIMessage("你好！有什么可以帮助你的吗？"),
#         HumanMessage("天气"),
#         AIMessage("...")
#     ]
# }

自动追加历史

# 你只需要传新消息
agent.invoke(
    {"messages": [{"role": "user", "content": "新问题"}]},
    config
)

此时，checkpointer 自动：

读取之前的历史
追加新消息
调用模型（传入完整历史）
保存新的历史

说明：checkpointer 会自动管理历史。

2.1.5 常见问题

1、为什么 Agent 不记得？

检查：

✅ 是否添加了 checkpointer=InMemorySaver()？
✅ 是否传入了 config 参数？
✅ 两次调用的 thread_id 是否相同？

# ❌ 错误：没有 checkpointer
agent = create_agent(model=model, tools=[])
agent.invoke({...})   # 不会记住

# ❌ 错误：没有 config
agent = create_agent(model=model, tools=[], checkpointer=InMemorySaver())
agent.invoke({...})   # 不会记住

# ❌ 错误：thread_id 不同
agent.invoke({...}, config={"configurable": {"thread_id": "1"}})
agent.invoke({...}, config={"configurable": {"thread_id": "2"}})   # 不同会话

# ✅ 正确
agent = create_agent(model=model, tools=[], checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "1"}}
agent.invoke({...}, config)
agent.invoke({...}, config)   # 记得！

2、InMemorySaver 会丢失数据吗？

会！ InMemorySaver 只保存在内存中：

✅ 同一进程内有效（不支持跨进程共享）
❌ 程序重启后丢失（或进程重启后丢失）
❌ 不同进程无法共享

解决方案：持久化（SQLite、PostgreSQL）

3、内存会无限增长吗？

会！默认情况下，InMemorySaver 会保存所有消息。

问题：

消息越来越多（无限增长，需要管理上下文）
Token 消耗增加，甚至会超过模型的 token 限制
响应速度变慢、成本增加

解决方案：上下文管理（修剪、摘要）

4、如何清空某个会话的历史？

目前 InMemorySaver 没有提供删除 API。

临时方案：使用新的 thread_id 或重新创建 Agent。

2.2 基于外部存储介质的持久化器

如果将状态检查点（checkpointer）保存在内存，进程结束则状态丢失，生产环境不可接受。因此，生产环境要用持久化的外部存储介质，如 PostgreSQL。

LangGraph 提供的 checkpointer 后端列表如下：

https://docs.langchain.com/oss/python/langgraph/persistence#checkpointer-libraries

此处选择 PostgreSQL 作为持久化器。

2.2.1 数据库环境准备

在此之前，先准备好 PostgreSQL 环境，此处在云服务器的 Ubuntu 系统安装 PostgreSQL。大家需要根据自己云服务器位置，更改 URL 中的 IP 即可。

具体云服务器安装和 PostgreSQL 的安装，见文件 02-资料下的《Linux云服务器安装与PostgreSQL安装》。

1	URL：postgresql://langchain_user:abcd1234@118.195.128.47:5432/langchain_db?sslmode=disable

说明：

1、上述 URL 中的用户名、密码、IP 地址，需要根据自己的情况替换。

2、要在 LangChain 中对接 PostgreSQL，还需要额外的依赖，比如：
1
pip install langgraph-checkpoint-postgres
其实我们在课程开始的 requirements.txt 中已经提供了相关依赖，此处不必也不能重新安装，否则版本可能冲突。

2.2.2 代码实现

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 从 .env 文件中加载环境变量
load_dotenv(override=True)

model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.postgres import PostgresSaver

DB_URL = "postgresql://langchain_user:abcd1234@118.195.128.47:5432/langchain_db?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    # 初始化 PostgreSQL 数据库
    checkpointer.setup()

    agent = create_agent(
        model=model,
        checkpointer=checkpointer
    )

    config = {"configurable": {"thread_id": "1"}}
    response1 = agent.invoke(
        {"messages": [HumanMessage("你好，我是老王")]},
        config=config
    )
    print("=" * 30, "-> 第一次调用 <-", "=" * 30)
    for msg in response1["messages"]:
        msg.pretty_print()

    response2 = agent.invoke(
        {"messages": [HumanMessage("你好，我是谁？")]},
        config=config
    )

    print("=" * 30, "-> 第二次调用 <-", "=" * 30)
    for msg in response2["messages"]:
        msg.pretty_print()

输出：

============================== -> 第一次调用 <- ==============================
================================ Human Message =================================

你好，我是老王
================================== Ai Message ==================================

你好，老王！很高兴认识你。有什么我可以帮你的？
============================== -> 第二次调用 <- ==============================
================================ Human Message =================================

你好，我是老王
================================== Ai Message ==================================

你好，老王！很高兴认识你。有什么我可以帮你的？
================================ Human Message =================================

你好，我是谁？
================================== Ai Message ==================================

你是老王。

setup() 用于初始化 PostgreSQL 数据库，首次运行会创建必要的表，重复执行不会重新建表，底层逻辑是 CREATE IF NOT EXISTS。相关源码如下：

def setup(self) -> None:
    """Set up the checkpoint database asynchronously.

    This method creates the necessary tables in the Postgres database if they don't already exist
    and runs database migrations. It MUST be called directly by the user the first time
    checkpointer is used.
    """
    ...

2.2.3 查看持久化数据

查看 PostgreSQL 数据库，可以通过命令行或图形化工具查看：

1	ubuntu@VM-0-6-ubuntu:~$ psql "postgresql://langchain_user:abcd1234@localhost:5432/langchain_db?sslmode=disable"

psql (16.13 (Ubuntu 16.13-0ubuntu0.24.04.1))
Type "help" for help.

langgraph_db=>

PostgreSQL 的存储结构是 Database -> Schema -> Table：

Database：数据库
Schema：分区
Table：表

需求 1：查看所有数据库

1	langgraph_db=> \l

List of databases
     Name      |     Owner     | Encoding | Locale Provider |  Collate   |   Ctype    | ICU Locale | ICU Rules |      Access privileges
---------------+----------------+----------+-----------------+------------+------------+------------+-----------+-----------------------------------
 langgraph_db  | langgraph_user | UTF8     | libc            | C.UTF-8    | C.UTF-8    |            |           | =Tc/langgraph_user +
               |                |          |                 |            |            |            |           | langgraph_user=CTc/langgraph_user
 postgres      | postgres       | UTF8     | libc            | C.UTF-8    | C.UTF-8    |            |           |
 template0     | postgres       | UTF8     | libc            | C.UTF-8    | C.UTF-8    |            |           | =c/postgres +
               |                |          |                 |            |            |            |           | postgres=CTc/postgres
 template1     | postgres       | UTF8     | libc            | C.UTF-8    | C.UTF-8    |            |           | =c/postgres +
               |                |          |                 |            |            |            |           | postgres=CTc/postgres
(4 rows)

命令行前缀 langgraph_db 标识了当前所在数据库。

需求 2：查看所有 schema

1	langgraph_db=> \dn

List of schemas
  Name  |       Owner
--------+-------------------
 public | pg_database_owner
(1 row)

需求 3：查看当前所处的 schema

1	langgraph_db=> select current_schema();

 current_schema
----------------
 public
(1 row)

需求 4：查看当前 schema 下的所有表

1	langgraph_db=> \dt

List of relations
 Schema |         Name          | Type  |     Owner
--------+-----------------------+-------+----------------
 public | checkpoint_blobs      | table | langgraph_user
 public | checkpoint_migrations | table | langgraph_user
 public | checkpoint_writes     | table | langgraph_user
 public | checkpoints           | table | langgraph_user
(4 rows)

这四张表都是 setup() 函数初始化时创建的。

checkpoints：这是主表，存每个 thread 在某个时刻的 checkpoint 快照。
checkpoint_blobs：这张表专门存不适合直接内联进 checkpoints.checkpoint 的较复杂 channel 值。
checkpoint_writes：这张表存的是中间写入 / pending writes，不是最终完整 checkpoint。
checkpoint_migrations：这张表不是业务数据表，而是迁移版本表。

2.3 对比两种方式

举例 1：基于内存存储

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

agent = create_agent(
    model=model,
    checkpointer=InMemorySaver()
)
config = {"configurable": {"thread_id": "1"}}

print("=" * 30, "-> 第一次调用 <-", "=" * 30)
response1 = agent.invoke(
    {"messages": [HumanMessage("你好，我是谁？")]},
    config=config
)
for msg in response1["messages"]:
    msg.pretty_print()

print("=" * 30, "-> 第二次调用 <-", "=" * 30)
response2 = agent.invoke(
    {"messages": [HumanMessage("我是老王")]},
    config=config
)
for msg in response2["messages"]:
    msg.pretty_print()

print("=" * 30, "-> 第三次调用 <-", "=" * 30)
response3 = agent.invoke(
    {"messages": [HumanMessage("你好，我是谁？")]},
    config
)

for msg in response3["messages"]:
    msg.pretty_print()

输出如下：

============================== -> 第一次调用 <- ==============================
================================ Human Message =================================

你好，我是谁？
================================== Ai Message ==================================

你好！我不知道你的真实身份，除非你告诉我。
如果你愿意，我可以根据你提供的信息，帮你判断你是谁、你像哪类人，或者帮你整理一段自我介绍。
============================== -> 第二次调用 <- ==============================
================================ Human Message =================================

你好，我是谁？
================================== Ai Message ==================================

你好！我不知道你的真实身份，除非你告诉我。
如果你愿意，我可以根据你提供的信息，帮你判断你是谁、你像哪类人，或者帮你整理一段自我介绍。
================================ Human Message =================================

我是老王
================================== Ai Message ==================================

你好，老王！很高兴认识你。
如果你愿意，我可以直接这样称呼你。
============================== -> 第三次调用 <- ==============================
================================ Human Message =================================

你好，我是谁？
================================== Ai Message ==================================

你好！我不知道你的真实身份，除非你告诉我。
如果你愿意，我可以根据你提供的信息，帮你判断你是谁、你像哪类人，或者帮你整理一段自我介绍。
================================ Human Message =================================

我是老王
================================== Ai Message ==================================

你好，老王！很高兴认识你。
如果你愿意，我可以直接这样称呼你。
================================ Human Message =================================

你好，我是谁？
================================== Ai Message ==================================

你好，老王。你是老王。

可以发现，每次执行的输出完全相同，而我们并没有更改 thread_id，之所以看不到上次运行的状态是因为每次运行创建新的 Saver()，历史 State 被丢弃了。

举例 2：基于外部存储器存储

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.postgres import PostgresSaver

DB_URL = "postgresql://langchain_user:abcd1234@118.195.128.47:5432/langchain_db?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    # 初始化数据库
    checkpointer.setup()

    agent = create_agent(
        model=model,
        checkpointer=checkpointer
    )

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

    print("=" * 30, "-> 第一次调用 <-", "=" * 30)
    response1 = agent.invoke(
        {"messages": [HumanMessage("你好，我是谁啊？")]},
        config
    )
    for msg in response1["messages"]:
        msg.pretty_print()

    print("=" * 30, "-> 第二次调用 <-", "=" * 30)
    response2 = agent.invoke(
        {"messages": [HumanMessage("我是老王~")]},
        config
    )
    for msg in response2["messages"]:
        msg.pretty_print()

    print("=" * 30, "-> 第三次调用 <-", "=" * 30)
    response3 = agent.invoke(
        {"messages": [HumanMessage("你好，我是谁？？")]},
        config
    )

    for msg in response3["messages"]:
        msg.pretty_print()

第一次输出：

============================== -> 第一次调用 <- ==============================
================================ Human Message =================================

你好，我是谁啊？
================================== Ai Message ==================================

你好！我只能知道：你现在是这次和我对话的用户。
如果你是想问"我记不记得你是谁"，那我其实**不会自动知道你的真实身份**，除非你在对话里告诉我。

你可以直接告诉我你想让我怎么称呼你。
============================== -> 第二次调用 <- ==============================
================================ Human Message =================================

你好，我是谁啊？
================================== Ai Message ==================================

你好！我只能知道：你现在是这次和我对话的用户。
如果你是想问"我记不记得你是谁"，那我其实**不会自动知道你的真实身份**，除非你在对话里告诉我。

你可以直接告诉我你想让我怎么称呼你。
================================ Human Message =================================

我是老王~
================================== Ai Message ==================================

你好，老王~ 很高兴认识你！
接下来我可以怎么帮你？
============================== -> 第三次调用 <- ==============================
================================ Human Message =================================

你好，我是谁啊？
================================== Ai Message ==================================

你好！我只能知道：你现在是这次和我对话的用户。
如果你是想问"我记不记得你是谁"，那我其实**不会自动知道你的真实身份**，除非你在对话里告诉我。

你可以直接告诉我你想让我怎么称呼你。
================================ Human Message =================================

我是老王~
================================== Ai Message ==================================

你好，老王~ 很高兴认识你！
接下来我可以怎么帮你？
================================ Human Message =================================

你好，我是谁？？
================================== Ai Message ==================================

你好，老王 🙂

第二次运行输出：

============================== -> 第一次调用 <- ==============================
================================ Human Message =================================

你好，我是谁啊？
================================== Ai Message ==================================

你好！我只能知道：你现在是这次和我对话的用户。
如果你是想问"我记不记得你是谁"，那我其实**不会自动知道你的真实身份**，除非你在对话里告诉我。

你可以直接告诉我你想让我怎么称呼你。
================================ Human Message =================================

我是老王~
================================== Ai Message ==================================

你好，老王~ 很高兴认识你！
接下来我可以怎么帮你？
================================ Human Message =================================

你好，我是谁？？
================================== Ai Message ==================================

你好，老王 🙂
================================ Human Message =================================

你好，我是谁啊？
================================== Ai Message ==================================

你好，老王。
============================== -> 第二次调用 <- ==============================
================================ Human Message =================================

你好，我是谁啊？
================================== Ai Message ==================================

你好！我只能知道：你现在是这次和我对话的用户。
如果你是想问"我记不记得你是谁"，那我其实**不会自动知道你的真实身份**，除非你在对话里告诉我。

你可以直接告诉我你想让我怎么称呼你。
================================ Human Message =================================

我是老王~
================================== Ai Message ==================================

你好，老王~ 很高兴认识你！
接下来我可以怎么帮你？
================================ Human Message =================================

你好，我是谁？？
================================== Ai Message ==================================

你好，老王 🙂
================================ Human Message =================================

你好，我是谁啊？
================================== Ai Message ==================================

你好，老王。
================================ Human Message =================================

我是老王~
================================== Ai Message ==================================

你好，老王~ 有什么我可以帮你的？
============================== -> 第三次调用 <- ==============================
================================ Human Message =================================

你好，我是谁啊？
================================== Ai Message ==================================

你好！我只能知道：你现在是这次和我对话的用户。
如果你是想问"我记不记得你是谁"，那我其实**不会自动知道你的真实身份**，除非你在对话里告诉我。

你可以直接告诉我你想让我怎么称呼你。
================================ Human Message =================================

我是老王~
================================== Ai Message ==================================

你好，老王~ 很高兴认识你！
接下来我可以怎么帮你？
================================ Human Message =================================

你好，我是谁？？
================================== Ai Message ==================================

你好，老王 🙂
================================ Human Message =================================

你好，我是谁啊？
================================== Ai Message ==================================

你好，老王。
================================ Human Message =================================

我是老王~
================================== Ai Message ==================================

你好，老王~ 有什么我可以帮你的？
================================ Human Message =================================

你好，我是谁？？
================================== Ai Message ==================================

你好，老王。

根据输出判断，状态是累积的。

由此可以得出结论：即便重新创建 Saver()，只要 thread_id 一致，历史状态就可以和当前调用串联起来。

总结：

InMemorySaver() 将状态持久化到内存，进程结束或重建 Saver() 则历史状态丢失。
基于外部存储介质（如 PostgreSQL）的持久化器，其存储的状态不会随进程终止而丢失，只要不显式删除历史状态，即可通过 thread_id 加载历史状态。

2.4 记忆治理策略（上下文管理）

随着对话的进行，历史消息不断累积，state 会持续增长，为模型带来挑战：

LLM 的上下文窗口是有限的，完整历史可能无法装入 LLM 的上下文窗口，导致上下文丢失或错误。
即便模型的上下文窗口够大，多数 LLM 在长上下文场景仍然表现不佳。模型会被陈旧或离题的内容”分散注意力”。
同时，会带来高昂的 Token 花费。

此时需要对上下文进行管理：对历史记录进行压缩、清理、重组等。

2.4.1 消息裁剪

调用模型前裁剪上下文。

目标是控制 Token 用量，通常保留系统初始消息和最近若干消息，或按 Token 数保留末尾内容。

适合成本敏感、对旧上下文依赖不强的场景。

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 从 .env 文件中加载环境变量
load_dotenv(override=True)

model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

from langchain_core.messages import HumanMessage
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig
from typing import Any

@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    messages = state["messages"]

    if len(messages) <= 3:
        return None

    first_msg = messages[0]
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    new_messages = [first_msg] + recent_messages

    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),
            *new_messages
        ]
    }

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

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

agent.invoke({"messages": [HumanMessage("你好，我是老王")]}, config)
agent.invoke({"messages": [HumanMessage("从现在起，你叫小王")]}, config)
agent.invoke({"messages": [HumanMessage("今天天气不错")]}, config)
final_response = agent.invoke({"messages": [HumanMessage("告诉我，你是谁？我是谁？")]}, config)

for msg in final_response["messages"]:
    msg.pretty_print()

假设有 5 条消息：[H1, A1, H2, A2, H3]（当前总数 5，奇数）。

执行 messages[-4:] 取出 [A1, H2, A2, H3]。加上第一条 H1，最终大模型看到的是 [H1, A1, H2, A2, H3]（相当于没剪掉什么）。

输出：

================================ Human Message =================================

你好，我是老王
================================== Ai Message ==================================

好的，从现在起你可以叫我小王。
那我也叫你老王，可以吗？
================================ Human Message =================================

今天天气不错
================================== Ai Message ==================================

是啊，天气不错的话，心情也容易好起来。
你今天有什么安排吗？
================================ Human Message =================================

告诉我，你是谁？我是谁？
================================== Ai Message ==================================

我是你的AI助手。
你是老王。

显然，裁剪生效了。

分析：

我们来看看连续 4 次 invoke 时，内存里到底发生了什么：

步骤	传入消息	触发前 State 中的 `messages` 列表（长度）	最终传给 LLM 的 `messages` 实际内容	触发与否与执行逻辑
1	`invoke("你好，我是老王")`	`[H: 你好，我是老王]`（1）	`[H: 你好，我是老王]`	不触发（长度 ≤ 3）
	模型自动回复（A1）	存储到 Memory 中，等待下次	`[H: 你好..., A1: 你好老王...]`（2）	—
2	`invoke("从现在起，你叫小王")`	`[H: 你好..., A1: ..., H: 从现在...]`（3）	`[H: 你好..., A1: ..., H: 从现在...]`	不触发（长度 = 3）
	模型自动回复（A2）	存储到 Memory 中，等待下次	`[H: 你好..., A1: ..., H: 从现在..., A2: 好的...]`（4）	—
3	`invoke("今天天气不错")`	`[H: 你好..., A1: ..., H: 从现在..., A2: ..., H: 今天...]`（5）	触发裁剪！ `[H: 你好，我是老王]`（第一条）+ `[A1, H: 从现在..., A2, H: 今天...]`（后 4 条）	长度 5 是奇数，取后 4 条（`messages[-4:]`）
	模型自动回复（A3）	存储到 Memory 中，此时 State 被重置为上 5 条 + 新回复的 A3（共 6 条）	—	—
4	`invoke("告诉我，你是谁？我是谁？")`	`[H: 你好..., A1: ..., H: 从现在..., A2: ..., H: 今天..., A3: ..., H: 告诉我...]`（7）	触发裁剪！ `[H: 你好，我是老王]`（第一条）+ `[A2, H: 今天..., A3, H: 告诉我...]`（后 4 条）	长度 7 是奇数，取后 4 条（`messages[-4:]`）

2.4.2 消息删除

消息裁剪强调”在模型调用前裁剪消息列表，控制模型可以看到的上下文范围”，而消息删除强调模型调用完成后将某些消息从消息列表中移除，永久更改状态。

适合明确要遗忘、清理、重置某些历史。

from langchain.messages import RemoveMessage
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig

@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    messages = state["messages"]
    # 保持最近的 5 条消息
    if len(messages) > 5:
        # 框架中通常使用 RemoveMessage 来标记删除，并返回更新状态。
        to_delete = len(messages) - 5
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:to_delete]]}
    return None

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

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

agent.invoke({"messages": "你好，我是老王"}, config)
agent.invoke({"messages": "从现在起，你叫小王"}, config)
agent.invoke({"messages": "今天天气不错"}, config)
final_response = agent.invoke({"messages": "告诉我，你是谁？我是谁？"}, config)

for msg in final_response["messages"]:
    msg.pretty_print()

输出：

================================== Ai Message ==================================

好的，从现在起你可以叫我小王。
================================ Human Message =================================

今天天气不错
================================== Ai Message ==================================

是啊，天气不错的时候心情也容易变好。你今天打算出去走走，还是在家休息一下？
================================ Human Message =================================

告诉我，你是谁？我是谁？
================================== Ai Message ==================================

我是小王，你是老王。

显然，模型调用后消息列表规模超过阈值触发了消息裁剪。

说明：只要消息总数超过 5 条，就计算超出几条（to_delete），然后精准地删掉最老的那几条，使剩下的消息总数永远保持在 5 条。

分析消息列表 messages 的数量变化：

📥 第一轮：

用户说：”你好，我是老王”（第 1 条）
AI 回复：”你好，老王！”（第 2 条）
数量：此时 len(messages) == 2（不大于 5）
结果：中间件不触发。（记忆：记住你是老王）

📥 第二轮：

用户说：”从现在起，你叫小王”（第 3 条）
AI 回复：”好的，我是小王。”（第 4 条）
数量：此时 len(messages) == 4（不大于 5）
结果：中间件不触发。（记忆：记住你是老王、我是小王）

📥 第三轮：

用户说：”今天天气不错”（第 5 条）
AI 回复：”是啊，天气好心情也好。”（第 6 条）
数量：此时 len(messages) == 6。触发中间件条件（> 5）！

清理发生：

计算 to_delete = 6 - 5 = 1
清理 messages[:1]，也就是只删除了第 1 条消息（”你好，我是老王”）
剩余消息（共 5 条）：
1. AI 回复：”你好，老王！”（第 2 条）
2. 用户说：”从现在起，你叫小王”（第 3 条）
3. AI 回复：”好的，我是小王。”（第 4 条）
4. 用户说：”今天天气不错”（第 5 条）
5. AI 回复：”是啊，天气好心情也好。”（第 6 条）

📥 第四轮（最终提问）：

用户输入：”告诉我，你是谁？我是谁？”（第 7 条）
大模型读取上下文：在 AI 做出最终回复前，大模型会读取当前留存的 5 条历史记录 + 用户的新提问。
大模型通过上下文里 AI 自己说过的”你好，老王！”，依然能够推断出”哦，原来坐在对面的人叫老王”。
AI 做出最终回复：比如”你是老王，我是小王啊。”（第 8 条）
数量：此时回复完后，总数 len(messages) == 8。再次触发中间件！

清理发生：

计算 to_delete = 8 - 5 = 3
清理 messages[:3]，即把目前列表中最老的前 3 条消息删除。
剩余消息：依然精准保持在最后 5 条：
1. AI 回复：”好的，我是小王。”（第 4 条）
2. 用户说：”今天天气不错”（第 5 条）
3. AI 回复：”是啊，天气好心情也好。”（第 6 条）
4. 用户说：”告诉我，你是谁？我是谁？”（第 7 条）
5. AI 回复：”你是老王，我是小王啊。”（第 8 条）

RemoveMessage 到底干了什么？

当你在中间件里返回 [RemoveMessage(id=m.id)] 时，你实际上是向框架发送了一个删除指令。

框架的底层处理逻辑如下：

[历史消息池（内存中持续存在）]
├── Message(id="1", content="你好，我是老王")
├── Message(id="2", content="...")
└── RemoveMessage(id="1")   <-- 这是一个新追加进去的"墓碑"标记

追加”墓碑”标记：框架收到 RemoveMessage(id="1") 后，并不会去内存的数组里把 id="1" 的对象删掉，而是把这个 RemoveMessage 作为一条新记录追加到当前线程的状态历史中。这个 RemoveMessage 就像是一个”墓碑”。
运行时过滤合并（Reducer）：当下一次你再次调用 agent.invoke 或者大模型要去读取上下文时，框架的内置合并器（Reducer）会把”原始消息”和”墓碑标记”放在一起进行计算：

1	原始消息 + 墓碑标记 → 对外隐藏

它在丢给大模型之前，会自动把被标记删除的消息过滤掉。

2.4.3 摘要

把早期历史压缩成摘要，再替换原始消息。

消息裁剪和删除都会导致上下文缺失，影响回答质量和用户体验。和它们相比，摘要是更适合长会话的折中方案：保语义，不保原文。官方推荐内置 SummarizationMiddleware。上一章已有讲解。

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 从 .env 文件中加载环境变量
load_dotenv(override=True)

model_out = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 从 .env 文件中加载环境变量
load_dotenv(override=True)

model_in = init_chat_model(
    model="gpt-4o-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

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

# 创建带摘要中间件的 Agent
agent = create_agent(
    model=model_out,
    tools=[],
    checkpointer=InMemorySaver(),
    middleware=[
        SummarizationMiddleware(
            model=model_in,
            trigger=[
                ("tokens", 100),   # 超过 100 tokens 就摘要
            ],
            keep=("messages", 2),
            summary_prompt="对历史消息摘要，消息列表如下\n{messages}",
        )
    ]
)

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

print("\n进行多轮对话...")
conversations = [
    "我叫张三，是工程师。这里是一段非常长非常长的废话..." * 20,   # 强制撑爆 100 tokens
    "请总结一下我的信息"
]

for msg in conversations:
    response = agent.invoke(
        {"messages": [{"role": "user", "content": msg}]},
        config=config
    )
    for msg in response["messages"]:
        msg.pretty_print()
    print("*" * 50)

输出：

进行多轮对话...

================================ Human Message =================================

我叫张三，是工程师。这里是一段非常长非常长的废话...（重复多次）
================================== Ai Message ==================================

你好，张三。你想让我基于这段内容做什么？
如果你愿意，我可以帮你：

1. **提炼关键信息**
2. **总结成一句话/一段话**
3. **改写成更正式的自我介绍**
4. **识别并去掉重复废话**
5. **翻译成英文**

例如，提炼后可以写成：

> 张三是一名工程师。

如果你告诉我目标，我可以直接帮你处理。
**************************************************
================================ Human Message =================================

Here is a summary of the conversation to date:

根据历史消息，用户"张三"自称是一名工程师，并重复发送了大量无实质内容的废话（如"这里是一段非常长非常长的废话"）。摘要如下：

**摘要**：用户张三（工程师）发送了重复的冗长废话，内容无实际信息。
================================== Ai Message ==================================

你好，张三。你想让我基于这段内容做什么？
如果你愿意，我可以帮你：

1. **提炼关键信息**
2. **总结成一句话/一段话**
3. **改写成更正式的自我介绍**
4. **识别并去掉重复废话**
5. **翻译成英文**

例如，提炼后可以写成：

> 张三是一名工程师。

如果你告诉我目标，我可以直接帮你处理。
================================ Human Message =================================

请总结一下我的信息
================================== Ai Message ==================================

根据你提供的内容，目前能总结出的信息是：

- **姓名**：张三
- **职业**：工程师
- **特点**：之前发送的内容以重复、冗长的无实质信息为主

如果你愿意，我还可以帮你把这些信息整理成：
- **正式简介**
- **一句话自我介绍**
- **简历风格描述**
- **更自然的个人介绍**
**************************************************

final_state = agent.get_state(config)
print("\n--- 最终保存在 State 中的真实消息 ---")
for msg in final_state.values.get("messages", []):
    msg.pretty_print()

输出：

--- 最终保存在 State 中的真实消息 ---
================================ Human Message =================================

Here is a summary of the conversation to date:

根据历史消息，用户"张三"自称是一名工程师，并重复发送了大量无实质内容的废话（如"这里是一段非常长非常长的废话"）。摘要如下：

**摘要**：用户张三（工程师）发送了重复的冗长废话，内容无实际信息。
================================== Ai Message ==================================

你好，张三。你想让我基于这段内容做什么？
如果你愿意，我可以帮你：

1. **提炼关键信息**
2. **总结成一句话/一段话**
3. **改写成更正式的自我介绍**
4. **识别并去掉重复废话**
5. **翻译成英文**

例如，提炼后可以写成：

> 张三是一名工程师。

如果你告诉我目标，我可以直接帮你处理。
================================ Human Message =================================

请总结一下我的信息
================================== Ai Message ==================================

根据你提供的内容，目前能总结出的信息是：

- **姓名**：张三
- **职业**：工程师
- **特点**：之前发送的内容以重复、冗长的无实质信息为主

如果你愿意，我还可以帮你把这些信息整理成：
- **正式简介**
- **一句话自我介绍**
- **简历风格描述**
- **更自然的个人介绍**

工作原理：

对话历史: [消息1, 消息2, ..., 消息20] (超过 100 tokens)
↓
SummarizationMiddleware 自动触发
↓
摘要旧消息: "用户是张三，在北京工作，喜欢编程..."
↓
新历史: [摘要, 最近消息] (减少到 100 tokens)

常见问题：

1. 摘要会丢失信息吗？

会有一些细节丢失，但：

重要信息会保留（姓名、关键事实）
最近的消息完整保留
对于大部分场景足够

2. 设置最大 Token 数触发摘要的标准是啥？

# 模型上下文窗口 4k → 设置 3000
# 模型上下文窗口 8k → 设置 6000
# 模型上下文窗口 16k → 设置 12000

# 留一些余量给工具调用和系统提示

3. 摘要成本高吗？

摘要只在超过阈值时触发
可以使用便宜的模型（如 gpt-4o-mini）
相比传输全部历史，通常更便宜

4. 摘要触发的频率要关注吗？

要关注，根据监控摘要触发频率，调整阈值。

如果频繁触发 → 提高阈值
如果从不触发 → 降低阈值

2.4.4 自定义过滤策略

通过中间件可以随意更改消息列表，因此可以实现任意的过滤策略。

此处不再演示。

2.5 了解：state 的理解

state 是 Agent 底层有状态运行图的状态信息，是 AgentState 类型的实例，定义如下：

class AgentState(TypedDict, Generic[ResponseT]):
    """State schema for the agent."""

    messages: Required[Annotated[list[AnyMessage], add_messages]]
    jump_to: NotRequired[Annotated[JumpTo | None, EphemeralValue, PrivateStateAttr]]
    structured_response: NotRequired[Annotated[ResponseT, OmitFromInput]]

说明：

AgentState 是 TypedDict 的子类，这意味着我们可以按照字典的读写方式访问其实例的元素。

该类有三个字段：

messages：截止到当前节点的历史会话消息记录，标记为 Required。

jump_to：中间件章节介绍过，表示跳转至运行图的指定节点，标记为 NotRequired，表示该字段可以为 None。

structured_response：结构化输出内容，当我们启用结构化输出时，结构化后的内容会被记录在这里。

举例：

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 从 .env 文件中加载环境变量
load_dotenv(override=True)

model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

from langchain.agents import create_agent
from langchain.agents.middleware import AgentState, before_model, wrap_tool_call, after_agent
from langchain.tools.tool_node import ToolCallRequest
from langchain.messages import HumanMessage, AIMessage, ToolMessage
from langchain.tools import tool
from langgraph.runtime import Runtime
from langgraph.types import Command
from typing import Any, Callable
from pydantic import BaseModel, Field

class WeatherInfo(BaseModel):
    """城市天气情况"""
    city: str = Field(description="城市名称")
    temperature: str = Field(description="气温")
    desc: str = Field(description="当日天气概述")

@tool(parse_docstring=True)
def get_weather(city: str):
    """
    获取当日天气

    Args:
        city: 城市名称
    """
    return f"[{city}] 今天气温9~16度，万里无云，天气不错适合外出"

@before_model(can_jump_to=["tools"])
def direct_tool_call(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    last_msg = state["messages"][-1]
    if isinstance(last_msg, HumanMessage) and "天气" in last_msg.text and "北京" in last_msg.text:
        fake_tool_call = AIMessage(
            content="人工构造的消息",
            tool_calls=[
                {
                    "name": "get_weather",
                    "args": {
                        "city": "北京"
                    },
                    "id": "direct_call_id"
                }
            ]
        )
        return {
            "messages": [fake_tool_call],
            "jump_to": "tools"
        }

    return None

@wrap_tool_call
def first_check(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command]
) -> ToolMessage | Command:
    print("=" * 30, '-> In first_check Middleware <-', "=" * 30)
    print(f"{request.state.get("jump_to", None) = }")
    print(f"{request.state.get("structured_response", None) = }")

    return handler(request)

@after_agent
def final_check(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print("=" * 30, '-> In final_check Middleware <-', "=" * 30)
    for msg in state["messages"]:
        msg.pretty_print()

    print("=" * 30, "-> 消息打印完毕 <-", "=" * 30)
    print(f"{state.get("jump_to", None) = }")
    print(f"{state.get("structured_response", None) = }")

    return None

agent = create_agent(
    model=model,
    response_format=WeatherInfo,
    middleware=[direct_tool_call, first_check, final_check],
    tools=[get_weather],
)

response = agent.invoke({
    "messages": [
        HumanMessage("请帮我查询北京当日天气")
    ]
})

输出：

============================== -> In first_check Middleware <- ==============================
request.state.get("jump_to", None) = 'tools'
request.state.get("structured_response", None) = None
============================== -> In final_check Middleware <- ==============================
================================ Human Message =================================

请帮我查询北京当日天气
================================== Ai Message ==================================

人工构造的消息
Tool Calls:
  get_weather (direct_call_id)
  Call ID: direct_call_id
  Args:
    city: 北京
================================= Tool Message =================================
Name: get_weather

[北京] 今天气温9~16度，万里无云，天气不错适合外出
================================== Ai Message ==================================

{"city":"北京","temperature":"9~16℃","desc":"万里无云，天气不错适合外出"}
============================== -> 消息打印完毕 <- ==============================
state.get("jump_to", None) = None
state.get("structured_response", None) = WeatherInfo(city='北京', temperature='9~16℃', desc='万里无云，天气不错适合外出')

分析：

我们在模型调用之前通过 jump_to 直接跳转至工具节点。
在工具调用前打印状态信息，此时的 jump_to 非空。
在 Agent 执行完毕之后打印状态信息，此时的 structured_response 非空而 jump_to 为空。

3、长期记忆

3.1 基本理解

3.1.1 什么是长期记忆

短期记忆记录的是会话级别（线程，Thread）的数据，会话间不共享。

而长期记忆记录的是用户特定或应用级别的数据，任何会话都可以随时访问。

比如：

你喜欢简短回答
你偏好 Python
某个用户是 VIP
某个流程过去怎么做效果更好

这类信息不属于某一条聊天，而属于”用户/组织/应用本身”。

3.1.2 类型划分

LangChain 参考 CoALA paper 将长期记忆划分为三类：

Memory Type	存什么	示例
Semantic（语义记忆）	事实	用户喜欢简洁回答；用户常用中文；某个公司属于哪个行业
Episodic（情景记忆）	经验	过去某个任务是怎么成功的；某种用户输入下，怎样回答效果最好
Procedural（程序性记忆）	规则/做事方法	Agent 的系统提示词；Agent 的工作流程；工具调用规则

类型 1：Semantic Memory（语义记忆）

即”事实类记忆”，记录事实/用户偏好/概念，如：

用户喜欢简洁回答
用户常用中文
某个公司属于哪个行业

类型 2：Episodic Memory（情景记忆）

即”经验类记忆”，记录 Agent 过去执行的动作，如：

过去某个任务是怎么成功的
某种用户输入下，怎样回答效果最好

在 Agent 里，这常常表现为 few-shot examples（少样本示例）：

不直接告诉模型规则，而是给它看几个”输入 -> 输出”的例子，让它照着学

类型 3：Procedural Memory（程序性记忆）

即”规则/做事方法”，如：

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

3.1.3 存储架构

长期记忆的存储是 Store -> Namespace -> Key -> Value 的四层架构。

第 1 层：Store（记忆仓库）

Store 是 langgraph.store.base.BaseStore 的子类实例，由 LangGraph 提供。

常用实现类：

InMemoryStore：将长期记忆存储在内存，适合测试
PostgresStore：将长期记忆存储在外部的 PostgreSQL 数据库，适合生产环境

开发期可用 InMemoryStore；生产建议数据库后端，如 PostgresStore。

第 2 层：Namespace（命名空间）

数据类型是由任意长度的 tuple[str, ...] 表示的层级路径。作用上很像”文件路径/文件夹层级”，用于给长期记忆分组和隔离。数据类型为字符串元组。

第 3 层：Key（键）

是该 namespace 下的唯一标识，单条记忆的唯一键，数据类型为字符串（str）。

第 4 层：Value（值）

是存储的值，数据类型为字典（dict[str, Any]）。

举例 1：

每个 namespace 存储的都是 key-value 键值对，通过 key 可以唯一标识一条 value。

namespace = ("users", "user_123", "preferences")   # 元组类型
key = "profile"                                    # 字符串类型
value = {                                          # 字典类型
    "language": "zh-CN",
    "style": "short_direct",
    "likes": ["python", "rag"]
}
store.put(namespace, key, value)

举例 2：

同一个 AI 应用通常会为每个独立会话维护各自的短期状态 State；而长期记忆通常可以共享同一个 Store 实例，再通过 namespace 区分不同用户、组织、业务域或会话相关数据。如下：

AI 应用
├─ thread_id = t1 -> state_1
│   ├─ messages = [
│   │   {"role": "user", "content": "我想去北京旅游"},
│   │   {"role": "assistant", "content": "你想玩几天？"}
│   │ ]
│   ├─ current_intent = "travel_planning"
│   └─ collected_slots = {"destination": "北京"}
│
├─ thread_id = t2 -> state_2
│   ├─ messages = [
│   │   {"role": "user", "content": "帮我写周报"}
│   │ ]
│   ├─ current_intent = "write_report"
│   └─ collected_slots = {}
│
├─ thread_id = t3 -> state_3
│   ├─ messages = [
│   │   {"role": "user", "content": "我喜欢简洁风格的UI"}
│   │ ]
│   ├─ current_intent = "ui_design"
│   └─ collected_slots = {"style": "minimal"}
│
└─ shared store
    ├─ namespace = (user_1, "memories")
    │   ├─ key = "profile"
    │   │   value = {
    │   │       "name": "张三",
    │   │       "city": "上海",
    │   │       "preferences": ["简洁风格", "中文回复"]
    │   │   }
    │   ├─ key = "travel_preference"
    │   │   value = {
    │   │       "favorite_cities": ["北京", "杭州"],
    │   │       "budget_level": "medium"
    │   │   }
    │   └─ key = "writing_style"
    │       value = {
    │           "tone": "professional",
    │           "language": "zh-CN"
    │       }
    │
    ├─ namespace = (user_2, "memories")
    │   ├─ key = "profile"
    │   │   value = {
    │   │       "name": "李四",
    │   │       "city": "深圳"
    │   │   }
    │   └─ key = "product_interest"
    │       value = {
    │           "topics": ["AI Agent", "RAG", "Workflow"]
    │       }
    │
    └─ namespace = (user_1, thread_3, "artifacts")
        ├─ key = "draft_v1"
        │   value = {
        │       "type": "html",
        │       "content": "<html>...</html>"
        │   }
        ├─ key = "ui_notes"
        │   value = {
        │       "summary": "用户偏好留白多、低饱和配色"
        │   }
        └─ key = "final_scheme"
            value = {
                "palette": ["#F5F1E8", "#1F3A5F", "#D97B2D"],
                "font_style": "clean"
            }

3.2 基础 API 的使用

LangChain 1.2.x 的长期记忆基于 store 持久化数据，相关的 API 有：

put()：负责写入
get()：负责读取
search()：负责检索

我们可以在 Agent 执行流程之外直接访问长期记忆。

3.2.1 `put()` / `get()`：写入/读取 API

① put() 源码剖析

直接调用 put() 函数即可，函数签名如下：

def put(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    index: Literal[False] | list[str] | None = None,
    *,
    ttl: float | None | NotProvided = NOT_PROVIDED,
) -> None:

参数说明：

namespace：文档所在的层级路径
key：该路径下的唯一键
value：要保存的 JSON-like 字典
index：控制语义检索索引
- None（默认选项）：使用 store 初始化时配置的索引配置，如果初始化时没有指定索引策略，则 index 参数将会被忽略
- False：不为该 item 建立语义索引
- list[str]：只对指定字段路径建索引
ttl：可选，过期时间；是否支持取决于具体 store 实现

举例：

store.put(
    ("users", "alice", "memories"),   # namespace
    "pref_food",                       # key
    {"category": "food", "text": "Alice likes sushi"}   # value
)

② get() 源码剖析

按照 namespace + key 精确查询，返回的不止是 value，而是完整对象。即 LangGraph 底层将数据封装为 Item 对象。

函数签名如下：

def get(
    self,
    namespace: tuple[str, ...],
    key: str,
    *,
    refresh_ttl: bool | None = None,
) -> Item | None:

参数说明：

namespace：文档所在的层级路径
key：该路径下的唯一键
refresh_ttl：是否刷新当前 item 的 TTL（time-to-live，存活时间）
- 默认为 None：表示采用创建 store 对象时指定的同名配置
- 如果没有配置 TTL，该参数被忽略

举例：

item = my_store.get(("users", "alice", "memories"), "pref_food")

if item is not None:
    print(item.value)
    # {'category': 'food', 'text': 'Alice likes sushi'}

③ 举例 1：基于 InMemoryStore

添加：

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

namespace = ("users",)
user_id = 'user-1'
username = "小蓝"
store.put(namespace, user_id, {"name": username})

print(store.get(namespace, user_id))

输出：

1	Item(namespace=['users'], key='user-1', value={'name': '小蓝'}, created_at='2026-06-11T15:20:52.542493+00:00', updated_at='2026-06-11T15:20:52.542494+00:00')

注意到，Item 对象新增了 created_at 和 updated_at 字段，分别为数据新增和更改时间。

注意：对于当前版本，InMemoryStore 每次 put 都会创建一个新的 Item 对象，无论 namespace 和 key 是否相同，所以 created_at 和 updated_at 始终是一致的。

更新：

对同一条数据进行更改后查询。

1 2	store.put(namespace, user_id, {"name": '小红'}) print(store.get(namespace, user_id))

输出：

1	Item(namespace=['users'], key='user-1', value={'name': '小红'}, created_at='2026-06-11T15:23:11.283981+00:00', updated_at='2026-06-11T15:23:11.283982+00:00')

④ 举例 2：基于 PostgresStore

PostgresStore 不同于 InMemoryStore，更改数据的逻辑是 UPDATE 而非覆盖，因此 created_at 固定为 Item 创建时间，而 updated_at 为更新时间，二者可能不同，符合直觉。

添加：

from langgraph.store.postgres import PostgresStore

namespace = ("users", )
user_id = "user-11"
username = "小蓝"

DB_URL = "postgresql://langchain_user:abcd1234@118.195.128.47:5432/langchain_db?sslmode=disable"
with PostgresStore.from_conn_string(DB_URL) as store:
    store.setup()
    store.put(namespace, user_id, {"name": username})
    print(store.get(namespace, user_id))

输出：

1	Item(namespace=['users'], key='user-11', value={'name': '小蓝'}, created_at='2026-06-11T23:24:31.132342+08:00', updated_at='2026-06-11T23:24:31.132342+08:00')

说明：这里的 DB_URL 大家测试时，需要替换为自己的。

更新：

更改同一条数据：

with PostgresStore.from_conn_string(DB_URI) as store:
    store.setup()
    store.put(namespace, user_id, {"name": "小红"})
    print(store.get(namespace, user_id))

输出：

1	Item(namespace=['users'], key='user-11', value={'name': '小红'}, created_at='2026-06-11T23:24:31.132342+08:00', updated_at='2026-06-11T23:26:11.956257+08:00')

可以看到，created_at 没变，但 updated_at 更改了。

3.2.3 `search()`：检索 API

① 源码剖析

函数签名如下：

def search(
    self,
    namespace_prefix: tuple[str, ...],
    /,
    *,
    query: str | None = None,
    filter: dict[str, Any] | None = None,
    limit: int = 10,
    offset: int = 0,
    refresh_ttl: bool | None = None,
) -> list[SearchItem]:

参数说明：

namespace_prefix：命名空间前缀，在该前缀下搜索。
query：语义检索时用于查询的自然语言。
filter：过滤条件，value 中的键值对组合，见下文举例。
limit：可以返回 item 的最大条数，效果等同于 SQL 中的 LIMIT。
offset：返回结果之前跳过的 item 数量。
refresh_ttl：同上。

它支持两种检索方式（对应上面的参数 2、3）：

按 filter 做结构化过滤，即用 value 中的键值筛选符合条件的记录。
按 query 做语义相似度检索，需要将输入转换为向量。

返回值：

返回匹配的 SearchItem 列表，并额外携带匹配分数等检索元信息。

② 举例 1：按照 namespace 前缀搜索

本节全部基于 InMemoryStore 测试。

1、准备 store

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

namespace1 = ("users", "Alice", "memories")
key1 = 'preferences'
value1 = {
    "course": "计算机组成原理",
    "sports": "跑步",
    "food": "紫光园奶皮子酸奶"
}

namespace2 = ("users", "Bob", "memories")
key2 = 'preferences'
value2 = {
    "course": "数字电路与模拟电路",
    "sports": "跑步",
    "food": "奶皮子糖葫芦"
}

namespace3 = ("users", "Black", "memories")
key3 = 'preferences'
value3 = {
    "course": "数字电路与模拟电路",
    "sports": "羽毛球",
    "food": "紫光园奶皮子酸奶"
}

store.put(namespace1, key1, value1)
store.put(namespace2, key2, value2)
store.put(namespace3, key3, value3)

如果在 Jupyter 环境，执行上述代码后，直接执行后续代码即可。

如果每次测试都在单独的 Python 文件中进行，则以下 1、2 小节的案例代码之前都要补充上述代码。

2、按照 namespace 前缀搜索

代码 1：

1
2
3

print('=' * 30, '-> (users, ) <-', "=" * 30)
for item in store.search(("users", )):
    print(item)

输出：

============================== -> (users, ) <- ==============================
Item(namespace=['users', 'Alice', 'memories'], key='preferences', value={'course': '计算机组成原理', 'sports': '跑步', 'food': '紫光园奶皮子酸奶'}, created_at='2026-06-11T15:27:24.734788+00:00', updated_at='2026-06-11T15:27:24.734789+00:00', score=None)
Item(namespace=['users', 'Bob', 'memories'], key='preferences', value={'course': '数字电路与模拟电路', 'sports': '跑步', 'food': '奶皮子糖葫芦'}, created_at='2026-06-11T15:27:24.734809+00:00', updated_at='2026-06-11T15:27:24.734810+00:00', score=None)
Item(namespace=['users', 'Black', 'memories'], key='preferences', value={'course': '数字电路与模拟电路', 'sports': '羽毛球', 'food': '紫光园奶皮子酸奶'}, created_at='2026-06-11T15:27:24.734829+00:00', updated_at='2026-06-11T15:27:24.734829+00:00', score=None)

代码 2：

1
2
3

print('=' * 30, '-> (users, Alice) <-', "=" * 30)
for item in store.search(("users", "Alice")):
    print(item)

输出：

1
2

============================== -> (users, Alice) <- ==============================
Item(namespace=['users', 'Alice', 'memories'], key='preferences', value={'course': '计算机组成原理', 'sports': '跑步', 'food': '紫光园奶皮子酸奶'}, created_at='2026-06-11T15:27:24.734788+00:00', updated_at='2026-06-11T15:27:24.734789+00:00', score=None)

③ 举例 2：按照 filter 过滤

代码 1：

1
2
3

print("=" * 30, '-> (users, ), filter=food <-", "=" * 30)
for item in store.search(("users", ), filter={"food": "紫光园奶皮子酸奶"}):
    print(item)

输出：

1
2
3

============================== -> (users, ), filter=food <-", "=" * 30)
Item(namespace=['users', 'Alice', 'memories'], key='preferences', value={'course': '计算机组成原理', 'sports': '跑步', 'food': '紫光园奶皮子酸奶'}, created_at='2026-06-11T15:27:24.734788+00:00', updated_at='2026-06-11T15:27:24.734789+00:00', score=None)
Item(namespace=['users', 'Black', 'memories'], key='preferences', value={'course': '数字电路与模拟电路', 'sports': '羽毛球', 'food': '紫光园奶皮子酸奶'}, created_at='2026-06-11T15:27:24.734829+00:00', updated_at='2026-06-11T15:27:24.734829+00:00', score=None)

代码 2：

1
2
3

print("=" * 30, '-> (users, ), filter=sports <-", "=" * 30)
for item in store.search(("users", ), filter={"sports": "跑步"}):
    print(item)

输出：

1
2
3

============================== -> (users, ), filter=sports <-", "=" * 30)
Item(namespace=['users', 'Alice', 'memories'], key='preferences', value={'course': '计算机组成原理', 'sports': '跑步', 'food': '紫光园奶皮子酸奶'}, created_at='2026-06-11T15:27:24.734788+00:00', updated_at='2026-06-11T15:27:24.734789+00:00', score=None)
Item(namespace=['users', 'Bob', 'memories'], key='preferences', value={'course': '数字电路与模拟电路', 'sports': '跑步', 'food': '奶皮子糖葫芦'}, created_at='2026-06-11T15:27:24.734809+00:00', updated_at='2026-06-11T15:27:24.734810+00:00', score=None)

代码 3：

1
2
3

print("=" * 30, '-> (users, ), filter=course <-", "=" * 30)
for item in store.search(("users", ), filter={"course": "数字电路与模拟电路"}):
    print(item)

输出：

1
2
3

============================== -> (users, ), filter=course <-", "=" * 30)
Item(namespace=['users', 'Bob', 'memories'], key='preferences', value={'course': '数字电路与模拟电路', 'sports': '跑步', 'food': '奶皮子糖葫芦'}, created_at='2026-06-11T15:27:24.734809+00:00', updated_at='2026-06-11T15:27:24.734810+00:00', score=None)
Item(namespace=['users', 'Black', 'memories'], key='preferences', value={'course': '数字电路与模拟电路', 'sports': '羽毛球', 'food': '紫光园奶皮子酸奶'}, created_at='2026-06-11T15:27:24.734829+00:00', updated_at='2026-06-11T15:27:24.734829+00:00', score=None)

④ 举例 3：按照语义搜索

举例 1：自定义嵌入函数

自定义嵌入函数，目的是查看嵌入向量。

from langgraph.store.memory import InMemoryStore

# 自定义嵌入函数
def embed(text: list[str]) -> list[list[float]]:
    return [[1.0] * 6 for _ in range(len(text))]

index_config = {
    "embed": embed,
    "dims": 6,
    "fields": ["$", "course"]
}
store = InMemoryStore(
    index=index_config
)

namespace1 = ("users", "Alice", "memories")
key1 = 'preferences'
value1 = {
    "course": "计算机组成原理",
    "sports": "跑步",
    "food": "紫光园奶皮子酸奶"
}

namespace2 = ("users", "Bob", "memories")
key2 = 'preferences'
value2 = {
    "course": "数字电路与模拟电路",
    "sports": "跑步",
    "food": "奶皮子糖葫芦"
}

namespace3 = ("users", "Black", "memories")
key3 = 'preferences'
value3 = {
    "course": "数字电路与模拟电路",
    "sports": "羽毛球",
    "food": "紫光园奶皮子酸奶"
}

store.put(namespace1, key1, value1)
store.put(namespace2, key2, value2)
store.put(namespace3, key3, value3)

初始化 Store 时通过 index 指定索引方式，后者源码如下：

class IndexConfig(TypedDict, total=False):
    dims: int
    embed: Embeddings | EmbeddingsFunc | AEmbeddingsFunc | str
    fields: list[str] | None

embeds：将输入文本转换为向量的嵌入函数，可以是自定义函数，也可以是嵌入模型对象，本例传递的是自定义嵌入函数。
dims：输出向量维度。
fields：用于计算向量的属性列表，这里的属性都是指 value 中的 key，value 是一个 JSON-like 字典，可取值如下：
- ["$"]：将 value 作为整体嵌入
- ["fields1", "fields2"]：单独指定某个一级字段
- ["parent.child"]：从内部的嵌套 JSON 对象中获取子字段的值
- ["array[*].field"]：从 JSON 数组的每个 JSON 对象中获取子字段的值
- 注意：上述四种形式可以同时出现，fields 列表的每个元素都会生成一个嵌入向量。

查看嵌入向量：

1 2	from pprint import pprint pprint(store._vectors)

输出：

defaultdict(<function InMemoryStore.__init__.<locals>.<lambda> at 0x000001C201898A40>,
            {('users', 'Alice', 'memories'): defaultdict(<class 'dict'>,
                {'preferences': {'$': [1.0, 1.0, 1.0, 1.0, 1.0, 1.0],
                                 'course': [1.0, 1.0, 1.0, 1.0, 1.0, 1.0]}}),
             ('users', 'Black', 'memories'): defaultdict(<class 'dict'>,
                {'preferences': {'$': [1.0, 1.0, 1.0, 1.0, 1.0, 1.0],
                                 'course': [1.0, 1.0, 1.0, 1.0, 1.0, 1.0]}}),
             ('users', 'Bob', 'memories'): defaultdict(<class 'dict'>,
                {'preferences': {'$': [1.0, 1.0, 1.0, 1.0, 1.0, 1.0],
                                 'course': [1.0, 1.0, 1.0, 1.0, 1.0, 1.0]}})})

还可以通过指定 namespace、key、index_config 的 fields 中指定的字段名查看特定向量。

1 2	from pprint import pprint pprint(store._vectors[('users', 'Alice', 'memories')]['preferences']['$'])

输出：

1	[1.0, 1.0, 1.0, 1.0, 1.0, 1.0]

举例 2：使用嵌入模型

本例要通过 CloseAI 平台调用 OpenAI 的嵌入模型 openai:text-embedding-3-large。

其嵌入维度为 3072，所以此处的 dims 应设置为 3072。

from langgraph.store.memory import InMemoryStore
from dotenv import load_dotenv
from langchain.embeddings import init_embeddings

import os
load_dotenv(override=True)

embedding_model = init_embeddings(
    model="openai:text-embedding-3-large",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)

index_config = {
    "embed": embedding_model,
    "dims": 3072,
    "fields": ["$"]
}
store = InMemoryStore(
    index=index_config
)

namespace1 = ("users", "Alice", "memories")
key1 = 'preferences'
value1 = {
    "course": "计算机组成原理",
    "sports": "跑步",
    "food": "紫光园奶皮子酸奶"
}

namespace2 = ("users", "Bob", "memories")
key2 = 'preferences'
value2 = {
    "course": "数字电路与模拟电路",
    "sports": "跑步",
    "food": "奶皮子糖葫芦"
}

namespace3 = ("users", "Black", "memories")
key3 = 'preferences'
value3 = {
    "course": "数字电路与模拟电路",
    "sports": "羽毛球",
    "food": "紫光园奶皮子酸奶"
}

store.put(namespace1, key1, value1)
store.put(namespace2, key2, value2)
store.put(namespace3, key3, value3)

1 2	for item in store.search(("users", ), query="数电模电"): print(item)

输出：

1
2
3

Item(namespace=['users', 'Bob', 'memories'], key='preferences', value={'course': '数字电路与模拟电路', 'sports': '跑步', 'food': '奶皮子糖葫芦'}, created_at='2026-06-11T15:45:13.161389+00:00', updated_at='2026-06-11T15:45:13.161391+00:00', score=0.22494289397943232)
Item(namespace=['users', 'Black', 'memories'], key='preferences', value={'course': '数字电路与模拟电路', 'sports': '羽毛球', 'food': '紫光园奶皮子酸奶'}, created_at='2026-06-11T15:45:13.748504+00:00', updated_at='2026-06-11T15:45:13.748506+00:00', score=0.21367212817763243)
Item(namespace=['users', 'Alice', 'memories'], key='preferences', value={'course': '计算机组成原理', 'sports': '跑步', 'food': '紫光园奶皮子酸奶'}, created_at='2026-06-11T15:45:12.599571+00:00', updated_at='2026-06-11T15:45:12.599573+00:00', score=0.1253981117634029)

注意：如果只是指定了 query，返回的是所有 namespace 前缀满足要求的 item。底层会按照向量相似度计算查询和候选的 score，输出结果按照 score 降序排列。可以结合 limit 或 filter 限制数据条数。这里不再演示。

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

我们可以在工具或中间件中访问长期记忆。

3.3.1 在工具中访问长期记忆

① 基于 InMemoryStore

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 从 .env 文件中加载环境变量
load_dotenv(override=True)

model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

from langchain_core.messages import HumanMessage
from typing import NotRequired

from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

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

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

    Args:
        name: 用户名

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

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

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

agent = create_agent(
    model=model,
    tools=[save_user_info, get_user_info],
    store=store,
    system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
    state_schema=CustomState,
)

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()

其中，CustomState 扩展了 Agent 的标准状态。除了默认的 messages（历史消息列表）之外，还额外增加了一个 user_id 字段。这样，Agent 在运行过程中随时随地都能知道当前和它说话的用户 ID 是什么。

输出：

============================== -> 第一个会话（线程） <- ==============================
================================ Human Message =================================

你好，很高兴认识你，我是小花
================================== Ai Message ==================================
Tool Calls:
  save_user_info (call_dUFA22P95tnIrBc0fi2I0WLF)
  Call ID: call_dUFA22P95tnIrBc0fi2I0WLF
  Args:
    name: 小花
================================= Tool Message =================================
Name: save_user_info

saved
================================== Ai Message ==================================

你好，小花，很高兴认识你！
============================== -> 第二个会话（线程） <- ==============================
================================ Human Message =================================

我是谁
================================== Ai Message ==================================
Tool Calls:
  get_user_info (call_cARiaT5NR0kUZDbMOpy0khqA)
  Call ID: call_cARiaT5NR0kUZDbMOpy0khqA
  Args:
================================= Tool Message =================================
Name: get_user_info

{'name': '小花'}
================================== Ai Message ==================================

你是小花。

两次 invoke 没有通过 config 串联起来，是两个独立的会话，但第二个会话可以访问第一个会话写入长期记忆的内容。

② 基于 PostgresStore

from typing import NotRequired

from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime
from langgraph.store.postgres import PostgresStore

DB_URI = "postgresql://langchain_user:abcd1234@118.195.128.47:5432/langchain_db?sslmode=disable"

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

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

    Args:
        name: 用户名

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

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

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

with PostgresStore.from_conn_string(DB_URI) as store:
    store.setup()
    agent = create_agent(
        model=model,
        tools=[save_user_info, get_user_info],
        store=store,
        system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
        state_schema=CustomState,
    )

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

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

输出：

============================== -> 第一个会话（线程） <- ==============================
================================ Human Message =================================

你好，很高兴认识你，我是小花
================================== Ai Message ==================================
Tool Calls:
  save_user_info (call_CmfPqf4aahyt5LQIgcaCSTDy)
  Call ID: call_CmfPqf4aahyt5LQIgcaCSTDy
  Args:
    name: 小花
================================= Tool Message =================================
Name: save_user_info

saved
================================== Ai Message ==================================

你好，小花，很高兴认识你！有什么我可以帮你的吗？
============================== -> 第二个会话（线程） <- ==============================
================================ Human Message =================================

我是谁
================================== Ai Message ==================================
Tool Calls:
  get_user_info (call_x6DPJGLsqwjMgRoXd62bD7ZZ)
  Call ID: call_x6DPJGLsqwjMgRoXd62bD7ZZ
  Args:
================================= Tool Message =================================
Name: get_user_info

{'name': '小花'}
================================== Ai Message ==================================

你是小花。

查看 PostgreSQL 数据表：

打开 Xshell 客户端：

ubuntu@VM-0-6-ubuntu:~$ psql postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable
psql (16.13 (Ubuntu 16.13-0ubuntu0.24.04.1))
Type "help" for help.

langgraph_db=>

查看表：

1	langgraph_db=> \dt

List of relations
 Schema |         Name          | Type  |     Owner
--------+-----------------------+-------+----------------
 public | checkpoint_blobs      | table | langgraph_user
 public | checkpoint_migrations | table | langgraph_user
 public | checkpoint_writes     | table | langgraph_user
 public | checkpoints           | table | langgraph_user
 public | store                 | table | langgraph_user
 public | store_migrations      | table | langgraph_user
(6 rows)

新增两张 Store 相关的表：store 和 store_migrations。

3.3.2 在中间件中访问长期记忆

① Node-style hooks 中访问

以 before_model 为例，其钩子函数签名如下：

1	def before_model(self, state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] \| None:

Runtime 定义如下：

@dataclass(**_DC_KWARGS)
class Runtime(Generic[ContextT]):
    context: ContextT = field(default=None)   # type: ignore[assignment]
    """Static context for the graph run, like `user_id`, `db_conn`, etc.

    Can also be thought of as 'run dependencies'."""

    store: BaseStore | None = field(default=None)
    """Store for the graph run, enabling persistence and memory."""

    stream_writer: StreamWriter = field(default=_no_op_stream_writer)
    """Function that writes to the custom stream."""

    previous: Any = field(default=None)
    """The previous return value for the given thread.

    Only available with the functional API when a checkpointer is provided.
    """
    ...

所以，我们可以通过 runtime.store 在中间件中访问长期记忆。

② Wrap-style hooks 中访问

1. wrap_model_call

钩子函数签名如下：

def wrap_model_call(
    self,
    request: ModelRequest[ContextT],
    handler: Callable[[ModelRequest[ContextT]], ModelResponse[ResponseT]],
) -> ModelResponse[ResponseT] | AIMessage | ExtendedModelResponse[ResponseT]:

ModelRequest 定义如下：

@dataclass(init=False)
class ModelRequest(Generic[ContextT]):
    """Model request information for the agent.

    Type Parameters:
        ContextT: The type of the runtime context. Defaults to `None` if not specified.
    """

    model: BaseChatModel
    messages: list[AnyMessage]   # excluding system message
    system_message: SystemMessage | None
    tool_choice: Any | None
    tools: list[BaseTool | dict[str, Any]]
    response_format: ResponseFormat[Any] | None
    state: AgentState[Any]
    runtime: Runtime[ContextT]
    model_settings: dict[str, Any] = field(default_factory=dict)
    ...

所以，可以通过 request.runtime.store 访问长期记忆。

2. wrap_tool_call

钩子函数签名如下：

def wrap_tool_call(
    self,
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
) -> ToolMessage | Command[Any]:

ToolCallRequest 定义如下：

@dataclass
class ToolCallRequest:
    """Tool execution request passed to tool call interceptors.

    Attributes:
        tool_call: Tool call dict with name, args, and id from model output.
        tool: BaseTool instance to be invoked, or None if tool is not
            registered with the `ToolNode`. When tool is `None`, interceptors can
            handle the request without validation. If the interceptor calls `execute()`,
            validation will occur and raise an error for unregistered tools.
        state: Agent state (`dict`, `list`, or `BaseModel`).
        runtime: LangGraph runtime context (optional, `None` if outside graph).
    """

    tool_call: ToolCall
    tool: BaseTool | None
    state: Any
    runtime: ToolRuntime
    ...

ToolRuntime 定义如下：

@dataclass
class ToolRuntime(_DirectlyInjectedToolArg, Generic[ContextT, StateT]):
    state: StateT
    context: ContextT
    config: RunnableConfig
    stream_writer: StreamWriter
    tool_call_id: str | None
    store: BaseStore | None

所以，可以通过 request.runtime.store 访问长期记忆。

3.4 何时写入记忆

官方介绍了两种方式。

1. 在主流程里写（hot path）

也就是：用户发消息，AI 一边回答，一边决定要不要记下来。

优点：

立即生效
下一轮马上能用
用户可感知，透明

缺点：

增加延迟
逻辑变复杂

2. 在后台写（background）

就是先回答用户，记忆整理放到后台异步做。

优点：

主流程更快
记忆逻辑更独立
更适合批量整理

缺点：

不能立刻生效
要决定多久整理一次
触发时机不好选

工程上通常这么选：

用户偏好、账号资料：可热路径写
对话摘要、经验沉淀、行为分析：更适合后台写

4、课后阅读：Static runtime Context

静态运行时上下文（Static runtime Context） 表示不可变的数据，如用户元数据、工具和传递给应用程序的数据库连接对象。通常在运行开始时通过 invoke / stream 的 context 参数传递，此类数据在运行期间不会更改。

4.1 中间件中访问

4.1.1 Node-style hooks

此类钩子都通过 runtime.context 访问上下文对象。

用户自定义 ContextSchema 用 @dataclass 修饰，在 Agent 创建时通过 state_schema 参数传递。

本例从长期记忆中查询用户额度，如果额度用尽则中断流程。

代码：

from dataclasses import dataclass
from typing import Any

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langchain.messages import AIMessage

from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore
from dotenv import load_dotenv
from loguru import logger

load_dotenv()

store = InMemoryStore()
namespace = ("users", "credits")
key1 = 'Ada Lovelace'
value1 = {
    "tokens_credit_left": 5000,
    "user_level": 5
}
key2 = 'Blackwell'
value2 = {
    "tokens_credit_left": 2999,
    "user_level": 5
}
key3 = 'Ampere'
value3 = {
    "tokens_credit_left": 1000,
    "user_level": 5
}
store.put(namespace, key1, value1)
store.put(namespace, key2, value2)
store.put(namespace, key3, value3)

@dataclass
class UserContext:
    username: str

class CheckCredit(AgentMiddleware):
    @hook_config(can_jump_to=["end"])
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        store = runtime.store
        context = runtime.context

        username = context.username
        credit = store.get(namespace, username)
        if not credit:
            return {
                "jump_to": "end",
                "messages": AIMessage("您尚未注册~")
            }

        if credit.value["tokens_credit_left"] < 3000:
            return {
                "jump_to": "end",
                "messages": AIMessage(f"{username}额度不足，请充值")
            }

        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        store = runtime.store
        context = runtime.context

        username = context.username
        credit = store.get(namespace, username)
        usage_metadata = state["messages"][-1].usage_metadata
        token_usage = usage_metadata["input_tokens"] + usage_metadata["output_tokens"] * 6
        credit.value["tokens_credit_left"] -= token_usage

        # 更新长期记忆
        store.put(namespace, username, credit.value)

        return None

    def after_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        store = runtime.store
        context = runtime.context

        username = context.username
        item = store.get(namespace, username)
        credit = item.value["tokens_credit_left"]

        logger.info(f"{username} 当前剩余额度：{credit}")
        return None

agent = create_agent(
    model="deepseek-chat",
    middleware=[CheckCredit()],
    store=store,
    context_schema=UserContext
)

response1 = agent.invoke(
    {"messages": ["你好啊，你知道 Ada Lovelace 的贡献吗？"]},
    context=UserContext(username="Ada Lovelace")
)
print("=" * 30, '-> Ada Lovelace messages <-', '=' * 30)
for msg in response1["messages"]:
    msg.pretty_print()

response2 = agent.invoke(
    {"messages": ["你好啊，你知道 Blackwell 的贡献吗？"]},
    context=UserContext(username="Blackwell")
)
print("=" * 30, '-> Blackwell messages <-', '=' * 30)
for msg in response2["messages"]:
    msg.pretty_print()

输出：

2026-04-14 11:39:18.876 | INFO     | __main__:after_agent:85 - Ada Lovelace 当前剩余额度：2721
2026-04-14 11:39:18.881 | INFO     | __main__:after_agent:85 - Blackwell 当前剩余额度：2999

============================== -> Ada Lovelace messages <- ==============================
================================ Human Message =================================

你好啊，你知道 Ada Lovelace 的贡献吗？
================================== Ai Message ==================================

是的，Ada Lovelace（1815-1852）被广泛认为是世界上第一位计算机程序员，她的贡献主要体现在以下几个方面：

---

### 1. 翻译与注释：解析"分析机"的潜力
- 1843年，她翻译了意大利工程师路易吉·费德里科·米纳布里关于查尔斯·巴贝奇"分析机"（早期机械通用计算机）的论文，并添加了大量自己的注释，篇幅远超原文。
- 她的注释中包含了**世界上第一个计算机算法**，用于计算伯努利数，这被视为最早的"计算机程序"概念。

---

### 2. 超越计算的前瞻性思想
- 她指出分析机不仅限于数值计算，还能处理符号、音乐甚至艺术，预见了计算机的**通用性**和**创造性潜力**。
- 她强调"机器只能执行指令，不能自主思考"，区分了人工智能与人类智能的界限。

---

### 3. 技术与人文的桥梁
- 作为诗人拜伦的女儿，她将数学逻辑与想象力结合，主张科技与艺术的融合，这一思想对后世计算科学影响深远。

---

### 4. 历史意义
- 尽管分析机在她有生之年未能建成，但她的工作在一个世纪后启发了艾伦·图灵等计算机科学先驱。
- 1980年，美国国防部将一种编程语言命名为"Ada"以纪念她。

---

### 争议与澄清
- 部分学者认为她的贡献被过度神话，但主流观点仍肯定她首次系统阐述了编程逻辑的核心思想，尤其在19世纪女性科学参与受限的背景下，她的成就尤为突出。

---

Ada Lovelace 的故事不仅是技术史上的里程碑，也象征着跨学科创新如何推动人类认知边界。如果需要更具体的细节，我可以进一步展开！
============================== -> Blackwell messages <- ==============================
================================ Human Message =================================

你好啊，你知道 Blackwell 的贡献吗？
================================== Ai Message ==================================

Blackwell额度不足，请充值

4.1.2 Wrap-style hooks

① wrap_model_call

本例通过 context 中记录的用户身份信息调整暴露给模型的工具集。

此处更改模型请求的方法是 transient request update。

from dataclasses import dataclass
from typing import Callable

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelResponse, ModelRequest
from langchain.messages import AIMessage
from langchain.tools import tool

from langgraph.store.memory import InMemoryStore
from dotenv import load_dotenv
from loguru import logger

load_dotenv()

store = InMemoryStore()
namespace = ("users", "credits")
key1 = 'Ada Lovelace'
value1 = {
    "tokens_left": 3000,   # 剩余大模型token额度
    "get_weather": "yes",  # 是否认证天气查询服务
    "get_news": "no"       # 是否认证网络爬虫服务，用于查询新闻
}
key2 = 'Blackwell'
value2 = {
    "tokens_left": 2000,
    "get_weather": "no",
    "get_news": "yes"
}
key3 = 'Ampere'
value3 = {
    "tokens_left": 3000,
    "get_weather": "yes",
    "get_news": "yes"
}
store.put(namespace, key1, value1)
store.put(namespace, key2, value2)
store.put(namespace, key3, value3)

@tool
def get_weather(city: str):
    """
    查询指定城市当日天气

    Args:
        city: 城市名称
    """
    return f"{city} 今天天气不错"

@tool()
def get_news():
    """查询当日新闻"""
    return "美伊尚未达成停战协议"

@dataclass
class UserContext:
    username: str

class CheckCredit(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        context = request.runtime.context
        store = request.runtime.store

        value = store.get(namespace, context.username).value
        tools = []

        for tool in request.tools:
            tool_name = tool.name
            if value[tool_name] == "yes":
                tools.append(tool)
            else:
                logger.warning(f"{context.username} 无权调用 {tool_name}")

        # 更改调用请求携带的工具集，仅本次生效
        request = request.override(tools=tools)

        return handler(request)

agent = create_agent(
    model="deepseek-chat",
    middleware=[CheckCredit()],
    tools=[get_weather, get_news],
    store=store,
    context_schema=UserContext
)

response1 = agent.invoke(
    {"messages": ["你好啊，今天北京天气如何，今日有哪些新闻？"]},
    context=UserContext(username="Ada Lovelace")
)
print("=" * 30, '-> Ada Lovelace messages <-', '=' * 30)
for msg in response1["messages"]:
    msg.pretty_print()

response2 = agent.invoke(
    {"messages": ["你好啊，今天北京天气如何，今日有哪些新闻？"]},
    context=UserContext(username="Blackwell")
)
print("=" * 30, '-> Blackwell messages <-', '=' * 30)
for msg in response2["messages"]:
    msg.pretty_print()

response3 = agent.invoke(
    {"messages": ["你好啊，今天北京天气如何，今日有哪些新闻？"]},
    context=UserContext(username="Ampere")
)
print("=" * 30, '-> Ampere messages <-', '=' * 30)
for msg in response3["messages"]:
    msg.pretty_print()

输出：

2026-04-14 11:40:07.527 | WARNING  | __main__:wrap_model_call:75 - Ada Lovelace 无权调用 get_news
2026-04-14 11:40:11.357 | WARNING  | __main__:wrap_model_call:75 - Ada Lovelace 无权调用 get_news
2026-04-14 11:40:15.049 | WARNING  | __main__:wrap_model_call:75 - Blackwell 无权调用 get_weather

============================== -> Ada Lovelace messages <- ==============================
================================ Human Message =================================

你好啊，今天北京天气如何，今日有哪些新闻？
================================== Ai Message ==================================

我来帮您查询北京的天气情况。不过关于今日新闻，我目前没有相关的查询功能。
Tool Calls:
  get_weather (call_00_OplicbLHGEaTanH3yaTDuiMp)
  Call ID: call_00_OplicbLHGEaTanH3yaTDuiMp
  Args:
    city: 北京
================================= Tool Message =================================
Name: get_weather

北京 今天天气不错
================================== Ai Message ==================================

根据查询结果，北京今天天气不错！

关于今日新闻，我目前无法为您提供实时的新闻信息。您可以：
1. 查看各大新闻网站或新闻应用
2. 关注官方新闻发布平台
3. 使用专门的新闻查询工具

如果您需要了解其他城市的天气情况，我很乐意为您查询。

2026-04-14 11:40:17.903 | WARNING  | __main__:wrap_model_call:75 - Blackwell 无权调用 get_weather

============================== -> Blackwell messages <- ==============================
================================ Human Message =================================

你好啊，今天北京天气如何，今日有哪些新闻？
================================== Ai Message ==================================

我目前无法查询天气信息，不过我可以帮您查询今天的新闻。让我为您获取一下今日的新闻内容。
Tool Calls:
  get_news (call_00_thc8aLWFXkeffY1vafIwMJLk)
  Call ID: call_00_thc8aLWFXkeffY1vafIwMJLk
  Args:
================================= Tool Message =================================
Name: get_news

美伊尚未达成停战协议
================================== Ai Message ==================================

根据查询结果，今天有一条重要新闻：**美伊尚未达成停战协议**。

关于北京的天气情况，我目前无法为您提供实时天气信息。建议您可以通过以下方式获取准确的北京天气信息：
1. 查看天气预报应用或网站
2. 使用搜索引擎搜索"北京天气"
3. 查看手机自带的天气应用

如果您还想了解其他新闻或者有其他问题，我很乐意为您提供帮助！
============================== -> Ampere messages <- ==============================
================================ Human Message =================================

你好啊，今天北京天气如何，今日有哪些新闻？
================================== Ai Message ==================================

我来帮您查询北京的天气和今日新闻。
Tool Calls:
  get_weather (call_00_ZyVpPMawPgdFdv49ojD3zIFt)
  Call ID: call_00_ZyVpPMawPgdFdv49ojD3zIFt
  Args:
    city: 北京
  get_news (call_01_kZaouXjSdhjrJGmzGWRbP3so)
  Call ID: call_01_kZaouXjSdhjrJGmzGWRbP3so
  Args:
================================= Tool Message =================================
Name: get_weather

北京 今天天气不错
================================= Tool Message =================================
Name: get_news

美伊尚未达成停战协议
================================== Ai Message ==================================

根据查询结果：

**北京天气：**
今天北京天气不错。

**今日新闻：**
- 美伊尚未达成停战协议

如果您需要更详细的天气信息（比如具体温度、湿度等）或者想了解其他新闻，请告诉我！

② wrap_tool_call

在上述案例的基础上，我们补充额度校验和长期记忆更新。

from dataclasses import dataclass
from typing import Callable, Any

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelResponse, ModelRequest, ToolCallRequest
from langchain.messages import AIMessage, ToolMessage
from langchain.tools import tool

from langgraph.store.memory import InMemoryStore
from langgraph.types import Command
from dotenv import load_dotenv
from loguru import logger

load_dotenv()

store = InMemoryStore()
namespace = ("users", "credits")
key1 = 'Ada Lovelace'
value1 = {
    "tokens_left": 3000,   # 剩余大模型token额度
    "get_weather": "yes",  # 是否认证天气查询服务
    "get_news": "no"       # 是否认证网络爬虫服务，用于查询新闻
}
key2 = 'Blackwell'
value2 = {
    "tokens_left": 2000,
    "get_weather": "no",
    "get_news": "yes"
}
key3 = 'Ampere'
value3 = {
    "tokens_left": 3000,
    "get_weather": "yes",
    "get_news": "yes"
}
store.put(namespace, key1, value1)
store.put(namespace, key2, value2)
store.put(namespace, key3, value3)

# 工具额度消耗
CREDIT_MAP = {
    "get_weather": 10,
    "get_news": 20
}

@tool
def get_weather(city: str):
    """
    查询指定城市当日天气

    Args:
        city: 城市名称
    """
    return f"{city} 今天天气不错"

@tool()
def get_news():
    """查询当日新闻"""
    return "美伊尚未达成停战协议"

@dataclass
class UserContext:
    username: str

class CheckCredit(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        context = request.runtime.context
        store = request.runtime.store

        value = store.get(namespace, context.username).value
        tools = []

        for tool in request.tools:
            tool_name = tool.name
            if value[tool_name] == "yes":
                tools.append(tool)
            else:
                logger.warning(f"{context.username} 无权调用 {tool_name}")

        # 更改调用请求携带的工具集，仅本次生效
        request = request.override(tools=tools)

        return handler(request)

class ToolGuard(AgentMiddleware):
    def wrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
    ) -> ToolMessage | Command[Any]:
        store = request.runtime.store
        context = request.runtime.context

        username = context.username
        value = store.get(namespace, username).value
        tool_name = request.tool.name

        # 本次调用花费
        credits_cost = CREDIT_MAP.get(tool_name)
        # 剩余额度
        credits_left = value["tokens_left"]

        if credits_left < credits_cost:
            logger.warning(f"{username} 额度不足，{tool_name} 调用失败！")
            return Command(
                update={
                    "messages": [
                        ToolMessage(
                            content="额度不足，无法调用工具，请充值~",
                            tool_call_id=request.runtime.tool_call_id
                        )
                    ]
                }
            )

        # 更新剩余额度
        credits_left -= credits_cost
        value["tokens_left"] = credits_left
        store.put(namespace, username, value)

        logger.info(f"{username} 调用 {tool_name} 花费 {credits_cost}，剩余额度 {credits_left}")
        logger.info(f"{username} 状态已更新，当前状态 {store.get(namespace, username).value}")

        return handler(request)

agent = create_agent(
    model="deepseek-chat",
    middleware=[CheckCredit(), ToolGuard()],
    tools=[get_weather, get_news],
    store=store,
    context_schema=UserContext
)

response = agent.invoke(
    {"messages": ["你好啊，今天北京天气如何，今日有哪些新闻？"]},
    context=UserContext(username="Ada Lovelace")
)
print("=" * 30, '-> Ada Lovelace messages <-', '=' * 30)
for msg in response["messages"]:
    msg.pretty_print()

输出：

2026-04-14 11:41:20.386 | WARNING  | __main__:wrap_model_call:82 - Ada Lovelace 无权调用 get_news
2026-04-14 11:41:23.859 | INFO     | __main__:wrap_tool_call:124 - Ada Lovelace 调用 get_weather 花费 10，剩余额度 2990
2026-04-14 11:41:23.860 | INFO     | __main__:wrap_tool_call:125 - Ada Lovelace 状态已更新，当前状态 {'tokens_left': 2990, 'get_weather': 'yes', 'get_news': 'no'}
2026-04-14 11:41:23.865 | WARNING  | __main__:wrap_model_call:82 - Ada Lovelace 无权调用 get_news

============================== -> Ada Lovelace messages <- ==============================
================================ Human Message =================================

你好啊，今天北京天气如何，今日有哪些新闻？
================================== Ai Message ==================================

我来帮您查询北京的天气情况。不过需要说明的是，我目前只能查询天气信息，无法提供新闻资讯。
Tool Calls:
  get_weather (call_00_HY2jENXsRyK21gITiQqeVSk9)
  Call ID: call_00_HY2jENXsRyK21gITiQqeVSk9
  Args:
    city: 北京
================================= Tool Message =================================
Name: get_weather

北京 今天天气不错
================================== Ai Message ==================================

根据查询结果，北京今天天气不错。

关于新闻资讯，我目前没有相关的查询功能。如果您想了解今日新闻，建议您：
1. 查看主流新闻网站或APP
2. 关注官方新闻媒体
3. 使用专门的新闻搜索工具

如果您还需要了解其他城市的天气情况，我很乐意为您查询。

4.1.3 特殊的便捷中间件 `dynamic_prompt`

如前所述，@dynamic_prompt 底层实现机制和通用钩子函数是统一的，对于动态提示词处理提供了更加便捷的入口。

本例通过静态上下文中携带的用户信息，动态更改系统提示词。

from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest

from langgraph.store.memory import InMemoryStore
from dotenv import load_dotenv
from loguru import logger

load_dotenv()

# 准备历史记忆
namespace = ("users", "preferences")
key1 = "Ada Lovelace"
value1 = {
    "chat_preferences": [
        "不喜欢啰嗦",
        "尽可能用最精简的文字解释清楚"
    ]
}

key2 = "Blackwell"
value2 = {
    "chat_preferences": [
        "喜欢长篇大论、引经据典",
        "不知道的东西要明确说'不知道'，不能胡编乱造"
    ]
}

store = InMemoryStore()
store.put(namespace, key1, value1)
store.put(namespace, key2, value2)

@dataclass
class UserContext:
    username: str

@dynamic_prompt
def personalized_prompt(request: ModelRequest) -> str:
    username = request.runtime.context.username
    store = request.runtime.store

    preferences = store.get(namespace, username).value["chat_preferences"]
    custom_prompt = f"# 用户偏好\n{'\n'.join(preferences)}"
    logger.info(f"{username} 自定义系统提示词\n{custom_prompt}")

    return custom_prompt

agent = create_agent(
    model="deepseek-chat",
    middleware=[personalized_prompt],
    store=store,
    context_schema=UserContext
)

response1 = agent.invoke(
    {"messages": ["为什么花儿这样红？"]},
    context=UserContext(username="Ada Lovelace")
)
print("=" * 30, '-> Ada Lovelace messages <-', '=' * 30)
for msg in response1["messages"]:
    msg.pretty_print()

response2 = agent.invoke(
    {"messages": ["为什么花儿这样红？"]},
    context=UserContext(username="Blackwell")
)
print("=" * 30, '-> Blackwell messages <-', '=' * 30)
for msg in response2["messages"]:
    msg.pretty_print()

输出：

2026-04-14 11:44:10.784 | INFO     | __main__:personalized_prompt:44 - Ada Lovelace 自定义系统提示词
# 用户偏好
不喜欢啰嗦
尽可能用最精简的文字解释清楚
2026-04-14 11:44:13.622 | INFO     | __main__:personalized_prompt:44 - Blackwell 自定义系统提示词
# 用户偏好
喜欢长篇大论、引经据典
不知道的东西要明确说'不知道'，不能胡编乱造

============================== -> Ada Lovelace messages <- ==============================
================================ Human Message =================================

为什么花儿这样红？
================================== Ai Message ==================================

花儿红的主要原因是花瓣细胞中的**花青素**在酸性环境下显红色，同时受光照、温度、养分等影响，为吸引传粉者进化而成。
============================== -> Blackwell messages <- ==============================
================================ Human Message =================================

为什么花儿这样红？
================================== Ai Message ==================================

"为什么花儿这样红"这一问题，可以从科学、文学、文化等多个维度展开探讨。以下将从不同角度进行详细分析，力求全面而深入。

---

### 一、科学视角：色彩形成的自然机制
花朵的红色主要源于植物色素、光学结构以及进化适应性的共同作用。

1. **色素基础：花青素的主导作用**
   - 花青素（Anthocyanin）是形成红、紫、蓝等颜色的关键色素，属于类黄酮化合物。它在不同酸碱度（pH）下呈现颜色变化：酸性环境中偏红（如玫瑰、牡丹），中性时呈紫色，碱性环境中则偏蓝。
   - 其他色素如类胡萝卜素（橙黄色）和甜菜色素（少数植物如仙人掌）也可能与红色相关，但花青素是最常见的红色来源。

2. **光学与结构增色**
   - 部分花朵的红色并非单纯依赖色素，而是通过花瓣表面的微观结构（如多层薄膜、衍射光栅）反射特定波长的光，增强色彩饱和度。例如红罂粟的花瓣具有蜂窝状结构，能强化红色视觉效果。

3. **进化与生态学意义**
   - **传粉者吸引**：红色对许多传粉昆虫（如蜜蜂）可见度较低，但对鸟类（如蜂鸟）和部分蝶类极具吸引力。在美洲、亚洲等地区，红色花朵常与鸟媒传粉相关。
   - **环境适应**：花青素可抵御紫外线伤害、抵抗病原体，并在低温或干旱条件下保护植物细胞。高海拔地区的红花比例较高，可能与紫外线防护有关。

---

### 二、文学与艺术中的象征意义
红色在人类文化中被赋予丰富内涵，花朵的红色常成为情感与哲思的载体。

1. **爱情与激情**
   - 红玫瑰在西方传统中象征炽热爱情，可追溯至希腊罗马神话中维纳斯与阿多尼斯的故事，玫瑰的红色被联想为爱神之血。波斯诗人萨迪在《蔷薇园》中亦以玫瑰隐喻激情。

2. **生命、牺牲与革命**
   - 中国文学中，红色梅花、杜鹃常被赋予坚贞或壮烈的意象（如"杜鹃啼血"）。近现代以来，红色花朵亦与革命精神关联，如《红岩》中的红梅象征烈士鲜血。

3. **美学与哲学隐喻**
   - 王阳明在《传习录》中提及"岩中花树"，探讨心与物的关系；歌德在《色彩论》中分析红色对情感的影响。红色花朵的视觉冲击力，常引发对生命短暂与绚烂的思考（如日本"物哀"美学中的樱花）。

---

### 三、文化差异与历史演变
不同文明对红色花朵的解读存在显著差异：

1. **东方文化**
   - 中国：红色象征吉祥、喜庆，牡丹被称为"国色天香"，红色牡丹尤其代表富贵。但李贺诗句"桃花乱落如红雨"亦暗含哀愁。
   - 日本：红椿花在江户时代被赋予"谦逊"与"永恒"之意，但亦因整朵凋落的特点与武士精神关联。

2. **西方文化**
   - 基督教艺术中，红色玫瑰象征殉道者的鲜血与神圣之爱（如但丁《神曲》中天堂的玫瑰）。
   - 近代欧洲，红罂粟因一战战场盛放，成为纪念阵亡将士的符号（如约翰·麦克雷诗作《在弗兰德战场》）。

---

### 四、现代科学的新发现
近年研究进一步揭示了花朵红色的复杂性：
- **基因调控**：MYB转录因子等基因可精确控制花青素合成，通过基因工程已培育出深浅各异的红色花卉。
- **气候响应**：全球变暖可能导致花朵颜色加深，因高温促进花青素积累（2020年《生态学通讯》研究）。
- **传粉者视觉模型**：蜂鸟对红色的偏好与其视网膜中特殊色素细胞（油滴滤光）相关，这解释了美洲红色花与蜂鸟的协同进化。

---

### 结语
"花儿这样红"既是自然选择的精巧结果，也是人类文明的情感投射。从花青素的生物化学路径到《诗经》中的"桃之夭夭"，从达尔文对传粉生态的观察到现代基因图谱的解析，这一问题串联起了科学与人文的对话。红色花朵的绚丽，最终在观察者的眼中完成了自然法则与心灵共鸣的统一。

**参考文献方向**（如需进一步研究可查阅）：
- 植物生理学：《植物生物化学与分子生物学》（B.B. Buchanan等编著）
- 文化象征：《花的语言》（Jack Goody，人类学著作）
- 进化生态学：《花的起源与多样性》（昆虫传粉协同进化理论）

dynamic_prompt 是动态更新本次调用前的系统提示词，并不会更改消息列表，所以根据模型输出，可以发现系统提示词生效，但消息列表看不到相关提示词。

4.2 工具中访问

在上文介绍工具的自定义 args_schema 时我们介绍了两种方式：

基于 Pydantic 模型
基于 JSON Schema

实际上，LangChain 底层在生成工具的 schema 时，如果参数类型注解是 Pydantic 模型，也会被一并解析。

from pydantic import BaseModel, Field
from typing import List

from langchain.tools import tool
from langgraph.prebuilt.tool_node import ToolRuntime
from langchain_core.utils.function_calling import convert_to_openai_tool

class UserInfo(BaseModel):
    username: str = Field(description="用户名")
    age: int = Field(description="年龄")
    hobbies: List[str] = Field(description="兴趣爱好")

@tool(parse_docstring=True)
def save_user_info(user_info: UserInfo, runtime: ToolRuntime) -> str:
    """
    保存用户信息

    Args:
        user_info: 用户信息对象
    """
    return "ok"

print(convert_to_openai_tool(save_user_info))

输出如下：

{
    "type": "function",
    "function": {
        "name": "save_user_info",
        "description": "保存用户信息",
        "parameters": {
            "properties": {
                "user_info": {
                    "properties": {
                        "username": {
                            "description": "用户名",
                            "type": "string"
                        },
                        "age": {
                            "description": "年龄",
                            "type": "integer"
                        },
                        "hobbies": {
                            "description": "兴趣爱好",
                            "items": {
                                "type": "string"
                            },
                            "type": "array"
                        }
                    },
                    "required": [
                        "username",
                        "age",
                        "hobbies"
                    ],
                    "type": "object",
                    "description": "用户信息对象"
                }
            },
            "required": [
                "user_info"
            ],
            "type": "object"
        }
    }
}

下面的案例中，我们通过工具读写长期记忆中保存的用户信息：

from pydantic import BaseModel, Field
from typing import List, Any
from dataclasses import dataclass

from langchain.agents import create_agent
from langchain.tools import tool
from langchain.messages import HumanMessage, SystemMessage

from langgraph.prebuilt.tool_node import ToolRuntime
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import InMemorySaver

from loguru import logger
from dotenv import load_dotenv
load_dotenv()

@dataclass
class UserContext:
    user_id: str

store = InMemoryStore()
namespace = ("users", "user_info")

class UserInfo(BaseModel):
    username: str = Field(description="用户名", default="unknown")
    age: int = Field(description="年龄", default=0)
    # 这里相当于告诉模型，每次创建新实例时，调用 list() 创建一个新的 list 传进去
    hobbies: List[str] = Field(description="兴趣爱好", default_factory=list)
    # 这种写法是用可变对象做默认值，会有多个对象共享一个列表对象的风险
    # hobbies: List[str] = Field(description="兴趣爱好", default=[])

@tool(parse_docstring=True)
def read_user_info(runtime: ToolRuntime[UserContext, Any]) -> UserInfo | str:
    """
    读取用户信息

    Returns:
        UserInfo | str: 如果找到用户信息，返回 UserInfo 对象；否则返回空字符串。
    """
    user_id = runtime.context.user_id
    store = runtime.store
    item = store.get(namespace, user_id)
    if item:
        return UserInfo(**item.value)
    return ''

@tool(parse_docstring=True)
def write_user_info(user_info: UserInfo, runtime: ToolRuntime[UserContext, Any]) -> bool:
    """
    将用户信息写入长期记忆

    Args:
        user_info：用户信息对象

    Returns:
        bool: 如果成功写入则返回 True，否则返回 False
    """
    user_id = runtime.context.user_id
    store = runtime.store
    try:
        logger.info(f"写入记忆, namespace: {namespace}, user_id: {user_id}, user_info: {user_info.model_dump()}")
        store.put(namespace, user_id, user_info.model_dump())
    except Exception as e:
        logger.error(e)
        return False
    return True

agent = create_agent(
    model="deepseek-chat",
    tools=[read_user_info, write_user_info],
    store=store,
    checkpointer=InMemorySaver(),
    context_schema=UserContext
)

config1 = {"configurable": {"thread_id": "thread_1"}}
config2 = {"configurable": {"thread_id": "thread_2"}}

agent.invoke(
    {"messages": [
        SystemMessage("如果输入包含用户信息，抽取并调用工具记录。在记录之前先查找历史信息，和新增信息合并后记录。"
                      "如果用户提问设计个人信息，调用工具查找"),
        HumanMessage("你好，我是韩立，我喜欢修仙")
    ]},
    context=UserContext(user_id='user_1'),
    config=config1
)

thread_1_response = agent.invoke(
    {"messages": ["我今年两百岁了，是你们口中的 "元婴老怪"，我喜欢跑步"]},
    context=UserContext(user_id='user_1'),
    config=config1
)

print("=" * 30, '-> thread_1 messages <-', '=' * 30)
for msg in thread_1_response["messages"]:
    msg.pretty_print()

thread_2_response = agent.invoke(
    {"messages": ["你还记得我吗？"]},
    context=UserContext(user_id='user_1'),
    config=config2
)

print("=" * 30, '-> thread_2 messages <-', '=' * 30)
for msg in thread_2_response["messages"]:
    msg.pretty_print()

输出：

2026-04-14 15:58:30.011 | INFO     | __main__:write_user_info:61 - 写入记忆, namespace: ('users', 'user_info'), user_id: user_1, user_info: {'username': '韩立', 'age': 0, 'hobbies': ['修仙']}
2026-04-14 15:58:41.439 | INFO     | __main__:write_user_info:61 - 写入记忆, namespace: ('users', 'user_info'), user_id: user_1, user_info: {'username': '韩立', 'age': 200, 'hobbies': ['修仙', '跑步']}

============================== -> thread_1 messages <- ==============================
================================ System Message =================================

如果输入包含用户信息，抽取并调用工具记录。在记录之前先查找历史信息，和新增信息合并后记录。如果用户提问设计个人信息，调用工具查找
================================ Human Message =================================

你好，我是韩立，我喜欢修仙
================================== Ai Message ==================================

你好韩立！很高兴认识你。我看到你喜欢修仙，这是一个很有趣的爱好。

为了让我能更好地记住你的信息，让我先查看一下是否已经有你的历史记录。
Tool Calls:
  read_user_info (call_00_DAqBrMPFS4iZVK5F7mvaOsBy)
  Call ID: call_00_DAqBrMPFS4iZVK5F7mvaOsBy
  Args:
================================= Tool Message =================================
Name: read_user_info

================================== Ai Message ==================================

看起来目前还没有你的历史记录。让我记录一下你的基本信息：
Tool Calls:
  write_user_info (call_00_Zoz8sLHivXkA0a0Py2yvDFJB)
  Call ID: call_00_Zoz8sLHivXkA0a0Py2yvDFJB
  Args:
    user_info: {'username': '韩立', 'hobbies': ['修仙']}
================================= Tool Message =================================
Name: write_user_info

true
================================== Ai Message ==================================

好的，我已经记录了你喜欢修仙这个爱好。修仙确实是一个很有深度的领域，涉及到修炼、功法、丹药、法宝等很多方面。你对修仙的哪个方面特别感兴趣呢？比如是修炼功法、炼丹术、阵法，还是其他方面？
================================ Human Message =================================

我今年两百岁了，是你们口中的 "元婴老怪"，我喜欢跑步
================================== Ai Message ==================================

哇！两百岁的元婴老怪，这真是令人惊叹！看来您在修仙道路上已经走了很长的路。

让我更新一下您的信息，包括年龄和新的爱好——跑步。
Tool Calls:
  read_user_info (call_00_I9AO3Y8RJSQLeHhwBGszkuiW)
  Call ID: call_00_I9AO3Y8RJSQLeHhwBGszkuiW
  Args:
================================= Tool Message =================================
Name: read_user_info

username='韩立' age=0 hobbies=['修仙']
================================== Ai Message ==================================

现在我来更新您的完整信息：
Tool Calls:
  write_user_info (call_00_3JpZC48aqJG7m0m6X2EnZFNH)
  Call ID: call_00_3JpZC48aqJG7m0m6X2EnZFNH
  Args:
    user_info: {'username': '韩立', 'age': 200, 'hobbies': ['修仙', '跑步']}
================================= Tool Message =================================
Name: write_user_info

true
================================== Ai Message ==================================

很好！我已经更新了您的信息：
- 姓名：韩立
- 年龄：200岁
- 爱好：修仙、跑步

作为一位元婴期的前辈，您一定有着丰富的修炼经验和人生智慧。跑步这个爱好很有意思，对于修仙者来说，这既是锻炼身体的好方法，也能让心境更加平和。您平时是在宗门内跑步，还是在山林间奔跑呢？
============================== -> thread_2 messages <- ==============================
================================ Human Message =================================

你还记得我吗？
================================== Ai Message ==================================

我来查看一下之前记录的用户信息。
Tool Calls:
  read_user_info (call_00_iJgxeOZKvXCNYkRjX3EFGp8q)
  Call ID: call_00_iJgxeOZKvXCNYkRjX3EFGp8q
  Args:
================================= Tool Message =================================
Name: read_user_info

username='韩立' age=200 hobbies=['修仙', '跑步']
================================== Ai Message ==================================

当然记得！你是韩立，200岁，喜欢修仙和跑步。很高兴再次见到你！最近修仙进展如何？还是经常跑步锻炼吗？

分析：

前两次 invoke 通过相同的 thread_id 串联为一个会话，共享短期记忆。
context 是运行时静态配置，每次 invoke 互相独立。
长期记忆可以在任意位置访问，全局共享同一份信息。
工具参数列表中的 ToolRuntime 是 LangChain / LangGraph 运行时注入的，在工具 schema 中没有体现。
ToolRuntime 的默认泛型是 [None, Any]，第一个是 Context 的泛型，第二个是 State 对象的泛型。如果不显式指定上下文对象的类型，则 LangChain / LangGraph 底层会认为上下文对象为 None，invoke 调用时传递了 context 就会抛出警告，所以这里将 UserContext 作为 ToolRuntime 的第一个泛型传递。

大模型

09-上下文与记忆

背景

第09章：上下文与记忆

1、概述

1.1 为什么需要记忆（Memory）

1.2 如何解决记忆问题

1.2.1 上下文工程

1.2.2 上下文类型及相关的 API

1.3 LangChain 的记忆

1.3.1 记忆的分类

1.3.2 记忆的管理

2、短期记忆

2.1 基于内存的持久化器

2.1.1 举例 1：没有记忆

2.1.2 举例 2：拥有记忆

2.1.3 关键步骤说明

2.1.4 工作原理

2.1.5 常见问题

2.2 基于外部存储介质的持久化器

2.2.1 数据库环境准备

2.2.2 代码实现

2.2.3 查看持久化数据

2.3 对比两种方式

2.4 记忆治理策略（上下文管理）

2.4.1 消息裁剪

2.4.2 消息删除

2.4.3 摘要

2.4.4 自定义过滤策略

2.5 了解：state 的理解

3、长期记忆

3.1 基本理解

3.1.1 什么是长期记忆

3.1.2 类型划分

3.1.3 存储架构

3.2 基础 API 的使用

3.2.1 put() / get()：写入/读取 API

3.2.3 search()：检索 API

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

3.3.1 在工具中访问长期记忆

3.3.2 在中间件中访问长期记忆

3.4 何时写入记忆

4、课后阅读：Static runtime Context

4.1 中间件中访问

4.1.1 Node-style hooks

4.1.2 Wrap-style hooks

4.1.3 特殊的便捷中间件 dynamic_prompt

4.2 工具中访问

喜欢这篇文章？打赏一下作者吧

目录

最新文章

分类

归档

标签

3.2.1 `put()` / `get()`：写入/读取 API

3.2.3 `search()`：检索 API

4.1.3 特殊的便捷中间件 `dynamic_prompt`