基于LangChain和FAISS的CSV数据检索增强型问答机器人构建指南-Python教程-PHP中文网

基于LangChain和FAISS的CSV数据检索增强型问答机器人构建指南

本教程详细介绍了如何利用langchain框架，结合faiss向量数据库和openai大型语言模型，构建一个能够基于csv文件内容进行智能问答的聊天机器人。文章涵盖了从csv数据向量化、faiss索引创建，到核心的检索增强生成（rag）机制集成，以及如何将检索到的相关信息有效融入语言模型提示词，从而实现精准、上下文感知的问答体验。

在构建能够基于特定文档（如CSV文件）进行问答的智能机器人时，检索增强生成（Retrieval Augmented Generation, RAG）是一种强大且高效的范式。它允许大型语言模型（LLM）在生成答案之前，先从一个外部知识库中检索相关信息，从而克服LLM知识时效性、减少幻觉并提供更精准的答案。本教程将引导您使用LangChain、FAISS和OpenAI API，逐步实现一个基于CSV数据的RAG问答机器人。

1. 构建知识库：CSV数据向量化与FAISS索引

首先，我们需要将CSV文件中的结构化数据转换为LLM可理解的向量表示，并构建一个高效的检索索引。LangChain提供了丰富的工具来简化这一过程。

步骤概述：

加载CSV数据： 使用CSVLoader加载本地CSV文件，将其转换为LangChain的Document对象列表。
文本切分： 将每个文档（在此例中为CSV的每一行）进行适当的切分，以便后续嵌入。
生成嵌入： 使用预训练的嵌入模型将文本块转换为高维向量。
创建向量存储： 利用FAISS库构建一个高效的向量索引，用于快速检索。
持久化索引： 将FAISS索引保存到本地磁盘，以便后续复用。

以下是实现这一过程的代码示例（通常位于 embeddings.py 文件中）：

from langchain.vectorstores import FAISS
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.document_loaders.csv_loader import CSVLoader
import os

# 定义FAISS索引的保存路径
DB_FAISS_PATH = "vectorstore/db_faiss"
# 确保目录存在
os.makedirs(DB_FAISS_PATH, exist_ok=True)

# 1. 加载CSV数据
# 假设您的CSV文件位于 ./data/cleanTripLisbon.csv
loader = CSVLoader(file_path="./data/cleanTripLisbon.csv", encoding="utf-8", csv_args={'delimiter': ','})
data = loader.load()

# 2. 文本切分
# 对于CSV数据，如果每行代表一个独立实体，可以考虑以行作为基本单位。
# CharacterTextSplitter 默认按字符切分，这里设置分隔符为换行符。
# 注意：对于复杂CSV，可能需要更精细的切分策略。
text_splitter = CharacterTextSplitter(separator='\n', chunk_size=1000, chunk_overlap=0)
text_chunks = text_splitter.split_documents(data)

# 3. 生成嵌入
# 使用HuggingFace提供的sentence-transformers模型生成文本嵌入
embeddings = HuggingFaceEmbeddings(model_name='sentence-transformers/all-MiniLM-L6-v2')

# 4. 创建FAISS向量存储
# 从切分后的文档和嵌入模型创建FAISS索引
docsearch = FAISS.from_documents(text_chunks, embeddings)

# 5. 持久化索引
docsearch.save_local(DB_FAISS_PATH)
print(f"FAISS 索引已成功创建并保存到 {DB_FAISS_PATH}")

# 也可以加载已保存的索引
# docsearch = FAISS.load_local(DB_FAISS_PATH, embeddings)

登录后复制

代码说明：

CSVLoader: 专门用于加载CSV文件，并将其每行转换为一个Document对象。
CharacterTextSplitter: 负责将加载的文档切分成更小的块（chunk），这对于处理长文档至关重要。separator='\n'在这里意味着它会尝试在换行符处分割，这对于按行处理CSV数据是合理的。
HuggingFaceEmbeddings: 利用HuggingFace模型库中的预训练模型（例如all-MiniLM-L6-v2）生成文本的向量嵌入。
FAISS.from_documents(): 使用生成的文本块及其嵌入来构建一个FAISS索引，FAISS是一个用于高效相似性搜索的库。
docsearch.save_local(): 将构建好的FAISS索引保存到指定路径，避免每次启动应用时都重新生成。

2. 设计语言模型助手：基础聊天机器人实现

