使用 Ollama、Huggingface、FAISS 和 Google Gemma 3 构建本地 RAG 💻 ✨
由 Mux 主办的 DEV 全球展示挑战赛：展示你的项目！
在上一篇文章中，我们了解了检索增强生成（RAG）。这项技术已成为增强大型语言模型（LLM）能力的强大工具。RAG 使 LLM 能够提供更准确、更相关、更符合上下文的答案，从而缓解“幻觉”和信息过时等问题。
这是三篇系列文章的第二部分。本文将详细介绍如何在本地 Python 环境中创建一个基于 RAG 的聊天应用程序。我们将探讨如何集成几项关键技术来构建一个交互式且信息丰富的工具：
- Reflex：一个现代化的纯 Python 框架，旨在快速构建和部署交互式 Web 应用程序，无需单独的前端专业知识（例如 JavaScript）。其响应式特性简化了状态管理。
- LangChain：一个专为开发基于语言模型的应用程序而设计的综合框架。它提供模块化组件和链，以简化复杂的流程（例如 RAG），从而显著简化管道构建。
- Ollama：一款越来越受欢迎的工具，使用户能够直接在本地计算机上下载、运行和管理各种开源 LLM（如 Google 的 Gemma、Meta 的 Llama 系列、Mistral 模型等），从而提升隐私性和离线功能。
- FAISS（Facebook AI 相似性搜索）：一个高度优化的库，用于在大规模向量数据集上高效执行相似性搜索。在我们的 RAG 环境中，根据用户的查询嵌入快速找到相关的文本段落至关重要。
- Hugging Face Datasets & Transformers：自然语言处理生态系统中的事实标准库。Datasets 提供对海量数据集的便捷访问，而句子转换器（基于 Transformers 构建）则提供生成高质量文本嵌入的便捷方法。
我们的目标是创建一个基于网页的聊天应用程序，用户可以在其中提出问题。该应用程序随后将：
- 将问题转换为数值向量（嵌入）。
- 搜索预先索引的向量存储（使用 FAISS 从数据集构建），以查找具有相似嵌入（即相关上下文）的文本段落。
- 将检索到的上下文连同原始问题一起提供给本地运行的 LLM（通过 Ollama）。
- 将 LLM 生成的答案（现在基于检索到的信息）显示给 Reflex 内置聊天界面中的用户。
初始代码结构和设置
注：您可以在此 GitHub 存储库中找到完整代码。
🔶文件夹/文件布局
我们的 RAG 应用程序布局。
rag_app/ # Root folder for this project
│
├── .env
├── requirements.txt
├── rxconfig.py
│
└── rag_gemma_reflex/
├── __init__.py
├── rag_logic.py
├── state.py
└── rag_gemma_reflex.py
🔶要求
请务必使用 uv 或通过基本的 pip 创建虚拟环境。
reflex
langchain
langchain-community
langchain-huggingface
datasets
faiss-cpu
sentence-transformers
ollama
python-dotenv
langchain-ollama
为了方便安装，请从此处下载 requirements.txt 文件。然后在您的虚拟环境中运行以下任一命令：
pip install -r requirements.txt
## if you're using uv
uv pip install -r requirements.txt
🔶每个图书馆都在发挥各自的作用：
- Reflex 构建交互式前端
- langchain 协调 RAG 流
- 数据集提供知识来源
- 句子转换器和 faiss-cpu 处理检索机制
- 奥拉玛经营着当地的法学硕士项目
- python-dotenv 有助于管理配置
深入代码世界💻
🔶下载本地 AI 模型
Ollama 提供了大量可供下载和入门的 AI 模型。本次演示我们将使用gemma3:4b-it-qat一个 40 亿参数的量化模型，它比非量化模型占用的内存少 3 倍。您可以根据需要更换为更大或更小的模型。
🔶本演示使用的数据集
：我们使用来自 Huggingface 的 neural-bridge/rag-dataset-12000数据集。还有许多其他数据集可供使用，您可以下载并在设置中更改名称。
🔶 Reflex 配置 (rxconfig.py)：
此文件包含 Reflex 应用程序的基本设置。对于本项目而言，它非常精简，主要只是为应用程序命名：
import reflex as rx
# Basic configuration defining the application's name
config = rx.Config(
app_name="rag_app",
)
🔶实现 RAG 核心逻辑
此脚本是我们应用程序的引擎，负责设置和执行整个 RAG 流水线。您可以在这里找到此文件的代码。
🔶关键步骤：
DEFAULT_OLLAMA_MODEL = "gemma3:4b-it-qat"
DATASET_NAME = "neural-bridge/rag-dataset-12000"
DATASET_SUBSET_SIZE = 100
EMBEDDING_MODEL_NAME = "all-MiniLM-L6-v2"
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", DEFAULT_OLLAMA_MODEL)
FAISS_INDEX_PATH = "faiss_index_neural_bridge"
1)配置：定义了数据集名称（neural-bridge/rag-dataset-12000）、嵌入模型（all-MiniLM-L6-v2）、默认 Ollama 模型（gemma3:4b-it-qat）和 FAISS 索引路径的常量。为了提高灵活性，添加了从环境变量（OLLAMA_MODEL）读取 Ollama 模型名称的逻辑。
def load_and_split_data():
"""
Loads the neural-bridge/rag-dataset-12000 dataset and converts
contexts into LangChain Documents.
"""
print(f"Loading dataset '{DATASET_NAME}'...")
try:
if DATASET_SUBSET_SIZE:
print(f"Loading only the first {DATASET_SUBSET_SIZE} entries.")
dataset = load_dataset(DATASET_NAME, split=f"train[:{DATASET_SUBSET_SIZE}]")
else:
print("Loading the full dataset...")
dataset = load_dataset(DATASET_NAME, split="train")
documents = [
Document(
page_content=row["context"],
metadata={"question": row["question"], "answer": row["answer"]},
)
for row in dataset
if row.get("context")
]
print(f"Loaded {len(documents)} documents.")
return documents
except Exception as e:
print(f"Error loading dataset '{DATASET_NAME}': {e}")
print(traceback.format_exc())
return []
def load_and_split_data():
"""
Loads the neural-bridge/rag-dataset-12000 dataset and converts
contexts into LangChain Documents.
"""
print(f"Loading dataset '{DATASET_NAME}'...")
try:
if DATASET_SUBSET_SIZE:
print(f"Loading only the first {DATASET_SUBSET_SIZE} entries.")
dataset = load_dataset(DATASET_NAME, split=f"train[:{DATASET_SUBSET_SIZE}]")
else:
print("Loading the full dataset...")
dataset = load_dataset(DATASET_NAME, split="train")
documents = [
Document(
page_content=row["context"],
metadata={"question": row["question"], "answer": row["answer"]},
)
for row in dataset
if row.get("context")
]
print(f"Loaded {len(documents)} documents.")
return documents
except Exception as e:
print(f"Error loading dataset '{DATASET_NAME}': {e}")
print(traceback.format_exc())
return []def load_and_split_data():
"""
Loads the neural-bridge/rag-dataset-12000 dataset and converts
contexts into LangChain Documents.
"""
print(f"Loading dataset '{DATASET_NAME}'...")
try:
if DATASET_SUBSET_SIZE:
print(f"Loading only the first {DATASET_SUBSET_SIZE} entries.")
dataset = load_dataset(DATASET_NAME, split=f"train[:{DATASET_SUBSET_SIZE}]")
else:
print("Loading the full dataset...")
dataset = load_dataset(DATASET_NAME, split="train")
documents = [
Document(
page_content=row["context"],
metadata={"question": row["question"], "answer": row["answer"]},
)
for row in dataset
if row.get("context")
]
print(f"Loaded {len(documents)} documents.")
return documents
except Exception as e:
print(f"Error loading dataset '{DATASET_NAME}': {e}")
print(traceback.format_exc())
return []
2)数据加载（load_and_split_data）：
- 从 Hugging Face 加载指定的数据集（datasets.load_dataset）。
- 处理加载子集（DATASET_SUBSET_SIZE）以加快测试速度的操作。
- 将每一行的“上下文”转换为 LangChain 文档对象，并将“问题”和“答案”存储在元数据中。（我们最初尝试过 rag-datasets/rag-mini-wikipedia，但由于加载复杂而切换回了原来的方案。）
def get_embeddings_model():
"""Initializes and returns the HuggingFace embedding model."""
print(f"Loading embedding model '{EMBEDDING_MODEL_NAME}'...")
model_kwargs = {"device": "cpu"}
encode_kwargs = {"normalize_embeddings": False}
embeddings = HuggingFaceEmbeddings(
model_name=EMBEDDING_MODEL_NAME,
model_kwargs=model_kwargs,
encode_kwargs=encode_kwargs,
)
print("Embedding model loaded.")
return embeddings
3) Embeddings (get_embeddings_model) : 使用 langchain_huggingface.HuggingFaceEmbeddings 初始化句子转换器模型 (all-MiniLM-L6-v2) 将文本文件转换为数值。
def create_or_load_vector_store(documents, embeddings):
"""Creates a FAISS vector store from documents or loads it if it exists."""
if os.path.exists(FAISS_INDEX_PATH) and os.listdir(FAISS_INDEX_PATH):
print(f"Loading existing FAISS index from '{FAISS_INDEX_PATH}'...")
try:
vector_store = FAISS.load_local(
FAISS_INDEX_PATH,
embeddings,
allow_dangerous_deserialization=True
)
print("FAISS index loaded.")
except Exception as e:
print(f"Error loading FAISS index: {e}")
print("Attempting to rebuild the index...")
vector_store = None
else:
vector_store = None
if vector_store is None:
if not documents:
print("Error: No documents loaded to create FAISS index.")
return None
print("Creating new FAISS index...")
vector_store = FAISS.from_documents(documents, embeddings)
print("FAISS index created.")
print(f"Saving FAISS index to '{FAISS_INDEX_PATH}'...")
try:
vector_store.save_local(FAISS_INDEX_PATH)
print("FAISS index saved.")
except Exception as e:
print(f"Error saving FAISS index: {e}")
return vector_store
4)向量存储（创建或加载向量存储）：
- 使用 FAISS（langchain_community.vectorstores.FAISS）从文档嵌入创建向量索引。
- 至关重要的是，它会检查 FAISS_INDEX_PATH 是否已存在索引，如果已存在则加载该索引，以避免每次运行时都重新处理。如果未找到，则创建并保存一个新索引。
def get_ollama_llm():
"""Initializes and returns the Ollama LLM using the new package."""
global OLLAMA_MODEL
current_ollama_model = os.getenv("OLLAMA_MODEL", DEFAULT_OLLAMA_MODEL)
if OLLAMA_MODEL != current_ollama_model:
print(f"Ollama model changed to '{current_ollama_model}'.")
OLLAMA_MODEL = current_ollama_model
global _rag_chain
_rag_chain = None
print(f"Initializing Ollama LLM with model '{OLLAMA_MODEL}'...")
try:
ollama_client.show(OLLAMA_MODEL)
print(f"Confirmed Ollama model '{OLLAMA_MODEL}' is available locally.")
except ollama_client.ResponseError as e:
if "model not found" in str(e).lower():
print(f"Error: Ollama model '{OLLAMA_MODEL}' not found locally.")
print(f"Please pull it first using: ollama pull {OLLAMA_MODEL}")
return None
else:
print(f"An error occurred while checking the Ollama model: {e}")
return None
except Exception as e:
print(f"An unexpected error occurred while checking Ollama model: {e}")
return None
ollama_base_url = os.getenv("OLLAMA_HOST")
if ollama_base_url:
print(f"Using Ollama host: {ollama_base_url}")
llm = Ollama(model=OLLAMA_MODEL, base_url=ollama_base_url)
else:
print("Using default Ollama host (http://localhost:11434).")
llm = Ollama(model=OLLAMA_MODEL)
print("Ollama LLM initialized.")
return llm
5) LLM 初始化 (get_ollama_llm):
- 连接到本地运行的 Ollama 服务。
- 使用 langchain_ollama.OllamaLLM 类（在从已弃用的 langchain_community 版本更新之后）。
- 指定所需的模型（OLLAMA_MODEL，默认为 gemma3:4b-it-qat）。
- 包含错误处理，用于检查指定的模型是否已在 Ollama 中拉取（ollama pull）。
def get_rag_chain():
"""Returns the initialized RAG chain, setting it up if necessary."""
if _rag_chain is None:
setup_rag_chain()
if _rag_chain is None:
print("Warning: RAG chain is not available.")
return _rag_chain
6)链设置（setup_rag_chain，get_rag_chain）：
- 检索器：从 FAISS 向量存储 (vector_store.as_retriever) 创建一个检索器，以查找与用户查询最相似的前 k 个文档。
- 提示模板：定义一个 ChatPromptTemplate，指示 LLM 如何使用提供的上下文回答问题。我们将模板变量从 {question} 更新为 {input}，以匹配 create_retrieval_chain 的预期。
- 内容文档链：使用 create_stuff_documents_chain 获取检索到的文档和用户输入，将它们格式化为提示，并将它们传递给 LLM。
- 检索链：使用 create_retrieval_chain 将检索器和文档链连接起来，形成最终的 RAG 链。
- get_rag_chain 函数充当延迟初始化器，仅在第一次请求或配置（如 Ollama 模型）更改时设置链。
Reflex 应用（状态和用户界面）
🔶管理 UI 状态 (state.py)
此文件定义了 Reflex State 类，该类保存 UI 数据和处理用户交互的逻辑。您可以在这里找到代码。
import reflex as rx
from . import rag_logic
import traceback
class QA(rx.Base):
"""A question and answer pair."""
question: str
answer: str
is_loading: bool = False
class State(rx.State):
"""Manages the application state for the RAG chat interface."""
question: str = ""
chat_history: list[QA] = []
is_loading: bool = False
async def handle_submit(self):
"""Handles the user submitting a question."""
if not self.question.strip():
return
user_question = self.question
self.chat_history.append(QA(question=user_question, answer="", is_loading=True))
self.question = ""
yield
try:
rag_chain = rag_logic.get_rag_chain()
if rag_chain is None:
raise Exception("RAG chain could not be initialized. Check logs.")
response = await rag_chain.ainvoke({"input": user_question})
answer = response.get("answer", "Sorry, I couldn't find an answer.")
self.chat_history[-1].answer = answer
self.chat_history[-1].is_loading = False
except Exception as e:
print(f"Error processing question: {e}")
print(traceback.format_exc())
self.chat_history[-1].answer = f"An error occurred: {e}. Check the console logs."
self.chat_history[-1].is_loading = False
finally:
if self.chat_history:
self.chat_history[-1].is_loading = False
- 问题：存储输入字段中的当前文本。
- chat_history：一个 QA 对象列表（一个简单的 rx.Base 模型，包含问题、答案和加载状态），用于显示对话。
- is_loading：全局加载状态的布尔值（尽管我们主要使用按消息加载）。
- handle_submit：用户提交表单时触发的异步事件处理程序。
🔶 Reflex UI
我们使用 Reflex 组件来构建可视化聊天界面。您可以参考 Reflex 样式指南，了解更多关于如何自定义 Reflex 组件样式的信息。代码如下：rag_gemma_reflex.py
import reflex as rx
from .state import State, QA
# --- UI Styles ---
colors = {
"background": "#0F0F10",
"text_primary": "#E3E3E3",
"text_secondary": "#BDC1C6",
"input_bg": "#1F1F21",
"input_border": "#3C4043",
"button_bg": "#8AB4F8",
"button_text": "#202124",
"button_hover_bg": "#AECBFA",
"user_bubble_bg": "#3C4043",
"bot_bubble_bg": "#1E1F21",
"bubble_border": "#5F6368",
"loading_text": "#9AA0A6",
"heading_gradient_start": "#8AB4F8",
"heading_gradient_end": "#C3A0F8",
}
base_style = {
"background_color": colors["background"],
"color": colors["text_primary"],
"font_family": "'Roboto', sans-serif",
"font_weight": "200",
"height": "100vh",
"width": "100%",
}
input_style = {
"background_color": colors["input_bg"],
"border": f"1px solid {colors['input_border']}",
"color": colors["text_primary"],
"border_radius": "24px",
"padding": "12px 18px",
"width": "100%",
"font_weight": "400",
"_placeholder": {
"color": colors["text_secondary"],
"font_weight": "300",
},
":focus": {
"border_color": colors["button_bg"],
"box_shadow": f"0 0 0 1px {colors['button_bg']}",
},
}
button_style = {
"background_color": colors["button_bg"],
"color": colors["button_text"],
"border": "none",
"border_radius": "24px",
"padding": "12px 20px",
"cursor": "pointer",
"font_weight": "500",
"font_family": "'Roboto', sans-serif",
"transition": "background-color 0.2s ease",
":hover": {
"background_color": colors["button_hover_bg"],
},
}
chat_box_style = {
"padding": "1em 0",
"flex_grow": 1,
"overflow_y": "auto",
"display": "flex",
"flex_direction": "column-reverse",
"width": "100%",
"&::-webkit-scrollbar": {
"width": "8px",
},
"&::-webkit-scrollbar-track": {
"background": colors["input_bg"],
"border_radius": "4px",
},
"&::-webkit-scrollbar-thumb": {
"background": colors["bubble_border"],
"border_radius": "4px",
},
"&::-webkit-scrollbar-thumb:hover": {
"background": colors["text_secondary"],
},
}
qa_style = {
"margin_bottom": "1em",
"padding": "12px 18px",
"border_radius": "18px",
"word_wrap": "break-word",
"max_width": "85%",
"box_shadow": "0 1px 3px 0 rgba(0, 0, 0, 0.15)",
"line_height": "1.6",
"font_weight": "400",
"code": {
"background_color": "rgba(255, 255, 255, 0.1)",
"padding": "0.2em 0.4em",
"font_size": "85%",
"border_radius": "4px",
"font_family": "monospace",
},
"a": {
"color": colors["button_bg"],
"text_decoration": "underline",
":hover": {
"color": colors["button_hover_bg"],
},
},
"p": {
"margin": "0",
},
}
question_style = {
**qa_style,
"background_color": colors["user_bubble_bg"],
"color": colors["text_primary"],
"align_self": "flex-end",
"border_bottom_right_radius": "4px",
}
answer_style = {
**qa_style,
"background_color": colors["bot_bubble_bg"],
"color": colors["text_primary"],
"align_self": "flex-start",
"border_bottom_left_radius": "4px",
}
loading_style = {
"color": colors["loading_text"],
"font_style": "italic",
"font_weight": "300",
}
# --- UI Components ---
def message_bubble(qa: QA):
"""Displays a single question and its answer."""
return rx.vstack(
rx.box(qa.question, style=question_style),
rx.cond(
qa.is_loading,
rx.box("Thinking...", style={**answer_style, **loading_style}),
rx.markdown(qa.answer, style=answer_style),
),
align_items="stretch",
width="100%",
spacing="1",
)
# --- Main Page ---
def index() -> rx.Component:
"""The main chat interface page."""
heading_style = {
"size": "7",
"margin_bottom": "0.25em",
"font_weight": "400",
"background_image": f"linear-gradient(to right, {colors['heading_gradient_start']}, {colors['heading_gradient_end']})",
"background_clip": "text",
"-webkit-background-clip": "text",
"color": "transparent",
"width": "fit-content",
}
return rx.container(
rx.vstack(
rx.box(
rx.heading("RAG Chat with Gemma", **heading_style),
rx.text(
"Ask a question based on the loaded context.",
color=colors["text_secondary"],
font_weight="300",
),
padding_bottom="0.5em",
width="100%",
text_align="center",
),
rx.box(
rx.foreach(State.chat_history, message_bubble),
style=chat_box_style,
),
rx.form(
rx.hstack(
rx.input(
name="question",
placeholder="Ask your question...",
value=State.question,
on_change=State.set_question,
style=input_style,
flex_grow=1,
height="50px",
),
rx.button(
"Ask",
type="submit",
style=button_style,
is_loading=State.is_loading,
height="50px",
),
width="100%",
align_items="center",
),
on_submit=State.handle_submit,
width="100%",
),
align_items="center",
width="100%",
height="100%",
padding_x="1em",
padding_y="1em",
spacing="4",
),
max_width="900px",
height="100vh",
padding=0,
margin="auto",
)
# --- App Setup ---
stylesheets = [
"https://fonts.googleapis.com/css2?family=Roboto:wght@200;300;400;500&display=swap",
]
app = rx.App(style=base_style, stylesheets=stylesheets)
app.add_page(index, title="Reflex Chat")
构建应用程序
您需要确保项目结构正确。然后在项目根目录下运行以下两条命令。您的应用即可启动并运行。
reflex init
reflx run
⚠️ 注意：请确保在运行此应用程序之前，Ollama 已启动并正常运行。成功的消息将显示如下。部分消息可能在您于聊天界面发送“Hi”后出现。
🔶聊天界面
向我们的 RAG 应用程序发送消息：
然后从数据集中验证该说法。
对于这个问题，我们的 RAG 应用给出了正确的答案和上下文；数据集截图来自 Huggingface。因此，我们可以说 RAG 运行良好。
如何让这款应用更准确/更符合生产环境要求？
为了使该应用程序运行得更好、更准确，可以遵循以下几个步骤。
- 使用 Ollama 的更大型号，例如 Gemma 27B、Llama 3.3 70B、Qwen 2.5 70B 或 DeepSeek V3，以获得更高的精度。
- 使用 Qdrant、Pinecone、Milvus 等专用向量数据库来存储和索引结果。
- 针对特定用例，应创建个性化专用数据集，而不是使用互联网语料库。请记住，RAG 的效果取决于您提供的数据质量。为了获得最佳结果，数据应经过充分清洗，并准备好用于人工智能。
- 视觉改造：您可以按照 reflex 的指南改进聊天界面，添加更多功能、额外状态、用于存储聊天记录的数据库等。
结论
通过这一迭代过程，我们成功构建了一个功能完善的本地部署 RAG 聊天应用程序。该项目展示了将 Reflex 用于快速 UI 开发、LangChain 用于复杂的 LLM 工作流编排、FAISS 用于高效的向量搜索以及 Ollama 用于本地 LLM 推理所提供的隐私和控制等技术相结合的强大功能。