在 LangGraph 中构建对话式 Agent 时，消息列表会随着对话轮次不断增加。如果不加以控制，消息数量会无限增长，可能导致：

• 超过 LLM 的上下文窗口（通常是几十 K 到几百 K tokens）
• 推理延迟增加（更多 tokens 需要处理）
• API 调用费用上涨（按 token 计费）

LangGraph 提供了内置的消息修剪（Pruning）功能，可以通过 add_messages reducer 配合 RemoveMessage 或者使用 LangGraph 内置的消息管理工具来实现。

🛠️ 方法一：使用 RemoveMessage 手动修剪（最灵活）

LangGraph 的 add_messages reducer 支持通过 RemoveMessage 对象来删除指定消息。

基本用法

from langgraph.graph.message import add_messages
from langchain_core.messages import RemoveMessage, AIMessage, HumanMessage
from typing import Annotated, TypedDict, List

class State(TypedDict):
    messages: Annotated[List, add_messages]

def trim_messages(state: State) -> dict:
    """修剪消息，只保留最近的 4 条"""
    messages = state["messages"]
    if len(messages) > 4:
        # 删除最早的消息（索引 0）
        return {"messages": [RemoveMessage(id=messages[0].id)]}
    return {}

# 在图中使用
from langgraph.graph import StateGraph

builder = StateGraph(State)
builder.add_node("trim", trim_messages)
builder.add_edge("trim", "__end__")

保留最近 N 条消息的完整示例

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import RemoveMessage, AnyMessage
from typing import Annotated, TypedDict, List

class State(TypedDict):
    messages: Annotated[List[AnyMessage], add_messages]

def keep_last_k_messages(state: State, k: int = 5) -> dict:
    """只保留最近 k 条消息，删除其余"""
    messages = state["messages"]
    if len(messages) <= k:
        return {}
    
    # 需要删除的消息 ID 列表
    to_remove = [RemoveMessage(id=m.id) for m in messages[:-k]]
    return {"messages": to_remove}

# 构建图
builder = StateGraph(State)
builder.add_node("chat", lambda state: {"messages": [llm.invoke(state["messages"])]})
builder.add_node("trim_msgs", keep_last_k_messages)

builder.add_edge(START, "chat")
builder.add_edge("chat", "trim_msgs")
builder.add_edge("trim_msgs", END)

🛠️ 方法二：使用 trim_messages 辅助函数（LangChain 提供）

LangChain 提供了一个独立的 trim_messages 函数，可以基于 token 或消息数量进行修剪。

安装依赖

pip install tiktoken  # token 计数需要

基于消息数量修剪

from langchain_core.messages import trim_messages, HumanMessage, AIMessage

messages = [
    HumanMessage(content="你好"),
    AIMessage(content="你好！有什么可以帮你？"),
    HumanMessage(content="今天天气怎么样？"),
    AIMessage(content="抱歉，我无法获取实时天气。"),
    HumanMessage(content="那你能做什么？"),
]

# 只保留最近 3 条消息
trimmed = trim_messages(
    messages,
    max_tokens=3,  # 保留3条消息（注意：这里的 token 不是实际token，而是消息条数）
    strategy="last",
    token_counter=len,  # 使用消息条数作为计数
)

print(len(trimmed))  # 输出 3

基于 Token 数量修剪（更精确）

from langchain_core.messages import trim_messages
import tiktoken

def count_tokens(messages):
    enc = tiktoken.encoding_for_model("gpt-3.5-turbo")
    total = 0
    for msg in messages:
        total += len(enc.encode(msg.content))
    return total

trimmed = trim_messages(
    messages,
    max_tokens=50,  # 最多保留 50 个 token
    strategy="last",
    token_counter=count_tokens,
    start_on="human",  # 确保以 human 消息开头
)

在 LangGraph 中使用 trim_messages

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import trim_messages
from typing import Annotated, TypedDict, List
from langchain_core.messages import AnyMessage

class State(TypedDict):
    messages: Annotated[List[AnyMessage], add_messages]

def smart_trim(state: State) -> dict:
    """智能修剪：保留最近 2000 个 token 的消息"""
    trimmed = trim_messages(
        state["messages"],
        max_tokens=2000,
        strategy="last",
        token_counter=count_tokens,  # 你需要定义这个函数
        include_system=True,          # 保留系统消息
    )
    return {"messages": trimmed}

🛠️ 方法三：在 Assistant 节点中直接过滤

你可以直接在调用 LLM 之前过滤历史消息，而不修改状态中的消息。

from langchain_core.messages import SystemMessage

def assistant(state: State) -> dict:
    """在调用 LLM 前动态修剪消息"""
    messages = state["messages"]
    
    # 方案 A：只保留最近 10 条
    if len(messages) > 10:
        messages = messages[-10:]
    
    # 方案 B：保留系统消息 + 最近 8 条
    sys_msgs = [m for m in messages if isinstance(m, SystemMessage)]
    other_msgs = [m for m in messages if not isinstance(m, SystemMessage)][-8:]
    messages = sys_msgs + other_msgs
    
    # 调用 LLM
    response = llm.invoke(messages)
    return {"messages": [response]}

🛠️ 方法四：按 Token 阈值修剪（生产推荐）

这是生产环境最常用的方法：在每次调用 LLM 前，估算当前消息的 token 数，如果超过阈值，删除最早的非系统消息。

import tiktoken
from langchain_core.messages import SystemMessage, RemoveMessage

class State(TypedDict):
    messages: Annotated[List[AnyMessage], add_messages]

def get_token_count(messages, model="gpt-3.5-turbo"):
    """计算消息列表的 token 数"""
    enc = tiktoken.encoding_for_model(model)
    total = 0
    for msg in messages:
        total += len(enc.encode(msg.content))
    return total

def auto_trim(state: State, max_tokens: int = 8000) -> dict:
    """自动修剪消息，确保不超过 token 限制"""
    messages = state["messages"]
    
    # 分离系统消息（通常需要保留）
    system_messages = [m for m in messages if isinstance(m, SystemMessage)]
    chat_messages = [m for m in messages if not isinstance(m, SystemMessage)]
    
    # 计算当前 token 数
    current_tokens = get_token_count(messages)
    
    if current_tokens <= max_tokens:
        return {}
    
    # 需要删除的数量（从最早的聊天消息开始）
    to_remove_ids = []
    tokens_to_remove = current_tokens - max_tokens
    
    removed_tokens = 0
    for i, msg in enumerate(chat_messages):
        if removed_tokens >= tokens_to_remove:
            break
        msg_tokens = len(tiktoken.encoding_for_model("gpt-3.5-turbo").encode(msg.content))
        to_remove_ids.append(RemoveMessage(id=msg.id))
        removed_tokens += msg_tokens
    
    return {"messages": to_remove_ids}

📊 各方法对比

💎 最佳实践建议

1. 设置合理的上限：根据你的模型上下文窗口（如 8K、32K、128K）设置 80% 阈值作为安全线。

2. 保留系统消息：系统提示词通常需要一直保留，不应被剪掉。

3. 考虑语义完整性：避免在对话中间突然切断，最好保留完整的问答对。

4. 使用 RemoveMessage 时注意 ID：确保 msg.id 存在（LangChain 的 BaseMessage 默认会有 ID）。

5. 测试先行：修剪策略对对话质量有明显影响，先用测试集验证效果。

如果你需要更具体的实现代码（适配你的 MessagesState 结构），可以把你的 State 定义发给我，我可以帮你写一个适配版的修剪函数。