4.2 实现检索与生成流水线

📌 本节目标

学完本节，你将能够亲手将之前准备好的检索器（向量数据库）与生成器（大语言模型）连接起来，编写代码实现一个完整的 RAG 问答流程。你将理解“检索-增强-生成”这三个核心步骤是如何在代码层面协同工作的，并能够处理用户提问，输出带有知识依据的答案。

1. 核心概念讲解

在动手编码之前，我们先来拆解一下 RAG 流水线的三个核心环节。你可以把它想象成一个“智能助手”的工作流程：

1.1 检索器

通俗解释：检索器就像是一个知识渊博的图书管理员。当你提问时，他不是直接回答，而是先跑到巨大的图书馆（知识库）里，根据你的问题关键词，快速找出最相关的几本书（文档片段）。
比如：你问“Python 列表和元组有什么区别？”，检索器会从知识库中找出包含“列表”、“元组”、“区别”、“可变”等关键词的文档片段。
代码体现：在我们的系统中，检索器通常是一个向量数据库（如 ChromaDB）加上一个嵌入模型（Embedding Model）。核心操作是 similarity_search。

1.2 生成器

通俗解释：生成器是那位在前台负责回答的专家。他不能凭空想象，必须参考图书管理员找来的资料，并结合自己的语言能力，组织成一段流畅、准确的回答。
比如：专家拿到管理员找来的几段关于“列表可变、元组不可变”的文字，然后他组织语言，告诉你：“列表用 [] 定义，可以增删改；元组用 () 定义，一旦创建就不能修改。”
代码体现：生成器就是大语言模型（LLM），比如 OpenAI 的 GPT 或开源的 Llama 模型。核心操作是调用其 generate 或 chat 接口。

1.3 流水线

通俗解释：流水线就是把“管理员”和“专家”的工作串联起来的一条传送带。你提一个问题，传送带把它送到管理员那里，管理员带回资料，再连问题带资料一起送到专家那里，专家给出最终答案。
比如：整个过程就是：用户提问 → 检索器搜索 → 返回相关文档 → 组装 prompt（提示词）→ 生成器回答 → 输出答案。
代码体现：这就是我们本节要编写的核心逻辑，一个将上述步骤串联起来的函数。

2. 深入理解

为什么 RAG 要采用这种“先检索，后生成”的流水线设计？这背后有几个关键考量：

2.1 原理：知识补给与事实约束

大语言模型（LLM）本身的知识是“静态”的，截止于其训练数据的时间点。RAG 通过检索外部知识库，相当于给模型注入了一剂“知识强心针”。更重要的是，检索到的文档为模型提供了“事实锚点”，能极大减少模型“胡编乱造”（即“幻觉”）的概率。模型不再是凭记忆作答，而是基于给定的材料做摘要、推理和总结。

2.2 为什么这样做？—— 对比“裸”用 LLM

2.3 初学者常见误区

误区一：检索到的文档越多越好。 实际上，送入生成器的上下文过长，会稀释关键信息，增加模型理解和生成的成本，甚至导致模型“迷失”在海量信息中。通常检索 3-5 个最相关的文档片段效果较好。
误区二：Prompt 不重要。 很多人以为 RAG 就是把文档和问题拼在一起。实际上，一个精心设计的 Prompt 至关重要。你需要明确告诉模型：“请根据以下上下文回答问题。如果上下文中没有相关信息，请说‘我不知道’。” 这会极大提升回答质量。
误区三：生成器可以处理任何格式的文档。 检索器返回的文档片段是纯文本。如果你的知识库包含 PDF、表格、图片，你需要先做好文本提取和清洗工作。

3. 实战演示

现在，让我们开始编码！我们将使用之前配置好的环境（假设你已经安装了 chromadb, openai, langchain 等库）。我们将构建一个完整的 RAG 函数。

准备工作：加载知识库和 LLM

首先，我们假设你已经有一个 ChromaDB 的向量数据库实例 vector_store，和一个 LLM 客户端 llm。如果你是从头开始，可以参考下面的简化代码：

# 假设你已经有了一个向量存储对象 vector_store
# 假设你已经有了一个大语言模型客户端 llm
# 例如：
# from langchain.embeddings import OpenAIEmbeddings
# from langchain.vectorstores import Chroma
# from langchain.llms import OpenAI

# embeddings = OpenAIEmbeddings()
# vector_store = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
# llm = OpenAI(temperature=0, model_name="gpt-3.5-turbo-instruct")

实现 RAG 流水线函数

我们将分三步实现：retrieve、generate_prompt、generate_answer，最后组合成 rag_pipeline。

# 1. 检索函数：从向量库中检索最相关的文档片段
def retrieve(query, k=4):
    """
    根据用户查询，从向量数据库中检索最相关的 k 个文档片段。
    
    Args:
        query (str): 用户的查询字符串
        k (int): 返回的文档片段数量，默认为 4
    
    Returns:
        list: 包含文档内容的字符串列表
    """
    # 使用向量库的 similarity_search 方法进行检索
    # 它内部会计算查询和文档的向量相似度，并返回最相似的文档
    results = vector_store.similarity_search(query, k=k)
    # 从结果对象中提取出文档的文本内容
    documents = [doc.page_content for doc in results]
    return documents

