短期记忆

Alpha 版本提示： 本文档涵盖的是 v1-alpha 版本。内容尚不完整，且可能随时变更。如需查阅最新稳定版本，请参阅 v0 版本的 LangChain Python 文档或 LangChain JavaScript 文档。

概述

记忆是一种能够记住先前交互信息的系统。对于人工智能代理而言，记忆至关重要，因为它使代理能够记住之前的交互、从反馈中学习并适应用户的偏好。随着代理处理的任务越来越复杂，涉及的用户交互越来越多，这种能力对于提高效率和用户满意度变得至关重要。短期记忆使您的应用程序能够在单个线程或对话中记住之前的交互。

线程用于组织一次会话中的多个交互，类似于电子邮件将消息分组到单个对话中的方式。

对话历史是最常见的短期记忆形式。然而，长对话对当今的大语言模型（LLM）构成了挑战；完整的历史记录可能无法容纳在LLM的上下文窗口内，从而导致上下文丢失或错误。即使您的模型支持完整的上下文长度，大多数LLM在处理长上下文时表现仍然不佳。它们容易被陈旧或离题的内容“分散注意力”，同时还会遭受响应速度变慢和成本增加的问题。聊天模型通过消息接收上下文，这些消息包括指令（系统消息）和输入（用户消息）。在聊天应用中，消息在用户输入和模型响应之间交替出现，从而形成一个随时间增长的消息列表。由于上下文窗口有限，许多应用可以通过使用技术来移除或“遗忘”陈旧信息而受益。

使用方法

要为代理添加短期记忆（线程级持久化），您需要在创建代理时指定一个 checkpointer。

LangChain 的代理将短期记忆作为代理状态的一部分进行管理。通过将这些状态存储在图的状态中，代理可以访问给定对话的完整上下文，同时保持不同线程之间的隔离。状态通过 checkpointer 持久化到数据库（或内存）中，以便随时恢复线程。短期记忆在代理被调用或完成一个步骤（如工具调用）时更新，并在每个步骤开始时读取状态。

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

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

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

在生产环境中

在生产环境中，请使用由数据库支持的 checkpointer：

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:
    agent = create_agent(
        "openai:gpt-5",
        [get_user_info],
        checkpointer=checkpointer,
    )

自定义代理记忆

默认情况下，代理使用 AgentState 来管理短期记忆，特别是通过 messages 键管理对话历史。用户可以通过继承 AgentState 来向状态添加额外字段。这种自定义状态随后可以通过工具和动态提示/模型函数访问。

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

class CustomAgentState(AgentState):
    user_id: str

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

常见模式

启用短期记忆后，长对话可能会超出LLM的上下文窗口。常见的解决方案包括：

裁剪消息

删除前N条或后N条消息（在调用LLM之前）

删除消息

从LangGraph状态中永久删除消息

总结消息

总结历史中的早期消息并用摘要替换它们

自定义策略

自定义策略（例如，消息过滤等）

这使得代理能够在不超出LLM上下文窗口的情况下跟踪对话。

裁剪消息

大多数LLM都有最大支持的上下文窗口（以token为单位）。决定何时截断消息的一种方法是计算消息历史中的token数量，并在接近该限制时进行截断。如果您使用的是LangChain，可以使用裁剪消息工具并指定要保留的token数量，以及用于处理边界的 strategy（例如，保留最后 maxTokens）。要在代理中裁剪消息历史，请使用 @[pre_model_hook][create_agent] 和 trim_messages 函数：

from langchain_core.messages.utils import trim_messages, count_tokens_approximately
from langchain_core.messages import BaseMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain_core.runnables import RunnableConfig

def pre_model_hook(state) -> dict[str, list[BaseMessage]]:
    """
    此函数将在每次调用LLM之前被调用，以准备LLM的消息。
    """
    trimmed_messages = trim_messages(
        state["messages"],
        strategy="last",
        token_counter=count_tokens_approximately,
        max_tokens=384,
        start_on="human",
        end_on=("human", "tool"),
    )
    return {"llm_input_messages": trimmed_messages}


checkpointer = InMemorySaver()
agent = create_agent(
    "openai:gpt-5-nano",
    tools=[],
    pre_model_hook=pre_model_hook,
    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()
"""
================================== Ai Message ==================================

您的名字是Bob。您之前告诉过我。
如果您希望我称呼您昵称或使用其他名字，请随时告诉我。
"""

删除消息

您可以从图状态中删除消息以管理消息历史。当您想要删除特定消息或清除整个消息历史时，这非常有用。要从图状态中删除消息，可以使用 RemoveMessage。为了使 RemoveMessage 正常工作，您需要使用带有 add_messages reducer 的状态键。默认的 AgentState 提供了这一点。要删除特定消息：

from langchain_core.messages import RemoveMessage

def delete_messages(state):
    messages = state["messages"]
    if len(messages) > 2:
        # 删除最早的两条消息
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}

