Jean's Blog

一个专注软件测试开发技术的个人博客

0%

LangChain核心组件Memory(记忆)

发表于 分类于
学习LangChain学习笔记第四讲

Langchain的v.0.3.x版本中是没有记忆模块的，在v1.x.x版本以后，Langchain与Langraph深度融合以后，Langchain有了短期记忆的模块，官方地址为：https://docs.langchain.com/oss/python/langchain/short-term-memory

因此本章节重点介绍LangChain中的短期记忆，在Langgraph的记忆模块中是支持短期记忆和长期记忆的，会在Langgraph中重点介绍

什么是记忆

从定义角度阐述记忆的重要性

  • 记忆的定义：记忆是一个能够存储和回忆以往交互信息的系统。对于人工智能代理而言，它就如同一个信息的“仓库”，将过往与用户的每一次互动都记录下来，以便后续能够随时调取使用。
  • 记忆的作用：记忆对于人工智能代理来说至关重要。首先，它使得代理能够记住之前的交互过程，当再次遇到类似情境时，可以凭借记忆中的经验来做出更合适的反应。其次，记忆为代理从反馈中学习提供了基础。代理可以根据记忆中的反馈信息，分析哪些行为是有效的、哪些需要改进，从而不断优化自身的行为模式。最后，记忆还能帮助代理适应用户的偏好。通过记住用户的喜好和习惯，代理能够为用户提供更加个性化、贴合需求的服务。

从任务复杂度角度分析记忆的必要性

  • 简单任务与复杂任务的对比：当人工智能代理处理较为简单的任务时，可能仅依靠预设的规则和程序就能完成。然而，随着任务复杂度的增加，涉及到大量的用户交互，记忆的作用就凸显出来了。例如，在一个简单的问答系统中，代理可能只需要根据关键词匹配答案即可。但在一个需要进行多轮对话、理解用户意图并提供连贯建议的复杂场景中，如果没有记忆功能，代理就无法跟踪对话的进展，无法理解用户的上下文信息，从而导致无法给出准确且符合用户需求的回答。
  • 对效率和用户满意度的影响：从效率方面来看，具备记忆能力的代理能够更快地找到解决问题的方法。因为它可以避免重复处理相同或类似的问题，而是直接利用记忆中的经验来快速响应。从用户满意度角度来说，用户更倾向于与能够记住自己偏好的代理进行交互。当代理能够根据记忆为用户提供个性化的服务时，用户会感到被尊重和理解，从而提升对代理的满意度。

短期记忆

  • 定义：短期记忆允许应用程序在单个线程或对话中记住之前的交互。
  • 组织方式：线程在会话中组织多次交互，类似于电子邮件将消息分组到单个对话中的方式。
  • 常见形式：对话历史是短期记忆最常见的形式。

长对话的挑战

  • 上下文窗口限制：对于现代的语言模型（LLMs），长对话是一个挑战。完整的对话历史可能无法完全放入LLM的上下文窗口中，这会导致上下文丢失或出现错误。
  • 性能问题：即使你的模型支持完整的上下文长度，大多数LLMs在处理长上下文时仍然表现不佳。它们会被过时的或离题的内容“分心”，同时还会遭受响应速度变慢和成本增加的问题。

对话的组织和处理

  • 消息交互：聊天模型通过消息来接收上下文，这些消息包括指令（系统消息）和输入（人类消息）。在聊天应用中，消息在人类输入和模型响应之间交替出现，随着时间的推移，消息列表会变得越来越长。
  • 上下文窗口限制的应对：由于上下文窗口是有限的，许多应用可以从使用技术来移除或“忘记”过时信息中受益。

短期记忆主要用于 线程级对话保持，确保模型持续拥有必要的历史上下文。

  • 短期记忆操作使用 runtime 对象的 state 属性, 也称之为状态(每一次调用 agent 的 state 是独立的)
  • LangChain的 Agent 将短期记忆作为代理状态的一部分来管理。通过将这些数据存储在图的状态中，代理可以在保持不同线程间的分离的同时，访问给定对话的完整上下文。状态通过检查点持久化到数据库（或内存），因此线程可以随时恢复。当调用代理或完成某一步（如工具调用）时，短期内存会更新，并在每步开始时读取状态。

使用

如何开启短期记忆：Checkpointer

基本使用（In-Memory）

官方示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langgraph.checkpoint.memory import InMemorySaver  

# 定义一个工具
@tool("get_user_info", description="用户获取当前用户信息的工具")
def get_user_info(runtime: ToolRuntime):
    """获取用户信息"""
    ...

