基于 LangChain 开发大模型 RAG 知识问答应用

3. 文档加载与分割

使用 DirectoryLoader 加载本地文档，并通过 RecursiveCharacterTextSplitter 进行分割。合理的 Chunk Size 和重叠率有助于保持语义完整性。

from langchain_community.document_loaders import DirectoryLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = DirectoryLoader("./data", glob="**/*.txt")
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ".", " ", ""]
)
chunks = splitter.split_documents(docs)

4. 文本嵌入与向量存储

使用阿里云百炼 DashScope Embedding 模型生成向量，并存储至 FAISS 向量数据库。

from langchain.embeddings import DashScopeEmbeddings
from langchain.vectorstores import FAISS

embeddings = DashScopeEmbeddings(model="text-embedding-v2")
vectorstore = FAISS.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

5. ReRank 重排优化

为了提升检索精度，引入 BAAI/bge-reranker-large 模型对初步检索结果进行重排序。需自定义 BaseDocumentCompressor。

from langchain.retrievers.document_compressors import BaseDocumentCompressor
from typing import List
from FlagEmbedding import FlagReranker

class BGERerankCompressor(BaseDocumentCompressor):
    def __init__(self, reranker_model: str):
        self.reranker = FlagReranker(reranker_model, use_fp16=True)

    def compress_documents(self, documents, query, top_n=2):
        scores = self.reranker.compute_score([[query, doc.page_content] for doc in documents])
        sorted_docs = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
        return [doc for doc, _ in sorted_docs[:top_n]]

reranker_compressor = BGERerankCompressor("BAAI/bge-reranker-large")

整合检索与重排逻辑：

from langchain.retrievers import ContextualCompressionRetriever

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

6. 构建 LCEL 链

利用 LangChain Expression Language (LCEL) 编排流程。使用 RunnableParallel 并行处理问题和检索内容，再通过 RunnablePassthrough 传递原始输入。

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

chain = (
    {"context": compression_retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
)

7. 回调函数与输出解析

继承 BaseCallbackHandler 可监控链的运行状态。输出端使用 StrOutputParser 提取模型生成的纯文本。

from langchain_core.output_parsers import StrOutputParser

output_chain = chain | StrOutputParser()

8. 效果测试

假设知识库中包含人物介绍信息，测试问答效果。

response = output_chain.invoke("林悦喜欢吃什么？")
print(response)

预期输出示例：

根据您的问题，我找到的相关答案是：林悦喜欢吃火锅，尤其是涮毛肚和鸭肠...

若知识库无相关内容：

在资料中没有相关信息。

总结

本教程展示了如何从零搭建一个基于 LangChain 的 RAG 系统。通过集成 DeepSeek 推理能力、阿里云 Embedding 服务以及 BGE 重排模型，实现了高质量的知识问答应用。关键点在于合理的文档分块策略、向量检索优化以及严格的提示词约束，有效降低了大模型的幻觉风险。