接下来，我们构建一个基础的聊天机器人框架，它能与OpenAI API交互并管理对话历史。

核心组件：

LobeHub

LobeChat brings you the best user experience of ChatGPT, OLLaMA, Gemini, Claude

201

查看详情

GPT_Helper 类： 封装了与OpenAI chat.completions API的交互逻辑，包括API密钥管理、消息历史维护和模型调用。
AttractionBot 类： 机器人的主要逻辑类，负责处理用户输入、生成响应，并管理系统行为。

以下是实现这些组件的代码示例（通常位于 chat_bot.py 文件中）：

from openai import OpenAI
# 假设 util.local_settings 包含 OPENAI_API_KEY
# from util import local_settings 

class GPT_Helper:
    def __init__(self, OPENAI_API_KEY: str, system_behavior: str = "", model="gpt-3.5-turbo"):
        self.client = OpenAI(api_key=OPENAI_API_KEY)
        self.messages = []
        self.model = model

        if system_behavior:
            self.messages.append({
                "role": "system",
                "content": system_behavior
            })

    def get_completion(self, prompt: str, temperature=0):
        """
        向OpenAI模型发送请求并获取响应。
        这里的prompt应该是一个完整的用户请求，可能已包含检索到的上下文。
        """
        # 为了RAG，我们将把检索到的上下文和用户问题组合成一个单一的prompt
        # 此时，我们不直接将prompt添加到messages作为"user"角色，
        # 而是创建一个新的消息列表，以更好地控制上下文。
        # 或者，如果RAG上下文是动态的，可以在调用时构造完整的消息列表。

        # 简单起见，这里假设传入的prompt已经包含了所有必要信息
        # 如果需要维护对话历史，这里可以做更复杂的处理

        current_messages = self.messages + [{"role": "user", "content": prompt}]

        completion = self.client.chat.completions.create(
            model=self.model,
            messages=current_messages, # 使用包含当前用户prompt的消息列表
            temperature=temperature,
        )

        response_content = completion.choices[0].message.content

        # 将用户的原始消息和模型的响应添加到历史中（如果需要维护对话历史）
        # 注意：如果每次调用都重新构建prompt，这里的历史维护需要更精细的设计
        # self.messages.append({"role": "user", "content": prompt}) # 这取决于prompt是否是原始用户消息
        # self.messages.append({"role": "assistant", "content": response_content})

        return response_content

class AttractionBot:
    def __init__(self, system_behavior: str, openai_api_key: str): # 移除 docsearch 参数，稍后添加
        self._system_behavior = system_behavior
        self._username = None

        self.engine = GPT_Helper(
            OPENAI_API_KEY=openai_api_key, # 传入API密钥
            system_behavior=system_behavior
        )

    def set_username(self, username):
        self._username = username

    def generate_response(self, message: str):
        # 这是一个基础的生成响应方法，不包含检索逻辑
        user_message_with_name = f"{self._username}: {message}" if self._username else message
        response = self.engine.get_completion(user_message_with_name)
        return response

    def reset(self):
        # 重置机器人状态，例如清空对话历史
        self.engine.messages = []
        if self._system_behavior:
            self.engine.messages.append({
                "role": "system",
                "content": self._system_behavior
            })

    @property
    def memory(self):
        return self.engine.messages

    @property
    def system_behavior(self):
        return self._system_behavior

    @system_behavior.setter
    def system_behavior(self, system_config: str):
        self._system_behavior = system_config
        self.engine.messages = [{"role": "system", "content": system_config}] # 更新系统行为并重置历史

登录后复制

代码说明：

GPT_Helper: 这是一个通用类，用于与OpenAI API进行通信。它维护一个messages列表来跟踪对话历史，并提供get_completion方法来发送请求并获取响应。
AttractionBot: 封装了特定于应用（例如景点问答）的逻辑。它内部使用GPT_Helper来与LLM交互。目前，generate_response方法直接将用户消息传递给LLM，尚未集成检索功能。

3. 整合检索增强生成（RAG）：连接知识库与聊天机器人

现在是核心部分：将我们构建的FAISS知识库与AttractionBot结合，实现检索增强生成。

RAG工作原理： 当用户提出问题时，机器人不再直接将问题发送给LLM。相反，它会：

检索： 使用用户的查询作为检索词，从FAISS向量存储中查找最相关的文档片段。
增强： 将检索到的文档内容作为额外上下文，与用户原始问题一起构建一个新的、更丰富的提示词（prompt）。
生成： 将这个增强后的提示词发送给LLM，LLM利用这些上下文生成更准确、更相关的答案。