agent = create_agent(
    "gpt-5",
    tools=[get_user_info],
    checkpointer=InMemorySaver(),  
)

agent.invoke(
    {"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
    {"configurable": {"thread_id": "1"}},  
)

练习代码

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

from src.core.llms import model_client

# 创建一个agent，开启短期记忆
agent = create_agent(
    model_client,
    # 配置智能体工具
    tools=[],
    # 开启短期记忆
    checkpointer=InMemorySaver()
)

# 调用agent,一定要通过参数config传入线程id
res = agent.invoke({"messages": [{"role": "user", "content": "我的名字叫小花"}]}, config={"configurable": {"thread_id": "1"}})
print(res)

生产环境

数据库来存储其记录的状态信息的，利用数据库的稳定性和可靠性来保证检查点数据的持久化和安全性。

安装提供的postgres的依赖

1
pip install langgraph-checkpoint-postgres

官方示例

1
2
3
4
5
6
7
8
9
10
11
12
13
from langchain.agents import create_agent

from langgraph.checkpoint.postgres import PostgresSaver  


DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup() # auto create tables in PostgresSql
    agent = create_agent(
        "gpt-5",
        tools=[get_user_info],
        checkpointer=checkpointer,  
    )

自定义Agent记忆

默认情况下，智能代理（agents）使用 AgentState 来管理短期记忆。这里的“短期记忆”主要指的是在一次会话（session）或对话线程（thread）中，代理能够记住之前交互的信息，例如用户说过的话、之前的回答等，以便在后续的交互中能够更好地理解和回应。

通过一个名为 messages 的键来存储对话历史。也就是说，AgentState 中有一个 messages 键，它的值是一个列表，包含了这次会话中所有交互的信息，比如用户发送的消息和代理的回复等。

你可以对 AgentState 进行扩展，以添加额外的字段。这意味着，除了默认的 messages 字段用于存储对话历史外，你还可以根据自己的需求，为 AgentState 添加其他字段，用于存储更多不同类型的信息，比如用户的偏好设置、会话的上下文信息等。

自定义的状态模式（state schemas）是通过 state_schema 参数传递给 create_agent 函数的。当你定义了自己的 AgentState 扩展类，包含了额外的字段后，你需要在创建代理时，通过 state_schema 参数将这个自定义的状态模式传递给 create_agent 函数，这样代理在运行时就能够按照你定义的模式来管理和使用状态信息。

官方示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver


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

agent = create_agent(
    "gpt-5",
    tools=[get_user_info],
    state_schema=CustomAgentState,  
    checkpointer=InMemorySaver(),
)

# Custom state can be passed in invoke
result = agent.invoke(
    {
        "messages": [{"role": "user", "content": "Hello"}],
        "user_id": "user_123",  
        "preferences": {"theme": "dark"}  
    },
    {"configurable": {"thread_id": "1"}})

源码中AgentState的定义

1
2
3
4
5
6
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]]

记忆访问

在工具中读取短期记忆

短期记忆（short-term memory）是指代理在当前对话线程（thread）中记住的交互信息。这些信息对于工具来说可能是有用的，因为它们可以帮助工具了解对话的上下文和历史。

  • ToolRuntime 参数：这是传递给工具的一个参数，它包含了代理的运行时信息，包括短期记忆（state）。通过这个参数，工具可以访问代理的短期记忆。
  • 隐藏的参数：tool_runtime 参数在工具的签名（signature）中是隐藏的，这意味着模型（model）不会直接看到这个参数。换句话说，模型在调用工具时，不会意识到这个参数的存在，但它仍然可以通过 tool_runtime 参数来访问短期记忆。

官方示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime


class CustomState(AgentState):
    user_id: str

@tool
def get_user_info(
    runtime: ToolRuntime
) -> str:
    """Look up user info."""
    # 读取短期记忆中的内容
    user_id = runtime.state["user_id"]
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

agent = create_agent(
    model="gpt-5-nano",
    tools=[get_user_info],
    state_schema=CustomState,
)

result = agent.invoke({
    "messages": "look up user information",
    "user_id": "user_123"
})
print(result["messages"][-1].content)
# > User is John Smith.

练习代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
# @Time：2025/12/19 10:31
# @Author：jinglv
"""
记忆的访问
在工具中传递数据
"""
from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime
from langgraph.checkpoint.memory import InMemorySaver

from src.core.llms import model_client


