从零开始学LangChain（五）：Memory 记忆管理

从零开始学LangChain（五）：Memory 记忆管理

本系列教程将带你从零开始学习LangChain框架，构建强大的AI应用程序。

为什么需要记忆？

想象一下与一个”健忘”的机器人对话的情景。用户说”我叫张三”，机器人回答”你好张三！”，但当用户接着问”我叫什么名字？”时，机器人却回答”对不起，我不知道你的名字”。这就是没有记忆的对话。

如果有记忆，当用户再次问”我叫什么名字？”时，机器人会正确回答”你叫张三。”这种能力对对话系统至关重要。

LangChain提供了多种记忆组件来保存和检索对话上下文。ConversationBufferMemory保存所有对话历史，适合短期对话场景。ConversationBufferWindowMemory只保留最近k轮对话，避免token超限，适用于长期对话。ConversationSummaryMemory对历史进行摘要节省token。ConversationKGMemory使用知识图谱存储复杂关系。

ConversationBufferMemory

最简单的记忆类型，保存所有对话历史。

基础用法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain

# 创建记忆
memory = ConversationBufferMemory()

# 创建对话链
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 第一轮对话
response1 = conversation.predict(input="你好，我叫张三")
print(response1)

# 第二轮对话
response2 = conversation.predict(input="我叫什么名字？")
print(response2)

输出会显示完整的对话历史，AI能够记住用户的名字。

手动管理记忆

1
2
3
4
5
6
7
8
9
10
11
from langchain.memory import ConversationBufferMemory

# 创建记忆
memory = ConversationBufferMemory()

# 手动添加对话
memory.save_context({"input": "我叫张三"}, {"output": "你好张三！"})
memory.save_context({"input": "我喜欢编程"}, {"output": "太棒了！"})

# 查看对话历史
print(memory.load_memory_variables({}))

返回的字典包含完整的对话历史字符串。

ConversationBufferWindowMemory

只保存最近k轮对话，避免token超限。

限制对话窗口

1
2
3
4
5
6
7
8
9
10
11
12
13
from langchain.memory import ConversationBufferWindowMemory

# 只保存最近2轮对话
memory = ConversationBufferWindowMemory(k=2)

# 添加多轮对话
memory.save_context({"input": "你好"}, {"output": "嗨！"})
memory.save_context({"input": "我叫张三"}, {"output": "你好张三！"})
memory.save_context({"input": "我喜欢Python"}, {"output": "Python很棒！"})
memory.save_context({"input": "我最喜欢什么？"}, {"output": "你最喜欢..."})

# 只保留最后2轮
print(memory.load_memory_variables({}))

前面的对话已被自动丢弃，只保留最近的两轮。

ConversationSummaryMemory

对对话历史进行摘要，节省token。

自动摘要对话

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 创建摘要记忆
memory = ConversationSummaryMemory(llm=llm)

# 添加对话（会自动摘要）
memory.save_context(
    {"input": "我叫张三，是一名程序员"},
    {"output": "你好张三！很高兴认识程序员朋友。"}
)

memory.save_context(
    {"input": "我主要用Python和Java开发"},
    {"output": "Python和Java都是优秀的语言！"}
)

# 查看摘要
print(memory.load_memory_variables({}))

返回的是对话历史的摘要文本，而不是完整对话。

在链中使用记忆

自定义链使用记忆

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from langchain.schema.runnable import RunnablePassthrough

llm = ChatOpenAI(model="gpt-3.5-turbo")

# 创建包含历史消息占位符的提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位友好的AI助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

# 创建记忆
memory = ConversationBufferMemory(return_messages=True)

# 创建链
chain = (
    RunnablePassthrough.assign(
        history=memory.load_memory_variables | lambda x: x["history"]
    )
    | prompt
    | llm
)

