基于LangChain和FAISS构建RAG问答机器人：CSV数据集成指南

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)

# 加载CSV文件，指定编码和分隔符
loader = CSVLoader(file_path="./data/cleanTripLisbon.csv", encoding="utf-8", csv_args={'delimiter': ','})
data = loader.load()

# 文本分割器，这里我们假设每行数据已经足够独立，或者根据实际需求进行更细致的分割
# 如果CSV每行代表一个完整的文档，可以简化分割策略
text_splitter = CharacterTextSplitter(separator='\n', chunk_size=1000, chunk_overlap=0)
text_chunks = text_splitter.split_documents(data)

print(f"Loaded {len(data)} documents from CSV, split into {len(text_chunks)} chunks.")

# 初始化HuggingFace嵌入模型
embeddings = HuggingFaceEmbeddings(model_name='sentence-transformers/all-MiniLM-L6-v2')

# 从文本块和嵌入模型创建FAISS向量数据库
docsearch = FAISS.from_documents(text_chunks, embeddings)

# 将FAISS索引保存到本地，以便后续加载和使用
docsearch.save_local(DB_FAISS_PATH)
print(f"FAISS vector store saved to {DB_FAISS_PATH}")

from openai import OpenAI
import os # 假设local_settings.py或环境变量中存储API密钥

# 为了教程的独立性，这里模拟local_settings.py
class LocalSettings:
    def __init__(self):
        self.OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") # 从环境变量获取
        if not self.OPENAI_API_KEY:
            raise ValueError("OPENAI_API_KEY environment variable not set.")

local_settings = LocalSettings()

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, temperature=0):
        # 将用户提示添加到消息历史
        self.messages.append({"role": "user", "content": prompt})

        # 调用OpenAI API获取完成
        completion = self.client.chat.completions.create(
            model=self.model,
            messages=self.messages,
            temperature=temperature,
        )

        # 将助手的回复添加到消息历史
        assistant_response = completion.choices[0].message.content
        self.messages.append(
            {
                "role": "assistant",
                "content": assistant_response
            }
        )
        return assistant_response

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

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

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

    def generate_response(self, message: str):
        user_message = f"{self._username}: {message}" if self._username else message
        response = self.engine.get_completion(user_message)
        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
        # 更新GPT_Helper的系统行为，并重置对话
        self.engine.messages = [{"role": "system", "content": system_config}]

# 修改AttractionBot类的__init__方法
class AttractionBot:
    def __init__(self, system_behavior: str, docsearch): # 接受docsearch对象
        self._system_behavior = system_behavior
        self._username = None

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

        # 将docsearch转换为检索器
        self.doc_retriever = docsearch.as_retriever() # 新增行

# 修改AttractionBot类的generate_response方法
class AttractionBot:
    # ... (init方法如上所示) ...

    def generate_response(self, message: str):
        user_message = f"{self._username}: {message}" if self._username else message

        # 使用检索器获取相关信息
        # retrieve方法返回的是Document对象列表
        relevant_docs = self.doc_retriever.invoke(user_message) 

        # 将检索到的文档内容提取出来，并格式化为LLM可理解的上下文
        context = "\n".join([doc.page_content for doc in relevant_docs])

        # 构建增强后的提示，将上下文和用户问题结合
        # 这是一个关键步骤，确保LLM能够利用检索到的信息
        augmented_prompt = f"""根据以下上下文信息，回答用户的问题。
        如果上下文没有足够的信息，请说明你无法回答。

        上下文:
        {context}

        用户问题:
        {user_message}
        """

        # 使用增强后的提示生成响应
        response = self.engine.get_completion(augmented_prompt)

        return response

    # ... (其他方法) ...

# 确保在运行此代码之前，已经执行了“数据准备与向量数据库构建”部分的代码
# 并且DB_FAISS_PATH路径下有保存的FAISS索引

# 从本地加载FAISS向量数据库
# 需要重新初始化embeddings，因为docsearch.load_local需要它
embeddings = HuggingFaceEmbeddings(model_name='sentence-transformers/all-MiniLM-L6-v2')
loaded_docsearch = FAISS.load_local(DB_FAISS_PATH, embeddings, allow_dangerous_deserialization=True) # 注意：如果FAISS版本较新，可能需要allow_dangerous_deserialization=True

# 定义系统行为
system_behavior_config = "你是一个关于葡萄牙里斯本景点的智能助手，请根据提供的信息回答问题。"

# 实例化AttractionBot，传入系统行为和加载的docsearch对象
attraction_bot = AttractionBot(system_behavior=system_behavior_config, docsearch=loaded_docsearch)

# 设置用户名（可选）
attraction_bot.set_username("用户A")

# 模拟用户交互
print("Bot: 你好！我是里斯本景点助手，有什么可以帮你的吗？")

while True:
    user_query = input("你: ")
    if user_query.lower() == '退出':
        print("Bot: 再见！")
        break

    response = attraction_bot.generate_response(user_query)
    print(f"Bot: {response}")

    # 可以查看当前的对话历史
    # print("\n--- 对话历史 ---")
    # for msg in attraction_bot.memory:
    #     print(f"{msg['role']}: {msg['content']}")
    # print("-----------------\n")