今天我们聚焦的是大模型应用开发中最核心的架构模式之一——RAG（Retrieval-Augmented Generation，检索增强生成）。在大模型落地的过程中，几乎所有团队都会遇到同一个问题：模型很聪明，但它的知识是"封闭"的。它不知道你公司的内部文档，不了解最新的行业动态，甚至会在回答中"一本正经地胡说八道"。RAG 正是为了解决这个问题而诞生的——它让大模型从"闭卷考试"变成了"开卷考试"，通过在生成回答之前先检索相关知识，大幅提升回答的准确性和可靠性。

  在整个 LLM 应用技术栈中，RAG 处于"应用架构层"，它不改变模型本身，而是通过外挂知识库的方式增强模型能力。这意味着你不需要花费大量算力去微调模型，也不需要等待模型更新训练数据——只要把最新的文档灌入检索系统，模型就能立刻"学会"这些知识。正因如此，RAG 已经成为企业级 LLM 应用中最广泛采用的知识增强架构之一。

  本课程共四章，沿着"为什么需要 → 是什么 → 怎么工作 → 亲手搭建 → 工程化落地"的主线展开。第一章从大模型的知识困境出发，帮助大家建立对 RAG 价值的直觉认知，并深入了解 RAG 的核心应用场景与产品集成价值；第二章深入 RAG 的核心架构，拆解每一个关键组件和数据流转过程，介绍主流开发框架与热门开源项目；第三章是全课程的动手核心——不借助任何 SDK 框架，全部用 Python + 基础依赖手写实现一个端到端的 RAG 问答系统；第四章则通过一个完整的工程化项目——RAG 代码库智能问答系统，展示手写知识如何迁移到真实产品中。

🎯 学员画像与前置知识 本课程面向具备 Python 基础、了解大模型基本概念（如 Prompt、Token、Chat Completions API）的开发者。学习前，建议先了解 LLM API 的基本调用方式。学完本课后，你将能够清晰说明 RAG 解决什么问题、由哪些组件构成、数据如何流转，并能不依赖任何框架从零实现一个端到端的 RAG 问答系统，最终理解从手写原型到工程化产品的完整迁移路径。

📅 时效性说明：本课件内容基于 2026 年 2 月的技术生态编写。RAG 领域发展迅速，具体框架版本和最佳实践可能随时间演进，建议结合官方文档获取最新信息。

第一章：从大模型的"知识困境"到检索增强 链接到标题

  在正式拆解 RAG 架构之前，我们需要先回答一个根本性的问题：大模型明明已经很强了，为什么还需要 RAG？这个问题的答案，直接决定了我们对 RAG 价值的理解深度。本章将从大模型的四大知识局限出发，对比三种主流的知识注入方案，最终帮助大家建立起"RAG 是当前最务实的知识增强路径"这一核心认知。

1.1 大模型的四大知识局限 链接到标题

  大语言模型（LLM）的能力令人印象深刻——它能写代码、做翻译、分析数据、生成报告。但当我们把它放到真实业务场景中时，很快就会撞上四堵"知识围墙"。

幻觉问题（Hallucination）

  所谓幻觉（Hallucination），是指大模型在生成文本时，产出看似流畅合理、但与事实不符或完全虚构的内容。这是最让开发者头疼的问题。当模型遇到它不确定的问题时，它不会说"我不知道"，而是会非常自信地编造一个看起来合理但完全错误的答案。比如你问它某个公司的内部政策，它可能会"创作"一份根本不存在的制度文件。在医疗、法律、金融等对准确性要求极高的领域，这种幻觉是不可接受的。

知识截止（Knowledge Cutoff）

  每个大模型都有一个训练数据的截止日期。以 GPT-5 为例，其主模型训练数据截止到 2024 年 10 月（GPT-5.2 系列已更新至 2025 年 8 月）；Claude 4.6 Sonnet 的训练数据截止到 2026 年 1 月；Qwen3.5（2026 年 2 月发布）的训练数据预计截止到 2025 年底或 2026 年初。截止日期之后发生的事情，模型一无所知。这意味着如果你问它"上周发布的新框架有什么特性"，它只能猜测或拒绝回答。对于需要实时信息的应用场景（如新闻摘要、市场分析），这是一个硬伤。

领域盲区（Domain Blindness）

  大模型的训练数据来自公开互联网，它对你公司的内部文档、私有数据库、行业专有知识几乎一无所知。一个通用大模型无法回答"我们公司的年假制度是什么"或"这个产品的技术规格参数是多少"——这些信息从未出现在它的训练集中。

上下文窗口限制（Context Window Limit）

  即使你想通过 Prompt 把所有知识"塞给"模型，也会撞上另一堵墙——上下文窗口的长度限制。目前大部分主流模型的上下文窗口在 128K token 左右（约 10 万个汉字），少数模型支持更长窗口但推理成本会急剧上升。当你的知识库有几百页甚至上千页文档时，不可能一次性全部放进 Prompt。这个物理限制直接催生了 RAG 的核心设计思路——不把所有知识塞给模型，而是每次只检索最相关的片段，在有限的窗口内实现精准回答。

⚠️ 常见误区：很多初学者以为"模型参数越大，幻觉就越少"。实际上，幻觉问题与模型规模没有简单的线性关系。即使是最先进的模型，在面对训练数据中未覆盖的领域知识时，仍然会产生幻觉。RAG 是目前在知识密集型问答场景下最有效的幻觉缓解手段之一。 也不要以为用了 RAG 就能 100% 消除幻觉。如果检索到的文档本身有错误，或者检索结果与问题不相关，模型仍然可能产生幻觉。RAG 降低幻觉的前提是：知识库质量高 + 检索准确。

1.2 三种知识注入方案对比 链接到标题

  面对大模型的知识局限，业界发展出了三种主流的解决方案。它们各有适用场景，理解它们之间的差异，是选择正确技术路线的前提。

三种知识注入方案对比

  从表格可以看出，Prompt Engineering 最简单但容量有限——当你的知识库有几千页文档时，不可能全部塞进一个 Prompt。Fine-tuning 能让模型"内化"领域知识，但更新成本高，且不擅长处理频繁变化的信息。RAG 的核心优势在于：它把"知识存储"和"知识推理"解耦了——模型负责推理，知识库负责存储，各司其职。

  我们可以从成本维度做一个更具体的对比：Fine-tuning 一个 7B 参数的模型，使用 LoRA 方案在单张 A100 上通常需要 2-4 小时的 GPU 时间，加上数据标注成本，一次迭代的综合成本在数百到数千元不等；而搭建一个 RAG 系统，核心成本在于向量数据库（如 Milvus 云服务月费约几十到几百元）和 Embedding API 调用（以 OpenAI text-embedding-3-small 为例，100 万 token 的向量化成本约 ¥0.15）。从幻觉控制角度看，RAG 的优势在于回答可溯源——每条回答都能追溯到具体的文档片段，便于人工审核和纠错；而 Fine-tuning 后的模型，其知识已"融入"参数中，出错时难以定位原因。

🔥 踩坑预警：不少团队一上来就想用 Fine-tuning 解决所有问题，花了大量时间标注数据、训练模型，最后发现效果还不如一个设计良好的 RAG 系统。经验法则是：先用 RAG 验证可行性，只有当 RAG 无法满足质量要求时，再考虑 Fine-tuning。

1.3 RAG 的核心思想：让模型"开卷考试" 链接到标题

  理解了三种方案的差异之后，我们可以用一个直觉性的比喻来抓住 RAG 的本质。传统的大模型推理就像"闭卷考试"——模型只能依赖训练时记住的知识来回答问题。而 RAG 则是"开卷考试"——在回答之前，先让模型翻阅一下相关的参考资料，然后基于这些资料来组织答案。

  这个过程可以拆解为三个核心步骤：

1. 检索（Retrieve）：根据用户的问题，从外部知识库中找到最相关的文档片段

2. 增强（Augment）：将检索到的文档片段与用户问题拼接在一起，构成增强后的 Prompt

3. 生成（Generate）：将增强后的 Prompt 送入大模型，生成最终回答

  这三个步骤就是 RAG 名称的由来——Retrieval-Augmented Generation。它的精妙之处在于：模型不需要"记住"所有知识，只需要在回答时"查阅"相关知识。这就像一个优秀的咨询顾问——他不需要背下所有法律条文，但他知道去哪里查、怎么查、查到之后怎么用。

  上图展示了 RAG 的核心数据流。用户的问题首先经过向量化处理，然后在向量数据库中进行相似度检索，找到最相关的文档片段。这些片段与原始问题一起构成增强后的 Prompt，最终由大模型生成回答。整个过程对用户是透明的——用户只看到一个"更聪明、更准确"的 AI 回答。

广义 RAG：联网搜索也是"开卷考试"

  讲到这里，你可能会想到一个问题：大模型的"联网搜索"功能，不也是先检索再生成吗？没错。从架构模式上看，联网搜索 + LLM 的链路和 RAG 完全一致——用户提问 → 从互联网检索相关内容 → 拼入 Prompt → LLM 生成回答。本质上，联网搜索就是广义的 RAG，只不过检索源从私有知识库换成了公开互联网。

  以 DeepSeek 的联网搜索为例：关闭联网搜索时，模型从预先抓取的静态数据库中检索信息；开启联网搜索时，系统会额外发起一次实时网络搜索，获取最新内容补充到上下文中。无论哪种模式，底层都是"检索 → 增强 → 生成"的 RAG 范式。

  理解了这一点之后，我们可以把 RAG 分为两个层次来认识：广义 RAG 泛指所有"先检索外部知识，再增强生成"的架构模式，联网搜索、Perplexity、Google AI Overview 都属于此类；狭义 RAG（也是本课程的重点）特指基于私有知识库 + 向量检索的增强生成系统。两者的核心差异不在架构模式，而在检索源的可控性——私有知识库你能控制数据质量、更新节奏和权限边界；互联网检索则依赖搜索引擎的索引覆盖和排序质量，不可控因素更多。本课程后续所有内容，都聚焦在狭义 RAG 的设计与实现上。

1.4 RAG 在 LLM 应用生态中的定位 链接到标题

  在结束第一章之前，我们需要把 RAG 放到更大的技术生态中来理解它的位置。这有助于我们在后续学习中保持全局视野，而不是陷入细节。

  RAG 位于 LLM 应用技术栈的应用架构层。它的上游是基础模型层（OpenAI、Qwen、Llama 等大模型提供推理能力）和数据层（企业文档、知识库、数据库提供知识来源）；它的下游是应用层（智能客服、文档问答、知识管理等具体产品）。在同层的替代/互补方案中，Agent（智能体）侧重于任务规划与工具调用，Fine-tuning 侧重于模型能力定制，而 RAG 侧重于知识增强。三者并非互斥，在复杂应用中经常组合使用。

  举一个具体的组合场景：假设我们要搭建一个智能客服系统，Agent 负责理解用户意图并规划处理步骤（判断是查询订单、退换货还是技术咨询），RAG 负责从产品文档和FAQ知识库中检索相关信息提供给模型，而 Fine-tuning 则让模型的回答风格符合企业的客服话术规范（如语气亲切、使用特定称呼）。三者各司其职，共同支撑一个完整的智能应用。

  到这里，我们已经建立了对 RAG 的全局认知：它为什么存在、它的核心思想是什么、它在技术生态中处于什么位置。但"知道 RAG 是什么"还不够——我们还需要回答一个更实际的问题：RAG 到底能用在哪些场景？它给产品带来的核心价值是什么？接下来两节将从应用场景和产品集成两个维度，帮助大家建立对 RAG 商业价值的直觉。

1.5 RAG 核心应用场景介绍 链接到标题

  理解了 RAG 的技术原理之后，我们需要把视角从"技术怎么实现"切换到"技术能解决什么问题"。RAG 的应用场景远比很多人想象的要广泛——只要你的业务涉及"基于特定知识回答问题"，RAG 就有用武之地。

企业知识库问答

  这是 RAG 最经典、也是落地最多的场景。企业内部积累了大量的制度文档、产品手册、技术规范、会议纪要，但员工往往找不到、不想翻、翻了也看不完。通过 RAG 构建企业知识库问答系统，员工可以用自然语言直接提问——“出差报销的流程是什么？““产品 X 的最大并发数是多少？"——系统从内部文档中检索相关内容，由大模型组织成清晰的回答。这比传统的全文搜索体验好得多，因为用户得到的不是一堆文档链接，而是一个直接的答案。

智能客服系统

  客服场景是 RAG 的另一个高频落地点。传统的客服机器人依赖人工编写的问答对（FAQ），维护成本高且覆盖面有限。基于 RAG 的智能客服可以直接对接产品文档、帮助中心、历史工单等知识源，面对用户的各种提问都能给出有据可查的回答。更重要的是，当产品更新时，只需要更新知识库中的文档，客服机器人的回答就会自动跟上——不需要重新训练模型，也不需要人工更新 FAQ。

代码库检索与技术问答

  对于开发团队来说，RAG 可以用来构建代码库检索助手。把项目的源代码、技术文档、API 文档灌入知识库，开发者就可以用自然语言查询——“用户认证模块在哪个文件？““这个接口的参数格式是什么？“这在大型项目中尤其有价值，新成员不需要花几周时间熟悉代码库，通过 RAG 助手就能快速定位关键信息。我们在后续的实战案例中，就会从零搭建一个这样的代码库检索助手。

垂直领域专业问答

  在法律、医疗、金融等专业领域，RAG 的价值更加突出。这些领域对回答的准确性要求极高，通用大模型的幻觉问题是不可接受的。通过 RAG 对接专业知识库（法律条文、临床指南、金融法规），模型的每一条回答都有明确的出处，专业人士可以快速验证。例如，一个法律 RAG 系统可以在回答中标注"根据《民法典》第 XXX 条”，让律师一眼就能判断回答是否可靠。

RAG 核心应用场景一览

⚠️ 常见误区：不少团队以为 RAG 只适合"文档问答"这一种场景。实际上，只要你的业务需要"基于特定知识生成内容”，RAG 都可以发挥作用——包括但不限于报告生成、邮件草拟、数据分析解读等。关键在于你能否把相关知识组织成可检索的知识库。

1.6 RAG 核心应用价值：问答机器人产品集成 链接到标题

  在了解了 RAG 的应用场景之后，我们进一步聚焦到一个最具代表性的产品形态——问答机器人。这是 RAG 技术落地最直接、最成熟的产品载体，也是大多数团队接触 RAG 的第一个项目。

  传统的问答机器人（如基于规则匹配或意图识别的对话系统）面临两个核心痛点：一是知识维护成本高——每新增一个问题，都需要人工编写答案并配置匹配规则；二是覆盖面有限——只能回答预设范围内的问题，超出范围就"答非所问"或"无法回答”。RAG 的引入从根本上改变了这个局面。

知识实时更新 vs 模型重训的成本对比

  这是 RAG 在产品集成中最核心的价值点。我们通过一个具体的对比来说明：

