Appearance
🧱 从 0 到 1 完整 RAG 系统：本地部署、检索、引用与评测
📌 这篇不是"RAG 概念介绍"——是可交付的实战教程。目标是看完之后你能在本地跑起来一套完整 RAG 系统：读文档 → 检索 → 回答 → 带来源引用 → 能评测。
01｜项目目标
做一个"公司内部知识库问答系统"。用户用自然语言提问，系统从公司文档里检索相关内容，给出带引用来源的回答。
最终效果：
- 用户问："我们的年假政策是什么？"
- 系统答："根据《员工手册 v2025》第 3.2 节，年假为每年 15 天，可累计最多 30 天。[来源: employee_handbook_2025.pdf, 第 12 页]"
02｜技术栈选择
python
# requirements.txt
langchain==0.3.0
langchain-openai==0.2.0
langchain-community==0.3.0
chromadb==0.5.0 # 轻量本地向量库（零配置）
unstructured==0.15.0 # 文档解析（支持 PDF/Word/HTML）
pypdf==4.0.0
tiktoken==0.7.0
ragas==0.2.0 # RAG 评测
fastapi==0.115.0
uvicorn==0.30.0
python-dotenv==1.0.0选择理由：
- ChromaDB：不需要装 Docker、不需要配服务器——
pip install chromadb就能用。适合本地开发和小规模部署 - Unstructured：能处理 PDF、Word、HTML、Markdown 等 20+ 种格式
- LangChain：RAG 的文档加载和链式调用最简单（Agent 逻辑会用 LangGraph 但 RAG 用 LangChain）
03｜项目目录结构
text
rag-system/
├── data/ # 原始文档
│ ├── handbook.pdf
│ └── faq.docx
├── src/
│ ├── ingest.py # 文档读取 + 清洗 + 索引
│ ├── retrieve.py # 检索模块
│ ├── generate.py # 生成模块
│ ├── app.py # FastAPI 接口
│ └── evaluate.py # 评测脚本
├── chroma_db/ # 向量库持久化目录
├── eval_set/ # 评测集
│ └── test_questions.json
├── requirements.txt
└── .env # OPENAI_API_KEY04｜Step 1：文档读取与清洗
python
# ingest.py
from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
# 1. 加载文档
loader = DirectoryLoader("data/", glob="**/*.pdf", loader_cls=PyPDFLoader)
raw_docs = loader.load()
print(f"Loaded {len(raw_docs)} documents")
# 2. 清洗
for doc in raw_docs:
doc.page_content = doc.page_content.replace("\n\n", "\n") # 去多余空行
doc.page_content = doc.page_content.replace("\t", " ") # Tab 转空格
# 3. 语义分块（chunk_size=500, chunk_overlap=100）
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, chunk_overlap=100,
separators=["\n\n", "\n", "。", ".", " ", ""]
)
chunks = splitter.split_documents(raw_docs)
print(f"Split into {len(chunks)} chunks")
# 4. Embedding + 向量库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
chunks, embeddings,
persist_directory="chroma_db/"
)
print("Indexing complete!")关键说明：
chunk_overlap=100确保相邻 chunk 之间有重叠——防止关键信息被切在两段之间separators优先按段落（\n\n）切，再按句号切，最后按空格——保持语义完整性
05｜Step 2：混合检索
python
# retrieve.py
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(persist_directory="chroma_db/", embedding_function=embeddings)
# Dense 检索（语义）
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
# Sparse 检索（关键词 BM25）
bm25_retriever = BM25Retriever.from_documents(all_chunks) # 需要缓存chunks
bm25_retriever.k = 10
# 混合检索 + Rerank
ensemble = EnsembleRetriever(
retrievers=[dense_retriever, bm25_retriever],
weights=[0.7, 0.3] # Dense 权重更高（语义匹配为主）
)
def retrieve(query: str, top_k: int = 5):
docs = ensemble.invoke(query)[:top_k]
return docs为什么需要混合检索：
- Dense（语义）：用户问"怎么请假"→ 能匹配到"休假流程"（语义接近）
- Sparse（关键词）：用户问"PTO 政策"→能匹配到精确包含"PTO"的文档片段（Dense 可能把 PTO 和"请假"搞混）
06｜Step 3：Prompt 组装与生成
python
# generate.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", """你是公司内部知识库助手。请只根据以下参考资料回答问题。
如果参考资料不足以回答问题，请诚实地说"根据现有资料无法回答这个问题"——不要编造。
参考资料：
{context}"""),
("user", "{question}")
])
def generate_answer(question: str, retrieved_docs: list):
# 拼装上下文 + 引用来源
context = ""
sources = []
for i, doc in enumerate(retrieved_docs):
context += f"[{i+1}] {doc.page_content}\n\n"
sources.append(doc.metadata.get("source", "Unknown"))
chain = prompt | llm
response = chain.invoke({"context": context, "question": question})
return {
"answer": response.content,
"sources": sources
}07｜Step 4：拒答机制
python
def should_answer(retrieved_docs, threshold=0.7):
"""用检索分数判断是否应该回答"""
# ChromaDB 的 similarity_search_with_score 返回 (doc, score)
# score 越小越相似（L2 距离）
if not retrieved_docs:
return False
# 如果第一个文档的相似度 < 0.7（阈值可调），拒答
avg_score = sum(d.metadata.get("score", 0) for d in retrieved_docs) / len(retrieved_docs)
return avg_score < threshold为什么需要拒答：Agent 检索不到相关文档时，如果没有拒答机制，LLM 会"自由发挥"——编造一个看起来像真的回答。拒答 = 保护用户免受幻觉。
08｜Step 5：FastAPI 接口
python
# app.py
from fastapi import FastAPI
from pydantic import BaseModel
from retrieve import retrieve
from generate import generate_answer
app = FastAPI()
class Query(BaseModel):
question: str
@app.post("/ask")
async def ask(query: Query):
docs = retrieve(query.question)
if not docs:
return {"answer": "未找到相关信息。", "sources": []}
result = generate_answer(query.question, docs)
return {"answer": result["answer"], "sources": result["sources"][:5]}
# 启动: uvicorn app:app --host 0.0.0.0 --port 800009｜Step 6：RAGAS 评测
python
# evaluate.py
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
from datasets import Dataset
# 准备评测数据
eval_data = Dataset.from_dict({
"question": ["年假是多少天？", "谁负责审批请假？"],
"answer": [
"根据《员工手册》第3.2节，年假为每年15天。", # Agent 的输出
"请假由直属经理审批。"
],
"contexts": [
["年假为每年15天，可累计..."],
["请假由直属经理审批..."]
],
"ground_truth": [
"年假15天",
"直属经理审批"
]
})
# 跑评测
result = evaluate(eval_data, metrics=[faithfulness, answer_relevancy, context_precision])
print(result)
# 期望值: faithfulness > 0.85, answer_relevancy > 0.8010｜错误排查表
| 症状 | 可能原因 | 排查方向 |
|---|---|---|
| 检索结果不相关 | Embedding 模型与文档领域不匹配 | 换个 Embedding 模型试试（bge-m3 对中文更好） |
| 回答编造数据 | 检索没找到相关文档，LLM 自由发挥了 | 加拒答机制（检查检索分数） |
| Chunk 内容不完整 | Chunk 太小（< 300 字） | 调大 chunk_size 到 800-1000 |
| 中文回答英文 | System Prompt 里没指定语言 | 加一条 "请用中文回答" |
| 向量库越来越大 | Chunking 太细导致 10 万+个向量 | 用 PCA 降维或换更大规模的向量库 |
11｜从 Demo 到生产需要补什么
当前版本能做的事情：
- ✅ 本地跑通 RAG 全流程
- ✅ 混合检索 + 拒答
- ✅ API 接口 + 评测
上生产需要补的：
- 增量索引（文档更新时不用全量重建向量库）
- 权限隔离（不同用户看到不同知识库）
- 全链路 trace + Dashboard
- 缓存层（高频问题缓存）
- Docker 化 + CI/CD
📌 RAG 不是"搭起来就完了"——是"搭起来之后开始持续优化检索质量和生成质量的过程"。评测集是你在这个过程中的唯一导航工具。
🍋 本文为 AI Agent 学习路线 · 完整 RAG 项目实战。© 2026 AI小柠檬。