前言

💡 痛点：个人积累了几 GB 的文档、笔记、PDF，但找不到想要的信息？

🎯 解决方案：搭建一个本地 RAG 系统，让 AI 帮你从所有文档中精准检索答案。

RAG（Retrieval-Augmented Generation） 是目前最实用的 AI 应用架构之一。本文将带你从零搭建一个生产级个人知识库系统，支持：

• 📚 多格式文档（PDF、Markdown、Word、网页）
• 🔍 语义搜索（不是关键词匹配，而是理解意图）
• 💬 对话式问答（带上下文记忆）
• 🚀 本地部署（数据不出本地，隐私安全）
• 📊 生产级优化（分块策略、重排序、评估体系）

一、RAG 系统架构详解

1.1 什么是 RAG？

传统 AI 问答 vs RAG 对比：

1.2 RAG 完整工作流程

1.3 系统架构图

1.4 技术选型对比

详细对比表：

二、环境准备与安装

2.1 系统要求检查清单

推荐配置 vs 最低配置：

2.2 安装 Ollama（本地 LLM 运行时）

Ollama 可以让你在本地运行开源大模型，无需 API Key。

安装步骤：

macOS/Linux：

# 一键安装
curl -fsSL https://ollama.com/install.sh | sh

# 验证安装
ollama --version
# 应显示：ollama version 0.5.7 或更高

# 启动 Ollama 服务（默认端口 11434）
ollama serve &

Windows (WSL2)：

# 在 WSL2 中执行（同上）
curl -fsSL https://ollama.com/install.sh | sh

手动下载（如果脚本失败）：

• 访问 https://ollama.com/download
• 下载对应系统版本
• 安装后运行 ollama serve

2.3 下载 LLM 和 Embedding 模型

执行命令：

# 1. 下载对话模型（Llama 3.1 8B，约 4.7GB）
ollama pull llama3.1:8b

# 2. 下载 Embedding 模型（用于向量化文本）
ollama pull nomic-embed-text

# 验证模型
ollama list
# 应显示：
# NAME              SIZE   MODIFIED
# llama3.1:8b      4.7 GB  ...
# nomic-embed-text  274 MB  ...

模型选择建议：

2.4 安装 Python 依赖

操作步骤：

# 创建虚拟环境（推荐）
python3 -m venv rag-env
source rag-env/bin/activate  # macOS/Linux
# rag-env\Scripts\activate  # Windows

# 安装依赖
pip install -U pip
pip install \
  langchain>=0.3.0 \
  langchain-ollama>=0.2.0 \
  langchain-community>=0.3.0 \
  chromadb>=0.5.0 \
  sentence-transformers>=3.0.0 \
  pypdf>=1.13.0 \
  python-docx>=1.1.0 \
  streamlit>=1.39.0 \
  tiktoken>=0.8.0 \
  rank-bm25>=0.2.2 \
  --index-url https://pypi.tuna.tsinghua.edu.cn/simple

依赖说明表：

三、核心功能实现

3.1 项目结构

完整目录结构：

personal-rag/
├── 📁 data/                      # 原始文档存放目录
│   ├── 📁 pdfs/                 # PDF 文件
│   ├── 📁 markdown/             # Markdown 文件
│   └── 📁 web/                 # 网页存档
├── 📁 vectorstore/              # Chroma 向量数据库文件
├── 📁 src/                     # 源代码
│   ├── 📄 __init__.py
│   ├── 📄 document_loader.py    # 文档加载模块
│   ├── 📄 text_splitter.py      # 文本分块模块
│   ├── 📄 embeddings.py         # Embedding 封装
│   ├── 📄 vector_store.py       # 向量数据库操作
│   ├── 📄 retriever.py         # 检索器（含重排序）
│   └── 📄 qa_chain.py          # 问答链
├── 📄 app.py                    # Streamlit Web UI
├── 📄 cli.py                    # 命令行界面
├── 📄 config.yaml               # 配置文件
├── 📄 requirements.txt          # 依赖清单
└── 📄 README.md                # 项目文档

3.2 配置文件详解

创建 config.yaml：

# config.yaml - RAG 系统配置

# ═════════════════════════════════════════
# 🤖 LLM 配置
# ═════════════════════════════════════════
llm:
  provider: "ollama"          # ollama / openai / anthropic
  model: "llama3.1:8b"
  base_url: "http://localhost:11434/v1"
  temperature: 0.1              # 创造性（0=精确，1=创造性）
  max_tokens: 2048

# ═════════════════════════════════════════
# 🔢 Embedding 配置
# ═════════════════════════════════════════
embedding:
  provider: "ollama"
  model: "nomic-embed-text"
  base_url: "http://localhost:11434/api"

# ═════════════════════════════════════════
# ✂️ 文本分块配置
# ═════════════════════════════════════════
text_splitter:
  chunk_size: 512           # 每个块的最大字符数
  chunk_overlap: 50         # 块之间的重叠字符数
  separator: "\n\n"        # 分割符（优先按段落分割）

