跳到主要内容

Agent 短期记忆

4.1 短期记忆介绍​

短期记忆（Short-term Memory），也称为线程范围记忆（Thread-scoped Memory），是指 Agent 在单个线程/会话中记住先前交互的能力。其核心机制通过 Checkpointer 实现状态持久化，使用 thread_id 隔离不同会话的上下文。

短期记忆中保存的状态不仅包含对话历史（messages），还可以存储自定义的业务数据（如用户信息、会话变量等），从而实现更精细化的会话管理。

基本用法：

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

# 创建内存检查点保存器
checkpointer = InMemorySaver()

# 创建 Agent，通过 checkpointer 参数启用短期记忆
agent = create_agent(
    model=deepseek_llm,
    tools=[],
    checkpointer=checkpointer
)

# 通过 thread_id 隔离不同用户的对话上下文
config = {"configurable": {"thread_id": "conversation_1"}}

# 第一轮对话：告知姓名
response = agent.invoke({"messages": [{"role": "user", "content": "你好，我叫张三"}]}, config)
print(response["messages"][-1].content)

print("="*20)

# 第二轮对话：验证 Agent 是否记住了姓名
response = agent.invoke({"messages": [{"role": "user", "content": "我叫什么名字?"}]}, config)
print(response["messages"][-1].content)

要点说明：

  • create_agent 中的 checkpointer 参数用于启用短期记忆功能
  • thread_id 隔离不同用户的对话上下文，相同 thread_id 共享记忆
  • 每次 invoke 后，会话状态自动保存到 Checkpointer 中

4.2 短期记忆使用方式​

Checkpointer 是短期记忆的核心架构组件，负责状态的持久化管理。根据存储介质不同，可以选择内存存储（开发测试）或数据库存储（生产环境）。

4.2.1 使用内存存储短期记忆​

InMemorySaver 将检查点存储在内存中，适合开发和测试环境。以下示例展示带工具的完整用法：

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.tools import tool
from init_llm import deepseek_llm

@tool
def get_user_info(name: str) -> str:
    """根据姓名查询用户信息"""
    user_db = {
        "张三": {"age": 28, "hobby": "旅游、滑雪、喝茶"},
        "李四": {"age": 32, "hobby": "编程、阅读、电影"}
    }
    info = user_db.get(name, {"age": "未知", "hobby": "未知"})
    return f"姓名: {name}, 年龄: {info['age']}岁, 爱好: {info['hobby']}"

# 创建内存检查点
checkpointer = InMemorySaver()

# 创建 Agent，绑定工具和检查点
agent = create_agent(
    model=deepseek_llm,
    tools=[get_user_info],
    checkpointer=checkpointer
)

# 配置会话 ID
config = {"configurable": {"thread_id": "user_123"}}

print("=== 第一轮对话 ===")
result1 = agent.invoke({"messages": "你好，我叫张三"}, config)
print(f"AI: {result1['messages'][-1].content}")

print("\n=== 第二轮对话 ===")
result2 = agent.invoke({"messages": "你知道我的信息吗？"}, config)
print(f"AI: {result2['messages'][-1].content}")

# 查看当前会话状态
print("="*20)
state = agent.get_state(config)
print(type(state))  # StateSnapshot 类型
print(state)        # 包含 messages 等属性

要点说明：

  • agent.get_state(config) 返回 StateSnapshot 对象，其中包含 messages 属性，记录了完整的对话历史
  • 内存存储在进程重启后会丢失，仅适用于开发测试场景

4.2.2 使用数据库存储短期记忆​

生产环境推荐使用数据库 Checkpointer，确保数据持久化。以下案例使用 MySQL（PyMySQLSaver）。

前置准备 - 创建数据库：

create database langchain_db;

安装依赖：

pip install langgraph-checkpoint-mysql==3.0.0 pymysql==1.1.2 cryptography==46.0.3

代码示例：

from langchain.agents import create_agent
from langchain_core.tools import tool
from init_llm import deepseek_llm
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver

@tool
def get_user_info(name: str) -> str:
    """根据姓名查询用户信息"""
    user_db = {
        "张三": {"age": 28, "hobby": "旅游、滑雪、喝茶"},
        "李四": {"age": 32, "hobby": "编程、阅读、电影"}
    }
    info = user_db.get(name, {"age": "未知", "hobby": "未知"})
    return f"姓名: {name}, 年龄: {info['age']}岁, 爱好: {info['hobby']}"

# MySQL 连接字符串
DB_URI = "mysql+pymysql://root:123456@localhost:3306/langchain_db?charset=utf8mb4"

# 使用上下文管理器确保连接正确关闭
with PyMySQLSaver.from_conn_string(DB_URI) as checkpointer:
    # 首次运行自动创建所需表结构
    checkpointer.setup()

    agent = create_agent(
        model=deepseek_llm,
        tools=[get_user_info],
        checkpointer=checkpointer
    )

    config = {"configurable": {"thread_id": "user_001"}}

    # 查看初始状态
    print(agent.get_state(config))

    # 第一轮对话
    agent.invoke({"messages": [{"role": "user", "content": "我是用户张三"}]}, config)

    # 第二轮对话，验证记忆持久化
    response = agent.invoke({"messages": [{"role": "user", "content": "我是谁？"}]}, config)
    print(f"AI响应: {response['messages'][-1].content}")

要点说明：

  • PyMySQLSaver.from_conn_string 创建 MySQL 持久化连接
  • checkpointer.setup() 在首次运行时自动创建所需的数据库表
  • 如果使用 PostgreSQL，可安装 langgraph-checkpoint-postgres 包，用法类似

4.3 自定义记忆状态​

默认的 AgentState 只包含 messages 字段。通过继承 AgentState 扩展自定义字段，可以实现更精细化的状态管理，存储业务相关数据。

实现步骤：

  1. 继承 AgentState，定义自定义字段
  2. 创建 Agent 时通过 state_schema 参数指定自定义状态类
  3. invoke 时传入自定义数据
from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver
from init_llm import deepseek_llm

# 步骤 1：扩展 AgentState，添加自定义字段
class CustomAgentState(AgentState):
    user_id: str
    hobby: list
    other_info: dict

# 步骤 2：创建 Agent，指定 state_schema
agent = create_agent(
    model=deepseek_llm,
    tools=[],
    state_schema=CustomAgentState,
    checkpointer=InMemorySaver()
)

config = {"configurable": {"thread_id": "user_001"}}

# 步骤 3：首次 invoke 时传入自定义字段
result = agent.invoke(
    {
        "messages":[{"role":"user","content":"你好，我是张三"}],
        "user_id": "user_001",
        "hobby": ["旅游、滑雪、喝茶"],
        "other_info": {"age": 28, "gender": "男"},
    },
    config=config
)
print("AI回复:", result["messages"][-1].content)
print("当前状态:", agent.get_state(config=config))

print("="*20)

# 后续对话无需再传自定义字段，同一 thread_id 自动携带
result = agent.invoke(
    {"messages": [{"role":"user","content":"使用十个字介绍你自己"}]},
    config=config
)
print("AI回复:", result["messages"][-1].content)
print("当前状态:", agent.get_state(config=config))

要点说明：

  • state_schema 参数用于指定自定义状态类
  • 首次 invoke 传入自定义字段后，同一 thread_id 的后续对话会自动携带这些字段
  • 自定义字段与 messages 同级，均作为状态的一部分被 Checkpointer 持久化

4.4 短期记忆访问和修改​

在 Agent 运行过程中，可以通过三种方式访问和修改短期记忆：Tools、@before_model 中间件、@after_model 中间件。这些方式基于 AgentState 和 ToolRuntime 实现。

4.4.1 通过 Tools 访问和修改短期记忆​

ToolRuntime 对象包含 Agent 的 state、context、store 等信息，作为工具的隐藏参数，不会暴露给 LLM。

4.4.1.1 通过 Tools 读取短期记忆​

以下示例定义两个工具，分别通过 ToolRuntime 读取不同的自定义状态字段：

from langchain.agents import AgentState, create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool, ToolRuntime
from init_llm import deepseek_llm

@tool
def get_info(runtime: ToolRuntime) -> str:
    """查询用户会员等级"""
    print("runtime:", runtime)
    user_level = runtime.state["user_level"]
    if user_level == "VIP":
        return "你的会员等级是VIP，你有免费退换货、专属客服通道等福利"
    else:
        return "你的会员等级是普通会员，你有积分翻倍活动等福利"

