本文介绍了如何使用Python和LangChain搭建一个基础的RAG系统。文章详细讲解了环境准备、技术选型、数据预处理、文档向量化与索引构建以及检索模块的实现。通过本文，读者可以了解RAG系统的核心流程，并掌握使用LangChain进行快速开发的技巧。教程示例采用ChromaDB作为向量数据库，BAAI/bge-small-zh-v1.5作为Embedding模型，并提供OpenAI API和本地部署ollama+Qwen1.5-1.8B-Chat模型的备选方案。

🛠️**环境准备与技术选型**：搭建RAG系统需要准备Python环境，并选择合适的依赖库，如langchain、sentence-transformers、faiss或chromadb等。同时，还需要考虑向量数据库、Embedding模型和大型语言模型的选型，例如OpenAI API或本地开源模型。

📚**数据准备与预处理**：原始知识文档需要经过加载和分割处理，以便RAG系统能够高效处理。LangChain提供了多种DocumentLoader来处理不同类型的文件，并使用RecursiveCharacterTextSplitter将长文档分割成更小的语义单元。

🗂️**文档向量化与索引构建**：将预处理后的文本块转换为数值向量，并存储到向量数据库中，构建可供高效检索的索引。文章示例使用HuggingFace Hub上的 BAAI/bge-small-zh-v1.5 模型进行向量化，并使用 ChromaDB 作为向量数据库。

🔍**实现检索模块**：根据用户的查询，从向量数据库中检索出top_k个最相关的文本块，作为后续LLM生成答案的上下文。LangChain的Retriever对象封装了用户查询向量化的过程，并执行相似度搜索。

本章节将带领读者一步步动手实践，使用Python和流行的开源框架Lang Chain， 搭建一个基础的RAG系统。我们将重点关注核心流程的实现，并提供相应的代码示例。请确保你已具备基本的Python编程能力， 并对LLM和向量检索有初步了解。

环境准备与技术选型

开发环境：

编程语言：Python(推荐版本3.8或更高)

核心依赖库及其用途：

langchain：一个强大的开源框架，用于快速开发基于语言模型的应用程序，它极大地简化了RAG流程中各个组件(如文档加载、文本分割、向量存储、LLM调用、链构建等)的集成和编排。(CSDN博客：基于LangChain搭建RAG系统)

sentence-transformers ： 一个流行的Python库，提供了大量预训练的SOTA(State-of-the-Art)文本嵌入模型，方便我们将文本转换为高质量的向量表示。

faiss-cpu/faiss-gpu：Facebook AI开发的非常高效的相似性搜索库，适合本地构建和查询向量索引。-cpu版本适用于无GPU环境。

chromadb：一个开源的、专为AI应用设计的向量数据库，易于在本地启动和使用，并能与LangChain很好地集成。

openai：如果选择使用OpenAI的LLM(如GPT-3.5-turbo，GPT-4) 和Embedding模型(如text-embedding-ada-002)，需要此库进行API调用。

transformers：Hugging Face提供的库，可以方便地加载和使用本地部署的开源LLM和Embedding模型。

pypdf ， unstructured， python-docx， beautifulsoup4：用于加载和解析不同格式的文档(PDF， TXT， DOCX， HTML等)。 unstructured尤其擅长处理复杂格式文档。

(可选)fastapi，uvicorn：如果需要将RAG系统部署为API服务，这两个库是构建高性能Web API的常用选择。

安装命令示例：

首先，建议创建一个独立的Python虚拟环境(例如使用venv 或 conda )。激活虚拟环境后，执行以下命令安装核心依赖：

pip install langchain langchain-community langchain-core sentence-trans# 如果使⽤OpenAI的LLM，可能还需要 specific langchain-openaipip install langchain-openai# 如果使⽤Ollama运⾏本地模型pip install langchain_community ollama# 注意：unstructured[local-inference] 会安装⼀些本地解析所需的依赖，如tesserac# 根据你实际选择的组件，可能还需要安装其他特定库。

技术选型考量(工具/模型)：

