第6章：记忆与对话

本章将带你深入了解LangChain的记忆机制，学习如何让AI记住对话历史，构建具有上下文感知能力的智能聊天机器人，并掌握记忆与RAG结合的高级技巧。

1. 为什么需要记忆

在大语言模型的应用中，记忆（Memory）是一个至关重要的组件。它允许AI记住之前的对话内容，从而在多轮交互中保持上下文连贯性。

没有记忆的问题

想象一下这样的对话场景：

上面的问题展示了没有记忆的AI的局限性：

• 上下文丢失：无法引用之前对话中的信息
• 体验割裂：每轮对话都像第一次交流
• 无法追问：用户无法基于之前的话题深入讨论
• 重复提问：AI可能反复询问相同的信息

记忆的作用

引入记忆机制后，AI可以：

• ✅ 记住用户信息：姓名、偏好、背景等
• ✅ 维护对话上下文：理解指代、承接话题
• ✅ 支持多轮交互：逐步澄清、深入探讨
• ✅ 个性化回应：基于历史调整回复风格

2. 记忆类型详解

LangChain提供了多种记忆类型，每种都有特定的使用场景。让我们逐一了解。

2.1 ConversationBufferMemory - 完整对话记忆

这是最基础的记忆类型，它会完整保存所有的对话历史。

from langchain.memory import ConversationBufferMemory

# 创建缓冲记忆
memory = ConversationBufferMemory()

# 保存对话上下文
# 用户消息
memory.save_context(
    {"input": "你好，我叫李四"},
    {"output": "你好李四！很高兴认识你。有什么我可以帮助你的吗？"}
)

# AI回复后，继续保存下一轮
memory.save_context(
    {"input": "我想学习Python"},
    {"output": "太好了！Python是非常流行的编程语言。你有编程基础吗？"}
)

# 查看保存的记忆
print("当前记忆内容：")
print(memory.load_memory_variables({}))

# 输出：
# {'history': 'Human: 你好，我叫李四\nAI: 你好李四！很高兴认识你。有什么我可以帮助你的吗？\nHuman: 我想学习Python\nAI: 太好了！Python是非常流行的编程语言。你有编程基础吗？'}

2.2 ConversationBufferWindowMemory - 窗口记忆

只保留最近的k轮对话，避免记忆过长导致token超限。

from langchain.memory import ConversationBufferWindowMemory

# 创建窗口记忆，只保留最近2轮对话
memory = ConversationBufferWindowMemory(k=2)

# 模拟多轮对话
conversations = [
    ("你好", "你好！有什么可以帮助你？"),
    ("今天天气怎么样？", "我无法获取实时天气信息。"),
    ("你会做什么？", "我可以回答问题、写作、编程等。"),
    ("教我Python", "好的，Python入门很简单..."),
    ("变量是什么？", "变量是用来存储数据的容器...")
]

for user_msg, ai_msg in conversations:
    memory.save_context({"input": user_msg}, {"output": ai_msg})

# 查看记忆 - 只保留最后2轮
print("窗口记忆内容（k=2）：")
print(memory.load_memory_variables({}))

# 输出只包含最后两轮：
# {'history': 'Human: 教我Python\nAI: 好的，Python入门很简单...\nHuman: 变量是什么？\nAI: 变量是用来存储数据的容器...'}

2.3 ConversationSummaryMemory - 对话摘要记忆

随着对话进行，自动生成摘要，用摘要代替原始对话历史，节省token。

from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI

# 需要LLM来生成摘要
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,
    api_key="your-api-key"
)

# 创建摘要记忆
memory = ConversationSummaryMemory(
    llm=llm,
    max_token_limit=100  # 摘要的最大token数
)

# 保存长对话
memory.save_context(
    {"input": "你好，我想预订酒店"},
    {"output": "好的，请问您想预订哪个城市的酒店？入住日期是哪天？"}
)

memory.save_context(
    {"input": "我想预订北京的酒店，明天入住，住3天"},
    {"output": "明白了。您对酒店有什么要求吗？比如价格区间、星级、位置等？"}
)

memory.save_context(
    {"input": "想要价格在500-800之间，四星级以上，最好靠近天安门"},
    {"output": "好的，我为您查找符合条件的酒店。请稍等..."}
)

# 查看摘要形式的记忆
print("摘要记忆内容：")
print(memory.load_memory_variables({}))

# 输出示例：
# {'history': '用户想要预订北京的酒店，明天入住，住3天。预算500-800元，要求四星级以上，靠近天安门。AI正在为用户查找符合条件的酒店。'}

2.4 VectorStoreRetrieverMemory - 向量检索记忆

将对话历史存储在向量数据库中，根据语义相似度检索相关的历史信息。

from langchain.memory import VectorStoreRetrieverMemory
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI

# 创建向量存储
embeddings = OpenAIEmbeddings(api_key="your-api-key")
vectorstore = Chroma(
    embedding_function=embeddings,
    collection_name="conversation_memory"
)

# 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 创建向量记忆
memory = VectorStoreRetrieverMemory(
    retriever=retriever,
    memory_key="history",
    input_key="input"
)

# 保存多轮对话（模拟历史对话）
historical_conversations = [
    ("我最喜欢的颜色是蓝色", "蓝色是很受欢迎的颜色，给人宁静的感觉。"),
    ("我养了一只叫小白的猫", "小白一定很可爱！猫是很多人喜欢的宠物。"),
    ("我在北京工作", "北京是繁华的大都市，工作机会很多。"),
    ("我喜欢吃川菜", "川菜以麻辣著称，非常开胃。"),
]

for user_msg, ai_msg in historical_conversations:
    memory.save_context({"input": user_msg}, {"output": ai_msg})

