智能体短期记忆管理实战指南：LangGraph三剑客解析

创作时间:

作者:

@小白创作中心

智能体短期记忆管理实战指南：LangGraph三剑客解析

引用

来源

https://www.h3blog.com/article/566/

在智能体系统设计中，短期记忆管理直接影响着对话质量与资源消耗。本文将通过LangGraph框架，演示三种渐进式的记忆优化策略，助你打造更高效的对话系统。

一、传统记忆机制的三大痛点

当前主流的ChatML模板虽能自动拼接对话历史，但存在明显缺陷：

🚨Token消耗失控：每次交互携带完整历史，成本指数级增长
💥上下文窗口溢出：超过模型处理上限（如GPT-4 Turbo的128k限制）
⏳响应延迟加剧：长文本处理耗时影响用户体验

"记忆不是存储所有细节，而是保留关键脉络" —— 本文核心设计理念

二、LangGraph解决方案实战

策略1：滑动窗口记忆法（RemoveMessage）

from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import RemoveMessage

def filter_messages(state):
    """保留最近3轮对话的滑动窗口"""
    return {"messages": [RemoveMessage(m.id) for m in state["messages"][:-3]]}

# 构建处理流程
builder = StateGraph(MessagesState)
builder.add_node("filter_messages", filter_messages)
builder.add_node("chat_model", lambda state: {"messages": llm.invoke(state["messages"])})
builder.add_edge(START, "filter_messages")
builder.add_edge("filter_messages", "chat_model")
builder.add_edge("chat_model", END)
graph = builder.compile()

特点：
✅ 固定内存占用
⚠️ 可能丢失关键历史信息

策略2：动态Token裁剪法（trim_messages）

from langchain_core.messages import trim_messages

def smart_filter(state):
    """按Token预算智能裁剪"""
    processed = trim_messages(
        state["messages"],
        max_tokens=1000,  # 根据模型窗口调整
        strategy="last",  # 保留尾部对话
        token_counter=ChatOpenAI(model="gpt-3.5-turbo")
    )
    return {"messages": processed}

# 替换基础策略中的filter_messages节点
builder.add_node("smart_filter", smart_filter)

优势：
📊 自适应模型上下文窗口
⚖️ 平衡信息完整性与资源消耗

策略3：增量摘要记忆法

def summary_processor(state):
    """每6轮对话生成增量摘要"""
    summary = state.get("summary", "")
    prompt = f"现有摘要：{summary}\n根据新对话更新摘要：" if summary else "创建对话摘要："
    messages = state["messages"][-2:] + [HumanMessage(content=prompt)]
    new_summary = llm.invoke(messages).content
    return {
        "summary": new_summary,
        "messages": [RemoveMessage(m.id) for m in state["messages"][:-2]]
    }

def should_summarize(state):
    return "summarize" if len(state["messages"]) > 6 else END

# 构建带条件判断的工作流
workflow = StateGraph(State)
workflow.add_conditional_edges(
    "conversation",
    should_summarize,
    {"summarize": "summary_processor", END: END}
)

技术要点：
🔍 定期提炼对话要点
🔄 保留关键上下文脉络
⏲️ 通过条件触发避免过度计算