本教程详细介绍了如何利用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交互并管理对话历史。
核心组件：
- 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}]关键修改点：
- AttractionBot.__init__: 现在接受 docsearch 对象作为参数。`self.