# 使用并保存记忆
def chat_with_memory(user_input):
    """带记忆的对话"""
    # 获取响应
    response = chain.invoke({"input": user_input})

    # 保存到记忆
    memory.save_context(
        {"input": user_input},
        {"output": response.content}
    )

    return response.content

# 测试
print(chat_with_memory("我叫张三"))
print(chat_with_memory("我叫什么名字？"))

使用LCEL简化记忆链

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from langchain.schema.runnable import RunnablePassthrough

llm = ChatOpenAI(model="gpt-3.5-turbo")

# 更简洁的LCEL写法
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位Python导师。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

memory = ConversationBufferMemory(return_messages=True)

chain = (
    RunnablePassthrough.assign(
        history=RunnablePassthrough.assign(
            history=memory.load_memory_variables | lambda x: x["history"]
        )
    )
    | prompt
    | llm
)

# 第一次对话
response1 = chain.invoke({"input": "什么是装饰器？"})
memory.save_context(
    {"input": "什么是装饰器？"},
    {"output": response1.content}
)

# 第二次对话（记住上下文）
response2 = chain.invoke({"input": "给我一个例子"})
print(response2.content)

构建聊天机器人

让我们创建一个完整的聊天机器人系统。

完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferWindowMemory
from langchain.schema.runnable import RunnablePassthrough, RunnableBranch
from typing import Dict

class ChatBot:
    """智能聊天机器人"""

    def __init__(self, system_message: str = "你是一位友好的AI助手。"):
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
        self.memory = ConversationBufferWindowMemory(
            k=5,
            return_messages=True
        )

        # 创建提示词模板
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", system_message),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

    def create_chain(self):
        """创建对话链"""

        def load_memory(input_dict):
            """加载记忆"""
            return {
                "history": self.memory.load_memory_variables({})["history"],
                "input": input_dict["input"]
            }

        chain = (
            RunnablePassthrough.assign(**load_memory)
            | self.prompt
            | self.llm
        )

        return chain

    def chat(self, user_input: str) -> str:
        """与机器人对话"""
        chain = self.create_chain()
        response = chain.invoke({"input": user_input})

        # 保存对话到记忆
        self.memory.save_context(
            {"input": user_input},
            {"output": response.content}
        )

        return response.content

    def clear_memory(self):
        """清除对话历史"""
        self.memory.clear()

# 使用示例
bot = ChatBot(system_message="你是一位Python编程导师，擅长用比喻解释概念。")

print("=== 开始对话 ===")
print(bot.chat("你好"))
print(bot.chat("我想学习Python"))
print(bot.chat("什么是生成器？"))
print(bot.chat("给我一个例子"))
print(bot.chat("生成器和列表有什么区别？"))

# 清除记忆开始新对话
bot.clear_memory()
print("\n=== 新对话 ===")
print(bot.chat("你好，我想学Java"))

记忆的高级用法

多用户记忆管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from langchain.memory import ConversationBufferMemory
from typing import Dict

class MultiUserMemory:
    """多用户记忆管理"""

    def __init__(self):
        self.memories: Dict[str, ConversationBufferMemory] = {}

    def get_memory(self, user_id: str) -> ConversationBufferMemory:
        """获取用户的记忆"""
        if user_id not in self.memories:
            self.memories[user_id] = ConversationBufferMemory()
        return self.memories[user_id]

    def clear_user(self, user_id: str):
        """清除特定用户的记忆"""
        if user_id in self.memories:
            del self.memories[user_id]

# 使用
multi_memory = MultiUserMemory()

# 用户A的对话
memory_a = multi_memory.get_memory("user_a")
memory_a.save_context({"input": "我叫张三"}, {"output": "你好张三"})

# 用户B的对话
memory_b = multi_memory.get_memory("user_b")
memory_b.save_context({"input": "我叫李四"}, {"output": "你好李四"})

# 各自独立
print("用户A:", memory_a.load_memory_variables({}))
print("用户B:", memory_b.load_memory_variables({}))