修改 AttractionBot 以集成 RAG：

我们需要修改 AttractionBot 的 __init__ 方法来接收 docsearch 对象，并将其转换为一个检索器（retriever）。同时，generate_response 方法需要更新以执行检索和提示词增强。

import os
from openai import OpenAI
from langchain.vectorstores import FAISS
from langchain.embeddings import HuggingFaceEmbeddings
# from util import local_settings # 假设 local_settings 包含 OPENAI_API_KEY

# 重新加载FAISS索引和嵌入模型 (在实际应用中，这些应该在主程序入口处加载一次)
DB_FAISS_PATH = "vectorstore/db_faiss"
embeddings = HuggingFaceEmbeddings(model_name='sentence-transformers/all-MiniLM-L6-v2')

# 确保索引文件存在，否则会报错
if os.path.exists(DB_FAISS_PATH):
    docsearch = FAISS.load_local(DB_FAISS_PATH, embeddings, allow_dangerous_deserialization=True)
else:
    raise FileNotFoundError(f"FAISS 索引文件在 {DB_FAISS_PATH} 不存在。请先运行 embeddings.py 创建索引。")

# 假设 local_settings.OPENAI_API_KEY 已定义
# 为了示例运行，我们这里直接定义一个占位符
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY") 
if OPENAI_API_KEY == "YOUR_OPENAI_API_KEY":
    print("警告：请设置 OPENAI_API_KEY 环境变量或在代码中替换为您的实际API密钥。")


class GPT_Helper:
    def __init__(self, OPENAI_API_KEY: str, system_behavior: str = "", model="gpt-3.5-turbo"):
        self.client = OpenAI(api_key=OPENAI_API_KEY)
        self.messages = []
        self.model = model

        if system_behavior:
            self.messages.append({
                "role": "system",
                "content": system_behavior
            })

    def get_completion(self, prompt: str, temperature=0):
        """
        向OpenAI模型发送请求并获取响应。
        这里假设传入的prompt已经包含了检索到的上下文和用户问题。
        """
        # 为了简洁和RAG的灵活性，我们在这里每次都构造一个新的消息列表
        # 如果需要维护长期的对话历史，需要更复杂的逻辑来决定何时清空或截断历史。
        current_messages = self.messages + [{"role": "user", "content": prompt}]

        completion = self.client.chat.completions.create(
            model=self.model,
            messages=current_messages,
            temperature=temperature,
        )
        return completion.choices[0].message.content

class AttractionBot:
    def __init__(self, system_behavior: str, openai_api_key: str, docsearch: FAISS):
        self._system_behavior = system_behavior
        self._username = None

        self.engine = GPT_Helper(
            OPENAI_API_KEY=openai_api_key,
            system_behavior=system_behavior
        )

        # 将 docsearch 转换为一个检索器
        self.doc_retriever = docsearch.as_retriever()

    def set_username(self, username):
        self._username = username

    def generate_response(self, message: str):
        """
        生成响应，集成RAG流程。
        """
        # 1. 使用检索器获取相关文档
        # LangChain的retriever通常使用 .invoke() 方法来执行检索
        relevant_docs = self.doc_retriever.invoke(message)

        # 2. 从文档中提取内容，构建上下文字符串
        context = "\n\n".join([doc.page_content for doc in relevant_docs])

        # 3. 构建一个增强的提示词，将检索到的上下文和用户问题结合起来
        # 这是一个示例提示词模板，您可以根据需求进行优化
        rag_prompt = f"""
        请根据以下提供的信息回答用户的问题。如果提供的信息不足以回答问题，请告知用户。

        上下文信息：
        {context}

        用户问题：
        {message}
        """

        # 4. 调用语言模型生成响应
        response = self.engine.get_completion(rag_prompt)

        return response

    def reset(self):
        self.engine.messages = []
        if self._system_behavior:
            self.engine.messages.append({
                "role": "system",
                "content": self._system_behavior
            })

    @property
    def memory(self):
        return self.engine.messages

    @property
    def system_behavior(self):
        return self._system_behavior

    @system_behavior.setter
    def system_behavior(self, system_config: str):
        self._system_behavior = system_config
        self.engine.messages = [{"role": "system", "content": system_config}]

登录后复制

关键修改点：