LangChain Memory — 对话记忆管理

记忆类型概览

类型	特点	适用场景
ConversationBufferMemory	保存完整历史	短对话
ConversationBufferWindowMemory	保留最近 K 轮	长对话节省 token
ConversationSummaryMemory	自动摘要历史	超长对话
ConversationSummaryBufferMemory	摘要+近期完整	生产推荐
VectorStoreRetrieverMemory	向量检索相关记忆	知识密集型

ConversationBufferWindowMemory

from langchain.memory import ConversationBufferWindowMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain

llm = ChatOpenAI(model="qwen-turbo", ...)

memory = ConversationBufferWindowMemory(
    k=5,                    # 保留最近 5 轮对话
    return_messages=True    # 返回 Message 对象而非字符串
)

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

conversation.predict(input="我想申请一笔 50 万的企业贷款")
conversation.predict(input="我们公司成立 3 年，年营收 200 万")
conversation.predict(input="需要什么材料？")  # 模型记得上下文

ConversationSummaryBufferMemory（生产推荐）

from langchain.memory import ConversationSummaryBufferMemory

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=1000,   # 超过此 token 数则压缩为摘要
    return_messages=True
)

# 手动添加历史
memory.save_context(
    {"input": "我是招商银行的客户经理"},
    {"output": "您好！请问有什么可以帮助您？"}
)

# 查看当前记忆
print(memory.load_memory_variables({}))

自定义记忆（Redis 持久化）

import redis
import json
from langchain_core.memory import BaseMemory
from typing import Dict, List, Any

class RedisMemory(BaseMemory):
    """基于 Redis 的持久化对话记忆"""
    
    redis_client: Any
    session_id: str
    max_history: int = 20
    
    class Config:
        arbitrary_types_allowed = True
    
    @property
    def memory_variables(self) -> List[str]:
        return ["history"]
    
    def load_memory_variables(self, inputs: Dict) -> Dict:
        key = f"chat_history:{self.session_id}"
        history = self.redis_client.lrange(key, -self.max_history * 2, -1)
        messages = [json.loads(m) for m in history]
        return {"history": messages}
    
    def save_context(self, inputs: Dict, outputs: Dict) -> None:
        key = f"chat_history:{self.session_id}"
        self.redis_client.rpush(
            key,
            json.dumps({"role": "user", "content": inputs["input"]}),
            json.dumps({"role": "assistant", "content": outputs["output"]})
        )
        self.redis_client.expire(key, 86400)  # 24小时过期

# 使用
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
memory = RedisMemory(redis_client=r, session_id="user_123")

LCEL 中的记忆管理

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import RedisChatMessageHistory

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是专业的金融客服助手"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

chain = prompt | llm

# 包装为带历史的链
chain_with_history = RunnableWithMessageHistory(
    chain,
    lambda session_id: RedisChatMessageHistory(
        session_id,
        url="redis://localhost:6379"
    ),
    input_messages_key="input",
    history_messages_key="history"
)

# 调用时传入 session_id
config = {"configurable": {"session_id": "user_456"}}
response = chain_with_history.invoke(
    {"input": "我想了解理财产品"},
    config=config
)