本地离线 RAG 系统完整实现指南

在上一篇文章中，我们了解了检索增强生成（Retrieval-Augmented Generation，RAG）。它已经成为增强大型语言模型（LLM）能力的一项强大技术。RAG 能够让 LLM 提供更准确、更相关且更符合特定上下文的回答，从而缓解“幻觉”和信息过时等问题。

这是三篇系列文章中的第二篇。本文将详细演示如何完全在本地 Python 环境中创建一款由 RAG 驱动的聊天应用。我们将探索如何集成以下几项关键技术，构建一个交互性强、信息丰富的工具：

Reflex：一个现代化的纯 Python 框架，旨在快速构建和部署交互式 Web 应用，无需掌握单独的前端技术（如 JavaScript）。它的响应式特性简化了状态管理。

LangChain：一个专门用于开发语言模型驱动应用的综合性框架。它提供模块化组件和链，用于简化 RAG 等复杂工作流，使流水线的构建变得更加容易。

Ollama：一款日益流行的工具，用户可以使用它直接在本地计算机上下载、运行和管理各种开源 LLM，例如 Google 的 Gemma、Meta 的 Llama 系列和 Mistral 模型等，从而保障隐私并支持离线使用。

FAISS（Facebook AI Similarity Search）：一个高度优化的库，用于在大型向量数据集中高效执行相似性搜索。在我们的 RAG 场景中，它对于根据用户查询的嵌入向量快速查找相关文本段落至关重要。

Hugging Face Datasets 与 Transformers：NLP 生态系统中事实上的标准库。Datasets 可以方便地访问海量数据集，而 sentence-transformers（基于 Transformers 构建）则提供了生成高质量文本嵌入向量的便捷方式。

我们的目标是创建一个基于 Web 的聊天应用，让用户可以提出问题。随后，应用将：

将问题转换为数值向量（嵌入向量）。

搜索预先建立索引的向量存储（使用 FAISS 根据数据集构建），查找嵌入向量相似的文本段落，也就是相关上下文。

将检索到的上下文与原始问题一起提供给本地运行的 LLM（通过 Ollama）。

在使用 Reflex 构建的聊天界面中，将基于检索信息生成的 LLM 回答展示给用户。

初始代码结构与设置

注意：你可以在这个 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 流程

• datasets 提供知识来源

• sentence-transformers 和 faiss-cpu 负责检索机制

• ollama 用于运行本地 LLM

• python-dotenv 用于管理配置

Ollama 提供了许多可供下载和使用的 AI 模型。在本次演示中，我们将使用 gemma3:4b-it-qat。这是一个拥有 40 亿参数的量化模型，与对应的非量化模型相比，所需内存减少了 3 倍。在本次演示中，你也可以自由地将其替换为更大或更小的模型。

本次演示使用 Hugging Face 上的 neural-bridge/rag-dataset-12000 数据集。还有许多其他可用的数据集，你可以下载它们，并在设置中修改数据集名称。

这个文件包含 Reflex 应用的基本设置。对于本项目而言，其内容非常精简，主要只是为应用命名：

import reflex as rx

# Basic configuration defining the application's name
config = rx.Config(
    app_name="rag_app",
)

这个脚本是应用的引擎，负责设置并执行完整的 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"

配置：定义数据集名称（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 []

数据加载（load_and_split_data）：

• 从 Hugging Face 加载指定的数据集（datasets.load_dataset）。

• 支持仅加载数据集的一个子集（DATASET_SUBSET_SIZE），以便更快地进行测试。

• 将每一行的 context 转换为 LangChain Document 对象，并把 question 和 answer 存储在元数据中。（我们最初尝试使用 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

嵌入（get_embeddings_model）：使用 langchain_huggingface.HuggingFaceEmbeddings 初始化 sentence transformer 模型（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

向量存储（create_or_load_vector_store）：

• 使用 FAISS（langchain_community.vectorstores.FAISS）根据文档嵌入向量创建向量索引。

• 更关键的是，它会检查 FAISS_INDEX_PATH 中是否已存在索引。如果存在，则直接加载，以避免每次运行时都重新处理。如果没有找到索引，则创建并保存一个新索引。

