RAG（Retrieval-Augmented Generation）是当前大模型落地最实用的方案之一——不用微调、不用海量 GPU，就能让模型基于你自己的文档回答问题。本文记录从零搭建一个本地 RAG 知识库问答系统的完整过程，所有组件都可以在本地运行，不依赖任何云端 API。
整体架构
用户提问
│
▼
Query Embedding ──→ 向量检索(ChromaDB) ──→ Top-K 相关文档片段
│
▼
Prompt 组装(Context + Question)
│
▼
本地 LLM (Ollama + Qwen2)
│
▼
回答
技术栈选型：
| 组件 | 选型 | 理由 |
|---|---|---|
| 文档加载 | LangChain Loaders | 支持 PDF/Markdown/Word/HTML 等 |
| 文本切分 | RecursiveCharacterTextSplitter | 对中文友好，支持语义分段 |
| Embedding | bge-small-zh-v1.5 | 中文效果好，模型小(~90MB)，CPU 可跑 |
| 向量数据库 | ChromaDB | 纯 Python，无需额外部署 |
| LLM | Ollama + Qwen2-7B | 本地部署，隐私安全，中文能力强 |
| 编排 | LangChain | 粘合各组件 |
环境准备
pip install langchain langchain-community chromadb sentence-transformers
pip install pymupdf python-docx unstructured
# 安装 Ollama (macOS/Linux)
curl -fsSL https://ollama.com/install.sh | sh
# 拉取 Qwen2-7B
ollama pull qwen2:7b
文档加载与切分
支持多种文档格式是实用系统的基本要求：
from langchain_community.document_loaders import (
PyMuPDFLoader,
UnstructuredMarkdownLoader,
Docx2txtLoader,
DirectoryLoader,
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
import os
def load_documents(docs_dir: str):
'''从目录加载所有支持格式的文档'''
documents = []
# PDF
for f in os.listdir(docs_dir):
path = os.path.join(docs_dir, f)
if f.endswith(".pdf"):
loader = PyMuPDFLoader(path)
documents.extend(loader.load())
elif f.endswith(".md"):
loader = UnstructuredMarkdownLoader(path)
documents.extend(loader.load())
elif f.endswith(".docx"):
loader = Docx2txtLoader(path)
documents.extend(loader.load())
return documents
def split_documents(documents, chunk_size=500, chunk_overlap=80):
'''
切分文档。
chunk_size=500 对中文比较合适，太大会稀释检索精度，
太小会丢失上下文。
'''
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", "。", "！", "？", ".", " ", ""],
)
return splitter.split_documents(documents)
RecursiveCharacterTextSplitter 的 separators 参数很关键。默认的分隔符对英文优化，处理中文时加上 "。" "！" "？" 可以让切分更贴合语义边界。
Embedding 模型
选择 bge-small-zh-v1.5，MTEB 中文榜排名靠前，且模型仅 90MB 左右，CPU 机器也能跑：
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
def get_embedding_model():
model_name = "BAAI/bge-small-zh-v1.5"
encode_kwargs = {"normalize_embeddings": True}
return HuggingFaceBgeEmbeddings(
model_name=model_name,
model_kwargs={"device": "cpu"},
encode_kwargs=encode_kwargs,
query_instruction="为这个句子生成表示以用于检索相关文章：",
)
query_instruction 是 bge 模型的特性——query 和 document 用不同的指令前缀，能提升检索效果。
向量库构建（ChromaDB）
from langchain_community.vectorstores import Chroma
def build_vector_store(docs, embedding, persist_dir="./chroma_db"):
'''构建并持久化向量库'''
vectorstore = Chroma.from_documents(
documents=docs,
embedding=embedding,
persist_directory=persist_dir,
collection_metadata={"hnsw:space": "cosine"},
)
vectorstore.persist()
print(f"向量库构建完成，共 {vectorstore._collection.count()} 条向量")
return vectorstore
def load_vector_store(embedding, persist_dir="./chroma_db"):
'''加载已有的向量库'''
return Chroma(
persist_directory=persist_dir,
embedding_function=embedding,
)
ChromaDB 默认用 HNSW 索引，对于几万条向量的规模完全够用。如果文档量上到百万级，需要换 Milvus 或 Qdrant。
本地 LLM 接入（Ollama）
from langchain_community.llms import Ollama
def get_llm():
return Ollama(
model="qwen2:7b",
temperature=0.1,
num_ctx=4096, # 上下文窗口
repeat_penalty=1.1, # 降低重复
)
temperature=0.1 让回答更确定性，适合知识库问答场景。num_ctx=4096 对于拼接 context + question 通常够用。
LangChain 整合 RAG Chain
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA
PROMPT_TEMPLATE = '''基于以下已知信息来回答用户的问题。
如果已知信息中没有足够的内容来回答，请直接说明"根据已有资料无法回答该问题"，不要编造答案。
已知信息：
{context}
用户问题：
{question}
请用中文回答：'''
def build_qa_chain(vectorstore, llm):
retriever = vectorstore.as_retriever(
search_type="mmr", # 最大边际相关性，减少冗余
search_kwargs={
"k": 4, # 返回 4 个最相关片段
"fetch_k": 10, # MMR 候选池大小
"lambda_mult": 0.7, # 相关性 vs 多样性平衡
},
)
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},
)
return qa_chain
这里用 search_type="mmr" 而不是默认的相似度搜索。MMR（Maximal Marginal Relevance）会在返回结果中兼顾相关性和多样性，避免检索到内容高度重复的片段。
完整使用流程
def main():
# 1. 加载文档
docs = load_documents("./knowledge_base")
print(f"加载了 {len(docs)} 个文档")
# 2. 切分
chunks = split_documents(docs)
print(f"切分为 {len(chunks)} 个片段")
# 3. 构建向量库
embedding = get_embedding_model()
vectorstore = build_vector_store(chunks, embedding)
# 4. 构建 QA Chain
llm = get_llm()
qa_chain = build_qa_chain(vectorstore, llm)
# 5. 问答
while True:
question = input("\n请输入问题 (q 退出): ")
if question.strip().lower() == "q":
break
result = qa_chain({"query": question})
print(f"\n回答：{result['result']}")
print(f"\n参考来源：")
for i, doc in enumerate(result["source_documents"]):
src = doc.metadata.get("source", "unknown")
print(f" [{i+1}] {src}")
if __name__ == "__main__":
main()
效果优化方向
实际使用中效果不理想的话，可以从这几个方向调优：
1. 切分策略
chunk_size 对效果影响很大。我的经验值：
- 技术文档、API 手册：300-500
- 长篇报告、论文：500-800
- FAQ/QA 对：尽量一个 QA 一个 chunk
2. 检索增强
# 混合检索：向量检索 + BM25 关键词检索
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
bm25 = BM25Retriever.from_documents(chunks, k=4)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
ensemble = EnsembleRetriever(
retrievers=[bm25, vector_retriever],
weights=[0.4, 0.6],
)
混合检索对于专有名词、代码片段等语义 Embedding 不擅长的内容有明显提升。
3. Reranker
检索后用 cross-encoder 模型做重排序，精度提升明显但会增加延迟：
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-base")
def rerank(query, docs, top_k=3):
pairs = [(query, doc.page_content) for doc in docs]
scores = reranker.predict(pairs)
ranked = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, _ in ranked[:top_k]]
4. Query 改写
用户的口语化提问往往检索效果不佳，可以让 LLM 先把问题改写成更适合检索的形式：
rewrite_prompt = '''请将以下用户问题改写为更适合在知识库中检索的查询语句，
保持原意但使用更精确的关键词。只输出改写后的查询，不要解释。
用户问题：{question}
改写查询：'''
总结
这套方案的硬件需求不高——8GB 内存 + 一块中端显卡（甚至纯 CPU）就能跑起来。实际效果取决于文档质量和切分策略，建议先用小规模文档验证效果，再逐步扩充知识库。生产环境部署时记得加上文档更新的增量索引机制，避免每次全量重建向量库。