# ═════════════════════════════════════════
# 🔍 检索配置
# ═════════════════════════════════════════
retriever:
  search_type: "mmr"       # similarity / mmr（最大边际相关性）
  k: 5                      # 检索返回的文档数
  fetch_k: 20               # MMR 候选集大小
  lambda_mult: 0.5          # MMR 多样性参数

# ═════════════════════════════════════════
# 🔄 重排序配置
# ═════════════════════════════════════════
reranking:
  enabled: true
  model: "BAAI/bge-reranker-base"  # 重排序模型
  top_n: 3                            # 重排序后保留的文档数

# ═════════════════════════════════════════
# 📊 向量数据库配置
# ═════════════════════════════════════════
vectorstore:
  provider: "chroma"        # chroma / pinecone / pgvector
  persist_directory: "./vectorstore"
  collection_name: "personal_knowledge_base"

# ═════════════════════════════════════════
# 📁 文件路径配置
# ═════════════════════════════════════════
paths:
  data_dir: "./data"
  pdf_dir: "./data/pdfs"
  md_dir: "./data/markdown"
  web_dir: "./data/web"

配置参数详解：

3.3 文档加载模块

创建 src/document_loader.py：

"""
文档加载模块 - 支持多种格式的文档加载
"""
import os
from pathlib import Path
from typing import List, Optional
from langchain.document import Document
from langchain_community.document_loaders import (
    PyPDFLoader,
    Docx2txtLoader,
    TextLoader,
    UnstructuredMarkdownLoader,
    WebBaseLoader,
    CSVLoader,
    JSONLoader,
)


class DocumentLoader:
    """统一的文档加载器"""

    def __init__(self, data_dir: str = "./data"):
        self.data_dir = Path(data_dir)
        self.supported_extensions = {
            ".pdf": self._load_pdf,
            ".docx": self._load_docx,
            ".txt": self._load_txt,
            ".md": self._load_markdown,
            ".csv": self._load_csv,
            ".json": self._load_json,
        }

    def load_file(self, file_path: str) -> List[Document]:
        """
        加载单个文件
        
        Args:
            file_path: 文件路径
            
        Returns:
            加载的文档列表
        """
        path = Path(file_path)
        extension = path.suffix.lower()

        if extension not in self.supported_extensions:
            raise ValueError(f"不支持的文件格式: {extension}")

        loader_func = self.supported_extensions[extension]
        return loader_func(str(path))

    def load_directory(self, directory: Optional[str] = None) -> List[Document]:
        """
        加载整个目录下的所有支持的文件
        
        Args:
            directory: 目录路径，默认为 self.data_dir
            
        Returns:
            加载的文档列表
        """
        target_dir = Path(directory) if directory else self.data_dir
        documents = []

        for file_path in target_dir.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)
                        doc.metadata["filename"] = file_path.name
                    documents.extend(docs)
                    print(f"✅ 已加载: {file_path.name}")
                except Exception as e:
                    print(f"❌ 加载失败 {file_path.name}: {e}")

        return documents

    def load_web_page(self, url: str) -> List[Document]:
        """
        加载网页内容
        
        Args:
            url: 网页 URL
            
        Returns:
            加载的文档列表
        """
        loader = WebBaseLoader(url)
        documents = loader.load()
        for doc in documents:
            doc.metadata["source"] = url
            doc.metadata["type"] = "web"
        return documents

    def _load_pdf(self, file_path: str) -> List[Document]:
        """加载 PDF 文件"""
        loader = PyPDFLoader(file_path)
        return loader.load()

    def _load_docx(self, file_path: str) -> List[Document]:
        """加载 Word 文档"""
        loader = Docx2txtLoader(file_path)
        return loader.load()

    def _load_txt(self, file_path: str) -> List[Document]:
        """加载纯文本文件"""
        loader = TextLoader(file_path, encoding="utf-8")
        return loader.load()

    def _load_markdown(self, file_path: str) -> List[Document]:
        """加载 Markdown 文件"""
        loader = UnstructuredMarkdownLoader(file_path)
        return loader.load()

    def _load_csv(self, file_path: str) -> List[Document]:
        """加载 CSV 文件"""
        loader = CSVLoader(file_path)
        return loader.load()

    def _load_json(self, file_path: str) -> List[Document]:
        """加载 JSON 文件"""
        loader = JSONLoader(
            file_path=file_path,
            jq_schema=".",  # 加载整个 JSON
            text_content=False
        )
        return loader.load()


# 使用示例
if __name__ == "__main__":
    loader = DocumentLoader("./data")
    
    # 加载单个文件
    docs = loader.load_file("./data/pdfs/example.pdf")
    print(f"加载了 {len(docs)} 个文档片段")
    
    # 加载整个目录
    all_docs = loader.load_directory()
    print(f"总共加载了 {len(all_docs)} 个文档片段")

支持的文档格式：

3.4 文本分块模块

创建 src/text_splitter.py：

"""
文本分块模块 - 将长文档分割成合适大小的块
"""
from typing import List
from langchain.document import Document
from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    MarkdownTextSplitter,
    PythonCodeTextSplitter,
)


