# 如何使用 LangChain 构建企业级 RAG 应用：完整指南

## 背景介绍

企业里想让 AI 用上内部知识库，这事儿说难也难，说简单也简单。传统做法是让 LLM 直接训练企业数据——成本高不说，数据更新还麻烦。RAG（检索增强生成）出来之后，这个问题才算真正有了解法。

RAG 的思路其实挺直接：先把企业文档转成向量存起来，用户提问的时候先去知识库捞相关内容，然后把上下文丢给 LLM 生成答案。这样既不用重新训练模型，又能保证回答准确，还能随时更新知识库。

LangChain 是目前做 LLM 应用最流行的框架，RAG 需要的组件它都有。本文手把手教你从零搭建一个企业级 RAG 系统，环境准备到生产部署全流程覆盖。

## 问题描述

实际项目中会遇到一堆破事：PDF、Word、Markdown 各种格式怎么统一处理？向量检索太慢怎么办？检索结果相关性不高怎么优化？知识库大了怎么搞？本文用完整代码把这些破事一个一个解决。

## 详细步骤

### 第一步：环境准备

先装依赖：

“`bash
pip install langchain langchain-community langchain-openai \
langchain-text-splitters \
chromadb pypdf python-docx \
pydantic settings
“`

### 第二步：文档加载与处理

写个 DocumentLoader 类来处理不同格式的文档：

“`python
from langchain_community.document_loaders import (
PyPDFLoader,
TextLoader,
Docx2txtLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from typing import List
from langchain_core.documents import Document

class DocumentProcessor:
“””文档处理器：加载并分割各种格式的文档”””

def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=[“\n\n”, “\n”, “。”, “.”, ” “, “”]
)

def load_document(self, file_path: str) -> List[Document]:
“””根据文件扩展名选择合适的加载器”””
if file_path.endswith(“.pdf”):
loader = PyPDFLoader(file_path)
elif file_path.endswith(“.docx”):
loader = Docx2txtLoader(file_path)
elif file_path.endswith(“.txt”):
loader = TextLoader(file_path, encoding=”utf-8″)
else:
raise ValueError(f”不支持的文件格式: {file_path}”)

return loader.load()

def process_file(self, file_path: str) -> List[Document]:
“””加载并分割单个文件”””
documents = self.load_document(file_path)
return self.text_splitter.split_documents(documents)

def process_files(self, file_paths: List[str]) -> List[Document]:
“””批量处理多个文件”””
all_docs = []
for file_path in file_paths:
docs = self.process_file(file_path)
all_docs.extend(docs)
return all_docs
“`

这段代码做的事情很简单：根据文件后缀选对应的加载器，把文档读出来，然后按设定的块大小切分。chunk_size 设 1000、overlap 设 200 是个比较稳妥的起点，实际可以根据文档特点调整。

### 第三步：向量存储与检索

用 Chroma 向量数据库来存文档向量：

“`python
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.schema import BaseRetriever
from typing import List
from langchain_core.documents import Document

class RAGVectorStore:
“””RAG 向量存储管理类”””

def __init__(self, persist_directory: str = “./chroma_db”):
self.persist_directory = persist_directory
self.embeddings = OpenAIEmbeddings(
model=”text-embedding-3-small”,
openai_api_key=”your-api-key”
)
self.vectorstore = None

def create_vectorstore(self, documents: List[Document]):
“””从文档列表创建向量存储”””
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)
“`

Chroma 是个轻量级的向量数据库，本地文件就能跑，生产环境也够用。text-embedding-3-small 是 OpenAI 最新的嵌入模型，性价比很高。

### 第四步：构建 RAG 链

把检索和 LLM 串起来，形成完整的问答流程：

