Agent上下文爆炸：从全量塞入到智能记忆管理

青云计划

186人浏览 · 2026-05-31 20:49:30

青云计划 · 2026-05-31 20:49:30 发布

大家好，我是程序员小策。

“给Agent的上下文越多，它就越聪明”——这几乎是每个刚接触Agent开发的人都会形成的直觉。

你随便问一个正在做Agent的同事，他都会告诉你："我把所有历史消息、工具调用结果、检索到的文档全部塞进prompt里，Agent自然就能做出更好的决策。"OpenAI的API文档里写满了"更大的上下文窗口"的卖点，各种技术博客也在教你怎么用128K token的模型处理长文档。

但真的是这样吗？

我们来看一个真实的账单：假设你做了一个客服Agent，每次对话10轮，每轮平均500 token，加上工具调用返回的2000 token，再加上检索到的文档片段5000 token——单次请求的上下文轻松突破12000 token。如果用户连续对话30轮呢？上下文膨胀到36000 token。而GPT-4的输入价格是每百万token 2.5美元——每30轮对话光上下文成本就接近0.1美元，而且响应延迟随着上下文增长呈线性上升。

更致命的是：上下文越大，模型反而越容易"分心"。 这就是著名的"Lost in the Middle"现象——模型对上下文开头和结尾的信息关注度高，对中间部分的信息关注度断崖式下降。你塞进去的"重要文档"，如果恰好落在中间位置，模型可能根本没注意到。

今天就来拆解一下这个话题——企业级的Agent项目是怎么解决上下文爆炸的。

一、问题的本质：上下文不是免费的

Agent上下文爆炸：随着对话轮次增加、工具调用累积、检索文档堆叠，Agent的上下文窗口被逐渐填满，导致Token成本飙升、响应延迟恶化、模型注意力分散的三重困境。

这个问题有三个维度：

成本维度。 每次LLM调用都要把整个上下文重新发送一遍。上下文从1K膨胀到100K，成本就是100倍的差距。这不是一次性开销，而是每次对话都要付的账单。

延迟维度。 Transformer的自注意力机制复杂度是O(n^2)，n是上下文长度。上下文翻倍，计算量翻四倍。用户的等待时间可不是线性增长。

质量维度。 如前所述，Lost in the Middle。大上下文里模型找不到重点，就像在一本没目录的百科全书里找一句话。

那么问题来了：企业级项目是怎么解决这个问题的？

二、朴素方案：全量上下文——为什么它不行

最直观的做法是什么？把所有的东西都塞进去。

# 朴素方案：把所有历史消息全量塞入上下文
def build_context_naive(messages, tools_results, documents):
    context = []
    context.append({"role": "system", "content": system_prompt})
    # 历史消息全部保留
    for msg in messages:
        context.append(msg)
    # 工具调用结果全部保留
    for tool_result in tools_results:
        context.append({"role": "tool", "content": tool_result})
    # 检索文档全部保留
    for doc in documents:
        context.append({"role": "system", "content": f"参考文档: {doc}"})
    return context

这个方案在对话轮次少的时候完全没问题。但当你的Agent需要处理几十轮对话、调用十几个工具、检索几十个文档片段时，上下文轻松突破模型上限。

全量上下文的致命问题是：它把"短期记忆"和"长期记忆"混在了一起。 就像你搬家时把所有东西——从今天的早饭到三年前的旧杂志——全部塞进同一个箱子里。到了新家，你翻箱倒柜找一个杯子，却要先把所有旧杂志翻一遍。

三、策略一：滑动窗口——最简单的"断舍离"

滑动窗口策略：只保留最近N轮对话，超出窗口的历史消息直接丢弃。这是解决上下文爆炸的最简单方案，代价是丢失早期对话的上下文信息。

这个策略就像手机相册的"最近删除"——只保留最近30天的照片，更早的自动清除。

# 滑动窗口策略：只保留最近N条消息
from collections import deque

class SlidingWindowContext:
    def __init__(self, max_messages: int = 20):
        self.max_messages = max_messages
        self.messages = deque(maxlen=max_messages)

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

    def get_context(self, system_prompt: str):
        return [{"role": "system", "content": system_prompt}] + list(self.messages)

优点： 实现简单，O(1)的内存开销，上下文大小完全可控。