知识更新方式成本对比

  从表格可以看出，RAG 方式在知识更新的敏捷性上具有压倒性优势。对于知识频繁变化的业务（如产品迭代快、政策更新频繁），RAG 几乎是唯一可行的方案。而"回答溯源"能力更是 RAG 的独特优势——当用户或审核人员质疑某条回答时，系统可以直接展示"这条回答基于《XX文档》第X页”，大幅提升了产品的可信度。

  在实际的产品集成中，一个典型的 RAG 问答机器人包含三层架构：底层是知识管理层（文档上传、解析、索引），中间是 RAG 引擎层（检索 + 增强 + 生成），上层是交互层（Web UI、API 接口、嵌入式组件）。后续我们在实战课程中搭建的 RAG 系统，就是这个架构的简化版——先把中间的引擎层跑通，再逐步扩展上下游。

  到这里，第一章的内容全部完成。我们从大模型的知识困境出发，理解了 RAG 的核心思想和技术定位，梳理了它的主要应用场景和产品集成价值。接下来的第二章，我们将深入 RAG 的内部架构，拆解每一个核心组件，理解数据从"原始文档"到"最终回答"的完整流转过程。

第二章：RAG 核心架构：检索增强生成的三大支柱 链接到标题

  在第一章中，我们理解了 RAG 的价值定位和核心思想。但"先检索再生成"这六个字背后，隐藏着一套精密的工程架构。一个生产级的 RAG 系统，从用户提问到最终回答，中间要经过文档加载、文本切分、向量化、存储、检索、生成等多个环节，每个环节的设计选择都会直接影响最终的回答质量。本章将沿着数据流转的路径，逐一拆解 RAG 的六大核心组件，并介绍 RAG 技术的四代演进脉络，帮助大家建立起完整的架构认知。

2.1 RAG 整体数据流：从文档到回答的完整链路 链接到标题

  在深入每个组件之前，我们先建立一个全局视图。一个标准的 RAG 系统包含两条数据流：

离线索引流（Indexing Pipeline）

  这是"准备知识"的过程，通常在系统初始化或知识更新时执行：原始文档 → 文档加载（Document Loader）→ 文本切分（Text Splitter）→ 向量化（Embedding Model）→ 存入向量数据库（Vector Store）。

在线查询流（Query Pipeline）

  这是"使用知识"的过程，每次用户提问时实时执行：用户问题 → 向量化 → 在向量数据库中检索相似文档（Retriever）→ 将检索结果与问题拼接为增强 Prompt → 大模型生成回答（LLM）。

  这两条流共享 Embedding Model 和 Vector Store 两个组件——索引时用 Embedding 把文档变成向量存进去，查询时用同一个 Embedding 把问题变成向量去检索。这就是为什么索引和查询必须使用同一个 Embedding 模型——如果用不同的模型，向量空间不一致，检索结果会完全错乱。

  上图清晰地展示了 RAG 的两阶段工作流程：左侧是离线的知识库构建阶段（数据接入 → 文档解析 → 文档分割 → 向量化 → 存储），右侧是在线的问答推理阶段（用户提问 → 问题向量化 → 相似度检索 → 构建增强 Prompt → LLM 生成回答）。两条流水线通过向量数据库和 Embedding 模型连接在一起，构成完整的 RAG 数据闭环。

⚠️ 常见误区：初学者经常在索引阶段和查询阶段使用不同的 Embedding 模型（比如索引用了 text-embedding-ada-002，查询时换成了 bge-large-zh），导致检索效果极差。这是 RAG 开发中最常见的"隐性 Bug"之一，因为系统不会报错，只是返回不相关的结果。

2.2 六大核心组件详解 链接到标题

  理解了整体数据流之后，我们逐一拆解六大核心组件。每个组件解决一个明确的问题，它们像流水线上的工位一样，依次处理数据。

RAG 六大核心组件功能对照

2.2.1 Document Loader：统一数据入口 链接到标题

  RAG 系统的第一步是把各种格式的原始文档"吃进来”。现实中的知识来源五花八门——PDF 报告、Word 文档、HTML 网页、Markdown 笔记、数据库记录、甚至 Notion 页面。Document Loader 的职责就是把这些异构数据统一转换为标准的 Document 对象（通常包含 page_content 文本内容和 metadata 元数据两个字段）。

  这一步看似简单，实际上是 RAG 质量的第一道关卡。如果文档解析不准确（比如 PDF 中的表格被解析成乱码、图片中的文字没有被 OCR 识别），后续所有环节都会受到影响。垃圾进，垃圾出（Garbage In, Garbage Out）——这条原则在 RAG 中体现得尤为明显。

2.2.2 Text Splitter：切分的艺术 链接到标题

  文档加载之后，我们面临一个关键问题：大模型的上下文窗口是有限的，我们不可能把整篇文档都塞进 Prompt。因此需要把长文档切分成合适大小的片段（Chunks）。

  切分策略直接影响检索质量。切得太大，检索时会引入大量无关信息；切得太小，语义会被割裂，模型无法理解上下文。业界常用的 RecursiveCharacterTextSplitter 会按照段落、句子、字符的优先级递归切分，尽量在语义边界处断开。根据 LangChain 社区和多个开源 RAG 项目的实践经验，典型的 chunk 大小在 500-1000 个 token 之间，并设置 10%-20% 的重叠（overlap）来保持上下文连贯性。所谓重叠，就是相邻两个片段之间共享一部分文本——比如第一个片段覆盖第 1-100 句，第二个片段覆盖第 80-180 句，中间的第 80-100 句就是重叠部分。这样即使切分点恰好落在一个完整概念的中间，重叠区域也能保证上下文不丢失。

🔥 踩坑预警：chunk_size 的选择没有万能公式，它取决于你的文档类型和查询模式。技术文档通常适合较大的 chunk（800-1000 tokens），因为技术概念需要完整的上下文；而 FAQ 类文档适合较小的 chunk（200-500 tokens），因为每个问答对本身就是一个独立单元。建议从 500 tokens 开始实验，根据检索效果逐步调整。

2.2.3 Embedding Model：从文字到向量 链接到标题

  切分后的文本片段还是"人类语言”，计算机无法直接比较两段文字的相似度。Embedding Model 的作用就是把文本转换为高维向量——可以把向量理解为一组数字坐标，维度越高，能捕捉的语义细节越丰富。不同模型输出的向量维度各不相同，例如 OpenAI text-embedding-3-small 为 1536 维，BGE-large-zh 为 1024 维，BGE-M3 为 1024 维（截至 2026 年 2 月）。Embedding 的核心目标是使得语义相近的文本在向量空间中距离更近。

  举个例子：“如何申请年假"和"请假流程是什么"这两句话，字面上没有重叠的词，但经过 Embedding 之后，它们的向量会非常接近——因为模型理解了它们的语义是相似的。这就是向量检索比传统关键词检索更强大的原因：它匹配的是"意思”，而不是"字面”。

2.2.4 Vector Store：向量的仓库 链接到标题

  有了向量之后，我们需要一个高效的存储和检索系统。Vector Store（向量数据库）就是专门为此设计的。它支持"给定一个查询向量，快速找到最相似的 K 个向量"这种操作，底层通常使用 ANN（Approximate Nearest Neighbor，近似最近邻）算法来实现毫秒级检索。为什么是"近似"而非精确？因为在百万级甚至千万级向量中做精确最近邻搜索太慢，ANN 通过牺牲极小的精度换取数量级的速度提升，实际召回率通常在 95% 以上，对 RAG 场景完全够用。

  常见的向量数据库包括：FAISS（Meta 开源，适合本地实验）、Chroma（轻量级，适合原型开发）、Milvus（分布式，适合生产环境）、Weaviate（支持混合检索）。选择哪个取决于你的数据规模和部署环境。

2.2.5 Retriever：检索策略 链接到标题

  Retriever 是连接"知识库"和"大模型"的桥梁。最基础的检索方式是纯向量检索——根据余弦相似度返回 Top-K 结果。所谓余弦相似度，就是衡量两个向量方向的接近程度，值越接近 1 表示语义越相似，越接近 0 表示越不相关。但在实际应用中，单一的向量检索往往不够。比如用户搜索一个精确的产品编号"SKU-20240315"，向量检索可能找不到精确匹配，因为 Embedding 模型更擅长语义匹配而非精确匹配。

  因此，生产级 RAG 系统通常采用混合检索（Hybrid Search）策略：向量检索负责语义匹配，BM25 等稀疏检索算法负责基于关键词的统计相关性匹配。BM25 是经典的基于词频和文档长度的概率检索模型，可以理解为"升级版的关键词搜索"——它不只看关键词是否出现，还会考虑词频、文档长度等统计特征来排序。两路检索结果最后通过 RRF（Reciprocal Rank Fusion，倒数排名融合）等算法进行融合——RRF 根据每条结果在各路检索中的排名来计算综合得分，排名越靠前得分越高。这种"两条腿走路"的方式能显著提升检索的召回率和准确率。

2.2.6 LLM：最终的生成引擎 链接到标题

  检索到相关文档片段后，最后一步是把它们和用户问题一起交给大模型生成回答。这一步的关键在于 Prompt 的构造——需要明确告诉模型"请基于以下参考资料回答问题，如果资料中没有相关信息，请说明无法回答"。这种指令可以有效减少幻觉，因为模型被引导去"引用"而非"创作"。

  上图是 RAG 完整流程的总结视图。在继续深入之前，有三个关键认知需要在这里锚定，它们是初学者最容易混淆的点：

📌 RAG 认知锚点（必须记住的三件事） 第一，整个 RAG 流程涉及两类模型：一类是用于将文本转换为向量的 Embedding 模型（如 text-embedding-v3、BGE），另一类是用于最终生成回答的 LLM 模型（如 GPT-5、Qwen3.5）。这两类模型职责完全不同，不要混为一谈。特别注意：知识库构建阶段（离线索引）和用户提问阶段使用的 Embedding 模型必须保持一致，否则向量空间不匹配，检索结果会完全错乱。 第二，LLM 只负责最后一步：知识库的构建（文档加载、切分、向量化、存储）全程不涉及 LLM。LLM 只在用户提问时才登场，负责将检索到的知识片段和用户问题整合后生成回答。 第三，RAG 的最终目的是构建一个高质量的 Prompt：前面所有的检索、切分、向量化操作，归根结底都是为了拼出一个包含相关知识的 Prompt，然后交给 LLM。Prompt 是与 LLM 交互的唯一方式——理解了这一点，就理解了 RAG 的本质。

  至此，我们已经走完了 RAG 的完整数据链路。六个组件各司其职，共同构成了一个从"原始文档"到"准确回答"的完整流水线。

2.3 四代 RAG 演进：从 Naive 到 Agentic 链接到标题

  了解了 RAG 的核心组件之后，我们还需要知道 RAG 技术本身也在快速演进。从最初的"简单拼接"到如今的"智能体驱动"，RAG 已经经历了四代发展。理解这个演进脉络，能帮助我们在实际项目中做出更合理的架构选择——不同代际解决的问题层次完全不同，选错架构要么杀鸡用牛刀，要么力不从心。

第一代：Naive RAG（朴素 RAG）

  最早期的 RAG 实现非常直接：把文档切块、向量化、存入数据库，查询时检索 Top-K 片段拼进 Prompt，交给 LLM 生成。这种方式实现简单，但存在明显的问题——检索质量不稳定（可能检索到不相关的内容）、没有对检索结果进行排序和过滤、切分策略粗糙导致语义割裂。对于简单的文档问答场景，Naive RAG 已经够用；但面对复杂查询，它的表现往往不尽如人意。

第二代：Advanced RAG（进阶 RAG）

  针对 Naive RAG 的不足，Advanced RAG 在检索前后引入了多个优化环节。检索前（Pre-Retrieval）：对用户查询进行改写、扩展或分解，提升检索命中率；检索后（Post-Retrieval）：对检索结果进行重排序（Reranker）、过滤、压缩，确保送入 LLM 的内容高度相关。此外，Advanced RAG 还引入了更精细的切分策略（如基于语义的切分）和混合检索（向量 + 关键词）。这一代 RAG 是目前大多数生产系统的主流选择。

第三代：Modular RAG（模块化 RAG）

  最新的 Modular RAG 把 RAG 系统拆解为可自由组合的功能模块——检索模块、路由模块、重排模块、生成模块等，每个模块可以独立替换和升级。它还引入了"自适应检索"的概念：不是每个问题都需要检索，模型可以先判断是否需要外部知识，再决定是否触发检索流程。这种架构更灵活，但也更复杂，适合对 RAG 有深度定制需求的团队。

第四代：Agentic RAG（智能体 RAG）

  如果说 Modular RAG 实现了"模块可插拔"，那 Agentic RAG 则更进一步——它引入了一个 LLM Agent 作为整个 RAG 流程的"大脑"，由 Agent 动态决定每一步该做什么。面对一个复杂问题，Agent 不再沿着固定流水线走，而是自主判断：是否需要检索？检索哪个知识库？检索结果够不够好？要不要换个查询重新检索？要不要调用其他工具（如计算器、代码执行器、数据库查询）来辅助回答？这种"规划—执行—反思—修正"的循环，让 RAG 系统具备了处理多跳推理、跨源整合、自我纠错等复杂任务的能力。Agentic RAG 与前三代的本质区别在于：前三代的流程是人设计的，而 Agentic RAG 的流程是 Agent 自己"想"出来的。代表框架包括 LangGraph（基于状态图的 Agent 编排）、LlamaIndex Agents（工具化检索 Agent）和 CrewAI（多 Agent 协作）。不过，Agent 的自主性也带来了可控性和延迟方面的挑战——Agent 可能"想多了"导致响应变慢，也可能走错路径产生不可预期的结果，因此目前更适合对回答质量要求极高、可以容忍较长响应时间的场景。

四代 RAG 技术演进对比

  从表格可以看出，四代 RAG 并非简单的"新的一定比旧的好"。选择哪一代架构，取决于你的业务复杂度、响应时间要求和团队能力。如果你刚开始接触 RAG，从 Naive RAG 入手快速验证想法是最务实的选择；当验证通过需要上线时，再逐步引入 Advanced RAG 的优化手段。Modular RAG 的代表框架中，LlamaIndex Workflows 通过可编排的事件驱动流水线实现模块化，Haystack 2.0 采用 Pipeline 架构支持组件自由组合，DSPy 则用编程范式自动优化 Prompt 和检索策略——它们从不同角度诠释了"模块化"的理念。而当你的场景涉及多跳推理（如"对比 A 公司和 B 公司最近三年的营收趋势"）、需要跨多个知识源整合信息、或者对回答准确性有极高要求时，Agentic RAG 就成了值得考虑的选项。LangGraph 通过状态图定义 Agent 的决策路径，LlamaIndex Agents 将检索封装为 Agent 可调用的工具，CrewAI 则支持多个 Agent 分工协作——选择时应根据团队技术栈和场景复杂度来决定。

  举一个具体的选型场景：假设你要为公司内部搭建一个 IT 运维知识问答系统，文档量约 500 篇，用户查询以自然语言为主。这种场景下，Advanced RAG（混合检索 + Reranker）是性价比最高的选择——Naive RAG 的检索精度不够，Modular RAG 的复杂度对这个规模来说过重，Agentic RAG 的响应延迟对运维场景也不太友好。但如果需求升级为"跨多个系统（运维文档 + 工单数据库 + 监控日志）进行关联分析并给出排障建议"，这时候就需要 Agentic RAG 出场了——Agent 可以先查文档确认故障类型，再查工单看历史处理方案，最后查监控日志定位根因，整个过程由 Agent 自主编排。