# 查询与"宠物"相关的记忆
print("查询与'宠物'相关的记忆：")
print(memory.load_memory_variables({"input": "我想养宠物"}))

# 查询与"食物"相关的记忆
print("\n查询与'食物'相关的记忆：")
print(memory.load_memory_variables({"input": "推荐美食"}))

2.5 记忆类型对比

3. ConversationChain 对话链

ConversationChain是LangChain提供的带记忆功能的对话链，它将LLM、记忆和提示词模板组合在一起。

3.1 基础用法

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

# 初始化组件
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.7,
    api_key="your-api-key"
)

memory = ConversationBufferMemory()

# 创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True  # 打印详细执行过程
)

# 进行对话
response1 = conversation.predict(input="你好，我叫王五")
print(f"AI: {response1}\n")

response2 = conversation.predict(input="我喜欢打篮球")
print(f"AI: {response2}\n")

response3 = conversation.predict(input="你知道我叫什么名字吗？")
print(f"AI: {response3}\n")

# 查看完整记忆
print("完整记忆：")
print(memory.buffer)

3.2 自定义提示词模板

from langchain.prompts import PromptTemplate

# 自定义对话提示词模板
custom_template = """你是一个友好、乐于助人的AI助手。请根据对话历史回答用户的问题。

当前对话历史：
{history}

用户：{input}
AI助手："""

CUSTOM_PROMPT = PromptTemplate(
    input_variables=["history", "input"],
    template=custom_template
)

# 使用自定义提示词创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    prompt=CUSTOM_PROMPT,
    verbose=False
)

response = conversation.predict(input="你好！")
print(response)

3.3 保存和加载对话历史

import json

# ========== 保存对话历史 ==========
def save_conversation(memory, filepath="conversation_history.json"):
    """保存对话历史到文件"""
    history = memory.load_memory_variables({})["history"]
    with open(filepath, "w", encoding="utf-8") as f:
        json.dump({"history": history}, f, ensure_ascii=False, indent=2)
    print(f"对话历史已保存到：{filepath}")

# 保存当前对话
save_conversation(memory, "my_chat_history.json")

# ========== 加载对话历史 ==========
def load_conversation(memory, filepath="conversation_history.json"):
    """从文件加载对话历史"""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            data = json.load(f)
            # 注意：这里只是示例，实际加载需要解析history格式
            # 不同记忆类型的加载方式可能不同
            print(f"已加载历史：{data['history'][:100]}...")
            return data
    except FileNotFoundError:
        print("历史文件不存在")
        return None

# 加载历史
loaded_data = load_conversation(memory, "my_chat_history.json")

# ========== 使用消息列表加载 ==========
from langchain.schema import HumanMessage, AIMessage

# 手动构建消息历史
messages = [
    HumanMessage(content="你好"),
    AIMessage(content="你好！很高兴见到你。"),
    HumanMessage(content="今天天气不错"),
    AIMessage(content="是的，适合出去走走。")
]

# 创建新的记忆并设置历史
new_memory = ConversationBufferMemory()
new_memory.chat_memory.messages = messages

print("\n从消息列表加载的记忆：")
print(new_memory.load_memory_variables({}))

4. 聊天机器人实战

本节将构建一个功能完善的聊天机器人，包含记忆功能、系统提示词（性格设定）和多轮对话处理。

4.1 基础聊天机器人

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

class SimpleChatBot:
    """简单聊天机器人"""
    
    def __init__(self, api_key: str, window_size: int = 5):
        """
        初始化聊天机器人
        
        Args:
            api_key: OpenAI API密钥
            window_size: 记忆窗口大小（保留的轮数）
        """
        self.llm = ChatOpenAI(
            model="gpt-3.5-turbo",
            temperature=0.7,
            api_key=api_key
        )
        
        # 使用窗口记忆，控制上下文长度
        self.memory = ConversationBufferWindowMemory(k=window_size)
        
        # 创建对话链
        self.conversation = ConversationChain(
            llm=self.llm,
            memory=self.memory,
            verbose=False
        )
    
    def chat(self, user_input: str) -> str:
        """
        处理用户输入并返回回复
        
        Args:
            user_input: 用户输入的消息
            
        Returns:
            AI的回复
        """
        response = self.conversation.predict(input=user_input)
        return response
    
    def reset(self):
        """重置对话历史"""
        self.memory.clear()
        print("🔄 对话历史已重置")


# 使用示例
if __name__ == "__main__":
    import os
    
    bot = SimpleChatBot(api_key=os.getenv("OPENAI_API_KEY"))
    
    print("🤖 聊天机器人已启动（输入 'quit' 退出）\n")
    
    while True:
        user_input = input("你：").strip()
        
        if user_input.lower() in ['quit', 'exit', '退出']:
            print("👋 再见！")
            break
        
        if user_input.lower() == 'reset':
            bot.reset()
            continue
        
        if not user_input:
            continue
        
        response = bot.chat(user_input)
        print(f"🤖：{response}\n")

4.2 带性格设定的聊天机器人

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema import SystemMessage, HumanMessage, AIMessage
from langchain.memory import ConversationBufferMemory
from langchain.chains import LLMChain
from langchain_openai import ChatOpenAI

