消息裁剪与过滤
Trim & Filter Messages

1 核心问题：无限增长的消息历史

在构建聊天机器人时，我们把对话消息存在 MessagesState 里，每轮对话都会追加新消息。表面上很完美——但隐藏着一个根本性的工程挑战：

下面用一个直观的 Token 消耗示意图来理解这个问题：

这个 Notebook 介绍了解决这一问题的三种策略，分别对应不同的使用场景和取舍：

2 MessagesState 与基础聊天图回顾

在深入讲解三种策略之前，先回顾 Notebook 建立的基础聊天图，它是后续所有变体的出发点。

定义初始消息列表

Notebook 首先创建了一个简单的消息列表，模拟一场关于"海洋哺乳动物"的对话：

from langchain_core.messages import AIMessage, HumanMessage

# 构建一个两条消息的初始对话历史
messages = [AIMessage("So you said you were researching ocean mammals?", name="Bot")]
messages.append(HumanMessage("Yes, I know about whales. But what others should I learn about?", name="Lance"))

for m in messages:
    m.pretty_print()
# 输出：
# =================================== Ai Message ==================================
# Name: Bot
# So you said you were researching ocean mammals?
# ================================ Human Message =================================
# Name: Lance
# Yes, I know about whales. But what others should I learn about?

最简单的聊天图（无任何消息管理）

Notebook 先建立了一个"裸奔"的聊天图：把 State 里所有消息原封不动地传给 LLM。

from langgraph.graph import MessagesState, StateGraph, START, END
from langchain_deepseek import ChatDeepSeek

llm = ChatDeepSeek(model="deepseek-v4-pro")

# 节点：直接把 state["messages"] 全部传给 LLM
def chat_model_node(state: MessagesState):
    return {"messages": llm.invoke(state["messages"])}
    # ↑ 无任何裁剪！所有历史都发给模型

builder = StateGraph(MessagesState)
builder.add_node("chat_model", chat_model_node)
builder.add_edge(START, "chat_model")
builder.add_edge("chat_model", END)
graph = builder.compile()

MessagesState 内置了什么？

MessagesState 是 LangGraph 提供的内置 State 类，等价于：

from langgraph.graph.message import add_messages
from typing_extensions import Annotated, TypedDict

class MessagesState(TypedDict):
    # Annotated + add_messages：新消息追加到列表，而非覆盖
    messages: Annotated[list, add_messages]

核心是 add_messages 这个 Reducer：它让每次节点返回的新消息追加到现有列表，而不是替换。这是消息积累的根本原因，也是为什么我们需要主动管理消息数量。

3 方案一：用 RemoveMessage 在图的 State 层删除消息

第一种策略是在图里插入一个专门的过滤节点，它负责从 State 中永久删除不需要的旧消息，然后再把更新后的 State 交给 LLM 节点。

RemoveMessage 是什么？

RemoveMessage 是 LangGraph 提供的特殊消息类型。当你把一个 RemoveMessage(id="某条消息的id") 返回给 State，add_messages Reducer 会识别出这是"删除指令"，并从 State 的消息列表中彻底移除对应 id 的消息。

from langchain_core.messages import RemoveMessage

# 过滤节点：只保留最近 2 条消息，删除其余所有旧消息
def filter_messages(state: MessagesState):
    # state["messages"][:-2] = 除最后2条外的所有消息
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
    return {"messages": delete_messages}
    # ↑ 返回一组"删除指令"，Reducer 会执行实际删除

