使用 LangChain 1.0 构建 RAG 流水线：实战指南

温馨提示：LangChain更新速度飞快，很多库的位置一直再变动，因此导包的时候需要结合自己的版本。

引言

检索增强生成（Retrieval-Augmented Generation, RAG）解决了 AI 开发中的一个核心难题：让大语言模型能够访问并利用其训练数据之外的特定信息。借助 LangChain 1.0，构建 RAG 流水线变得更加简洁高效、功能强大。
本指南将带你从零开始搭建一套 RAG 系统——它能从你的文档中检索相关信息，并以此为基础生成准确且上下文相关的回答。LangChain 1.0 引入了更清晰的 API 设计和更优的性能表现，使得 RAG 的实现速度更快、稳定性更高。

什么是 RAG？为什么它如此重要？

RAG 将搜索能力与生成式 AI 相结合，整个流程包含三个关键步骤：

1. 索引（Indexing）：将文档转换为向量嵌入（vector embeddings），并存入向量数据库
2. 检索（Retrieval）：根据用户查询，找出最相关的信息片段
3. 生成（Generation）：结合检索到的上下文与模型自身能力，生成最终回答
   RAG 的价值体现在以下几点：

• 为语言模型提供最新信息
• 通过基于事实数据生成答案，显著减少“幻觉”（hallucinations）
• 无需对模型进行完整微调，即可集成私有数据
• 生成更准确、更贴合场景的 AI 回答

环境配置

首先安装 LangChain 1.0 及其必要依赖：

# 安装核心包pip install langchain==1.0.0langchain-community chromadb openai tiktoken# 安装文档加载所需工具pip install unstructured pdf2image pillow pytesseract

基础项目结构

建议按如下结构组织你的项目：

rag-project/ ├── data/ # 存放你的文档文件 ├── index.py # 向量数据库创建逻辑 ├── retrieve.py # 检索功能实现 ├── generate.py # 带上下文的生成逻辑 └── main.py # 主应用入口

构建你的第一个 RAG 流水线

1. 加载并预处理文档

首先，我们将文档加载进来，并切分为便于处理的文本块（chunks）：

# document_loader.pyfromlangchain_community.document_loadersimportDirectoryLoaderfromlangchain_text_splittersimportRecursiveCharacterTextSplitterdefload_documents(directory_path):# 从指定目录加载文档loader=DirectoryLoader(directory_path)documents=loader.load()# 将文档切分为文本块text_splitter=RecursiveCharacterTextSplitter(chunk_size=1000,# 每块最多 1000 个字符chunk_overlap=200# 相邻块重叠 200 个字符，保持上下文连贯)chunks=text_splitter.split_documents(documents)print(f"成功加载{len(documents)}份文档，共切分为{len(chunks)}个文本块")returnchunks

2. 创建向量嵌入

接下来，将这些文本块转换为向量嵌入，并存入向量数据库：

# index.pyimportosfromlangchain_community.embeddingsimportOpenAIEmbeddingsfromlangchain_community.vectorstoresimportChromadefcreate_vector_db(chunks,persist_directory="./chroma_db"):# 初始化嵌入模型（使用 OpenAI）embeddings=OpenAIEmbeddings()# 创建并向磁盘持久化向量数据库vectordb=Chroma.from_documents(documents=chunks,embedding=embeddings,persist_directory=persist_directory)# 显式持久化（确保数据写入磁盘）vectordb.persist()print(f"向量数据库已创建，共包含{len(chunks)}个文本块")returnvectordb

3. 配置检索器（Retriever）

现在，我们从已保存的向量数据库中加载数据，并创建一个检索器：

# retrieve.pyfromlangchain_community.vectorstoresimportChromafromlangchain_community.embeddingsimportOpenAIEmbeddingsdefsetup_retriever(persist_directory="./chroma_db"):# 加载已持久化的向量数据库embeddings=OpenAIEmbeddings()vectordb=Chroma(persist_directory=persist_directory,embedding_function=embeddings)# 创建检索器，使用相似度搜索，返回最相关的 4 个结果retriever=vectordb.as_retriever(search_type="similarity",search_kwargs={"k":4})returnretriever

4. 构建 RAG 链（RAG Chain）

最后，我们将检索器、提示模板和大语言模型串联成一条完整的 RAG 链：

# generate.pyfromlangchain_community.chat_modelsimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.runnableimportRunnablePassthroughfromlangchain_core.output_parserimportStrOutputParserdefcreate_rag_chain(retriever):# 初始化大语言模型（LLM）llm=ChatOpenAI(model_name="gpt-3.5-turbo")# 定义提示模板：仅基于提供的上下文回答问题template="""请仅根据以下上下文回答问题： {context} 问题：{question} """prompt=ChatPromptTemplate.from_template(template)# 构建 RAG 链：输入 → 检索上下文 + 原始问题 → 提示 → LLM → 字符串输出rag_chain=({"context":retriever,"question":RunnablePassthrough()}|prompt|llm|StrOutputParser())returnrag_chain