@tool
def get_user_id(runtime: ToolRuntime) -> str:
    """查询用户唯一标识"""
    print("runtime:", runtime)
    user_id = runtime.state["user_id"]
    return f"你的用户唯一标识是：{user_id}"

# 自定义状态，包含用户 ID 和会员等级
class CustomerState(AgentState):
    user_id: str
    user_level: str

agent = create_agent(
    model=deepseek_llm,
    tools=[get_info, get_user_id],
    state_schema=CustomerState,
    checkpointer=InMemorySaver()
)

config = {"configurable": {"thread_id": "session_001"}}

# 首次调用传入自定义状态字段
response1 = agent.invoke(
    {
        "messages": [{"role": "user", "content": "我的会员等级是什么？"}],
        "user_id": "user_123",
        "user_level": "VIP"
    },
    config=config
)
print(response1["messages"][-1].content)

# 后续调用无需再传自定义字段
response2 = agent.invoke(
    {"messages": [{"role": "user", "content": "查看我的用户唯一标识"}]},
    config=config
)
print(response2["messages"][-1].content)

要点说明：

  • ToolRuntime 是隐藏参数，LLM 不会看到，仅供工具内部使用
  • 只需在首次调用时传入自定义状态字段，后续调用自动从 Checkpointer 加载

4.4.1.2 通过 Tools 修改短期记忆​

工具可以通过返回 Command 对象来修改状态。注意：Command 的 update 中必须包含 messages 字段（ToolMessage 类型）。

from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool, ToolRuntime
from langgraph.types import Command
from langchain.messages import ToolMessage
from init_llm import deepseek_llm

class CustomerState(AgentState):
    user_name: str = ""
    hobby: list = []

@tool
def update_user_profile(runtime: ToolRuntime, name: str, hobby: list) -> Command:
    """更新用户档案信息并持久化到记忆状态"""
    # 参数校验
    if not name or not hobby:
        return Command(update={
            "messages": [ToolMessage(content="错误：姓名和爱好不能为空", tool_call_id=runtime.tool_call_id)]
        })

    # 构建状态更新，必须包含 messages 字段
    updates = {
        "user_name": name,
        "hobby": hobby,
        "messages": [ToolMessage(content=f"已更新用户档案：姓名={name}, 爱好={','.join(hobby)}", tool_call_id=runtime.tool_call_id)]
    }
    return Command(update=updates)

agent = create_agent(
    model=deepseek_llm,
    tools=[update_user_profile],
    state_schema=CustomerState,
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "session_001"}}

# 第一轮：触发工具更新用户档案
result1 = agent.invoke({"messages": [{"role": "user", "content": "我叫王五，我的爱好是钓鱼和唱歌"}]}, config)
print("模型回复:", result1['messages'][-1].content)

print("="*20)

# 第二轮：继续对话，状态已更新
result2 = agent.invoke({"messages": [{"role": "user", "content": "我也喜欢旅游"}]}, config)
print("模型回复:", result2['messages'][-1].content)

print("="*20)

# 查看完整状态
print("当前状态:", agent.get_state(config=config))

要点说明：

  • Command 对象的 update 字典中必须包含 messages 字段，且类型为 ToolMessage
  • ToolMessage 需要通过 tool_call_id 关联到对应的工具调用

4.4.2 @before_model 中间件操作短期记忆​

@before_model 是前置拦截器，在模型调用前触发。其参数为 AgentState 和 Runtime，可用于预加载数据、注入上下文或修改状态。

from langchain.agents import AgentState
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from typing import Any, Dict
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from init_llm import deepseek_llm

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气是晴朗的，温度是25摄氏度"

# 自定义状态，包含工具调用计数
class CustomState(AgentState):
    tool_call_count: int

@before_model
def manage_state(state: CustomState, runtime: Runtime) -> Dict[str, Any] | None:
    """在模型调用前统计工具调用次数并更新状态"""
    print("before_model_state：", state)
    print("before_model_runtime：", runtime)

    # 读取当前工具调用次数
    tool_call_count = state.get("tool_call_count", 0)
    print("状态中工具调用次数:", tool_call_count)

    # 统计消息中实际的 ToolMessage 数量
    tool_call_count = len([msg for msg in state["messages"] if isinstance(msg, ToolMessage)])
    return {"tool_call_count": tool_call_count}