缺点： 窗口外的信息永久丢失。如果用户在第1轮说"我叫张三"，在第30轮问"我叫什么名字？"——Agent会一脸茫然。

这个方案适用于对话轮次少、不需要跨轮次记忆的简单场景，比如单次问答、翻译任务。但对于需要持续对话的Agent——比如客服、个人助手——这个方案让Agent变成了"金鱼记忆"。

四、策略二：摘要压缩——把"历史"变成"简历"

摘要压缩策略：当对话历史超过一定长度时，调用LLM对历史对话进行摘要，用摘要替代原始对话，从而在保留关键信息的同时大幅压缩Token数量。

这个策略来自 Letta（前身MemGPT，GitHub 15K+ stars）的核心设计思想。Letta的设计灵感来自操作系统：把上下文窗口当作RAM，把外部存储当作硬盘，通过"换页"机制在两者之间搬运数据。

Letta的Agent有一个核心机制：self-editing memory（自编辑记忆）。Agent不仅接收消息，还会主动编辑自己的记忆块。当对话历史过长时，Agent会触发一个"反思"步骤，将冗长的对话历史压缩为一段精炼的摘要。

# Letta风格：摘要压缩 + 记忆块管理
from openai import OpenAI

client = OpenAI()

class SummaryCompressionContext:
    def __init__(self, max_recent_messages: int = 10, max_total_tokens: int = 4000):
        self.max_recent_messages = max_recent_messages
        self.max_total_tokens = max_total_tokens
        self.recent_messages = []  # 最近N条完整消息
        self.summary = ""  # 更早对话的摘要
        self.memory_blocks = {
            "human": "",  # 关于用户的关键信息
            "persona": ""  # Agent的自我认知
        }

    def add_message(self, role: str, content: str):
        self.recent_messages.append({"role": role, "content": content})
        # 超出滑动窗口时，触发摘要压缩
        if len(self.recent_messages) > self.max_recent_messages:
            self._compress()

    def _compress(self):
        # 将超出窗口的消息送去摘要
        overflow = self.recent_messages[:-self.max_recent_messages]
        self.recent_messages = self.recent_messages[-self.max_recent_messages:]

        summary_prompt = f"""将以下对话历史压缩为一段简洁的摘要，保留关键信息（人名、决策、重要事实）。
现有摘要：{self.summary if self.summary else '无'}

待压缩的对话：
{self._format_messages(overflow)}

请生成更新后的摘要："""

        response = client.chat.completions.create(
            model="gpt-4o-mini",  # 压缩用便宜的小模型
            messages=[{"role": "user", "content": summary_prompt}]
        )
        self.summary = response.choices[0].message.content

    def get_context(self, system_prompt: str):
        context = [{"role": "system", "content": system_prompt}]
        if self.summary:
            context.append({"role": "system", "content": f"[历史对话摘要] {self.summary}"})
        context.extend(self.recent_messages)
        return context

为什么用GPT-4o-mini做压缩？ 压缩任务本身不需要强大的推理能力——它只需要提取关键信息、组织语言。用便宜的小模型做压缩，用大模型做主推理，这是成本最优解。

Letta的记忆块设计巧妙在哪？ 它把记忆分为"human"（关于用户的事实）和"persona"（Agent的自我定义）两块。这两块独立存储、独立更新，Agent可以在对话中主动修改它们——比如"我了解到用户叫张三"，Agent就会更新human块。

这个方案的边界情况： 摘要本身也是Token。如果摘要不断累积，它自己也会膨胀。Letta的解法是：摘要也分层——近期摘要保留细节，远期摘要进一步压缩。就像简历：最近的工作经历写详细，十年前的工作经历一行带过。

五、策略三：外部记忆 + 按需检索——把"记忆"变成"数据库"

外部记忆策略：将对话历史、用户偏好、领域知识存储在外部向量数据库中，每次对话时根据当前问题检索最相关的记忆片段，只将检索结果注入上下文，而不是全部历史。

这是 Mem0（mem0ai/mem0，GitHub 20K+ stars）的核心策略。Mem0的定位是"AI Agent的通用记忆层"——它不直接管理上下文窗口，而是在上下文窗口之外维护一个独立的记忆系统。

Mem0的核心理念是：记忆不应该住在上下文窗口里，它应该住在向量数据库里。 上下文窗口只保留"当前工作所需"的记忆，其余全部外置。