class TextSplitter:
    """智能文本分块器"""

    def __init__(
        self,
        chunk_size: int = 512,
        chunk_overlap: int = 50,
        separator: str = "\n\n"
    ):
        """
        初始化文本分块器
        
        Args:
            chunk_size: 每个块的最大字符数
            chunk_overlap: 块之间的重叠字符数（保持上下文连贯）
            separator: 分割符优先级
        """
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.separator = separator

    def split_documents(
        self,
        documents: List[Document],
        doc_type: str = "general"
    ) -> List[Document]:
        """
        将文档列表分块
        
        Args:
            documents: 原始文档列表
            doc_type: 文档类型（general/markdown/python）
            
        Returns:
            分块后的文档列表
        """
        if doc_type == "markdown":
            splitter = MarkdownTextSplitter(
                chunk_size=self.chunk_size,
                chunk_overlap=self.chunk_overlap
            )
        elif doc_type == "python":
            splitter = PythonCodeTextSplitter(
                chunk_size=self.chunk_size,
                chunk_overlap=self.chunk_overlap
            )
        else:
            splitter = RecursiveCharacterTextSplitter(
                chunk_size=self.chunk_size,
                chunk_overlap=self.chunk_overlap,
                separators=["\n\n", "\n", "。", "；", " ", ""]
            )

        split_docs = splitter.split_documents(documents)
        
        # 添加分块元数据
        for i, doc in enumerate(split_docs):
            doc.metadata["chunk_id"] = i
            doc.metadata["chunk_size"] = len(doc.page_content)
            
        print(f"✅ 分块完成: {len(documents)} 个文档 → {len(split_docs)} 个块")
        return split_docs

    def estimate_tokens(self, text: str) -> int:
        """
        估算 token 数量（用于分块大小规划）
        
        Args:
            text: 文本内容
            
        Returns:
            估算的 token 数
        """
        # 粗略估算：1 个中文字符 ≈ 2 个 token，1 个英文单词 ≈ 1.3 个 token
        chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        english_words = len(text.split())
        return int(chinese_chars * 2 + english_words * 1.3)

分块策略可视化：

分块大小选择指南：

chunk_overlap 的作用：

┌─────────────────────────────────────┐
│ 块 1: "...人工智能是...['未来'] 发展的方向..."     │
│                          ↑ chunk_overlap=50      │
│ 块 2: "['未来'] 发展的方向...将会深刻影响..."     │
└─────────────────────────────────────┘
→ 避免分割关键句子，保持语义连贯

3.5 Embedding 模块

创建 src/embeddings.py：

"""
Embedding 模块 - 文本向量化
"""
from langchain.ollama import OllamaEmbeddings
from langchain.embeddings.base import Embeddings


class EmbeddingManager:
    """Embedding 管理器"""

    def __init__(
        self,
        model: str = "nomic-embed-text",
        base_url: str = "http://localhost:11434/api"
    ):
        """
        初始化 Embedding 模型
        
        Args:
            model: Embedding 模型名称（Ollama 拉取的模型）
            base_url: Ollama API 地址
        """
        self.model = model
        self.base_url = base_url
        self.embeddings = OllamaEmbeddings(
            model=model,
            base_url=base_url
        )

    def get_embeddings(self) -> Embeddings:
        """
        获取 LangChain Embeddings 对象
        
        Returns:
            LangChain Embeddings 实例
        """
        return self.embeddings

    def embed_query(self, text: str) -> List[float]:
        """
        将查询文本向量化（用于检索）
        
        Args:
            text: 查询文本
            
        Returns:
            向量表示
        """
        return self.embeddings.embed_query(text)

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """
        将文档列表向量化（用于存储）
        
        Args:
            texts: 文档文本列表
            
        Returns:
            向量列表
        """
        return self.embeddings.embed_documents(texts)

Embedding 模型对比：

详细对比表：

3.6 向量数据库模块

创建 src/vector_store.py：

"""
向量数据库模块 - Chroma 封装
"""
from pathlib import Path
from typing import List, Optional
from langchain.document import Document
from langchain.vectorstores import Chroma
from langchain.embeddings.base import Embeddings


