本文手把手带你从零搭建**生产级个人私有知识库 RAG 系统**，使用 Python + LangChain + Chroma 实现，支持 PDF / Word / TXT 本地文档读取，解决大模型**知识过时、易幻觉、无法读取私有数据**三大痛点。

全文提供**完整可运行代码**，包含缓存机制、中文适配、异常处理、MMR 检索优化、批量文件夹读取等实战特性，新手也能一键跑通，快速拥有属于自己的本地 AI 问答系统。

上一篇《大模型入门与原理》里我提到，RAG（检索增强生成）是解决大模型 "知识过时、易幻觉、不懂私有数据" 的核心方案。

今天就带大家**从零到一搭建一个可落地、可直接用于生产的个人私有知识库**。不用复杂环境，不用高深算法，纯 Python 实现，看完就能跑通，让你的 AI 真正读懂你的本地（PDF/Word/TXT）。

本文特色：代码可直接运行、生产级优化、中文友好、支持缓存、支持批量文档、带完整异常处理与报错解决方案。

一、RAG 是什么？为什么需要它？

1. RAG 核心定义

RAG（Retrieval-Augmented Generation）：检索增强生成，简单说就是 「先从你的私有数据里找答案，再让大模型基于找到的内容回答」。

2. 解决的核心问题

问题	原生大模型	RAG 方案
私有数据	不知道你的本地文档、公司资料	✅ 精准读取本地知识
知识时效	训练数据截止到某个时间点	✅ 实时更新知识库
幻觉问题	容易"一本正经地胡说八道"	✅ 只基于检索内容回答
落地成本	需要重新训练/微调	✅ 无需训练，即插即用

3. RAG 核心流程

本地文档 → 文档加载与拆分 → 文本向量化（Embedding） → 存入向量数据库

                                                                                                          ↓

用户提问 → 问题向量化 → 向量数据库检索相关内容 → 拼接上下文+问题发给大模型 → 大模型生成精准回答

二、环境准备

1. 安装依赖包

先确保你的电脑装了 Python 3.8+，然后创建 requirements.txt 文件：

# requirements.txt
langchain==0.1.0
langchain-community==0.0.20
sentence-transformers==2.2.2
chromadb==0.4.22
pypdf==3.17.4
python-docx==1.1.0
openai==1.6.1

执行安装命令：

pip install -r requirements.txt

2. 依赖说明（新手必看）

依赖	作用	说明
langchain	RAG 开发的核心框架	封装了全套流程，开箱即用
sentence-transformers	开源的文本向量化模型	本地运行，无需联网
chromadb	轻量级本地向量数据库	无需部署，自动持久化
pypdf/python-docx	解析 PDF/Word 文档	支持常见文档格式
openai	大模型 API（可选）	可替换为本地模型

3. 准备测试文档

找 1-2 个本地文档（PDF/Word/TXT 都可以，比如你的学习笔记、技术文档），记住文件路径（比如 ./docs/我的笔记.pdf）。

三、完整代码实现

# -*- coding: utf-8 -*-
"""
生产级个人私有知识库 RAG 系统
支持：PDF / Word / TXT、向量库缓存、MMR 检索、中文优化、异常处理、编码兼容
"""
import os
from typing import List, Optional, Tuple
from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import SentenceTransformerEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.schema.document import Document


class RAGConfig:
    """统一配置项"""
    chunk_size = 500
    chunk_overlap = 50
    k_retrieval = 3
    retrieval_score_threshold = 0.5
    persist_directory = "./chroma_db"
    embedding_model = "all-MiniLM-L6-v2"
    llm_model = "gpt-3.5-turbo"
    temperature = 0
    encoding = "utf-8"