class CharacterChatBot:
    """带性格设定的聊天机器人"""
    
    def __init__(
        self,
        api_key: str,
        character_description: str = None,
        window_size: int = 10
    ):
        """
        初始化角色聊天机器人
        
        Args:
            api_key: OpenAI API密钥
            character_description: 角色设定描述
            window_size: 记忆窗口大小
        """
        # 默认角色设定
        default_character = """你是一个知识渊博、幽默风趣的AI助手。
你擅长用通俗易懂的方式解释复杂概念，偶尔会用恰当的比喻。
你的回答简洁明了，但富有洞察力。"""
        
        self.character = character_description or default_character
        
        self.llm = ChatOpenAI(
            model="gpt-3.5-turbo",
            temperature=0.8,  # 稍高的温度增加创造性
            api_key=api_key
        )
        
        # 创建带记忆的提示词模板
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", self.character),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{input}")
        ])
        
        # 初始化记忆
        self.memory = ConversationBufferWindowMemory(
            k=window_size,
            return_messages=True,  # 返回消息对象列表
            memory_key="history"
        )
        
        # 创建链
        self.chain = LLMChain(
            llm=self.llm,
            prompt=self.prompt,
            memory=self.memory,
            verbose=False
        )
    
    def chat(self, user_input: str) -> str:
        """处理用户输入"""
        response = self.chain.predict(input=user_input)
        return response
    
    def get_history(self) -> list:
        """获取对话历史"""
        return self.memory.load_memory_variables({})["history"]


# 使用不同性格的机器人
if __name__ == "__main__":
    import os
    
    # 定义不同的角色
    characters = {
        "1": {
            "name": "温柔姐姐",
            "desc": """你是一位温柔体贴的大姐姐，说话轻声细语，总是给予鼓励和支持。
你善于倾听，会耐心地帮助对方解决问题。你的话语中常带有温暖和关怀。"""
        },
        "2": {
            "name": "极客程序员",
            "desc": """你是一个资深程序员，技术宅，说话直接简洁。
你热爱代码和技术，经常引用编程概念来解释日常事物。
你会用技术术语，但会解释清楚。"""
        },
        "3": {
            "name": "历史老师",
            "desc": """你是一位博学的历史老师，喜欢将话题与历史事件联系起来。
你的回答常包含历史典故，引经据典，让对话富有文化底蕴。"""
        }
    }
    
    # 选择角色
    print("请选择聊天角色：")
    for key, char in characters.items():
        print(f"  {key}. {char['name']}")
    
    choice = input("\n输入编号（默认1）：").strip() or "1"
    selected = characters.get(choice, characters["1"])
    
    print(f"\n🎭 已选择：{selected['name']}")
    print("=" * 50)
    
    # 创建机器人
    bot = CharacterChatBot(
        api_key=os.getenv("OPENAI_API_KEY"),
        character_description=selected["desc"]
    )
    
    # 开始对话
    print(f"你好！我是{selected['name']}，很高兴和你聊天~\n")
    
    while True:
        user_input = input("你：").strip()
        
        if user_input.lower() in ['quit', 'exit']:
            print("👋 再见！")
            break
        
        if not user_input:
            continue
        
        response = bot.chat(user_input)
        print(f"{selected['name']}：{response}\n")

4.3 多轮对话处理技巧

class AdvancedChatBot:
    """高级聊天机器人，支持更复杂的多轮对话处理"""
    
    def __init__(self, api_key: str):
        self.llm = ChatOpenAI(
            model="gpt-3.5-turbo",
            temperature=0.7,
            api_key=api_key
        )
        
        # 使用摘要记忆处理长对话
        self.memory = ConversationSummaryMemory(
            llm=self.llm,
            max_token_limit=200,
            return_messages=True
        )
        
        # 系统提示词包含对话管理指令
        system_prompt = """你是一个智能助手。在对话中请注意：
1. 记住用户提到的关键信息（如名字、偏好、需求）
2. 如果用户的问题指代不清，根据上下文合理推断或礼貌询问
3. 保持对话连贯，自然地承接上文话题
4. 如果话题发生变化，平滑过渡"""
        
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", system_prompt),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{input}")
        ])
        
        self.chain = LLMChain(
            llm=self.llm,
            prompt=self.prompt,
            memory=self.memory
        )
        
        # 存储用户关键信息
        self.user_profile = {}
    
    def chat(self, user_input: str) -> str:
        """处理输入，包含信息提取"""
        # 提取关键信息（简单示例）
        self._extract_info(user_input)
        
        # 生成回复
        response = self.chain.predict(input=user_input)
        return response
    
    def _extract_info(self, text: str):
        """从对话中提取用户关键信息"""
        # 简单的关键词提取，实际可用NER等更复杂方法
        if "我叫" in text or "我是" in text:
            # 提取名字逻辑（简化版）
            import re
            match = re.search(r"我叫(\w+)", text)
            if match:
                self.user_profile["name"] = match.group(1)
        
        if "我喜欢" in text:
            match = re.search(r"我喜欢(.+?)[。，]", text)
            if match:
                likes = self.user_profile.get("likes", [])
                likes.append(match.group(1))
                self.user_profile["likes"] = likes
    
    def get_profile(self) -> dict:
        """获取用户画像"""
        return self.user_profile


# 使用示例
def demo_multiturn():
    """演示多轮对话"""
    import os
    
    bot = AdvancedChatBot(api_key=os.getenv("OPENAI_API_KEY"))
    
    # 模拟多轮对话
    conversations = [
        "你好，我是小明",
        "我想学习编程",
        "哪种语言适合初学者？",
        "Python怎么样？",
        "你能教我Python吗？",
        "我们之前聊了什么？",  # 测试记忆
        "我叫什么名字？",      # 测试信息提取
    ]
    
    print("🤖 多轮对话演示\n")
    print("=" * 50)
    
    for user_msg in conversations:
        print(f"👤 用户：{user_msg}")
        response = bot.chat(user_msg)
        print(f"🤖 AI：{response}")
        print("-" * 50)
    
    print(f"\n📊 提取的用户画像：{bot.get_profile()}")


if __name__ == "__main__":
    demo_multiturn()

5. 记忆与RAG结合