2.4 主流框架选型概览 链接到标题

  最后，我们简要了解一下当前 RAG 开发中最常用的两个框架，为后续的实战课程做好准备。

  LangChain 是目前最流行的 LLM 应用开发框架，它提供了完整的 RAG 工具链——从 Document Loader 到 Vector Store 到 Chain，开箱即用。它的优势是生态丰富、社区活跃、集成了几乎所有主流的模型和数据库；劣势是抽象层次较多，调试时有时需要深入源码。LangChain 适合快速原型开发和大多数标准 RAG 场景。

  LlamaIndex 则更专注于数据索引和检索优化。它在文档解析、索引结构、检索策略方面提供了更精细的控制——比如树状索引、知识图谱索引、自动合并检索等高级特性。如果你的 RAG 系统需要处理复杂的文档结构（如多层级目录、表格密集型文档），LlamaIndex 可能是更好的选择。不过它的学习曲线相对较陡，社区规模不及 LangChain，且文档更新有时滞后于代码变更，上手时需要多参考源码和示例。

  除了这两个主流框架，Haystack（deepset 开源，Pipeline 架构清晰，适合生产部署）、Semantic Kernel（微软出品，与 Azure 生态深度集成）、Dify（开源 LLMOps 平台，提供可视化 RAG 编排）等也值得关注。选择框架时，核心考量是团队技术栈、生态集成需求和定制深度。

  两个框架并非互斥。在实际项目中，不少团队会用 LlamaIndex 构建索引和检索层，用 LangChain 构建上层的 Chain 和 Agent 逻辑，取两者之长。

RAG 落地的四条路径

  除了选择框架，还需要明确 RAG 的落地方式。根据团队的技术能力和业务需求，当前主流的 RAG 实践路径可以归纳为四种：

RAG 四种落地路径对比

  这四条路径并非互斥。一个常见的演进路线是：先用低代码平台（如 Dify）快速验证业务可行性，确认 RAG 能解决问题后，再用 LangChain 或手写方式重构为可控的生产系统。本课程的实战部分会从"手动搭建"入手，让你彻底理解每个环节的原理，之后再接触框架时就能做到"知其然也知其所以然"。

实际部署前的四个关键考量

  在选定技术路径之后、正式开发之前，还需要从业务视角评估四个维度，它们决定了你的 RAG 系统能否真正落地：

1. 数据质量：是否有高质量、结构化的知识源？如果原始文档质量差（扫描版 PDF、格式混乱的 Word），文档解析环节就会成为瓶颈

2. 更新频率：知识需要多频繁更新？如果知识库每天都在变化（如产品文档、政策法规），需要设计自动化的索引更新流程

3. 准确度要求：错误回答的代价有多大？医疗、法律、金融等场景对准确度要求极高，需要更精细的检索策略和人工审核机制

4. 技术复杂度：团队是否具备相应的技术能力？如果团队没有 ML 工程师，低代码平台可能是更务实的起点

2.5 RAG 热门开源项目产品介绍 链接到标题

  了解了开发框架之后，我们还需要知道：如果你不想从零开发，市面上已经有哪些开箱即用的 RAG 产品？这些开源项目可以帮助团队快速验证 RAG 的业务价值，也可以作为自研系统的参考架构。

MaxKB：开箱即用的知识库问答产品

📌 项目定位：以 RAG 为核心的产品化知识库问答系统，面向非研发背景用户，零代码即可搭建知识库机器人。⭐ GitHub Star：约 20.2k（2026 年 2 月数据，会持续增长）

  MaxKB 是飞致云（FIT2CLOUD）开源的知识库问答系统，主打"开箱即用"。它提供了完整的 Web 管理界面，支持文档上传、自动解析、向量化索引、对话问答等全流程功能，无需编写代码即可搭建一个可用的知识库问答机器人。MaxKB 支持对接多种大模型（OpenAI、通义千问、文心一言等），内置了文档解析和切分逻辑，适合希望快速上线 RAG 产品的团队。需要注意的是，MaxKB 的核心优势是部署简单、上手门槛低，但在检索策略和 Prompt 定制方面的灵活性相对有限，更适合验证阶段或中小规模场景，而非定制化需求强的生产系统。

RAGFlow：企业级 RAG 编排平台

📌 项目定位：以深度文档理解为核心差异化能力的 RAG 专项引擎，覆盖 RAG 全链路，曾入选 GitHub 2025 年度增长最快开源项目。⭐ GitHub Star：约 70k（2026 年 2 月，全球 RAG 专项项目 Star 数第一）

  RAGFlow 是 InfiniFlow 开源的企业级 RAG 引擎，它的核心亮点在于文档解析能力——支持复杂 PDF（含表格、图片、多栏排版）的深度解析，这是很多 RAG 系统的薄弱环节。RAGFlow 提供了可视化的 RAG 流程编排界面，用户可以自定义检索策略、切分方式、Prompt 模板等。它还内置了"文档理解"模块，能自动识别文档结构并进行智能切分，而不是简单的按字数切割。值得关注的是，RAGFlow 是这三个项目中唯一一个以 RAG 为全部核心功能、面向全球市场的专项 RAG 引擎，其 70k Star 也印证了社区对"深度文档理解 + 精细编排"这一路线的高度认可。适合对文档解析质量要求高、需要精细化控制 RAG 流程的企业场景。

LangChain-Chatchat：基于 LangChain 的本地知识库对话系统

📌 项目定位：以 RAG 知识库问答为核心功能、面向数据安全敏感场景的本地化中文对话系统，是国内 RAG 开源社区的标志性项目。⭐ GitHub Star：约 37.3k（2026 年 2 月数据）

  LangChain-Chatchat（原 langchain-ChatGLM）是国内社区最早的开源 RAG 项目之一，以 RAG 知识库问答为核心功能，以 LangChain 作为底层框架构建。它支持本地部署大模型（如 ChatGLM、Qwen）+ 本地向量数据库（FAISS），实现完全离线的知识库问答。对于数据安全要求高、不允许调用外部 API 的场景（如政务、军工、金融内网），这是一个重要的选择。需要注意的是，LangChain-Chatchat 主要面向中文场景和国产大模型生态，与 ChatGLM 的深度绑定历史使其在国际社区知名度相对有限；但在国内开发者群体中，它仍然是"本地化 RAG"方向的首选参考项目。项目提供了 Streamlit Web UI，支持多轮对话和知识库管理。

热门 RAG 开源项目对比

  这三个项目代表了 RAG 产品化的三种路线：MaxKB 走"低门槛快速上线"路线，RAGFlow 走"深度文档理解 + 精细编排"路线，LangChain-Chatchat 走"完全本地化部署"路线。在实际选型时，建议先明确你的核心需求——是快速验证、深度定制还是数据安全——再选择对应的项目作为起点。

🔥 踩坑预警：开源 RAG 项目迭代非常快，功能和架构可能在几个月内发生重大变化。在选型时，除了看功能列表，还要关注项目的社区活跃度（GitHub Star 增长趋势、Issue 响应速度）和最近的提交频率。一个半年没有更新的项目，即使功能看起来很全，也要谨慎选择。

🎉 目前，你已经掌握了：

• ✅ 大模型的四大知识局限（幻觉、知识截止、领域盲区、上下文窗口限制），以及为什么 RAG 是当前最务实的解决方案
• ✅ RAG 的核心思想——“先检索再生成”，以及它与 Prompt Engineering、Fine-tuning 的本质区别
• ✅ 广义 RAG 与狭义 RAG 的关系——联网搜索本质上也是 RAG，区别在于检索源的可控性
• ✅ RAG 的核心应用场景（企业知识库、智能客服、代码检索、垂直领域问答）
• ✅ RAG 与问答机器人产品集成的核心价值，以及知识更新 vs 模型重训的成本对比
• ✅ RAG 的完整数据流（离线索引 + 在线查询）和六大核心组件的职责
• ✅ RAG 系统中两类模型（Embedding 模型 vs LLM）的职责划分
• ✅ 三代 RAG 的演进脉络（Naive → Advanced → Modular）和选型依据
• ✅ LangChain 与 LlamaIndex 两大框架的定位差异，以及四种落地路径的选择
• ✅ MaxKB、RAGFlow、LangChain-Chatchat 三大热门开源项目的特点与选型

  到这里，我们完成了 RAG 的全部认知建设——从"为什么需要"到"由什么组成"再到"怎么演进"。接下来，是时候把这些认知真正转化为可运行的代码了。第三章我们将不借助任何 SDK 框架，全部用 Python + 基础依赖手写实现一个端到端的 RAG 问答系统。

📅 时效性与环境说明：本课件基于 2026 年 2 月的技术生态编写，测试环境为 Python 3.11 + openai 1.109.1（1.x 分支；2.x 已发布但存在破坏性变更，本课件维持 1.x API）+ numpy 2.4.2 + faiss-cpu 1.13.2 + pypdf 6.7.3。

  在运行本章任何代码之前，请先执行以下环境准备 cell，requirements.txt已为大家提供，安装所有必要依赖：

# 环境依赖安装（版本已在 2026-02 验证，如遇冲突请参考上方时效性说明）
!pip install -r requirements.txt

  安装完成后，验证关键包版本是否符合预期：

!python --version

import openai, numpy, faiss, pypdf
print(f"openai:    {openai.__version__}")
print(f"numpy:     {numpy.__version__}")
print(f"faiss:     {faiss.__version__}")
print(f"pypdf:     {pypdf.__version__}")

  预期输出：openai 1.x、numpy 2.4.x、faiss 1.13.x、pypdf 6.7.x。如果版本不符，优先检查当前 Python 环境是否与其他项目存在依赖冲突，建议使用独立的虚拟环境（venv 或 conda）运行本课件。

第三章：从零到一手动搭建 RAG 系统 链接到标题

  本章是整个 RAG 课程从"知道"走向"会做"的关键一章。在前两章建立了完整认知框架之后，我们需要把每一个组件的职责"落地"为真实可运行的代码。本章的 10 节内容严格沿着第二章介绍的两条数据流展开——前半段对应离线索引流（知识库构建），后半段对应在线查询流（实时问答），每一节都是独立的功能单元，也是最终系统的一块拼图。

  具体来说，3.1-3.2 是两阶段共用的基础准备（Embedding 认知与 API 接入）；3.3-3.6 对应离线索引阶段，从文档加载、文本切分、余弦相似度理解，到 FAISS 向量索引的构建与持久化——跑完这四节，你的知识库就"建好了"；3.7-3.10 对应在线查询阶段，从 Top-K 检索、Prompt 设计、Token 预算控制，到完整 Pipeline 串联——跑完这四节，你的 RAG 系统就能"回答问题了"。

3.1 Embedding 模型原理与选型 链接到标题

  在动手写代码之前，我们需要先搞清楚一个基础问题：为什么计算机能比较两段文字的"相似度"？答案正是 Embedding——文本向量化技术。这一节是纯概念建立，不写代码，但它是理解后续所有步骤的数学基础。

什么是 Embedding

  Embedding（向量化）是将文本映射为固定长度浮点数数组的过程。可以把它理解为：把一段文字"翻译"成一组坐标，让语义相近的文字在这个坐标空间里距离更近。举个直觉性的例子——“如何申请年假"和"请假流程是什么”，虽然用词完全不同，但经过 Embedding 之后，它们对应的坐标（向量）会非常接近；而"如何申请年假"和"Python 列表排序"的向量则会相距很远。这就是 RAG 能够做到"语义检索"的核心机制。

  上图直观地展示了 Embedding 的核心思想：文本被转换为向量空间中的点，语义相近的文本在空间中距离更近，语义无关的文本则相距较远。向量之间的距离远近，就对应着语义相似度的大小。

向量维度的含义

  Embedding 模型输出的向量有固定的维度（Dimension），维度越高，能捕捉的语义细节越丰富，但计算和存储成本也越高。常见的模型维度从 512 到 3072 不等。对于中文 RAG 场景，1024 维通常已经足够，不必盲目追求更高维度。

  上图展示了向量维度的物理含义：每个下标 i 代表一个维度，整个浮点数数组对应 n 维空间中的一个点。维度越多，这个"坐标系"就越精细，能区分的语义差异也越细微。但在实际选型中，维度并非越高越好——需要在精度和计算成本之间找到平衡。

主流 Embedding 模型对比

主流 Embedding 模型对比（2026 年 2 月）

  从表格可以看出，对于以中文为主的 RAG 场景，Qwen text-embedding-v3 在语义匹配效果上通常优于 OpenAI 的模型——这主要是因为 Qwen Embedding 的训练数据中包含更大比例的中文语料，在 C-MTEB（中文文本嵌入基准）等评测中，中文专用模型的平均得分普遍高于通用多语言模型。维度上 1024 也比 1536 更紧凑，存储和检索成本更低。本章的代码将同时支持两种模型，你可以根据实际场景自由切换。

⚠️ 常见误区：很多初学者以为"维度越高越好"。实际上，对同一模型来说更高维度确实保留更多信息；但跨模型比较时，低维度的专业中文模型（如 text-embedding-v3）可能比高维度的通用模型效果更好。选模型要看使用场景和中文覆盖，而不是只看维度数字。

3.2 Embedding API 接入实战 链接到标题

  理解了 Embedding 的原理之后，我们来正式接入 API。本节的核心目标是封装一个通用的 get_embedding() 函数，让它既能调用 OpenAI，也能调用 Qwen——只需切换 client 和 model 参数，无需修改任何业务代码。

统一入口：用 openai SDK 对接两个平台

  Qwen Embedding 兼容 OpenAI 协议，这意味着我们只需要一套 openai SDK，通过切换 base_url（即 API 的服务地址，不同平台的模型托管在不同的服务器上，切换这个地址就相当于切换了后端服务商）就能对接两个平台。这是本章的核心工程决策之一——学员只需掌握一套调用范式，就能无缝切换主流平台。在初始化 client 之前，我们先从本地 .env 文件中加载 API Key，切勿将 Key 硬编码在代码中，否则一旦上传到代码仓库就会造成密钥泄露。

  在项目根目录创建 .env 文件，填入以下内容（不要加引号，不要提交到 Git）：

OPENAI_API_KEY=sk-你的OpenAI密钥
DASHSCOPE_API_KEY=sk-你的阿里云密钥

  然后在 Notebook 的第一个 cell 中加载环境变量，再初始化两个 client：

import os
from dotenv import load_dotenv
from openai import OpenAI