# Mem0风格：外部记忆 + 按需检索
from mem0 import Memory
from openai import OpenAI

client = OpenAI()
memory = Memory()

class ExternalMemoryContext:
    def __init__(self, user_id: str, max_context_messages: int = 10):
        self.user_id = user_id
        self.max_context_messages = max_context_messages
        self.recent_messages = []

    def add_message(self, role: str, content: str):
        self.recent_messages.append({"role": role, "content": content})
        if len(self.recent_messages) > self.max_context_messages:
            self.recent_messages = self.recent_messages[-self.max_context_messages:]

    def chat(self, user_input: str) -> str:
        # 关键步骤：从外部记忆库检索相关记忆
        relevant_memories = memory.search(
            query=user_input,
            filters={"user_id": self.user_id},
            top_k=3  # 只取最相关的3条记忆
        )
        memories_str = "\n".join(
            f"- {entry['memory']}" for entry in relevant_memories["results"]
        )

        # 将检索到的记忆注入system prompt
        system_prompt = f"""你是一个智能助手。以下是关于用户的历史记忆：
{memories_str}

最近的对话记录：
{self._format_recent()}

请基于以上信息回答问题。"""

        messages = [{"role": "system", "content": system_prompt}]
        messages.append({"role": "user", "content": user_input})

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages
        )
        assistant_response = response.choices[0].message.content

        # 对话结束后，将新信息存入外部记忆
        self.add_message("user", user_input)
        self.add_message("assistant", assistant_response)
        memory.add(
            [
                {"role": "user", "content": user_input},
                {"role": "assistant", "content": assistant_response}
            ],
            user_id=self.user_id
        )

        return assistant_response

Mem0的检索不是简单的向量搜索。 它用的是多信号融合检索：

语义检索（向量相似度）
BM25关键词检索（精确匹配）
实体链接（将"张三"和"他"关联起来）
时间推理（能区分"当前状态"和"过去事件"）

为什么top_k=3？ 这是Mem0论文中的关键参数。Top-K太小，丢失关键信息；Top-K太大，噪声过多，Token浪费。Mem0的benchmark显示，单次检索平均消耗约7000 token——相比全量上下文动辄几万token，这是数量级的压缩。

这个方案的边界情况： 如果检索不准确，Agent拿到的记忆是无关的，回答质量反而下降。Mem0的解法是：只做ADD操作，不做UPDATE/DELETE——记忆只增不删，避免误删关键信息。通过多信号融合检索来保证召回率。

六、策略四：状态检查点 + 多Agent隔离——LangGraph的企业级方案

状态检查点策略：将Agent的完整状态（包括对话历史）持久化到外部存储，每次对话时从检查点恢复状态，对话结束后保存新状态。这样上下文窗口可以很小，但Agent仍然"记得"一切。

多Agent上下文隔离：每个Agent只管理自己的上下文，Agent之间通过消息传递协作，而不是共享上下文。

这是 LangGraph（langchain-ai/langgraph，GitHub 10K+ stars）的核心理念。LangGraph没有直接实现"压缩"或"检索"，而是提供了一套底层基础设施，让开发者可以灵活组合各种策略。

LangGraph做对了两件事：

第一，Durable Execution（持久化执行）。 每个Agent的执行状态（包括消息历史、中间结果）都通过checkpointer持久化到外部存储。Agent可以随时"断点续传"——即使上下文窗口只保留最近N条消息，完整的历史仍然在数据库中。

第二，多Agent的上下文隔离。 在一个多Agent系统中，每个Agent有自己独立的上下文窗口。Agent A不需要知道Agent B的完整对话历史——它只需要知道Agent B的输出结果。这就是"子图"（subgraph）模式。

# LangGraph风格：多Agent上下文隔离 + 状态检查点
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
import operator

# 每个Agent的独立状态
class OrderAgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 订单查询Agent的上下文
    order_result: str

class RefundAgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 退款Agent的上下文
    refund_result: str

# 全局协调状态（只保留必要信息，不包含子Agent的完整上下文）
class SupervisorState(TypedDict):
    user_input: str
    order_result: str
    refund_result: str
    final_response: str

# 创建checkpointer，持久化执行状态
checkpointer = MemorySaver()