在实际应用中，我们通常需要AI既能记住对话历史，又能查阅知识库。这就需要将记忆和RAG结合起来。

5.1 ConversationalRetrievalQA链

LangChain提供了ConversationalRetrievalQA链，专门为这种场景设计。

from langchain.chains import ConversationalRetrievalQA
from langchain.memory import ConversationBufferMemory
from langchain_chroma import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

def create_conversational_rag():
    """创建带记忆功能的RAG问答系统"""
    
    # ========== 1. 初始化组件 ==========
    llm = ChatOpenAI(
        model="gpt-3.5-turbo",
        temperature=0,
        api_key="your-api-key"
    )
    
    embeddings = OpenAIEmbeddings(api_key="your-api-key")
    
    # 加载已有的向量存储（假设之前已创建）
    vectorstore = Chroma(
        persist_directory="./rag_vector_db",
        embedding_function=embeddings
    )
    
    # ========== 2. 创建记忆 ==========
    memory = ConversationBufferMemory(
        memory_key="chat_history",  # 关键：必须使用chat_history
        return_messages=True,        # 返回消息对象
        output_key="answer"          # 指定输出键
    )
    
    # ========== 3. 创建ConversationalRetrievalQA链 ==========
    qa_chain = ConversationalRetrievalQA.from_llm(
        llm=llm,
        retriever=vectorstore.as_retriever(
            search_kwargs={"k": 4}
        ),
        memory=memory,
        return_source_documents=True,  # 返回来源文档
        verbose=True
    )
    
    return qa_chain, memory


# 使用示例
def chat_with_knowledge_base():
    """与知识库进行多轮对话"""
    
    qa_chain, memory = create_conversational_rag()
    
    print("🤖 知识库问答助手（输入 'quit' 退出）\n")
    print("=" * 60)
    
    while True:
        question = input("\n❓ 你的问题：").strip()
        
        if question.lower() in ['quit', 'exit']:
            print("👋 再见！")
            break
        
        if not question:
            continue
        
        # 执行问答
        result = qa_chain.invoke({"question": question})
        
        print("\n📝 回答：")
        print(result["answer"])
        
        if result["source_documents"]:
            print("\n📚 参考来源：")
            sources = set()
            for doc in result["source_documents"]:
                source = doc.metadata.get('source', 'unknown')
                sources.add(source)
            for src in sources:
                print(f"  • {src}")


if __name__ == "__main__":
    import os
    os.environ["OPENAI_API_KEY"] = "your-api-key"
    chat_with_knowledge_base()

5.2 自定义带记忆的RAG提示词

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chains import ConversationalRetrievalQA
from langchain.chains.question_answering import load_qa_chain

class CustomConversationalRAG:
    """自定义带记忆的RAG系统"""
    
    def __init__(self, vectorstore, api_key: str):
        self.llm = ChatOpenAI(
            model="gpt-3.5-turbo",
            temperature=0,
            api_key=api_key
        )
        
        # 自定义组合文档的提示词
        self.combine_docs_prompt = ChatPromptTemplate.from_template("""基于以下检索到的资料回答问题。
如果资料中没有相关信息，请明确说明。

资料：
{context}

问题：{question}

请给出准确、简洁的回答：""")
        
        # 自定义最终问答提示词（包含对话历史）
        self.qa_prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一个专业的知识库助手。结合对话历史和检索资料回答问题。"),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", """基于以下资料回答问题：
{context}

问题：{question}""")
        ])
        
        # 创建文档组合链
        doc_chain = load_qa_chain(
            llm=self.llm,
            chain_type="stuff",
            prompt=self.combine_docs_prompt
        )
        
        # 创建记忆
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True,
            output_key="answer"
        )
        
        # 创建问答链
        self.qa_chain = ConversationalRetrievalQA(
            combine_docs_chain=doc_chain,
            retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
            memory=self.memory,
            return_source_documents=True
        )
    
    def ask(self, question: str) -> dict:
        """提问"""
        return self.qa_chain.invoke({"question": question})

5.3 记忆与RAG结合的完整示例

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
带记忆的智能客服系统
结合知识库检索和对话历史
"""

import os
from typing import List, Dict
from dataclasses import dataclass

from langchain.chains import ConversationalRetrievalQA
from langchain.memory import ConversationBufferWindowMemory
from langchain_chroma import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.schema import Document


@dataclass
class ChatResponse:
    """聊天响应数据结构"""
    answer: str
    source_documents: List[Document]
    chat_history: List


class SmartCustomerService:
    """智能客服系统 - 结合知识库和对话记忆"""
    
    def __init__(
        self,
        db_path: str = "./customer_service_db",
        api_key: str = None,
        window_size: int = 5
    ):
        """
        初始化智能客服
        
        Args:
            db_path: 向量数据库路径
            api_key: OpenAI API密钥
            window_size: 对话记忆窗口大小
        """
        self.api_key = api_key or os.getenv("OPENAI_API_KEY")
        if not self.api_key:
            raise ValueError("请提供API密钥")
        
        # 初始化LLM
        self.llm = ChatOpenAI(
            model="gpt-3.5-turbo",
            temperature=0.2,  # 客服场景，降低创造性提高准确性
            api_key=self.api_key
        )
        
        # 初始化嵌入模型
        self.embeddings = OpenAIEmbeddings(api_key=self.api_key)
        
        # 加载向量存储
        if not os.path.exists(db_path):
            raise FileNotFoundError(
                f"知识库不存在：{db_path}\n请先构建知识库索引"
            )
        
        self.vectorstore = Chroma(
            persist_directory=db_path,
            embedding_function=self.embeddings
        )
        
        # 创建窗口记忆
        self.memory = ConversationBufferWindowMemory(
            k=window_size,
            memory_key="chat_history",
            return_messages=True,
            output_key="answer"
        )
        
        # 创建RAG链
        self.qa_chain = ConversationalRetrievalQA.from_llm(
            llm=self.llm,
            retriever=self.vectorstore.as_retriever(
                search_type="similarity",
                search_kwargs={"k": 3}
            ),
            memory=self.memory,
            return_source_documents=True,
            verbose=False
        )
        
        # 系统提示词（用于指导AI的客服行为）
        self.system_prompt = """你是一家科技公司的智能客服助手。请遵循以下原则：