# 加载 .env 文件中的环境变量（override=True 确保 .env 优先级高于系统环境变量）
load_dotenv(override=True)

# ── OpenAI Embedding ──────────────────────────────────────────
client_openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# ── Qwen Embedding（兼容 OpenAI 协议，只需切换 base_url）──────
client_qwen = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

  os.getenv("OPENAI_API_KEY") 会从已加载的环境变量中读取对应的值。如果返回 None（说明 .env 文件中没有这个 key 或未正确加载），OpenAI() 会抛出 AuthenticationError——这是一个有用的快速失败机制，比程序运行到 API 调用时才报错要好得多。两个 client 的对象类型完全相同，后续所有调用 client.embeddings.create() 的地方，只需传入不同的 client 实例和 model 名称，业务逻辑代码无需任何改动。

  有了 client 之后，我们封装两个核心函数：单文本向量化（get_embedding）和批量向量化（get_embeddings_batch）。单次调用适合实时查询，批量调用适合离线文档索引——两种场景在 RAG 中都会用到：

def get_embedding(text: str, client, model: str) -> list[float]:
    """单文本向量化：返回一个浮点数列表"""
    response = client.embeddings.create(input=[text], model=model)
    return response.data[0].embedding


def get_embeddings_batch(texts: list[str], client, model: str) -> list[list[float]]:
    """批量向量化：一次 API 调用处理多个文本，节省延迟和成本"""
    response = client.embeddings.create(input=texts, model=model)
    # 注意：返回结果的顺序与 input 列表顺序严格一致
    return [item.embedding for item in response.data]

  get_embedding 函数将 text 包装在列表中传给 API（input=[text]），然后取返回值的第一个 embedding 字段；get_embeddings_batch 则一次性传入所有文本，遍历 response.data 拼接结果。两个函数的返回值格式完全一致（list[float]），可以无缝传给后续的 FAISS 索引构建函数。

验证 Embedding 的效果

  让我们用几行代码验证 Embedding 的物理含义。运行下面的代码，打印向量的维度和前 5 个数值，直观感受"向量"这个概念：

# 请确认模型名称为当前可用版本（查询日期：2026-02）
# text-embedding-v3 仍有效；如需更强中文能力可升级为 text-embedding-v4
# openai的embedding模型为：text-embedding-3-small
MODEL = "text-embedding-v3"
client = client_qwen   # 或切换为 client_openai

vec1 = get_embedding("如何申请年假", client, MODEL)
vec2 = get_embedding("请假流程是什么", client, MODEL)
vec3 = get_embedding("今天天气怎么样", client, MODEL)

print(f"向量维度：{len(vec1)}")       # 应为 1024
print(f"前 5 个数值：{vec1[:5]}")     # 浮点数数组，正负均有

  预期输出：向量维度：1024，前 5 个数值 为一组接近 0 的浮点数（如 [-0.028, 0.042, ...]）。维度是 1024 说明调用了正确的模型；至于 vec1 和 vec2 是否真的比 vec1 和 vec3 更接近，我们将在 3.5 节手写余弦相似度后正式验证。

🔥 踩坑预警：调用 Qwen Embedding 时，model 参数必须填 "text-embedding-v3"（而非 "Qwen-embedding-v3" 或其他变体）。若填错模型名，API 会返回 404 model not found 错误，且报错信息往往不够直观。建议第一次接入时直接查阅阿里云百炼控制台的"模型列表"页面确认正确的模型 ID。

🔥 踩坑预警：load_dotenv() 默认在当前工作目录查找 .env 文件。如果你的 Notebook 不在项目根目录下运行，需要传入绝对路径：load_dotenv("/path/to/.env")。加载失败时不会报错，os.getenv() 会返回 None，直到 API 调用时才抛出 AuthenticationError——排查时请优先检查 .env 文件路径是否正确。

📦 离线索引阶段：构建知识库 链接到标题

  Embedding 的概念和 API 工具已经就绪，接下来我们正式进入 RAG 系统的离线索引阶段——也就是第二章中介绍的 Indexing Pipeline（Pipeline 即"流水线"，指数据依次经过多个处理步骤的链式流程）。这个阶段的目标是把原始文档转化为可检索的向量索引，整个过程只需要执行一次（或在知识库更新时重新执行），不涉及用户提问，也不涉及 LLM。

  接下来的四节（3.3-3.6）将沿着 原始文档 → 文档加载 → 文本切分 → 向量化 → FAISS 索引持久化 的链路逐步推进。跑完这四节，你的知识库就"建好了"，可以保存到磁盘随时加载使用。

3.3 文档加载：手写 Document Loader 链接到标题

  能调通 Embedding API 之后，下一步是把知识库中的文档"读进来"。这一步被称为 Document Loader，它的职责是把各种格式的文件统一转换为结构化的文本对象。看起来简单，却是影响 RAG 系统质量的第一道关卡——垃圾进，垃圾出（Garbage In, Garbage Out），文档解析不准确，后续所有步骤都会受到影响。

设计目标：统一输出结构

  无论输入是 PDF、TXT 还是 Markdown，我们的 Document Loader 都应该输出相同结构的字典对象：{"content": 文本内容, "metadata": 来源信息}。其中 metadata 记录文件路径、文件类型等溯源信息，为后续检索结果追溯提供依据——当用户问"这个回答的依据是什么"时，我们需要能指向具体文件。以下是完整实现：

import os
from pypdf import PdfReader  # 注意：pypdf 是 PyPDF2 的官方继任者（PyPDF2 已停止维护），如果你在网上搜到的教程使用 import PyPDF2，请改为 import pypdf，API 基本兼容


def load_txt(file_path: str) -> dict:
    """加载 TXT / Markdown 文件，统一编码为 utf-8"""
    with open(file_path, 'r', encoding='utf-8') as f:
        return {
            "content": f.read(),
            "metadata": {"source": file_path, "type": "txt"}
        }


def load_pdf(file_path: str) -> dict:
    """加载 PDF 文件：逐页提取文本，用换行符拼接各页内容"""
    reader = PdfReader(file_path)
    text = "\n".join(page.extract_text() or "" for page in reader.pages)
    return {
        "content": text,
        "metadata": {"source": file_path, "type": "pdf", "pages": len(reader.pages)}
    }


def load_document(file_path: str) -> dict:
    """统一入口：根据扩展名自动选择加载器"""
    ext = os.path.splitext(file_path)[1].lower()
    loaders = {".txt": load_txt, ".md": load_txt, ".pdf": load_pdf}
    loader = loaders.get(ext)
    if not loader:
        raise ValueError(f"不支持的文件格式: {ext}（当前支持：.txt .md .pdf）")
    return loader(file_path)

  代码的核心设计是"分发器"模式：load_document() 作为统一入口，根据文件扩展名自动选择对应的加载函数。外部调用者只需传入文件路径，无需关心底层格式。.md 文件和 .txt 文件共用同一个加载器，因为 Markdown 本质上也是纯文本，直接读取即可。

验证步骤一：先加载结构简单的文件，建立基准认知

  我们先从 TXT 和 Markdown 文件开始验证，这两种格式提取效果最干净，能给我们一个清晰的"成功基准"：

doc_txt = load_document("data/python_faq.txt")

print(f"TXT 字符数：{len(doc_txt['content'])}")
print(f"TXT 前 150 字：\n{doc_txt['content'][:150]}")

  TXT 文件验证通过后，我们用同样的方式加载 Markdown 文件，确认 load_document() 的格式分发逻辑正确：

doc_md  = load_document("data/product_spec.md")

print(f"\nMD 字符数：{len(doc_md['content'])}")
print(f"MD 前 150 字：\n{doc_md['content'][:150]}")
print(f"MD metadata：{doc_md['metadata']}")

  预期输出：TXT 文件开头是"Python 常见问题 FAQ"，MD 文件开头是"# DataFlow Pro 数据处理平台…"，内容干净，结构清晰，metadata 中 source 和 type 字段均正确填充。这说明 load_document() 对纯文本格式工作正常。

验证步骤二：加载 PDF，先看"成功"的部分

  接下来加载 PDF 文件。从字符数和页数来看，提取似乎也成功了：

doc_pdf = load_document("data/company_leave_policy.pdf")

print(f"PDF 页数：{doc_pdf['metadata']['pages']}")
print(f"PDF 字符总数：{len(doc_pdf['content'])}")
print(f"\nPDF 内容前 200 字：\n{doc_pdf['content'][:200]}")

  预期看到页数为 4，字符总数约 2000+，内容里能看到"第一章 总则"等正文文字——表面上一切正常。但如果仔细看前几行，会发现一些奇怪的内容。我们放大来看。

验证步骤三：放大镜——主动暴露两个真实的坑

  用下面的代码仔细检查提取结果，你会发现两个在生产环境中非常常见的问题：

# ── 坑一：页眉页脚混入正文 ──────────────────────────────────
print("=== 提取内容的前 4 行（用 repr 显示，方便看清空白字符）===")
for line in doc_pdf['content'].split('\n')[:4]:
    if line.strip():
        print(repr(line))

print()

# ── 坑二：表格结构丢失 ──────────────────────────────────────
content = doc_pdf['content']
idx = content.find("假期类型")
print(f"=== '假期类型' 附近的 300 字（原本是一张 5 列对齐的表格）===")
print(content[idx:idx+300])

  运行后你会看到两个让人意外的结果：

  坑一的实际输出（页眉页脚混入正文）：

'XX科技有限公司 内部文件 — 保密级别：内部'
'第 1 页 / 共 3 页'
'文件编号：HR-2026-001  版本：v3.2  生效日期：2026年1月1日'
' XX 科技有限公司'

  提取内容的前三行根本不是正文，而是页眉和页脚信息。pypdf 按照 PDF 内部的文字流顺序提取，页眉恰好排在最前面。这些噪声会被原封不动地切分进 chunk，最终影响检索质量——想象一下，用户问"公司文件编号是什么"，RAG 系统可能会把页脚里的"HR-2026-001"当成答案返回。

  坑二的实际输出（表格结构丢失）：

假期类型
天数
适用条件
是否带薪
申请提前期
年假
5-15天
工作满1年，按工龄递增
是
提前3个工作日
婚假
3天
...

  原本是一张 5 列对齐的表格，提取后每个单元格变成了独立的一行，列与列之间的对应关系完全丢失。如果用户问"年假需要提前几天申请"，这段乱序文字里虽然包含"年假"和"提前3个工作日"，但它们之间的关联已经被破坏，LLM 很难从中准确提取答案。

🔥 踩坑预警：页眉页脚噪声和表格结构丢失是 PDF 解析中最普遍的两个问题，也是 RAG 系统"答非所问"的常见根源之一。本课件的示例数据刻意保留了这两个问题，让你在安全的学习环境中亲眼看到它们。生产环境的处理方案包括：用正则过滤页眉页脚模式、对含表格的 PDF 改用 pdfplumber 或 pymupdf 提取、或将表格转为 Markdown 格式后再入库——这些进阶技巧将在后续章节的结构化数据 RAG 中详细讲解。

⚠️ 常见误区：pypdf 的 extract_text() 在处理扫描版 PDF 时会返回空字符串，而不是报错。这意味着你的知识库可能静默地装入了"空文档"，而系统不会告警。生产环境中建议在 load_pdf 中加入内容长度检查：

if len(doc["content"].strip()) < 50:
    print(f"⚠️  警告：{file_path} 提取内容过短，可能是扫描版 PDF")

3.4 文本切分：手写 Text Splitter 链接到标题

  文档加载之后，我们面临一个关键决策：把长文档切成多大的片段？切得太大，检索时会引入大量无关信息；切得太小，语义会被割裂，模型无法从碎片中理解上下文。本节我们手写一个支持递归分隔符和 overlap 的文本切分器，理解这个决策背后的工程逻辑。

两个关键参数：chunk_size 与 overlap

  chunk_size 控制每个片段的最大字符数，overlap 控制相邻两个片段共享的字符数。overlap 的作用是防止关键概念恰好被切分点割裂——如果一个完整概念横跨两个片段的边界，重叠区域能保证两个片段都包含这个概念的部分信息，提升检索时的命中率。

  上图直观地展示了文档切分的物理过程：长文档被切成多个 chunk，相邻 chunk 之间有一段重叠区域（overlap）。重叠区域确保了即使切分点落在一个完整概念的中间，两个相邻片段都能保留这个概念的部分信息。

递归分隔符策略

  递归分隔符切分的思路是：优先在"语义边界"处切开，按照 \n\n（段落）→ \n（行）→ 。（句子）→ （词）的优先级依次尝试。只有在大分隔符切出的片段仍然超长时，才降级用小分隔符继续切，这样能最大程度地保留语义完整性。以下是完整实现：