# 2. 组装 Prompt 函数：将问题和检索到的文档拼接成 Prompt
def generate_prompt(query, documents):
    """
    根据用户查询和检索到的文档，生成一个结构化的 Prompt。
    
    Args:
        query (str): 用户的查询字符串
        documents (list): 检索到的文档内容列表
    
    Returns:
        str: 组装好的 Prompt 字符串
    """
    # 将多个文档片段合并成一个上下文文本，用分隔符隔开
    context = "\n\n---\n\n".join(documents)
    # 构建一个清晰的 Prompt，指导模型如何基于上下文回答
    prompt = f"""你是一个专业的问答助手。请根据以下提供的上下文信息，回答用户的问题。
如果上下文中没有足够的信息来回答问题，请直接说“我无法从提供的资料中找到答案”，不要编造答案。

上下文信息：
{context}

用户问题：
{query}

请给出一个准确、简洁的回答："""
    return prompt

# 3. 生成答案函数：调用大语言模型生成最终答案
def generate_answer(prompt):
    """
    将组装好的 Prompt 发送给大语言模型，并获取生成的回答。
    
    Args:
        prompt (str): 完整的 Prompt 字符串
    
    Returns:
        str: 模型生成的回答
    """
    # 调用 LLM 的 generate 方法，传入 Prompt
    # 这里我们只取第一个生成结果
    response = llm.generate([prompt])
    # 从响应中提取生成的文本
    answer = response.generations[0][0].text.strip()
    return answer

# 4. 将上述步骤组合成完整的 RAG 流水线
def rag_pipeline(query):
    """
    完整的 RAG 问答流水线：检索 -> 组装 Prompt -> 生成答案。
    
    Args:
        query (str): 用户的查询字符串
    
    Returns:
        tuple: (最终答案, 检索到的文档列表)
    """
    print(f"🔍 正在检索与问题相关的资料: {query}")
    # 第一步：检索
    retrieved_docs = retrieve(query)
    print(f"📄 已检索到 {len(retrieved_docs)} 个相关文档片段")
    
    # 第二步：组装 Prompt
    prompt = generate_prompt(query, retrieved_docs)
    print("📝 已组装好 Prompt，准备发送给生成器...")
    
    # 第三步：生成答案
    answer = generate_answer(prompt)
    print("✨ 答案生成完成！")
    
    # 返回答案和相关的文档，方便后续检查
    return answer, retrieved_docs

# --- 测试我们的 RAG 系统 ---
if __name__ == "__main__":
    # 模拟一个用户提问
    user_question = "什么是RAG技术？它有什么优点？"
    
    # 调用我们的流水线
    final_answer, source_docs = rag_pipeline(user_question)
    
    # 打印最终答案
    print("\n" + "="*50)
    print("🤖 最终答案：")
    print(final_answer)
    print("="*50)
    
    # 打印检索到的文档来源（可选）
    # print("\n📚 参考文档片段：")
    # for i, doc in enumerate(source_docs):
    #     print(f"\n--- 文档 {i+1} ---")
    #     print(doc[:200] + "...")  # 只打印前200个字符

4. 关键要点

模块化设计：将检索、Prompt 组装、生成拆分成独立的函数，代码更清晰、易于测试和修改。
Prompt 工程：RAG 的核心在于 Prompt 的设计。明确告诉模型“基于上下文回答”和“不知道就说不知道”，是控制质量和减少幻觉的关键。
检索数量控制：k 值（检索文档数量）是重要的超参数，通常 3-5 个效果最好，需要根据实际情况调整。
错误处理意识：在实际应用中，要考虑到检索结果为空、LLM 调用失败等情况，并添加相应的异常处理逻辑。
可解释性：RAG 的一大优势是可解释性。在返回答案的同时返回检索到的文档来源，有助于用户验证答案的准确性。

✏️ 练习任务

练习 1：基础 RAG 流程实现（必做）

目标：基于上述代码，实现一个能回答“LangChain 是什么？”这个问题的 RAG 流水线。
前提：你需要一个包含 LangChain 相关文档片段的 ChromaDB 知识库。
输入：query = "LangChain 是什么？"
期望输出：一个基于知识库内容生成的、关于 LangChain 定义的准确回答。例如：“LangChain 是一个用于开发由语言模型驱动的应用程序的框架...”。同时，你可以在控制台看到检索到文档的提示信息。

练习 2：优化与调试（选做）

挑战：修改 generate_prompt 函数，增加一个指令，要求模型在回答时用序号列出要点。例如，修改 Prompt 为“请用 1. 2. 3. 的格式，分点列出答案。”。
进阶：观察当 k 值设置为 1 和设置为 10 时，回答质量（准确性、完整性）和模型响应速度的变化。尝试解释为什么会产生这种差异。

💡 下一节我们将学习如何评估我们搭建的 RAG 系统的性能，包括检索准确率和生成答案的相关性，让你的系统从“能用”走向“好用”。

本节由 StudyAI8 AI 课程团队生成并审核。