LLM 初始化（get_ollama_llm）：

连接到本地运行的 Ollama 服务。

使用 langchain_ollama.OllamaLLM 类（从已弃用的 langchain_community 版本更新后）。

指定所需的模型（OLLAMA_MODEL，默认为 gemma3:4b-it-qat）。

包含错误处理，用于检查指定的模型是否已在 Ollama 中拉取（ollama pull <model_name>）。

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

链设置（setup_rag_chain、get_rag_chain）：

检索器：从 FAISS 向量存储（vector_store.as_retriever）创建检索器，以查找与用户查询最相似的前 k 个文档。

检索器：从 FAISS 向量存储（vector_store.as_retriever）创建检索器，以查找与用户查询最相似的前 k 个文档。

提示模板：定义一个 ChatPromptTemplate，指示 LLM 如何使用提供的上下文来回答问题。我们将模板变量从 {question} 更新为 {input}，以匹配 create_retrieval_chain 的期望。

提示模板：定义一个 ChatPromptTemplate，指示 LLM 如何使用提供的上下文来回答问题。我们将模板变量从 {question} 更新为 {input}，以匹配 create_retrieval_chain 的期望。

填充文档链：使用 create_stuff_documents_chain 获取检索到的文档和用户输入，将它们格式化到提示中，并将其传递给 LLM。

填充文档链：使用 create_stuff_documents_chain 获取检索到的文档和用户输入，将它们格式化到提示中，并将其传递给 LLM。

检索链：使用 create_retrieval_chain 将检索器和填充文档链结合到最终的 RAG 链中。

检索链：使用 create_retrieval_chain 将检索器和填充文档链结合到最终的 RAG 链中。

get_rag_chain 函数充当惰性初始化程序，仅在第一次请求或配置更改（如 Ollama 模型）时设置链。

get_rag_chain 函数充当惰性初始化程序，仅在第一次请求或配置更改（如 Ollama 模型）时设置链。

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

Reflex 应用（State 和 UI）

此文件定义了 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

question：存储输入字段中的当前文本。

chat_history：一个 QA 对象列表（一个简单的 rx.Base 模型，保存问题、答案和加载状态），用于显示对话。

is_loading：用于全局加载状态的布尔值（尽管我们主要使用每条消息的加载状态）。

handle_submit：用户提交表单时触发的异步事件处理程序。

对于 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.2

构建应用

你需要确保遵循正确的结构。然后从项目的根目录运行这两个命令。你的应用将会启动并运行。

注意：在运行应用之前，请确保 Ollama 已启动并运行。成功的消息看起来像这样。一些消息可能会在你在聊天界面中输入"Hi"后出现。

向我们的 RAG 应用发送消息：

然后验证数据集中的声明。

对于这个答案，我们的 RAG 应用会回答并提供正确的上下文；数据集截图来自 Huggingface。所以，我们可以说 RAG 运行得相当不错。

如何使这个应用更准确/生产就绪？

为了使这个应用性能更好、更准确，可以遵循以下几个步骤。

• 使用 Ollama 中更大的模型，如 Gemma 27B、Llama 3.3 70B、Qwen 2.5 70B 或 DeepSeek V3，以获得更好的准确性。

• 使用专用的向量数据库，如 Qdrant、Pinecone、Milvus 等来存储和索引结果。

• 为特定用例创建个性化和专用的数据集，而不是使用互联网语料库。记住，RAG 的效果取决于你提供的数据。为了获得最佳结果，数据应该经过充分清理和优化以用于 AI。

• 视觉升级：你可以按照 Reflex 的指南改进聊天界面，以添加更多功能、额外的状态、存储聊天的数据库等。

通过这个迭代过程，我们成功构建了一个功能完整、本地托管的 RAG 聊天应用。这个项目展示了将 Reflex 用于快速 UI 开发、LangChain 用于复杂的 LLM 工作流编排、FAISS 用于高效的向量搜索，以及 Ollama 提供的隐私和本地 LLM 推理控制力的结合。

LangChain Python Docs

准备好开始了吗？

规模化你的集成战略，并在创纪录的时间内提供你的客户需要的集成。