LangChain 实战：构建 RAG 应用

什么是 RAG

RAG（Retrieval-Augmented Generation，检索增强生成）是目前最实用的 AI 应用模式之一。它的核心思想很简单：先从你的文档中检索相关内容，再把这些内容作为上下文交给 LLM 生成回答。

为什么需要 RAG？因为 LLM 有两个天然的局限：

知识截止日期——它不知道训练数据之后发生的事
没有你的私有数据——它不知道你公司的文档、代码库、知识库里有什么

RAG 解决了这两个问题。你可以把任何文档喂给 RAG 系统，然后基于这些文档进行问答。

RAG 的工作流程

文档 → 切分 → Embedding → 存入向量数据库
                                    ↓
用户提问 → Embedding → 向量检索 → 获取相关文档片段
                                    ↓
                    相关文档 + 用户问题 → LLM → 回答

整个流程分为两个阶段：

阶段	步骤	说明
索引阶段	加载文档	读取各种格式的文档
	切分文档	将长文档切成小块
	生成 Embedding	将文本转换为向量
	存入向量库	存储向量以供检索
查询阶段	用户提问	接收用户的问题
	向量检索	找到最相关的文档片段
	生成回答	LLM 基于检索结果回答

环境准备

安装依赖

pip install langchain langchain-openai langchain-community
pip install chromadb          # 向量数据库
pip install pypdf             # PDF 文档加载
pip install tiktoken          # Token 计数
pip install unstructured      # 通用文档加载

配置

import os
os.environ["OPENAI_API_KEY"] = "your-api-key"

第一步：文档加载

LangChain 提供了丰富的文档加载器（Document Loaders）。

加载 PDF

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("docs/technical-spec.pdf")
documents = loader.load()

print(f"加载了 {len(documents)} 页")
print(f"第一页内容预览: {documents[0].page_content[:200]}")
print(f"元数据: {documents[0].metadata}")
# {'source': 'docs/technical-spec.pdf', 'page': 0}

加载 Markdown

from langchain_community.document_loaders import UnstructuredMarkdownLoader

loader = UnstructuredMarkdownLoader("docs/README.md")
documents = loader.load()

加载网页

from langchain_community.document_loaders import WebBaseLoader

loader = WebBaseLoader("https://docs.python.org/3/tutorial/classes.html")
documents = loader.load()

加载整个目录

from langchain_community.document_loaders import DirectoryLoader

loader = DirectoryLoader(
    "docs/",
    glob="**/*.md",
    show_progress=True
)
documents = loader.load()
print(f"加载了 {len(documents)} 个文档")

常用加载器一览

加载器	支持格式	包
PyPDFLoader	PDF	`pypdf`
TextLoader	TXT	内置
CSVLoader	CSV	内置
UnstructuredMarkdownLoader	Markdown	`unstructured`
WebBaseLoader	网页	`beautifulsoup4`
GitLoader	Git 仓库	`gitpython`
NotionDirectoryLoader	Notion 导出	内置

第二步：文档切分

加载的文档通常很长，需要切分成小块。切分策略直接影响检索质量。

基本切分

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,       # 每块最大字符数
    chunk_overlap=200,     # 块之间的重叠字符数
    length_function=len,
    separators=["\n\n", "\n", "。", ".", " ", ""]
)

chunks = text_splitter.split_documents(documents)
print(f"切分为 {len(chunks)} 个块")
print(f"第一块: {chunks[0].page_content[:100]}...")

为什么需要 overlap

overlap（重叠）是为了避免重要信息被切断。比如一个段落刚好在切分点被分成两块，有了 overlap，两块都会包含这个段落的内容。

切分策略选择

策略	适用场景	说明
RecursiveCharacterTextSplitter	通用	按层级分隔符递归切分
MarkdownHeaderTextSplitter	Markdown	按标题层级切分
TokenTextSplitter	精确控制	按 token 数切分
CodeTextSplitter	代码	按编程语言语法切分

代码文档的切分

如果你的文档包含代码，使用专门的代码切分器：

from langchain_text_splitters import Language, RecursiveCharacterTextSplitter

python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1000,
    chunk_overlap=100
)

# 切分 Python 代码
code_chunks = python_splitter.split_documents(code_documents)

第三步：生成 Embedding

Embedding 是将文本转换为向量（一组数字），语义相似的文本会有相似的向量。

使用 OpenAI Embedding

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 测试 Embedding
test_text = "Python 是一种编程语言"
vector = embeddings.embed_query(test_text)
print(f"向量维度: {len(vector)}")  # 1536
print(f"前 5 个值: {vector[:5]}")

Embedding 模型选择

模型	维度	价格	质量
text-embedding-3-small	1536	$0.02/1M tokens	好
text-embedding-3-large	3072	$0.13/1M tokens	更好
text-embedding-ada-002	1536	$0.10/1M tokens	旧版

对于大多数场景，text-embedding-3-small 就够用了，性价比最高。

第四步：存入向量数据库

使用 Chroma

Chroma 是一个轻量级的向量数据库，适合开发和小规模应用：

from langchain_community.vectorstores import Chroma

# 创建向量数据库并存入文档
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 持久化到磁盘
)

print(f"存入 {vectorstore._collection.count()} 个文档块")

加载已有的向量数据库

# 下次使用时直接加载
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

向量数据库选择

数据库	类型	适用场景
Chroma	嵌入式	开发、小规模
FAISS	嵌入式	大规模、高性能
Pinecone	云服务	生产环境、托管
Weaviate	自托管/云	企业级
Qdrant	自托管/云	高性能、过滤
pgvector	PostgreSQL 扩展	已有 PG 的项目