在搭建RAG系统时，我们需要对关键组件进行选型。以下是一些常见的考量维度和选项：

向量数据库(Vector Database)：

对比维度：部署复杂度(本地vs.云托管)、性能(查询延迟、吞吐量)、可扩展性、功能特性(如元数据过滤、混合检索支持、备份恢复)、社区支持、成本。

选项示例：

本地/轻量级： FAISS S(库，极快，适合原型)， ChromaDB(易用， 适合入门和中小型项目)， SQLite+ sqlite-vss(基于SQLite的向量搜索扩展)。

可本地可云/生产级：Milvus(功能强大， 可扩展性好)， Qdrant (Rust编写，性能优秀)，weaviate (支持GraphQL，混合检索)。

云原生/托管服务：Pinecone ， Amazon OpenSearch Service (with k-NN)， Google Vertex AI Vector Search。

Embedding模型(Text Embedding Model)：

对比维度：模型效果(尤其是在目标语言和任务上的表现，可参考MTEB、C-MTEB等榜单) 、向量维度(影响存储和计算)、计算资源需求、推理速度、是否支持中文及多语言、许可协议。

选项示例：

中文优先：BAAI/bge-large-zh-v1.5，BAAI/bge-m3(支持中英等多语言)， shibing624/text2vec-base-chinese，阿里云DashScope的文本嵌入服务。

英文/通用：OpenAI的 text-embedding-ada-002， text-embedding-3-small/large， Sentence Transformers库中的all-MiniLM-L6-v2(轻量高效)， all-mpnet-base-v2(效果较好)。

多语言：sentence-transformers/paraphrase-multilingual-mpnet-base-v2，intfloat/multilingual-e5-large.

**大型语言模型(LLM-Generator) **：

对比维度：生成质量(遵循指令能力、连贯性、创造性)、推理成本(API费用或本地部署资源消耗)、上下文窗口大小、API易用性或本地部署可行性、对中文等特定语言的支持程度、响应速度。

选项示例：

OpenAIAPI：gpt-3.5-turbo bo(性价比高)， gpt-4，(能力更强，上下文更长，成本更高)。

国内模型API：阿里云通义千问(qwen-turbo，qwen-plus )，智谱AI ChatGLM系列(glm-4)，百度文心系列，腾讯混元等。

本地开源模型：Llama-2/3系列， Mistral AI的Mistral-7B， Mixtral-8x7B，清华大学的ChatGLM3-6B， 阿里巴巴的Qwen1.5系列(如Qwen1.5-1.8B-Chat，Qwen1.5-7B-Chat)。本地部署需要相应的硬件资源 (GPU) 和配置工作(如使用 transformers ， Ollama ， vLLM等工具)。

本教程选型(示例)：

为了方便入门和本地快速实践，本教程将采用以下技术栈作为示例。读者可以根据自己的实际需求和环境进行替换。

RAG框架：LangChain(版本>0.1.0)

向量数据库：ChromaDB(本地运行，无需复杂配置)

Embedding模型：BAAI/bge-small-zh-v1.5 (HuggingFace模型，兼顾中文效果与本地资源消耗，使用：sentence-transformers加载)

LLM模型：

优先展示使用 OpenAI API (gpt-3.5-turbo)，因其易用性和广泛性。(需要自行准备OpenAI API Key)

同时，会提供使用本地部署的ollama+Qwen1.5-1.8B-Chat模型的备选方案，方便没有APIKey或希望完全本地化的读者。(需要自行安装Ollama并拉取模型)

请确保在开始后续步骤前，已按照上述安装命令安装了必要的库，并准备好了所选LLM的运行环境(APIKey或本地模型服务)。

步骤一：数据准备与预处理

目标：将我们的原始知识文档(例如，一组关于某个开源项目的介绍文档，可能是TXT或PDF格式)加载进来，并将其分割成适合RAG系统处理的、带有元数据的文本块(chunks)。

具体操作：

1. 数据加载(Data Loading)

