LangChain 内存管理优化ConversationBufferMemory 的内存膨胀问题根治一、深度引言与场景痛点大家好我是赵咕咕。上周我的一个生产环境出了问题——一个电商客服 Agent 在跟用户聊了四十分钟后服务突然 OOM 崩溃了。翻日志一看好家伙用户的对话历史已经积攒了 600 多条消息全部堆在一个 ConversationBufferMemory 里。每次新的对话还要把这些历史全部塞进 Prompt 里发给 LLMToken 消耗已经高达 4 万多。内存和费用双双爆炸。这就是 ConversationBufferMemory 的经典问题它没有任何遗忘机制。对话有多长它就能记得多长。就像一个从不清空的记事本写得越厚翻起来越慢。今天我们就来根治这个记忆膨胀问题从 LangChain 的内存结构剖析到生产级的记忆管理方案。二、底层机制与原理深度剖析2.1 ConversationBufferMemory 的数据流ConversationBufferMemory 本质上是一个用 Python list 维护的对话历史存储flowchart TD A[用户输入消息] -- B[HumanMessage] B -- C[append 到 buffer] D[LLM 返回响应] -- E[AIMessage] E -- C C -- F[chat_memory.messages] F -- G{是否满足触发条件?} G --|每次调用| H[load_memory_variables] H -- I[拼接所有历史消息为字符串] I -- J[注入 Prompt 模板] J -- K[发送完整 Prompt 到 LLM] K -- L{Token 数检查} L --|超过上下文窗口| M[截断或报错] L --|正常| N[LLM 处理并返回] M -.- O[⚠️ 可能丢失早期重要信息] style O fill:#ffe0e0 style F fill:#e0f0ff关键问题是buffer只增不减。每次load_memory_variables时所有历史消息都会被拼接到 Prompt 中。这意味着每次 LLM 调用的 Token 开销随对话增长而线性增加上下文窗口有上限如 GPT-4 的 128K超了就报错早期对话信息即使已经不相关仍然占据大量 Token 配额2.2 为什么不能简单地截断有人会说只保留最近 N 条消息不就行了这就是 ConversationBufferWindowMemory 的思路。但问题没这么简单截断会把用户的初始需求也扔掉。当用户说帮我查一下上周的订单这条关键指令在 20 轮对话后就被截掉了Agent 后续行为可能完全偏离截断是一刀切的不管消息重要不重要超了就扔对话中有回指和转折截断后 LLM 会断章取义我们需要的是一个有选择性的遗忘机制——忘记不重要的记住关键的。2.3 记忆管理的三种进阶方案flowchart LR A[完整对话历史] -- B{记忆管理策略} B -- C[摘要记忆] C -- C1[定期对历史做摘要用摘要替代原文] C1 -- C2[Token 稳定但可能丢失细节] B -- D[混合记忆] D -- D1[近期消息保留原文 远期消息保留摘要] D1 -- D2[兼顾细节和上下文较复杂] B -- E[向量记忆] E -- E1[将历史消息向量化存储按语义检索] E1 -- E2[不丢重要信息但检索成本高] style C2 fill:#ffffcc style D2 fill:#ccffcc style E2 fill:#cce5ff这三种方案不是互斥的。生产环境中我推荐混合记忆 向量记忆的组合。下面讲具体实现。三、生产级代码实现import asyncio import hashlib from dataclasses import dataclass, field from typing import Any from langchain.memory import ConversationSummaryMemory from langchain.memory.chat_memory import BaseChatMemory from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage # ── 1. 智能摘要记忆 ──────────────────────────────── class SummaryBufferMemory(BaseChatMemory): 基于 LLM 摘要的滑动窗口记忆 保留最近 window_size 条原始消息更早的消息定期生成摘要。 def __init__( self, llm, window_size: int 10, summary_interval: int 20, max_summary_length: int 500, **kwargs, ): super().__init__(**kwargs) self._llm llm self._window_size window_size self._summary_interval summary_interval self._max_summary_length max_summary_length self._summary: str self._message_count_since_summary: int 0 property def memory_variables(self) - list[str]: return [history] def _should_summarize(self) - bool: return self._message_count_since_summary self._summary_interval async def _generate_summary(self) - str: 调 LLM 对溢出窗口的消息生成摘要 messages self.chat_memory.messages overflow messages[: -self._window_size] if not overflow: return self._summary conversation_text \n.join( f{用户 if isinstance(m, HumanMessage) else 助手}: {m.content} for m in overflow ) summary_prompt f请将以下对话压缩为一段简洁摘要不超过 {self._max_summary_length} 字保留 1. 用户的初始需求和关键偏好 2. 已完成的操作和结果 3. 待处理的事项 对话 {conversation_text} 现有摘要{self._summary} 合并后的摘要 try: new_summary await self._llm.agenerate([summary_prompt]) return new_summary.generations[0][0].text.strip() except Exception: return self._summary async def _update_summary_if_needed(self): if self._should_summarize(): self._summary await self._generate_summary() self._message_count_since_summary 0 async def aload_memory_variables(self, inputs: dict) - dict: await self._update_summary_if_needed() messages self.chat_memory.messages recent messages[-self._window_size :] if len(messages) self._window_size else messages history_parts [] if self._summary: history_parts.append(f[对话摘要]\n{self._summary}) history_parts.extend( f{用户 if isinstance(m, HumanMessage) else 助手}: {m.content} for m in recent ) return {history: \n\n.join(history_parts)} async def asave_context(self, inputs: dict, outputs: dict): await super().asave_context(inputs, outputs) self._message_count_since_summary 2 # ── 2. 基于向量检索的记忆 ───────────────────────── class VectorRetrievalMemory: 将历史对话向量化存储按语义检索最相关的历史 def __init__(self, embedding_model, vector_store, top_k: int 5): self._embedding embedding_model self._store vector_store self._top_k top_k async def store(self, message: BaseMessage, metadata: dict | None None): 将一条消息向量化并存储 try: text message.content embedding await self._embedding.aembed_query(text) doc_id hashlib.md5(text.encode()).hexdigest()[:16] await self._store.aadd_texts( texts[text], metadatas[{ role: message.type, timestamp: asyncio.get_event_loop().time(), **(metadata or {}), }], ids[doc_id], ) except Exception: pass # 向量存储失败不应阻塞主流程 async def retrieve(self, query: str) - list[str]: 检索与当前查询最相关的历史消息 try: results await self._store.asimilarity_search(query, kself._top_k) return [doc.page_content for doc in results] except Exception: return [] # ── 3. 组合记忆混合所有策略 ────────────────────── class HybridMemory(BaseChatMemory): 生产级混合记忆 - 近期 N 条消息保留原文 - 更早消息 LLM 摘要 - 关键信息向量检索 def __init__( self, llm, embedding_model, vector_store, window_size: int 10, summary_interval: int 20, top_k: int 5, **kwargs, ): super().__init__(**kwargs) self._summary_memory SummaryBufferMemory( llmllm, window_sizewindow_size, summary_intervalsummary_interval, chat_memoryself.chat_memory, ) self._vector_memory VectorRetrievalMemory( embedding_modelembedding_model, vector_storevector_store, top_ktop_k, ) self._window_size window_size property def memory_variables(self) - list[str]: return [history, relevant_context] async def aload_memory_variables(self, inputs: dict) - dict: # 1. 获取摘要 最近消息 summary_result await self._summary_memory.aload_memory_variables(inputs) history summary_result[history] # 2. 向量检索最相关的历史上下文 current_query inputs.get(input, ) relevant await self._vector_memory.retrieve(current_query) relevant_context \n.join( f- {r} for r in relevant ) if relevant else 无相关历史信息 return { history: history, relevant_context: relevant_context, } async def asave_context(self, inputs: dict, outputs: dict): # 保存到聊天记忆供摘要使用 await self._summary_memory.asave_context(inputs, outputs) # 同时向量化存储输入和输出 user_msg HumanMessage(contentinputs.get(input, )) ai_msg AIMessage(contentoutputs.get(output, )) await asyncio.gather( self._vector_memory.store(user_msg), self._vector_memory.store(ai_msg), ) async def aclear(self): await super().aclear() # 向量存储的清理由外部管理 # ── 4. Token 预算控制器 ──────────────────────────── class TokenBudgetController: 在记忆注入 Prompt 前做最后把关确保不超 Token 预算 def __init__(self, max_tokens: int 8000): self._max_tokens max_tokens def estimate_tokens(self, text: str) - int: 粗略 Token 估算英文 1 token ≈ 4 字符中文 1 token ≈ 1 字符 import re en_chars len(re.findall(r[a-zA-Z0-9], text)) cn_chars len(text) - en_chars return en_chars // 4 cn_chars def trim(self, history: str, relevant: str) - tuple[str, str]: 按预算裁剪记忆内容优先保留 relevant budget self._max_tokens # 先算 relevant 的 Token relevant_tokens self.estimate_tokens(relevant) if relevant_tokens budget * 0.3: # relevant 不能超过 30% 预算 relevant self._truncate_text(relevant, int(budget * 0.3)) relevant_tokens self.estimate_tokens(relevant) remaining budget - relevant_tokens - 200 # 留 200 token 余量 history_tokens self.estimate_tokens(history) if history_tokens remaining: history self._truncate_text(history, max(remaining, 200)) return history, relevant def _truncate_text(self, text: str, max_tokens: int) - str: # 从尾部截取保留最近的内容 if self.estimate_tokens(text) max_tokens: return text # 按行截取 lines text.split(\n) result [] current_tokens 0 for line in reversed(lines): line_tokens self.estimate_tokens(line) if current_tokens line_tokens max_tokens: break result.insert(0, line) current_tokens line_tokens return \n.join(result)四、边界分析与架构权衡摘要记忆的取舍LLM 生成的摘要不可避免地会丢失一些细节。对于医疗、法律等对精确性要求极高的场景摘要记忆可能不够用。这时候需要把摘要局限在辅助定位的角色真正精确的信息仍然从向量记忆中检索。向量记忆的成本向量存储和检索有额外的延迟开销。如果每次对话都做一次向量检索对 P99 延迟会有 50-100ms 的影响。可以考虑异步存储 同步检索存储不阻塞主流程检索用缓存加速。对于频繁出现的查询模式可以把检索结果缓存 30 秒。Token 预算的估算误差Token 估算函数是粗略的不同模型的分词方式不一样。Claude 和 GPT 对中文的分词差异尤其大。生产环境中建议用对应模型的 tiktoken 库做精确计算或者在估算上预留 15% 的安全边际。什么时候该完全清空记忆不是所有对话都需要长期记忆。对于今天天气怎么样这种一次性查询完全不需要历史。应该在 Agent 的路由层加一个判断如果用户意图是一次性的就用空的或不带历史的记忆。记忆持久化的代价如果你的 Agent 需要跨会话保持记忆用户第二天回来还能接着聊记忆就需要持久化到数据库。但持久化带来了一致性问题——多个 Agent 实例同时更新同一个用户的记忆怎么办这时需要引入乐观锁或最后写入获胜策略。五、总结ConversationBufferMemory 的内存膨胀问题不是无解的。核心思想是把全量记忆变成智能记忆——用摘要压缩远期信息用向量检索定位关键信息用 Token 预算做最后兜底。三个关键策略分层记忆近期原文 中期摘要 远期向量检索Token 预算在记忆注入 Prompt 前做硬限制不让任何一个维度超过预算按需遗忘根据对话类型判断是否需要保持长记忆记住记忆管理的目标不是记得越多越好而是在有限的 Token 预算内记住对当前任务最有用的信息。如果你的 Agent 还在用默认的 ConversationBufferMemory 直接跑生产环境建议今晚就去改——省下来的 Token 费够你请全组喝一周的咖啡。下一篇预告Python 多进程协程混合架构GPU 推理与异步 I/O 的协作调度。