# 聊天节点：此时 State 里只剩最近 2 条，正常传给 LLM
def chat_model_node(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

# 图：先过滤，再聊天
builder = StateGraph(MessagesState)
builder.add_node("filter", filter_messages)
builder.add_node("chat_model", chat_model_node)
builder.add_edge(START, "filter")   # 先过滤
builder.add_edge("filter", "chat_model")  # 再聊天
builder.add_edge("chat_model", END)
graph = builder.compile()

实际运行演示

Notebook 构造了一个 4 条消息的列表，并且显式指定了每条消息的 id（id="1"、id="2"……）——这是因为 RemoveMessage 需要凭 id 来找到目标消息：

# 构建 4 条消息的对话，显式指定 id 以便后续删除
messages = [AIMessage("Hi.", name="Bot", id="1")]
messages.append(HumanMessage("Hi.", name="Lance", id="2"))
messages.append(AIMessage("So you said you were researching ocean mammals?", name="Bot", id="3"))
messages.append(HumanMessage("Yes, I know about whales. But what others should I learn about?", name="Lance", id="4"))

# 运行：filter 节点会删除 id="1" 和 id="2" 的消息
output = graph.invoke({'messages': messages})

4 方案二：filter_messages——在节点内过滤，State 保持完整

第二种策略比第一种更"温和"：State 里的消息列表原封不动，只在节点内部临时过滤，传给 LLM 的是一个经过筛选的子集，但 State 本身不受影响。

核心思想

在聊天节点里，不把 state["messages"] 全部传给 LLM，而是先做一次简单的切片，只取最后 1 条（或 N 条）：

# 节点：只把最后 1 条消息传给 LLM
def chat_model_node(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"][-1:])]}
    #                                            ↑ [-1:] 只取最后 1 条
    #   state 里仍然保存着全部历史，只是传给 LLM 的只有最后 1 条

builder = StateGraph(MessagesState)
builder.add_node("chat_model", chat_model_node)
builder.add_edge(START, "chat_model")
builder.add_edge("chat_model", END)
graph = builder.compile()

State 与 LLM 入参的分离

这是理解此方案的关键：State 里存的 ≠ LLM 收到的。

Notebook 中的完整演示

Notebook 在构建 filter 图之后，进一步追加了新消息来测试多轮对话：

# 将上一次 LLM 的回复追加到消息列表
messages.append(output['messages'][-1])
# 追加新的用户问题
messages.append(HumanMessage("Tell me more about Narwhals!", name="Lance"))

# 此时 messages 列表有多条消息
for m in messages:
    m.pretty_print()

# 再次调用图（State 里有全部历史，但 LLM 只收到最后 1 条）
output = graph.invoke({'messages': messages})

过滤的灵活性：不只是切片

除了简单的列表切片 messages[-1:]，你还可以按消息类型过滤，例如只保留 HumanMessage，或者按 id 精确过滤：

from langchain_core.messages import filter_messages

# 只保留 HumanMessage（过滤掉所有 AI 消息）
human_only = filter_messages(state["messages"], include_types=[HumanMessage])

# 只保留 AIMessage
ai_only = filter_messages(state["messages"], include_types=[AIMessage])

# 排除某些类型
no_system = filter_messages(state["messages"], exclude_types=["system"])

# 按 name 过滤（只取某个 agent 的消息）
lance_msgs = filter_messages(state["messages"], include_names=["Lance"])

5 方案三：trim_messages——按 Token 数量精确裁剪

第三种策略是最精细化的管控方式：使用 trim_messages 函数，精确指定允许发送给 LLM 的最大 token 数，由函数自动决定保留哪些消息。

trim_messages 的工作原理

不同于手动切片（messages[-1:]），trim_messages 使用模型本身的 tokenizer 来计算每条消息的 token 数，然后从指定的方向（最新的或最旧的）开始，在不超过 max_tokens 限制的前提下，尽量多保留消息。

from langchain_core.messages import trim_messages

# 节点：使用 trim_messages 裁剪后再传给 LLM
def chat_model_node(state: MessagesState):
    messages = trim_messages(
        state["messages"],
        max_tokens=100,          # 最多允许 100 个 token
        strategy="last",          # 优先保留最新的消息
        token_counter=ChatDeepSeek(model="deepseek-v4-pro"),
                                   # 用这个模型来计算 token 数
        allow_partial=False,      # 不允许截断单条消息（要么整条保留，要么整条丢弃）
    )
    return {"messages": [llm.invoke(messages)]}

关键参数详解

strategy="last" 的行为可视化

假设有 5 条消息，总计 280 tokens，max_tokens=100，strategy="last"：

直接调用 trim_messages 观察结果

Notebook 也展示了在图之外直接调用 trim_messages 来预览裁剪效果：

# 直接调用 trim_messages 预览裁剪结果（不经过图）
trimmed = trim_messages(
    messages,
    max_tokens=100,
    strategy="last",
    token_counter=ChatDeepSeek(model="deepseek-v4-pro"),
    allow_partial=False
)

# 查看裁剪后的消息列表
# 会输出保留下来的 N 条消息，其余旧消息被排除在外
for m in trimmed:
    m.pretty_print()

strategy="first" 的用法

当你的场景需要保留对话开头（例如重要的系统指令或背景信息），可以改用 strategy="first"：

# strategy="first": 优先保留最旧的消息（从开头往后取）
trimmed = trim_messages(
    messages,
    max_tokens=100,
    strategy="first",  # ← 改为 first
    token_counter=llm,
    allow_partial=False,
    include_system=True  # 始终保留 SystemMessage
)

与图的集成：完整示例

# 使用 trim 方案构建图
def chat_model_node(state: MessagesState):
    trimmed = trim_messages(
        state["messages"],
        max_tokens=100,
        strategy="last",
        token_counter=ChatDeepSeek(model="deepseek-v4-pro"),
        allow_partial=False,
    )
    return {"messages": [llm.invoke(trimmed)]}

builder = StateGraph(MessagesState)
builder.add_node("chat_model", chat_model_node)
builder.add_edge(START, "chat_model")
builder.add_edge("chat_model", END)
graph = builder.compile()

# 在消息列表上追加更多消息后再调用
messages.append(output['messages'][-1])
messages.append(HumanMessage("Tell me where Orcas live!", name="Lance"))

# 调用图（State 包含全部历史，LLM 只收到 ≤100 token 的子集）
messages_out = graph.invoke({'messages': messages})

6 三种方案对比与选型指南

三种策略各有千秋，适用于不同的场景。下面从多个维度进行对比：

选型决策树

• 1

  你需要保留完整的对话历史吗？

  是 → 选 Filter 或 Trim；否（确定历史不再有用）→ 考虑 RemoveMessage

• 2

  你的消息长度是否差异很大？

  是（有长有短，不好预估 token）→ 选 trim_messages；否（消息相对均匀）→ Filter 切片已够用

• 3

  你是在做 PoC 还是生产环境？

  PoC/快速原型 → Filter 切片（最简单）；生产环境 → trim_messages（精确、可预测费用）

• 4

  需要处理 SystemMessage（系统提示词）吗？

  是 → 使用 trim_messages + include_system=True，确保系统提示词始终被保留

7 完整流程：消息列表随对话演化

Notebook 并非孤立地演示每种方案，而是贯穿了一个持续增长的消息列表，让你看到消息是如何在多轮对话中积累的，以及过滤/裁剪在哪个环节介入。

消息列表的完整演化时间线

filter 与 trim 在 LangSmith 中的行为差异

Notebook 分别提供了 filter 和 trim 两个版本的 LangSmith 追踪链接。关键观察点：

8 核心概念总结与最佳实践

最佳实践清单