1. 基于知识库资料回答，不要编造信息
2. 记住用户之前说过的话，保持对话连贯
3. 如果用户问题需要澄清，礼貌地询问
4. 回答简洁专业，避免冗长
5. 如果用户不满，先表示理解和歉意"""
        
        print("✅ 智能客服系统初始化完成")
        print(f"   知识库路径：{db_path}")
        print(f"   记忆窗口：{window_size}轮")
    
    def handle_query(self, question: str) -> ChatResponse:
        """
        处理用户查询
        
        Args:
            question: 用户问题
            
        Returns:
            ChatResponse对象
        """
        # 执行问答
        result = self.qa_chain.invoke({"question": question})
        
        # 构建响应
        response = ChatResponse(
            answer=result["answer"],
            source_documents=result.get("source_documents", []),
            chat_history=self.memory.load_memory_variables({})["chat_history"]
        )
        
        return response
    
    def get_context(self) -> str:
        """获取当前对话上下文（用于调试）"""
        vars = self.memory.load_memory_variables({})
        history = vars.get("chat_history", [])
        
        if not history:
            return "（无对话历史）"
        
        context = []
        for msg in history[-6:]:  # 只显示最近6条
            role = "用户" if msg.type == "human" else "AI"
            context.append(f"{role}：{msg.content[:50]}...")
        
        return "\n".join(context)
    
    def reset_conversation(self):
        """重置对话"""
        self.memory.clear()
        print("🔄 对话已重置")
    
    def run_interactive(self):
        """运行交互式客服"""
        print("\n" + "=" * 60)
        print("🎧 智能客服系统")
        print("=" * 60)
        print("\n命令：")
        print("  • 输入 'quit' 退出")
        print("  • 输入 'reset' 重置对话")
        print("  • 输入 'context' 查看当前上下文")
        print("=" * 60 + "\n")
        
        while True:
            try:
                user_input = input("👤 客户：").strip()
                
                if not user_input:
                    continue
                
                if user_input.lower() in ['quit', 'exit', '退出']:
                    print("\n👋 感谢使用，再见！")
                    break
                
                if user_input.lower() == 'reset':
                    self.reset_conversation()
                    continue
                
                if user_input.lower() == 'context':
                    print("\n📋 当前对话上下文：")
                    print(self.get_context())
                    print()
                    continue
                
                # 处理问题
                print("🤔 思考中...")
                response = self.handle_query(user_input)
                
                print(f"🤖 客服：{response.answer}\n")
                
                # 显示来源（调试用，生产环境可关闭）
                if response.source_documents:
                    sources = set()
                    for doc in response.source_documents:
                        src = doc.metadata.get('source', '未知')
                        sources.add(src)
                    print(f"   📚 参考：{', '.join(sources)}\n")
                
            except KeyboardInterrupt:
                print("\n\n👋 再见！")
                break
            except Exception as e:
                print(f"\n❌ 错误：{e}\n")


# 演示对话场景
def demo_conversation():
    """演示多轮对话场景"""
    
    # 模拟的知识库查询（实际应使用真实向量存储）
    print("🎬 演示：多轮客服对话\n")
    print("=" * 60)
    
    # 这里展示的是典型的对话流程
    demo_flow = [
        ("你好，我想退货", "您好！我可以帮您处理退货。请问您购买的是什么产品？订单号是多少？"),
        ("我买了一台笔记本，订单号是12345", "好的，订单12345。请问退货原因是什么呢？"),
        ("屏幕有坏点", "了解了，屏幕质量问题。您的订单还在7天无理由退货期内，可以办理退货。"),
        ("怎么操作？", "您可以在订单页面点击'申请退货'，选择退货原因，我们会安排快递上门取件。"),
        ("运费谁出？", "因质量问题退货，运费由我们承担。您无需支付任何费用。"),
        ("好的，谢谢", "不客气！还有其他问题吗？"),
    ]
    
    for user_msg, ai_reply in demo_flow:
        print(f"👤 客户：{user_msg}")
        print(f"🤖 客服：{ai_reply}\n")
    
    print("=" * 60)
    print("\n在这个对话中，AI记住了：")
    print("  • 用户想要退货")
    print("  • 订单号是12345")
    print("  • 产品是笔记本")
    print("  • 退货原因是屏幕坏点")
    print("\n这使得后续对话能够自然衔接，无需重复询问。")


if __name__ == "__main__":
    import sys
    
    if len(sys.argv) > 1 and sys.argv[1] == "--demo":
        demo_conversation()
    else:
        # 正常运行模式
        try:
            service = SmartCustomerService()
            service.run_interactive()
        except Exception as e:
            print(f"启动失败：{e}")
            print("\n请确保：")
            print("  1. 已设置 OPENAI_API_KEY 环境变量")
            print("  2. 知识库索引已构建")
            sys.exit(1)

6. LangChain记忆机制原理

理解记忆机制的工作原理，有助于我们更好地使用和定制记忆功能。

6.1 记忆如何注入提示词

LangChain的记忆机制本质上是在每次调用时将历史信息注入到提示词中：

# 简化示意：记忆注入提示词的过程

# 1. 用户输入
user_input = "我叫什么名字？"

# 2. 加载记忆
memory_vars = memory.load_memory_variables({"input": user_input})
# memory_vars = {"history": "Human: 你好\nAI: 你好！\nHuman: 我叫张三\nAI: 你好张三！"}

# 3. 构建完整提示词
prompt_template = """根据以下对话历史，回答用户问题：

