基于 RAG + LangChain 搭建企业级私有知识库问答系统（2026 实战版）

用户10354340

发布于 2026-08-15 20:59:32

2280

💡 作者按：这是我在多个企业知识库项目落地后的实战总结。RAG 看似简单——"文档切片、向量化、检索、生成"，但要真正达到生产可用，里面的坑比想象中深得多。本文给出完整可运行的代码，并穿插十年开发经验中总结的架构决策与避坑指南。

一、为什么企业要自建 RAG 知识库

通用大模型有两个致命问题：知识滞后和幻觉。企业内部的制度文档、技术手册、客户资料，ChatGPT/DeepSeek 统统不知道。RAG（Retrieval-Augmented Generation，检索增强生成）的核心思路是：

不让大模型只依赖训练知识，而是回答前先从企业知识库检索相关资料，再基于真实资料生成答案。

一个真正能上线的 RAG 系统，远不止"文档切块+向量化+调 LLM"这么简单，还要处理：文档解析与版本更新、关键词与向量混合检索、权限过滤、结果重排、引用来源、资料不足时拒答、效果评测与监控。

本文带你从零搭建一套本地可运行、生产可扩展的 RAG 知识库系统。

二、系统架构

┌─────────────┐   ┌──────────────┐   ┌─────────────┐
│ 文档上传     │ → │ 文本分割      │ → │ 向量化存储   │
│ PDF/Word/MD  │   │ Chunk Split  │   │ ChromaDB    │
└─────────────┘   └──────────────┘   └─────────────┘
                                                ↓
┌─────────────┐   ┌──────────────┐   ┌─────────────┐
│ 最终回答     │ ← │ DeepSeek/LLM │ ← │ 向量检索     │
│ 含来源引用   │   │ 生成回答      │   │ Top-K 结果  │
└─────────────┘   └──────────────┘   └─────────────┘

技术栈选型（2026 主流组合）：

编排框架：LangChain 0.3.x
向量数据库：ChromaDB（开发）/ Milvus（生产）
Embedding：BAAI/bge-small-zh-v1.5（中文效果好的本地小模型）
LLM：DeepSeek（OpenAI 兼容接口）或 Ollama 本地模型

三、环境准备

pip install langchain langchain-community langchain-openai
pip install chromadb
pip install pypdf python-docx
pip install sentence-transformers

四、核心代码实战

4.1 文档加载与分割

大模型有上下文窗口限制，必须把长文档切成小块（Chunk）。

