一、短期记忆是什么？

▎ 短期记忆让您的应用程序能够记住单个线程或对话中的先前交互。

打个比方：短期记忆就是"这一通电话里的记性"——你跟智能体在这一个对话中聊过的所有内容，它都记得。但换一个对话（挂断重拨）
，就从头开始。

关键概念：线程（thread）

▎ 线程将一次会话中的多次交互组织起来，类似于电子邮件将消息归入单个对话中的方式。

一句话：一个线程 = 一次连续对话。短期记忆就是"按线程保存的记忆"。

---

二、为什么需要专门管理短期记忆？

几个痛点：

1. 上下文窗口有限：完整对话历史可能塞不进 LLM 的上下文窗口，导致"上下文丢失或错误"。
2. 长上下文表现变差：即使模型支持长上下文，也会因陈旧/无关内容"分心"，且响应变慢、成本增加。
3. 消息会不断增长：聊天中人类输入和模型回复交替，消息列表越来越长。

所以需要"移除或遗忘陈旧信息"的技术来管理记忆。

▎ 区分：要记住跨对话的信息，请用长期记忆（store），本文讲的是短期记忆（单线程内）。

---

三、怎么启用短期记忆？—— checkpointer

核心一句话：

▎ 要为智能体添加短期记忆，需要在创建智能体时指定一个 checkpointer（检查点器）。

checkpointer 就是负责"把状态存下来、下次能恢复"的组件。最简单的内存版：

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

  agent = create_agent(
      "gpt-5.4",
      tools=[get_user_info],
      checkpointer=InMemorySaver(),   # 关键：加 checkpointer
  )

  # 调用时用 thread_id 标识"这是哪个对话"
  agent.invoke(
      {"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
      {"configurable": {"thread_id": "1"}},   # 线程 ID
  )

两个关键点：

• checkpointer：决定记忆存哪（内存/数据库）
• thread_id：标识哪个对话线程，相同 thread_id = 同一个对话

生产环境：用数据库版 checkpointer

InMemorySaver 重启就丢，生产要用持久化的，比如 Postgres：

  pip install langgraph-checkpoint-postgres

  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()   # 自动建表
      agent = create_agent(
          "gpt-5.4",
          tools=[get_user_info],
          checkpointer=checkpointer,
      )

短期记忆的工作机制

• 短期记忆作为智能体状态的一部分管理，存在图状态里
• 通过 checkpointer 持久化到数据库（或内存），线程可随时恢复
• 更新时机：调用智能体或完成一步（如工具调用）时更新
• 读取时机：每一步开始时读取状态

---

四、自定义智能体记忆

默认智能体用 AgentState 管理短期记忆（通过 messages 键存对话历史）。你可以扩展 AgentState 添加额外字段：

  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.4",
      tools=[get_user_info],
      state_schema=CustomAgentState,   # 传入自定义状态
      checkpointer=InMemorySaver(),
  )

  # 调用时可以传入自定义状态字段
  result = agent.invoke(
      {
          "messages": [{"role": "user", "content": "Hello"}],
          "user_id": "user_123",
          "preferences": {"theme": "dark"}
      },
      {"configurable": {"thread_id": "1"}}
  )

这样除了对话历史，还能在这个线程里记住 user_id、preferences 等额外信息。

---

五、长对话的管理策略（核心！）

启用短期记忆后，长对话可能超出上下文窗口。页面给出四种常见策略：

策略	做法	适用场景
截断消息	调用 LLM 前移除部分消息（保留最近的）	防止超出 token 上限
删除消息	永久删除 LangGraph 状态中的消息	移除特定消息或清空历史
总结消息	总结早期消息，用摘要替换它们	既省空间又尽量不丢信息
自定义策略	如消息过滤等	特殊需求

---

策略 1：截断消息（@before_model 中间件）

用 @before_model 中间件，在调用模型之前修剪消息历史。思路是：保留第一条消息 + 最近几条，中间的丢掉：

  from langchain.messages import RemoveMessage
  from langgraph.graph.message import REMOVE_ALL_MESSAGES
  from langchain.agents import create_agent, AgentState
  from langchain.agents.middleware import before_model
  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

      first_msg = messages[0]                              # 保留第一条
      recent_messages = messages[-3:]                       # 保留最近几条
      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(),
  )

效果：即使中间聊了很多轮，模型也只看到"开头 + 最近几条"

---

策略 2：删除消息（RemoveMessage）

用 RemoveMessage 删除特定消息或全部消息。需要状态键带 add_messages reducer（默认 AgentState 已提供）。