def load_documents(file_path: str) -> Optional[List[Document]]:
    if not os.path.exists(file_path):
        print(f"文件不存在：{file_path}")
        return None

    try:
        if file_path.endswith('.pdf'):
            print("检测到 PDF 文件")
            loader = PyPDFLoader(file_path)
        elif file_path.endswith('.docx'):
            print("检测到 Word 文件")
            loader = Docx2txtLoader(file_path)
        elif file_path.endswith('.txt'):
            print("检测到 TXT 文件")
            try:
                loader = TextLoader(file_path, encoding="utf-8")
            except UnicodeDecodeError:
                loader = TextLoader(file_path, encoding="gbk")
        else:
            print("不支持的文件格式")
            print("支持格式：.pdf, .docx, .txt")
            return None

        documents = loader.load()
        print(f"成功加载文档：{file_path}")
        total_len = sum(len(d.page_content) for d in documents)
        print(f"文档信息：共 {len(documents)} 页/段，总字符数：{total_len}")
        return documents

    except Exception as e:
        print(f"加载文档失败：{str(e)}")
        return None


def split_documents(documents: List[Document]) -> List[Document]:
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=RAGConfig.chunk_size,
        chunk_overlap=RAGConfig.chunk_overlap,
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", "、", " ", ""],
        length_function=len
    )
    splits = text_splitter.split_documents(documents)
    print("文本拆分完成")
    print(f"拆分信息：共拆分成 {len(splits)} 个片段")
    return splits


def get_or_create_vector_db(splits: List[Document]) -> Tuple[Chroma, SentenceTransformerEmbeddings]:
    print(f"\n初始化向量化模型：{RAGConfig.embedding_model}")

    embeddings = SentenceTransformerEmbeddings(model_name=RAGConfig.embedding_model)

    if os.path.exists(RAGConfig.persist_directory) and os.listdir(RAGConfig.persist_directory):
        print(f"检测到已存在的向量库，直接加载：{RAGConfig.persist_directory}")
        vectordb = Chroma(
            persist_directory=RAGConfig.persist_directory,
            embedding_function=embeddings
        )
        print(f"向量库信息：包含 {vectordb._collection.count()} 个向量")
    else:
        print(f"未检测到向量库，新建并保存：{RAGConfig.persist_directory}")
        vectordb = Chroma.from_documents(
            documents=splits,
            embedding=embeddings,
            persist_directory=RAGConfig.persist_directory
        )
        vectordb.persist()
        print(f"向量库信息：包含 {vectordb._collection.count()} 个向量")

    return vectordb, embeddings


def create_retriever(vectordb: Chroma):
    retriever = vectordb.as_retriever(
        search_type="mmr",
        search_kwargs={
            "k": RAGConfig.k_retrieval,
            "fetch_k": RAGConfig.k_retrieval * 3,
            "lambda_mult": 0.7
        }
    )
    print(f"检索器创建完成，检索策略：MMR，召回数量：{RAGConfig.k_retrieval}")
    return retriever


def build_rag_chain(retriever):
    print("\n初始化大模型...")
    llm = ChatOpenAI(
        model_name=RAGConfig.llm_model,
        temperature=RAGConfig.temperature
    )

    prompt_template = """
你是一个专业的问答助手，请严格基于以下【上下文内容】回答用户的问题。

【约束条件】
1. 如果上下文中包含答案，请用简洁准确的语言回答，并指出信息在文档中的位置。
2. 如果上下文中没有相关信息，请直接回答：根据现有知识库，无法回答该问题。
3. 严禁编造或使用外部知识。
4. 回答保持客观，不添加个人观点。

【上下文内容】
{context}

【用户问题】
{question}

【回答】：
"""
    prompt = PromptTemplate(template=prompt_template, input_variables=["context", "question"])

    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True,
        chain_type_kwargs={"prompt": prompt}
    )
    print("RAG问答链构建完成")
    return qa_chain


def rag_qa(qa_chain, question: str):
    print(f"\n正在检索：'{question}'")
    result = qa_chain.invoke({"query": question})
    answer = result["result"]
    sources = result["source_documents"]

    print("\n" + "="*60)
    print("回答：")
    print("-"*60)
    print(answer)
    print("\n答案来源：")
    print("-"*60)
    for i, doc in enumerate(sources, 1):
        source_info = f"{i}. {doc.metadata.get('source', '未知来源')}"
        if 'page' in doc.metadata:
            source_info += f" (第 {doc.metadata['page'] + 1} 页)"
        preview = doc.page_content[:120] + "..." if len(doc.page_content) > 120 else doc.page_content
        print(source_info)
        print(f"   预览：{preview}")
    print("="*60)


