# 使用 LangChain + ChromaDB 构建本地 RAG 知识库问答系统
## 背景介绍
企业日常运营中会产生大量的内部文档、技术手册、会议记录。当员工需要从这些海量文档中查找特定信息时，传统的关键词搜索效果往往不理想——它无法理解查询的语义意图，也无法处理自然语言提问。
检索增强生成（Retrieval-Augmented Generation，RAG）技术解决了这个问题。RAG 将大语言模型的语义理解能力与向量检索的精确匹配能力结合起来，用户可以用自然语言向知识库提问，获得准确、相关的回答。
本文手把手教你如何在本地搭建一个完整的 RAG 问答系统。我们使用 LangChain 作为应用框架，ChromaDB 作为向量数据库，OpenAI 的 GPT 模型作为生成引擎。整个系统可以在本地运行，数据隐私有保障。
## 问题描述
企业在构建 RAG 系统时通常面临几个实际困难：
数据导入很繁琐。不同的文档格式（PDF、Word、Markdown）需要不同的解析方式，处理不当会导致信息丢失。检索效果不稳定，简单的向量相似度搜索可能返回与问题相关但不足以回答的内容。系统集成复杂，检索、生成、提示词模板等多个组件串联起来需要不少胶水代码。
本文通过完整的代码示例演示如何解决这些问题。我们构建一个本地知识库问答系统，支持上传文档、向量化存储、自然语言提问等功能。
## 详细步骤
### 环境准备
安装必要的 Python 依赖库：
“`bash
pip install langchain langchain-openai chromadb pypdf python-docx tiktoken
“`
这些库分别提供：langchain 核心框架、OpenAI API 集成、向量数据库支持、PDF 解析、Word 解析、文本分词功能。
### 项目结构
创建以下项目结构：
“`
rag_project/
├── main.py # 主程序入口
├── ingest.py # 文档处理和导入
├── query.py # 问答查询
├── requirements.txt # 依赖列表
└──data/ # 文档存储目录
“`
### 核心代码实现
#### 1. 文档加载与处理（ingest.py）
“`python
from langchain_community.document_loaders import (
PyPDFLoader,
TextLoader,
Docx2txtLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
import os
class DocumentIngester:
def __init__(self, persist_directory=”./chroma_db”):
self.persist_directory = persist_directory
self.embeddings = OpenAIEmbeddings(
model=”text-embedding-3-small”
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
)
def load_document(self, file_path):
“””根据文件类型选择对应的加载器”””
ext = os.path.splitext(file_path)[1].lower()
if ext == ‘.pdf’:
loader = PyPDFLoader(file_path)
elif ext == ‘.docx’:
loader = Docx2txtLoader(file_path)
elif ext == ‘.txt’:
loader = TextLoader(file_path, encoding=’utf-8′)
else:
raise ValueError(f”不支持的文件类型: {ext}”)
return loader.load()
def process_documents(self, file_paths):
“””处理多个文档并返回分块后的文本”””
documents = []
for file_path in file_paths:
docs = self.load_document(file_path)
documents.extend(docs)
# 文本分块
texts = self.text_splitter.split_documents(documents)
return texts
def create_vectorstore(self, texts, collection_name=”knowledge_base”):
“””创建向量数据库”””
vectorstore = Chroma.from_documents(
documents=texts,
embedding=self.embeddings,
persist_directory=self.persist_directory,
collection_name=collection_name
)
return vectorstore
“`
#### 2. 问答系统核心（query.py）
“`python
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
class RAGQuerySystem:
def __init__(self, persist_directory=”./chroma_db”):
self.embeddings = OpenAIEmbeddings(
model=”text-embedding-3-small”
)
self.llm = ChatOpenAI(
model=”gpt-3.5-turbo”,
temperature=0
)
# 加载向量数据库
self.vectorstore = Chroma(
persist_directory=persist_directory,
embedding_function=self.embeddings
)
# 定义提示词模板
self.prompt_template = “””你是一个专业的技术文档问答助手。请根据以下参考文档回答用户的问题。
参考文档：
{context}
用户问题：{question}
要求：
1. 只根据提供的参考文档内容回答，不要编造信息
2. 如果参考文档中没有相关信息，请明确告知用户
3. 回答要准确、简洁、有条理
回答：”””
self.prompt = PromptTemplate(
template=self.prompt_template,
input_variables=[“context”, “question”]
)
# 创建问答链
self.qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type=”stuff”,
retriever=self.vectorstore.as_retriever(
search_kwargs={“k”: 3}
),
chain_type_kwargs={“prompt”: self.prompt}
)
def query(self, question):
“””执行问答查询”””
result = self.qa_chain.invoke({“query”: question})
return result[“result”]
“`
#### 3. 主程序（main.py）
“`python
from ingest import DocumentIngester
from query import RAGQuerySystem
import os
def main():
# 初始化
data_dir = “./data”
persist_dir = “./chroma_db”
# 步骤1：导入文档（首次运行时执行）
if not os.path.exists(persist_dir):
print(“正在导入文档…”)
ingester = DocumentIngester(persist_directory=persist_dir)
# 扫描数据目录下的所有文档
file_paths = []
for root, dirs, files in os.walk(data_dir):
for file in files:
file_path = os.path.join(root, file)
if file.endswith((‘.pdf’, ‘.docx’, ‘.txt’)):
file_paths.append(file_path)
if file_paths:
texts = ingester.process_documents(file_paths)
print(f”成功处理 {len(texts)} 个文档片段”)
vectorstore = ingester.create_vectorstore(texts)
print(“向量数据库创建完成”)
else:
print(“未找到文档文件”)
return
# 步骤2：启动问答系统
print(“\n=== 本地 RAG 知识库问答系统 ===”)
print(“输入问题进行查询，输入 ‘quit’ 退出\n”)
qa_system = RAGQuerySystem(persist_directory=persist_dir)
while True:
question = input(“请输入问题: “)
if question.lower() == ‘quit’:
break
answer = qa_system.query(question)
print(f”\n回答: {answer}\n”)
print(“-” * 50)
if __name__ == “__main__”:
main()
“`
## 运行结果
运行主程序后，系统首先检查向量数据库是否已存在。如果不存在，会自动扫描 data 目录下的文档并创建向量索引。初始化完成后，系统进入交互模式。
“`
$ python main.py
正在导入文档…
成功处理 45 个文档片段
向量数据库创建完成
=== 本地 RAG 知识库问答系统 ===
输入问题进行查询，输入 ‘quit’ 退出
请输入问题: 如何安装 Python 环境?
回答: 根据文档，安装 Python 环境的步骤如下：
1. 访问 python.org 下载最新版本的 Python
2. 运行安装程序，勾选 “Add Python to PATH” 选项
3. 打开终端，输入 python –version 验证安装
请输入问题: 项目的代码规范是什么?
回答: 根据技术规范文档，本项目采用以下代码规范：
– 遵循 PEP 8 Python 代码规范
– 使用 Black 进行代码格式化
– 使用 Flake8 进行代码检查
– 函数和类必须有完整的文档字符串
请输入问题: quit
“`
系统能够准确理解自然语言问题，从向量数据库中检索相关文档片段，生成准确、相关的回答。整个过程完全在本地执行，不需要将敏感文档上传到第三方服务。
## 总结
本文介绍了如何使用 LangChain 和 ChromaDB 在本地构建 RAG 知识库问答系统。通过这个系统，用户可以：
将各种格式的文档（PDF、Word、TXT）导入到向量数据库中，系统会自动进行文本提取和分块处理。使用语义搜索而不是关键词匹配，可以更准确地找到与问题相关的内容。基于检索到的内容生成自然语言回答大大提高问答的准确性。
这个架构有良好的可扩展性。可以更换不同的 embedding 模型，比如本地部署的 BGE，或者用开源的大语言模型（如 Llama 2）替代 OpenAI 的 API，真正实现完全本地化部署。
如果对这个系统感兴趣，可以尝试添加更多功能：支持更多文档格式、添加对话历史记录、实现多轮对话等。这些增强功能可以让系统更好地满足实际业务需求。