class VectorStoreManager:
    """向量数据库管理器"""

    def __init__(
        self,
        embeddings: Embeddings,
        persist_directory: str = "./vectorstore",
        collection_name: str = "personal_knowledge_base"
    ):
        """
        初始化向量数据库
        
        Args:
            embeddings: Embedding 模型
            persist_directory: 数据库持久化目录
            collection_name: 集合名称
        """
        self.embeddings = embeddings
        self.persist_directory = Path(persist_directory)
        self.collection_name = collection_name
        
        # 确保目录存在
        self.persist_directory.mkdir(parents=True, exist_ok=True)
        
        self.vectorstore = None

    def create_vectorstore(self, documents: List[Document]) -> Chroma:
        """
        创建向量数据库（首次索引）
        
        Args:
            documents: 分块后的文档列表
            
        Returns:
            Chroma 向量数据库实例
        """
        self.vectorstore = Chroma.from_documents(
            documents=documents,
            embedding=self.embeddings,
            persist_directory=str(self.persist_directory),
            collection_name=self.collection_name
        )
        print(f"✅ 向量数据库创建成功！共索引 {len(documents)} 个文档块")
        return self.vectorstore

    def load_vectorstore(self) -> Optional[Chroma]:
        """
        加载已有的向量数据库
        
        Returns:
            Chroma 向量数据库实例，如果不存在返回 None
        """
        if not self.persist_directory.exists():
            print("⚠️ 向量数据库不存在，请先创建")
            return None
            
        self.vectorstore = Chroma(
            persist_directory=str(self.persist_directory),
            embedding_function=self.embeddings,
            collection_name=self.collection_name
        )
        count = self.vectorstore._collection.count()
        print(f"✅ 向量数据库加载成功！当前有 {count} 个文档块")
        return self.vectorstore

    def add_documents(self, documents: List[Document]):
        """
        向已有数据库添加新文档
        
        Args:
            documents: 新文档列表
        """
        if self.vectorstore is None:
            raise ValueError("向量数据库未初始化，请先调用 create_vectorstore 或 load_vectorstore")
        
        self.vectorstore.add_documents(documents)
        print(f"✅ 已添加 {len(documents)} 个新文档块")

    def similarity_search(
        self,
        query: str,
        k: int = 5,
        filter: Optional[dict] = None
    ) -> List[Document]:
        """
        相似度搜索
        
        Args:
            query: 查询文本
            k: 返回的最相似文档数
            filter: 元数据过滤条件
            
        Returns:
            最相似的文档列表
        """
        if self.vectorstore is None:
            raise ValueError("向量数据库未初始化")
        
        return self.vectorstore.similarity_search(
            query=query,
            k=k,
            filter=filter
        )

    def similarity_search_with_score(
        self,
        query: str,
        k: int = 5
    ) -> List[tuple]:
        """
        带相似度分数的搜索
        
        Args:
            query: 查询文本
            k: 返回的最相似文档数
            
        Returns:
            (文档, 相似度分数) 元组列表
        """
        if self.vectorstore is None:
            raise ValueError("向量数据库未初始化")
        
        return self.vectorstore.similarity_search_with_score(
            query=query,
            k=k
        )

    def delete_collection(self):
        """删除整个集合（慎用）"""
        if self.vectorstore:
            self.vectorstore.delete_collection()
            print("✅ 集合已删除")

Chroma vs 其他向量数据库对比：

详细对比表：

3.7 检索器模块（含重排序）

创建 src/retriever.py：

"""
检索器模块 - 包含重排序等高级检索技术
"""
from typing import List
from langchain.document import Document
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import DocumentCompressorPipeline
from langchain_community.document_compressors import BgeRerank
from langchain.retrievers import MergerRetriever
from langchain.retrievers.multi_query import MultiQueryRetriever


class AdvancedRetriever:
    """高级检索器（含重排序、多查询等）"""

    def __init__(
        self,
        vectorstore,
        search_type: str = "mmr",
        k: int = 5,
        fetch_k: int = 20,
        lambda_mult: float = 0.5
    ):
        """
        初始化检索器
        
        Args:
            vectorstore: 向量数据库实例
            search_type: 搜索类型（similarity/mmr）
            k: 返回的文档数
            fetch_k: MMR 候选集大小
            lambda_mult: MMR 多样性参数（0=最多样，1=最相关）
        """
        self.vectorstore = vectorstore
        self.search_type = search_type
        self.k = k
        self.fetch_k = fetch_k
        self.lambda_mult = lambda_mult

    def get_basic_retriever(self):
        """获取基础检索器"""
        if self.search_type == "mmr":
            return self.vectorstore.as_retriever(
                search_type="mmr",
                search_kwargs={
                    "k": self.k,
                    "fetch_k": self.fetch_k,
                    "lambda_mult": self.lambda_mult
                }
            )
        else:
            return self.vectorstore.as_retriever(
                search_kwargs={"k": self.k}
            )

    def get_reranking_retriever(self, top_n: int = 3):
        """
        获取带重排序的检索器
        
        重排序（Reranking）的作用：
        - 基础检索可能返回不太相关的文档
        - 重排序模型会重新打分，提高最相关文档的排名
        
        Args:
            top_n: 重排序后保留的文档数
            
        Returns:
            带重排序的检索器
        """
        # 基础检索器
        base_retriever = self.get_basic_retriever()
        
        # 重排序模型（需提前下载：pip install sentence-transformers）
        reranker = BgeRerank(
            model="BAAI/bge-reranker-base",
            top_n=top_n
        )
        
        # 压缩管道
        pipeline = DocumentCompressorPipeline(
            transformers=[reranker]
        )
        
        # 上下文压缩检索器
        compression_retriever = ContextualCompressionRetriever(
            base_compressor=pipeline,
            base_retriever=base_retriever
        )
        
        return compression_retriever

    def get_multi_query_retriever(self, llm):
        """
        获取多查询检索器
        
        多查询检索的作用：
        - 自动生成多个不同角度的查询
        - 提高召回率
        
        Args:
            llm: 语言模型实例
            
        Returns:
            多查询检索器
        """
        return MultiQueryRetriever.from_llm(
            retriever=self.get_basic_retriever(),
            llm=llm
        )

检索优化技术栈：

