LangChain 短期记忆机制深度解读：从原理到生产实践

背景

随着基于大语言模型（LLM）的 AI Agent 应用逐渐从简单的问答演示走向复杂的实际业务场景，对话的连续性和上下文理解能力成为了决定用户体验的关键因素。然而，LLM 本身是无状态的，且受限于上下文窗口（Context Window）的大小，直接保留完整的对话历史不仅成本高昂，还容易因“长上下文性能下降”导致模型注意力分散、响应变慢甚至报错。

为了解决这一痛点，LangChain 框架引入了 Memory（记忆）系统，特别是短期记忆（Short-term Memory）机制。短期记忆旨在让 Agent 能够在单一线程或对话中记住先前的交互信息，从而从反馈中学习并适应用户偏好。本文基于 LangChain 和 LangGraph 的最新实践，深入解析短期记忆的工作原理、实现方式、状态管理扩展以及记忆清理策略。

核心内容

1. 短期记忆的基础概念

记忆的定义与作用 Memory 是一个记录先前交互信息的系统。对于 AI Agent 而言，记忆至关重要，因为它赋予了 Agent 以下能力：

• 记住之前的交互：保持对话的连贯性。
• 从反馈中学习：根据用户纠正或反馈调整行为。
• 适应用户偏好：个性化服务，如记住用户的口味、名字等。

随着 Agent 处理的任务复杂度增加，记忆能力对效率和用户满意度变得不可或缺。

短期记忆与 Thread

• Short-term memory：让应用能够记住单一线程（Thread）或对话中的先前交互。
• Thread：将会话中的多次交互组织在一起，类似于电子邮件客户端将消息归组到同一对话线程中的方式。
• Conversation history：是短期记忆最常见的形式。

为什么需要短期记忆？ 长对话对当前 LLM 构成了双重挑战：

1. 上下文窗口不足：完整的对话历史可能超出 LLM 的上下文窗口限制，导致上下文丢失或 API 报错。
2. 长上下文性能下降：即使模型支持极长的上下文，大多数 LLM 在长上下文下表现不佳。过多的历史消息会分散模型注意力（被过时或离题内容干扰），导致响应速度变慢且计算成本更高。

消息结构 Chat 模型通过 messages 接收上下文，主要包含以下类型：

• System message：提供指令或系统提示。
• Human message：用户输入。

在聊天应用中，消息在用户输入和模型回复之间交替出现，形成一条不断增长的消息列表。由于上下文窗口有限，许多应用需要采用技术来移除或“遗忘”过时信息。

2. 创建短期记忆的实现

在创建 Agent 时，需要显式指定一个 checkpointer（检查点保存器），用于持久化或管理状态。

内存中的短期记忆（测试环境） 适用于快速测试，使用 langgraph.checkpoint.memory 中的 InMemorySaver。需要注意的是，必须通过实例化 RunnableConfig 对象并指定 configurable 参数（如 thread_id）来区分不同的对话线程。

from langchain.messages import HumanMessage
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver

# 配置线程 ID
config = RunnableConfig(configurable={"thread_id": "1"})
# 创建带有内存检查点的 Agent
agent = create_agent(base_model, checkpointer=InMemorySaver())

# 第一次交互：记住名字
res = agent.invoke({"messages": HumanMessage(content="我叫张三")}, config=config)
print("第一次运行：", res["messages"][-1].content)

# 第二次交互：利用记忆回答
res1 = agent.invoke({"messages": HumanMessage(content="我叫什么名字？")}, config=config)
print("第二次运行：", res1["messages"][-1].content)

生产环境中的短期记忆 在生产环境中，通常使用专门的数据库（如 PostgreSQL、SQLite）来管理短期记忆，以确保数据的持久性和并发安全性。以 PostgreSQL 为例，使用 PostgresSaver。

from langgraph.checkpoint.postgres import PostgresSaver

config2 = RunnableConfig(configurable={"thread_id": "2"})

# 使用上下文管理器连接数据库并设置检查点
with PostgresSaver.from_conn_string(DATABASE_URL) as checkpointer:
    checkpointer.setup()
    agent = create_agent(base_model, checkpointer=checkpointer)
    
    # 第一次交互
    res = agent.invoke({"messages": [HumanMessage(content="我叫张三")]}, config=config2)
    print("第一次运行：", res["messages"][-1].content)

# 重新连接以验证记忆持久性
with PostgresSaver.from_conn_string(DATABASE_URL) as checkpointer:
    agent = create_agent(base_model, checkpointer=checkpointer)
    
    # 第二次交互：即使 Agent 实例重建，记忆依然存在
    res1 = agent.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config2)
    print("第二次运行：", res1["messages"][-1].content)
    
    # 查看检查点状态
    checkpoints = list(checkpointer.list(config2))
    print("checkpoint 数量：", len(checkpoints))
    latest = checkpointer.get_tuple(config2)
    print("latest tuple:", latest)

3. 自定义 Agent 状态（Customizing Agent Memory）

除了标准的消息历史，我们往往需要在 Agent 状态中存储额外的用户信息（如用户 ID、偏好设置等）。这可以通过继承 langchain.agents 中的 AgentState 来实现。

扩展 AgentState 定义一个新的 TypedDict 类，继承自 AgentState，并添加自定义字段。

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

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

# 创建 Agent 时传入自定义状态 schema
agent = create_agent(
    base_model,
    system_prompt="你是一个个人助手，帮助用户管理他们的个人信息和偏好设置。",
    state_schema=MyAgentState,
    checkpointer=InMemorySaver()
)

config3 = RunnableConfig(configurable={"thread_id": 3})

# 初始化状态并传入自定义字段
res = agent.invoke(
    MyAgentState(
        messages=[HumanMessage(content="我喜欢吃辣的")], 
        user_id="user123", 
        preferences={"food": "spicy"}
    ),
    config=config3
)
print("模型回复：", res['messages'][-1].content)

# 后续交互中，Agent 可以访问之前存储的 preferences
res = agent.invoke(
    MyAgentState(
        messages=[HumanMessage(content="我的名字是user123你能告诉我我的偏好吗？")], 
    ),
    config=config3
)
print("模型回复：", res['messages'][-1].content)

执行结果解读

• 第一次回复确认记住了“喜欢吃辣的”，并给出了相应的推荐策略。
• 第二次回复成功调用了之前存储的偏好信息，表明自定义状态字段已正确参与 Agent 的状态管理。

源码深度解析：为什么自定义 State 生效？

1. 输入约束：invoke 的输入参数类型标注为 _InputAgentState，其本质是一个最小输入约束，只要求包含 messages 字段。MyAgentState 继承自 AgentState（包含 messages），因此满足最低要求。
2. Schema 解析机制：在 langchain/agents/factory.py 中，create_agent() 接收 state_schema 参数。源码逻辑如下：
   • 如果显式传入 state_schema，则 base_state 变为传入的自定义类（如 MyAgentState）。
   • create_agent() 不会直接将单个 schema 交给 StateGraph，而是调用 _resolve_schemas() 对多个 schema 进行统一处理。
   • 最终交给 StateGraph