我们需要从文件系统中加载文档。Lang Chain提供了多种DocumentLoader`来处理不同类型的文件。

首先，假设我们有一个名为knowledge_base`的文件夹，里面存放了我们的知识文档。例如，可以创建两个简单的.txt文件：

knowledge_base/project_intro.txt：

RAGFlow是一个基于Python的开源检索增强生成框架。

它致力于简化RAG应用的开发、评估和部署流程。

RAGFlow的核心特性包括模块化设计、多种检索策略支持以及易于扩展的接口。

该项目由ABCLab于2024年发起。

knowledge_base/features.txt：

RAGFlow的主要功能包括：

数据接⼊：⽀持PDF, TXT, Markdown等多种⽂档格式。智能分块：提供多种⽂本分割策略，如递归分割、语义分割。向量化与索引：集成主流Embedding模型和向量数据库。灵活检索：⽀持向量检索、关键词检索及混合检索。LLM集成：⽅便对接OpenAI、HuggingFace等多种⼤语⾔模型。评估套件：内置常⽤RAG评估指标和⼯具。

现在，我们使⽤LangChain的 DirectoryLoader 和 TextLoader 来加载这些⽂档

from langchain_community.document_loaders import DirectoryLoader, TextLoaderfrom langchain_core.documents import Document # Document typefrom typing import List# 定义知识库⽬录knowledge_base_path = "./knowledge_base"# 注意：请确保在Python脚本的同级⽬录下创建 knowledge_base ⽂件夹，并将上述txt⽂件放⼊其# 使⽤DirectoryLoader加载⽬录下所有.txt⽂件，指定使⽤TextLoaderloader = DirectoryLoader(knowledge_base_path,glob="**/*.txt", # 匹配所有txt⽂件loader_cls=TextLoader, # 使⽤TextLoader加载loader_kwargs={'encoding': 'utf-8'}, # TextLoader的参数，确保UTF-8编码use_multithreading=True, # 可以加速加载多个⽂件show_progress=True # 显⽰加载进度)documents: List[Document] = loader.load()if documents:print(f"成功加载 {len(documents)} 个⽂档.")for i, doc in enumerate(documents):print(f"\n--- ⽂档 {i+1}: {doc.metadata.get('source', '未知来源')} ---print(f"内容预览 (前100字符): {doc.page_content[:100]}")else:print(f"未能从 '{knowledge_base_path}' 加载到任何⽂档。请检查路径和⽂件。")# 对于PDF⽂件加载，可以使⽤PyPDFLoader (需要 pip install pypdf)# from langchain_community.document_loaders import PyPDFLoader# pdf_loader = PyPDFLoader("path/to/your/document.pdf")# pdf_documents = pdf_loader.load()

说明：DirectoryLoader可以方便地加载整个目录的文件。每个加载的Document对象通常包含两部分：page_content(文档的文本内容)和metadata(一个字典，通常包含如 source等元信息，即文件名)。

2.文本分割(Text Splitting)

长文档需要被切分成更小的语义单元(chunks)， 以便Embedding模型处理并提高检索的精确性。LangChain的 RecursiveCharacterTextSplitter是一个常用的选择，它会尝试按预设的一系列分隔符(如换行符、句号等)递归地分割文本，以求在满足大小限制的同时尽量保持语义的完整性。

from langchain.text_splitter import RecursiveCharacterTextSplitter# 初始化⽂本分割器text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, # 每个块的最⼤字符数。应根据Embedding模型和LLM的上下⽂窗⼝进⾏调# 对于中⽂，⼀个汉字通常算1个token，但具体取决于模型的分词器。# BGE等模型通常有512 tokens的输⼊限制。chunk_overlap=20, # 相邻块之间的重叠字符数，帮助保留上下⽂连贯性。separators=["\n\n", "\n", "。", "！", "？", "，", "、", " ", ""], # 中⽂场景length_function=len, # 使⽤字符⻓度计算chunk_sizeadd_start_index=True # 在metadata中添加块在原⽂中的起始位置，可选)if documents: # 确保前⾯加载成功document_chunks: List[Document] = text_splitter.split_documents(documentprint(f"\n⽂档被分割成 {len(document_chunks)} 个⽂本块.")if document_chunks:print(f"第⼀个⽂本块预览: {document_chunks[0].page_content}")print(f"第⼀个⽂本块元数据: {document_chunks[0].metadata}")else:print("没有⽂档可供分割。")document_chunks = [] # 初始化为空列表以防后续代码出错