RAG 检索优化技术详解：

3.8 问答链模块

创建 src/qa_chain.py：

"""
问答链模块 - 构建 RAG 问答系统
"""
from typing import List, Dict, Any
from langchain.document import Document
from langchain.prompts import PromptTemplate
from langchain_ollama import ChatOllama
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.memory import ConversationBufferMemory
from langchain.chains import create_history_aware_retriever
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema import StrOutputParser


class RAGQuestionAnswering:
    """RAG 问答系统"""

    def __init__(
        self,
        llm,
        retriever,
        system_prompt: Optional[str] = None
    ):
        """
        初始化问答系统
        
        Args:
            llm: 语言模型实例
            retriever: 检索器实例
            system_prompt: 系统提示词（可选）
        """
        self.llm = llm
        self.retriever = retriever
        
        # 默认系统提示词
        self.system_prompt = system_prompt or """你是回答问题的助手。
使用以下检索到的上下文来回答问题。
如果不知道答案，就说不知道，不要编造答案。
保持答案简洁，最多 3-5 句话。

上下文：
{context}

问题：
{question}

回答："""

    def create_basic_qa_chain(self):
        """
        创建基础 QA 链（无对话历史）
        
        Returns:
            QA 链
        """
        prompt = PromptTemplate(
            template=self.system_prompt,
            input_variables=["context", "question"]
        )
        
        # 创建文档合并链
        document_chain = create_stuff_documents_chain(
            llm=self.llm,
            prompt=prompt
        )
        
        # 创建检索 QA 链
        qa_chain = create_retrieval_chain(
            retriever=self.retriever,
            combine_docs_chain=document_chain
        )
        
        return qa_chain

    def create_conversational_qa_chain(self):
        """
        创建对话式 QA 链（带历史记忆）
        
        Returns:
            对话式 QA 链
        """
        # 对话历史感知检索器
        history_aware_retriever = create_history_aware_retriever(
            self.llm,
            self.retriever,
            PromptTemplate(
                template="根据对话历史和最新问题，生成一个独立的搜索查询。",
                input_variables=["chat_history", "input"]
            )
        )
        
        # QA 提示词
        qa_prompt = PromptTemplate(
            template="""根据以下上下文和对话历史回答问题：

上下文：{context}

对话历史：{chat_history}

问题：{input}

回答：""",
            input_variables=["context", "chat_history", "input"]
        )
        
        # 文档合并链
        document_chain = create_stuff_documents_chain(self.llm, qa_prompt)
        
        # 检索 QA 链
        qa_chain = create_retrieval_chain(
            history_aware_retriever,
            document_chain
        )
        
        return qa_chain

    def answer(self, question: str, chat_history: List[tuple] = None) -> Dict[str, Any]:
        """
        回答问题
        
        Args:
            question: 用户问题
            chat_history: 对话历史 [(human, ai), ...]
            
        Returns:
            包含答案和检索文档的字典
        """
        if chat_history:
            qa_chain = self.create_conversational_qa_chain()
            result = qa_chain.invoke({
                "input": question,
                "chat_history": chat_history
            })
        else:
            qa_chain = self.create_basic_qa_chain()
            result = qa_chain.invoke({"input": question})
        
        return {
            "answer": result["answer"],
            "source_documents": result["context"]
        }

提示词工程详解：

RAG 提示词最佳实践：

优化后的提示词模板：

你是我的个人知识库助手。

任务：
1. 根据检索到的上下文回答问题
2. 如果上下文不包含答案，说"文档中没有相关信息"
3. 回答时引用来源（格式：[来源: filename.pdf, 第 3 页]）
4. 用中文回答，技术术语保留英文
5. 代码用 Markdown 代码块包裹

上下文：
{context}

问题：{question}

回答：

四、整合：构建完整应用

4.1 主程序入口（CLI）

创建 cli.py（命令行界面）：

"""
命令行界面 - 与知识库交互
"""
import yaml
from pathlib import Path
from src.document_loader import DocumentLoader
from src.text_splitter import TextSplitter
from src.embeddings import EmbeddingManager
from src.vector_store import VectorStoreManager
from src.retriever import AdvancedRetriever
from src.qa_chain import RAGQuestionAnswering
from langchain_ollama import ChatOllama