对话历史：
{history}

用户问题：{input}

AI回答："""

final_prompt = prompt_template.format(
    history=memory_vars["history"],
    input=user_input
)

print("最终发送给LLM的提示词：")
print(final_prompt)
# 输出：
# 根据以下对话历史，回答用户问题：
# 
# 对话历史：
# Human: 你好
# AI: 你好！
# Human: 我叫张三
# AI: 你好张三！
#
# 用户问题：我叫什么名字？
#
# AI回答：

6.2 记忆裁剪策略

当对话过长时，需要控制记忆长度以适配LLM的上下文窗口限制：

from langchain.memory import ConversationBufferMemory
from langchain.memory.utils import get_buffer_string

# 不同的记忆类型采用不同的裁剪策略

# ========== 策略1：截断（Window）==========
# 只保留最近k轮
window_memory = ConversationBufferWindowMemory(k=3)
# 实现原理：内部维护一个定长队列，超过则丢弃最早的

# ========== 策略2：摘要（Summary）==========
# 用LLM生成摘要
summary_memory = ConversationSummaryMemory(llm=llm, max_token_limit=300)
# 实现原理：当token数超过限制，触发LLM总结对话历史

# ========== 策略3：向量化检索（Vector）==========
# 语义搜索相关历史
vector_memory = VectorStoreRetrieverMemory(retriever=retriever)
# 实现原理：将历史存入向量库，查询时检索最相关的片段

# ========== 自定义记忆裁剪 ==========
class CustomPruningMemory(ConversationBufferMemory):
    """自定义记忆，基于token数裁剪"""
    
    def __init__(self, max_tokens=2000, **kwargs):
        super().__init__(**kwargs)
        self.max_tokens = max_tokens
        self.token_counter = 0  # 简化的token计数
    
    def save_context(self, inputs, outputs):
        # 先保存
        super().save_context(inputs, outputs)
        
        # 然后裁剪
        self._prune_if_needed()
    
    def _prune_if_needed(self):
        """如果超过token限制，裁剪早期对话"""
        buffer = get_buffer_string(self.chat_memory.messages)
        
        # 简化：按字符数估算（实际应该用tiktoken）
        while len(buffer) > self.max_tokens * 4 and len(self.chat_memory.messages) > 2:
            # 移除最早的一轮对话（2条消息）
            self.chat_memory.messages = self.chat_memory.messages[2:]
            buffer = get_buffer_string(self.chat_memory.messages)


# 使用自定义记忆
custom_memory = CustomPruningMemory(max_tokens=1000)
conversation = ConversationChain(llm=llm, memory=custom_memory)

7. 完整实战：AI面试官系统

本节将构建一个完整的"AI面试官"系统，它具备：性格设定、对话记忆、知识库检索（简历）等功能。

7.1 项目设计

ai_interviewer/
├── data/
│   └── resumes/           # 候选人简历目录
├── db/
│   └── resume_vectors/    # 简历向量数据库
├── interviewer.py         # 面试官核心类
├── main.py               # 主程序入口
└── utils.py              # 工具函数

7.2 核心代码 interviewer.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AI面试官系统
具备性格设定、记忆功能、简历检索能力
"""

import os
from typing import List, Optional, Dict
from dataclasses import dataclass, field
from datetime import datetime

from langchain.chains import ConversationalRetrievalQA
from langchain.memory import ConversationSummaryMemory
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_chroma import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.document_loaders import TextLoader, PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter


@dataclass
class InterviewSession:
    """面试会话数据"""
    candidate_name: str
    position: str
    start_time: datetime
    questions_asked: List[str] = field(default_factory=list)
    answers_recorded: List[str] = field(default_factory=list)
    evaluations: List[str] = field(default_factory=list)


