当多轮对话的上下文越来越长，超出了 LLM 的 Context Window（上下文窗口），在 LangGraph 的 State 中应该如何设计历史消息的截断或总结机制？

当多轮对话的上下文越来越长，超出了 LLM 的 Context Window（上下文窗口），在 LangGraph...

知识点图片

在 LangGraph 中处理多轮对话上下文超出 LLM 窗口（Context Window）的问题，通常有两种核心策略：滑动窗口截断（Sliding Window / Truncation） 和 对话总结（Summarization）。

在 LangGraph 的设计中，因为状态（State）通常使用 add_messages 这个 reducer，它默认是追加消息的。因此，要管理上下文，我们需要利用 LangGraph 的消息删除机制或动态过滤机制。

以下是具体的架构设计和代码实现方案：

方案一：动态截断（只保留最近 N 条消息或 Token 限制）

如果你不需要长期的历史记忆，只需要最近几轮的上下文，最简单的做法是在传给 LLM 之前对消息列表进行截断。

设计思路：
不需要修改全局的 State（不删除原本的记录，如果需要保存完整日志的话），仅仅在调用 LLM 之前，使用 LangChain 提供的 trim_messages 工具对传给模型的 Message 列表进行修剪。

代码示例：

python

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import SystemMessage
from langchain_core.messages.utils import trim_messages
from langchain_openai import ChatOpenAI

# 1. 定义 State
class State(TypedDict):
    messages: Annotated[list, add_messages]

llm = ChatOpenAI(model="gpt-4o-mini")

# 2. 定义截断工具
# 这里配置保留最近的 4500 个 token，确保 system message 永远保留
trimmer = trim_messages(
    max_tokens=4500,
    strategy="last",
    token_counter=llm,
    include_system=True, # 确保系统提示词不被截断
    allow_partial=False, # 不截断单条消息的一半
    start_on="human"     # 确保截断后的第一条消息是用户的，避免模型报错
)

# 3. 定义 LLM 节点
def call_model(state: State):
    # 在这里动态截断，不改变原 State，只改变喂给 LLM 的数据
    trimmed_messages = trimmer.invoke(state["messages"])
    
    # 调用模型
    response = llm.invoke(trimmed_messages)
    return {"messages": [response]}

# 4. 构建 Graph
workflow = StateGraph(State)
workflow.add_node("agent", call_model)
workflow.add_edge(START, "agent")
workflow.add_edge("agent", END)
app = workflow.compile()

方案二：总结与清理混合机制（Hybrid Summarization）—— 推荐做法

截断会导致丢失早期的关键信息。更高级的做法是：当消息数量（或 Token 数）达到阈值时，触发一个“总结节点”，让 LLM 将早期的对话总结成一段文本，存入 State，然后从 State 中真正删除旧消息。

设计思路：

在 State 中增加一个 summary 字段。
使用条件边（Conditional Edge）检查消息长度，如果超长，则路由到 summarize_conversation 节点。
使用 LangGraph 的 RemoveMessage 从状态中永久移除已被总结的旧消息。

代码示例：

python

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage
from langchain_openai import ChatOpenAI

# 1. 定义带有 Summary 的 State
class State(TypedDict):
    messages: Annotated[list, add_messages]
    summary: str  # 保存早期的对话总结

llm = ChatOpenAI(model="gpt-4o-mini")

# 2. 主 LLM 节点
def call_model(state: State):
    summary = state.get("summary", "")
    if summary:
        # 将总结作为 System Message 注入
        system_message = f"这是之前的对话总结: {summary}"
        messages = [SystemMessage(content=system_message)] + state["messages"]
    else:
        messages = state["messages"]
        
    response = llm.invoke(messages)
    return {"messages": [response]}

# 3. 检查是否需要总结的逻辑 (Conditional Edge)
def should_summarize(state: State) -> Literal["summarize", END]:
    """如果消息超过 6 条，就触发总结"""
    if len(state["messages"]) > 6:
        return "summarize"
    return END

# 4. 总结并删除旧消息的节点
def summarize_conversation(state: State):
    summary = state.get("summary", "")
    messages = state["messages"]
    
    # 提取旧消息（留下最近的 2 条消息作为当前上下文）
    messages_to_summarize = messages[:-2]
    
    # 构建总结的 Prompt
    if summary:
        prompt = f"结合这段之前的总结: {summary}\n\n请将以下新的对话内容补充进总结中："
    else:
        prompt = "请用简洁的语言总结以下对话的核心信息："
        
    summary_prompt = [HumanMessage(content=prompt)] + messages_to_summarize
    
    # 调用 LLM 生成新总结
    response = llm.invoke(summary_prompt)
    new_summary = response.content
    
    # 核心：使用 RemoveMessage 删除已经被总结的旧消息
    # 这样 State 中的 messages 列表就会变短
    delete_messages = [RemoveMessage(id=m.id) for m in messages_to_summarize]
    
    return {
        "summary": new_summary,
        "messages": delete_messages
    }

# 5. 构建 Graph
workflow = StateGraph(State)
workflow.add_node("agent", call_model)
workflow.add_node("summarize", summarize_conversation)

workflow.add_edge(START, "agent")
# agent 执行完后，检查是否需要总结
workflow.add_conditional_edges("agent", should_summarize)
# 总结完后结束（或者你可以指回 agent，视具体业务而定）
workflow.add_edge("summarize", END)

app = workflow.compile()

方案三：引入外部记忆（向量数据库 RAG）

如果你的应用（例如心理咨询机器人、长期个人助理）需要无限期的、精确的历史记忆，单纯的总结也是不够的，因为细节会被丢失。

设计思路：
不再依赖大模型的 Context Window 携带所有历史，而是将历史存入向量数据库。

短期记忆（State 中的 Messages）：同方案一，使用 trim_messages 只保留最近的 N 轮。
长期记忆（Vector Store）：在 LangGraph 中增加一个异步节点，将每一轮完成的对话向量化并存入数据库。
记忆提取：在调用主 LLM 之前，增加一个 retrieve_memory 节点，将当前用户的 Query 去向量数据库召回相关的历史对话，拼接到 System Prompt 中。

最佳实践建议：

对于轻量级应用：直接使用 方案一（trim_messages），简单且不会出错。结合 Token 计算而不是条数计算，能最大化利用 Context Window 且不爆显存/报错。
对于复杂 Agent / 多轮业务：推荐 方案二（Hybrid Summarization）。这是 LangGraph 官方最推荐的模式，利用 RemoveMessage 保持状态整洁，同时用 summary 维持长期目标和上下文。
关于 RemoveMessage 的提示：在 LangGraph 的早期版本中，修改 messages 列表比较困难，但在较新的版本中引入了 RemoveMessage(id=...)。确保你在 State 中为被删除的消息提供准确的 id，LangGraph 的 add_messages reducer 会自动处理删除操作。

播面