本文梳理了 LangChain 与 LangGraph 中的上下文剪裁核心策略（裁剪、删除、内置工具与摘要中间件）及 Token 预算规划方法，并结合“黑白梦博客助手”的业务场景，详解其采用的“Redis 短期快照 + MySQL 长期档案”双存储解耦架构与冷启动恢复实践。

为什么需要 Checkpoint 与上下文剪裁？

在构建对话 Agent 时，大语言模型（LLM）受限于最大上下文窗口（以 token 计）。随着对话轮次增加，消息列表持续增长，会导致以下问题：

1. 超出上下文窗口：对话请求超过模型限制导致 API 报错。
2. 注意力分散：无关的历史上下文干扰模型注意力，影响回答质量。
3. 延迟与成本增加：更长的 token 导致更高的首字响应延迟（TTFT）与 API 计费费用。

因此，主动管理对话历史（上下文剪裁，Context Trimming）是构建生产级 Agent 的必要环节。

在 LangChain/LangGraph 中，Checkpoint（检查点） 用于持久化保存多轮对话的运行状态。通过在写入 Checkpoint 时配合上下文剪裁策略，可以精简输入给 LLM 的历史消息。

上下文剪裁策略对比

针对不同业务场景，LangChain 与 LangGraph 提供了多种上下文管理策略：

在实际工程中，对于长对话且对上下文依赖较高的场景，使用 SummarizationMiddleware 可以更完整地保留语义；对于需要 token 级精确控制或在中间件中进行状态替换的场景，结合 trim_messages 或手动状态覆盖更为合适。

Token 预算规划与容量计算

在设置裁剪阈值或摘要触发条件时，需合理分配 token 预算。

模型支持上限与实际发送预算

虽然部分模型支持百万级的上下文窗口，但在工程实施中不宜将每轮对话都发满：

• 经济成本：如果不做裁剪，后续请求会重复发送大量历史消息，导致 API 计费增长。
• 首字响应延迟 (TTFT)：超长 Prompt 的预处理（Pre-fill）会导致首字输出延迟增加。
• 注意力噪声：过多无关的历史消息会引入噪声，增加模型产生幻觉的概率。

Token 预算计算公式

一次 LLM 请求的 token 占用主要由以下部分构成：

模型上下文窗口 = 系统提示词 + 工具定义 + 对话历史预算 + 模型输出预留 + 安全余量

因此，对话历史的可用预算计算方式为：

对话历史预算 = 模型上下文上限 - 系统提示词 - 工具定义 - 输出预留 - 安全余量

常见场景的对话历史预算建议

核心架构设计

在生产环境中，我们将 LangGraph Checkpoint（短期运行记忆）与 MySQL（长期全局档案）进行分离：

Redis Checkpoint 的角色：保存引擎当前运行快照。只保留精简后的上下文状态供下一次 LLM 调用，通过 @before_model 中间件配合 RemoveMessage(id=REMOVE_ALL_MESSAGES) 避免消息历史无限增长。

MySQL 的角色：充当数据归档库（Source of Truth），持久化保存完整的原始对话记录及结构化引用（Citations）数据。

关键实现机制

Checkpoint 基础配置与会话隔离

在初始化图结构（Graph）时，配置 RedisSaver 作为 checkpointer。通过结合用户 ID 和会话 ID 生成唯一的 thread_id（即 user_id:session_id），实现会话级别的状态隔离：

config = {"configurable": {"thread_id": f"{user_id}:{session_id}"}}
async for event in agent.astream({"messages": [HumanMessage(content=query)]}, config):
    # 处理流式输出

动态剪裁与截断 (@before_model)

在中间件实现中，利用 LangChain 的消息裁剪能力与 LangGraph 的状态更新机制：

1. Token 计算与精准裁剪 (trim_messages)：使用 langchain_core.messages.trim_messages 工具函数在模型调用前按 token 数量进行精准截取。该函数既支持在中间件中被调用，也支持通过管道符 | 直接接入 LCEL 链（如 chain = trimmer | model）。其 token_counter 支持 4 种计数模式：
   • "approximate"：基于字符数的近似计数（无需依赖特定模型）。
   • len：按消息条数而非 token 数进行截取。
   • Callable 自定义函数：针对特定语言或场景自定义计数逻辑。
   • 模型实例：传入特定模型（如 ChatOpenAI），通过模型的 tokenizer 进行精确定数。
2. 长工具输出截断：对于返回内容过长的 ToolMessage（如查询文章全文），中间件将超出指定字符数（如 2000 字符）的内容进行截断，防止单条工具消息占用过高上下文空间。
3. 状态替换与清空：由于 LangGraph 默认会对消息进行追加（Append），直接返回裁剪后的消息列表会在 Checkpoint 中保留原有历史。通过返回 RemoveMessage(id=REMOVE_ALL_MESSAGES) 清除当前 Checkpoint 中的全部消息记录，再重新写入裁剪后的消息列表，实现状态的压缩更新。

