LangChain对话记忆原理

发布时间: 2025-03-29 19:08

↑

# LangChain对话记忆原理

本文将详细介绍LangChain中的对话记忆机制，帮助你理解和使用记忆组件来构建具有上下文感知能力的AI应用。

## 记忆组件简介

记忆组件（Memory）是LangChain中的核心功能之一，它使AI能够：
- 记住对话历史
- 维护上下文信息
- 实现连续对话
- 提供个性化响应

## 记忆类型

### 1. ConversationBufferMemory

最基础的记忆类型，存储完整对话历史：

```python
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain.llms import OpenAI

# 创建记忆组件
memory = ConversationBufferMemory()

# 创建对话链
conversation = ConversationChain(
    llm=OpenAI(),
    memory=memory,
    verbose=True
)

# 进行对话
conversation.predict(input="你好，我叫小明")
conversation.predict(input="还记得我的名字吗？")
```

### 2. ConversationBufferWindowMemory

带窗口的记忆，只保留最近k轮对话：

```python
from langchain.memory import ConversationBufferWindowMemory

# 创建窗口记忆
memory = ConversationBufferWindowMemory(k=2)

# 使用记忆组件
conversation = ConversationChain(
    llm=OpenAI(),
    memory=memory
)

# 对话超过窗口大小时，早期对话会被遗忘
conversation.predict(input="第一句话")
conversation.predict(input="第二句话")
conversation.predict(input="第三句话")  # 第一句话被遗忘
```

### 3. ConversationSummaryMemory

通过总结压缩对话历史：

```python
from langchain.memory import ConversationSummaryMemory

# 创建总结记忆
memory = ConversationSummaryMemory(llm=OpenAI())

# 使用记忆组件
conversation = ConversationChain(
    llm=OpenAI(),
    memory=memory
)

# 对话会被自动总结
for _ in range(5):
    conversation.predict(input="讲一个有趣的故事")

# 查看总结
print(memory.buffer)
```

### 4. ConversationTokenBufferMemory

基于token数量限制的记忆：

```python
from langchain.memory import ConversationTokenBufferMemory

# 创建token限制记忆
memory = ConversationTokenBufferMemory(
    llm=OpenAI(),
    max_token_limit=1000
)

# 使用记忆组件
conversation = ConversationChain(
    llm=OpenAI(),
    memory=memory
)

# 当对话token数超过限制时，早期对话会被删除
```

## 高级特性

### 1. 自定义记忆键

```python
from langchain.memory import ConversationBufferMemory

# 自定义记忆键
memory = ConversationBufferMemory(
    memory_key="chat_history",    # 历史记录键
    input_key="input",           # 输入键
    output_key="output"         # 输出键
)
```

### 2. 记忆持久化

```python
from langchain.memory import ConversationBufferMemory
import json

def save_memory(memory, file_path):
    # 保存记忆到文件
    with open(file_path, 'w') as f:
        json.dump(memory.chat_memory.messages, f)

def load_memory(file_path):
    # 从文件加载记忆
    memory = ConversationBufferMemory()
    with open(file_path, 'r') as f:
        messages = json.load(f)
        memory.chat_memory.messages = messages
    return memory
```

### 3. 记忆过滤

```python
from langchain.memory import ConversationBufferMemory

def filter_memory(memory, filter_func):
    # 过滤记忆内容
    filtered_messages = [
        msg for msg in memory.chat_memory.messages
        if filter_func(msg)
    ]
    memory.chat_memory.messages = filtered_messages
    return memory

# 使用示例
def sensitive_filter(message):
    # 过滤敏感信息
    sensitive_words = ["密码", "账号", "银行卡"]
    return not any(word in message.content for word in sensitive_words)
```

## 最佳实践

### 1. 记忆选择

根据应用场景选择合适的记忆类型：

1. **短对话**：ConversationBufferMemory
2. **长对话**：ConversationSummaryMemory
3. **固定窗口**：ConversationBufferWindowMemory
4. **Token限制**：ConversationTokenBufferMemory

### 2. 记忆管理

```python
class MemoryManager:
    def __init__(self, memory_type="buffer", **kwargs):
        if memory_type == "buffer":
            self.memory = ConversationBufferMemory(**kwargs)
        elif memory_type == "window":
            self.memory = ConversationBufferWindowMemory(**kwargs)
        elif memory_type == "summary":
            self.memory = ConversationSummaryMemory(**kwargs)
        else:
            raise ValueError(f"不支持的记忆类型：{memory_type}")
    
    def add_message(self, message):
        self.memory.chat_memory.add_message(message)
    
    def clear(self):
        self.memory.clear()
    
    def get_history(self):
        return self.memory.load_memory_variables({})
```

### 3. 错误处理

```python
from typing import Optional

def safe_memory_operation(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            print(f"记忆操作错误：{e}")
            return None
    return wrapper

@safe_memory_operation
def get_chat_history(memory) -> Optional[str]:
    return memory.load_memory_variables({}).get("history")
```

## 应用示例

### 1. 智能客服

```python
from langchain.chains import ConversationChain
from langchain.memory import ConversationSummaryMemory

def create_customer_service():
    # 创建带总结记忆的客服
    memory = ConversationSummaryMemory(
        llm=OpenAI(),
        prompt=PromptTemplate(
            input_variables=["summary"],
            template="客服记录摘要：{summary}"
        )
    )
    
    # 创建对话链
    service = ConversationChain(
        llm=OpenAI(),
        memory=memory,
        prompt=PromptTemplate(
            input_variables=["history", "input"],
            template="""你是一个专业的客服代表。
            
            对话历史：{history}
            客户：{input}
            客服："""
        )
    )
    
    return service
```

### 2. 个性化助手

```python
from langchain.memory import ConversationBufferMemory

def create_personal_assistant():
    # 创建个性化记忆
    memory = ConversationBufferMemory(
        input_key="input",
        memory_key="chat_history",
        return_messages=True
    )
    
    # 创建助手
    assistant = ConversationChain(
        llm=OpenAI(),
        memory=memory,
        prompt=PromptTemplate(
            input_variables=["chat_history", "input"],
            template="""你是一个个性化AI助手。
            根据对话历史理解用户偏好。
            
            对话历史：{chat_history}
            用户：{input}
            助手："""
        )
    )
    
    return assistant
```

## 总结

LangChain的记忆组件提供了强大的对话管理能力：
- 多种记忆类型
- 灵活的配置选项
- 持久化支持
- 记忆过滤机制

掌握记忆组件的使用，将帮助你构建更智能和个性化的AI应用。

元素码农