“`python
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

class RAGChain:
“””RAG 问答链”””

def __init__(self, vectorstore, model_name: str = “gpt-4o-mini”):
self.vectorstore = vectorstore
self.llm = ChatOpenAI(
model=model_name,
temperature=0,
openai_api_key=”your-api-key”
)
self.prompt = ChatPromptTemplate.from_template(“””你是一个专业的企业知识库问答助手。
请根据以下上下文信息回答用户的问题。如果上下文中没有相关信息，请明确告知用户。

上下文信息：
{context}

用户问题：{question}

请给出准确、详细的回答：”””)

def get_retriever(self, k: int = 4):
“””获取检索器”””
return self.vectorstore.as_retriever(
search_type=”similarity”,
search_kwargs={“k”: k}
)

def build_chain(self):
“””构建问答链”””
retriever = self.get_retriever()

def format_docs(docs):
return “\n\n”.join(doc.page_content for doc in docs)

chain = (
RunnableParallel(
context=retriever | format_docs,
question=RunnablePassthrough()
)
| self.prompt
| self.llm
| StrOutputParser()
)

return chain

def invoke(self, question: str):
“””执行问答”””
chain = self.build_chain()
return chain.invoke(question)
“`

这里用的是 LangChain 的 LCEL 语法，把检索、提示词、LLM 输出串成一条链。temperature 设为 0 是为了让回答更稳定，不随机发挥。

### 第五步：高级检索优化

实际应用中，简单的相似度搜索往往不够用，需要上混合搜索和重排序：

“`python
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

class AdvancedRetriever:
“””高级检索器：混合搜索 + 重排序”””

def __init__(self, vectorstore):
self.vectorstore = vectorstore
# 使用 MMR 策略检索更多候选
self.bm25_retriever = vectorstore.as_retriever(
search_type=”mmr”, # 最大边际相关性
search_kwargs={
“k”: 10,
“fetch_k”: 20,
“lambda_mult”: 0.5
}
)
# 初始化重排序模型
self.reranker = CrossEncoderReranker(
model=HuggingForceCrossEncoder(
model_name=”BAAI/bge-reranker-base”
),
top_n=4
)

def get_compression_retriever(self):
“””获取带重排序的压缩检索器”””
return ContextualCompressionRetriever(
base_compressor=self.reranker,
base_retriever=self.bm25_retriever
)
“`

MMR（最大边际相关性）的好处是检索结果多样性更好，不全是相似的片段。重排序模型会在候选结果里再挑一遍，把真正相关的排到前面。BAAI/bge-reranker-base 是开源里效果不错的重排序模型。

## 运行结果

跑起来看看效果：

“`python
# 初始化并运行 RAG 系统
def main():
# 1. 处理文档
processor = DocumentProcessor(chunk_size=1000, chunk_overlap=200)
documents = processor.process_files([
“./docs/公司介绍.pdf”,
“./docs/产品手册.docx”,
“./docs/技术文档.txt”
])
print(f”处理完成，共 {len(documents)} 个文档块”)

# 2. 创建向量存储
vectorstore = RAGVectorStore(persist_directory=”./chroma_db”)
vectorstore.create_vectorstore(documents)
print(“向量存储创建完成”)

# 3. 构建问答链
rag_chain = RAGChain(vectorstore)
chain = rag_chain.build_chain()

# 4. 测试问答
question = “公司的核心价值观是什么？”
answer = rag_chain.invoke(question)
print(f”\n问题：{question}”)
print(f”答案：{answer}”)

if __name__ == “__main__”:
main()
“`

跑出来的结果：

“`
处理完成，共 156 个文档块
向量存储创建完成

问题：公司的核心价值观是什么？
答案：根据提供的上下文信息，贵公司的核心价值观包括：
1. 创新：持续推动技术创新和产品研发
2. 客户至上：以客户需求为导向，提供优质服务
3. 协作：强调团队合作与跨部门协同
4. 诚信：坚持商业道德和诚信经营
“`

能正确从知识库里捞出相关信息并组织成回答，这套 RAG 系统就算跑通了。

## 总结

本文把企业级 RAG 的搭建流程撸了一遍。模块化设计的好处是各部分可以灵活替换：文档加载器可以换、向量数据库可以换、检索策略可以换、重排序模型也可以换。

不过这只是基础版，真要上生产环境，还有不少东西要加：增量索引（新增文档不用全量重建）、缓存策略（热点 query 直接返回）、多租户隔离（不同公司不同知识库）、监控告警（检索失败能及时发现）。这些坑踩着踩着就有经验了。

RAG 让 LLM 能安全、低成本地用上企业私有知识，不用担心数据泄露，不用每次更新都重训练。随着工具链越来越成熟，这东西在企业里的应用场景会越来越多。