说明：：chunk_size 和 chunk_overlap 的选择对RAG性能有显著影响，需要根据具体数据、Embedding模型的能力以及LLM的上下文窗口大小进行实验调整。对于中文文本，合理设置separators尤为重要，以尽可能在自然的语义边界(如段落、句子)进行切分。

步骤⼆：⽂档向量化与索引构建

⽬标： 将预处理后的⽂本块（chunks）转换为数值向量（embeddings），并将这些向量及

其对应的⽂本内容存储到向量数据库中，构建⼀个可供⾼效检索的索引。

具体操作:

1. 初始化Embedding模型

我们将使⽤HuggingFace Hub上的 BAAI/bge-small-zh-v1.5 模型作为⽰例，它是⼀个对中⽂⽀持较好且相对轻量级的模型。LangChain通过 HuggingFaceBgeEmbeddings 类可以方便地加载此类模型。

from langchain_community.embeddings import HuggingFaceBgeEmbeddingsembedding_model_name = "BAAI/bge-small-zh-v1.5"# 如果你的机器有GPU并且安装了CUDA版本的PyTorch，可以设置为'cuda'# 否则， sentence-transformers 会⾃动检测或使⽤ 'cpu'model_kwargs = {'device': 'cpu'}encode_kwargs = {'normalize_embeddings': True} # BGE模型通常推荐对输出向量进⾏归⼀try:embedding_model = HuggingFaceBgeEmbeddings(model_name=embedding_model_name,model_kwargs=model_kwargs,encode_kwargs=encode_kwargs)print(f"\n成功加载Embedding模型: {embedding_model_name}")# 测试⼀下Embedding模型example_text = "这是⼀个⽰例⽂本，⽤于测试向量化功能。"query_vector = embedding_model.embed_query(example_text)print(f"⽰例⽂本的向量维度: {len(query_vector)}")# print(f"向量预览 (前5维): {query_vector[:5]}")except Exception as e:print(f"加载Embedding模型失败: {e}")print("请确保已安装sentence-transformers且模型名称正确，或⽹络连接正常可以下载模型embedding_model = None # 标记模型加载失败

说明： ⾸次运⾏此代码时， sentence-transformers 库会⾃动从HuggingFace Hub下载并缓存BAAI/bge-small-zh-v1.5 模型⽂件，这可能需要⼀些时间，并取决于⽹络连接。如果遇到问题，请检查⽹络或尝试⼿动下载模型到本地指定路径。

2. 构建并填充向量索引

我们将使⽤ ChromaDB 作为向量数据库。ChromaDB易于在本地运⾏和集成。LangChain的Chroma 类提供了便捷的接⼝。

from langchain_community.vectorstores import Chromaimport shutil # ⽤于清理旧的数据库⽬录# 定义ChromaDB的持久化存储路径和集合名称persist_directory = './chroma_db_store'collection_name = 'my_rag_collection_v1'# (可选) 清理旧的数据库⽬录，以便每次运⾏时都创建⼀个新的# if True: # 设置为False则不清空，会尝试加载现有数据# try:# shutil.rmtree(persist_directory)# print(f"已清理旧的数据库⽬录: {persist_directory}")# except FileNotFoundError:# print(f"数据库⽬录 {persist_directory} 不存在，⽆需清理。")# except Exception as e:# print(f"清理⽬录 {persist_directory} 失败: {e}")vector_db = None # 初始化if document_chunks and embedding_model: # 确保前⾯的步骤成功print(f"\n开始构建向量数据库和索引，使⽤集合名: {collection_name}...")try:vector_db = Chroma.from_documents(documents=document_chunks, # 前⾯分割好的⽂本块列表embedding=embedding_model, # 初始化好的Embedding模型实例collection_name=collection_name,persist_directory=persist_directory # 指定持久化路径)# Chroma.from_documents 会⾃动处理持久化，但显式调⽤ persist() 确保写⼊磁盘# vector_db.persist() # 对于某些版本的ChromaDB或特定⽤法可能需要print(f"向量数据库 '{collection_name}' 构建完成并已持久化到 '{persist_dirprint(f"数据库中包含 {vector_db._collection.count()} 个向量条⽬.")except Exception as e:print(f"构建ChromaDB失败: {e}")向量数据库(e.g.,ChromaDB)构建⾼效检索索引else:print("\n由于⽂档块列表为空或Embedding模型未加载，跳过向量数据库构建。")