class AIInterviewer:
    """
    AI面试官
    
    特点：
    - 专业但友好的面试官性格
    - 记住候选人的每轮回答
    - 能根据简历提出针对性问题
    - 面试结束生成评估报告
    """
    
    # 面试官性格设定
    INTERVIEWER_PERSONA = """你是一位经验丰富、专业严谨的HR面试官，同时也很友善。

你的面试风格：
1. 专业但不严肃，让候选人感到舒适
2. 善于追问，深入了解候选人的能力和经历
3. 根据简历针对性提问，不问无关问题
4. 认真倾听，记住候选人之前的回答
5. 适时给予反馈和鼓励
6. 面试结束时给出中肯的评价和建议

当前面试职位：{position}
候选人姓名：{candidate_name}

请根据候选人的简历和之前的对话，提出恰当的面试问题。
如果候选人回答不够详细，请礼貌地追问。
如果某个话题已经聊透，自然地切换到下一个话题。"""
    
    def __init__(
        self,
        api_key: str,
        resume_db_path: str = "./resume_vectors",
        position: str = "软件工程师",
        interview_language: str = "中文"
    ):
        """
        初始化AI面试官
        
        Args:
            api_key: OpenAI API密钥
            resume_db_path: 简历向量数据库路径
            position: 面试职位
            interview_language: 面试语言
        """
        self.api_key = api_key
        self.position = position
        self.language = interview_language
        
        # 初始化LLM
        self.llm = ChatOpenAI(
            model="gpt-3.5-turbo",
            temperature=0.7,  # 适当创造性，使对话更自然
            api_key=api_key
        )
        
        # 初始化嵌入模型
        self.embeddings = OpenAIEmbeddings(api_key=api_key)
        
        # 加载简历向量库
        self.resume_store = self._load_resume_store(resume_db_path)
        
        # 面试会话状态
        self.session: Optional[InterviewSession] = None
        self.qa_chain = None
        
        print("✅ AI面试官已就绪")
    
    def _load_resume_store(self, db_path: str) -> Optional[Chroma]:
        """加载简历向量存储"""
        if not os.path.exists(db_path):
            print(f"⚠️  简历库不存在：{db_path}")
            return None
        
        return Chroma(
            persist_directory=db_path,
            embedding_function=self.embeddings
        )
    
    def start_interview(self, candidate_name: str) -> str:
        """
        开始一场面试
        
        Args:
            candidate_name: 候选人姓名
            
        Returns:
            开场白
        """
        # 初始化会话
        self.session = InterviewSession(
            candidate_name=candidate_name,
            position=self.position,
            start_time=datetime.now()
        )
        
        # 创建记忆（使用摘要记忆，适合长对话）
        memory = ConversationSummaryMemory(
            llm=self.llm,
            max_token_limit=400,
            memory_key="chat_history",
            return_messages=True,
            output_key="answer"
        )
        
        # 创建RAG链（如果有简历库）
        if self.resume_store:
            retriever = self.resume_store.as_retriever(
                search_kwargs={
                    "k": 3,
                    "filter": {"candidate": candidate_name}  # 只检索该候选人的简历
                }
            )
            
            # 自定义提示词，融入性格设定
            custom_prompt = ChatPromptTemplate.from_messages([
                ("system", self.INTERVIEWER_PERSONA.format(
                    position=self.position,
                    candidate_name=candidate_name
                )),
                MessagesPlaceholder(variable_name="chat_history"),
                ("human", """基于候选人简历：
{context}

请提出下一个面试问题，或对候选人的回答进行回应。
候选人说：{question}""")
            ])
            
            self.qa_chain = ConversationalRetrievalQA.from_llm(
                llm=self.llm,
                retriever=retriever,
                memory=memory,
                combine_docs_chain_kwargs={"prompt": custom_prompt},
                return_source_documents=False,
                verbose=False
            )
        else:
            # 无简历库，使用普通对话链
            from langchain.chains import ConversationChain
            from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", self.INTERVIEWER_PERSONA.format(
                    position=self.position,
                    candidate_name=candidate_name
                )),
                MessagesPlaceholder(variable_name="history"),
                ("human", "{input}")
            ])
            
            self.qa_chain = ConversationChain(
                llm=self.llm,
                memory=memory,
                prompt=prompt
            )
        
        # 生成开场白
        opening = self._generate_opening(candidate_name)
        return opening
    
    def _generate_opening(self, candidate_name: str) -> str:
        """生成面试开场白"""
        if self.resume_store:
            # 检索简历关键信息
            resume_summary = self._get_resume_summary(candidate_name)
            
            opening_prompt = f"""你是面试官，请生成一段开场白。

候选人：{candidate_name}
应聘职位：{self.position}
简历摘要：{resume_summary}

要求：
1. 自我介绍，表示欢迎
2. 简要提及对候选人简历的初步印象（表示你已经看过简历）
3. 简要说明面试流程
4. 提出第一个面试问题

用自然、专业的语气："""
        else:
            opening_prompt = f"""你是面试官，请生成面试开场白。

候选人：{candidate_name}
应聘职位：{self.position}

要求：欢迎候选人，自我介绍，提出第一个问题。"""
        
        opening = self.llm.predict(opening_prompt)
        return opening
    
    def _get_resume_summary(self, candidate_name: str) -> str:
        """获取简历摘要"""
        if not self.resume_store:
            return "无简历信息"
        
        # 检索候选人简历的关键信息
        results = self.resume_store.similarity_search(
            f"{candidate_name} 的教育背景 工作经验 技能",
            k=5,
            filter={"candidate": candidate_name}
        )
        
        if results:
            summary = " | ".join([doc.page_content[:100] for doc in results[:3]])
            return summary
        return "简历信息有限"
    
    def respond(self, candidate_answer: str) -> str:
        """
        处理候选人回答，返回回应
        
        Args:
            candidate_answer: 候选人的回答
            
        Returns:
            面试官的回应/下一个问题
        """
        if not self.session:
            raise RuntimeError("面试尚未开始，请先调用 start_interview()")
        
        # 记录回答
        self.session.answers_recorded.append(candidate_answer)
        
        # 获取回应
        if hasattr(self.qa_chain, 'invoke'):
            result = self.qa_chain.invoke({"question": candidate_answer})
            response = result["answer"]
        else:
            response = self.qa_chain.predict(input=candidate_answer)
        
        # 记录问题（从回答中提取或标记）
        self.session.questions_asked.append(response)
        
        return response
    
    def generate_evaluation(self) -> str:
        """
        生成面试评估报告
        
        Returns:
            评估报告文本
        """
        if not self.session:
            return "没有面试记录"
        
        eval_prompt = f"""作为面试官，请基于以下面试记录生成评估报告。

候选人：{self.session.candidate_name}
应聘职位：{self.session.position}
面试时间：{self.session.start_time.strftime('%Y-%m-%d %H:%M')}

面试问答记录：
"""
        for i, (q, a) in enumerate(zip(
            self.session.questions_asked,
            self.session.answers_recorded
        ), 1):
            eval_prompt += f"\n第{i}轮：\n问：{q[:100]}...\n答：{a[:200]}...\n"
        
        eval_prompt += """
请生成评估报告，包含：
1. 总体印象（技术能力、沟通能力等）
2. 优点和亮点
3. 需要提升的地方
4. 是否建议进入下一轮
5. 综合评分（1-10分）
"""
        
        evaluation = self.llm.predict(eval_prompt)
        self.session.evaluations.append(evaluation)
        
        return evaluation
    
    def get_session_summary(self) -> Dict:
        """获取会话摘要"""
        if not self.session:
            return {}
        
        return {
            "candidate": self.session.candidate_name,
            "position": self.session.position,
            "duration": str(datetime.now() - self.session.start_time),
            "questions_count": len(self.session.questions_asked),
            "has_evaluation": len(self.session.evaluations) > 0
        }