def print_stream_result(response):
    """流式结果的输出"""
    for item in response:
        for key, value in item.items():
            if key == "model":
                print("-----执行步骤：调用大模型")
                if value['messages'][0].content:
                    print(f"-----大模型分析的结果:{value["messages"][0].content}")
                elif value['messages'][0].tool_calls:
                    print("-----大模型分析的结果为调用以下工具：")
                    for tool_ in value['messages'][0].tool_calls:
                        print(f"工具名称：{tool_['name']},调用工具的入参:{tool_['args']}")
            elif key == "tools":
                print(f"智能体执行工具：{value['messages'][0].name}")
                print(f"工具执行结果：{value['messages'][0].content}")


# 定义短期记忆(runtime.state)中其他额外的字段
class UserInfo(AgentState):
    user_id: str
    nickname: str


# 定义一个工具
@tool("get_user_info", description="用户获取当前用户信息的工具")
def get_user_info(runtime: ToolRuntime):
    """获取用户信息"""
    # 读取短期记忆中的内容
    return {"user_id": runtime.state.get("user_id"), "nickname": runtime.state.get("nickname"), "age": 18}


# 创建一个agent，开启短期记忆
agent = create_agent(
    model_client,
    # 配置智能体工具
    tools=[get_user_info],
    # 指定短期记忆中额外的字段
    state_schema=UserInfo,
    # 开启短期记忆
    checkpointer=InMemorySaver()
)

# 调用agent,一定要通过参数config出入线程id
res = agent.stream(
    # agent调用的时候第一个参数input传递的只，最终会保存到runtime.state(短期记忆)中
    input={"messages": [{"role": "user", "content": "获取用户的信息"}], "user_id": "9527", "nickname": "花花"},
    config={"configurable": {"thread_id": "1"}}
)
print_stream_result(res)

执行结果

1
2
3
4
5
6
7
8
9
10
11
12
13
-----执行步骤：调用大模型
-----大模型分析的结果:我来帮您获取用户信息。
智能体执行工具：get_user_info
工具执行结果：{"user_id": "9527", "nickname": "花花", "age": 18}
-----执行步骤：调用大模型
-----大模型分析的结果:根据获取到的用户信息，以下是您的详细信息：

**用户信息：**
- **用户ID：** 9527
- **昵称：** 花花
- **年龄：** 18岁

如果您需要了解其他信息或有其他问题，请随时告诉我！

在工具中写入短期记忆

  • 在智能代理执行任务的过程中，可以通过工具直接返回状态更新，以此来修改代理的短期记忆（state）。
  • 这种做法非常有用，主要有两个方面的益处：
    • 持久化中间结果：可以将一些在执行过程中产生的中间结果保存下来，这样在后续的执行过程中就可以直接使用这些中间结果，而无需重新计算，从而提高执行效率。
    • 使信息对后续工具或提示可用：可以将一些重要的信息通过状态更新传递给后续的工具或提示，让它们能够基于这些信息进行更准确的操作或生成更合适的提示内容，增强智能代理的整体性能和用户体验。

官方示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from langchain.tools import tool, ToolRuntime
from langchain_core.runnables import RunnableConfig
from langchain.messages import ToolMessage
from langchain.agents import create_agent, AgentState
from langgraph.types import Command
from pydantic import BaseModel


class CustomState(AgentState):  
    user_name: str

class CustomContext(BaseModel):
    user_id: str

@tool
def update_user_info(
    runtime: ToolRuntime[CustomContext, CustomState],
) -> Command:
    """Look up and update user info."""
    user_id = runtime.context.user_id
    name = "John Smith" if user_id == "user_123" else "Unknown user"
    return Command(update={  
        "user_name": name,
        # update the message history
        "messages": [
            ToolMessage(
                "Successfully looked up user information",
                tool_call_id=runtime.tool_call_id
            )
        ]
    })

@tool
def greet(
    runtime: ToolRuntime[CustomContext, CustomState]
) -> str | Command:
    """Use this to greet the user once you found their info."""
    user_name = runtime.state.get("user_name", None)
    if user_name is None:
       return Command(update={
            "messages": [
                ToolMessage(
                    "Please call the 'update_user_info' tool it will get and update the user's name.",
                    tool_call_id=runtime.tool_call_id
                )
            ]
        })
    return f"Hello {user_name}!"

agent = create_agent(
    model="gpt-5-nano",
    tools=[update_user_info, greet],
    state_schema=CustomState, 
    context_schema=CustomContext,
)

agent.invoke(
    {"messages": [{"role": "user", "content": "greet the user"}]},
    context=CustomContext(user_id="user_123"),
)

