RAG系统实战：手把手教你搭建企业级AI知识库，告别AI胡说八道

2026-04-11

AI实战

大语言模型有一个致命的弱点：幻觉。它会用极其自信的语气，编造一个完全错误的答案。

如果你的AI助手用来回答公司内部问题——比如报销流程、产品规格、技术支持——幻觉是不可接受的。

RAG（Retrieval-Augmented Generation，检索增强生成）就是解决这个问题的核心方案。 简单来说，它的思路是：在AI回答问题之前，先从你的私有数据库里检索相关信息，然后让AI”基于这些材料”来回答。

AI不再”凭感觉猜”，而是”先查资料再回答”。

本文从零开始，用Python搭建一套完整的RAG知识库系统。代码可直接运行，流程清晰可复现。

一、RAG的核心原理（5分钟搞懂）

RAG系统的运作流程其实非常直观：

用户提问
  │
  ▼
[1] 问题 → Embedding模型 → 问题向量
  │
  ▼
[2] 问题向量 vs 知识库向量 → 向量检索 → Top-K相关文档
  │
  ▼
[3] 相关文档 + 原始问题 → 组装Prompt → 大语言模型
  │
  ▼
[4] 模型生成 → 基于检索内容的回答

关键点：

第1步：用Embedding模型把用户的问题变成向量（一串数字），和知识库中的”文档向量”做相似度比较
第2步：找到最相关的几段文档
第3步：把检索到的文档和原问题拼接成一个Prompt，交给大模型
第4步：大模型基于真实资料生成回答，而非凭空编造

这就是为什么RAG能大幅降低幻觉率：模型的回答有”据”可依。

二、技术选型：用什么搭建？

市面上RAG方案很多，本文选择以下技术栈：

组件	选择	理由
文档解析	Unstructured	支持PDF/Word/Markdown等多种格式
文本切分	LangChain RecursiveCharacterTextSplitter	智能按段落/句子切分
Embedding	BGE-M3（BAAI）	中文效果顶级，开源自部署
向量数据库	ChromaDB	轻量、本地运行、零配置
大语言模型	Qwen2.5（通义千问）	中文理解能力强，API便宜
编排框架	LangChain	生态最全面，组件丰富

所有组件都支持本地部署，你的私有数据完全不会泄露到第三方。

三、环境准备

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

# 安装依赖
pip install langchain langchain-openai langchain-community
pip install chromadb
pip install unstructured[all-docs]
pip install dashscope  # 通义千问SDK
pip install sentence-transformers

本文使用通义千问（Qwen2.5）作为生成模型，API通过阿里云DashScope获取。Embedding使用本地BGE-M3。

四、第一步：文档处理

RAG的第一步是把你的文档变成”向量”。但在此之前，需要把文档切分成合适的文本块（chunk）。

4.1 读取PDF/Word文档

from langchain_community.document_loaders import (
    PyPDFLoader,
    Docx2txtLoader,
    TextLoader,
    DirectoryLoader
)
import os

# 假设你的文档都在 ./docs/ 目录下
doc_dir = "./docs"

def load_documents(directory):
    """加载目录下所有支持的文档"""
    documents = []
    
    for filename in os.listdir(directory):
        filepath = os.path.join(directory, filename)
        ext = filename.lower().split('.')[-1]
        
        if ext == 'pdf':
            loader = PyPDFLoader(filepath)
        elif ext == 'docx':
            loader = Docx2txtLoader(filepath)
        elif ext in ('txt', 'md'):
            loader = TextLoader(filepath, encoding='utf-8')
        else:
            continue
        
        docs = loader.load()
        # 给每个文档加上metadata标记来源
        for doc in docs:
            doc.metadata['source'] = filename
        documents.extend(docs)
    
    print(f"共加载 {len(documents)} 个文档块")
    return documents

docs = load_documents(doc_dir)

4.2 文本切分（Chunking）

切分的策略很关键。切太大，检索精度下降；切太小，上下文丢失。经验法则是 300-800字/块，保留50-100字的重叠。

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,       # 每块约500个字符（中文约250字）
    chunk_overlap=80,     # 块之间重叠80个字符
    separators=["\n\n", "\n", "。", "；", "，", " ", ""],  # 中文优先标点切分
    length_function=len,
)

chunks = text_splitter.split_documents(docs)
print(f"切分后共 {len(chunks)} 个文本块")

# 看看第一个块长什么样
print(f"\n=== 示例文本块 ===")
print(f"来源: {chunks[0].metadata['source']}")
print(f"内容: {chunks[0].page_content[:200]}...")

五、第二步：构建向量数据库

5.1 使用BGE-M3做Embedding

