在构建客服 AI 助手时，最棘手的挑战并非模型推理能力，而是上下文长度的硬性限制。当用户会话持续超过 20 轮，全量历史消息导致 Token 膨胀，既增加推理成本又可能触发截断错误。LangGraph 通过状态机管理会话流转，结合滑动窗口机制，可实现动态消息裁剪，确保上下文始终处于模型有效窗口内。
一、核心概念：LangGraph 状态图与记忆机制
LangGraph 将 Agent 逻辑建模为有向图，节点代表处理步骤，边定义流转路径。其核心优势在于对会话状态的显式控制。与 LangChain 链式调用不同，Graph 允许定义多个检查点，实现会话的持久化与回溯。记忆部分通常由 Checkpoint 机制承担，存储当前对话历史及中间变量。
处理长会话需明确定义状态 schema。State 对象包含消息列表、用户意图及上下文元数据。在节点执行前后，通过状态读写实现数据流转。关键在于如何对 State 中的 `messages` 字段进行增量更新与裁剪，这是避免 Token 溢出的核心环节。
二、构建基础状态图
首先定义最小化的客服会话图，包含两个节点：接收用户输入与生成 LLM 响应。使用 `MessagesState` 作为默认状态，简化消息管理。
from langgraph.graph import StateGraph, MessagesState
from typing import TypedDict, Annotated
from operator import add
# 定义状态：messages 使用 reduce 操作，即追加而非覆盖
class CustomerServiceState(MessagesState):
intent: str
history_len: int
# 初始化图
graph = StateGraph(CustomerServiceState)
# 定义节点处理函数
def receive_input(state: CustomerServiceState) -> dict:
"""模拟用户输入处理"""
return {"intent": "inquiry"}
def generate_response(state: CustomerServiceState) -> dict:
"""模拟 LLM 生成回复，此处省略具体 LLM 调用"""
return {"messages": ("assistant", "您好，请问有什么可以帮助您的？")}
graph.add_node("receive", receive_input)
graph.add_node("generate", generate_response)
graph.set_entry_point("receive")
graph.add_edge("receive", "generate")
graph.add_edge("generate", "__end__")
app = graph.compile()三、实现滑动窗口消息裁剪
滑动窗口策略保留最近 N 轮对话，丢弃早期消息。实现方式在节点入口处修改 State 中的消息列表。需保留系统提示词（System Prompt），仅裁剪用户与助手的交替消息。
import math
def truncate_history(state: CustomerServiceState, window_size=10) -> dict:
"""
裁剪消息历史，保留 System Prompt 及最近 window_size 条消息
"""
messages = state["messages"]
# 分离 System Prompt 与对话历史
system_msgs = [m for m in messages if m.role == "system"]
chat_msgs = [m for m in messages if m.role != "system"]
# 保留最近 N 条聊天消息
truncated_chat = chat_msgs[-window_size:]
# 重新组装消息列表
new_messages = system_msgs + truncated_chat
# 计算当前 Token 估算（简单字符数/4 估算，生产环境建议使用 tiktoken）
token_estimate = sum(len(m.content) for m in new_messages) // 4
return {
"messages": new_messages,
"history_len": len(new_messages),
}
# 将裁剪逻辑添加到图节点前
graph.add_node("truncate", truncate_history)
graph.add_edge("generate", "truncate") # 在生成前执行裁剪四、动态 Token 控制策略
固定窗口大小无法适应不同长度的消息。更精细的策略是基于 Token 预算动态调整保留消息数量。定义最大 Token 阈值，从最新消息向前累加，直到超出阈值则停止。这种方法比固定条数更精准。
| 策略类型 | 实现复杂度 | 适用场景 | 缺点 |
|---|---|---|---|
| 固定条数窗口 | 低 | 消息长度均匀的业务 | 长消息易导致溢出 |
| Token 预算裁剪 | 中 | 消息长度波动大 | 需估算 Token，存在误差 |
| 摘要压缩 | 高 | 超长期会话 | 调用额外 LLM，延迟增加 |
在资源受限或离线场景，可参考 本地大模型量化部署 技术，选择轻量级模型配合更严格的 Token 预算，降低整体成本。若需进一步优化检索性能，可结合 Pinecone 向量库延迟优化 方案管理长期记忆。
五、检查点与会话持久化
LangGraph 支持 Checkpoint 机制，将状态序列化至数据库，实现会话恢复。这对于客服场景至关重要，用户重新接入时可延续上次对话。使用 SQLite 或 PostgreSQL 作为检查点存储，确保状态一致性。
# 注意：需安装对应检查点后端包，如 pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.base import BaseCheckpointSaver
# 初始化检查点存储器
checkpointer = SqliteSaver.from_conn_string("chat.db")
# 编译图时传入 checkpointer
graph_with_memory = graph.compile(checkpointer=checkpointer)
# 运行时指定线程 ID 区分不同会话
config = {
"configurable": {
"thread_id": "user_123_session_456"
}
}
# 第一次调用
result = graph_with_memory.invoke(
{"messages": [("user", "我的订单还没发货")], "intent": "order_status"},
config
)
# 第二次调用，自动加载历史状态
result = graph_with_memory.invoke(
{"messages": [("user", "那什么时候能到？")], "intent": "delivery_eta"},
config
)六、错误处理与降级策略
当 Token 估算失败或模型 API 返回上下文超限错误时，需具备降级能力。实现重试机制，进一步裁剪消息直至成功。若仍失败，向用户发送预设的截断提示，或转接人工客服。该逻辑可封装为独立的异常处理节点。
在多 Agent 调试复杂 Bug 的场景中，类似的状态管理技巧同样适用，可参考 用 LangGraph 实现多 Agent 协作调试复杂 Bug 中的错误传播与恢复模式。对于需要动态检索外部知识的场景，建议研究 用 LangGraph 编排 RAG 动态检索工作流，将检索结果作为额外上下文注入而非仅依赖对话历史。
七、性能监控与调优
生产环境需监控每次调用的 Token 消耗、窗口大小及推理延迟。在 State 中记录 `token_estimate` 并写入日志。调整 `window_size` 或 Token 预算时，A/B 测试不同策略对客服解决率的影响。通常保留最近 6-10 轮对话是平衡上下文完整性与成本的合理起点。
若本地部署可视化演示用于内部测试，可集成 Open WebUI 实战：自建本地大模型可视化平台，直观观察多轮对话中的上下文变化。
八、小结
通过 LangGraph 状态图结合滑动窗口记忆，可有效管理客服长会话上下文。核心在于状态裁剪逻辑与检查点持久化的协同。选择固定窗口或 Token 预算策略需根据消息分布特征决定。生产环境务必加入降级路径与监控指标。该架构具备良好扩展性，可后续接入向量库增强长期记忆能力，适应更复杂的客服场景需求。