def load_folder_documents(folder_path: str) -> List[Document]:
    if not os.path.exists(folder_path):
        print(f"文件夹不存在：{folder_path}")
        return []

    all_docs = []
    supported = ('.pdf', '.docx', '.txt')
    print(f"\n扫描文件夹：{folder_path}")
    for file in os.listdir(folder_path):
        fp = os.path.join(folder_path, file)
        if os.path.isfile(fp) and fp.lower().endswith(supported):
            print(f"发现文档：{file}")
            docs = load_documents(fp)
            if docs:
                all_docs.extend(docs)
    print(f"\n文件夹加载完成，共加载 {len(all_docs)} 页/段")
    return all_docs


def main():
    print("个人私有知识库 RAG 系统 v2.0")
    print("="*60)
    print()

    DOC_PATH = "./docs/RAG教程.pdf"
    documents = load_documents(DOC_PATH)
    # documents = load_folder_documents("./docs")

    if not documents:
        print("文档加载失败，程序退出")
        return

    splits = split_documents(documents)
    vectordb, embeddings = get_or_create_vector_db(splits)
    retriever = create_retriever(vectordb)
    qa_chain = build_rag_chain(retriever)

    print("\n问答交互已启动（输入 q 退出）")
    while True:
        q = input("\n请输入你的问题：")
        if q.lower() in ['q', 'quit', 'exit']:
            print("感谢使用，再见！")
            break
        if not q.strip():
            print("问题不能为空，请重新输入")
            continue
        rag_qa(qa_chain, q)


if __name__ == "__main__":
    main()

四、代码运行步骤 & 注意事项

1. 运行前准备

修改文档路径：将代码中的 DOC_PATH 替换为你的实际文档路径

设置 API Key（如果用OpenAI）：

# Windows
set OPENAI_API_KEY=你的key
# Mac/Linux
export OPENAI_API_KEY=你的key

首次运行：会自动下载向量化模型（约100MB），请耐心等待

2. 运行效果示例

个人私有知识库 RAG 系统 v2.0
============================================================

文档路径：./docs/RAG教程.pdf
检测到 PDF 文件
成功加载文档：./docs/RAG教程.pdf
文档信息：共 5 页/段，总字符数：3250

文本拆分完成
拆分信息：共拆分成 12 个片段

初始化向量化模型：all-MiniLM-L6-v2
检测到已存在的向量库，直接加载：./chroma_db
向量库信息：包含 12 个向量
检索器创建完成，检索策略：MMR，召回数量：3

初始化大模型...
RAG问答链构建完成

问答交互已启动（输入 q 退出）

请输入你的问题：RAG的核心步骤有哪些？

正在检索：'RAG的核心步骤有哪些？'

============================================================
回答：
------------------------------------------------------------
RAG的核心步骤包括：1. 文档加载与拆分；2. 文本向量化；3. 存入向量数据库；4. 问题向量化并检索相关内容；5. 拼接上下文和问题发给大模型；6. 大模型基于检索内容生成回答。

答案来源：
------------------------------------------------------------
1. ./docs/RAG教程.pdf (第 2 页)
   预览：RAG的核心流程：首先需要加载本地文档，然后将文档拆分成小片段，接着通过向量化模型将文本转换为向量...

2. ./docs/RAG教程.pdf (第 3 页)
   预览：在检索阶段，系统会将用户的问题也向量化，然后在向量数据库中查找最相似的片段...

3. ./docs/RAG教程.pdf (第 1 页)
   预览：RAG（Retrieval-Augmented Generation）检索增强生成，是一种结合信息检索和文本生成的技术...
============================================================

3. 常见踩坑点及解决方案