def load_config(config_path: str = "config.yaml") -> dict:
    """加载配置文件"""
    with open(config_path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


def main():
    """主程序"""
    # 1. 加载配置
    config = load_config()
    print("✅ 配置加载成功")
    
    # 2. 初始化组件
    print("\n🚀 正在初始化组件...")
    
    # Embedding 模型
    embedding_manager = EmbeddingManager(
        model=config["embedding"]["model"],
        base_url=config["embedding"]["base_url"]
    )
    embeddings = embedding_manager.get_embeddings()
    print("  ✅ Embedding 模型已加载")
    
    # LLM
    llm = ChatOllama(
        model=config["llm"]["model"],
        base_url=config["llm"]["base_url"],
        temperature=config["llm"]["temperature"]
    )
    print("  ✅ LLM 已加载")
    
    # 向量数据库
    vectorstore_manager = VectorStoreManager(
        embeddings=embeddings,
        persist_directory=config["vectorstore"]["persist_directory"],
        collection_name=config["vectorstore"]["collection_name"]
    )
    
    # 3. 索引文档（如果数据库不存在）
    vectorstore = vectorstore_manager.load_vectorstore()
    
    if vectorstore is None:
        print("\n📚 开始索引文档...")
        loader = DocumentLoader(config["paths"]["data_dir"])
        documents = loader.load_directory()
        
        print("\n✂️ 开始分块...")
        text_splitter = TextSplitter(
            chunk_size=config["text_splitter"]["chunk_size"],
            chunk_overlap=config["text_splitter"]["chunk_overlap"]
        )
        split_documents = text_splitter.split_documents(documents)
        
        print("\n💾 创建向量数据库...")
        vectorstore = vectorstore_manager.create_vectorstore(split_documents)
    
    # 4. 创建检索器和 QA 系统
    print("\n🔍 初始化检索器...")
    retriever = AdvancedRetriever(
        vectorstore=vectorstore,
        search_type=config["retriever"]["search_type"],
        k=config["retriever"]["k"]
    ).get_reranking_retriever(
        top_n=config["reranking"]["top_n"]
    ) if config["reranking"]["enabled"] else AdvancedRetriever(
        vectorstore=vectorstore
    ).get_basic_retriever()
    
    print("\n💬 初始化 QA 系统...")
    qa_system = RAGQuestionAnswering(
        llm=llm,
        retriever=retriever
    )
    
    # 5. 交互式问答
    print("\n" + "="*50)
    print("🎯 个人知识库 RAG 系统已就绪！")
    print("="*50)
    print("输入你的问题，输入 'exit' 退出\n")
    
    chat_history = []
    
    while True:
        question = input("🧑 你: ").strip()
        
        if question.lower() in ["exit", "quit", "q"]:
            print("👋 再见！")
            break
        
        if not question:
            continue
        
        # 回答问题
        result = qa_system.answer(question, chat_history)
        
        # 显示答案
        print(f"\n🤖 AI: {result['answer']}\n")
        
        # 显示来源
        if result["source_documents"]:
            print("📚 参考来源：")
            for i, doc in enumerate(result["source_documents"], 1):
                source = doc.metadata.get("source", "未知来源")
                print(f"  [{i}] {source}")
        
        print()
        
        # 更新对话历史
        chat_history.append((question, result["answer"]))


if __name__ == "__main__":
    main()

CLI 工作流程：

4.2 Web UI（Streamlit）

创建 app.py：

"""
Web UI - 使用 Streamlit 构建可视化界面
"""
import yaml
import streamlit as st
from pathlib import Path
from src.document_loader import DocumentLoader
from src.text_splitter import TextSplitter
from src.embeddings import EmbeddingManager
from src.vector_store import VectorStoreManager
from src.retriever import AdvancedRetriever
from src.qa_chain import RAGQuestionAnswering
from langchain_ollama import ChatOllama


# 页面配置
st.set_page_config(
    page_title="个人知识库 RAG 系统",
    page_icon="📚",
    layout="wide"
)

# 加载配置
@st.cache_resource
def load_config():
    with open("config.yaml", "r", encoding="utf-8") as f:
        return yaml.safe_load(f)

@st.cache_resource
def init_components(config):
    """初始化组件（带缓存）"""
    # Embedding
    embedding_manager = EmbeddingManager(
        model=config["embedding"]["model"],
        base_url=config["embedding"]["base_url"]
    )
    embeddings = embedding_manager.get_embeddings()
    
    # LLM
    llm = ChatOllama(
        model=config["llm"]["model"],
        base_url=config["llm"]["base_url"],
        temperature=config["llm"]["temperature"]
    )
    
    # 向量数据库
    vectorstore_manager = VectorStoreManager(
        embeddings=embeddings,
        persist_directory=config["vectorstore"]["persist_directory"],
        collection_name=config["vectorstore"]["collection_name"]
    )
    
    vectorstore = vectorstore_manager.load_vectorstore()
    
    return llm, vectorstore_manager, vectorstore

# 主界面
def main():
    st.title("📚 个人知识库 RAG 系统")
    
    config = load_config()
    
    # 侧边栏
    with st.sidebar:
        st.header("⚙️ 配置")
        
        # 索引管理
        st.subheader("📦 文档索引")
        if st.button("🔄 重新索引所有文档"):
            with st.spinner("正在索引文档..."):
                loader = DocumentLoader(config["paths"]["data_dir"])
                documents = loader.load_directory()
                
                text_splitter = TextSplitter(
                    chunk_size=config["text_splitter"]["chunk_size"],
                    chunk_overlap=config["text_splitter"]["chunk_overlap"]
                )
                split_docs = text_splitter.split_documents(documents)
                
                _, vectorstore_manager, _ = init_components(config)
                vectorstore_manager.create_vectorstore(split_docs)
                
                st.success(f"✅ 索引完成！共 {len(split_docs)} 个文档块")
                st.experimental_rerun()
        
        # 检索配置
        st.subheader("🔍 检索配置")
        k = st.slider("检索文档数", min_value=1, max_value=10, value=config["retriever"]["k"])
        reranking_enabled = st.checkbox("启用重排序", value=config["reranking"]["enabled"])
        
        st.divider()
        st.caption("Powered by LangChain + Ollama + Chroma")
    
    # 主界面
    tab1, tab2, tab3 = st.tabs(["💬 问答", "📤 上传文档", "ℹ️ 关于"])
    
    with tab1:
        st.header("对话式问答")
        
        # 初始化对话历史
        if "messages" not in st.session_state:
            st.session_state.messages = []
        
        # 显示对话历史
        for message in st.session_state.messages:
            with st.chat_message(message["role"]):
                st.markdown(message["content"])
                if "sources" in message:
                    with st.expander("📚 参考来源"):
                        for src in message["sources"]:
                            st.caption(src)
        
        # 用户输入
        if prompt := st.chat_input("输入你的问题..."):
            # 显示用户消息
            st.session_state.messages.append({"role": "user", "content": prompt})
            with st.chat_message("user"):
                st.markdown(prompt)
            
            # 生成回答
            with st.chat_message("assistant"):
                message_placeholder = st.empty()
                with st.spinner("🤔 思考中..."):
                    # 初始化组件
                    llm, vectorstore_manager, vectorstore = init_components(config)
                    
                    if vectorstore is None:
                        message_placeholder.error("⚠️ 请先索引文档！")
                        return
                    
                    # 创建检索器和 QA 系统
                    retriever = AdvancedRetriever(
                        vectorstore=vectorstore,
                        k=k
                    ).get_reranking_retriever(top_n=3) if reranking_enabled else AdvancedRetriever(
                        vectorstore=vectorstore,
                        k=k
                    ).get_basic_retriever()
                    
                    qa_system = RAGQuestionAnswering(llm=llm, retriever=retriever)
                    
                    # 获取对话历史
                    chat_history = [
                        (m["content"], st.session_state.messages[i+1]["content"])
                        for i, m in enumerate(st.session_state.messages[:-1])
                        if m["role"] == "user" and i+1 < len(st.session_state.messages)
                    ]
                    
                    # 回答问题
                    result = qa_system.answer(prompt, chat_history)
                    
                    # 显示答案
                    message_placeholder.markdown(result["answer"])
                    
                    # 显示来源
                    sources = []
                    if result["source_documents"]:
                        with st.expander("📚 参考来源"):
                            for i, doc in enumerate(result["source_documents"], 1):
                                source = doc.metadata.get("source", "未知来源")
                                st.caption(f"[{i}] {source}")
                                sources.append(source)
                    
                    # 保存到历史
                    st.session_state.messages.append({
                        "role": "assistant",
                        "content": result["answer"],
                        "sources": sources
                    })
    
    with tab2:
        st.header("📤 上传文档")
        st.info("将文档放入 `data/` 目录下对应的子文件夹，然后点击侧边栏的「重新索引」按钮")
        
        # 显示当前文档列表
        data_dir = Path(config["paths"]["data_dir"])
        if data_dir.exists():
            st.subheader("当前已索引的文档：")
            for file_path in data_dir.rglob("*"):
                if file_path.is_file():
                    st.text(f"📄 {file_path.relative_to(data_dir)}")
    
    with tab3:
        st.header("ℹ️ 关于")
        st.markdown("""
        **个人知识库 RAG 系统**
        
        技术栈：
        - 🦜 LangChain - RAG 编排框架
        - 🦙 Ollama - 本地 LLM 运行时
        - 🎨 Chroma - 向量数据库
        - 🎈 Streamlit - Web UI
        
        功能特性：
        - ✅ 多格式文档支持（PDF、Word、Markdown、TXT）
        - ✅ 语义搜索（理解意图，不是关键词匹配）
        - ✅ 对话式问答（带上下文记忆）
        - ✅ 重排序优化（提高检索精准度）
        - ✅ 本地部署（数据不出本地）
        """)

if __name__ == "__main__":
    main()

Streamlit Web UI 界面预览：

五、运行与测试

5.1 首次运行（索引文档）

操作步骤：

# 1. 准备文档
mkdir -p data/{pdfs,markdown,web}
# 将你的 PDF、Markdown 等文档放入对应目录

# 2. 启动 Ollama 服务
ollama serve &

# 3. 运行 CLI 版本
python cli.py
# 首次运行会自动索引文档

# 4. 运行 Web UI 版本
streamlit run app.py
# 访问 http://localhost:8501

5.2 测试问答

示例对话：

🧑 你: 公司年假政策是什么？

🤖 AI: 根据公司员工手册第 3 章第 2 节的规定：
- 入职满 1 年：5 天年假
- 入职满 3 年：10 天年假  
- 入职满 5 年：15 天年假

[来源: employee_handbook.pdf, 第 12 页]

六、生产级优化

6.1 分块策略优化

# 针对不同文档类型使用不同分块策略

# PDF 文档（按段落分块）
pdf_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=100,
    separators=["\n\n", "\n", ".", "。", " ", ""]
)