from langchain.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoaderfrom langchain.text_splitter import RecursiveCharacterTextSplitterimport osdef load_documents(file_path: str):    """根据文件类型加载文档"""
    ext = os.path.splitext(file_path)[1].lower()    if ext == '.pdf':
        loader = PyPDFLoader(file_path)    elif ext in ['.docx', '.doc']:
        loader = Docx2txtLoader(file_path)    elif ext in ['.txt', '.md']:
        loader = TextLoader(file_path, encoding='utf-8')    else:        raise ValueError(f"不支持的文件类型: {ext}")    return loader.load()def split_documents(documents, chunk_size=500, chunk_overlap=50):    """将文档切割成小块"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "!", "?", " ", ""],
        length_function=len
    )    return splitter.split_documents(documents)# 使用示例docs = load_documents("company_manual.pdf")
chunks = split_documents(docs)print(f"共切割为 {len(chunks)} 个文本块")

⚠️ 十年经验提示：chunk_size 不是越小越精准。太小会切断语义完整性，太大则检索精度下降。中文场景建议 500-800 字符，overlap 取 10%-20%。

4.2 向量化与存储

使用本地 Embedding 模型（免费，不调用 API）：

from langchain_community.vectorstores import Chromafrom langchain_community.embeddings import HuggingFaceEmbeddingsdef create_vector_store(chunks, persist_dir="./chroma_db"):    """创建向量数据库"""
    embeddings = HuggingFaceEmbeddings(
        model_name="BAAI/bge-small-zh-v1.5",
        model_kwargs={'device': 'cpu'},
        encode_kwargs={'normalize_embeddings': True}
    )
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir
    )
    vectorstore.persist()    print(f"向量库已保存至 {persist_dir}")    return vectorstore

4.3 接入 DeepSeek 大模型

DeepSeek 兼容 OpenAI SDK 格式，配置非常简单：

from langchain_openai import ChatOpenAIfrom langchain.chains import RetrievalQAdef build_qa_chain(vectorstore, api_key: str):    """构建检索问答链"""
    llm = ChatOpenAI(
        model_name="deepseek-chat",
        openai_api_key=api_key,
        openai_api_base="https://api.deepseek.com/v1",
        temperature=0.1
    )
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
        return_source_documents=True
    )    return qa_chain

4.4 完整流水线整合

把上述模块串成端到端管线：

import osdef main():    # 1. 加载并分割文档
    docs = load_documents("company_manual.pdf")
    chunks = split_documents(docs, chunk_size=500, chunk_overlap=50)    
    # 2. 构建向量库
    vectorstore = create_vector_store(chunks, persist_dir="./chroma_db")    
    # 3. 构建问答链（替换为你自己的 DeepSeek API Key）
    qa_chain = build_qa_chain(vectorstore, api_key="sk-your-deepseek-api-key")    
    # 4. 问答循环
    while True:
        query = input("\n请输入问题（输入 q 退出）: ")        if query.lower() == 'q':            break
        result = qa_chain({"query": query})        print(f"\n🤖 回答: {result['result']}")        print(f"\n📚 参考来源:")        for i, doc in enumerate(result['source_documents'], 1):            print(f"[{i}] {doc.page_content[:100]}...")if __name__ == "__main__":
    main()

五、进阶：生产级优化

上面是最简版本。在企业落地时，还需要考虑以下优化点：

5.1 混合检索（向量 + 关键词）

纯向量检索会漏掉精确关键词匹配，混合检索能显著提升召回率：

from langchain.retrievers import EnsembleRetrieverfrom langchain_community.retrievers import BM25Retrieverdef create_hybrid_retriever(documents, vector_store):
    vector_retriever = vector_store.as_retriever(search_kwargs={"k": 6})
    keyword_retriever = BM25Retriever.from_documents(documents)
    keyword_retriever.k = 6
    ensemble_retriever = EnsembleRetriever(
        retrievers=[vector_retriever, keyword_retriever],
        weights=[0.7, 0.3]  # 偏向语义检索
    )    return ensemble_retriever

5.2 结果重排（Rerank）

用 Cross-Encoder 对召回结果重排，把最相关的排在前面：

from langchain.retrievers import ContextualCompressionRetrieverfrom langchain.retrievers.document_compressors import CrossEncoderRerankerfrom langchain_community.cross_encoders import HuggingFaceCrossEncoderdef create_rerank_retriever(base_retriever):
    compressor = CrossEncoderReranker(
        model=HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-large"),
        top_n=5
    )    return ContextualCompressionRetriever(
        base_compressor=compressor,
        base_retriever=base_retriever
    )

5.3 带引用的自定义 Prompt

让 LLM 基于上下文回答，并在资料不足时拒答：

from langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.runnables import RunnablePassthroughdef build_advanced_rag_chain(retriever, llm):
    prompt = ChatPromptTemplate.from_template(        "基于以下上下文回答问题。如果上下文没有相关信息，请回答"
        "'抱歉，知识库中未找到相关信息'。\n\n"
        "上下文:\n{context}\n\n问题: {question}\n\n回答:"
    )    def format_docs(docs):        return "\n\n".join(doc.page_content for doc in docs)
    
    rag_chain = (
        {"context": retriever | format_docs, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )    return rag_chain

六、避坑指南（血泪经验）

⚠️ 六大常见误区，每一个都是我踩过的坑：

文档入库只做一次 —— 企业文档会更新，必须做增量更新机制
分片越小越精准 —— 过小会切断语义，建议 500-800 字符
向量维度越高越好 —— 维度高≠效果好，bge-small-zh 在企业场景够用
只使用向量搜索 —— 纯向量检索会漏掉精确关键词，必须混合检索
召回结果越多越好 —— k 值过大会引入噪声，一般 k=3~5
接入 RAG 后就不会产生幻觉 —— RAG 大幅降低幻觉，但不等于消除，仍需拒答机制

七、效果评估

RAG 系统上线前必须评估，否则你根本不知道它在变好还是变坏：

检索评估：召回率、MRR、NDCG
回答评估： faithfulness（忠实度）、answer relevancy
工具推荐：Ragas

# 评估示例（伪代码）from ragas import evaluatefrom ragas.metrics import faithfulness, answer_relevancy

results = evaluate(
    dataset=test_dataset,
    metrics=[faithfulness, answer_relevancy]
)print(results)

本文系转载，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python