LangChain对话检索链中聊天历史与内存的深度解析

花韻仙語

发布时间：2025-10-16 10:22:01

303人浏览过

来源于php中文网

原创

LangChain对话检索链中聊天历史与内存的深度解析

本文深入探讨了langchain `conversationalretrievalchain`在集成提示模板和内存时，为何仍需显式提供`chat_history`输入的问题。通过详细解析`conversationbuffermemory`、`faiss`检索器、自定义提示模板以及链的初始化参数，提供了一套完整的解决方案和代码示例，帮助开发者构建功能完善的对话式检索系统，并避免常见的`valueerror: missing some input keys: {'chat_history'}`错误。

理解 LangChain ConversationalRetrievalChain 的聊天历史管理

在构建基于LangChain的对话式检索系统时，ConversationalRetrievalChain是一个核心组件，它结合了对话记忆和文档检索功能，使得语言模型能够根据历史对话和外部知识库进行智能回复。然而，开发者在使用自定义提示模板并配置了内存（Memory）时，常会遇到ValueError: Missing some input keys: {'chat_history'}的错误。这引发了一个常见疑问：既然已经配置了内存，为何还需要在调用链时显式传入chat_history？

这个问题的核心在于ConversationalRetrievalChain的内部机制以及它如何与提示模板、内存和输入参数协同工作。简而言之，尽管ConversationBufferMemory负责维护和管理对话历史，但如果您的提示模板（promptTemplate）明确引用了{chat_history}变量，那么ConversationalRetrievalChain在执行时，会期望在其输入字典中找到一个名为chat_history的键。这是因为链的输入接口需要满足提示模板的所有占位符要求。get_chat_history参数则用于定义如何从内部内存中提取并格式化历史记录以供提示模板使用。

核心组件解析与配置

为了构建一个功能完善的对话检索链并解决上述问题，我们需要正确配置以下几个关键组件：

1. 记忆模块 (Memory)

ConversationBufferMemory是LangChain中常用的对话记忆类型，它以列表形式存储消息。关键在于设置memory_key，它应该与您的提示模板中引用聊天历史的变量名一致。

from langchain.memory import ConversationBufferMemory

# 初始化对话记忆，memory_key应与提示模板中的变量名一致
memory = ConversationBufferMemory(
    memory_key='chat_history',  # 必须与提示模板中的 {chat_history} 匹配
    return_messages=True,       # 返回消息对象列表
    output_key='answer'         # 如果需要，指定链的输出键
)

2. 检索器 (Retriever)

ConversationalRetrievalChain需要一个检索器来从您的知识库中获取相关文档。通常，这涉及先构建一个向量数据库索引，例如FAISS。

from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import VertexAIEmbeddings # 或其他嵌入模型
import os

# 假设embeddings已被初始化
# embeddings = VertexAIEmbeddings(...)

# 加载预先构建的FAISS索引
# 确保在运行对话链之前，FAISS索引已经通过以下方式构建并保存：
# store = FAISS.load_local("faiss_index", embeddings)
# retriever = store.as_retriever(
#     search_type="similarity",
#     search_kwargs={"k": 2}
# )

关于如何构建FAISS索引，将在后续章节详细介绍。

3. 提示模板 (Prompt Template)

提示模板定义了LLM接收输入的格式。它通常包含上下文（context）、聊天历史（chat_history）和用户问题（question）。

from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate

promptTemplate = """请根据提供的上下文和聊天历史，回答用户的问题。

上下文:
{context}

聊天历史:
{chat_history}

用户问题:
{question}
"""

messages = [
    SystemMessagePromptTemplate.from_template(promptTemplate),
    HumanMessagePromptTemplate.from_template("{question}")
]
qa_prompt = ChatPromptTemplate.from_messages(messages)

请注意，这里的{chat_history}必须与ConversationBufferMemory中设置的memory_key保持一致。

B12

B12是一个由AI驱动的一体化网站建设平台

下载

4. 链的初始化 (Chain Initialization)

ConversationalRetrievalChain.from_llm是创建对话检索链的关键。以下参数至关重要：

llm: 您使用的语言模型实例。
retriever: 上面定义的检索器。
memory: 上面初始化的内存对象。
get_chat_history: 这是一个函数，用于从内存中提取并格式化聊天历史，以便将其注入到提示模板中。简单的lambda h: h通常足够，它会将内存中的消息列表直接传递给提示模板，由ChatPromptTemplate负责格式化。
combine_docs_chain_kwargs={"prompt": qa_prompt}: 将自定义的提示模板传递给文档组合链。

