LangChain 框架大项目用起来有多痛苦？本文整理了一套工程化方案
前言
"老王，为什么本文们的 LangChain 服务一重启就丢记忆？" 全栈工程师小李一脸无奈。
本文看了看他的代码，发现他用的是默认的 BufferMemory。"你这是把记忆存在内存里啊！服务一重启当然就没了！"
"那应该怎么弄？"
看来得从 LangChain 的工程化讲起了。今天本文们聊聊 LangChain 长期记忆的工程化方案。
一、底层原理
1.1 LangChain 的长期记忆困境
LangChain 的核心是把 LLM 调用链式编排，但长期记忆缺乏工程化支持：
graph TD A["用户输入"] --> B["LangChain Chain"] B --> C["LLM 调用"] C --> D{"有记忆吗？"} D -->|有| E["ConversationBufferMemory"] D -->|无| F["每次都重新来"] E --> G["内存中累积"] G --> H["Token 超限"] H --> I["崩溃"] F --> J["用户体验差"]核心问题：
- ConversationBufferMemory 无限增长
- 没有自动压缩机制
- 状态丢失无法恢复
- prompt 版本难管理
1.2 LangChain 长期记忆方案对比
| 方案 | 容量 | 持久化 | 复杂度 |
|---|---|---|---|
| BufferMemory | 小 | 否 | 低 |
| SummaryMemory | 中 | 否 | 中 |
| 向量存储记忆 | 大 | 是 | 高 |
| 自定义记忆 | 可控 | 是 | 最高 |
二、快速上手
看默认的 BufferMemory 问题：
from langchain.memory import ConversationBufferMemory memory = ConversationBufferMemory() memory.chat_memory.add_user_message("你好") memory.chat_memory.add_ai_message("你好！") # 对话越长，memory 越大...改进版，加摘要：
from langchain.memory import ConversationSummaryMemory from langchain.llms import OpenAI llm = OpenAI() memory = ConversationSummaryMemory(llm=llm) memory.save_context({"input": "你好"}, {"output": "你好！"}) memory.save_context({"input": "今天天气怎么样？"}, {"output": "今天是晴天。"}) # 自动总结 print(memory.load_memory_variables({}))三、核心 API / 深水区
3.1 LangChain 记忆方案速查
| 记忆类型 | 特点 | 适合场景 |
|---|---|---|
| BufferMemory | 保存所有 | 短对话 |
| SummaryMemory | 压缩总结 | 中长对话 |
| VectorStoreMemory | 向量检索 | 知识问答 |
| CombinedMemory | 混合使用 | 复杂场景 |
3.2 自定义持久化记忆
import json import os from langchain.memory import BaseMemory class FilePersistedMemory(BaseMemory): def __init__(self, file_path="memory.json"): self.file_path = file_path self.memory = self._load() def _load(self): if os.path.exists(self.file_path): with open(self.file_path) as f: return json.load(f) return {"history": []} def save(self): with open(self.file_path, "w") as f: json.dump(self.memory, f, ensure_ascii=False) @property def memory_variables(self): return ["history"] def load_memory_variables(self, inputs): return self.memory def save_context(self, inputs, outputs): self.memory["history"].append({ "input": inputs.get("input", ""), "output": outputs.get("output", "") }) self.save() if len(self.memory["history"]) > 100: self._compress() def _compress(self): self.memory["history"] = self.memory["history"][-50:]3.3 多轮记忆管理
from typing import Dict, List, Any from dataclasses import dataclass @dataclass class TurnContext: turn: int user_input: str assistant_output: str retrieved_docs: List[str] class MultiTurnMemory: def __init__(self, llm, max_turns=20): self.llm = llm self.max_turns = max_turns self.turns: List[TurnContext] = [] def add_turn(self, user_input: str, output: str, docs=None): self.turns.append(TurnContext( turn=len(self.turns) + 1, user_input=user_input, assistant_output=output, retrieved_docs=docs or [] )) if len(self.turns) > self.max_turns: self._summarize_old() def _summarize_old(self): old = self.turns[:5] text = "\n".join(f"U: {t.user_input}\nA: {t.assistant_output}" for t in old) prompt = f"总结前五轮对话：\n{text}" summary = self.llm(prompt) self.turns = self.turns[5:] print(f"历史摘要：{summary}") def get_context(self) -> str: recent = self.turns[-5:] return "\n".join( f"用户：{t.user_input}\n助手：{t.assistant_output}" for t in recent )四、实战演练
完整工程化 LangChain 记忆系统：
from typing import Dict, List, Optional import json import time class EngineeringMemory: def __init__(self, llm, persist_path="memory.json"): self.llm = llm self.persist_path = persist_path self.short_term: List[Dict] = [] self.long_term: List[Dict] = [] self.summaries: List[str] = [] self._load() def add_interaction(self, user_input: str, output: str): interaction = { "user": user_input, "assistant": output, "timestamp": time.time() } self.short_term.append(interaction) self._balance() self._save() def _balance(self): if len(self.short_term) > 20: old_interactions = self.short_term[:-10] if old_interactions: text = "\n".join( f"U: {i['user']}\nA: {i['assistant']}" for i in old_interactions ) prompt = f"总结对话：{text[:1500]}" summary = self.llm(prompt) self.summaries.append(summary) self.short_term = self.short_term[-10:] def get_memory_context(self) -> str: parts = [] if self.summaries: parts.append("【历史摘要】") parts.extend(self.summaries[-3:]) if self.short_term: parts.append("【最近对话】") for i in self.short_term[-5:]: parts.append(f"用户：{i['user']}") parts.append(f"助手：{i['assistant']}") return "\n".join(parts) def _save(self): data = { "short_term": self.short_term, "summaries": self.summaries } with open(self.persist_path, "w") as f: json.dump(data, f, ensure_ascii=False) def _load(self): try: with open(self.persist_path) as f: data = json.load(f) self.short_term = data.get("short_term", []) self.summaries = data.get("summaries", []) except: pass memory = EngineeringMemory(llm) # 模拟 50 轮对话 for i in range(50): memory.add_interaction(f"第{i}次提问", f"第{i}次回复") print(memory.get_memory_context())五、避坑指南与最佳实践
💡 **技巧：做好持久化
进程重启后记忆丢失，一定要持久化。
⚠️ **警告：BufferMemory 不要用太多
对话一长，Token 就爆了。
✅ **推荐：分层记忆
短期 + 短期摘要 + 长期，三级架构。
六、综合实战演示
生产级 LangChain 记忆系统：
from typing import Dict, List, Optional from datetime import datetime import json class LangChainPersistentMemory: def __init__(self, llm, user_id: str): self.llm = llm self.user_id = user_id self.short_memory: List[Dict] = [] self.long_memory: Dict[str, str] = {} self.last_summary_time = 0 def add(self, role: str, content: str): msg = { "role": role, "content": content, "time": datetime.now().isoformat() } self.short_memory.append(msg) if len(self.short_memory) >= 10: self._compress() def _compress(self): to_compress = self.short_memory[:-5] if not to_compress: return text = "\n".join(m["content"] for m in to_compress) summary = self.llm(f"总结：{text}") key = f"summary_{len(self.long_memory)}" self.long_memory[key] = summary self.short_memory = self.short_memory[-5:] def get_context(self, token_limit=3000) -> str: parts = [] # 长期记忆 for k, v in self.long_memory.items(): part = f"[历史] {v}" if sum(len(p) for p in parts) + len(part) > token_limit: break parts.append(part) # 短期记忆 for m in self.short_memory: part = f"{m['role']}: {m['content']}" parts.append(part) return "\n".join(parts) def to_dict(self) -> Dict: return { "user_id": self.user_id, "short": self.short_memory, "long": self.long_memory } memory = LangChainPersistentMemory(llm, "user_001") for i in range(15): memory.add("user", f"问题{i}") memory.add("assistant", f"回答{i}") print(memory.get_context())七、总结
LangChain 的长期记忆工程化要点：
- 不要只用 BufferMemory：它会无限增长，最终导致 Token 超限
- 自动压缩 + 摘要：定期总结历史对话，控制上下文大小
- 持久化存储：进程重启后记忆不丢失
- 分层记忆架构：短期记忆 + 短期摘要 + 长期记忆
搞好了这些，LangChain 项目才能在真实场景稳定运行。