def split_text(text: str, chunk_size: int = 500, overlap: int = 100,
               separators: list[str] = None, metadata: dict = None) -> list[dict]:
    """
    递归字符切分器（保真版）

    设计约束：
    - chunk 的最终长度不超过 chunk_size
    - overlap 包含在 chunk_size 内
    - 不吞掉原文中的分隔符和连续空白
    """
    # 先做参数校验，避免出现无限切分或 overlap 挤占掉全部有效内容。
    if chunk_size <= 0:
        raise ValueError("chunk_size must be greater than 0")
    if overlap < 0:
        raise ValueError("overlap must be greater than or equal to 0")
    if overlap >= chunk_size:
        raise ValueError("overlap must be smaller than chunk_size")

    # 默认按“段落 -> 换行 -> 句子 -> 词 -> 字符”的粒度逐级降级。
    if separators is None:
        separators = ["\n\n", "\n", "。", ".", " ", ""]
    metadata = metadata or {}

    # 预留 overlap 空间，确保最终 content 长度仍不超过 chunk_size。
    payload_size = chunk_size - overlap if overlap > 0 else chunk_size

    def _split_keep_separator(value: str, sep: str) -> list[str]:
        """
        根据指定分隔符切分文本，并确保分隔符保留在切分后的片段末尾。
        这样做可以避免在 chunk 边界处丢失原文的结构信息（如句号、换行符等）。
        """
        pieces = []
        start = 0
        while True:
            # 从当前起始位置查找分隔符
            index = value.find(sep, start)

            # 如果找不到分隔符，说明已到达文本末尾
            if index == -1:
                tail = value[start:]
                # 如果末尾还有剩余字符，则作为最后一个片段加入
                if tail:
                    pieces.append(tail)
                break
            
            # 计算包含分隔符在内的片段结束位置
            end = index + len(sep)

            # 截取从 start 到 end 的子串，确保分隔符被包含在内
            pieces.append(value[start:end])

            # 将下一次查找的起始位置更新为当前片段的结束位置
            start = end
            
        # 如果 pieces 为空（例如输入为空字符串），则返回包含原字符串的列表作为兜底
        return pieces or [value]

    def _split_recursive(value: str, seps: list[str]) -> list[str]:
        # 当前片段已经足够小，直接返回，不再继续降级切分。
        if len(value) <= payload_size:
            return [value]

        # 所有分隔符都用完后，才退化为按固定宽度硬切。
        if not seps:
            return [value[i:i + payload_size] for i in range(0, len(value), payload_size)]

        sep = seps[0]

        # 空字符串表示最后兜底：按字符窗口切分。
        if sep == "":
            return [value[i:i + payload_size] for i in range(0, len(value), payload_size)]

        result = []
        for piece in _split_keep_separator(value, sep):
            # 当前层切出来的片段放得下就收下，放不下才递归到更细粒度。
            if len(piece) <= payload_size:
                result.append(piece)
            else:
                result.extend(_split_recursive(piece, seps[1:]))
        return result

    def _merge_splits(splits: list[str]) -> list[str]:
        """
        将递归切分后的细小片段进行贪心合并。
        目标是在不超过 payload_size 的前提下，尽可能让每个 chunk 更完整，减少碎片化。
        """
        chunks = []
        current = ""
        for split in splits:
            # 如果当前缓存为空，直接放入第一个片段
            if not current:
                current = split
            # 如果当前缓存加上新片段的长度仍未超过预留的 payload_size，则进行合并
            elif len(current) + len(split) <= payload_size:
                current += split
            # 否则，说明当前缓存已达到合并上限，存入结果集，并开启新的缓存
            else:
                chunks.append(current)
                current = split
        
        # 处理最后一个残余的片段
        if current:
            chunks.append(current)
        return chunks

    # 先递归切到合法大小，再做一次贪心合并，减少碎片化。
    raw_chunks = _merge_splits(_split_recursive(text, separators))

    chunks = []
    current_start = 0
    for i, raw_chunk in enumerate(raw_chunks):
        # 计算当前 chunk 的实际内容：
        # overlap 不再依赖“前一个 raw_chunk 的尾巴”，而是直接根据当前片段在原文中的起点回溯。
        # 这样即使前一个 raw_chunk 很短，也能跨越多个 raw_chunk 从原文中拿满 overlap。
        # 由于 raw_chunk 的长度已在 merge 阶段限制在 payload_size (chunk_size - overlap) 之内，
        # 回溯 overlap 后的 content 总长度依然严格保证不超过原始定义的 chunk_size。
        if overlap == 0:
            # 无重叠情况：直接使用当前切分的片段内容
            content = raw_chunk
        else:
            # 有重叠情况：从原文起始位置回溯 overlap 长度，确保片段间存在上下文交叠
            start_with_overlap = max(0, current_start - overlap)
            current_end = current_start + len(raw_chunk)
            content = text[start_with_overlap:current_end]
        
        # 更新当前片段在原文中的起始位点，供后续片段计算回溯使用
        current_start += len(raw_chunk)
        
        # 将文本内容与元数据封装为字典，方便后续向量化及检索溯源
        chunks.append({
            "content": content,
            "metadata": {
                **metadata,
                "chunk_index": i,
                "chunk_total": len(raw_chunks)
            }
        })
        current_start += len(raw_chunk)

    return chunks

# ── 测试文本准备 ──────────────────────────────────────────────
sample_text = """人工智能（AI）正在深刻改变各个行业。

机器学习是人工智能的核心子领域。它通过算法让计算机从数据中自动学习规律，无需人工显式编程。常见的机器学习算法包括线性回归、决策树、随机森林和神经网络等。

深度学习是机器学习的一个分支，使用多层神经网络来处理复杂的模式识别任务。它在图像识别、语音识别和自然语言处理领域取得了突破性进展。典型的深度学习模型有卷积神经网络（CNN）、循环神经网络（RNN）和Transformer架构。

大语言模型（LLM）是基于Transformer架构的超大规模语言模型。GPT、BERT、LLaMA等都是代表性的大语言模型。它们通过在海量文本上进行预训练，获得了强大的语言理解和生成能力。在实际应用中，LLM被用于问答、摘要、翻译、代码生成等多种任务。"""


# ── 测试用例 ──────────────────────────────────────────────────
def run_tests():
    print("=" * 60)

    # ── Case 1：基础切分（验证 chunk 大小不超限）──
    print("【Case 1】基础切分，chunk_size=100, overlap=0")
    chunks = split_text(sample_text, chunk_size=100, overlap=0)
    for c in chunks:
        content = c["content"]
        status = "✅" if len(content) <= 100 else "❌ 超长!"
        print(f"  [{c['metadata']['chunk_index']+1}/{c['metadata']['chunk_total']}] "
              f"len={len(content)} {status} | {content[:30].strip()}...")
    print()

    # ── Case 2：带 overlap（验证相邻 chunk 存在重叠内容）──
    print("【Case 2】overlap 验证，chunk_size=150, overlap=30")
    chunks = split_text(sample_text, chunk_size=150, overlap=30)
    for i, c in enumerate(chunks):
        if i > 0:
            prev_tail = chunks[i - 1]["content"][-30:]
            cur_head = c["content"][:30]
            overlap_ok = prev_tail in c["content"]
            print(f"  chunk[{i}] 头部包含前一片段末尾: {'✅' if overlap_ok else '❌'}")
            print(f"    前一片段末尾: {repr(prev_tail)}")
            print(f"    当前片段头部: {repr(cur_head)}")
    print()

    # ── Case 3：短文本（不应切分）──
    print("【Case 3】短文本，不应被切分")
    short_text = "这是一段很短的文字，不需要切分。"
    chunks = split_text(short_text, chunk_size=500)
    assert len(chunks) == 1, f"❌ 期望1个chunk，实际{len(chunks)}个"
    print(f"  ✅ 仅产生 1 个 chunk，len={len(chunks[0]['content'])}")
    print()

    # ── Case 4：无分隔符的长文本（兜底强制截断）──
    print("【Case 4】纯长字符串，无自然分隔符，兜底按字符截断")
    no_sep_text = "A" * 350  # 350个字符，chunk_size=100 → 应产生4个chunk
    chunks = split_text(no_sep_text, chunk_size=100, overlap=0)
    print(f"  chunk 数量: {len(chunks)}（期望: 4）{'✅' if len(chunks) == 4 else '❌'}")
    for c in chunks:
        print(f"    len={len(c['content'])} {'✅' if len(c['content']) <= 100 else '❌'}")
    print()

    # ── Case 5：metadata 透传验证 ──
    print("【Case 5】metadata 透传验证")
    meta = {"source": "test_doc.pdf", "author": "张三"}
    chunks = split_text(sample_text, chunk_size=200, overlap=0, metadata=meta)
    first = chunks[0]["metadata"]
    assert first["source"] == "test_doc.pdf", "❌ source 未透传"
    assert first["author"] == "张三", "❌ author 未透传"
    assert "chunk_index" in first, "❌ 缺少 chunk_index"
    assert "chunk_total" in first, "❌ 缺少 chunk_total"
    print(f"  ✅ metadata 透传正常: {first}")
    print()

    # ── Case 6：小片段合并验证（核心改动）──
    print("【Case 6】小片段合并验证，多个小段落应合并为一个 chunk")
    # 每行约15字，chunk_size=100，3行合计约45字应合并为1个chunk
    small_para_text = "第一段内容很短。\n第二段内容也很短。\n第三段内容同样很短。"
    chunks = split_text(small_para_text, chunk_size=100, overlap=0)
    print(f"  chunk 数量: {len(chunks)}（期望: 1，若>1说明合并逻辑失效）"
          f"  {'✅' if len(chunks) == 1 else '❌ 合并失效!'}")
    for c in chunks:
        print(f"    len={len(c['content'])} | {c['content']}")
    print()

    print("=" * 60)
    print("✅ 全部测试完成")


run_tests()

  函数先用 _split_keep_separator() 按当前分隔符切分，但把分隔符保留在结果里，避免 chunk 边界把原文结构吃掉。接着 _split_recursive() 只在片段仍然超长时，才降级到下一级分隔符继续切；若所有分隔符都不适用，最后才按字符硬切。最后 _merge_splits() 把这些原子片段贪心合并到 payload_size = chunk_size - overlap 以内，再在外层给后续 chunk 头部补上前一块的尾部 overlap，这样既保留了重叠区，又保证最终 chunk 长度不超过 chunk_size。

运行验证

  对加载的文档执行切分，观察片段数量和长度分布是否合理：

# 读取pdf文件
doc = load_document("data/company_leave_policy.pdf")

# 分块大小为500，重叠部分为100
chunks = split_text(doc["content"], chunk_size=500, overlap=100, metadata=doc["metadata"])

lengths = [len(c["content"]) for c in chunks]
print(f"总片段数：{len(chunks)}")
print(f"长度范围：{min(lengths)} ~ {max(lengths)} 字符")
print(f"平均长度：{sum(lengths) / len(lengths):.0f} 字符")
print(f"\n片段 0 metadata：{chunks[0]['metadata']}")
print("=" * 60)
print(f"片段 0 前 100 字：{chunks[0]['content']}")
print("=" * 60)
print(f"片段 1 前 50 字（应包含片段 0 末尾内容）：{chunks[1]['content'][:50]}")

  预期：chunk_total 与 len(chunks) 相等，最大长度不超过 chunk_size，片段 1 的开头应能看到片段 0 末尾的文字——这正是 overlap 生效的证明。

🔥 踩坑预警：chunk_size 的单位是字符数，不是 token 数。中文约 1.5 个字符对应 1 个 token，所以 500 个中文字符约对应 333 个 token。如果你的 LLM 对 token 数有严格限制（如 4096 token 的上下文窗口），需要结合 3.9 节的 estimate_tokens() 函数做二次验证，避免最终拼接的 context 超出 LLM 的窗口上限。

3.5 余弦相似度原理与实现 链接到标题

  切分后的文本已经可以向量化了。但在把向量存入 FAISS 之前，我们先用 NumPy 手写一遍余弦相似度。这不只是为了"懂原理"——理解余弦相似度的数值范围和直觉含义，将直接帮你在 3.7 节设置合理的 threshold 过滤阈值。

余弦相似度的数学直觉

  余弦相似度衡量两个向量"方向"的接近程度，而不是"大小"。值域为 [-1, 1]：数值越接近 1，表示两段文本语义越相似；接近 0 表示语义无关；接近 -1 在 Embedding 空间中极少出现（表示语义完全相反）。公式是两个向量的点积除以各自模长之积。重要特性：余弦相似度只关注方向，不受向量长度影响——这就是为什么我们在 3.6 节使用 FAISS IndexFlatIP 前需要先做 L2 归一化：归一化后向量模长均为 1，此时内积在数值上等价于余弦相似度，FAISS 的内积检索结果就直接对应语义相似度排序。

  上面两张图分别展示了向量间相似度计算的直觉含义和余弦相似度的数学公式。第一张图可以看到，方向越接近的向量，余弦值越大；第二张图是公式的精确表达——分子是两个向量的点积，分母是各自模长的乘积。理解了这个公式，后面手写的 cosine_similarity 函数就不再神秘了。

import numpy as np

def cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float:
    """
    计算两个向量之间的余弦相似度（NumPy 实现）
    
    Args:
        vec_a: 向量 A 的数值列表
        vec_b: 向量 B 的数值列表
        
    Returns:
        float: 余弦相似度得分，取值范围为 [-1, 1]，越接近 1 表示越相似
    """
    # 将输入列表转换为 NumPy 数组以便进行向量化计算
    a, b = np.array(vec_a), np.array(vec_b)
    
    # 计算向量点积 (Dot Product): A · B
    dot_product = np.dot(a, b)
    
    # 计算向量的 L2 范数 (模长): ||A|| 和 ||B||
    norm_a = np.linalg.norm(a)
    norm_b = np.linalg.norm(b)
    
    # 根据公式计算余弦相似度: cos(θ) = (A · B) / (||A|| * ||B||)
    # 注意：实际应用中应考虑分母为 0 的情况
    return float(dot_product / (norm_a * norm_b))

  这 5 行代码精确对应了余弦相似度公式的每一步：np.dot(a, b) 计算点积，np.linalg.norm() 计算向量的 L2 范数（模长），两者相除得到余弦值。返回 float 而不是 np.float64 是为了确保后续 JSON 序列化时不报错。

三组对比实验

  用以下代码对比三组句子对的相似度得分，直观感受"语义相近 → 分数高"：

# 请确认模型名称为当前可用版本（查询日期：2026-02）
MODEL = "text-embedding-v3"

pairs = [
    ("如何申请年假",     "请假流程是什么"),        # 中文同义，语义相近
    ("如何申请年假",     "今天天气怎么样"),        # 语义无关
    ("Python 列表排序",  "Python list sort"),      # 中英文同义
]

for text_a, text_b in pairs:
    # 调用 Embedding 接口获取文本 A 的向量表示
    vec_a = get_embedding(text_a, client_qwen, MODEL)
    # 调用 Embedding 接口获取文本 B 的向量表示
    vec_b = get_embedding(text_b, client_qwen, MODEL)
    
    # 计算两个向量之间的余弦相似度（Score 越接近 1 表示语义越相关）
    score = cosine_similarity(vec_a, vec_b)
    
    # 打印对比结果，展示不同文本组合下的语义匹配得分
    print(f"  {text_a!r:20s} vs {text_b!r:20s}  →  {score:.4f}")

  典型的预期输出（基于 Qwen text-embedding-v3 模型）：语义相近的中文句对得分在 0.8 以上，语义无关的句对得分在 0.4 左右，中英文同义句对得分在 0.8 左右。若切换为 OpenAI text-embedding-3-small，语义相近句对的得分可能在 0.65-0.85 之间，语义无关句对可能在 0.2-0.4 之间——不同模型的绝对分数不可直接比较，但相对排序应保持一致。这组数据给了我们一个重要参考：在 3.7 节设置 threshold=0.5 时，低于此分数的检索结果将被过滤掉，而真正相关的文档片段通常得分远高于 0.5。

补充：余弦相似度 vs 欧氏距离

  除了余弦相似度，另一种常见的向量度量方式是欧氏距离（L2 Distance）——它衡量的是两个向量在空间中的"直线距离"，数值越小表示越相似。两者的核心区别在于：余弦相似度只关注向量的方向，忽略长度；欧氏距离同时受方向和长度的影响。在 Embedding 场景中，我们关心的是"语义方向是否一致"，而不是"向量绝对值大小"，因此余弦相似度是更自然的选择。

  这也解释了为什么我们在 3.6 节选择 FAISS IndexFlatIP（内积索引）+ normalize_L2（归一化）的组合：向量经过 L2 归一化后模长均为 1，此时内积在数值上等价于余弦相似度，而欧氏距离也与余弦相似度呈单调关系——三种度量在归一化后本质上是等价的。选择内积索引是因为 FAISS 对内积计算做了深度优化，检索速度最快。

余弦相似度 vs 欧氏距离对比

3.6 FAISS 向量索引构建与持久化 链接到标题

  有了向量和余弦相似度的直觉之后，我们来引入真正的向量检索引擎——FAISS。单独用 NumPy 做余弦相似度计算在文档量较多时会很慢（需要逐一与每个向量比对）；FAISS 的价值在于它能对大规模向量集合做近似最近邻检索，实现毫秒级响应。

