TellWorldAI 世界档案
问档案English

首页›实践›实践

实践

LangChain 智能体的短期记忆

用 checkpointer 给 LangChain 智能体加上线程级记忆,生产环境改用数据库存状态;对话变长时用裁剪、删除或摘要控制历史。

这篇适合谁、读完能做到什么

这篇适合已经能用 LangChain 写出简单智能体、想让它记住同一段对话前面说过什么的开发者。读完你能给智能体加上线程级的短期记忆,把状态存进数据库,并知道对话变长以后该用裁剪、删除还是摘要来控制历史长度。

什么是短期记忆

短期记忆让应用记住同一个线程(thread)或同一段对话里之前的交互。线程可以类比邮件里的一个会话:它把一次会话里的多轮交互组织在一起,就像邮件软件把同一个话题的往来信件归在一起。[1]

对话历史是最常见的短期记忆形式。但对话一长就会出问题。模型一次能读入的内容有上限,叫上下文窗口(context window)。完整历史可能放不进去,结果是上下文丢失或直接报错。即便模型支持的上下文足够长,多数大语言模型在长上下文里的表现也会变差,容易被过时或跑题的内容带偏,同时响应变慢、费用升高。[1]

聊天模型接收的上下文是消息列表:系统消息(system message)放指令,人类消息(human message)放输入。每一轮里,用户消息和模型回复交替出现,列表越来越长。[1]

短期记忆和长期记忆要分清。短期记忆只管一个线程之内的事。如果需要跨对话记住信息,应该使用长期记忆,它把用户级或应用级的数据存下来,跨线程、跨会话地调取。[1]

上手:给智能体加上线程级记忆

要给智能体加短期记忆,在创建智能体时指定一个 checkpointer 即可。[1] LangChain 的智能体把短期记忆当作自身状态的一部分保存:状态放在图(graph)的 state 里,因此智能体能拿到当前对话的完整上下文,不同线程之间又互相隔离。状态通过 checkpointer 持久化,线程随时可以恢复。[1]

  1. 创建 checkpointer。本地试验可以用内存版,示例用的是 langgraph.checkpoint.memory 里的 InMemorySaver。[1]
  2. 创建智能体时,把它作为 checkpointer 参数传入。
  3. 每次调用时,在配置里传入 thread_id。同一个 thread_id 对应同一段对话的记忆。[1]

下面是一个自定义状态的完整 Python 例子,里面就包含了 InMemorySaver 和 thread_id 的用法:

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

class CustomAgentState(AgentState):  # [!code highlight]
    user_id: str  # [!code highlight]
    preferences: dict  # [!code highlight]

agent = create_agent(
    "gpt-6.1-sol",
    tools=[get_user_info],
    state_schema=CustomAgentState,  # [!code highlight]
    checkpointer=InMemorySaver(),
)

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

状态的更新时机也要清楚:短期记忆在智能体被调用时更新,工具调用这类步骤完成后也会更新;每一步开始时读取状态。[1]

生产环境换成数据库 checkpointer

内存版(InMemorySaver)把状态放在本进程的内存里,进程结束后这段对话的记忆就不在了,只适合本地试验和单元测试。调试时可以先用它把流程跑通,确认消息结构和工具之间的配合没有问题,再换成数据库版。数据库版把状态写到外部数据库里,线程可以在不同的运行中恢复,这才是上线需要的。生产环境应使用数据库支持的 checkpointer,可选的有 SQLite、Postgres 和 Azure Cosmos DB。[1] 下面以 Postgres 为例,Python 用 pip 安装:

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

用 uv 的话是:

uv add langgraph-checkpoint-postgres "psycopg[binary]"

默认安装的 langgraph-checkpoint-postgres 会带上不含 extras 的 psycopg(Psycopg 3),多数用户应使用 psycopg[binary]。[1] 接下来在代码里创建 PostgresSaver,并把它交给智能体:

from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver  # [!code highlight]

def get_user_info() -> str:
    """Look up information about the current user."""
    return "No user profile on file."

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup() # auto create tables in PostgreSQL
    agent = create_agent(
        "gpt-6.1-sol",
        tools=[get_user_info],
        checkpointer=checkpointer,  # [!code highlight]
    )

checkpointer.setup() 会在 PostgreSQL 里自动建表,所以第一次运行前不需要手动建表。[1] JavaScript 版本用连接字符串创建 PostgresSaver:

import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";

const DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable";
const checkpointer = PostgresSaver.fromConnString(DB_URI);

自定义智能体状态

默认情况下,智能体用 AgentState 管理短期记忆,对话历史放在 messages 键下。[1] 如果还要记录用户 ID、偏好之类的字段,就扩展 AgentState。Python 里通过 state_schema 参数把自定义状态传给 create_agent。[1] JavaScript 里则要通过带 stateSchema 的中间件(middleware)来加字段,推荐用 StateSchema 类定义,也支持普通的 Zod 对象。[1]

对话变长以后:裁剪、删除和摘要

常见做法有四种:裁剪消息(trim)、删除消息(delete)、摘要(summarize),以及按消息过滤等自定义策略。[1]

裁剪。 Python 里用 @before_model 装饰器定义一个中间件函数,它在每次调用模型之前运行。[1] 下面的例子保留第一条消息和最近的几条:

@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
        ]
    }

删除。 RemoveMessage 只有在状态键使用了 add_messages 归并器(reducer)时才会生效,默认的 AgentState 已经满足这一点。[1] 要一次清空所有消息,返回带 REMOVE_ALL_MESSAGES 的 RemoveMessage 即可,它从 langgraph.graph.message 导入。[1] 删除后要保证历史仍然合法:有的模型提供商要求历史以用户消息开头,多数要求带工具调用的助手消息后面紧跟对应的工具结果。[1]

摘要。 裁剪和删除都可能丢掉有用的信息,所以有些应用改用摘要。内置的 SummarizationMiddleware 用一个聊天模型对早先的历史做摘要,示例里设置的触发条件是 trigger=("tokens", 4000),保留设置是 keep=("messages", 20)。[1]

在工具和提示词里读写记忆

工具通过 runtime 参数读取状态,它的类型是 ToolRuntime。这个参数在工具签名里是隐藏的,模型看不到它,但工具能从中拿到状态。[1] 工具也能写回状态:返回一个 Command,里面带上状态更新,比如写入用户名,同时把一条 ToolMessage 加入消息历史。[1] 中间件还可以读取状态,生成动态提示词,例如用运行时上下文里的用户名称呼用户。[1]

限制和常见问题

  • 短期记忆的作用范围是单个线程。要跨对话保留信息,需要长期记忆。[1]
  • 内存版 checkpointer 不适合生产环境,应换成数据库支持的版本。[1]
  • 对话过长会超出上下文窗口,需要主动裁剪、删除或摘要。[1]
  • 删除消息后,历史必须仍然符合模型提供商的要求,否则可能出错。[1]

下一步

想看 checkpointer 的完整列表、持久化的细节,以及中间件的全部配置,可以接着读 LangChain 官方文档中的对应章节。[1]

本文改编自 LangChain 官方文档《Short-term memory》[1],原文以 MIT 许可证发布。

来源

  1. 1
    Short-term memoryLangChain docs (MIT)

引用本页: LangChain 智能体的短期记忆. TellWorld, https://www.tellworld.net/zh/practice/practice-langchain-memory. 更新于 2026-11-14

相关动态