@before_model
def trim_messages_middleware(state: AgentState, runtime: Runtime) -> dict | None:
    messages = state["messages"]
    
    # 裁剪消息列表，保留系统消息与最近的对话
    trimmed = trim_messages(
        messages,
        max_tokens=4000,
        strategy="last",
        token_counter="approximate",
        include_system=True,
        start_on="human",
    )
    
    if len(trimmed) < len(messages):
        return {
            "messages": [
                RemoveMessage(id=REMOVE_ALL_MESSAGES),
                *trimmed
            ]
        }
    return None

摘要中间件的集成方式 (SummarizationMiddleware)

除了基于 token 或条数的直接裁剪外，对于需要长效维持历史摘要的场景，可以引入内置的 SummarizationMiddleware：

from langchain.agents.middleware import SummarizationMiddleware

summarizer = SummarizationMiddleware(
    model="openai:qwen3.5-flash",     # 用于生成摘要的轻量模型
    trigger=("tokens", 4000),          # 触发摘要的 token 阈值
    keep=("messages", 20),             # 保留最近消息条数
    summary_prompt="请将以下对话历史总结为简洁的中文摘要，保留核心信息：\n\n{messages}",
)

当对话达到 trigger 条件时，系统会自动调用摘要模型对旧消息进行压缩，并以摘要形式保留上下文语义。在配置 trigger 时，除指定固定 token 数或消息数外，还支持多条件联合触发（如 trigger=[("tokens", 3000), ("messages", 6)]，满足任一即触发）及按上下文比例触发（如 trigger=("fraction", 0.8)，达到模型上限的 80% 时触发）。

冷启动恢复 (Cold Start Recovery)

当 Redis 中的 Checkpoint 到期销毁（如 TTL 到期）后，系统支持从 MySQL 中恢复历史会话状态。

在 stream_chat 初始化阶段，先检查 Checkpoint 状态。若状态为空，则从 MySQL 加载最近的历史记录并注入状态：

async def _cold_start_recovery(config: dict, user_id: str, session_id: str):
    """从 MySQL 加载历史消息并重建 Checkpoint"""
    history = await load_recent_messages(user_id, session_id, limit=20)
    if not history:
        return
    
    seed_messages = []
    for msg in history:
        if msg.role == "human":
            seed_messages.append(HumanMessage(content=msg.content))
        elif msg.role == "ai":
            seed_messages.append(AIMessage(content=msg.content))
    
    if seed_messages:
        await agent.aupdate_state(config, {"messages": seed_messages})

此恢复流程只载入主要对白消息（HumanMessage 与 AIMessage），无需重新还原中间过程的工具消息。

流式断流防护 (Stream Safety)

在 SSE 流式传输过程中，客户端连接可能意外中断。为避免残缺回复写入全局数据库：

1. 维护 stream_completed 标记，默认设为 False。
2. 只有当流式遍历完整结束时，才将 stream_completed 更新为 True。
3. 若捕获到 asyncio.CancelledError 或异常，终止存储流程。
4. 仅在 stream_completed == True 且有有效产出时，才将用户输入与 AI 完整响应一次性写入 MySQL 数据库。

MySQL 历史表是否存储 AI Tool 消息分析

在 MySQL 持久化层中，设计上只存储用户的提问（human）和 AI 的回复（ai）及引用数据（citations），未包含中间的 AIMessage (tool_calls) 与 ToolMessage。原因如下：

职责分工

设计考量

1. 展示需求：前端 /history 接口主要展示人类与 AI 的最终对话过程，工具调用属于内部推理步骤，无需向终端用户直接展示。
2. 容量优化：工具调用的返回内容（如搜索结果或长文章）数据量较大，存储在 MySQL 会增加不必要的存储开销。
3. 恢复简化：冷启动重建 Checkpoint 时，仅重建 HumanMessage 与 AIMessage 即可恢复对话走向。避免恢复带有 tool_call_id 的工具消息引发现象不一致。

若后续需要追踪工具调用，可通过独立的审计日志（Audit Log）或专门的工具日志表实现，与主对话业务表解耦。

小结

在 AI Agent 系统构建中，合理的上下文管理是平衡回答质量、响应速度与运营成本的关键：

• 理论与选型：明确 Token 预算规划公式，并根据业务场景在 trim_messages 精确裁剪与 SummarizationMiddleware 语义摘要等策略之间做出权衡。
• 场景落地：黑白梦博客助手通过“Redis Checkpoint 管运行快照，MySQL 管长期档案”的双存储架构，搭配 @before_model 动态压缩与流式防护，既实现了上下文窗口的精准控制，又保证了全局对话数据的安全归档与冷启动可恢复性。