要删除所有消息：

from langgraph.graph.message import REMOVE_ALL_MESSAGES

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

删除消息时，请确保最终的消息历史是有效的。检查您所使用的LLM提供商的限制。例如：

某些提供商期望消息历史以 user 消息开头
大多数提供商要求带有工具调用的 assistant 消息后必须跟随相应的 tool 结果消息。

from langchain_core.messages import RemoveMessage
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig


def delete_messages(state):
    messages = state["messages"]
    if len(messages) > 2:
        # 删除最早的两条消息
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}


agent = create_agent(
    "openai:gpt-5-nano",
    tools=[],
    prompt="请简洁明了。",
    post_model_hook=delete_messages,
    checkpointer=InMemorySaver(),
)

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

for event in agent.stream(
    {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

for event in agent.stream(
    {"messages": [{"role": "user", "content": "what's my name?"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

[('human', "hi! I'm bob")]
[('human', "hi! I'm bob"), ('ai', '嗨，Bob！很高兴认识你。今天有什么我可以帮你的吗？我可以回答问题、头脑风暴、起草文本、解释事物或帮助编写代码。')]
[('human', "hi! I'm bob"), ('ai', '嗨，Bob！很高兴认识你。今天有什么我可以帮你的吗？我可以回答问题、头脑风暴、起草文本、解释事物或帮助编写代码。'), ('human', "what's my name?")]
[('human', "hi! I'm bob"), ('ai', '嗨，Bob！很高兴认识你。今天有什么我可以帮你的吗？我可以回答问题、头脑风暴、起草文本、解释事物或帮助编写代码。'), ('human', "what's my name?"), ('ai', '您的名字是Bob。今天有什么我可以帮您的吗，Bob？')]
[('human', "what's my name?"), ('ai', '您的名字是Bob。今天有什么我可以帮您的吗，Bob？')]

总结消息

如上所述，裁剪或删除消息的问题在于，您可能会因消息队列的削减而丢失信息。因此，一些应用受益于使用聊天模型对消息历史进行更复杂的摘要处理方法。要在代理中总结消息历史，请使用 @[pre_model_hook][create_agent] 和预构建的 SummarizationNode 抽象：

from langmem.short_term import SummarizationNode, RunningSummary
from langchain_core.messages.utils import count_tokens_approximately
from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableConfig

model = ChatOpenAI(model="gpt-4o-mini")

summarization_node = SummarizationNode(
    token_counter=count_tokens_approximately,
    model=model,
    max_tokens=384,
    max_summary_tokens=128,
    output_messages_key="llm_input_messages",
)

class State(AgentState):
    # 为SummarizationNode添加，以便能够跟踪运行中的摘要信息
    context: dict[str, RunningSummary]


checkpointer = InMemorySaver()

agent = create_agent(
    model=model,
    tools=[],
    pre_model_hook=summarization_node,
    state_schema=State,
    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)

print(final_response.keys())

final_response["messages"][-1].pretty_print()
print("\n摘要:", final_response["context"]["running_summary"].summary)

访问

您可以通过以下几种不同方式访问代理的短期记忆：

工具
预模型钩子
后模型钩子

工具

在工具中读取短期记忆

通过在工具签名中使用 InjectedState 注解将代理的状态注入工具签名，从而在工具中访问短期记忆（状态）。此注解会将状态从工具签名中隐藏（因此模型看不到它），但工具可以访问它。

from typing import Annotated
from langchain.agents import create_agent, AgentState
from langchain.agents.tool_node import InjectedState

class CustomState(AgentState):
    user_id: str

def get_user_info(
    state: Annotated[CustomState, InjectedState]
) -> str:
    """查找用户信息。"""
    user_id = state["user_id"]
    return "用户是John Smith" if user_id == "user_123" else "未知用户"

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

result = agent.invoke({
    "messages": "查找用户信息",
    "user_id": "user_123"
})
print(result["messages"][-1].content)
# > 用户是John Smith。

从工具中写入短期记忆

要在执行过程中修改代理的短期记忆（状态），您可以直接从工具中返回状态更新。这对于持久化中间结果或使信息可供后续工具或提示访问非常有用。

from typing import Annotated
from langchain_core.tools import InjectedToolCallId
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import ToolMessage
from langchain.agents import create_agent, AgentState
from langchain.agents.tool_node import InjectedState
from langgraph.runtime import get_runtime
from langgraph.types import Command
from pydantic import BaseModel

class CustomState(AgentState):
    user_name: str

class CustomContext(BaseModel):
    user_id: str

def update_user_info(
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
    """查找并更新用户信息。"""
    runtime = get_runtime(CustomContext)
    user_id = runtime.context.user_id
    name = "John Smith" if user_id == "user_123" else "Unknown user"
    return Command(update={
        "user_name": name,
        # 更新消息历史
        "messages": [
            ToolMessage(
                "成功查找用户信息",
                tool_call_id=tool_call_id
            )
        ]
    })

def greet(
    state: Annotated[CustomState, InjectedState]
) -> str:
    """找到用户信息后使用此工具向用户打招呼。"""
    user_name = state["user_name"]
    return f"你好 {user_name}!"

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

agent.invoke(
    {"messages": [{"role": "user", "content": "向用户打招呼"}]},
    context=CustomContext(user_id="user_123"),
)

提示

通过将代理的状态注入提示函数签名，在动态提示函数中访问短期记忆（状态）。

from langchain_core.messages import AnyMessage
from langchain.agents import create_agent, AgentState
from langgraph.runtime import get_runtime
from typing import TypedDict


class CustomContext(TypedDict):
    user_name: str


def get_weather(city: str) -> str:
    """获取城市的天气。"""
    return f"{city}的天气总是晴朗的！"


def prompt(state: AgentState) -> list[AnyMessage]:
    user_name = get_runtime(CustomContext).context["user_name"]
    system_msg = f"您是一个有用的助手。请称呼用户为 {user_name}。"
    return [{"role": "system", "content": system_msg}] + state["messages"]


agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[get_weather],
    prompt=prompt,
    context_schema=CustomContext,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "旧金山的天气如何？"}]},
    context=CustomContext(user_name="John Smith"),
)
for msg in result["messages"]:
    msg.pretty_print()

输出

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

旧金山的天气如何？
================================== Ai Message ==================================
工具调用：
  get_weather (call_WFQlOGn4b2yoJrv7cih342FG)
 调用ID: call_WFQlOGn4b2yoJrv7cih342FG
  参数：
    city: 旧金山
================================= Tool Message =================================
名称: get_weather

旧金山的天气总是晴朗的！
================================== Ai Message ==================================

嗨，John Smith，旧金山的天气总是晴朗的！

预模型钩子

通过将代理的状态注入钩子签名，在预模型钩子中访问短期记忆（状态）。

from langchain_core.messages.utils import trim_messages, count_tokens_approximately
from langchain_core.messages import BaseMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState


def pre_model_hook(state: AgentState) -> dict[str, list[BaseMessage]]:
    """
    此函数将在每次调用LLM之前被调用，以准备LLM的消息。
    """
    trimmed_messages = trim_messages(
        state["messages"],
        strategy="last",
        token_counter=count_tokens_approximately,
        max_tokens=384,
        start_on="human",
        end_on=("human", "tool"),
    )
    return {"llm_input_messages": trimmed_messages}

agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[],
    pre_model_hook=pre_model_hook,
    checkpointer=InMemorySaver(),
)

result = agent.invoke({"messages": "hi, my name is bob"}, {"configurable": {"thread_id": "1"}})
print(result["messages"][-1].content)

后模型钩子

通过将代理的状态注入钩子签名，在后模型钩子中访问短期记忆（状态）。

from langchain.agents import create_agent, AgentState

STOP_WORDS = ["密码", "秘密"]

def validate_response(state: AgentState) -> dict[str, list[BaseMessage]]:
    """确认响应内容不包含停用词列表中的任何内容。"""
    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 {}

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

快速开始

核心组件

进阶使用

在生产环境使用

概述

使用方法

在生产环境中

自定义代理记忆

常见模式

裁剪消息

删除消息

总结消息

自定义策略

裁剪消息

删除消息

总结消息

访问

工具

在工具中读取短期记忆

从工具中写入短期记忆

提示

预模型钩子

后模型钩子

​概述

​使用方法

​在生产环境中

​自定义代理记忆

​常见模式

裁剪消息

删除消息

总结消息

自定义策略

​裁剪消息

​删除消息

​总结消息

​访问

​工具

​在工具中读取短期记忆

​从工具中写入短期记忆

​提示

​预模型钩子

​后模型钩子

概述

使用方法

在生产环境中

自定义代理记忆

常见模式

裁剪消息

删除消息

总结消息

访问

工具

在工具中读取短期记忆

从工具中写入短期记忆

提示

预模型钩子

后模型钩子