持久化记忆

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from langchain.memory import ConversationBufferMemory
import json
import os

class PersistentMemory:
    """持久化记忆到文件"""

    def __init__(self, filepath: str = "chat_history.json"):
        self.filepath = filepath
        self.memory = ConversationBufferMemory()
        self.load()

    def save(self):
        """保存到文件"""
        history = self.memory.load_memory_variables({})
        with open(self.filepath, 'w', encoding='utf-8') as f:
            json.dump(history, f, ensure_ascii=False, indent=2)

    def load(self):
        """从文件加载"""
        if os.path.exists(self.filepath):
            with open(self.filepath, 'r', encoding='utf-8') as f:
                history = json.load(f)
                # 恢复记忆（需要解析history字符串）

    def add_message(self, user_input: str, ai_response: str):
        """添加对话并保存"""
        self.memory.save_context(
            {"input": user_input},
            {"output": ai_response}
        )
        self.save()

# 使用
persistent_memory = PersistentMemory("my_chat.json")
persistent_memory.add_message("你好", "你好！")
persistent_memory.add_message("我叫张三", "你好张三！")

实战示例：客服机器人

创建一个带记忆的智能客服机器人。

完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferWindowMemory
from langchain.schema.runnable import RunnablePassthrough

class CustomerServiceBot:
    """客服机器人"""

    def __init__(self, company_name: str, product_info: str):
        self.company_name = company_name
        self.product_info = product_info

        # 系统提示词
        system_prompt = f"""你是{company_name}的客服代表。

产品信息：
{product_info}

你的职责：
- 友好耐心地回答客户问题
- 主动推荐合适的产品
- 记住客户的偏好和历史对话
- 遇到无法解决的问题，礼貌地转人工客服

注意事项：
- 保持专业和礼貌
- 不随意承诺无法兑现的服务
- 记住客户提到的个人信息"""

        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.5)
        self.memory = ConversationBufferWindowMemory(
            k=10,
            return_messages=True
        )

        # 创建提示词模板
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", system_prompt),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{input}")
        ])

        self.chain = (
            RunnablePassthrough.assign(
                history=self.memory.load_memory_variables | lambda x: x["history"]
            )
            | self.prompt
            | self.llm
        )

    def respond(self, user_input: str) -> str:
        """响应客户"""
        response = self.chain.invoke({"input": user_input})
        reply = response.content

        # 保存对话
        self.memory.save_context(
            {"input": user_input},
            {"output": reply}
        )

        return reply

    def get_conversation_summary(self) -> str:
        """获取对话摘要"""
        history = self.memory.load_memory_variables({})["history"]
        return f"对话轮数：{len(history) // 2}"

# 使用示例
bot = CustomerServiceBot(
    company_name="TechStore",
    product_info="我们销售电脑、手机、平板等电子产品，提供7天无理由退换货服务。"
)

print("=== 客服对话 ===")
print("客户: 你们有什么电脑？")
print("客服:", bot.respond("你们有什么电脑？"))

print("\n客户: 我想要一款轻薄本，预算8000元")
print("客服:", bot.respond("我想要一款轻薄本，预算8000元"))

print("\n客户: 有红色的吗？")
print("客服:", bot.respond("有红色的吗？"))

print("\n", bot.get_conversation_summary())

常见问题

Q1: 记忆占用太多token怎么办？

可以使用ConversationBufferWindowMemory限制记忆窗口大小，只保留最近的几轮对话。或者使用ConversationSummaryMemory，让LLM自动总结历史对话，节省token。

Q2: 如何让AI记住特定信息？

将关键信息添加到系统提示中，或者手动保存到记忆里。例如：

1
2
3
4
5
6
7
8
9
10
from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory()

# 添加关键信息到系统提示
key_info = "用户姓名：张三，职业：程序员，偏好：Python"
memory.save_context(
    {"input": f"重要信息：{key_info}"},
    {"output": "已记录。"}
)

系列导航