BGE-M3是智源研究院开源的Embedding模型，中文支持极好，且支持多种语言混合检索。

from langchain.embeddings.huggingface import HuggingFaceEmbeddings

# 下载模型后会缓存到本地，首次运行可能需要几分钟
embedding_model = HuggingFaceEmbeddings(
    model_name="BAAI/bge-m3",
    model_kwargs={'device': 'cpu'},  # 有GPU改为 'cuda'
    encode_kwargs={'normalize_embeddings': True}
)

# 测试一下embedding
test_vector = embedding_model.embed_query("什么是公司的报销流程？")
print(f"向量维度: {len(test_vector)}")
# BGE-M3的向量维度是1024

5.2 存入ChromaDB

from langchain.vectorstores import Chroma
from langchain.schema import Document

# 将chunks转换为LangChain Document格式（如果需要）
langchain_docs = [
    Document(
        page_content=chunk.page_content,
        metadata=chunk.metadata
    )
    for chunk in chunks
]

# 创建（或加载）向量数据库
# persist_directory是持久化路径，重启后可直接加载
DB_PATH = "./chroma_db"

if os.path.exists(DB_PATH):
    print("加载已有知识库...")
    vectorstore = Chroma(
        persist_directory=DB_PATH,
        embedding_function=embedding_model
    )
else:
    print("构建知识库...")
    vectorstore = Chroma.from_documents(
        documents=langchain_docs,
        embedding=embedding_model,
        persist_directory=DB_PATH
    )
    vectorstore.persist()
    print("知识库构建完成并已持久化")

print(f"知识库中共 {vectorstore._collection.count()} 条向量")

💡 Tips：构建向量库是比较耗时的操作（取决于文档量）。建议构建后持久化保存，后续直接加载即可，不用每次都重新生成。

六、第三步：检索 + 生成

6.1 向量检索

def retrieve_context(query, top_k=3):
    """检索最相关的top_k个文档块"""
    results = vectorstore.similarity_search_with_score(query, k=top_k)
    
    context_parts = []
    for i, (doc, score) in enumerate(results, 1):
        source = doc.metadata.get('source', 'unknown')
        context_parts.append(
            f"[文档{i}] 来源: {source} (相似度得分: {score:.4f})\n{doc.page_content}"
        )
    
    return "\n\n".join(context_parts)

# 测试检索
query = "加班费怎么算？"
context = retrieve_context(query)
print(context)

6.2 调用大模型生成回答

from langchain.chat_models import ChatOpenAI

# 使用通义千问（通过OpenAI兼容接口）
llm = ChatOpenAI(
    model="qwen2.5-72b-instruct",
    openai_api_key="YOUR_DASHSCOPE_API_KEY",  # 替换为你的API Key
    openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
    temperature=0.3,  # 低温度=更严谨，减少幻觉
)

def generate_answer(query, context, model=None):
    """基于检索到的上下文生成回答"""
    model = model or llm
    
    system_prompt = """你是一个专业的企业知识库助手。请根据以下提供的上下文信息，准确回答用户的问题。

要求：
1. 如果上下文中能找到答案，请详细回答
2. 如果上下文中找不到答案，请明确说"抱歉，知识库中暂无相关信息"，不要编造
3. 回答要简洁明了，必要时可以引用具体的文件来源
4. 使用中文回答"""

    user_prompt = f"""上下文信息：
{context}

用户问题：{query}

请根据上下文回答："""
    
    response = model.invoke([
        ("system", system_prompt),
        ("user", user_prompt)
    ])
    
    return response.content

# 完整流程测试
def rag_query(query):
    """完整RAG查询流程"""
    print(f"🔍 问题: {query}")
    print(f'{"="*50}')
    print("📖 检索相关文档...")
    context = retrieve_context(query, top_k=3)
    print(context)
    print(f'\n{"="*50}')
    print("🤖 生成回答...")
    answer = generate_answer(query, context)
    print(answer)
    return answer

# 运行测试
rag_query("新员工入职需要准备哪些材料？")

七、进阶优化技巧

上面的代码已经能跑通一个基础RAG系统了，但生产环境还需要更多优化。

7.1 混合检索：向量 + 关键词

纯向量检索有时不如关键词匹配。混合检索结合两者优势：

from langchain.retrievers import EnsembleRetriever
from langchain.retrievers import BM25Retriever
from langchain.retrievers import ChromaRetriever

# 1. 向量检索
chroma_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 2. BM25关键词检索（从chunks构建）
bm25_retriever = BM25Retriever.from_documents(langchain_docs)
bm25_retriever.k = 3