# ========== 简历索引构建工具 ==========
def build_resume_index(
    resumes_dir: str,
    db_path: str,
    api_key: str
):
    """
    从简历目录构建向量索引
    
    Args:
        resumes_dir: 简历文件目录
        db_path: 向量数据库保存路径
        api_key: OpenAI API密钥
    """
    print(f"📄 正在索引简历：{resumes_dir}")
    
    # 加载简历文件
    loaders = []
    for ext in ["*.txt", "*.pdf"]:
        try:
            loader = DirectoryLoader(
                resumes_dir,
                glob=ext,
                loader_cls=TextLoader if ext == "*.txt" else PyPDFLoader
            )
            loaders.append(loader)
        except:
            pass
    
    documents = []
    for loader in loaders:
        try:
            docs = loader.load()
            # 添加候选人标识到元数据
            for doc in docs:
                # 从文件名提取候选人姓名（简化处理）
                filename = os.path.basename(doc.metadata.get('source', ''))
                candidate_name = filename.split('.')[0]
                doc.metadata['candidate'] = candidate_name
            documents.extend(docs)
        except Exception as e:
            print(f"  加载出错：{e}")
    
    if not documents:
        print("❌ 没有找到简历文件")
        return
    
    print(f"✅ 加载了 {len(documents)} 个简历文件")
    
    # 分割文档
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50
    )
    chunks = splitter.split_documents(documents)
    
    # 创建向量存储
    embeddings = OpenAIEmbeddings(api_key=api_key)
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=db_path
    )
    
    print(f"✅ 简历索引构建完成：{db_path}")
    print(f"   文档块数：{vectorstore._collection.count()}")


# ========== 主程序 ==========
def main():
    """运行AI面试官"""
    import sys
    
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        print("请设置 OPENAI_API_KEY 环境变量")
        sys.exit(1)
    
    # 检查是否需要构建索引
    if len(sys.argv) > 1 and sys.argv[1] == "--build-index":
        build_resume_index(
            resumes_dir="./resumes",
            db_path="./resume_vectors",
            api_key=api_key
        )
        return
    
    # 初始化面试官
    interviewer = AIInterviewer(
        api_key=api_key,
        position="Python开发工程师",
        resume_db_path="./resume_vectors"
    )
    
    # 开始面试
    candidate_name = input("请输入候选人姓名：").strip()
    print("\n" + "=" * 60)
    
    opening = interviewer.start_interview(candidate_name)
    print(f"🤖 面试官：{opening}\n")
    
    # 面试对话循环
    while True:
        candidate_input = input("👤 候选人：").strip()
        
        if candidate_input.lower() in ['结束', 'end', 'quit']:
            print("\n" + "=" * 60)
            print("📝 正在生成评估报告...\n")
            evaluation = interviewer.generate_evaluation()
            print(evaluation)
            break
        
        if not candidate_input:
            continue
        
        # 获取面试官回应
        print("🤔 面试官思考中...")
        response = interviewer.respond(candidate_input)
        print(f"🤖 面试官：{response}\n")
    
    # 打印会话摘要
    print("\n" + "=" * 60)
    print("📊 面试摘要：")
    summary = interviewer.get_session_summary()
    for key, value in summary.items():
        print(f"  {key}: {value}")


if __name__ == "__main__":
    main()

7.3 使用示例

# 1. 准备简历文件
mkdir -p resumes
echo "张三，5年Python开发经验，精通Django、Flask..." > resumes/张三.txt
echo "李四，3年Java开发经验，熟悉Spring Boot..." > resumes/李四.txt

# 2. 构建简历索引
python interviewer.py --build-index

# 3. 开始面试
python interviewer.py

# 交互示例：
# 请输入候选人姓名：张三
# ============================================================
# 🤖 面试官：你好张三，欢迎参加今天的面试！...（基于简历的开场白）
# 
# 👤 候选人：谢谢，很高兴参加这次面试
# 🤔 面试官思考中...
# 🤖 面试官：好的，我们先从你的简历开始。我看到你有5年Python经验...
#
# 👤 候选人：是的，我主要做Web后端开发...
# ...
# 👤 候选人：结束
# ============================================================
# 📝 正在生成评估报告...
# 
# [评估报告内容]
# ============================================================
# 📊 面试摘要：
#   candidate: 张三
#   position: Python开发工程师
#   duration: 0:15:32
#   questions_count: 8
#   has_evaluation: True

本章小结

本章我们深入学习了LangChain的记忆机制和对话系统：

• 记忆的必要性：让AI记住对话历史，保持上下文连贯
• 四种记忆类型：BufferMemory完整保存、WindowMemory窗口限制、SummaryMemory自动摘要、VectorMemory语义检索
• ConversationChain：带记忆的对话链，支持自定义提示词
• 聊天机器人实战：带性格设定、支持多轮对话的机器人开发
• 记忆与RAG结合：ConversationalRetrievalQA链实现知识库+对话历史
• 记忆机制原理：理解记忆如何注入提示词，以及裁剪策略
• 完整项目：AI面试官系统，综合运用记忆、RAG、角色设定