RAG知识库实战：从0到1搭建本地AI知识检索系统

RAG知识库实战：从0到1搭建本地AI知识检索系统

RAG（Retrieval-Augmented Generation，检索增强生成）是目前大模型落地最实用的技术之一。它的核心思想很简单：不让大模型”凭记忆”回答问题，而是先从知识库检索相关内容，再让模型基于检索结果生成答案。

这套方案特别适合：企业内部知识库、技术文档问答、私域数据检索等场景。本文从原理到实战，手把手搭建一个完整的 RAG 系统。

一、RAG 核心原理

RAG 的工作流程分为三个阶段：

1
2
3
4
5
6
7
用户提问
  ↓
[检索阶段] 从向量数据库中检索最相关的 K 个文档片段
  ↓
[增强阶段] 将检索结果与用户问题组装成 Prompt
  ↓
[生成阶段] 大模型基于增强后的 Prompt 生成答案

这个流程解决了大模型的三个核心问题：知识过时、幻觉（胡编乱造）、无法访问私有数据。

二、技术选型：向量数据库

向量数据库是 RAG 的存储层，负责把文本向量化和高效检索。

数据库 特点 适用场景
Chroma 轻量、Python 原生、单机友好 原型验证、个人项目
Qdrant 高性能、支持云原生、分布式 生产环境
Milvus 超大规模、社区成熟 超大数据量
FAISS Facebook 开源、内存级 离线分析、实验

本文用 Chroma 做演示（最简单），代码稍改就能切换到 Qdrant 或 Milvus。

三、完整实战：从文档到答案

3.1 环境准备

1
2
3
pip install langchain langchain-community langchain-openai \
    chromadb sentence-transformers pypdf python-dotenv \
    unstructured tiktoken

3.2 文档加载与分块

RAG 的效果很大程度上取决于分块策略（Chunking）。分块太小，丢失上下文；分块太大，引入噪声。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document

# 加载多种格式的文档
def load_documents(directory: str) -> list[Document]:
    documents = []
    import os

    for filename in os.listdir(directory):
        filepath = os.path.join(directory, filename)
        if filename.endswith('.pdf'):
            loader = PyPDFLoader(filepath)
            documents.extend(loader.load())
        elif filename.endswith('.txt'):
            loader = TextLoader(filepath, encoding='utf-8')
            documents.extend(loader.load())
        elif filename.endswith('.md'):
            loader = TextLoader(filepath, encoding='utf-8')
            documents.extend(loader.load())

    return documents

# 智能分块：按段落切分，保留重叠避免割裂语义
def split_documents(
    documents: list[Document],
    chunk_size: int = 500,
    chunk_overlap: int = 100
) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        separators=["\n\n", "\n", "。", "！", "？", " ", ""],
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len
    )
    return splitter.split_documents(documents)

# 示例
docs = load_documents("./knowledge_base/")
chunks = split_documents(docs)
print(f"共加载 {len(docs)} 个文档，切分成 {len(chunks)} 个块")

3.3 向量化与存储

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma

# 使用 HuggingFace 本地 Embedding 模型（无需 API key）
def create_vectorstore(chunks: list[Document], persist_dir: str = "./chroma_db"):
    # 使用中文效果好的 Embedding 模型
    embeddings = HuggingFaceEmbeddings(
        model_name="sentence-transformers/all-MiniLM-L6-v2",
        model_kwargs={"device": "cpu"},
        encode_kwargs={"normalize_embeddings": True}
    )

    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir
    )
    vectorstore.persist()
    return vectorstore

# 构建向量数据库
vs = create_vectorstore(chunks)
print("向量数据库构建完成")

3.4 检索与生成

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
import os

# 构建检索增强问答链
def build_qa_chain(vectorstore):
    # 使用 GPT-4o-mini（便宜、快速）
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        temperature=0,
        api_key=os.getenv("OPENAI_API_KEY")
    )

    # 自定义 Prompt，让回答更可靠
    prompt_template = """你是一个专业问答助手。基于以下检索到的参考资料，
回答用户的问题。如果参考资料中没有相关信息，请明确说明"没有找到相关内容"，
不要编造答案。

参考内容：
{context}

用户问题：{question}

回答："""
    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": 5}  # 检索最相关的 5 个块
        ),
        return_source_documents=True,
        chain_type_kwargs={"prompt": prompt}
    )
    return qa_chain