agent = create_agent(
    model=deepseek_llm,
    tools=[get_weather],
    middleware=[manage_state],
    checkpointer=InMemorySaver(),
    state_schema=CustomState
)

config = {"configurable": {"thread_id": "session_001"}}

response1 = agent.invoke({"messages": [{"role": "user", "content": "北京今天天气如何？"}]}, config=config)
print(response1["messages"][-1].content)
print("***" * 20)

response2 = agent.invoke({"messages": [{"role": "user", "content": "上海今天天气如何？"}]}, config=config)
print(response2["messages"][-1].content)
print("***" * 20)

# 查看最终状态，包含工具调用次数
print(agent.get_state(config=config))

要点说明：

  • @before_model 装饰器的函数参数为 AgentState 和 Runtime
  • 创建 Agent 时需通过 state_schema 指定自定义状态类型
  • 中间件通过 middleware 参数传入，支持多个中间件组合

4.4.3 @after_model 中间件操作短期记忆​

@after_model 是后置拦截器，在模型生成响应后触发。适合用于后处理结果、提取结构化数据、更新业务状态等场景。

以下案例展示一个订单查询 + 库存查询场景：通过 @after_model 从结构化输出中提取商品名并更新到状态中，供后续工具使用。

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import after_model
from langchain.agents.structured_output import ToolStrategy
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import ToolRuntime
from langgraph.runtime import Runtime
from pydantic import BaseModel, Field
from typing import Dict, Any, Union
from init_llm import deepseek_llm

# 结构化输出模型：订单查询结果
class OrderQueryResult(BaseModel):
    order_id: str
    product_name: str
    price: float
    status: str

# 结构化输出模型：库存查询结果
class InventoryQueryResult(BaseModel):
    product_name: str
    stock_quantity: int

# 模拟数据库
MOCK_DATABASE = {
    "orders": {
        "order_001": OrderQueryResult(order_id="order_001", product_name="华为手机", price=1999.00, status="已发货"),
        "order_002": OrderQueryResult(order_id="order_002", product_name="苹果电脑", price=2999.00, status="待发货"),
        "order_003": OrderQueryResult(order_id="order_003", product_name="三星显示器", price=3999.00, status="已签收"),
    },
    "inventory": {
        "华为手机": InventoryQueryResult(product_name="华为手机", stock_quantity=50),
        "苹果电脑": InventoryQueryResult(product_name="苹果电脑", stock_quantity=20),
        "三星显示器": InventoryQueryResult(product_name="三星显示器", stock_quantity=30)
    }
}

# 自定义状态，存储当前查询的商品名
class OrderState(AgentState):
    product_name: str

@tool
def get_order_info(order_id: str) -> OrderQueryResult:
    """获取订单详情"""
    order_data = MOCK_DATABASE["orders"].get(order_id)
    if order_data:
        return order_data
    else:
        raise ValueError("订单不存在")

@tool
def get_product_inventory(runtime: ToolRuntime) -> InventoryQueryResult:
    """查询商品库存"""
    print("runtime:", runtime)
    # 从状态中获取商品名，无需用户再次输入
    product_name = runtime.state["product_name"]
    inventory_data = MOCK_DATABASE["inventory"].get(product_name)
    if inventory_data:
        return inventory_data
    else:
        raise ValueError("商品不存在")

@after_model
def manage_order_state(state: AgentState, runtime: Runtime) -> Dict[str, Any] | None:
    """模型响应后，从结构化输出中提取商品名更新到状态"""
    print("state:", state)
    if "structured_response" not in state:
        return None
    structured_response = state["structured_response"]
    # 如果是订单查询结果，提取商品名存入状态
    if isinstance(structured_response, OrderQueryResult):
        product_name = structured_response.product_name
        return {"product_name": product_name}
    else:
        return None