说明： Chroma.from_documents ⽅法会⾃动对传⼊的 document_chunks 进⾏向量化（使⽤我们提供的 embedding_model ），并将⽂本内容、向量以及元数据⼀同存储到ChromaDB中。通过指定 persist_directory ，数据库的内容会被保存在本地磁盘，方便后续加载使用，避免每次重新计算。如果持久化⽬录已存在且包含同名集合，Chroma通常会尝试加载现有数据，除⾮我们⼿动清空⽬录。

步骤三：实现检索模块 (Retriever Implementation)

⽬标： 根据⽤⼾的查询 (query)，从已经构建好的向量数据库中检索出 top_k 个最相关的⽂本块，这些⽂本块将作为后续LLM⽣成答案的上下⽂。

具体操作：

1. ⽤⼾查询向量化： 这⼀步通常由LangChain的Retriever对象在内部封装。当我们调⽤检索器时，它会⾃动使⽤与索引构建时相同的Embedding模型来向量化⽤⼾的查询。

2. 执⾏相似度搜索： LangChain中的VectorStore对象本⾝就可以很⽅便地转换成⼀个Retriever。

# 假设前⾯的 vector_db (Chroma实例) 已经成功创建或加载if vector_db:user_query = "RAGFlow的核⼼特性有哪些？"k_results = 3 # 我们希望检索最相关的3个⽂档块# 将ChromaDB实例包装成⼀个Retriever对象# search_type默认为"similarity"，也可以指定为"mmr" (Maximal Marginal Releva# search_kwargs⽤于传递给底层向量存储的搜索参数，如kretriever = vector_db.as_retriever(search_kwargs={"k": k_results})print(f"\n为查询 '{user_query}' 配置的检索器将返回 {k_results} 个结果。")try:# 执⾏检索# invoke是LCEL推荐的⽅法, get_relevant_documents是旧版但仍可⽤relevant_docs: List[Document] = retriever.invoke(user_query)print(f"\n成功检索到 {len(relevant_docs)} 个相关⽂档块：")for i, doc in enumerate(relevant_docs):print(f"\n--- 相关⽂档 {i+1} ---")# ChromaDB的as_retriever可能不直接在metadata中返回score，# 如果需要分数，可能要⽤ vector_db.similarity_search_with_score()# print(f"Score: {doc.metadata.get('score', 'N/A')}") # ⽰例，实际print(f"来源: {doc.metadata.get('source', '未知')}, 块起始位置: {doprint(f"内容: {doc.page_content}")except Exception as e:print(f"检索失败: {e}")relevant_docs = [] # 初始化为空列表else:print("\n向量数据库未初始化，⽆法执⾏检索。")relevant_docs = [] # 初始化为空列表

说明： vector_db.as_retriever() ⽅法⾮常便捷，它将我们的ChromaDB实例转换成了⼀个标准的LangChain Retriever对象。通过 search_kwargs={"k": k_results} ，我们指定了希望检索的⽂档块数量。检索操作返回的是⼀个 Document 对象列表，每个对象包含了其原始内容 page_content 和元数据 metadata （例如来源⽂件名）。这些⽂档块将作为LLM⽣成答案的关键上下⽂。

步骤四：集成LLM⽣成答案 (Generator Integration)

⽬标： 将检索模块获取到的相关上下⽂信息 (retrieved contexts) 与⽤⼾的原始查询整合起来，通过⼀个⼤型语⾔模型（LLM）来⽣成⼀个连贯且准确的答案。