索引类型选择：IndexFlatIP + normalize_L2

  IndexFlatIP 是内积（Inner Product）索引，存储向量后按内积大小检索。关键点在于：将向量存入前先做 L2 归一化（faiss.normalize_L2），归一化后向量的模长均为 1，此时向量的内积等价于余弦相似度。这意味着 IndexFlatIP 返回的分数实际上就是余弦相似度值，与我们 3.5 节手写的函数在数学上完全一致。以下是构建索引和持久化的完整代码：

• 两个内积的计算公式： a · b = a₁×b₁ + a₂×b₂ + a₃×b₃ + … + aₙ×bₙ

• 模长（L2 范数）的计算公式是：||v|| = √(v₁² + v₂² + v₃² + … + vₙ²)

# 此处已更新，以课件为主！
import faiss
import numpy as np

# 查询向量 A
A = np.array([1, 2, 3, 4, 5], dtype=np.float32)

# 4种关系：相同、反向、正交、部分相似
embeddings = np.array([
    [1, 2, 3, 4, 5],     # v0: 与A完全相同 -> cos = 1
    [-1, -2, -3, -4, -5],# v1: 与A方向相反 -> cos = -1
    [2, -1, 0, 0, 0],    # v2: 与A正交(点积=0) -> cos = 0
    [5, 4, 3, 2, 1],     # v3: 与A部分相似 -> cos = 0.636364
], dtype=np.float32)

dim = embeddings.shape[1]
print(f"dim={dim}, 向量数={len(embeddings)}")

# 归一化前模长
print("\n归一化前模长:")
for i, n in enumerate(np.linalg.norm(embeddings, axis=1)):
    print(f"v{i}: {n:.6f}")

# 关键：库向量和查询向量都要L2归一化
faiss.normalize_L2(embeddings)

query = A.reshape(1, -1).astype(np.float32)
faiss.normalize_L2(query)

# IndexFlatIP + 归一化 => 内积 = 余弦相似度
index = faiss.IndexFlatIP(dim)
index.add(embeddings)

k = 4
scores, ids = index.search(query, k)

print("\nTop-k (score=cosine):")
for rank, (idx, score) in enumerate(zip(ids[0], scores[0]), start=1):
    print(f"{rank}. id={idx}, score={score:.6f}")

# 归一化后，IndexFlatIP 的得分就是余弦相似度；1 表示同向，0 表示正交，-1 表示反向。

为什么“向量先归一化，再做内积”，就等价于余弦相似度。

• 第1部分：先建立直觉 在向量检索里，我们最关心的是“方向像不像”，不是“长度大不大”。 余弦相似度本质上就是看两个向量夹角：

$$ \cos(a,b)=\frac{a\cdot b}{|a||b|} $$

• 结果接近 1：方向几乎一致
• 结果接近 0：方向几乎无关（正交）
• 结果接近 -1：方向完全相反

• 第2部分：代码每一步在干什么

1. 准备多个向量（数据库里的候选向量）
2. 转成 float32（Faiss 常用类型）
3. 看归一化前模长（长度）
4. faiss.normalize_L2(vectors)：把每个向量长度变成 1
5. 建立 IndexFlatIP（IP = Inner Product，内积检索）
6. 查询向量也做同样归一化，再 search

• 第3部分：最关键的一句话 因为归一化后 $|a|=|b|=1$，所以：

$$ \cos(a,b)=\frac{a\cdot b}{1\cdot1}=a\cdot b $$

所以在 Faiss 里： L2 归一化 + IndexFlatIP = 余弦相似度检索

• 第4部分：纠正常见误区（重点） 很多人会误以为 [5,4,3,2,1] 和 [1,2,3,4,5] 是“方向相反”，其实不是。 它们的余弦是 0.636364，说明“有一定相似”，不是反向。 真正反向是 [-1,-2,-3,-4,-5]，这时余弦才是 -1。 真正正交要满足点积为 0，余弦才是 0。

• 第5部分：如何解读检索结果 当查询向量是 A 时：

• 得分 ≈ 1：几乎同方向（最相似）
• 得分 ≈ 0.6：有一定相关，但不是很像
• 得分 ≈ 0：基本无关
• 得分 < 0：方向相反，不应认为语义接近

• 收尾总结 今天记住三件事就够了：

1. 余弦相似度看“方向”
2. 归一化是为了去掉“长度影响”
3. Faiss 中 normalize_L2 + IndexFlatIP 就是在算余弦相似度

import faiss
import numpy as np
import json


def build_faiss_index(embeddings: list[list[float]]) -> faiss.IndexFlatIP:
    """
    构建 FAISS 内积索引。
    通过对向量进行 L2 归一化，使得内积计算等价于余弦相似度。

    Args:
        embeddings: 嵌入向量列表，每个元素为一个浮点数列表。

    Returns:
        faiss.IndexFlatIP: 构建好的 FAISS 索引对象。
    """
    # 获取向量维度（特征长度）1024
    dim = len(embeddings[0])

    # 将输入列表转换为 float32 类型的 numpy 数组，这是 FAISS 指定的数值类型
    vectors = np.array(embeddings, dtype=np.float32)

    # 执行原地 L2 归一化：||v|| = 1。归一化后的向量点积即为余弦相似度
    faiss.normalize_L2(vectors)

    # 创建暴力搜索的内积索引 (IndexFlatIP)
    index = faiss.IndexFlatIP(dim)

    # 将处理后的向量添加到索引库中
    index.add(vectors)

    return index


def save_index(index: faiss.IndexFlatIP, chunks: list[dict],
               index_path: str, metadata_path: str) -> None:
    """
    持久化索引与元数据双文件方案。
    index_path 存储 FAISS 向量索引数据，metadata_path 存储对应的文本块及原始信息。

    Args:
        index: FAISS 索引对象。
        chunks: 包含 "content" 和 "metadata" 的原始文本块列表。
        index_path: 索引文件的保存路径（二进制格式）。
        metadata_path: 元数据文件的保存路径（JSON 格式）。
    """
    # 将向量索引二进制流写入磁盘
    faiss.write_index(index, index_path)

    # 将文本块及元数据序列化为 JSON，确保非 ASCII 字符（如中文）正常显示
    with open(metadata_path, 'w', encoding='utf-8') as f:
        json.dump(chunks, f, ensure_ascii=False, indent=2)

    print(f"✅ 索引保存成功：{index_path}（包含 {index.ntotal} 条向量）")
    print(f"✅ 元数据保存成功：{metadata_path}")


def load_index(index_path: str, metadata_path: str) -> tuple[faiss.IndexFlatIP, list[dict]]:
    """
    从磁盘恢复索引和关联的文本块数据。

    Args:
        index_path: 索引二进制文件路径。
        metadata_path: 元数据 JSON 文件路径。

    Returns:
        tuple: (index, chunks) 包含加载后的 FAISS 索引和对应的文本块列表。
    """
    # 读取二进制索引文件
    index = faiss.read_index(index_path)

    # 读取 JSON 格式的文本元数据
    with open(metadata_path, 'r', encoding='utf-8') as f:
        chunks = json.load(f)

    return index, chunks

  持久化采用"双文件"策略：FAISS 索引文件只存储向量矩阵，用二进制格式高效保存；chunks 列表存为 JSON 文件，保存每个片段的原始文本和 metadata。两个文件通过相同的下标对应——索引中第 i 个向量对应 chunks[i] 中的文本。这种设计使每个部分都可以独立更新：向量维度变了（换了 Embedding 模型），重建索引文件即可，不需要重新整理 chunks；文档内容更新了，只需重新切分和向量化，重建两个文件。

⚠️ 常见误区：faiss.normalize_L2(vectors) 是原地操作，会直接修改传入的 numpy 数组。如果你后续还需要使用原始未归一化的向量（例如做其他距离计算），请先用 vectors.copy() 保存副本再调用归一化。

⚠️ 常见误区：FAISS 还提供了 IndexFlatL2（欧氏距离索引），初学者容易与 IndexFlatIP 混淆。由于我们已经做了 L2 归一化，内积和欧氏距离在数学上等价（见 3.5 节对比表），而 IndexFlatIP 的返回值直接就是余弦相似度分数（0-1 之间，越大越相似），比 IndexFlatL2 的距离值（越小越相似）更直观，因此选择 IP 索引。

构建 → 保存 → 加载验证

  以下代码演示完整的构建和持久化流程，最后通过断言验证数据一致性：

# 对所有 chunks 批量向量化（一次 API 调用，比循环逐个调用快得多）
texts = [c["content"] for c in chunks]
embeddings = get_embeddings_batch(texts, client_qwen, MODEL)

# 构建索引并持久化
index = build_faiss_index(embeddings)
save_index(index, chunks, "faiss_data/rag.index", "faiss_data/rag_chunks.json")

# 重新加载，验证一致性
index2, chunks2 = load_index("faiss_data/rag.index", "faiss_data/rag_chunks.json")
assert index2.ntotal == len(chunks2), "❌ 索引和 chunks 数量不匹配！"
print(f"✅ 索引持久化验证通过：{index2.ntotal} 条向量 = {len(chunks2)} 个 chunks")

  运行后应看到两条保存成功的日志和最后的验证通过提示。assert 断言是一个好习惯——索引向量数和 chunks 列表长度必须严格相等，否则后续 search() 函数用 idx 索引 chunks 时会出现越界或数据错位的问题。

🎉 阶段性成果：到这里，离线索引阶段全部完成！我们已经走通了 原始文档 → 文档加载 → 文本切分 → 批量向量化 → FAISS 索引构建 → 双文件持久化 的完整链路。索引文件（.index）和元数据文件（.json）已经保存到磁盘，后续可以随时加载使用，无需重复构建。

🔍 在线查询阶段：实时问答 链接到标题

  知识库已经建好并持久化到磁盘，接下来我们进入 RAG 系统的在线查询阶段——也就是第二章中介绍的 Query Pipeline。这个阶段的代码在每次用户提问时都会执行：加载索引 → 将用户问题向量化 → 在 FAISS 中检索 Top-K → 构建增强 Prompt → 调用 LLM 生成回答。

  与离线阶段不同，在线阶段的核心关注点是响应速度和回答质量。接下来的四节（3.7-3.10）将依次实现检索、Prompt 设计、Token 预算控制和完整 Pipeline 串联。跑完这四节，你的 RAG 系统就能真正"回答问题"了。

3.7 Top-K 检索与结果后处理 链接到标题

  索引构建完成，我们来实现 RAG 系统的检索核心。搜索的过程分两步：先用 FAISS 拿到 Top-K 候选，再对候选结果进行相似度过滤和排序，得到最终可用的检索结果。

⚠️ 结构约定：chunks 参数中每项必须是完整的 chunk 对象（{"content": ..., "metadata": {...}}），即 3.6 节 save_index() 传入的 split_text() 返回值。若只传了 metadata 部分，chunks[idx]["content"] 会触发 KeyError。

  以下是完整的检索函数实现，覆盖了向量化、FAISS 检索、过滤和排序四个环节：

def search(query: str, client, model: str, index: faiss.IndexFlatIP,
           chunks: list[dict], top_k: int = 5,
           threshold: float = 0.5) -> list[dict]:
    """
    RAG 检索函数：向量化查询 → FAISS 检索 → 相似度过滤 → 排序
    参数：
      threshold: 余弦相似度阈值（0~1），低于此分数的结果被丢弃
    返回：
      按相似度降序排列的 chunk 列表，每项含 content / metadata / score
    """
    # Step 1: 查询向量化 + 归一化（必须与索引构建时用同一模型和同样的归一化）
    query_vec = np.array([get_embedding(query, client, model)], dtype=np.float32)
    faiss.normalize_L2(query_vec)

    # Step 2: FAISS 检索 Top-K
    # 注意：当 top_k > index.ntotal 时，多余的返回位置填充 -1（需过滤）
    scores, indices = index.search(query_vec, top_k)

    # Step 3: 过滤 + 组装输出
    results = []
    for score, idx in zip(scores[0], indices[0]):
        if idx == -1:            # FAISS 填充的无效位置（top_k > 实际向量数时出现）
            continue
        if score < threshold:    # 相似度低于阈值，丢弃
            continue
        results.append({
            "content":  chunks[idx]["content"],
            "metadata": chunks[idx]["metadata"],
            "score":    float(score)
        })

    # Step 4: 按相似度降序排序（FAISS 已返回排序结果，此处为显式保证）
    results.sort(key=lambda x: x["score"], reverse=True)
    return results

  函数的关键细节有三点：第一，查询向量必须用与构建索引完全相同的 Embedding 模型和 normalize_L2 归一化——向量空间不一致时检索结果会完全错误，且系统不会报错；第二，if idx == -1 的过滤处理了 top_k > index.ntotal 的边界情况（比如知识库只有 3 条向量但请求了 5 个结果）；第三，返回的 score 字段转为 Python 原生 float，确保后续 JSON 序列化时不报类型错误。

🔥 踩坑预警：查询向量和索引向量必须使用同一个 Embedding 模型。如果你用 text-embedding-v3 构建索引，查询时却用了 text-embedding-3-small，两个模型的向量空间完全不同（维度都不一样），检索结果会毫无意义——但系统不会报错，只会返回看似正常实则随机的结果。这是 RAG 系统中最隐蔽的 bug 之一。

运行验证

  用一个自然语言问题执行检索，打印 Top-3 结果，验证检索结果的相关性：

query = "年假可以申请几天？"

results = search(
    query,          # 查询文本
    client_qwen,    # 嵌入模型客户端
    MODEL,          # 嵌入模型名称
    index,          # 向量索引
    chunks,         # 原始文本块
    top_k=5,        # 返回最相关的结果数量
    threshold=0.5   # 相似度阈值
)

print(f"检索到 {len(results)} 条相关结果")
for i, r in enumerate(results[:3]):
    source = r['metadata'].get('source', '未知')
    print(f"\n── Top {i+1}（得分 {r['score']:.4f}，来源：{source}）")
    print(r["content"][:150] + "...")

  预期看到：得分最高的结果与"年假"直接相关，得分通常在 0.7 以上；与查询语义无关的片段要么被 threshold=0.5 过滤掉，要么排在靠后位置。如果检索结果全部不相关，优先检查 Embedding 模型是否与建索引时一致。

🔥 踩坑预警：threshold=0.5 是比较宽松的阈值，适合教学演示。在生产环境中，如果阈值设太低，无关片段会进入 context 并干扰 LLM 的回答；如果设太高，真正相关的片段可能被过滤掉。建议用 20-30 条带标注的测试问题，通过实验在精确率和召回率之间找到平衡点，通常中文场景的合理阈值在 0.6~0.75 之间。

3.8 RAG Prompt 模版设计 链接到标题

  检索到相关片段后，如何把它们"喂给"大模型？这一步的核心是 Prompt 设计。一个好的 RAG Prompt 要做到三件事：明确告知模型参考资料在哪、引导模型"引用"而非"创作"、处理资料不足时的兜底情况。这三点决定了 RAG 系统的幻觉控制效果。