5. 整合所有组件

将上述模块组合起来，形成完整的应用：

# main.pyimportosfromdocument_loaderimportload_documentsfromindeximportcreate_vector_dbfromretrieveimportsetup_retrieverfromgenerateimportcreate_rag_chain# 设置 OpenAI API 密钥os.environ["OPENAI_API_KEY"]="your-api-key"defmain():# 1. 加载并处理文档chunks=load_documents("./data")# 2. 创建向量数据库create_vector_db(chunks)# 3. 初始化检索器retriever=setup_retriever()# 4. 构建 RAG 链rag_chain=create_rag_chain(retriever)# 5. 测试提问question="RAG 系统的主要优势有哪些？"answer=rag_chain.invoke(question)print(f"问题：{question}")print(f"回答：{answer}")if__name__=="__main__":main()

高级 RAG 技巧

提升检索质量

通过更精细的参数调整，获得更好的检索结果：

# 使用 MMR（最大边际相关性）提升多样性与相关性的平衡retriever=vectordb.as_retriever(search_type="mmr",# Maximum Marginal Relevancesearch_kwargs={"k":6,# 最终返回 6 个结果"fetch_k":10,# 先从数据库取 10 个候选"lambda_mult":0.5,# 控制多样性 vs 相关性（0=完全多样，1=完全相关）"filter":{"source":"trusted_document.pdf"}# 仅检索来自特定文件的片段})

上下文预处理

在送入 LLM 之前，对检索到的文档进行格式化：

defformat_docs(docs):# 将多个文档内容用双换行符拼接return"\n\n".join(doc.page_contentfordocindocs)# 在 RAG 链中加入格式化步骤rag_chain=({"context":retriever|format_docs,"question":RunnablePassthrough()}|prompt|llm|StrOutputParser())

评估与反馈

引入评估机制，量化 RAG 系统的表现：

fromlangchain.evaluation.qaimportQAEvalChain# 定义评估用例examples=[{"question":"什么是 RAG？","answer":"RAG 是 Retrieval-Augmented Generation（检索增强生成）的缩写……"},# 可添加更多样例]# 初始化评估器evaluator=QAEvalChain.from_llm(llm=ChatOpenAI())# 获取模型预测结果predictions={ex["question"]:rag_chain.invoke(ex["question"])forexinexamples}# 执行评估eval_result=evaluator.evaluate(examples,predictions)print("评估结果：",eval_result)

RAG 实际运行过程可视化

当系统处理如下查询时，内部发生了什么？

用户提问：“RAG 中的向量嵌入是如何工作的？”

1. 查询被转换为向量：[0.023, -0.129, 0.47, …]
2. 检索到的相关文档片段：

• “向量嵌入将文本转换为数值向量……”
• “语义搜索通过计算嵌入之间的相似度来查找相关内容……”

3. 生成的回答：

“在 RAG 中，向量嵌入将文本转化为能捕捉语义含义的数值表示。系统通过余弦相似度等方法，比较查询向量与文档向量的相似性，从而找到相关上下文，并在此基础上生成回答。”

常见 RAG 挑战与解决方案

性能优化建议

要获得更佳的 RAG 表现，可尝试以下策略：

• 分块策略：实验不同的 chunk_size 和 chunk_overlap 组合
• 嵌入模型：对比 OpenAI、BERT、Sentence Transformers 等不同嵌入模型的效果
• 检索方法：采用混合搜索（Hybrid Search），结合关键词匹配与向量相似度
• 提示工程：在提示词中明确要求模型引用来源或说明依据
• 向量数据库调优：根据实际场景选择合适的索引结构（如 HNSW、IVF）

结语

使用 LangChain 1.0 构建 RAG 流水线，能够让你的 AI 系统基于真实、相关的文档生成更准确、更可信的回答。这种方法有效弥补了传统大语言模型无法获取最新或私有信息的短板。

得益于 LangChain 1.0 的模块化设计，你可以轻松定制 RAG 系统的每一个环节——从文档加载、向量索引、检索策略到最终生成。建议从本文的基础实现起步，再根据具体业务需求逐步扩展和优化。

参考文献

https://langchain-tutorials.com/lessons/rag-applications/lesson-11
https://www.geeksforgeeks.org/artificial-intelligence/rag-with-langchain/
https://markaicode.com/building-rag-pipelines-langchain/
https://docs.langchain.com/oss/python/langchain/rag