实操审校说明

本文内容属于可实操教程。RAG 的核心流程“文档解析 → 分块 → 向量化 → 检索 → 拼接上下文 → 交给大模型回答”是成熟方案。教程中的实现可以跑通一个最小知识库 Demo，但不同模型、Embedding 服务和向量数据库的接口会有差异。

建议测试环境

实操前置条件

1. 能创建 Python 虚拟环境。
2. 能安装 chromadb、langchain 等依赖。
3. 准备 3-5 篇测试文档，内容越具体越容易验证。
4. 如果使用在线模型，需要提前准备 API Key。

可验证结果

完成教程后，至少应该能验证以下结果：

• 文档被成功读取并切分。
• 向量数据库里能查询到已入库文本。
• 输入问题后能返回相关文档片段。
• 模型回答能引用或基于检索到的内容，而不是凭空编造。

版本风险提示

RAG 最容易出问题的是依赖版本、Embedding 模型维度不一致、PDF 解析质量和 API Key 配置。新手建议先用少量 Markdown/TXT 文档跑通，再升级到 PDF、网页和多用户知识库。

官方参考：

• Chroma Getting Started：https://docs.trychroma.com/docs/overview/getting-started
• Chroma Query and Get：https://docs.trychroma.com/docs/querying-collections/query-and-get
• Chroma Clients：https://docs.trychroma.com/docs/run-chroma/clients

RAG 知识库搭建教程：让 AI 读懂你的私有文档

📅 更新时间：2026 年 6 月

👤 适合人群：有 Python 基础的开发者、AI 应用开发新手

⏱ 预计阅读时间：40 分钟

一、RAG 是什么？

RAG（Retrieval-Augmented Generation，检索增强生成） 是目前 AI 应用开发最核心的技术之一。

🤔 问题背景：为什么需要 RAG？

AI 大模型（GPT、Claude、DeepSeek）的两大局限：

1. 知识截止日期：不知道最新的信息（比如今天的新闻）
2. 不知道你的私有数据：不了解你公司的内部文档、产品手册

RAG 就是解决这两个问题的方案。

💡 RAG 的核心思路

传统问答：
用户提问 → AI 凭记忆回答（可能过时、可能不知道）

RAG 问答：
用户提问 → 从文档库里找到相关内容 → 把内容给 AI → AI 基于真实内容回答

🏢 应用场景

二、RAG 工作原理

阶段一：文档入库（离线处理）

原始文档（PDF/Word/TXT）
    ↓ 文档解析
文本内容
    ↓ 文本分块（Chunk）
小段文本（500-1000 字/块）
    ↓ 向量化（Embedding）
每块文本的数字向量
    ↓ 存入向量数据库
向量数据库（Chroma/Faiss/Milvus）

阶段二：检索生成（在线查询）

用户提问
    ↓ 将问题向量化
问题向量
    ↓ 在向量数据库中相似度搜索
找到最相关的 3-5 个文档块
    ↓ 拼接成 Context 发给 LLM
LLM（DeepSeek/GPT/Claude）
    ↓ 基于 Context 生成回答
最终答案（有据可查）

为什么用"向量相似度"而不是关键词搜索？

• 关键词搜索："如何退款" 找不到 "退款流程说明"（词不同）
• 向量搜索：理解语义，"如何退款" 和 "退款流程说明" 的向量很相近 ✅

三、环境准备

3.1 安装依赖

# 创建虚拟环境
python -m venv rag-env
source rag-env/bin/activate  # Windows: rag-env\Scripts\activate

# 安装必要库
pip install langchain langchain-community
pip install chromadb                    # 向量数据库
pip install sentence-transformers       # 本地 Embedding 模型
pip install pypdf                       # PDF 解析
pip install openai                      # 调用 GPT/DeepSeek
pip install python-dotenv               # 管理 API Key

3.2 准备 API Key

创建 .env 文件（不要提交到 Git！）：

# 选一个你有的 API
OPENAI_API_KEY=sk-xxx
DEEPSEEK_API_KEY=sk-xxx