# 订单查询Agent：独立的上下文窗口
def order_agent(state: SupervisorState):
    # 这个Agent只关心订单查询，不需要知道退款Agent的历史
    order_context = [
        {"role": "system", "content": "你是订单查询助手，只负责查询订单信息。"},
        {"role": "user", "content": f"查询订单: {state['user_input']}"}
    ]
    # ... LLM调用
    return {"order_result": "订单#12345: 已发货，预计明天到达"}

# 退款Agent：独立的上下文窗口
def refund_agent(state: SupervisorState):
    refund_context = [
        {"role": "system", "content": "你是退款处理助手，只负责处理退款请求。"},
        {"role": "user", "content": f"处理退款: {state['user_input']}"}
    ]
    # ... LLM调用
    return {"refund_result": "退款申请已提交，预计3个工作日内到账"}

# 构建工作流，with checkpointer
graph = StateGraph(SupervisorState)
graph.add_node("order", order_agent)
graph.add_node("refund", refund_agent)
graph.add_node("summarize", summarize_results)
graph.set_entry_point("order")
graph.add_edge("order", "refund")
graph.add_edge("refund", "summarize")
graph.add_edge("summarize", END)

# 编译时传入checkpointer
app = graph.compile(checkpointer=checkpointer)

# 同一个thread_id确保断点续传
config = {"configurable": {"thread_id": "user-123-session-456"}}
result = app.invoke({"user_input": "我的订单#12345发货了吗？我要退款"}, config)

LangGraph的checkpointer解决了什么问题？ 没有checkpointer时，Agent的上下文全部在内存中——服务重启、容器迁移，上下文全部丢失。有了checkpointer，状态持久化到数据库，Agent可以随时恢复。这意味着你可以把上下文窗口设得很小（比如只保留最近5轮），完整历史在数据库中，需要时再检索。

多Agent上下文隔离的核心价值： 3个Agent各自维护1000 token的上下文，总开销是3000 token。但如果3个Agent共享一个上下文窗口，开销可能是10000+ token（因为每个Agent需要看到所有Agent的完整对话）。上下文隔离不是1+1=2，而是避免了1+1=10的爆炸。

七、实战场景：一个日均5000+用户的智能客服系统

前面讲了四种策略，说"实战中往往是组合使用"。这句话不落地就是一句正确的废话。下面用一个真实的项目场景来展示：四层策略是怎么在一个智能客服系统中组合起来的。

项目背景： 某电商平台的智能客服系统，日均5000+活跃用户，覆盖订单查询、退款处理、物流追踪、商品推荐四个核心场景。单用户平均对话轮次15轮，高峰期并发对话600+。

场景拆解：

这个系统上线第一版时，用的是朴素的全量上下文方案。上线两周后问题爆发了：

平均对话到第12轮，上下文超过25000 token，GPT-4单次调用的输入成本超过0.06美元
高峰期10台服务器打满，GPU限流导致响应延迟从2秒飙升到15秒
用户在第3轮说"我要退的是上周买的那件红色夹克"，到第10轮Agent开始问"请问您要退哪个订单？"——Lost in the Middle

这三个问题是同时发生的，必须用一个完整的方案来解决。我们的选型是：滑动窗口（最近8轮） + 摘要压缩（8轮之前的对话） + 外部记忆（用户画像+历史订单偏好） + 多Agent上下文隔离（订单/退款/物流三个子Agent各自独立上下文）。

下面是这个系统落地时的核心代码——不是教学Demo，是可以直接在项目里用的集成版本：

# 智能客服系统的上下文管理集成方案
# 组合策略：滑动窗口 + 摘要压缩 + 外部记忆 + 多Agent隔离
from openai import OpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from mem0 import Memory
from collections import deque
from typing import TypedDict, Annotated, Optional
import operator
import json
import hashlib

client = OpenAI()
checkpointer = MemorySaver()
memory = Memory()

# ========== 配置层 ==========
# 生产环境的上下文窗口配置——根据实际压测数据调优
class ContextConfig:
    MAX_RECENT_MESSAGES = 8       # 滑动窗口大小：保留最近8轮完整对话
    SUMMARY_TRIGGER_ROUNDS = 12   # 摘要触发阈值：累积12轮后触发压缩
    MEMORY_TOP_K = 5              # 外部记忆检索条数
    COMPRESSION_MODEL = "gpt-4o-mini"  # 压缩用便宜模型，成本敏感场景也可用 gpt-3.5-turbo
    REASONING_MODEL = "gpt-4o"         # 主推理用强模型