具体操作：

1. 构建Prompt (Prompt Engineering)

Prompt是指导LLM如何⾏动的关键。⼀个好的Prompt模板应该清晰地告诉LLM它的⻆⾊、任务、可⽤的上下⽂信息，以及在特定情况（如信息不⾜）下应该如何回应。下⾯是⼀个针对RAG场景的Prompt模板⽰例：

你是⼀个智能问答助⼿。你的任务是根据下⾯提供的【上下⽂信息】来回答【⽤⼾问题】。请严格依据【上下⽂信息】进⾏回答，不要依赖任何外部知识或进⾏猜测。如果【上下⽂信息】中没有⾜够的内容来回答【⽤⼾问题】，请直接回复：“抱歉，根据我⽬前掌握的信息，确保你的回答简洁、相关，并且直接针对【⽤⼾问题】。【上下⽂信息】:---{context_str}---【⽤⼾问题】: {user_query}【你的回答】:

我们可以使⽤LangChain的 ChatPromptTemplate 来创建这个模板：

from langchain_core.prompts import ChatPromptTemplateprompt_template_str = """你是⼀个智能问答助⼿。你的任务是根据下⾯提供的【上下⽂信息】来请严格依据【上下⽂信息】进⾏回答，不要依赖任何外部知识或进⾏猜测。如果【上下⽂信息】中没有⾜够的内容来回答【⽤⼾问题】，请直接回复：“抱歉，根据我⽬前掌握的信息，确保你的回答简洁、相关，并且直接针对【⽤⼾问题】。【上下⽂信息】:---{context_str}---【⽤⼾问题】: {user_query}【你的回答】:"""prompt_template = ChatPromptTemplate.from_template(prompt_template_str)# 模拟：假设已经有了检索到的⽂档列表 relevant_docs 和⽤⼾查询 user_query# (确保 user_query 和 relevant_docs 在此作⽤域内有效，通常它们来⾃上⼀步)if relevant_docs and 'user_query' in locals() and user_query:# 将检索到的多个⽂档块内容合并为⼀个字符串context_str = "\n\n---\n\n".join([doc.page_content for doc in relevant_d# 使⽤模板格式化Prompt，准备输⼊给LLMformatted_prompt_messages = prompt_template.format_messages(context_str=context_str,user_query=user_query)print("\n--- 为LLM准备的格式化Prompt ---")for msg in formatted_prompt_messages:print(f"类型: {msg.type}, 内容:\n{msg.content}")print("------------------------------")else:print("\n相关⽂档列表为空或⽤⼾查询未定义，⽆法格式化Prompt。")formatted_prompt_messages = None

2. 调⽤LLM模型获取答案

接下来，我们初始化选定的LLM模型，并将格式化后的Prompt传递给它以⽣成答案。我们将展⽰如何使⽤OpenAI API和本地Ollama服务两种⽅式。