from langchain.chains import ConversationalRetrievalChain
# 假设code_llm已被初始化
# code_llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0) # 示例LLM

# ... (上述 memory, store, retriever, qa_prompt 的初始化代码) ...

qa_chain = ConversationalRetrievalChain.from_llm(
    code_llm,
    retriever,
    memory=memory,
    get_chat_history=lambda h : h, # 关键：定义如何从内存中获取历史
    combine_docs_chain_kwargs={"prompt": qa_prompt} # 使用自定义提示模板
)

解决 ValueError: Missing some input keys: {'chat_history'}

即使配置了memory和get_chat_history，如果您的提示模板中包含{chat_history}，ConversationalRetrievalChain的__call__方法仍然期望在输入字典中接收一个chat_history键。这个键可以是一个空列表，它作为占位符满足链的输入要求，而实际的对话历史内容则由内部的memory和get_chat_history函数提供。

# 示例：初始化一个空的聊天历史列表，用于传递给链的输入
history = [] # 这个列表在每次调用时会被传递，但实际的历史由memory管理

# 模拟用户提问
question = "什么是LangChain？"

# 调用链
answer_obj = qa_chain({"question": question, "chat_history": history})

# 从结果中提取答案
response_text = answer_obj['answer']
print(f"Human: {question}")
print(f"AI: {response_text}")

# 更新外部历史列表（可选，用于外部显示或进一步处理）
history.append((question, response_text))

# 再次提问，模拟对话
question_2 = "它有什么主要功能？"
answer_obj_2 = qa_chain({"question": question_2, "chat_history": history})
response_text_2 = answer_obj_2['answer']
print(f"Human: {question_2}")
print(f"AI: {response_text_2}")
history.append((question_2, response_text_2))

通过在qa_chain的调用中显式传入"chat_history": history（即使history初始为空），我们就满足了链的输入要求，从而避免了ValueError。链内部会利用memory和get_chat_history来获取并格式化真实的对话历史，以填充提示模板中的{chat_history}。

FAISS 索引构建详解

在运行ConversationalRetrievalChain之前，您需要一个可用的检索器，通常是基于向量数据库的。以下是使用FAISS构建和保存本地索引的示例。

from langchain_community.embeddings import VertexAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter, Language
from langchain_community.vectorstores import FAISS
import os

# 1. 初始化嵌入模型
# 确保您已配置Vertex AI认证，例如通过gcloud auth application-default login
EMBEDDING_QPM = 100
EMBEDDING_NUM_BATCH = 5
embeddings = VertexAIEmbeddings(
    requests_per_minute=EMBEDDING_QPM,
    num_instances_per_batch=EMBEDDING_NUM_BATCH,
    model_name="textembedding-gecko",
    max_output_tokens=512,
    temperature=0.1,
    top_p=0.8,
    top_k=40
)

# 2. 初始化文本分割器
# 根据您的文档类型选择合适的分割器和参数
text_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON, # 示例：如果您的训练数据是Python代码或类似结构
    chunk_size=2000,
    chunk_overlap=500
)

# 3. 加载并分割训练数据
docs = []
training_data_dir = "training/facts/" # 假设您的训练数据文件在此目录下
if not os.path.exists(training_data_dir):
    os.makedirs(training_data_dir)
    # 创建一些示例文件以便代码运行
    with open(os.path.join(training_data_dir, "fact1.txt"), "w") as f:
        f.write("LangChain是一个用于开发由大型语言模型（LLM）驱动的应用程序的框架。")
    with open(os.path.join(training_data_dir, "fact2.txt"), "w") as f:
        f.write("LangChain的主要功能包括：链（Chains）、代理（Agents）、内存（Memory）、文档加载器（Document Loaders）和向量存储（Vector Stores）。")

trainingData = os.listdir(training_data_dir)

for training_file in trainingData:
    file_path = os.path.join(training_data_dir, training_file)
    with open(file_path, 'r', encoding='utf-8') as f:
        print(f"Add {f.name} to dataset")
        texts = text_splitter.create_documents([f.read()])
        docs.extend(texts)

# 4. 从文档创建FAISS索引并保存
if docs:
    store = FAISS.from_documents(docs, embeddings)
    store.save_local("faiss_index")
    print("FAISS index created and saved to 'faiss_index' directory.")