System Prompt 设计原则

  System Prompt 负责设定模型的"角色"和"行为规则"。在 RAG 场景中，最关键的规则只有三条：只用参考资料中的内容回答；如果资料不够，明确说无法回答；回答时标注来源。第三条"标注来源"看起来可选，但在需要可信度的场景（如医疗、法律、企业内部知识库）中至关重要——用户需要能验证答案。

SYSTEM_PROMPT = """你是一个专业的知识库问答助手。请严格基于【参考资料】中的内容回答用户问题。

规则：
1. 只使用参考资料中的信息回答，不要编造内容
2. 如果参考资料中没有相关信息，请明确说"根据现有资料，我无法回答这个问题"
3. 回答时尽量引用原文关键语句，并在括号中标注来源文件名"""


def build_rag_prompt(query: str, retrieved_chunks: list[dict]) -> list[dict]:
    """
    构建 RAG 增强后的消息列表（供 chat.completions.create 使用）
    返回标准的 OpenAI messages 格式：[{"role": ..., "content": ...}, ...]
    """
    if not retrieved_chunks:
        # 兜底：没有检索结果时，直接告知无相关资料
        context = "（当前知识库中未找到与该问题相关的资料）"
    else:
        # 提取并格式化检索到的知识片段
        context_parts = []
        
        for i, chunk in enumerate(retrieved_chunks):
            # 获取来源元数据，若不存在则显示“未知来源”
            source = chunk["metadata"].get("source", "未知来源")

            # 从文件路径中提取文件名（处理跨平台路径分隔符）
            source_name = source.split("/")[-1] if "/" in source else source

            # 组合编号、来源和具体内容
            context_parts.append(f"【资料{i+1}】（来源：{source_name}）\n{chunk['content']}")
        
        # 将多个资料片段用换行符连接成完整的上下文字符串
        context = "\n\n".join(context_parts)

    user_content = f"""
    【参考资料】
    {context}

    【用户问题】
    {query}"""

    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user",   "content": user_content}
    ]

  build_rag_prompt 函数将检索到的每个 chunk 格式化为 【资料N】（来源：文件名） 的形式，并用空行分隔，方便 LLM 区分不同来源的内容。空列表时的兜底设计很重要——当检索结果为空时，仍然构建一个合法的 messages 对象，让 LLM 告知用户"无相关资料"，而不是让程序崩溃。

打印完整 Prompt 结构

  运行以下代码打印 Prompt 结构，直观看到 LLM 实际"看到"的内容：

query = "年假可以申请几天？"

messages = build_rag_prompt(query, results)

print("── System Prompt ──────────────────────────")
print(messages[0]["content"])
print("\n── User Message（前 400 字）──────────────")
print(messages[1]["content"])

  预期看到：System Prompt 包含清晰的行为规则，User Message 包含格式化的参考资料和用户问题。这段内容就是最终发送给 LLM 的完整输入——理解这个结构，是理解 RAG 系统工作机制的关键。

3.9 Token 预算控制与上下文拼接 链接到标题

  从检索到 Prompt 构建之间，还有一个容易被忽视的环节：Token 预算控制。检索到的片段可能有多个，如果全部放进 Prompt，累积的 token 数量可能超出 LLM 的上下文窗口限制，导致 API 报错；即使不超出限制，塞入过多内容也会增加 LLM 的"注意力分散"问题，影响回答质量。

简易 Token 估算

  精确的 Token 计数需要 tiktoken 等专业工具，但对于 RAG 的预算控制场景，粗略估算已经足够：中文约 1.5 字/Token，英文约 4 字符/Token。这个比例基于 GPT 系列 cl100k_base tokenizer 的经验值，Qwen 系列 tokenizer 的中文 token 化效率略有不同（约 1.2-1.8 字/Token），此处取中间值 1.5 作为通用近似。精确计数请使用 tiktoken（OpenAI）或对应模型的 tokenizer。以下是估算函数和预算筛选函数的实现：

def estimate_tokens(text: str) -> int:
    """
    简易 Token 估算函数（无需依赖外部库如 tiktoken）

    估算规则：
    1. 中文字符：通常占比较多 token，按 1.5 个字符对应 1 个 token 估算
    2. 其他字符（英文、数字、标点）：按 4 个字符对应 1 个 token 估算

    Args:
        text (str): 需要估算的文本字符串

    Returns:
        int: 估算出的 token 总数
    """
    # 统计中文字符数量（利用 Unicode 范围：\u4e00-\u9fff）
    chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')

    # 计算非中文字符数量
    other_chars = len(text) - chinese_chars

    # 根据加权比例计算总 token 数并取整
    return int(chinese_chars / 1.5 + other_chars / 4)


def select_chunks_within_budget(chunks: list[dict],
                                max_tokens: int = 3000) -> list[dict]:
    """
    在指定的 Token 预算内，采用贪心策略从检索结果中筛选文本片段 (Chunks)

    策略说明：
    1. 假设输入 chunks 已按相关性（如相似度得分）降序排列
    2. 遍历 chunks 并累加 token 数，一旦超过 max_tokens 预算即停止
    3. 确保在上下文长度限制内，优先保留最相关的片段

    Args:
        chunks (list[dict]): 包含文本片段的字典列表，每个字典需含有 "content" 键
        max_tokens (int): 允许的最大 token 预算，默认为 3000

    Returns:
        list[dict]: 筛选后符合预算要求的文本片段列表
    """
    selected = []
    total_tokens = 0

    for chunk in chunks:
        # 估算当前片段的 token 消耗
        chunk_tokens = estimate_tokens(chunk["content"])

        # 检查加入当前片段后是否会超出总预算
        if total_tokens + chunk_tokens > max_tokens:
            # 若超出则停止筛选，确保不突破上下文窗口
            break

        # 将片段加入已选列表并更新总 token 计数
        selected.append(chunk)
        total_tokens += chunk_tokens

    return selected

  select_chunks_within_budget 使用贪心策略：因为输入的 chunks 已按相似度降序排列（由 3.7 节 search() 保证），函数按顺序逐个累加，一旦超出 max_tokens 就停止。这确保了在预算有限时，优先纳入相关性最高的片段。max_tokens=3000 是一个保守的默认值，为 System Prompt 和 LLM 回答预留了足够的 token 空间。

验证 Token 截断效果

  以下代码演示当知识库内容较多时，Token 预算控制如何截断结果：

# 用较低的 threshold 拉取更多候选结果（模拟内容丰富的知识库）
results_all = search(query, client_qwen, MODEL, index, chunks, top_k=5, threshold=0.3)
selected = select_chunks_within_budget(results_all, max_tokens=3000)

print(f"检索到 {len(results_all)} 条，Token 预算内选中 {len(selected)} 条")
total = sum(estimate_tokens(c['content']) for c in selected)
print(f"选中片段合计约 {total} tokens")

  预期看到：检索到的片段数 ≥ 选中的片段数，合计 token 数在 3000 以内。如果所有片段的 token 总量本身就低于 3000，那么 selected 与 results_all 完全相同，说明在当前知识库规模下无需截断——这也是正常的运行结果。

3.10 完整 RAG Pipeline 串联与演示 链接到标题

  九块拼图，现在全部到位。这一节我们把前 9 节的所有模块串联为一个端到端的 RAG 问答系统，并通过三个测试问题演示系统的行为——包括"能答"和"不能答"两种情况。

一函数搞定从提问到回答

  rag_query 函数是整个系统的顶层入口，它按顺序调用 search()、select_chunks_within_budget()、build_rag_prompt() 和 chat.completions.create()，把五步操作封装为一次函数调用：

def rag_query(query: str,
              client,
              embed_model: str,
              chat_model: str,
              index: faiss.IndexFlatIP,
              chunks: list[dict],
              top_k: int = 5,
              max_context_tokens: int = 3000) -> dict:
    """
    完整 RAG Pipeline：提问 → 检索 → Token 预算控制 → Prompt → LLM 生成
    返回：{
      "answer":          str,       # LLM 生成的回答
      "sources":         list[dict], # 实际使用的 chunk 列表
      "total_retrieved": int         # 检索到的候选总数
    }
    """
    # Step 1: 检索相关片段
    results = search(query, client, embed_model, index, chunks, top_k=top_k)

    # Step 2: Token 预算控制
    selected = select_chunks_within_budget(results, max_tokens=max_context_tokens)

    # Step 3: 构建增强 Prompt
    messages = build_rag_prompt(query, selected)

    # Step 4: 调用 LLM 生成回答
    response = client.chat.completions.create(
        model=chat_model,
        messages=messages
    )
    answer = response.choices[0].message.content

    return {
        "answer":          answer,
        "sources":         selected,
        "total_retrieved": len(results)
    }

  函数的返回值包含三部分：answer 是 LLM 的回答文本，sources 是实际使用的 chunk 列表（而非全部检索结果），total_retrieved 记录了经过 threshold 过滤后的候选总数。三个字段配合使用，让调用方既能展示答案，也能展示引用来源，还能了解系统实际检索到了多少相关内容。

端到端演示

  以三个测试问题演示系统行为，覆盖知识库内检索、语义检索（措辞不同但语义相同）和超出知识库范围的兜底逻辑：

# 请确认模型名称为当前可用版本（查询日期：2026-02）
EMBED_MODEL = "text-embedding-v3"   # Qwen Embedding，仍有效；可选升级 text-embedding-v4
CHAT_MODEL  = "qwen3.5-plus"           # 或 "gpt-5-mini"（GPT-5 家族的轻量变体，更低成本），支持 OpenAI 协议的模型均可

test_queries = [
    "年假可以申请几天？",                        # 知识库内有答案（PDF）
    "年假的申请流程是什么？",                     # 换一种措辞提问，测试语义检索（PDF）
    "量子纠缠是什么？",                          # 故意超出知识库范围，测试兜底逻辑
]

for q in test_queries:
    print(f"\n{'='*60}")
    print(f"❓ 问题：{q}")
    result = rag_query(q, client_qwen, EMBED_MODEL, CHAT_MODEL, index, chunks)

    print(f"✅ 回答：\n{result['answer']}")
    print(f"\n📚 引用来源（检索到 {result['total_retrieved']} 条，实际使用 {len(result['sources'])} 条）：")
    for src in result["sources"]:
        print(f"   - {src['metadata'].get('source')} (score={src['score']:.4f})")

  三个问题各有侧重，预期行为如下：第一个"年假可以申请几天"直接命中 PDF 中的假期制度，LLM 会从中提取年假天数并标注来源；第二个"年假的申请流程是什么"换了一种措辞提问（原文是"年假申请流程"章节），这是对语义检索能力的直接考验——即使措辞不同，Embedding 模型也应能检索到正确的内容；第三个"量子纠缠"故意超出知识库范围，检索结果为空或 score 低于 threshold 被过滤，LLM 应按 System Prompt 的指导回答"根据现有资料，我无法回答这个问题"。

完整数据流回顾

Naive RAG → Advanced RAG 主要优化方向

🎉 学完本章，你已经掌握了：

• ✅ Embedding 的本质与主流模型对比，能用 openai SDK 统一对接 OpenAI 和 Qwen 两个平台
• ✅ 手写 Document Loader，支持 PDF / TXT / Markdown 三种格式的统一加载
• ✅ 手写 Text Splitter，实现递归分隔符切分 + overlap，理解 chunk_size 对检索质量的影响
• ✅ 用 NumPy 手写余弦相似度，建立"向量方向 = 语义相似"的直觉，了解 threshold 设置的依据
• ✅ 用 FAISS 构建 IndexFlatIP 内积索引，掌握 normalize_L2 + 双文件持久化方案
• ✅ 实现 Top-K 检索 + 相似度过滤 + 排序的完整后处理逻辑，理解 -1 padding 的边界处理
• ✅ 设计 RAG Prompt 模版，引导模型"引用而非创作"，实现空结果兜底逻辑
• ✅ 实现简易 Token 预算控制，确保 context 始终在 LLM 窗口范围内
• ✅ 将所有模块串联为端到端 rag_query() 函数，完整覆盖"提问 → 检索 → 生成"链路

⚠️ 常见误区：别把两类模型搞混了：回顾整个 Pipeline，你会发现我们用到了两类完全不同的模型——Embedding 模型（如 text-embedding-v3）负责将文本转换为向量，用于索引构建和检索；Chat 模型（如 qwen3.5-plus）负责最终的回答生成。这两类模型职责独立、互不替代。一个关键点是：知识库的构建（文档加载、切分、向量化、FAISS 索引）全程不涉及 LLM，LLM 只在 rag_query() 的最后一步才登场。初学者经常误以为"RAG 系统需要用大模型来建索引"，实际上索引阶段只需要轻量的 Embedding 模型，成本远低于 LLM 推理。

  在下一章的实战案例中，我们将把本章的所有模块"迁移"到一个真实的应用场景——从零搭建代码库检索助手。你会发现，手写版的每一个函数都有直接的用武之地，只需要针对代码场景做一些定制：代码文件的扫描策略、按文件/函数级别的切分方式、以及记录文件名和行号的 metadata 设计。

综合实战：从手写到工程化 链接到标题

  第三章我们用 NumPy、FAISS 和 openai SDK 手写了 RAG 系统的全部核心模块，从文档加载到 Pipeline 串联，每一步都亲手实现。但手写版毕竟是"教学原型"——没有界面、没有持久化会话、没有流式输出，离"能给别人用的产品"还有距离。

  接下来的第四章，我们不会重新写一遍 RAG，而是直接展示一个基于第三章全部知识构建的完整工程化项目——RAG 代码库智能问答系统。你会看到，第三章手写的每一个函数，在这个项目中都有对应的"工程化升级版"。更重要的是，你可以在本地直接运行这个项目，亲手体验从"添加代码仓库 → 自动索引 → 智能问答"的完整流程。

第四章：综合实战——RAG 代码库智能问答系统 链接到标题

  本章的目标很明确：把第三章的手写知识"收拢"到一个可运行、可交互、可观察的真实项目中。我们不会逐行讲解项目代码，而是聚焦三件事——第一，让你在本地把项目跑起来；第二，帮你建立"手写模块 → 工程实现"的完整映射；第三，通过项目独有的 RAG 过程可视化功能，让你"看见"检索和生成的每一步。

  这个项目叫做 RAG Code Search，是一个前后端分离的全栈应用。它的核心能力是：给定一个 Python 代码仓库，自动完成代码切分、向量化、索引构建，然后通过自然语言对话的方式回答关于代码的问题。整个系统的底层逻辑，就是第三章我们手写的那条 文档加载 → 文本切分 → Embedding → 向量索引 → 检索 → Prompt → LLM 生成 链路。

4.1 项目概览与技术栈 链接到标题

  RAG Code Search 是一个面向 Python 代码仓库的智能问答系统，采用前后端分离架构。后端负责代码索引、向量检索和 LLM 对话，前端提供可视化的交互界面。项目的技术选型如下：