agent = create_agent(
    model=deepseek_llm,
    tools=[get_order_info, get_product_inventory],
    response_format=ToolStrategy(Union[OrderQueryResult, InventoryQueryResult]),
    middleware=[manage_order_state],
    state_schema=OrderState,
    checkpointer=InMemorySaver()
)

config = {"configurable": {"thread_id": "user_001"}}

# 查询订单 order_001，@after_model 提取商品名到状态
response1 = agent.invoke({"messages": [{"role": "user", "content": "查询订单order_001信息"}]}, config=config)
print("response1:", response1["structured_response"])
print("***" * 20)

# 查询库存，工具从状态中自动获取商品名"华为手机"
response2 = agent.invoke({"messages": [{"role": "user", "content": "这个订单中商品库存是多少"}]}, config=config)
print("response2:", response2["structured_response"])
print("***" * 20)

# 切换到 order_002，状态中商品名更新为"苹果电脑"
response3 = agent.invoke({"messages": [{"role": "user", "content": "查询订单order_002信息"}]}, config=config)
print("response3:", response3["structured_response"])
print("***" * 20)

# 再次查询库存，自动使用"苹果电脑"
response4 = agent.invoke({"messages": [{"role": "user", "content": "商品库存是多少"}]}, config=config)
print("response4:", response4["structured_response"])
print("***" * 20)

# 查看最终状态中的商品名
final_state = agent.get_state(config=config)
print("最终 product_name 状态:", final_state.values["product_name"])

4.5 State 和 Context 区别​

调用 agent.invoke 时可以传入 State（状态数据）和 Context（上下文数据），两者的核心区别如下：

  • State：管理短期记忆，动态变化，由 Checkpointer 自动持久化
  • Context：提供静态/半静态上下文信息，默认不自动持久化

传参方式：

agent.invoke(
    {
        "messages": [{"role": "user", "content": "你的问题"}],
        "state_key1": "state_value1",
    },
    config=config,
    context={
        "context_key1": "context_value1",
    }
)

对比表：

对比项StateContext
设计目的记录会话中动态演变的信息（聊天历史、购物车商品等）提供稳定背景信息与环境参数（地理位置、设备信息等）
可变性高，动态变化低，相对静态
持久化与生命周期由 Checkpointer 管理，自动持久化，同一 thread_id 自动加载默认不自动持久化，每次调用需传入
会话特点会话隔离，与 thread_id 绑定invoke 执行内共享，被所有工具和中间件访问

完整案例：在工具和 @after_model 中间件中分别获取 Context 和 State，并在中间件中将 Context 值设置到 State 中以便后续调用获取。

from langchain.agents import AgentState, create_agent
from langchain_core.tools import tool
from langchain.agents.middleware import after_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import ToolRuntime
from langgraph.runtime import Runtime
from typing import Any, Dict
from init_llm import deepseek_llm

# 自定义状态，包含用户名、渠道和 LLM 调用次数
class ConversationState(AgentState):
    user_name: str
    channel: str
    call_llm_count: int

@tool
def get_weather(city: str, runtime: ToolRuntime) -> str:
    """获取指定城市的天气"""
    context = runtime.context
    state = runtime.state

    # 优先从 context 获取，若无则从 state 获取
    if context:
        user_name = context.get("user_name", "未知用户")
        channel = context.get("channel", "未知渠道")
        print(f"[get_weather] 获取到上下文：用户 {user_name} 来自 {channel}")
    else:
        user_name = state.get("user_name", "未知用户")
        channel = state.get("channel", "未知渠道")
        print(f"[get_weather] 获取到状态：用户 {user_name} 来自 {channel}")

    current_call_llm_count = state.get("call_llm_count", 0)
    print(f"[get_weather] 获取到状态：当前大模型调用次数 {current_call_llm_count}")
    return f"{city} 的天气晴朗！"

@after_model
def my_middleware(state: AgentState, runtime: Runtime) -> Dict[str, Any] | None:
    """将 context 中的值同步到 state，确保后续调用可获取"""
    context = runtime.context

    if context:
        user_name = context.get('user_name', "未知用户")
        channel = context.get('channel', "未知渠道")
    else:
        user_name = state.get('user_name', "未知用户")
        channel = state.get('channel', "未知渠道")

    call_llm_count = state.get('call_llm_count', 0)
    current_llm_call_count = call_llm_count + 1

    # 将 context 值写入 state，后续调用不传 context 也能获取
    return {
        "user_name": user_name,
        "channel": channel,
        "call_llm_count": current_llm_call_count
    }