# 执行问答
def ask_question(chain, question: str):
    result = chain.invoke({"query": question})

    print(f"问题：{question}")
    print(f"\n答案：{result['result']}")
    print(f"\n参考来源：")
    for i, doc in enumerate(result['source_documents'], 1):
        print(f"  [{i}] {doc.metadata.get('source', 'unknown')}")
        print(f"      {doc.page_content[:100]}...")

    return result

# 使用
chain = build_qa_chain(vs)
ask_question(chain, "这个项目的部署流程是什么？")

四、检索优化技巧

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

纯向量检索有时找不到专有名词（如”API v2”），结合 BM25 关键词检索效果更好：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

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

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

# 按 6:4 权重混合
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4]
)

# 使用混合检索
query = "如何配置数据库连接池？"
results = ensemble_retriever.invoke(query)

4.2 重排序（Reranker）

初步检索后，用重排序模型优化结果顺序：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

# 使用 Cross-Encoder 做重排序
cross_encoder = HuggingFaceCrossEncoder(
    model_name="BAAI/bge-reranker-base"
)

def rerank_results(query: str, documents: list[Document], top_k: int = 3):
    doc_texts = [doc.page_content for doc in documents]
    scores = cross_encoder.score(query, doc_texts)

    # 按分数排序
    scored_docs = list(zip(scores, documents))
    scored_docs.sort(key=lambda x: x[0], reverse=True)

    return [doc for _, doc in scored_docs[:top_k]]

# 先检索 20 个，再用 Reranker 精选 3 个
initial_results = vs.similarity_search(query, k=20)
final_results = rerank_results(query, initial_results, top_k=3)

4.3 元数据过滤

在检索时按元数据过滤，大幅提升精准度：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 按文件类型过滤
filtered_retriever = vs.as_retriever(
    search_kwargs={
        "k": 5,
        "filter": {"source": {"$contains": "API"}}
    }
)

# 按日期过滤（获取最近更新的文档）
filtered_retriever = vs.as_retriever(
    search_kwargs={
        "k": 5,
        "filter": {"date": {"$gte": "2026-01-01"}}
    }
)

五、本地部署版本（无需 OpenAI）

如果不想依赖 OpenAI API，可以用 Ollama 本地模型：

1
2
3
4
5
6
7
8
9
10
11
from langchain_ollama import ChatOllama, OllamaEmbeddings

# 使用本地 Ollama 模型
embeddings = OllamaEmbeddings(model="nomic-embed-text")
llm = ChatOllama(model="qwen3:8b", temperature=0)

# 其他代码完全相同
vs = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

六、RAG 系统评估

上线前一定要评估系统效果，避免”看起来能用但实际不靠谱”：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from langchain.evaluation import QAEvalChain

# 准备测试问题集（人工标注期望答案）
test_questions = [
    "项目的依赖版本要求是什么？",
    "如何启动本地开发环境？",
    "数据库迁移的步骤？",
]

ground_truths = [
    "Python 3.11+，FastAPI 0.100+，...",
    "运行 docker-compose up -d",
    "alembic upgrade head",
]

# 自动评估
eval_chain = QAEvalChain.from_llm(llm)

predictions = [
    chain.invoke({"query": q})["result"]
    for q in test_questions
]

graded_results = eval_chain.evaluate(
    test_questions,
    predictions,
    ground_truths
)

for i, result in enumerate(graded_results):
    status = "✅" if result['results'] == "correct" else "❌"
    print(f"{status} 问题{i+1}: {result['results']}")

总结

RAG 系统的质量由三个因素决定：分块策略（决定检索粒度）、Embedding 模型（决定语义匹配质量）、检索优化（重排序、混合检索、元数据过滤）。

本文的代码是一个可用的最小化系统，在此基础上可以继续优化：换成更强大的 Embedding 模型、添加知识图谱做关系推理、实现多轮对话记忆等。