练习代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
# @Time：2025/12/19 10:31
# @Author：jinglv
"""
在工具中写入短期记忆
"""
from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime
from langchain_core.messages import ToolMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

from src.core.llms import model_client


def print_stream_result(response):
    """流式结果的输出"""
    for item in response:
        for key, value in item.items():
            if key == "model":
                print("-----执行步骤：调用大模型")
                if value['messages'][0].content:
                    print(f"-----大模型分析的结果:{value["messages"][0].content}")
                elif value['messages'][0].tool_calls:
                    print("-----大模型分析的结果为调用以下工具：")
                    for tool_ in value['messages'][0].tool_calls:
                        print(f"工具名称：{tool_['name']},调用工具的入参:{tool_['args']}")
            elif key == "tools":
                print(f"智能体执行工具：{value['messages'][0].name}")
                print(f"工具执行结果：{value['messages'][0].content}")


# 定义短期记忆(runtime.state)中其他额外的字段
class UserInfo(AgentState):
    user_id: str
    username: str


# 定义一个工具
@tool("get_user_info", description="用户获取当前用户信息的工具")
def get_user_info(runtime: ToolRuntime):
    """获取用户信息"""
    return {"user_id": runtime.state.get("user_id"), "username": runtime.state.get("username"), "age": 18}


@tool("save_user_info")
def save_user_info(username: str, runtime: ToolRuntime):
    """保存用户信息"""
    # 在短期记忆中保存记忆
    # state中数据的写入
    return Command(update={
        "username": username,
        "messages": [
            ToolMessage(
                "用户保存成功",
                tool_call_id=runtime.tool_call_id
            )
        ]
    })


# 创建一个agent，开启短期记忆
agent = create_agent(
    model_client,
    # 配置智能体工具
    tools=[get_user_info, save_user_info],
    # 指定短期记忆中额外的字段
    state_schema=UserInfo,
    # 开启短期记忆
    checkpointer=InMemorySaver()
)

# 调用agent,一定要通过参数config出入线程id
res = agent.stream(
    # agent调用的时候第一个参数input传递的只，最终会保存到runtime.state(短期记忆)中
    input={"messages": [{"role": "user", "content": "我的名字叫小红帽，并输出我的用户信息"}], "user_id": "9527"},
    config={"configurable": {"thread_id": "1"}}
)
print_stream_result(res)

执行结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
-----执行步骤：调用大模型
-----大模型分析的结果:我来帮您保存用户名并获取您的用户信息。
智能体执行工具：save_user_info
工具执行结果：用户保存成功
-----执行步骤：调用大模型
-----大模型分析的结果:现在让我获取您的用户信息：
智能体执行工具：get_user_info
工具执行结果：{"user_id": "9527", "username": "小红帽", "age": 18}
-----执行步骤：调用大模型
-----大模型分析的结果:您好，小红帽！您的用户信息如下：

- **用户ID**: 9527
- **用户名**: 小红帽  
- **年龄**: 18岁

您的用户名"小红帽"已经成功保存到系统中了！

记忆处理中间件

动态系统提示词

使用 dynamic_prompt 可以根据短期记忆动态构建构建系统提示词

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from langchain.agents import create_agent
from typing import TypedDict
from langchain.agents.middleware import dynamic_prompt, ModelRequest


class CustomContext(TypedDict):
    user_name: str


def get_weather(city: str) -> str:
    """Get the weather in a city."""
    return f"The weather in {city} is always sunny!"


@dynamic_prompt
def dynamic_system_prompt(request: ModelRequest) -> str:
    user_name = request.runtime.context["user_name"]
    system_prompt = f"You are a helpful assistant. Address the user as {user_name}."
    return system_prompt


agent = create_agent(
    model="gpt-5-nano",
    tools=[get_weather],
    middleware=[dynamic_system_prompt],
    context_schema=CustomContext,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    context=CustomContext(user_name="John Smith"),
)
for msg in result["messages"]:
    msg.pretty_print()

练习代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
# @Time：2025/12/29 14:18
# @Author：jinglv
"""
使用 dynamic_prompt 可以根据短期记忆动态构建构建系统提示词
"""
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from langgraph.checkpoint.memory import InMemorySaver

from src.core.llms import model_client