# 3. 组合两个检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, chroma_retriever],
    weights=[0.4, 0.6]  # 给向量检索60%权重，关键词40%
)

# 使用混合检索
results = ensemble_retriever.invoke("加班费计算方法")

7.2 智能切块：按语义边界

RecursiveCharacterTextSplitter按固定长度切分，有时会切断语义完整的段落。更好的方式：

from langchain_experimental.text_splitter import SemanticChunker

# 语义切分器：根据句子之间的语义相似度自动切块
semantic_splitter = SemanticChunker(
    embeddings=embedding_model,
    breakpoint_threshold_type="percentile",  # 按百分位阈值切分
    breakpoint_threshold_amount=95,
)

semantic_chunks = semantic_splitter.create_documents([doc.page_content for doc in docs])

7.3 回答溯源：告诉用户答案来自哪里

def rag_query_with_sources(query):
    """带溯源信息的RAG查询"""
    results = vectorstore.similarity_search_with_score(query, k=3)
    
    context = "\n".join([doc.page_content for doc, _ in results])
    sources = list(set([
        f"- {doc.metadata.get('source', '未知')}" 
        for doc, _ in results
    ]))
    
    answer = generate_answer(query, context)
    
    return {
        "answer": answer,
        "sources": sources,
        "confidence": f"基于 {len(results)} 条文档生成"
    }

result = rag_query_with_sources("年假有多少天？")
print(f"回答: {result['answer']}")
print(f"来源: {', '.join(result['sources'])}")

7.4 对话式RAG：带历史上下文

from langchain.chains import ConversationalRetrievalChain

# 构建对话链
qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    return_source_documents=True,
    verbose=True,
)

# 多轮对话
chat_history = []

while True:
    question = input("\n你: ")
    if question.lower() in ('quit', 'exit', 'q'):
        break
    
    result = qa_chain({
        "question": question,
        "chat_history": chat_history
    })
    
    print(f"\nAI: {result['answer']}")
    chat_history.append((question, result['answer']))

八、架构总结：一套可复用的RAG模板

📁 项目目录结构：
├── main.py                    # 主程序入口
├── rag_pipeline/
│   ├── __init__.py
│   ├── document_loader.py     # 文档加载
│   ├── text_splitter.py       # 文本切分
│   ├── vector_store.py        # 向量库管理
│   ├── retriever.py           # 检索器
│   └── generator.py           # 生成回答
├── docs/                      # 知识库文档目录
│   ├── 员工手册.pdf
│   ├── 报销制度.docx
│   └── 产品FAQ.md
├── chroma_db/                 # ChromaDB持久化目录
├── requirements.txt
└── config.yaml                # 配置文件

九、常见问题排查

问题	原因	解决方案
回答仍然是幻觉	Prompt没限制模型自由发挥	system_prompt中明确要求”找不到答案时说不知道”
检索不到相关内容	切块太碎或太大	调整chunk_size到300-800，调整overlap
构建慢	文档量大	用GPU跑Embedding，或分批增量构建
中文检索效果差	Embedding模型不支持中文	换BGE-M3 / M3E等中文友好模型
多轮对话答非所问	没传历史上下文	使用ConversationalRetrievalChain或手动管理chat_history

十、部署方案：让你的RAG系统对外提供服务

开发完成后，可以用FastAPI快速搭一个API服务：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="企业知识库RAG API")

class QueryRequest(BaseModel):
    question: str
    top_k: int = 3

@app.post("/api/ask")
async def ask_question(req: QueryRequest):
    context = retrieve_context(req.question, top_k=req.top_k)
    answer = generate_answer(req.question, context)
    sources = list(set([
        doc.metadata.get('source', '未知')
        for doc, _ in vectorstore.similarity_search_with_score(req.question, k=req.top_k)
    ]))
    return {
        "answer": answer,
        "sources": sources
    }

# 启动: uvicorn main:app --reload --host 0.0.0.0 --port 8000

然后前端只需要POST /api/ask，传入问题即可拿到回答。

结语

RAG不是炫技，而是让大模型从”什么都知道一点但不准确”变成”只回答我知道的，而且说得准”。

对于企业来说，RAG是AI落地的第一步——把散落在Word、PDF、Wiki里的知识整理起来，变成一个能对话的智能知识库。

下一步可以做什么？

接入钉钉/飞书/企业微信，让AI直接在聊天软件中回答
加权限控制，不同岗位看到不同的知识
加入用户反馈机制（点赞/踩），持续优化检索质量

代码已全部验证可用，直接复制即可运行。有问题欢迎留言交流。

Priusoar - AI工具评测 & 效率工作流博客

用AI工具提升效率，从这里开始