四、从零搭建一个 RAG 系统

4.1 项目结构

rag-project/
├── documents/        ← 放你的文档（PDF/TXT/MD）
│   ├── manual.pdf
│   └── policy.txt
├── vectorstore/      ← 向量数据库存储（自动生成）
├── .env              ← API Key（不提交！）
├── ingest.py         ← 文档入库脚本
└── chat.py           ← 问答脚本

4.2 文档入库脚本（ingest.py）

"""
ingest.py - 将文档处理并存入向量数据库
运行：python ingest.py
"""

import os
from pathlib import Path
from dotenv import load_dotenv
from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    DirectoryLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma

load_dotenv()

# 配置
DOCUMENTS_DIR = "./documents"
VECTORSTORE_DIR = "./vectorstore"
# 使用免费的本地 Embedding 模型（无需 API Key）
EMBEDDING_MODEL = "BAAI/bge-small-zh-v1.5"  # 中文 Embedding 模型

def load_documents():
    """加载 documents 文件夹中的所有文档"""
    documents = []
    doc_dir = Path(DOCUMENTS_DIR)

    # 加载 PDF 文件
    for pdf_file in doc_dir.glob("*.pdf"):
        print(f"📄 加载 PDF：{pdf_file.name}")
        loader = PyPDFLoader(str(pdf_file))
        documents.extend(loader.load())

    # 加载 TXT 文件
    for txt_file in doc_dir.glob("*.txt"):
        print(f"📄 加载 TXT：{txt_file.name}")
        loader = TextLoader(str(txt_file), encoding="utf-8")
        documents.extend(loader.load())

    print(f"✅ 共加载 {len(documents)} 个文档片段")
    return documents

def split_documents(documents):
    """将文档切分成小块"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,      # 每块 500 字
        chunk_overlap=50,    # 相邻块重叠 50 字（保持上下文连贯）
        length_function=len,
    )
    chunks = splitter.split_documents(documents)
    print(f"✅ 切分为 {len(chunks)} 个文本块")
    return chunks

def create_vectorstore(chunks):
    """创建向量数据库"""
    print("🔄 正在加载 Embedding 模型（首次运行会下载，需要几分钟）...")

    embeddings = HuggingFaceEmbeddings(
        model_name=EMBEDDING_MODEL,
        model_kwargs={"device": "cpu"},
        encode_kwargs={"normalize_embeddings": True}
    )

    print("🔄 正在生成向量并存入数据库...")
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=VECTORSTORE_DIR
    )

    print(f"✅ 向量数据库已保存到 {VECTORSTORE_DIR}")
    return vectorstore

if __name__ == "__main__":
    print("=== 开始处理文档 ===")
    docs = load_documents()
    chunks = split_documents(docs)
    vectorstore = create_vectorstore(chunks)
    print("\n🎉 文档入库完成！现在可以运行 chat.py 开始问答了。")

4.3 问答脚本（chat.py）

"""
chat.py - 基于知识库的智能问答
运行：python chat.py
"""

import os
from dotenv import load_dotenv
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

load_dotenv()

# 配置（和 ingest.py 保持一致）
VECTORSTORE_DIR = "./vectorstore"
EMBEDDING_MODEL = "BAAI/bge-small-zh-v1.5"

# 自定义 Prompt 模板
PROMPT_TEMPLATE = """你是一个专业的知识库助手。请严格根据以下参考资料回答用户的问题。
如果参考资料中没有相关内容，请如实说"我在知识库中没有找到相关信息"，不要编造答案。

参考资料：
{context}

用户问题：{question}

