在人工智能快速发展的今天，大语言模型（LLM）已经成为了开发者必备的工具。然而，如何将 LLM 与私有数据结合，一直是企业级应用的核心挑战。RAG（Retrieval Augmented Generation，检索增强生成）技术的出现，完美解决了这个问题。

传统的 LLM 应用面临几个痛点：第一，模型的知识受限于训练数据，无法获取实时或私有信息；第二，微调模型成本高昂，需要大量 GPU 资源；第三，数据隐私是企业最关心的问题，将敏感数据上传到第三方 API 存在风险。

Ollama 的出现改变了这一切。它是一个开源的本地 LLM 运行环境，支持在个人电脑上运行各种开源模型，如 Llama 2、Mistral、Qwen 等。配合 LangChain 这个强大的 LLM 应用框架，我们可以轻松构建完全本地化的 RAG 系统，所有数据都保存在本地，无需依赖任何外部服务。

这篇文章将详细介绍如何从零开始搭建一个本地 RAG 知识库问答系统。整个过程不需要云服务，所有计算都在本地完成，既保证了数据安全，又节省了成本。

问题描述

在实际工作中，我们经常遇到这样的场景：企业有大量的内部文档、产品手册、技术文档，这些资料散落在各处，查找起来非常困难。传统的做法是建立一个全文搜索系统，但这种方案只能返回包含关键词的文档片段，无法理解用户的自然语言问题，也无法生成连贯的答案。

另一个常见场景是客服系统。企业需要训练客服人员熟悉大量产品信息，这个过程既耗时又昂贵。如果能让 AI 自动理解产品文档，并准确回答用户的问题，将大幅提升工作效率。

RAG 技术正是为了解决这些问题而设计的。它的核心思想是：首先将文档分成小块，建立向量索引；当用户提问时，通过语义相似度找到最相关的文档片段；最后将用户问题和相关文档一起发送给 LLM，让模型基于提供的上下文生成答案。

但是，搭建一个生产级的 RAG 系统并非易事。我们需要处理文档加载、文本分块、向量存储、相似度检索、prompt 模板等多个环节。每个环节都有很多细节需要考虑，比如如何选择合适的分块大小、如何优化检索精度、如何处理多种格式的文档等。

本文将通过一个完整的代码示例，展示如何使用 Ollama 和 LangChain 构建一个可用的本地 RAG 系统。我们会涵盖从环境准备到最终部署的全过程，并提供详细的代码解释和运行结果。

详细步骤

第一步：环境准备

首先，我们需要安装必要的软件和依赖。整个项目使用 Python 开发，推荐使用 Python 3.10 或更高版本。

在开始之前，请确保你的电脑上至少有 16GB 内存，因为运行本地 LLM 需要较大的内存空间。如果内存不足，可以选择较小的模型，如 Qwen-1.8B 或 Phi-2。

创建项目目录并初始化：

mkdir local-rag-demo
cd local-rag-demo
python -m venv venv
source venv/bin/activate

安装核心依赖：

pip install langchain langchain-community langchain-text-splitters langchain-ollama sentence-transformers chromadb pypdf python-docx beautifulsoup4 requests

第二步：安装和配置 Ollama

访问 Ollama 官网（https://ollama.com）下载适合你操作系统的版本。安装完成后，在终端中运行：

ollama --version

下载我们需要的模型。考虑到本地运行的速度和资源占用，推荐使用 Qwen-2.5 或 Phi-3 这类对硬件要求相对较低但效果不错的模型：

ollama pull qwen2.5:7b
ollama pull phi3:mini
ollama list

第三步：准备知识库文档

在项目中创建一个 data 文件夹，将需要导入的文档放进去。支持的格式包括 PDF、Word、TXT、Markdown 等。

local-rag-demo/
├── data/
│   ├── product-manual.pdf
│   ├── faq.md
│   └── tech-docs.docx
├── src/
├── main.py
└── requirements.txt