agent = create_agent(
    model=deepseek_llm,
    tools=[get_weather],
    middleware=[my_middleware],
    checkpointer=InMemorySaver(),
    state_schema=ConversationState
)

config = {"configurable": {"thread_id": "session_123"}}

# 第一次调用：传入 context 和 state
response1 = agent.invoke(
    {
        "messages": [{"role": "user", "content": "北京今天天气如何？"}],
        "call_llm_count": 0,
    },
    config=config,
    context={"user_name": "张三", "channel": "App"}
)
print("response1:", response1["messages"][-1].content)
print("当前完整状态:", agent.get_state(config=config))

print("=" * 50)

# 第二次调用：不传 context，中间件从 state 中获取之前保存的值
response2 = agent.invoke(
    {"messages": [{"role": "user", "content": "上海呢？"}]},
    config=config,
)
print("response2:", response2["messages"][-1].content)
print("当前完整状态:", agent.get_state(config=config))

要点说明：

  • Context 不由 Checkpointer 自动保存，每次调用都需要显式传入
  • 可以在中间件中将 Context 的值设置到 State 中，实现"首次传入，后续自动携带"的效果

4.6 超出 LLM 上下文解决方案​

随着对话轮次增加，state["messages"] 会不断增长，可能导致超出 LLM 上下文窗口限制。以下提供四种解决方案：消息截断、消息删除、消息摘要、自定义策略。

4.6.1 Trim Message - 消息截断​

使用 @before_model 中间件，在模型调用前保留最近 N 条消息，通过 RemoveMessage 删除旧消息。

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from init_llm import deepseek_llm

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气是晴朗的，温度是25摄氏度"

@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict:
    """消息截断：保留最近 N 条消息，删除更早的消息"""
    print("当前state：", state)
    messages = state["messages"]

    # 超过 5 条消息时触发截断
    if len(messages) > 5:
        retain_msg_count = 3
        # 避免保留的第一条是 ToolMessage，导致工具调用链截断
        if messages[-retain_msg_count].type == "tool":
            retain_msg_count = 2
        print("删除的消息：", messages[:-retain_msg_count])
        return {"messages": [RemoveMessage(id=msg.id) for msg in messages[:-retain_msg_count]]}
    return None

agent = create_agent(
    model=deepseek_llm,
    tools=[get_weather],
    middleware=[trim_messages],
    checkpointer=InMemorySaver()
)

config = {"configurable": {"thread_id": "session_001"}}

response1 = agent.invoke({"messages": [{"role": "user", "content": "你好，我是张三"}]}, config=config)
print(response1["messages"][-1].content)
print("***"*20)

response2 = agent.invoke({"messages": [{"role": "user", "content": "今天北京天气好吗？"}]}, config=config)
print(response2["messages"][-1].content)
print("***"*20)

response3 = agent.invoke({"messages": [{"role": "user", "content": "上海天气怎么样？"}]}, config=config)
print(response3["messages"][-1].content)
print("***"*20)

# 由于早期消息被截断，Agent 可能无法回忆起姓名
final_response = agent.invoke({"messages": [{"role": "user", "content": "我的名字叫什么？"}]}, config=config)
print(final_response["messages"][-1].content)

要点说明：

  • 截断时需避免保留的第一条消息是 ToolMessage，否则会导致工具调用链不完整
  • RemoveMessage 通过标记消息 ID 来指示需要删除的消息

4.6.2 Delete Message - 消息删除​

使用 @after_model 中间件，实现永久删除特定消息。典型场景：用户说"删除历史聊天记录"时清空所有消息。

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langchain.messages import RemoveMessage
from langchain_core.messages import ToolMessage, AIMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.prebuilt import ToolRuntime
from langgraph.runtime import Runtime
from langgraph.types import Command
from init_llm import deepseek_llm

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气是晴朗的，温度是25摄氏度"