问题	可能原因	解决方案
❌ 找不到文件	路径错误或包含中文空格	使用绝对路径，或对路径进行转义
❌ 模型下载慢	网络问题	手动下载到本地，或使用国内镜像
❌ OpenAI API 报错	API Key无效/网络不通	检查API Key，确认能访问OpenAI
❌ 回答"未找到答案"	文档无关/检索参数不当	调整chunk_size，增大k值
❌ 内存不足	文档过大	减小chunk_size，分批处理

五、进阶优化（新手可选）

1. 替换成本地大模型（免API，全私有化）

# 替换 build_rag_chain 函数中的LLM部分

# 方案1：使用 ChatGLM（需要本地部署）
from langchain_community.llms import ChatGLM
llm = ChatGLM(
    endpoint_url="http://127.0.0.1:8000",  # 本地模型的API地址
    max_token=8192,
    temperature=0.1
)

# 方案2：使用 Ollama（推荐，最简单）
from langchain_community.llms import Ollama
llm = Ollama(
    model="qwen:7b",  # 或 llama2, mistral 等
    temperature=0
)

# 方案3：使用 HuggingFace 本地模型
from langchain_community.llms import HuggingFacePipeline
# 需要先加载模型到内存

2. 中文场景优化

# 修改 RAGConfig
class RAGConfig:
    # 使用中文向量模型
    embedding_model = "BAAI/bge-large-zh"  # 中文效果更好
    
    # 调整分隔符（中文友好）
    separators = ["\n\n", "\n", "。", "！", "？", "；", "，", "、"]
    
    # 调整chunk_size（中文一个字算一个字符）
    chunk_size = 300  # 约150-200个汉字

3. Web界面封装（使用Streamlit）

# web_app.py
import streamlit as st
from rag_knowledge_base import *

st.title("个人私有知识库 RAG 系统")

# 文件上传
uploaded_file = st.file_uploader("上传文档", type=['pdf', 'docx', 'txt'])

if uploaded_file:
    # 保存临时文件并处理
    with open(f"./temp/{uploaded_file.name}", "wb") as f:
        f.write(uploaded_file.getbuffer())
    
    # 加载文档...
    
# 问答界面
question = st.text_input("请输入你的问题")
if question:
    answer, sources = rag_qa(qa_chain, question)
    st.write("回答：", answer)
    with st.expander("查看来源"):
        for source in sources:
            st.write(source)

4. 检索效果优化参数调优指南

参数	作用	推荐值	调整说明
chunk_size	文本片段大小	300-500	片段越小越精准，但可能丢失上下文
chunk_overlap	片段重叠	10%-20%	保证边界信息不丢失
k	召回数量	3-5	越多信息越全，但可能引入噪声
相似度阈值	过滤低相关结果	0.5-0.7	越高越精准，但可能漏掉答案

六、总结 & 延伸

1. 核心收获

通过这篇教程，你掌握了：

✅ RAG 的核心逻辑和落地流程

✅ 用 LangChain+Chroma 搭建私有知识库的完整代码

✅ 生产级优化技巧（缓存、异常处理、中文适配）

✅ 解决大模型幻觉、私有数据问答的核心方案

2. 下一步学习方向

入门阶段：

学习向量数据库进阶用法（Milvus、Pinecone、Qdrant）

尝试大模型本地部署 + RAG 全私有化方案

把 RAG 封装成 Web 界面（用 Streamlit/Flask）

进阶阶段：

实现多文档混合检索（PDF+网页+数据库）

引入重排序（Rerank）提升检索精度

添加对话历史，实现多轮问答

对接企业级数据源（Confluence、Notion、钉钉文档）

专家阶段：

混合检索（向量+关键词）提高召回率

查询理解与改写

知识库自动更新机制

性能优化与分布式部署

3. 资源推荐

📚 LangChain 官方文档：https://python.langchain.com/

📚 ChromaDB 文档：https://docs.trychroma.com/

📚 中文向量模型榜单：https://huggingface.co/spaces/mteb/leaderboard

七、下期预告

下一篇我会写：《 RAG 个人知识库 v3.0：零成本接入本地大模型，彻底告别 OpenAI 》