# Markdown 文档（按标题分块）
md_splitter = MarkdownTextSplitter(
    chunk_size=1024,
    chunk_overlap=100
)

# 代码文件（按函数/类分块）
code_splitter = PythonCodeTextSplitter(
    chunk_size=1024,
    chunk_overlap=200
)

分块策略选择树：

6.2 Hybrid 检索（向量 + BM25）

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# 向量检索器
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# BM25 关键词检索器
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# 混合检索器（加权融合）
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.7, 0.3]  # 向量检索权重 0.7，BM25 权重 0.3
)

Hybrid 检索原理：

6.3 查询重写

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# 查询重写提示词
rewrite_prompt = PromptTemplate(
    template="""用户原始查询：{query}

请将其重写为更适合向量检索的查询（保持原意，补充同义词）：""",
    input_variables=["query"]
)

rewrite_chain = LLMChain(llm=llm, prompt=rewrite_prompt)

# 使用
rewritten_query = rewrite_chain.run(query)
results = retriever.get_relevant_documents(rewritten_query)

查询重写示例：

6.4 评估体系

from langchain.evaluation import QAEvaluator, EmbeddingDistanceEvalChain

# 1. 答案相关性评估
qa_evaluator = QAEvaluator()

# 2. 检索准确率评估（需要标注数据集）
ground_truth = {
    "question": "公司年假政策",
    "expected_answer": "入职满 1 年 5 天，满 3 年 10 天...",
    "relevant_docs": ["employee_handbook.pdf"]
}