@tool
def update_delete_history_state(runtime: ToolRuntime, delete_history: bool) -> Command:
    """是否清空聊天历史记录"""
    updates = {
        "delete_history": delete_history,
        "messages": [ToolMessage(content=f"已更新删除聊天历史记录状态：{delete_history}", tool_call_id=runtime.tool_call_id)]
    }
    return Command(update=updates)

@after_model
def delete_messages(state: AgentState, runtime: Runtime) -> dict:
    """检测删除标志，清空所有聊天历史"""
    print("当前state：", state)
    delete_history = state.get("delete_history")
    if delete_history:
        return {
            "delete_history": False,
            "messages": [
                RemoveMessage(id=REMOVE_ALL_MESSAGES),
                AIMessage(content="聊天历史记录已经成功删除，现在我们的对话将从新的状态开始，有什么其他我可以帮助你的吗？")
            ]
        }
    return None

# 自定义状态，包含删除标志
class CustomState(AgentState):
    delete_history: bool

agent = create_agent(
    model=deepseek_llm,
    tools=[get_weather, update_delete_history_state],
    middleware=[delete_messages],
    checkpointer=InMemorySaver(),
    state_schema=CustomState
)

config = {"configurable": {"thread_id": "session_001"}}

response1 = agent.invoke({"messages": [{"role": "user", "content": "你好，我是张三"}], "delete_history": False}, config=config)
print(response1["messages"][-1].content)
print("***"*20)

response2 = agent.invoke({"messages": [{"role": "user", "content": "今天北京天气好吗？"}]}, config=config)
print(response2["messages"][-1].content)
print("***"*20)

response3 = agent.invoke({"messages": [{"role": "user", "content": "我的名字叫什么？"}]}, config=config)
print(response3["messages"][-1].content)
print("***"*20)

# 触发删除历史记录
response4 = agent.invoke({"messages": [{"role": "user", "content": "请给我删除聊天历史记录"}]}, config=config)
print(response4["messages"][-1].content)
print("***"*20)

# 删除后，Agent 无法回忆任何之前的信息
final_response = agent.invoke({"messages": [{"role": "user", "content": "我的名字叫什么？"}]}, config=config)
print(final_response["messages"][-1].content)

要点说明：

  • REMOVE_ALL_MESSAGES 可以高效地清空所有消息
  • 删除所有消息后，需要添加一条 AIMessage 作为新的起始消息，避免空消息列表导致回复错误

4.6.3 Summarize Message - 消息摘要​

使用 LangChain 内置的 SummarizationMiddleware，自动将早期对话压缩为摘要，保留关键信息。

基本用法：

agent = create_agent(
    model="gpt-4.1",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4.1-mini",
            trigger=("tokens", 4000),
            keep=("messages", 20)
        ),
    ],
)

参数说明：

参数说明
model必填，用于生成摘要的 LLM
trigger必填，触发摘要的条件（tokens / messages / fraction）
keep保留消息策略，默认 ("messages", 20)
token_counter自定义 token 计数函数
summary_prompt自定义摘要提示词模板（需包含 {messages} 占位符）
trim_tokens_to_summarize摘要最大 token 数，默认 4000

完整案例：

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model, SummarizationMiddleware, before_model
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from init_llm import deepseek_llm

@before_model
def print_before_model_state(state: AgentState, runtime: Runtime) -> dict|None:
    """打印模型调用前的状态，便于调试"""
    print("before_model_state：", state)
    messages = state["messages"]
    return {"messages": messages}

@after_model
def print_after_model_state(state: AgentState, runtime: Runtime) -> dict|None:
    """打印模型调用后的状态，便于调试"""
    print("after_model_state：", state)
    messages = state["messages"]
    return {"messages": messages}

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气是晴朗的，温度是25摄氏度"

agent = create_agent(
    model=deepseek_llm,
    tools=[get_weather],
    middleware=[
        print_before_model_state,
        print_after_model_state,
        SummarizationMiddleware(
            model=deepseek_llm,
            trigger=('messages', 5),       # 消息数超过 5 条触发摘要
            keep=('messages', 2),          # 保留最新 2 条消息
            summary_prompt="请总结以下对话内容：{messages}"
        )
    ],
    checkpointer=InMemorySaver()
)

config = {"configurable": {"thread_id": "session_001"}}