# ========== 通用上下文管理器 ==========
class HybridContextManager:
    """混合上下文管理器：滑动窗口 + 摘要压缩 + 外部记忆"""
    
    def __init__(self, user_id: str, session_id: str, config: ContextConfig = ContextConfig()):
        self.user_id = user_id
        self.session_id = session_id
        self.config = config
        self.recent_messages = deque(maxlen=config.MAX_RECENT_MESSAGES)
        self.summary = ""
        self.round_count = 0
    
    def add_message(self, role: str, content: str):
        self.recent_messages.append({"role": role, "content": content})
        self.round_count += 1
        
        # 超过摘要触发阈值时，触发增量压缩
        if self.round_count >= self.config.SUMMARY_TRIGGER_ROUNDS and self.round_count % 4 == 0:
            self._incremental_compress()
    
    def _incremental_compress(self):
        """增量压缩：将窗口外的对话合并进现有摘要"""
        recent_list = list(self.recent_messages)
        # 保留窗口内的，其余送去压缩
        overflow = recent_list[:-self.config.MAX_RECENT_MESSAGES]
        if len(overflow) < 2:
            return
        
        compress_prompt = f"""将以下新增对话合并到现有摘要中，保留关键信息：
现有摘要：{self.summary if self.summary else '无'}

新增对话：
{self._format_messages(overflow)}

生成更新后的摘要（不超过200字）："""
        
        response = client.chat.completions.create(
            model=self.config.COMPRESSION_MODEL,
            messages=[{"role": "user", "content": compress_prompt}],
            max_tokens=300
        )
        self.summary = response.choices[0].message.content
        self.recent_messages = deque(
            recent_list[-self.config.MAX_RECENT_MESSAGES:],
            maxlen=self.config.MAX_RECENT_MESSAGES
        )
    
    def retrieve_memories(self, query: str) -> str:
        """从Mem0外部记忆库检索相关记忆"""
        try:
            results = memory.search(
                query=query,
                filters={"user_id": self.user_id},
                top_k=self.config.MEMORY_TOP_K
            )
            if results.get("results"):
                return "\n".join(
                    f"- {entry['memory']}" for entry in results["results"]
                )
        except Exception:
            pass  # 外部记忆检索失败不阻塞主流程
        return "暂无历史记忆"
    
    def build_system_prompt(self, user_input: str, agent_role: str) -> str:
        """构建完整的system prompt：摘要 + 外部记忆 + 角色定义"""
        memories_str = self.retrieve_memories(user_input)
        summary_block = f"[对话历史摘要]\n{self.summary}" if self.summary else ""
        
        return f"""{agent_role}

{summary_block}

[用户历史记忆]
{memories_str}

注意：以上记忆可能包含过时信息，请结合当前对话判断。"""
    
    def store_conversation_to_memory(self, user_msg: str, assistant_msg: str):
        """对话结束后异步存储到外部记忆"""
        try:
            memory.add(
                [
                    {"role": "user", "content": user_msg},
                    {"role": "assistant", "content": assistant_msg}
                ],
                user_id=self.user_id,
                metadata={"session_id": self.session_id}
            )
        except Exception:
            pass
    
    def _format_messages(self, messages: list) -> str:
        return "\n".join(
            f"{m['role']}: {m['content'][:200]}" for m in messages
        )

# ========== 多Agent上下文隔离 ==========
# 三个子Agent各自维护独立的上下文管理器实例

class OrderAgentState(TypedDict):
    order_context_id: str
    order_result: str

class RefundAgentState(TypedDict):
    refund_context_id: str
    refund_result: str

class LogisticsAgentState(TypedDict):
    logistics_context_id: str
    logistics_result: str

class SupervisorState(TypedDict):
    user_id: str
    session_id: str
    user_input: str
    intent: str
    order_result: str
    refund_result: str
    logistics_result: str
    final_response: str

# 订单查询Agent：独立的上下文窗口
def order_agent(state: SupervisorState, ctx_mgr: HybridContextManager):
    ctx_mgr.add_message("user", state["user_input"])
    
    system_prompt = ctx_mgr.build_system_prompt(
        state["user_input"],
        "你是订单查询助手，只负责查询订单状态、订单详情、物流信息。不要处理退款请求。"
    )
    
    response = client.chat.completions.create(
        model=ContextConfig.REASONING_MODEL,
        messages=[
            {"role": "system", "content": system_prompt},
            *list(ctx_mgr.recent_messages)
        ]
    )
    result = response.choices[0].message.content
    ctx_mgr.add_message("assistant", result)
    return {"order_result": result}