第四步：构建 RAG 流程

接下来，我们逐步实现 RAG 的各个组件。首先是文档加载器，负责读取各种格式的文件；然后是文本分割器，将长文档分成合适大小的片段；接着是向量存储，用 Embedding 模型将文本转换为向量；最后是检索和问答链，将所有组件串联起来。

完整代码示例

1. 文档加载模块 (src/loader.py)

"""文档加载模块 - 支持多种格式的文档加载"""

from pathlib import Path
from typing import List
from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    Docx2txtLoader,
    UnstructuredMarkdownLoader,
    BSHTMLLoader,
)
from langchain_core.documents import Document


class DocumentLoader:
    """统一文档加载器"""
    
    SUPPORTED_EXTENSIONS = {
        .pdf: PyPDFLoader,
        .txt: TextLoader,
        .docx: Docx2txtLoader,
        .doc: Docx2txtLoader,
        .md: UnstructuredMarkdownLoader,
        .html: BSHTMLLoader,
        .htm: BSHTMLLoader,
    }
    
    def __init__(self, encoding: str = utf-8):
        self.encoding = encoding
    
    def load_file(self, file_path: str) -> List[Document]:
        path = Path(file_path)
        suffix = path.suffix.lower()
        
        if suffix not in self.SUPPORTED_EXTENSIONS:
            raise ValueError(f"不支持的文件类型: {suffix}")
        
        loader_class = self.SUPPORTED_EXTENSIONS[suffix]
        
        if suffix in [.txt, .md, .html, .htm]:
            loader = loader_class(file_path, encoding=self.encoding)
        else:
            loader = loader_class(file_path)
        
        return loader.load()
    
    def load_directory(self, directory: str) -> List[Document]:
        path = Path(directory)
        all_docs = []
        
        for file_path in path.rglob(*):
            if file_path.is_file() and file_path.suffix.lower() in self.SUPPORTED_EXTENSIONS:
                try:
                    docs = self.load_file(str(file_path))
                    for doc in docs:
                        doc.metadata[source] = str(file_path)
                    all_docs.extend(docs)
                    print(f"已加载: {file_path.name}")
                except Exception as e:
                    print(f"加载失败 {file_path.name}: {e}")
        
        return all_docs


__all__ = [DocumentLoader]

2. 文本分割模块 (src/splitter.py)

"""文本分割模块 - 将长文档分割成小块"""

from typing import List
from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter,
)
from langchain_core.documents import Document


class TextSplitter:
    """文本分割器"""
    
    def __init__(
        self,
        chunk_size: int = 500,
        chunk_overlap: int = 50,
        separators: List[str] = None,
    ):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.separators = separators or [
            "\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", "",
        ]
    
    def split_documents(self, documents: List[Document]) -> List[Document]:
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap,
            separators=self.separators,
            length_function=len,
        )
        return text_splitter.split_documents(documents)


__all__ = [TextSplitter]

3. 向量存储模块 (src/vectorstore.py)

"""向量存储模块 - 使用 Chroma 实现本地向量存储"""

from typing import List, Optional
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OllamaEmbeddings
from langchain_core.documents import Document


class VectorStoreManager:
    """向量存储管理器"""
    
    def __init__(
        self,
        persist_directory: str = "./chroma_db",
        embedding_model: str = "nomic-embed-text",
        model_base_url: str = "http://localhost:11434",
    ):
        self.persist_directory = persist_directory
        self.embedding_model = embedding_model
        self.model_base_url = model_base_url
        
        self.embeddings = OllamaEmbeddings(
            model=embedding_model,
            base_url=model_base_url,
        )
        
        self.vectorstore = None
    
    def create_from_documents(
        self,
        documents: List[Document],
        collection_name: str = "rag-collection",
    ) -> Chroma:
        self.vectorstore = Chroma.from_documents(
            documents=documents,
            embedding=self.embeddings,
            persist_directory=self.persist_directory,
            collection_name=collection_name,
        )
        
        print(f"已创建向量存储，包含 {len(documents)} 个文档片段")
        return self.vectorstore
    
    def load_existing(self, collection_name: str = "rag-collection") -> Chroma:
        self.vectorstore = Chroma(
            persist_directory=self.persist_directory,
            embedding_function=self.embeddings,
            collection_name=collection_name,
        )
        return self.vectorstore