response1 = agent.invoke({"messages": [{"role": "user", "content": "你好，我是张三"}]}, config=config)
print(response1["messages"][-1].content)
print("***"*20)

response2 = agent.invoke({"messages": [{"role": "user", "content": "今天北京天气好吗？"}]}, config=config)
print(response2["messages"][-1].content)
print("***"*20)

response3 = agent.invoke({"messages": [{"role": "user", "content": "我的名字叫什么？"}]}, config=config)
print(response3["messages"][-1].content)

要点说明：

  • SummarizationMiddleware 在模型调用后、回复前进行摘要处理
  • 保留最新 N 条消息不做摘要，只对之前的消息进行压缩

4.6.4 Custom Strategies - 自定义策略​

使用 @after_model 实现自定义摘要策略，可以保留完整的工具调用链，实现更灵活的裁剪逻辑。

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model, before_model
from langchain_core.messages import RemoveMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime
from langchain_core.messages import SystemMessage
from init_llm import deepseek_llm, tongyi_llm

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气是晴朗的，温度是25摄氏度"

@before_model
def print_before_model_state(state: AgentState, runtime: Runtime) -> dict | None:
    """打印模型调用前的状态"""
    print("before_model_state：", state)
    messages = state["messages"]
    return {"messages": messages}

@after_model
def custom_summarizer(state: AgentState, runtime) -> dict | None:
    """自定义摘要：保留最近 N 条消息（含完整工具链），对之前消息摘要"""
    print("after_model_state：", state)
    messages = state["messages"]
    threshold = 5
    max_retain = 2

    # 未达到阈值，不做处理
    if len(messages) <= threshold:
        return None

    # 确定保留的最近消息
    recent_messages = messages[-max_retain:]

    # 如果保留的第一条是 ToolMessage，向前扩展以保留完整工具链
    while True:
        if isinstance(recent_messages[0], ToolMessage):
            max_retain += 1
            recent_messages = messages[-max_retain:]
        else:
            break

    # 需要摘要的早期消息
    early_messages = messages[:-max_retain]

    # 构建摘要提示词
    summary_prompt = f"""请将以下对话内容总结成一段简洁的摘要，保留重要信息和细节：
        对话历史：
        {"".join([f"{msg.type}: {msg.content}" for msg in early_messages])}
        摘要要求：
        1. 保留人物、地点、关键事件等重要信息
        2. 保持第三人称叙述
        3. 长度不超过200字
        4. 使用中文总结"""

    try:
        # 使用另一个 LLM 生成摘要
        summary_response = tongyi_llm.invoke(summary_prompt)
        summary_content = f"对话摘要: {summary_response.content}"
        summary_message = SystemMessage(content=summary_content)

        # 清空所有消息，用摘要 + 最近消息替代
        new_messages = [summary_message] + recent_messages
        return {
            "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *new_messages]
        }
    except Exception as e:
        print(f"摘要生成失败: {e}")
        return None

agent = create_agent(
    model=deepseek_llm,
    tools=[get_weather],
    middleware=[print_before_model_state, custom_summarizer],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "session_001"}}

response1 = agent.invoke({"messages": [{"role": "user", "content": "你好，我是张三"}]}, config=config)
print(response1["messages"][-1].content)
print("***" * 20)

response2 = agent.invoke({"messages": [{"role": "user", "content": "今天北京天气好吗？"}]}, config=config)
print(response2["messages"][-1].content)
print("***" * 20)

response3 = agent.invoke({"messages": [{"role": "user", "content": "我的名字叫什么？"}]}, config=config)
print(response3["messages"][-1].content)

要点说明：

  • 保留最近 N 条消息时，需检查是否包含完整的工具调用链（避免 ToolMessage 与对应的 AI 调用分离）
  • REMOVE_ALL_MESSAGES 清空所有消息后，使用 *new_messages 展开新消息列表
  • 可以使用不同的 LLM 生成摘要（如使用更便宜的模型），降低成本

裁剪策略对比​

策略优点缺点适用场景
Trim简单高效丢失早期上下文实时聊天
Delete精确控制需手动判断敏感信息清理
Summarize保留关键信息需额外 LLM 调用超长对话
Custom灵活可控实现复杂特殊业务需求