请用中文给出清晰、准确的回答："""

def load_vectorstore():
    """加载已存在的向量数据库"""
    embeddings = HuggingFaceEmbeddings(
        model_name=EMBEDDING_MODEL,
        model_kwargs={"device": "cpu"},
        encode_kwargs={"normalize_embeddings": True}
    )

    vectorstore = Chroma(
        persist_directory=VECTORSTORE_DIR,
        embedding_function=embeddings
    )
    return vectorstore

def create_qa_chain(vectorstore):
    """创建问答链"""
    # 使用 DeepSeek（便宜）或 GPT
    # 切换为 DeepSeek：修改 base_url 和 model
    llm = ChatOpenAI(
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
        model="deepseek-v4-flash",
        temperature=0.1  # 低温度 = 更准确，更少发挥
    )

    prompt = PromptTemplate(
        template=PROMPT_TEMPLATE,
        input_variables=["context", "question"]
    )

    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=vectorstore.as_retriever(
            search_kwargs={"k": 3}  # 检索最相关的 3 个文档块
        ),
        chain_type_kwargs={"prompt": prompt},
        return_source_documents=True  # 返回来源，方便验证
    )

    return qa_chain

def main():
    print("📚 加载知识库...")
    vectorstore = load_vectorstore()
    qa_chain = create_qa_chain(vectorstore)

    print("\n✅ 知识库助手已就绪！输入 'quit' 退出。\n")
    print("=" * 50)

    while True:
        question = input("\n🙋 你的问题：").strip()

        if question.lower() in ["quit", "exit", "退出"]:
            print("👋 再见！")
            break

        if not question:
            continue

        print("\n🤔 正在检索知识库...")
        result = qa_chain.invoke({"query": question})

        print(f"\n💡 回答：\n{result['result']}")

        # 显示来源文档
        if result.get("source_documents"):
            print("\n📌 来源文档：")
            for i, doc in enumerate(result["source_documents"], 1):
                source = doc.metadata.get("source", "未知")
                page = doc.metadata.get("page", "")
                print(f"  [{i}] {source} {'第' + str(page+1) + '页' if page != '' else ''}")

if __name__ == "__main__":
    main()

4.4 运行！

# 第一步：把你的文档放进 documents/ 文件夹

# 第二步：处理文档
python ingest.py

# 第三步：开始问答
python chat.py

五、进阶优化技巧

5.1 提高检索质量

# 使用混合检索（关键词 + 语义）
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

bm25_retriever = BM25Retriever.from_documents(chunks, k=3)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 融合两种检索结果
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]
)

5.2 文档重排序（Reranking）

# 安装：pip install sentence-transformers
from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-base")

def rerank_documents(query, documents, top_k=3):
    """对检索结果重排序，提高精准度"""
    pairs = [(query, doc.page_content) for doc in documents]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, documents), reverse=True)
    return [doc for _, doc in ranked[:top_k]]

5.3 添加 Web UI（Streamlit）

pip install streamlit

创建 app.py：

import streamlit as st
# 把上面的 chat.py 逻辑封装进来

st.title("📚 我的 AI 知识库助手")

if "messages" not in st.session_state:
    st.session_state.messages = []

for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.write(msg["content"])

if question := st.chat_input("问我任何关于知识库的问题..."):
    st.session_state.messages.append({"role": "user", "content": question})
    with st.chat_message("user"):
        st.write(question)

    with st.chat_message("assistant"):
        with st.spinner("检索知识库..."):
            # 调用 qa_chain
            result = qa_chain.invoke({"query": question})
            answer = result["result"]
        st.write(answer)

    st.session_state.messages.append({"role": "assistant", "content": answer})

streamlit run app.py
# 自动打开浏览器，就有好看的聊天界面了！

六、常见问题

Q：Embedding 模型下载太慢？

使用镜像：

import os
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"

Q：PDF 中文乱码？

pip install pdfminer.six  # 备用解析器

Q：回答质量差？

调整参数：

• 增大 chunk_size（从 500 → 1000）
• 增大 k（检索更多文档，从 3 → 5）
• 修改 Prompt，让 AI 更严格遵守文档内容

总结

🎯 下一步：参考 Datawhale 的 All-in-RAG 开源课程深入学习！

*参考：Datawhale All-in-RAG | CSDN RAG 完整实战指南（2026.5）| DeepToAI RAG 系列教程*