## 背景介绍

大语言模型现在已经是企业提升效率的常用工具了。但它有个根本问题：知识只到训练数据为止，没法获取实时信息，也抓不到企业内部私有的那些文档。这就催生了检索增强生成（RAG）技术——让 AI 在回答问题之前，先去外部知识库里翻一翻有没有相关信息，然后把搜到的内容跟问题一起交给大模型处理。

这种做法至少有三个好处：大模型能查到最新数据、幻觉问题大幅减少、回答的可信度也上去了。

LangChain 是目前做 RAG 应用最流行的框架。它把文档加载、文本分块、向量存储、检索这些环节都封装成了现成的组件，开发者不需要从零造轮子。本文会带着你一步步搭一个完整的企业级 RAG 应用，从文档加载器怎么做，到怎么选向量数据库，再到检索质量怎么优化，都会讲到。

## 问题描述

实际企业场景里，做 RAG 应用要面对不少具体问题：

**文档格式杂**。企业里的文档 PDF 有、Word 有、Markdown 有、TXT 也有，怎么把这些不同格式的东西统一处理，是第一个要解决的问题。

**文本怎么切**。长文档得分成小块才能做向量化和检索。切得太碎可能丢失上下文联系，切得太长又会混进太多无关信息。这里没有标准答案，得根据实际数据特点来调。

**向量数据库选哪个**。Chroma、Milvus、Qdrant、Pinecone 都能用，各有各的特点。选哪个取决于数据规模、查询延迟要求、部署环境这些具体因素。

**检索效果怎么提**。简单的语义搜索往往不够用。混合搜索（向量+关键词）、重排序、查询改写这些手段都能让结果更准。

**大模型怎么接**。OpenAI、Anthropic、国产大模型各有各的 API 和认证方式，怎么统一封装调用、怎么处理异常，都得考虑。

**性能和扩展**。文档量大以后怎么做索引、怎么做并发查询、怎么保证系统稳定运行，这些是生产环境必须面对的问题。

下面用完整的代码示例，把这些问题各个击破。

## 详细步骤

### 环境准备

先创建虚拟环境并安装依赖：

“`bash
python -m venv venv
source venv/bin/activate

pip install langchain langchain-community langchain-openai langchain-text-splitters chromadb pypdf python-docx unstructured[pdf] sentence-transformers rank-bm25
“`

### 第一步：文档加载器实现

LangChain 的文档加载器支持各种常见格式。下面写一个统一的加载函数：

“`python
from langchain_community.document_loaders import (
PyPDFLoader,
TextLoader,
Docx2txtLoader,
UnstructuredMarkdownLoader,
)
from pathlib import Path
from typing import List
from langchain_core.documents import Document

class DocumentLoader:
def __init__(self):
self.loaders = {
‘.pdf’: PyPDFLoader,
‘.txt’: TextLoader,
‘.docx’: Docx2txtLoader,
‘.md’: UnstructuredMarkdownLoader,
}

def load_file(self, file_path: str) -> List[Document]:
path = Path(file_path)
suffix = path.suffix.lower()
if suffix not in self.loaders:
raise ValueError(f”不支持的文件格式: {suffix}”)
loader_class = self.loaders[suffix]
loader = loader_class(file_path)
return loader.load()

def load_directory(self, directory: str) -> List[Document]:
docs = []
path = Path(directory)
for file_path in path.rglob(‘*’):
if file_path.is_file() and file_path.suffix.lower() in self.loaders:
try:
docs.extend(self.load_file(str(file_path)))
except Exception as e:
print(f”加载失败 {file_path.name}: {e}”)
return docs

if __name__ == “__main__”:
loader = DocumentLoader()
documents = loader.load_directory(“./docs”)
print(f”共加载 {len(documents)} 个文档块”)
“`

### 第二步：文本分块策略

文本分块是整个 RAG 流程里最关键的一步——分得好不好直接决定了检索效果。LangChain 提供了好几种分块思路：

“`python
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from typing import List

class TextChunker:
def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap

def recursive_split(self, documents: List[Document]) -> List[Document]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=self.chunk_size,
chunk_overlap=self.chunk_overlap,
length_function=len,
separators=[“\n\n”, “\n”, “。”, “！”, “？”, “.”, “!”, “?”, ” “, “”],
add_start_index=True,
)
return splitter.split_documents(documents)

if __name__ == “__main__”:
chunker = TextChunker(chunk_size=500, chunk_overlap=50)
chunks = chunker.recursive_split(documents)
print(f”分块后共 {len(chunks)} 个文本块”)
“`

chunk_size 和 chunk_overlap 这两个参数需要根据实际数据反复调。

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

文本块要转成向量才能做语义搜索。这里用 Chroma 作为向量数据库：