RAG Code Search 技术栈一览

  从技术栈可以看出，后端的核心依赖（openai SDK、向量数据库、FastAPI）与第三章的手写版一脉相承，只是从"脚本级"升级到了"服务级"。前端则提供了第三章完全没有的可视化能力——你可以直观地看到检索结果、Prompt 构建过程和 Token 消耗。

4.2 本地运行：三步跑通项目 链接到标题

  本节是整个第四章的起点——先跑起来，再理解原理。 项目提供了一键启动脚本 scripts/start-macos-linux.sh（macOS/Linux）和 scripts/start-windows.bat（Windows）， 脚本会自动完成虚拟环境创建、后端依赖安装、前端 npm install 和双服务启动， 你只需要关注三件事：确认项目目录完整、配置 API 密钥、运行启动脚本。环境要求为 Python 3.10+ 和 Node.js 18+。

📅 时效性说明：本节内容基于 FastAPI 0.128.0、ChromaDB 1.5.1、React 18.3.1、Vite 5.4.2 环境测试。若使用其他版本，部分命令可能需要调整。

步骤一：进入项目目录

  项目位于课程仓库的 rag-code-search/ 目录下，首先确认目录结构完整：

# 进入项目目录并查看结构
!cd rag-code-search/ && ls
# 预期看到：backend/  frontend/  scripts/  README.md

  如果能看到 backend/、frontend/、scripts/ 三个目录，说明项目文件完整。

步骤二：配置 API 密钥

  项目需要 LLM 和 Embedding 两个 API 的密钥。你可以先启动项目，然后在前端的"系统配置"页面中填写，也可以提前了解默认配置：

API 配置项说明

⚠️ 常见误区：LLM 和 Embedding 可以使用不同平台的 API。例如 Embedding 用通义千问（便宜），LLM 用 OpenAI（质量高），只要分别填写正确的 Key 和 Base URL 即可。这与第三章 3.2 节中我们分别配置两个 OpenAI 客户端的做法完全一致。

步骤三：一键启动前后端服务

  项目提供了一键启动脚本，自动完成虚拟环境创建、依赖安装和前后端服务启动。 在终端中进入项目根目录后执行即可：

# ⚠️ 以下命令为长驻服务，会阻塞 Jupyter Kernel，请在独立终端中执行

# macOS / Linux（首次运行需添加执行权限）
# chmod +x scripts/start-macos-linux.sh
# ./scripts/start-macos-linux.sh

# Windows
# scripts\start-windows.bat

  脚本会依次完成四个步骤：检查 Python/Node.js 环境 → 创建虚拟环境并安装后端依赖 → 安装前端依赖 → 启动前后端服务。 全部完成后，终端会显示服务地址。在浏览器中打开 http://localhost:5173，看到项目界面即表示启动成功。 按 Ctrl+C 可同时停止前后端服务。

🔥 踩坑预警：如果后端报 Address already in use，说明 8000 端口被占用， 可以改用手动启动方式并指定 --port 8001；如果前端报端口冲突，Vite 会自动切换到 5174 端口。 如需开发调试或单独控制前后端，也可以参考项目 README.md 中的"手动分步启动"章节。

4.3 系统架构与数据流 链接到标题

  项目跑起来之后，我们来看看它的内部结构。如果你还记得第三章开头介绍的两条数据流——离线索引流和在线查询流——那么理解这个项目的架构会非常轻松，因为它的后端服务层就是按这两条流来组织的。

4.3.1 后端服务分层 链接到标题

  后端采用经典的"路由层 → 服务层 → 存储层"三层架构，每个服务文件对应一个明确的职责：

后端服务层职责划分

  可以看到，五个服务文件几乎与第三章的十节内容一一对应。这不是巧合——好的工程架构本身就应该反映业务逻辑的自然分层。

4.3.2 两条数据流的工程化实现 链接到标题

  第三章我们用"离线索引"和"在线查询"两个阶段来组织教学，这个项目的 API 路由也严格沿着这两条流设计：

离线索引流（对应 3.3-3.6）：

  添加仓库 → 扫描 .py 文件 → 按行边界切分 → 批量 Embedding → ChromaDB 持久化

  对应 API：POST /api/repos → POST /api/repos/{id}/index → 轮询 GET /api/repos/{id}/index/status

在线查询流（对应 3.7-3.10）：

  用户提问 → 问题向量化 → ChromaDB Top-K 检索 → 构建四段式 Prompt → LLM 流式生成

  对应 API：POST /api/chat（SSE 流式响应，依次返回 retrieval → prompt → chunk → done 四类事件）

4.4 知识映射：从手写到工程化 链接到标题

  这是本章的核心环节。第三章我们亲手实现了 RAG 的每一个模块，现在来看看这些"手写知识"在真实项目中是如何落地的。下面这张映射表覆盖了全部关键模块，右侧的"工程化升级点"正是从教学原型到生产系统的核心差异：

第三章手写模块 → 工程化实现完整映射

  从这张表可以清晰地看到：项目没有引入任何第三章未涉及的"新概念"，所有升级都是工程层面的。接下来我们挑几个最值得关注的升级点展开说明。

4.4.1 FAISS → ChromaDB：为什么换向量数据库？ 链接到标题

  第三章我们用 FAISS 构建索引，需要手动管理两个文件——.index（向量）和 .json（元数据）。这在教学场景下没问题，但在工程场景下有明显短板：新增文档需要重建整个索引、元数据与向量分离容易出现不一致、不支持按条件过滤检索。

  ChromaDB 解决了这些问题。它是一个专为 AI 应用设计的向量数据库，向量和元数据存储在一起，支持增量写入、条件过滤和自动持久化。在本项目中，每个代码仓库对应一个 ChromaDB Collection，索引构建完成后数据自动保存到 backend/data/chroma/ 目录，重启服务后无需重新索引。

⚠️ 常见误区：FAISS 和 ChromaDB 不是"谁好谁差"的关系。FAISS 在超大规模数据（百万级以上）和极致性能场景下仍然是首选；ChromaDB 的优势在于开发体验和元数据管理。选型取决于场景——本项目的代码仓库通常在几千到几万个 chunk 的量级，ChromaDB 完全胜任。

4.4.2 同步调用 → SSE 流式输出 链接到标题

  第三章的 rag_query() 函数是同步的——调用后等待 LLM 返回完整回答，期间用户只能干等。在工程项目中，这种体验是不可接受的。本项目使用 SSE（Server-Sent Events） 实现流式输出，LLM 每生成一个 token 就立即推送到前端，用户可以实时看到回答"逐字出现"。

  更巧妙的是，SSE 流不仅传输 LLM 的回答文本，还依次传输四类事件：retrieval（检索结果）→ prompt（完整 Prompt + Token 估算）→ chunk（LLM 回答的每个 token）→ done（完成信号）。这意味着前端在 LLM 开始生成之前，就已经拿到了检索结果和 Prompt 信息，可以同步展示在侧边栏中。

4.4.3 单次调用 → 会话管理 链接到标题

  第三章的 Pipeline 是"无状态"的——每次调用 rag_query() 都是独立的，不记得之前问过什么。本项目引入了会话（Session）管理：每个会话绑定一个代码仓库，对话历史自动保存到 backend/data/sessions.json，构建 Prompt 时会将历史消息作为 History 段注入，实现多轮对话的上下文连贯。

4.5 功能演示与操作指南 链接到标题

  项目启动后，浏览器打开 http://localhost:5173，左侧导航栏提供五个页面。接下来我们按照"先建索引、再搜索、再对话"的使用顺序，逐一介绍每个页面的功能和操作方式。

4.5.1 仓库管理（RepoManager） 链接到标题

  这是使用项目的第一步——告诉系统"你要检索哪个代码仓库"。页面提供两种添加方式：

• 本地路径：输入代码仓库的绝对路径（如 /Users/xxx/my-project），系统会递归扫描其中的 .py 文件

• ZIP 上传：拖拽或点击上传 .zip 压缩包，系统自动解压并识别根目录

  添加仓库后，点击"开始索引"按钮，系统会在后台执行完整的离线索引流程：扫描文件 → 代码切分 → 批量 Embedding → ChromaDB 写入。索引过程中状态会从 pending → indexing → indexed 变化，前端每 1.5 秒自动轮询状态。索引完成后，仓库卡片上会显示文件数和 chunk 数。

🔥 踩坑预警：索引过程需要调用 Embedding API，如果你还没有在"系统配置"页面填写 API Key，索引会失败并显示 error 状态。建议先配置好 API 再添加仓库。

4.5.2 代码块浏览器（ChunksExplorer） 链接到标题

  索引完成后，可以在这个页面查看系统是如何切分代码的。左侧面板列出所有 chunk，支持按文件路径和内容关键词过滤；右侧面板展示选中 chunk 在原始文件中的上下文位置，高亮显示 chunk 区域和 overlap 区域。

  这个页面的教学价值在于：你可以直观地验证第三章 3.4 节学到的切分逻辑——chunk 是否在行边界对齐？overlap 区域是否覆盖了上一个 chunk 的尾部？页面顶部的统计栏还会显示 chunk 总数、文件数、Embedding 维度、chunk_size 和 overlap 参数。

4.5.3 向量搜索实验场（SearchPlayground） 链接到标题

  这个页面让你可以单独测试向量检索的效果，而不需要经过 LLM 生成环节。输入一个自然语言查询（如"用户认证是怎么实现的"），选择目标仓库和 Top-K 数量，点击搜索后会看到：

• 查询向量预览：你的问题被 Embedding 后的前 10 个维度数值

• 检索结果卡片：每个结果包含文件路径、行号范围、相似度分数（0-1）、距离值，以及可展开的代码片段

• Prompt 预览：系统会同时构建完整的四段式 Prompt（System / Context / History / User），并估算 Token 消耗

  这个页面对应的正是第三章 3.7 节的 Top-K 检索逻辑，但增加了可视化的分数展示和 Prompt 预览能力。你可以反复调整查询措辞，观察相似度分数的变化，建立对语义检索能力边界的直觉。

4.5.4 智能问答对话（ChatPage） 链接到标题

  这是项目的核心页面，也是第三章 rag_query() 函数的完整工程化体现。页面采用三栏布局：

• 左栏（280px）：仓库选择器 + 会话列表，支持新建、切换和删除会话

• 中栏：对话消息流，用户消息和 AI 回答交替展示，AI 回答支持 Markdown 渲染和代码语法高亮

• 右栏（340px）：RAG 过程面板，实时展示检索结果和 Prompt 构建详情

  使用流程：选择一个已索引的仓库 → 新建会话 → 输入问题（如"这个项目的 API 路由是怎么组织的？"）→ 回车发送。你会看到右侧面板先显示检索到的代码片段和相似度分数，然后中栏开始逐字显示 LLM 的回答。这个过程完整复现了第三章的 检索 → Prompt 构建 → LLM 生成 链路，只是从终端 print 升级为了可视化界面。

4.5.5 系统配置（SettingsPage） 链接到标题

  配置页面集中管理所有可调参数，分为四个区域：

• LLM 配置：API Key（密码输入框）、Base URL、模型名称

• Embedding 配置：API Key、Base URL、模型名称（可与 LLM 使用不同平台）

• 检索参数：chunk_size（默认 1000）、chunk_overlap（默认 200）、top_k（默认 5）

• 系统提示词：可自定义 System Prompt，附带"恢复默认"按钮

  这些参数与第三章中我们手动传入函数的参数完全对应——chunk_size 对应 3.4 节的切分粒度、top_k 对应 3.7 节的检索数量、System Prompt 对应 3.8 节的 Prompt 模板。区别在于，第三章需要改代码才能调参，这里只需要在界面上修改并保存。

4.6 RAG 过程可视化：让检索"看得见" 链接到标题

  这是本项目最具教学价值的功能，也是与第三章手写版最大的体验差异。在第三章中，检索过程是"黑盒"的——rag_query() 函数内部完成了检索、拼接、生成，你只能看到最终的回答文本。而在本项目中，检索的每一步都被"摊开"展示在界面上。

4.6.1 检索结果面板 链接到标题

  对话页面右栏的"检索结果"标签页，会在每次提问后实时展示 Top-K 检索结果。每个结果卡片包含：

• 排名编号：第几个被检索到

• 文件路径 + 行号范围：精确定位到源代码位置

• 相似度分数：0-1 之间的数值，附带可视化进度条

• 代码片段：可展开查看完整的 chunk 内容，带语法高亮

  这直接对应第三章 3.7 节 search() 函数的返回结果，但从终端的 print(f"Score: {score:.4f}") 升级为了可视化的分数条和代码预览。

4.6.2 Prompt 预览面板 链接到标题

  右栏的"Prompt 构建"标签页展示了发送给 LLM 的完整 Prompt，按四个子标签页组织：

• System：系统提示词（对应 3.8 节的 SYSTEM_PROMPT）

• Context：检索到的代码片段拼接结果（对应 3.9 节的上下文拼接）

• History：多轮对话历史（第三章未涉及，这是工程化新增的能力）

• User：用户的原始问题

  面板底部还会显示 Token 估算值，让你直观感受上下文窗口的消耗情况。这与第三章 3.9 节的 estimate_tokens() 函数逻辑一致，只是从代码中的数字变成了界面上的实时指标。

  通过这套可视化机制，你可以清晰地回答以下问题：LLM 的回答是基于哪些代码片段生成的？这些片段的相似度分数是多少？完整的 Prompt 长什么样？Token 消耗了多少？——这些在第三章只能通过 print 调试的信息，现在全部一目了然。

4.7 本章总结 链接到标题

  本章我们没有写一行新的 RAG 逻辑，而是通过一个完整的工程化项目，验证了第三章手写知识的"迁移能力"。让我们回顾一下从手写原型到工程产品的完整路径：

🎉 知识闭环：从手写到工程化的完整映射

• ✅ 第三章手写的 load_documents() → 项目中的仓库管理 + 文件扫描服务
• ✅ 第三章手写的 split_text() → 项目中的行边界对齐切分 + overlap 追踪
• ✅ 第三章手写的余弦相似度 + FAISS → 项目中的 ChromaDB 向量数据库
• ✅ 第三章手写的 search() → 项目中的结构化检索 + 可视化分数展示
• ✅ 第三章手写的 Prompt 模板 → 项目中的四段式 Prompt + 实时预览
• ✅ 第三章手写的 rag_query() → 项目中的 SSE 流式对话 + 会话管理

  更重要的是，你现在拥有了一个可以在本地运行的完整 RAG 应用。你可以用它来检索自己的代码仓库，也可以基于它进行二次开发。以下是一些值得探索的扩展方向：

• 支持更多文件类型：当前只扫描 .py 文件，可以扩展到 .js、.ts、.java、.go 等

• 优化切分策略：从固定长度切分升级为按函数/类级别的 AST 语法树切分

• 引入 Reranker：在 Top-K 检索后增加重排序步骤，提升检索精度（对应第三章末尾提到的 Advanced RAG 优化方向）

• 部署上线：前端 npm run build 生成静态文件，后端用 Docker 容器化，即可部署到云服务器

本章新增术语表