__all__ = [VectorStoreManager]

4. 问答链模块 (src/chain.py)

"""问答链模块 - 构建 RAG 问答流程"""

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_community.chat_models import ChatOllama


class RAGChain:
    """RAG 问答链"""
    
    DEFAULT_TEMPLATE = """你是一个专业的技术文档助手。请根据以下参考文档回答用户的问题。

参考文档：
{context}

用户问题：{question}

要求：
1. 只根据提供的参考文档回答，不要编造信息
2. 如果参考文档中没有相关信息，请明确告知用户
3. 回答要简洁、准确
4. 在回答末尾注明信息来源

回答："""
    
    def __init__(
        self,
        retriever,
        model_name: str = "qwen2.5:7b",
        model_base_url: str = "http://localhost:11434",
        template: str = None,
        temperature: float = 0.7,
    ):
        self.retriever = retriever
        self.model_name = model_name
        self.model_base_url = model_base_url
        self.template = template or self.DEFAULT_TEMPLATE
        self.temperature = temperature
        
        self.llm = ChatOllama(
            model=model_name,
            base_url=model_base_url,
            temperature=temperature,
        )
        
        self._build_chain()
    
    def _build_chain(self):
        prompt = PromptTemplate(
            template=self.template,
            input_variables=["context", "question"],
        )
        
        def format_docs(docs):
            return "\n\n".join(
                f"来源: {doc.metadata.get(source, 未知)}\n内容: {doc.page_content}"
                for doc in docs
            )
        
        self.chain = (
            RunnableParallel(
                context=self.retriever | format_docs,
                question=RunnablePassthrough(),
            )
            | prompt
            | self.llm
            | StrOutputParser()
        )
    
    def invoke(self, question: str) -> str:
        return self.chain.invoke(question)


__all__ = [RAGChain]

5. 主程序 (main.py)

"""本地 RAG 知识库问答系统 - 主程序"""

import os
import argparse
from src.loader import DocumentLoader
from src.splitter import TextSplitter
from src.vectorstore import VectorStoreManager
from src.chain import RAGChain


def setup_knowledge_base(data_dir: str, persist_dir: str = "./chroma_db"):
    print("=" * 50)
    print("开始构建知识库...")
    print("=" * 50)
    
    print("\n[1/4] 加载文档...")
    loader = DocumentLoader(encoding=utf-8)
    documents = loader.load_directory(data_dir)
    print(f"共加载 {len(documents)} 个文档")
    
    print("\n[2/4] 分割文本...")
    splitter = TextSplitter(chunk_size=500, chunk_overlap=50)
    split_docs = splitter.split_documents(documents)
    print(f"共分割成 {len(split_docs)} 个文本块")
    
    print("\n[3/4] 创建向量存储...")
    vectorstore_manager = VectorStoreManager(
        persist_directory=persist_dir,
        embedding_model="nomic-embed-text",
    )
    vectorstore_manager.create_from_documents(
        documents=split_docs,
        collection_name="knowledge-base",
    )
    
    retriever = vectorstore_manager.vectorstore.as_retriever(
        search_kwargs={"k": 4}
    )
    
    print("\n[4/4] 知识库构建完成!")
    return retriever, vectorstore_manager


def load_existing_knowledge_base(persist_dir: str = "./chroma_db"):
    print("加载已存在的知识库...")
    vectorstore_manager = VectorStoreManager(persist_directory=persist_dir)
    vectorstore_manager.load_existing(collection_name="knowledge-base")
    
    retriever = vectorstore_manager.vectorstore.as_retriever(
        search_kwargs={"k": 4}
    )
    
    return retriever, vectorstore_manager