from langchain_core.output_parsers import StrOutputParserimport os # ⽤于读取环境变量# 初始化LLM和输出解析器llm = Noneoutput_parser = StrOutputParser()# --- ⽅案1: 使⽤OpenAI API (gpt-3.5-turbo) ---# 需要设置环境变量 OPENAI_API_KEY# 同时可能需要设置 OPENAI_API_BASE 如果使⽤代理use_openai = True # 改为False以使⽤Ollama，或根据条件判断if use_openai:from langchain_openai import ChatOpenAItry:# openai_api_key = os.getenv("OPENAI_API_KEY")# openai_api_base = os.getenv("OPENAI_API_BASE") # 例如 "https://api# if not openai_api_key:# raise ValueError("请设置 OPENAI_API_KEY 环境变量。")llm = ChatOpenAI(# openai_api_key=openai_api_key,# openai_api_base=openai_api_base,model_name="gpt-3.5-turbo",temperature=0.1, # 低temperature使回答更确定、更基于事实max_tokens=500)print("\n已配置使⽤ OpenAI GPT-3.5-Turbo。")except Exception as e:print(f"配置OpenAI失败: {e}. 将尝试Ollama。")use_openai = False # 配置失败则尝试Ollama# --- ⽅案2: 使⽤本地Ollama服务 (假设已运⾏并拉取了qwen:1.8b模型) ---if not use_openai: # 如果不使⽤OpenAI或OpenAI配置失败from langchain_community.llms import Ollamatry:llm = Ollama(model="qwen:1.8b", # 确保Ollama服务中有名为qwen:1.8b的模型# 可以通过 `ollama list` 查看可⽤模型# 或通过 `ollama pull qwen:1.8b` 拉取temperature=0.1)print("\n已配置使⽤本地Ollama (qwen:1.8b)。请确保Ollama服务正在运⾏。")except Exception as e:print(f"配置Ollama失败: {e}. LLM未能初始化。")llm = None # 标记LLM初始化失败# 调⽤LLM (如果Prompt和LLM都已准备好)if formatted_prompt_messages and llm:# 构建链: prompt -> llm -> output_parser# (这⾥简化，后⾯会⽤LCEL正式构建链)# messages_for_llm = prompt_template.format_prompt(context_str=context_stry:# response = llm.invoke(formatted_prompt_messages) # LCEL⻛格 invoke# final_answer_from_llm = output_parser.invoke(response)# # 为了直接展⽰，这⾥⼿动组合⼀个简单的调⽤，完整链在下⼀步# # 此处仅为演⽰LLM调⽤，⾮最终RAG链temp_chain = prompt_template | llm | output_parserfinal_answer_from_llm = temp_chain.invoke({"context_str": context_stprint("\n--- LLM⽣成的初步回答 ---")print(final_answer_from_llm)print("--------------------------")except Exception as e:print(f"LLM调⽤失败: {e}")if "OPENAI_API_KEY" in str(e).upper():print("提⽰: 如果使⽤OpenAI，请确保正确设置了OPENAI_API_KEY环境变量，并elif "CONNECTION REFUSED" in str(e).upper() and "OLLAMA" in str(llmprint("提⽰: 如果使⽤Ollama，请确保Ollama服务已在本地启动并运⾏。")elif not llm:print("\nLLM模型未能成功初始化，⽆法⽣成答案。")else:print("\n格式化Prompt未准备好，⽆法调⽤LLM。")

说明： 上述代码块提供了连接OpenAI和本地Ollama的选项。 temperature=0.1 设置使得LLM的输出更具确定性，这对于基于事实的问答通常是期望的。 StrOutputParser ⽤于从LLM的复杂输出对象中提取纯⽂本回答。请确保你已正确配置所选LLM的环境（如API密钥或本地服务）

步骤五：组装完整的RAG Pipeline

⽬标： 将前⾯实现的各个步骤（数据加载、分割、向量化、索引、检索、Prompt构建、LLM⽣成）顺畅地串联起来，形成⼀个端到端的、可执⾏的RAG⼯作流。LangChain的表达式语⾔（LangChain Expression Language, LCEL）使得这个过程非常简洁和声明式。