删除特定消息（最早的 2 条）：

  from langchain.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)]}

⚠️ 重要提醒：删除消息后要确保消息历史仍然有效，否则模型会报错：

• 某些厂商要求消息历史以 user 消息开头
• 大多数厂商要求带工具调用的 assistant 消息后必须跟对应的 tool 结果消息（不能拆散）

实战中常用 @after_model 中间件在模型回复后删除旧消息：

  from langchain.agents.middleware import after_model

  @after_model
  def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
      """Remove old messages to keep conversation manageable."""
      messages = state["messages"]
      if len(messages) > 2:
          return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
      return None

  agent = create_agent(
      "gpt-5-nano",
      tools=[],
      system_prompt="Please be concise and to the point.",
      middleware=[delete_old_messages],
      checkpointer=InMemorySaver(),
  )

---

策略 3：总结消息（SummarizationMiddleware）

截断/删除的问题：会丢失信息。更聪明的方法是用模型把早期消息总结成摘要，再用摘要替换它们。

LangChain 提供内置的 SummarizationMiddleware：

  from langchain.agents import create_agent
  from langchain.agents.middleware import SummarizationMiddleware

  agent = create_agent(
      model="gpt-5.4",
      tools=[],
      middleware=[
          SummarizationMiddleware(
              model="gpt-5.4-mini",        # 用小模型做总结，省钱
              trigger=("tokens", 4000),     # 触发条件：token 达到 4000
              keep=("messages", 20)         # 保留最近 20 条消息不总结
          )
      ],
      checkpointer=InMemorySaver(),
  )

三个参数：

• model：用哪个模型做总结（推荐用便宜的小模型）
• trigger：什么时候触发总结（这里设为 token 数到 4000）
• keep：保留多少条最近消息不被总结

效果：长对话中，早期内容被压缩成摘要，模型既能看到历史梗概，又不会超出窗口

---

六、访问短期记忆的三种途径

短期记忆（状态）可以通过三个地方访问和修改：

途径 1：工具里访问（runtime.state）

用 ToolRuntime 参数在工具中读写状态（对模型隐藏）：

  from langchain.tools import tool, ToolRuntime
  from langchain.agents import create_agent, AgentState

  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"

工具还能写入短期记忆——直接返回 Command(update={…})：

  from langgraph.types import Command
  from langchain.messages import ToolMessage

  @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,                    # 写入自定义状态字段
          "messages": [
              ToolMessage(
                  "Successfully looked up user information",
                  tool_call_id=runtime.tool_call_id
              )
          ]
      })

---

途径 2：提示里访问（@dynamic_prompt 中间件）

根据状态/上下文生成动态系统提示：

  from langchain.agents.middleware import dynamic_prompt, ModelRequest

  class CustomContext(TypedDict):
      user_name: str

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

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

效果：系统提示会变成 “Address the user as John Smith”，模型回复时就会称呼用户名字。

---

途径 3：模型执行前/后访问（@before_model / @after_model）

• @before_model：调用模型前处理消息（如截断历史）
• @after_model：调用模型后处理消息（如删除含敏感词的回复）

  from langchain.agents.middleware import after_model
  @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

---

七、访问记忆途径总览

途径	装饰器/方式	时机	典型用途
工具	runtime: ToolRuntime	工具执行时	在工具中读写状态
提示	@dynamic_prompt	生成系统提示时	根据状态/上下文定制提示
模型执行前	@before_model	调用模型前	截断/修剪消息历史
模型执行后	@after_model	调用模型后	删除敏感词、清理消息

---

八、一句话总结

▎ 短期记忆 = 单个线程（对话）内的记忆，靠 checkpointer + thread_id 实现"同一对话记得住、能恢复"。
▎ 长对话靠四种策略管理：截断 / 删除 / 总结 /自定义。
▎ 访问记忆有四个切入点：工具（runtime.state）、动态提示（@dynamic_prompt）、模型执行前（@before_model）、模型执行后（@after_model）。

---

九、和前几讲的关系

把这几讲串起来：

• 短期记忆：线程内对话历史，存 state[“messages”]，靠 checkpointer 持久化
• 长期记忆（store）：跨线程永久保存，靠 runtime.store
• 工具访问上下文：工具用 runtime.state / runtime.context / runtime.store 读写这些记忆
• 消息（messages）：短期记忆的载体，四种类型组成对话历史

一句话：短期记忆是"这次通话的草稿纸"，checkpointer 是"草稿纸的存档"，thread_id 是"通话编号"。