第五步：检索

基本检索

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}  # 返回最相关的 4 个文档块
)

# 检索
docs = retriever.invoke("如何配置数据库连接？")
for doc in docs:
    print(f"来源: {doc.metadata.get('source', 'unknown')}")
    print(f"内容: {doc.page_content[:100]}...")
    print("---")

检索策略

策略	说明	参数
similarity	余弦相似度	`k`: 返回数量
mmr	最大边际相关性	`k`, `fetch_k`, `lambda_mult`
similarity_score_threshold	带阈值的相似度	`k`, `score_threshold`

MMR（Maximal Marginal Relevance）会在相关性和多样性之间取平衡，避免返回内容重复的文档：

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,
        "fetch_k": 20,       # 先获取 20 个候选
        "lambda_mult": 0.5   # 多样性权重
    }
)

第六步：生成回答

完整的 RAG Chain

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# RAG 提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个知识库问答助手。
    请根据以下检索到的文档内容回答用户的问题。
    如果文档中没有相关信息，请明确说明你无法从现有文档中找到答案。
    不要编造信息。

    检索到的文档：
    {context}"""),
    ("human", "{question}")
])

# 格式化文档
def format_docs(docs):
    return "\n\n---\n\n".join(
        f"来源: {doc.metadata.get('source', 'unknown')}\n{doc.page_content}"
        for doc in docs
    )

# 构建 RAG Chain
rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 使用
answer = rag_chain.invoke("如何配置数据库连接？")
print(answer)

带来源引用的 RAG

from langchain_core.runnables import RunnableParallel

# 同时返回答案和来源文档
rag_chain_with_sources = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
).assign(
    answer=lambda x: (
        prompt.format_messages(
            context=format_docs(x["context"]),
            question=x["question"]
        )
        |> llm
        |> StrOutputParser()
    )
)

# 或者更简单的方式
def rag_with_sources(question: str):
    docs = retriever.invoke(question)
    context = format_docs(docs)

    answer = (prompt | llm | StrOutputParser()).invoke({
        "context": context,
        "question": question
    })

    sources = list(set(
        doc.metadata.get("source", "unknown")
        for doc in docs
    ))

    return {
        "answer": answer,
        "sources": sources
    }

result = rag_with_sources("如何配置数据库连接？")
print(f"回答: {result['answer']}")
print(f"来源: {result['sources']}")

完整代码示例

把所有步骤整合在一起：

"""
完整的 RAG 应用示例
从文档加载到问答的完整流程
"""

import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 配置
os.environ["OPENAI_API_KEY"] = "your-api-key"
DOCS_DIR = "./docs"
DB_DIR = "./chroma_db"

def build_index():
    """构建文档索引（只需运行一次）"""
    # 1. 加载文档
    loader = DirectoryLoader(DOCS_DIR, glob="**/*.pdf", loader_cls=PyPDFLoader)
    documents = loader.load()
    print(f"加载了 {len(documents)} 个文档")

    # 2. 切分
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200
    )
    chunks = splitter.split_documents(documents)
    print(f"切分为 {len(chunks)} 个块")

    # 3. 生成 Embedding 并存入向量库
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=DB_DIR
    )
    print(f"索引构建完成，共 {vectorstore._collection.count()} 个块")
    return vectorstore

def create_rag_chain(vectorstore):
    """创建 RAG Chain"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

    prompt = ChatPromptTemplate.from_messages([
        ("system", """根据以下文档内容回答问题。
        如果文档中没有相关信息，请说明。

        文档内容：
        {context}"""),
        ("human", "{question}")
    ])

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

    chain = (
        {"context": retriever | format_docs, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )
    return chain

# 使用
if __name__ == "__main__":
    # 首次运行：构建索引
    # vectorstore = build_index()

    # 后续运行：加载已有索引
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma(
        persist_directory=DB_DIR,
        embedding_function=embeddings
    )

    chain = create_rag_chain(vectorstore)

    # 交互式问答
    while True:
        question = input("\n请输入问题（输入 q 退出）: ")
        if question.lower() == 'q':
            break
        answer = chain.invoke(question)
        print(f"\n回答: {answer}")

优化技巧

1. 调整 chunk_size

chunk_size 太小会丢失上下文，太大会引入噪音。一般建议：

技术文档：800-1200 字符
对话记录：400-600 字符
代码文档：按函数/类切分

2. 优化检索

# 使用 Ensemble Retriever 组合多种检索策略
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# 关键词检索
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 4

# 向量检索
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 组合
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]
)

3. 添加重排序

检索后对结果进行重排序，提高相关性：

from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

model = HuggingFaceCrossEncoder(model_name="cross-encoder/ms-marco-MiniLM-L-6-v2")
compressor = CrossEncoderReranker(model=model, top_n=3)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

4. 处理中文文档

中文文档切分需要注意分隔符的选择：

chinese_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=100,
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
)

常见问题

Q: 检索结果不相关怎么办？

检查 chunk_size 是否合适
尝试不同的检索策略（MMR、Ensemble）
优化提示词，让 LLM 更好地利用上下文
添加重排序步骤

Q: 回答出现幻觉怎么办？

在提示词中强调”只根据文档内容回答”
降低 temperature（设为 0）
要求 LLM 引用来源
添加后处理验证步骤

Q: 文档量很大怎么办？

使用更高效的向量数据库（FAISS、Qdrant）
实现增量索引
添加元数据过滤，缩小检索范围
考虑分层检索（先粗筛再精排）

RAG 不是银弹，但它是目前让 LLM 理解你私有数据最实用的方案。从简单开始，根据实际效果逐步优化。

加载中...