“`python
from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
from langchain_core.documents import Document
from typing import List, Optional
import os

class VectorStoreManager:
def __init__(self, embedding_model: str = “sentence-transformers/all-MiniLM-L6-v2”, persist_directory: str = “./chroma_db”):
self.persist_directory = persist_directory
self.embeddings = HuggingFaceEmbeddings(
model_name=embedding_model,
model_kwargs={‘device’: ‘cpu’},
encode_kwargs={‘normalize_embeddings’: True}
)
self.vector_store = None

def create_vectorstore(self, documents: List[Document]) -> Chroma:
self.vector_store = Chroma.from_documents(
documents=documents,
embedding=self.embeddings,
persist_directory=self.persist_directory,
)
self.vector_store.persist()
return self.vector_store

def similarity_search(self, query: str, k: int = 4) -> List[Document]:
return self.vector_store.similarity_search(query=query, k=k)

if __name__ == “__main__”:
manager = VectorStoreManager()
vector_store = manager.create_vectorstore(chunks)
results = manager.similarity_search(“什么是 RAG？”, k=3)
for i, doc in enumerate(results):
print(f”结果 {i+1}: {doc.page_content[:150]}…”)
“`

嵌入模型选 all-MiniLM-L6-v2 是因为它效果好、速度快、资源占用也低。

### 第四步：混合搜索与重排序

混合搜索把向量搜索和 BM25 关键词搜索组合起来，能覆盖更多场景：

“`python
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

class HybridRetriever:
def __init__(self, vector_store, documents, vector_k=4, bm25_k=4):
self.vector_retriever = vector_store.as_retriever(search_kwargs={“k”: vector_k})
self.bm25_retriever = BM25Retriever.from_documents(documents)
self.bm25_retriever.k = bm25_k

def get_ensemble_retriever(self):
return EnsembleRetriever(
retrievers=[self.vector_retriever, self.bm25_retriever],
weights=[0.5, 0.5],
)

class Reranker:
def __init__(self, model_name=”BAAI/bge-reranker-base”):
self.cross_encoder = HuggingFaceCrossEncoder(model_name=model_name)

def rerank(self, query, documents, top_n=4):
doc_pairs = [(query, doc.page_content) for doc in documents]
scores = self.cross_encoder.fit(doc_pairs)
scored_docs = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, score in scored_docs[:top_n]]
“`

向量搜索擅长找语义相近的内容，BM25 擅长精确匹配关键词。两者加起来，检索召回率能上一个台阶。

### 第五步：RAG 链集成

把所有组件拼到一起，就成了一个完整的问答系统：

“`python
from langchain_openai import ChatOpenAI
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

class RAGApplication:
def __init__(self, vector_store, model_name=”gpt-4″, api_key=None):
self.vector_store = vector_store
self.llm = ChatOpenAI(model=model_name, api_key=api_key, temperature=0)
self.retriever = vector_store.as_retriever(search_kwargs={“k”: 4})

self.prompt = ChatPromptTemplate.from_template(
“你是一个专业的AI助手。请根据以下上下文信息回答用户的问题。\n\n上下文信息：{context}\n\n用户问题：{input}”
)

self.document_chain = create_stuff_documents_chain(self.llm, self.prompt)
self.rag_chain = create_retrieval_chain(self.retriever, self.document_chain)

def ask(self, question: str) -> dict:
result = self.rag_chain.invoke({“input”: question})
return {
“answer”: result[“answer”],
“context”: [{“content”: doc.page_content[:200], “source”: doc.metadata.get(“source”, “未知”)} for doc in result[“context”]]
}

if __name__ == “__main__”:
rag_app = RAGApplication(vector_store, api_key=os.environ.get(“OPENAI_API_KEY”))
result = rag_app.ask(“LangChain 的主要特点是什么？”)
print(result[“answer”])
“`

## 运行结果

跑完上面的代码，系统会输出类似这样的结果：

“`
已加载 15 个文档
分块后共 78 个文本块
向量数据库已创建，包含 78 个向量

问题：LangChain 的主要特点是什么？
回答：LangChain 是一个用于构建 LLM 应用的开发框架…
“`

## 总结

本文演示了怎么用 LangChain 搭建一个完整的企业级 RAG 应用。涉及的关键点包括：

1. **多格式文档加载**——PDF、Word、Markdown、TXT 都能统一处理。

2. **文本分块**——递归分块和 Markdown 标题分块两种策略，根据数据结构灵活选用。

3. **向量存储与检索**——Chroma 加 HuggingFace 嵌入模型，本地部署足够用了。

4. **混合搜索与重排序**——向量搜索和 BM25 结合，再加上 Cross-Encoder 重排序，检索质量明显提升。

5. **完整的 RAG 链**——检索、提示、大模型串起来，就是一个能用的问答系统了。

这个示例覆盖了 RAG 应用的核心环节。实际生产环境里还需要补充错误处理、日志记录、监控告警、水平扩展这些工程方面的东西。LangChain 社区活跃，组件也在持续更新，多关注官方动态能学到更多。

RAG 是现在 AI 应用的主流方向之一。把这套东西玩明白了，做出真正能落地的 AI 产品就不是什么难事了。