def print_stream_result(response):
    """流式结果的输出"""
    for item in response:
        for key, value in item.items():
            if key == "model":
                print("-----执行步骤：调用大模型")
                if value['messages'][0].content:
                    print(f"-----大模型分析的结果:{value["messages"][0].content}")
                elif value['messages'][0].tool_calls:
                    print("-----大模型分析的结果为调用以下工具：")
                    for tool_ in value['messages'][0].tool_calls:
                        print(f"工具名称：{tool_['name']},调用工具的入参:{tool_['args']}")
            elif key == "tools":
                print(f"智能体执行工具：{value['messages'][0].name}")
                print(f"工具执行结果：{value['messages'][0].content}")


# 定义短期记忆(runtime.state)中其他额外的字段
class UserInfo(AgentState):
    user_id: str
    username: str
    role: str


# 定义一个动态配置系统提示词的中间件
@dynamic_prompt
def get_system_prompt(request: ModelRequest) -> str:
    """动态获取系统提示词的中间件"""
    # 通过state中传递的role来设计提示词, ModelRequest：可以获取上下问信息的
    role = request.state.get("role")
    return f"你是一位{role}"


# 创建一个agent，开启短期记忆
agent = create_agent(
    model_client,
    # 指定短期记忆中额外的字段
    state_schema=UserInfo,
    # 添加中间件
    middleware=[get_system_prompt],
    # 开启短期记忆
    checkpointer=InMemorySaver()
)

# 调用agent,一定要通过参数config出入线程id
res = agent.stream(
    # agent调用的时候第一个参数input传递的只，最终会保存到runtime.state(短期记忆)中
    input={"messages": [{"role": "user", "content": "请问什么是变量"}], "role": "程序员"},
    config={"configurable": {"thread_id": "1"}}
)
print_stream_result(res)

模型前处理

在模型调用前，访问@before_model中间件中的短期内存（状态）以处理消息。比如消息裁剪

before_model官方说明：https://reference.langchain.com/python/langchain/middleware/#langchain.agents.middleware.before_model

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
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 langchain_core.runnables import RunnableConfig
from langgraph.runtime import Runtime
from typing import Any


@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """Keep only the last few messages to fit context window."""
    messages = state["messages"]

    if len(messages) <= 3:
        return None  # No changes needed

    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(
    "gpt-5-nano",
    tools=[],
    middleware=[trim_messages],
    checkpointer=InMemorySaver()
)

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

agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()
"""
================================== Ai Message ==================================

Your name is Bob. You told me that earlier.
If you'd like me to call you a nickname or use a different name, just say the word.
"""

模型后处理

after_model 中间件可以在模型调用后处理记忆，例如屏蔽敏感词输出

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.runtime import Runtime


@after_model
def validate_response(state: AgentState, runtime: Runtime) -> dict | None:
    """Remove messages containing sensitive words."""
    STOP_WORDS = ["password", "secret"]
    last_message = state["messages"][-1]
    if any(word in last_message.content for word in STOP_WORDS):
        return {"messages": [RemoveMessage(id=last_message.id)]}
    return None

agent = create_agent(
    model="gpt-5-nano",
    tools=[],
    middleware=[validate_response],
    checkpointer=InMemorySaver(),
)

短期记忆管理对话消息

agent 的任务中如果出现长对话，可能超过 模型 上下文的最大限制，因此需要管理消息历史：

技术 描述
截断消息 保留最近 N 条消息
删除消息 删除不需要的消息
总结消息 用 summary 替换早期历史
自定义策略 过滤、压缩、优先级等

截断消息：before_model 中间件

只保留最近几条消息，示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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:
    """Keep only the last few messages to fit context window."""
    messages = state["messages"]

    if len(messages) <= 3:
        return None  # No changes needed

    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,
    tools=tools,
    middleware=[trim_messages],
    checkpointer=InMemorySaver(),
)

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

agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()
"""
================================== Ai Message ==================================

Your name is Bob. You told me that earlier.
If you'd like me to call you a nickname or use a different name, just say the word.
"""

删除消息：RemoveMessage

  • 删除特定消息，示例代码

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

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

  • 删除所有消息，示例代码

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

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

总结消息：SummarizationMiddleware

在消息过长时自动总结。示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig


checkpointer = InMemorySaver()

agent = create_agent(
    model="gpt-4o",
    tools=[],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",
            trigger={"tokens": 4000},
            keep={"messages": 20},
        )
    ],
    checkpointer=checkpointer,
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()

注意

注意一点，这里和之前提到过的运行上下文的区别

  • runtime.context：运行上下文信息
    • 运行上下文用来传递不可变的值（数据），在工具中不能修改context中的值
  • runtime.state：存储在短期记忆中的内容
    • 短期记忆state中除了agent运行的message（消息列表）外，还可以像context一样传递数据（短期记忆中可以在工具中修改state的值）