# 3. 向量相似度评估
embedding_eval = EmbeddingDistanceEvalChain(embeddings=embeddings)

评估指标详解：

七、部署到生产环境

7.1 使用 PostgreSQL + PGVector

# 安装 PGVector
pip install pgvector langchain-postgres

# 创建向量表
CREATE EXTENSION IF NOT EXISTS vector;

# 使用 PGVector 替代 Chroma
from langchain.vectorstores.pgvector import PGVector

vectorstore = PGVector(
    embeddings=embeddings,
    collection_name="knowledge_base",
    connection_string="postgresql://user:password@localhost:5432/rag_db"
)

PGVector 优势：

7.2 API 服务（FastAPI）

创建 api.py：

"""
REST API - 部署为微服务
"""
from fastapi import FastAPI
from pydantic import BaseModel
from src.qa_chain import RAGQuestionAnswering
from src.vector_store import VectorStoreManager
from src.embeddings import EmbeddingManager
from langchain_ollama import ChatOllama

app = FastAPI(title="RAG API")

# 请求模型
class QueryRequest(BaseModel):
    question: str
    chat_history: list = []

class QueryResponse(BaseModel):
    answer: str
    sources: list

@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    """问答接口"""
    # 初始化组件（实际部署中应缓存）
    config = load_config()
    embeddings = EmbeddingManager(...).get_embeddings()
    vectorstore = VectorStoreManager(...).load_vectorstore()
    retriever = AdvancedRetriever(vectorstore=vectorstore).get_basic_retriever()
    qa_system = RAGQuestionAnswering(llm=ChatOllama(...), retriever=retriever)
    
    # 回答问题
    result = qa_system.answer(request.question, request.chat_history)
    
    return {
        "answer": result["answer"],
        "sources": [doc.metadata.get("source") for doc in result["source_documents"]]
    }

# 启动：uvicorn api:app --host 0.0.0.0 --port 8000

FastAPI 部署架构：

八、常见问题 FAQ

Q1：检索结果不相关怎么办？

尝试以下优化：

1. 调整分块大小：太大 → 噪声多，太小 → 丢失上下文
2. 启用重排序：reranking.enabled: true
3. 使用 Hybrid 检索：向量 + BM25 结合
4. 优化 Embedding 模型：换用 bge-large-zh-v1.5（中文场景）

Q2：回答不准确或有幻觉？

优化提示词：

如果检索到的上下文无法回答问题，明确说"根据现有文档无法回答"
不要编造答案！

Q3：索引速度太慢？

优化建议：

1. 使用批量 Embedding（embed_documents 而非循环调用 embed_query）
2. 换用更快的 Embedding 模型（all-MiniLM-L6-v2）
3. 使用 GPU 加速（需安装 sentence-transformers[gpu]）

Q4：如何支持更大规模的知识库？

迁移到生产级向量数据库：

• 中型（10万+ 文档）：PGVector
• 大型（百万+ 文档）：Pinecone / Milvus

Q5：如何保护隐私？

本文方案已经是完全本地部署：

• Ollama 本地运行，无需上传文档到云端
• Chroma 本地存储，数据不出本地

额外建议：

• 对敏感文档加密存储
• 添加用户认证（FastAPI 中间件）
• 定期备份向量数据库

总结

下一步行动

资源链接

• 📚 LangChain 官方文档：https://python.langchain.com/docs/
• 🦙 Ollama 官网：https://ollama.com/
• 🎨 Chroma 文档：https://docs.trychroma.com/
• 📖 RAG 优化技术论文：https://arxiv.org/abs/2005.11401

本文基于 LangChain 0.3+、Ollama 0.5+、Chroma 0.5+ 编写。如有问题欢迎评论区讨论！