# 退款Agent：独立的上下文窗口
def refund_agent(state: SupervisorState, ctx_mgr: HybridContextManager):
    ctx_mgr.add_message("user", state["user_input"])
    
    system_prompt = ctx_mgr.build_system_prompt(
        state["user_input"],
        "你是退款处理助手，负责处理退款申请、查询退款进度、解释退款政策。"
    )
    
    response = client.chat.completions.create(
        model=ContextConfig.REASONING_MODEL,
        messages=[
            {"role": "system", "content": system_prompt},
            *list(ctx_mgr.recent_messages)
        ]
    )
    result = response.choices[0].message.content
    ctx_mgr.add_message("assistant", result)
    return {"refund_result": result}

# 物流Agent：独立的上下文窗口
def logistics_agent(state: SupervisorState, ctx_mgr: HybridContextManager):
    ctx_mgr.add_message("user", state["user_input"])
    
    system_prompt = ctx_mgr.build_system_prompt(
        state["user_input"],
        "你是物流追踪助手，负责查询快递进度、异常物流、预计送达时间。"
    )
    
    response = client.chat.completions.create(
        model=ContextConfig.REASONING_MODEL,
        messages=[
            {"role": "system", "content": system_prompt},
            *list(ctx_mgr.recent_messages)
        ]
    )
    result = response.choices[0].message.content
    ctx_mgr.add_message("assistant", result)
    return {"logistics_result": result}

# ========== 工作流编排 ==========
# 关键设计：每个子Agent传入独立的上下文管理器，实现上下文隔离
# thread_id 使用 session_id + agent_name 保证每个Agent独立持久化

def build_customer_service_graph(user_id: str, session_id: str):
    """构建智能客服工作流图"""
    # 每个子Agent有独立的上下文管理器
    order_ctx = HybridContextManager(user_id, session_id)
    refund_ctx = HybridContextManager(user_id, session_id)
    logistics_ctx = HybridContextManager(user_id, session_id)
    
    graph = StateGraph(SupervisorState)
    
    # 使用闭包将上下文管理器注入到节点函数中
    graph.add_node("order", lambda s: order_agent(s, order_ctx))
    graph.add_node("refund", lambda s: refund_agent(s, refund_ctx))
    graph.add_node("logistics", lambda s: logistics_agent(s, logistics_ctx))
    
    def route_intent(state: SupervisorState):
        intent = state.get("intent", "order")
        if "退款" in state["user_input"]:
            return "refund"
        elif "物流" in state["user_input"] or "快递" in state["user_input"]:
            return "logistics"
        return "order"
    
    def summarize(state: SupervisorState) -> dict:
        # 汇总三个子Agent的结果，只传递结构化结论，不传递完整上下文
        parts = []
        if state.get("order_result"):
            parts.append(f"[订单] {state['order_result']}")
        if state.get("refund_result"):
            parts.append(f"[退款] {state['refund_result']}")
        if state.get("logistics_result"):
            parts.append(f"[物流] {state['logistics_result']}")
        
        response = client.chat.completions.create(
            model=ContextConfig.REASONING_MODEL,
            messages=[{
                "role": "user",
                "content": f"请将以下各模块的结果整合为一段自然的客服回复：\n" + "\n".join(parts)
            }]
        )
        final = response.choices[0].message.content
        
        # 存储本轮对话到外部记忆
        order_ctx.store_conversation_to_memory(state["user_input"], final)
        
        return {"final_response": final}
    
    graph.add_node("summarize", summarize)
    graph.add_conditional_edges("__start__", route_intent, {
        "order": "order",
        "refund": "refund", 
        "logistics": "logistics"
    })
    graph.add_edge("order", "summarize")
    graph.add_edge("refund", "summarize")
    graph.add_edge("logistics", "summarize")
    graph.add_edge("summarize", END)
    
    return graph.compile(checkpointer=checkpointer)

# ========== 使用示例 ==========
app = build_customer_service_graph(user_id="user-8891", session_id="sess-20240531-001")
config = {"configurable": {"thread_id": "user-8891-sess-20240531-001"}}

