前言
大语言模型（LLM）虽然强大，但存在两个核心痛点：知识截断和幻觉问题。模型的训练数据有截止日期，且无法准确回答其训练数据之外的问题。RAG（Retrieval-Augmented Generation，检索增强生成）正是解决这两个问题的关键技术——它让模型在回答问题前，先从你提供的知识库中检索相关信息，再基于这些信息生成回答。
本文将带你从零构建一个完整的本地 RAG 问答系统，技术栈为 LangChain + ChromaDB + OpenAI Embeddings，所有代码可直接运行。
环境准备
首先安装必要的依赖包：
1
2
3
4
5
6
| # 创建虚拟环境
python -m venv rag-env
source rag-env/bin/activate
# 安装依赖
pip install langchain langchain-openai langchain-community chromadb unstructured pypdf
|
项目目录结构如下：
1
2
3
4
5
6
| rag-knowledge-base/
├── docs/ # 知识库文档目录（放入 PDF/TXT/MD 文件）
├── chroma_db/ # ChromaDB 持久化存储
├── rag_core.py # RAG 核心逻辑
├── chat.py # 交互式问答入口
└── requirements.txt
|
第一步：文档加载与预处理
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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
| # rag_core.py
from langchain_community.document_loaders import (
PyPDFLoader,
TextLoader,
UnstructuredMarkdownLoader,
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
import os
from pathlib import Path
def load_documents(docs_dir: str) -> list[Document]:
"""加载 docs 目录下所有支持格式的文档"""
loaders_map = {
".pdf": PyPDFLoader,
".txt": TextLoader,
".md": UnstructuredMarkdownLoader,
}
documents = []
docs_path = Path(docs_dir)
for file_path in docs_path.rglob("*"):
suffix = file_path.suffix.lower()
if suffix in loaders_map:
try:
loader = loaders_map[suffix](str(file_path))
documents.extend(loader.load())
print(f"✅ 已加载: {file_path.name} ({len(documents)} 页)")
except Exception as e:
print(f"❌ 加载失败 {file_path.name}: {e}")
print(f"\n共加载 {len(documents)} 个文档片段")
return documents
def split_documents(
documents: list[Document],
chunk_size: int = 500,
chunk_overlap: int = 100,
) -> list[Document]:
"""将文档切分为小块，每个 chunk 之间有重叠以保持语义连贯"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],
)
chunks = text_splitter.split_documents(documents)
print(f"切分完成: {len(documents)} 个文档 → {len(chunks)} 个片段")
return chunks
|
关键参数说明：
chunk_size=500：每个文本块最大 500 字符。太大会引入噪声，太小会丢失上下文chunk_overlap=100：相邻块重叠 100 字符，避免在句子中间截断导致语义断裂separators 列表按优先级使用中文分隔符
第二步：向量化存储（ChromaDB）
ChromaDB 是一个轻量级的向量数据库，无需独立部署服务，直接嵌入 Python 进程：
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
| import chromadb
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
def build_vector_store(
chunks: list[Document],
persist_dir: str = "./chroma_db",
collection_name: str = "knowledge_base",
) -> Chroma:
"""将文档片段向量化并存入 ChromaDB"""
# 初始化 Embedding 模型
# 使用 OpenAI 的 text-embedding-3-small，性价比最高
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 构建向量数据库（自动调用 Embedding API 向量化）
vector_store = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory=persist_dir,
collection_name=collection_name,
)
# 持久化到磁盘
vector_store.persist()
print(f"✅ 向量数据库已构建: {persist_dir}")
print(f" 共 {vector_store._collection.count()} 个向量")
return vector_store
def load_vector_store(
persist_dir: str = "./chroma_db",
collection_name: str = "knowledge_base",
) -> Chroma:
"""从磁盘加载已有的向量数据库"""
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma(
persist_directory=persist_dir,
embedding_function=embeddings,
collection_name=collection_name,
)
print(f"✅ 已加载向量数据库: {vector_store._collection.count()} 个向量")
return vector_store
|
💡 成本提示：text-embedding-3-small 模型每 1M token 约 $0.02，一个中等规模的知识库（100 页 PDF）向量化成本通常不到 $0.1。
第三步：构建 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
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
58
59
60
| from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
def format_docs(docs: list[Document]) -> str:
"""将检索到的文档片段格式化为纯文本"""
formatted = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "未知来源")
page = doc.metadata.get("page", "")
page_info = f"（第 {page} 页）" if page else ""
formatted.append(f"[文档{i}] {source}{page_info}:\n{doc.page_content}")
return "\n\n---\n\n".join(formatted)
def build_rag_chain(vector_store: Chroma, k: int = 4):
"""构建 RAG 检索增强生成链"""
# 1. 配置 LLM
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0, # 知识问答场景使用低温度保证准确性
)
# 2. 配置检索器
retriever = vector_store.as_retriever(
search_type="similarity", # 余弦相似度检索
search_kwargs={"k": k}, # 返回最相关的 k 个片段
)
# 3. 定义 Prompt 模板
prompt = ChatPromptTemplate.from_template("""
你是一个专业的技术助手。请根据以下提供的参考资料回答用户的问题。
要求：
1. 严格基于提供的资料回答，不要编造信息
2. 如果资料中没有相关内容，请明确告知用户
3. 回答时引用具体的文档来源
参考资料：
{context}
用户问题：{question}
请回答：""")
# 4. 使用 LCEL（LangChain Expression Language）组装链
rag_chain = (
{
"context": retriever | format_docs, # 检索 → 格式化
"question": RunnablePassthrough(), # 问题直接传递
}
| prompt # 填充 Prompt
| llm # 调用 LLM
| StrOutputParser() # 提取文本输出
)
return rag_chain, retriever
|
检索策略解析：
search_type="similarity" 使用余弦相似度匹配，适合大多数场景k=4 表示返回 4 个最相关的片段，足够覆盖问题所需信息- 也可以使用
search_type="mmr"（最大边际相关性）来增加检索结果的多样性
第四步：交互式问答
把所有组件串联起来，实现一个命令行交互式问答：
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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
| # chat.py
from rag_core import (
load_documents,
split_documents,
build_vector_store,
load_vector_store,
build_rag_chain,
)
import sys
import os
def initialize(docs_dir: str = "./docs", force_rebuild: bool = False):
"""初始化 RAG 系统"""
persist_dir = "./chroma_db"
if os.path.exists(persist_dir) and not force_rebuild:
print("📦 加载已有向量数据库...")
vector_store = load_vector_store(persist_dir)
else:
print("📚 开始构建知识库...")
documents = load_documents(docs_dir)
if not documents:
print("❌ docs 目录为空，请先放入文档")
sys.exit(1)
chunks = split_documents(documents)
vector_store = build_vector_store(chunks, persist_dir)
rag_chain, retriever = build_rag_chain(vector_store)
return rag_chain, retriever
def interactive_chat():
"""交互式问答循环"""
rag_chain, retriever = initialize()
print("\n" + "=" * 60)
print("🤖 RAG 知识库问答系统已启动")
print(" 输入问题开始对话，输入 'quit' 退出")
print(" 输入 'debug' 查看检索到的文档片段")
print("=" * 60 + "\n")
debug_mode = False
while True:
try:
question = input("👤 你的问题: ").strip()
except (EOFError, KeyboardInterrupt):
print("\n👋 再见！")
break
if not question:
continue
if question.lower() == "quit":
print("👋 再见！")
break
if question.lower() == "debug":
debug_mode = not debug_mode
print(f"🔧 调试模式: {'开启' if debug_mode else '关闭'}")
continue
if debug_mode:
# 展示检索结果
docs = retriever.invoke(question)
print("\n📄 检索到的文档片段:")
for i, doc in enumerate(docs, 1):
print(f"\n--- 片段 {i} ---")
print(f"来源: {doc.metadata.get('source', '未知')}")
print(f"内容: {doc.page_content[:200]}...")
# RAG 生成回答
print("\n🤖 回答:")
try:
response = rag_chain.invoke(question)
print(response)
except Exception as e:
print(f"❌ 生成失败: {e}")
print()
if __name__ == "__main__":
interactive_chat()
|
第五步：进阶优化
5.1 混合检索（Hybrid Search）
结合语义检索和关键词检索，提升召回率：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
def build_hybrid_retriever(chunks: list[Document], vector_store: Chroma, k: int = 4):
"""混合检索：向量相似度 + BM25 关键词匹配"""
# BM25 关键词检索器
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = k
# 向量语义检索器
vector_retriever = vector_store.as_retriever(search_kwargs={"k": k})
# 集成检索器，按权重混合
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.3, 0.7], # 语义检索权重更高
)
return ensemble_retriever
|
5.2 查询重写（Query Rewriting）
用户的问题往往口语化，重写查询可以提升检索效果：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| rewrite_prompt = ChatPromptTemplate.from_template("""
请将以下用户问题改写为更适合技术文档检索的查询语句。
只输出改写后的查询，不要解释。
原始问题：{question}
改写后的查询：""")
def build_query_rewriter(llm):
"""构建查询重写链"""
return rewrite_prompt | llm | StrOutputParser()
# 在 RAG 链中使用：
# rewritten = query_rewriter.invoke(user_question)
# results = retriever.invoke(rewritten)
|
5.3 响应源追溯
让用户知道回答来自哪些文档：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| def build_rag_chain_with_sources(vector_store: Chroma):
"""带源文档追溯的 RAG 链"""
retriever = vector_store.as_retriever(search_kwargs={"k": 4})
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_template("""
基于以下参考资料回答问题。在回答末尾标注引用的文档编号。
参考资料：
{context}
问题：{question}
回答：""")
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
return rag_chain, retriever
|
性能与成本参考
| 环节 | 耗时 | 成本 |
|---|
| 文档加载（100页 PDF） | ~2秒 | - |
| 文本切分 | <0.1秒 | - |
| Embedding 向量化 | ~30秒 | ~$0.05 |
| 单次检索（4个片段） | ~0.2秒 | - |
| GPT-4o-mini 生成 | ~2秒 | ~$0.002 |
| 总计（首次构建+问答） | ~32秒 | ~$0.05 |
| 后续问答（仅检索+生成） | ~2.2秒 | ~$0.002/次 |
常见问题排查
Q: ChromaDB 持久化后数据丢失？
确认调用了 vector_store.persist()，且 persist_directory 路径存在。ChromaDB v0.4+ 可能需要显式调用 chromadb.PersistentClient(path=...)。
Q: 检索结果不相关？
- 检查
chunk_size 是否合适（代码类文档建议 800-1000） - 尝试混合检索（5.1 节）
- 调整
k 值，增加检索数量
Q: Embedding API 调用报错？
确认 OPENAI_API_KEY 环境变量已设置。国内网络需要配置代理或使用兼容 API：
1
2
| export OPENAI_API_KEY="sk-xxx"
export OPENAI_BASE_URL="https://api.openai.com/v1" # 或代理地址
|
总结
本文完整演示了如何使用 LangChain + ChromaDB 构建一个本地 RAG 知识库问答系统，核心流程为：
- 文档加载：支持 PDF/TXT/MD 等多格式
- 文本切分：递归字符分割，保留语义连贯性
- 向量化存储：ChromaDB 本地持久化，无需额外部署
- 检索增强生成：相似度检索 + Prompt 工程 + LLM 生成
- 进阶优化：混合检索、查询重写、源追溯
整套方案零运维成本，适合个人知识管理、团队文档问答、产品 FAQ 助手等场景。只需在 docs/ 目录放入你的文档，即可快速搭建私有知识库。