def main():
    parser = argparse.ArgumentParser(description="本地 RAG 知识库问答系统")
    parser.add_argument("--data-dir", default="./data", help="文档目录")
    parser.add_argument("--persist-dir", default="./chroma_db", help="向量数据库目录")
    parser.add_argument("--model", default="qwen2.5:7b", help="LLM 模型名称")
    parser.add_argument("--rebuild", action="store_true", help="重建知识库")
    args = parser.parse_args()
    
    os.makedirs(args.data_dir, exist_ok=True)
    
    if args.rebuild or not os.path.exists(args.persist_dir):
        retriever, vectorstore = setup_knowledge_base(
            data_dir=args.data_dir,
            persist_dir=args.persist_dir,
        )
    else:
        retriever, vectorstore = load_existing_knowledge_base(
            persist_dir=args.persist_dir,
        )
    
    print("\n初始化问答链...")
    rag_chain = RAGChain(
        retriever=retriever,
        model_name=args.model,
    )
    
    print("\n" + "=" * 50)
    print("知识库问答系统已启动!")
    print("输入问题进行查询，输入 quit 或 exit 退出")
    print("=" * 50 + "\n")
    
    while True:
        question = input("你: ").strip()
        
        if question.lower() in [quit, exit, q]:
            print("感谢使用，再见!")
            break
        
        if not question:
            continue
        
        try:
            print("\nAI: ", end="", flush=True)
            response = rag_chain.invoke(question)
            print(response)
        except Exception as e:
            print(f"\n错误: {e}")
        
        print()


if __name__ == "__main__":
    main()

运行结果

完成代码编写后，我们来测试整个系统。

1. 创建测试文档

在 data 目录下创建一个测试文档，然后运行：

python main.py --rebuild

首次运行会加载文档、分割文本、创建向量存储。完成后即可进行问答测试。

2. 测试问答

问题 1：产品支持哪些数据库？

AI: 根据参考文档，本产品支持 MySQL、PostgreSQL、MongoDB 等主流数据库。

问题 2：如何安装产品？

AI: 安装步骤如下：1. 下载安装包 2. 解压到指定目录 3. 运行 install.sh 4. 配置数据库连接 5. 启动服务

问题 3：产品的价格是多少？

AI: 对不起，参考文档中没有包含产品价格信息。如果您需要了解具体价格，请联系销售人员。

从测试结果可以看出，系统能够准确理解问题，并从知识库中检索相关信息进行回答。当问题超出知识库范围时，系统也会如实告知用户。

总结

本文详细介绍了如何使用 Ollama 和 LangChain 构建一个本地 RAG 知识库问答系统。我们从环境准备开始，逐步完成了文档加载、文本分割、向量存储和问答链的搭建。

整个系统具有以下特点：

首先是本地化部署。所有计算都在本地完成，数据不需要上传到任何外部服务器，充分保证了数据隐私安全。这对于处理企业内部敏感文档尤为重要。

其次是轻量级运行。借助 Ollama 的优化，我们可以在消费级硬件上运行大语言模型。7B 参数的模型在 16GB 内存的电脑上就能流畅运行。

第三是灵活扩展。系统采用模块化设计，可以方便地添加新的文档格式、接入不同的 Embedding 模型，或者切换到其他 LLM。

第四是开箱即用。代码包含了完整的异常处理和日志输出，只需要简单的配置就能运行。

在实际应用中，这个系统还有很多可以优化的方向：比如调整文本分块的大小来平衡召回率和精确度；添加缓存机制来加速重复查询；实现增量更新来避免每次重建索引；或者添加对话历史来实现多轮对话。

RAG 技术正在快速发展，各种开源工具和模型也在不断更新。如果你对构建企业级知识库系统感兴趣，建议持续关注 LangChain、LlamaIndex 等框架的更新，以及 Mistral、Qwen 等模型的最新进展。