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 项目才能在真实场景稳定运行。