else:
    print("No documents found to create FAISS index.")

# 5. 加载已保存的FAISS索引以供检索
# store = FAISS.load_local("faiss_index", embeddings)
# retriever = store.as_retriever(search_type="similarity", search_kwargs={"k": 2})

这段代码首先初始化嵌入模型和文本分割器，然后读取指定目录下的训练数据文件，将其分割成小块（chunks），最后使用这些文本块和嵌入模型构建FAISS索引并保存到本地。

完整示例代码

将上述所有组件整合，形成一个完整的LangChain对话检索链示例：

import os
from langchain_community.embeddings import VertexAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter, Language
from langchain_community.vectorstores import FAISS
from langchain.memory import ConversationBufferMemory
from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
from langchain.chains import ConversationalRetrievalChain
from langchain_openai import ChatOpenAI # 示例LLM，您可以使用其他LLM

# --- 1. 初始化嵌入模型和LLM ---
# 确保您已配置Vertex AI认证或OpenAI API密钥
EMBEDDING_QPM = 100
EMBEDDING_NUM_BATCH = 5
embeddings = VertexAIEmbeddings(
    requests_per_minute=EMBEDDING_QPM,
    num_instances_per_batch=EMBEDDING_NUM_BATCH,
    model_name="textembedding-gecko",
    max_output_tokens=512,
    temperature=0.1,
    top_p=0.8,
    top_k=40
)

# 示例LLM，请替换为您的实际LLM配置
# code_llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)
# 或者使用 Vertex AI LLM
from langchain_google_vertexai import ChatVertexAI
code_llm = ChatVertexAI(model_name="gemini-pro", temperature=0.1)


# --- 2. 构建或加载FAISS索引 ---
FAISS_INDEX_DIR = "faiss_index"
if not os.path.exists(FAISS_INDEX_DIR):
    print("FAISS index not found. Building new index...")
    # 创建示例训练数据目录和文件
    training_data_dir = "training/facts/"
    if not os.path.exists(training_data_dir):
        os.makedirs(training_data_dir)
        with open(os.path.join(training_data_dir, "fact1.txt"), "w", encoding='utf-8') as f:
            f.write("LangChain是一个用于开发由大型语言模型（LLM）驱动的应用程序的框架。")
        with open(os.path.join(training_data_dir, "fact2.txt"), "w", encoding='utf-8') as f:
            f.write("LangChain的主要功能包括：链（Chains）、代理（Agents）、内存（Memory）、文档加载器（Document Loaders）和向量存储（Vector Stores）。")
        with open(os.path.join(training_data_dir, "fact3.txt"), "w", encoding='utf-8') as f:
            f.write("FAISS是Facebook AI Research开发的一个用于高效相似性搜索和密集向量聚类的库。")
        with open(os.path.join(training_data_dir, "fact4.txt"), "w", encoding='utf-8') as f:
            f.write("ConversationalRetrievalChain结合了检索增强生成（RAG）和对话记忆，以支持多轮对话。")

    text_splitter = RecursiveCharacterTextSplitter.from_language(
        language=Language.PYTHON, chunk_size=1000, chunk_overlap=200
    )

    docs = []
    for training_file in os.listdir(training_data_dir):
        file_path = os.path.join(training_data_dir, training_file)
        with open(file_path, 'r', encoding='utf-8') as f:
            print(f"Adding {f.name} to dataset")
            texts = text_splitter.create_documents([f.read()])
            docs.extend(texts)

    if docs:
        store = FAISS.from_documents(docs, embeddings)
        store.save_local(FAISS_INDEX_DIR)
        print("FAISS index built and saved.")
    else:
        raise ValueError("No documents found to build FAISS index.")
else:
    print("Loading existing FAISS index...")

store = FAISS.load_local(FAISS_INDEX_DIR, embeddings, allow_dangerous_deserialization=True) # allow_dangerous_deserialization=True for newer FAISS versions
retriever = store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 2}
)

# --- 3. 初始化对话记忆 ---
memory = ConversationBufferMemory(
    memory_key='chat_history',

Python日期时间进阶教程_时区转换与日期运算实践

如何在 Python 中高效生成并筛选随机键值对列表

Flask 中使用 url_for 生成 URL 时的常见错误及正确用法

Python配置文件合并_优先级说明【指导】

Python函数接口设计原则_可维护性解析【教程】