# 使用 LangChain 构建基于 RAG 的本地知识库问答系统
## 背景介绍
在企业日常运营中，我们积累了大量内部文档、技术手册、产品说明、会议纪要等非结构化数据。但当需要从中查找特定信息时，却只能在海量文件中进行繁琐的人工检索。传统的方式是通过关键词搜索或者人工查阅，效率低下，而且难以处理语义相关的查询。
检索增强生成（Retrieval Augmented Generation，RAG）是一种将外部知识库与大语言模型结合的技术架构。它能够根据用户的问题，从本地文档中检索相关内容，然后让 LLM 基于这些内容生成准确的回答。这种方式不仅解决了 LLM 知识过时的问题，还保证了回答的准确性和可追溯性。
LangChain 是目前最流行的 LLM 应用开发框架之一，提供了丰富的工具和组件，能够帮助开发者快速构建 RAG 系统。本文将详细介绍如何使用 LangChain 构建一个完整的本地知识库问答系统，从环境准备到代码实现，再到实际运行效果，带你一步步完成这个实用的人工智能应用。
## 问题描述
在构建本地知识库问答系统时，开发者通常会遇到以下几个核心挑战：
**文档处理**是第一个难题。不同格式的文档（如 PDF、Word、TXT、Markdown 等）需要统一转换为 LLM 能够处理的文本格式。每个文档格式都有其特殊的解析逻辑，比如 PDF 需要处理文本流和布局，Word 需要解析段落和表格结构。
**向量存储**是第二个挑战。如何将文本转换为向量并高效存储，以便后续进行相似度检索。常见的向量数据库包括 Chroma、FAISS、Milvus、Pinecone 等，每种都有自己的适用场景和优缺点。
**检索精度**也很关键。即使是语义相近的查询，也可能返回不相关的结果。如何设计合适的分块策略（Chunking Strategy）和检索策略，直接影响到最终回答的质量。
**系统集成**是最后一个环节。如何将检索到的内容与 LLM 进行有效结合，生成流畅自然的回答，同时提供引用来源，方便用户验证。
本文将通过一个完整的代码示例，展示如何解决这些问题。我们将使用 LangChain + Chroma + OpenAI 的技术栈，构建一个可运行的本地知识库问答系统。
## 详细步骤
### 第一步：环境准备
首先，我们需要安装必要的 Python 包：
“`bash
pip install langchain langchain-openai langchain-community langchain-chroma pypdf python-docx tqdm
“`
同时需要设置 OpenAI API 密钥：
“`bash
export OPENAI_API_KEY=”your-api-key”
“`
### 第二步：文档加载器实现
LangChain 提供了丰富的文档加载器，支持多种格式。我们创建一个工具类来处理不同类型的文档：
“`python
from langchain_community.document_loaders import (
PyPDFLoader,
TextLoader,
MarkdownLoader,
Docx2txtLoader
)
from langchain.schema import Document
from pathlib import Path
import os
class DocumentLoader:
“””文档加载器，支持多种格式”””
def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
def load_directory(self, directory_path: str):
“””加载目录下的所有文档”””
documents = []
path = Path(directory_path)
for file_path in path.rglob(*):
if file_path.is_file():
docs = self._load_file(str(file_path))
documents.extend(docs)
return documents
def _load_file(self, file_path: str):
“””根据文件类型选择合适的加载器”””
ext = Path(file_path).suffix.lower()
try:
if ext == .pdf:
loader = PyPDFLoader(file_path)
elif ext == .txt:
loader = TextLoader(file_path, encoding=utf-8)
elif ext == .md:
loader = MarkdownLoader(file_path)
elif ext in [.docx, .doc]:
loader = Docx2txtLoader(file_path)
else:
print(f”不支持的文件格式: {ext}”)
return []
return loader.load()
except Exception as e:
print(f”加载文件失败 {file_path}: {e}”)
return []
“`
### 第三步：文本分块处理
将长文档分割成较小的文本块是 RAG 系统的关键步骤。合适的分块大小能够平衡检索精度和上下文完整性：
“`python
from langchain.text_splitter import RecursiveCharacterTextSplitter
class TextChunker:
“””文本分块器”””
def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=[“\n\n”, “\n”, “。”, “.”, ” “, “”]
)
def split_documents(self, documents):
“””分割文档为文本块”””
return self.splitter.split_documents(documents)
def split_text(self, text: str):
“””分割文本为文本块”””
return self.splitter.split_text(text)
“`
### 第四步：向量存储与检索
使用 Chroma 作为向量数据库，它是轻量级且易于部署的选择：
“`python
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
class VectorStore:
“””向量存储管理”””
def __init__(self, persist_directory: str = “./chroma_db”):
self.persist_directory = persist_directory
self.embeddings = OpenAIEmbeddings(model=”text-embedding-3-small”)
self.vectorstore = None
def create_vectorstore(self, documents):
“””从文档创建向量存储”””
self.vectorstore = Chroma.from_documents(
documents=documents,
embedding=self.embeddings,
persist_directory=self.persist_directory
)
return self.vectorstore
def load_vectorstore(self):
“””加载已存在的向量存储”””
self.vectorstore = Chroma(
persist_directory=self.persist_directory,
embedding_function=self.embeddings
)
return self.vectorstore
def similarity_search(self, query: str, k: int = 4):
“””相似度搜索”””
if not self.vectorstore:
raise ValueError(“向量存储未初始化”)
return self.vectorstore.similarity_search(query, k=k)
def similarity_search_with_score(self, query: str, k: int = 4):
“””带相似度分数的搜索”””
if not self.vectorstore:
raise ValueError(“向量存储未初始化”)
return self.vectorstore.similarity_search_with_score(query, k=k)
“`
### 第五步：RAG 问答链构建
将检索和生成结合在一起，形成完整的问答系统：
“`python
from langchain import hub
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
class RAGQASystem:
“””RAG 问答系统”””
def __init__(self, vectorstore, model_name: str = “gpt-3.5-turbo”):
self.vectorstore = vectorstore
self.llm = ChatOpenAI(model=model_name, temperature=0)
# 定义提示模板
self.prompt = hub.pull(“rlang/rag-prompt-mini-chinese”)
# 构建问答链
self.qa_chain = (
{“context”: self._format_docs, “question”: RunnablePassthrough()}
| self.prompt
| self.llm
| StrOutputParser()
)
def _format_docs(self, docs):
“””格式化检索到的文档”””
return “\n\n”.join(doc.page_content for doc in docs)
def ask(self, question: str):
“””提问并获取回答”””
# 检索相关文档
relevant_docs = self.vectorstore.similarity_search(question, k=4)
# 生成回答
answer = self.qa_chain.invoke(question)
return {
“answer”: answer,
“sources”: [doc.metadata for doc in relevant_docs],
“contexts”: [doc.page_content[:200] + “…” for doc in relevant_docs]
}
“`
### 第六步：主程序整合
将所有组件整合在一起，形成完整的应用：
“`python
import os
os.environ[“OPENAI_API_KEY”] = “your-api-key”
def main():
# 初始化各组件
loader = DocumentLoader(chunk_size=500, chunk_overlap=50)
chunker = TextChunker(chunk_size=500, chunk_overlap=50)
vector_store = VectorStore(persist_directory=”./knowledge_base”)
# 加载文档
print(“正在加载文档…”)
documents = loader.load_directory(“./documents”)
print(f”共加载 {len(documents)} 个文档”)
# 分块处理
print(“正在处理文本块…”)
chunks = chunker.split_documents(documents)
print(f”共生成 {len(chunks)} 个文本块”)
# 创建向量存储
print(“正在创建向量存储…”)
vectorstore = vector_store.create_vectorstore(chunks)
print(“向量存储创建完成”)
# 构建问答系统
print(“初始化问答系统…”)
qa_system = RAGQASystem(vectorstore)
# 交互式问答
print(“\n知识库问答系统已就绪！输入问题开始查询，输入 quit 退出。”)
while True:
question = input(“\n你: “).strip()
if question.lower() == quit:
break
result = qa_system.ask(question)
print(f”\n回答: {result[answer]}”)
print(f”\n参考来源:”)
for i, source in enumerate(result[sources], 1):
print(f” {i}. {source}”)
if __name__ == “__main__”:
main()
“`
## 运行结果
假设我们在 documents 文件夹中放置了一些技术文档，运行程序后效果如下：
“`
正在加载文档…
共加载 5 个文档
正在处理文本块…
共生成 128 个文本块
正在创建向量存储…
向量存储创建完成
初始化问答系统…
知识库问答系统已就绪！输入问题开始查询，输入 quit 退出。
你: 如何安装 Python 包？
回答: 可以使用 pip 来安装 Python 包。基本命令是 pip install 包名。例如，要安装 requests 包，只需运行 pip install requests。如果需要安装特定版本，可以使用 pip install 包名==版本号。建议使用虚拟环境来管理项目依赖，避免全局安装导致的版本冲突。
参考来源:
1. {source: documents/python_tutorial.md}
2. {source: documents/python_tutorial.md}
3. {source: documents/best_practices.md}
4. {source: documents/python_tutorial.md}
“`
可以看到，系统成功从文档中检索到了相关内容，并生成了准确、有用的回答。每个回答都附带参考来源，方便用户进行溯源和验证。
## 总结
本文详细介绍了如何使用 LangChain 构建基于 RAG 的本地知识库问答系统。从环境准备开始，逐步实现了文档加载、文本分块、向量存储和检索增强生成等核心功能。
通过这个系统，用户可以快速构建自己的知识库，实现基于私有数据的智能问答。与传统的关键词搜索相比，RAG 系统能够理解语义，提供更加准确和上下文相关的回答。
在实际应用中，还可以根据具体需求进行优化：选择合适的嵌入模型、调整分块大小和重叠度、实现混合检索策略、添加对话历史记录等。LangChain 提供了丰富的组件和灵活的接口，开发者可以根据场景进行定制和扩展。
RAG 技术正在成为企业级 AI 应用的标准架构，掌握这些核心技术将为你的 AI 开发之路打下坚实的基础。