from langchain_core.runnables import RunnablePassthrough, RunnableParallel# 确保 retriever, prompt_template, llm, output_parser 已在前序步骤中定义并初始化成# 定义⼀个辅助函数，将检索到的Document列表格式化为单个字符串def format_docs(docs: List[Document]) -> str:"""将Document列表合并为⼀个⽤分隔符隔开的字符串。"""return "\n\n---\n\n".join([d.page_content for d in docs])if 'retriever' in locals() and retriever and \'prompt_template' in locals() and prompt_template and \'llm' in locals() and llm and \'output_parser' in locals() and output_parser:# 使⽤LCEL构建RAG链# 链的输⼊是⽤⼾的查询字符串 (user_query)# 1. retriever处理user_query得到相关⽂档 (relevant_docs)# 2. format_docs将relevant_docs格式化为context_str# 3. user_query通过RunnablePassthrough()直接传递# 4. context_str和user_query被送⼊prompt_template# 5. 格式化后的prompt送⼊llm# 6. llm的输出送⼊output_parser得到最终字符串答案rag_chain_simple = (# RunnableParallel允许并⾏或按指定结构准备后续组件的输⼊字典{"context_str": retriever | format_docs, "user_query": RunnablePasst| prompt_template| llm| output_parser)print("\n简单的RAG链已构建 (rag_chain_simple)。")# 同时构建⼀个可以返回源⽂档的链，⽅便溯源rag_chain_with_sources = RunnableParallel({"context_docs": retriever, "user_query": RunnablePassthrough()}).assign(answer = RunnableParallel({"context_str": lambda x: format_docs(x["context_docs"]), "user_) | prompt_template | llm | output_parser)print("带溯源信息的RAG链已构建 (rag_chain_with_sources)。")# --- 测试完整的RAG Pipeline ---test_query = "RAGFlow有哪些主要功能？" # 使⽤之前定义好的 user_query，或者新的测print(f"\n--- 开始测试RAG链，查询: '{test_query}' ---")# 测试简单链try:print("\n--- 测试 rag_chain_simple ---")final_answer_simple = rag_chain_simple.invoke(test_query)print("\n【最终RAG回答 (简单链)】:")print(final_answer_simple)except Exception as e:print(f"执⾏ rag_chain_simple 失败: {e}")# 测试带溯源信息的链try:print("\n--- 测试 rag_chain_with_sources ---")result_with_sources = rag_chain_with_sources.invoke(test_query)print("\n【最终RAG回答 (带溯源信息)】:")print(f"回答: {result_with_sources['answer']}")print("\n【参考上下⽂来源】:")if result_with_sources['context_docs']:for i, doc_item in enumerate(result_with_sources['context_docs']print(f" 来源 {i+1}: {doc_item.metadata.get('source', '未知'print(f" 内容⽚段 (前50字符): {doc_item.page_content[:50].selse:print(" 未能检索到相关上下⽂。")except Exception as e:print(f"执⾏ rag_chain_with_sources 失败: {e}")else:print("\n由于⼀个或多个核⼼组件（retriever, prompt, llm, parser）未能初始化，⽆

说明与分析：

LCEL的威⼒： LangChain表达式语⾔（LCEL）通过管道操作符 | 使得链的构建⾮常直观。它允许将不同的“可运⾏单元（” Runnables，如检索器、Prompt模板、LLM模型、输出解析器等）组合起来。

输⼊与输出流：

rag_chain_simple：输⼊是⼀个字符串（⽤⼾查询）。RunnablePassthrough() ⽤于将链的输⼊（即⽤⼾查询）直接作为 user_query字段的值传递给Prompt模板。 retriever | format_docs 这部分则先⽤原始查询通过 retriever 获取⽂档，然后通过 format_docs 函数处理成 context_str 。这两部分构成Prompt模板所需的输⼊字典。

rag_chain_with_sources ：这个链稍微复杂⼀些，因为它旨在同时返回LLM的回答和检索到的源⽂档。它⾸先使⽤ RunnableParallel 来并⾏获取context_docs （通过 retriever ）和 user_query （通过RunnablePassthrough ）。然后，使⽤ .assign() ⽅法在保持原始context_docs 和 user_query 的同时，计算出⼀个新的字段 answer 。计算answer 的过程本⾝⼜是⼀个⼦链，它从⽗链的输出中提取 context_docs 和user_query ，格式化上下⽂，然后通过Prompt模板、LLM和输出解析器得到答案。

可组合性与灵活性： LCEL的强⼤之处在于其⾼度的可组合性。你可以轻松地在链中插⼊或替换组件，例如加⼊查询改写步骤、重排步骤等，⽽⽆需⼤幅修改整体结构。

通过这五个步骤，我们已经成功搭建了⼀个基础但完整的RAG系统。它能够加载本地⽂档，构建向量索引，根据⽤⼾查询检索相关信息，并利⽤LLM⽣成基于这些信息的回答。接下来的章节将探讨如何部署这个系统以及如何进⼀步优化其性能。