result = app.invoke(
    {"user_id": "user-8891", "session_id": "sess-20240531-001",
     "user_input": "我上周买的红色夹克发货了吗？如果没发我想退款"},
    config
)
print(result["final_response"])

落地中踩到的坑与最佳实践：

坑一：摘要压缩的触发时机。 第一版我们设为每轮对话都触发摘要压缩，结果GPT-4o-mini的调用次数跟主模型一样多——压缩成本反而超过了上下文节省的成本。后来改成"每4轮触发一次"，只在轮次超过阈值后才启动。

坑二：外部记忆的检索延迟。 Mem0的search是同步调用，高峰期向量检索耗时200-500ms。如果放在主流程里同步等待，用户体验很差。实战中给检索加了500ms超时，超时后降级为空记忆继续对话——宁可少给一条记忆，也不能让用户等。

坑三：子Agent错误传播。 有一次订单Agent返回了"该订单不支持退款"，退款Agent拿到这个结论后直接拒绝了用户——但实际上"不支持退款"指的是"已发货订单需先退货再退款"，不是"不能退款"。解法是：summarize节点拿到子Agent结果后，不做直接拼接，而是让主模型重新理解一遍——相当于加了一道"语义审核"。

八、四种策略对比

策略	核心思路	Token节省	信息保留	实现复杂度	适用场景
滑动窗口	只保留最近N轮	高（O(1)固定窗口）	低（旧信息丢失）	极低	单次问答、翻译
摘要压缩	LLM压缩历史为摘要	中高（压缩比5-10倍）	中（细节丢失，要点保留）	中	长期对话，如个人助手
外部记忆+检索	记忆存向量库，按需检索	高（只取top-K）	高（检索质量决定）	高	多轮客服、个性化推荐
状态检查点+隔离	持久化+Agent独立上下文	高（各Agent窗口独立）	高（完整历史可恢复）	高	多Agent协作系统

什么时候选什么？

对话短、不需要跨轮记忆 → 滑动窗口足够，别过度设计
单Agent需要长期记忆但成本敏感 → 摘要压缩，用便宜的小模型做压缩
需要精确记忆、个性化要求高 → 外部记忆+检索，Mem0这类方案
多Agent系统、需要断点续传 → LangGraph的checkpointer + 上下文隔离

实战中往往是组合使用。 上面智能客服系统的例子就是典型：滑动窗口放最近8轮 + 摘要压缩处理更早的对话 + 外部记忆存储用户画像 + 多Agent间上下文隔离。四层策略叠加，才是企业级的完整方案。

九、面试追问

追问1：滑动窗口和摘要压缩能不能结合？怎么结合？

回答方向：可以，这就是"ConversationSummaryBufferMemory"模式。最近N轮保留完整消息，超过N轮的消息送去摘要，摘要作为system prompt的一部分注入。这样既保留了最近的细节，又不会丢失更早的上下文。问题在于摘要的更新频率——每轮都更新摘要成本太高，通常是每K轮触发一次。

追问2：Mem0的"只增不删"策略有什么风险？如果用户说"我搬家了，新地址是XXX"，旧地址怎么处理？

回答方向：Mem0通过时间推理来处理——检索时会根据时间戳判断"当前状态"vs"历史状态"。旧地址不会被删除，但检索时新地址的权重更高。这避免了误删的风险，但长期运行后记忆库会膨胀，需要定期清理或归档。这是一个"一致性vs可用性"的权衡。

追问3：多Agent上下文隔离有什么新的问题？

回答方向：Agent A的输出可能是Agent B的输入，但Agent B不知道Agent A的完整推理过程——如果Agent A犯了错误，Agent B可能基于错误结果继续推理，形成"错误传播"。解法是让关键决策点加入人工审核（Human-in-the-Loop），或者在Agent之间传递结构化结果而非纯文本。

追问4：128K上下文窗口的模型越来越便宜，这些策略还有必要吗？

回答方向：有必要。即使上下文窗口变大，三个问题仍然存在：成本（128K token的费用是1K的100倍以上）、延迟（O(n^2)的注意力计算）、质量（Lost in the Middle现象在大上下文中更严重）。更大的上下文窗口不是"不需要管理上下文"的理由，而是"需要更精细地管理上下文"的理由。