RAG入门与从零到一搭建RAG系统 (v2)

课程说明：

体验课内容节选自《2025大模型Agent智能体开发实战》(7月班) 完整版付费课程

体验课时间有限，若想深度学习大模型技术，欢迎大家报名由我主讲的《2025大模型Agent智能体开发实战》(7月班)

《2025大模型Agent智能体开发实战》(7月班) 为【100+小时】体系大课，总共20大模块精讲精析，零基础直达大模型企业级应用！

部分项目成果演示

from IPython.display import Video

MateGen项目演示

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/MG%E6%BC%94%E7%A4%BA%E8%A7%86%E9%A2%91.mp4", width=800, height=400)

智能客服项目演示

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/%E6%99%BA%E8%83%BD%E5%AE%A2%E6%9C%8D%E6%A1%88%E4%BE%8B%E8%A7%86%E9%A2%91.mp4", width=800, height=400)

Dify项目演示

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/2f1b47f42c65fd59e8d3a83e6cb9f13b_raw.mp4", width=800, height=400)

LangChain&LangGraph搭建Multi-Agnet

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/%E5%8F%AF%E8%A7%86%E5%8C%96%E6%95%B0%E6%8D%AE%E5%88%86%E6%9E%90Multi-Agent%E6%95%88%E6%9E%9C%E6%BC%94%E7%A4%BA%E6%95%88%E6%9E%9C.mp4", width=800, height=400)

此外，若是对大模型底层原理感兴趣，也欢迎报名由我和菜菜老师共同主讲的《2025大模型原理与实战课程》(夏季班)

两门大模型课程夏季班目前上新特惠，直播间下单可享618平价钜惠，合购还有更多优惠哦~详细信息扫码添加助教，回复“大模型”，即可领取课程大纲&查看课程详情👇

RAG技术实战公开课

Part 1.RAG快速入门

一、RAG（Retrieval-Augmented Generation，检索增强生成）技术综述

RAG，Retrieval-Augmented Generation，也被称作检索增强生成技术，最早在 Facebook AI（Meta AI）在 2020 年发表的论文《Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks》（ https://arxiv.org/abs/2005.11401 ）中正式提出，这种方法的核心思想是借助一些文本检索策略，让大模型每次问答前都带入相关文本，以此来改善大模型回答时的准确性。这项技术刚发布时并未引发太大关注，而伴随2022年大模型技术大爆发，RAG技术才逐渐进入人们视野，并且由于早期大模型技术应用均已“知识库问答”为主，而RAG技术是最易上手、并且上限极高的技术，因此很快就成为了大模型技术人必备的技术之一。

1. RAG技术极简实现流程

时至今日，RAG技术已经是非常庞大的技术体系了，从简答的文档切分、存储、匹配，再到复杂的入GraphRAG（基于知识图谱的检索增强），以及复杂文档解析+多模态识别技术等等等等。

而对于初学者来说，为了更好的上手学习RAG技术，我们首先需要对RAG技术最简单的实现形式有个基础的了解。一个最简单的RAG技术实现流程如下所示：

我们需要围绕给定的文档（往往是非常长的文档）先进行切分，然后将切分的文档转化为计算机能识别的形式，也就是将其转化为一个数值型向量（也被称为词向量），然后当用户询问问题的时候，我们再将用户的问题转化为词向量，并和段落文档的词向量进行相似度匹配，借此找出和当前用户问题最相关的原始文档片段，然后将用户的问题和匹配的到的原文片段都带入大模型，进行最终的问答。由此便可实现一次完整的文档检索增强执行流程。a

具体执行过程如下所示：

3. RAG技术核心应用场景：拓展模型知识边界与减少问答幻觉

那这样的一个检索增强流程到底有什么用呢？这就不得不从当代大模型本身的三项技术缺陷开始说起了。

缺陷一：大模型幻觉

相信大家在使用大模型的时候，都会遇到大模型无中生有胡编乱造答案的情况，例如胡乱生成一些概念、一些论文甚至是一些实时等，这就是所谓的大模型幻觉。

而其中，第一代DeepSeek R1模型的幻觉是非常严重的，平均七次回答中就会有一次的回答存在幻觉，这可以说是第一版R1模型最大的短板。

大型语言模型之所以会产生幻觉，主要是因为它们的训练方式和内在机制决定了它们并不具备真正理解和验证事实的能力。模型在训练过程中，通过分析大规模文本数据来学习不同词语和句子之间的概率关系，也就是在某种程度上掌握“在什么上下文中，什么样的回答听起来更合理”。然而，模型并没有接入实时的知识库或事实核查工具，当它遇到陌生的问题、模糊的描述或者上下文不完整的输入时，就会基于概率和语料库中似是而非的关联去“编造”一个看似正确的答案。由于这些输出往往语法流畅、逻辑连贯，人类读者很容易误以为它是真实可信的内容，这就是我们通常说的“模型幻觉”。

缺陷二：有限的最大上下文

而除此之外，大模型在实际应用中还会另一个“障碍”，那就是最大上下文限制。由于大模型的本质其实是一个算法，不管是让大模型“知道”有哪些外部工具，还是要给大模型进行“背景设置”，或者是要给模型添加历史对话消息，以及本次对话的输出，都需要占用这个上下文窗口。这就使得我们在一次对话中能够给大模型灌输的知识（文本）其实是有限的。

大型语言模型还存在最大上下文限制，这是由它们的架构和计算方式决定的。每次生成回答时，模型需要把输入文本转换成固定长度的数字序列（称为token），并在内部一次性加载到模型的“上下文窗口”中进行处理。这个窗口的大小是有限的，不同模型一般在几千到几万token之间。如果输入内容超出这个长度，模型要么截断最前面的部分，要么丢弃部分信息，这就会造成对话历史、长文档或先前提到的重要细节的遗失。因为它无法跨越上下文窗口无限地保留信息，所以在面对长对话或者大量背景知识时，模型常常出现上下文断裂、回答不连贯或者忽略先前条件的情况。

早些时候的大模型普遍是8k最大上下文，相当于是8-10页中文PDF，伴随着大模型预训练技术的不断发展，顶尖的大模型，如Gemini 2.5 Pro和GPT-4.1等模型，已经达到了1M的最大上下文长度，相当于是一千页的PDF，相当于1.5本《红楼梦》，而普通的模型，也基本达到64K或128K最大上下文，相当于60-100也左右的PDF。

但是，模型上下文的增长也是有限度的，对于开发者来说，能够一次性输入的信息都会有限制。

缺陷三：模型专业知识与时效性知识不足

大型语言模型虽然在通用领域展现出令人瞩目的语言理解和生成能力，但其在特定领域的专业知识掌握往往存在明显局限。其根本原因在于，模型的训练依赖于预先收集的大规模语料，这些语料覆盖面虽广，却很难保证在所有专业领域中具有足够的深度和准确性。某些领域，如医学、法律或前沿科技，知识更新速度快且门槛较高，公开可获取的高质量数据本身就有限，模型难以在此基础上形成系统性和权威性的认知。此外，模型训练通常在固定的时间点结束，因此其所掌握的知识具有天然的时效性，无法实时反映新近出现的研究成果、政策变化或行业动态。这种静态的知识存储模式，决定了大模型在面对最新或高度专业化的问题时，往往难以提供全面、精确的解答。

基于此，我们再回顾RAG的技术实现流程，就不难发现其背后的技术价值了：如果我们能在每次对话的时候，为当前模型输入最精准的问题相关的文档，那就能拓展模型的知识边界，无论是提升模型专业知识的准确性、给模型灌输一些时效性的知识、或者消除模型幻觉，都将大有助益，而在其他一些对话场景中，无论是需要围绕海量的文本搭建本地问答知识库、还是在构建无限上下文的聊天机器人，RAG技术都是最佳解决方案。

4. 问答机器人标配：RAG系统

正因为知识库检索的广泛的使用需求，RAG技术几乎成了现在各项聊天机器人的标配，无论是面向普通用户的聊天问答应用Cherry Studio：

还是面向企业应用场景的通用开源前端Open-WebUI:

都毫无例外都配置了RAG功能，而对于OpenAI-WebUI这种企业级前端，还为用户展示了RAG检索过程诸多技术细节：

尽管这些项目能让用户更加快速的使用RAG系统，但这种传统的RAG流程（也被称作Native RAG），在长期的应用过程中也逐渐展露出很多问题，例如对于非结构化的文本（例如包含图片、公式的文本）无法进行检索，而对于超大规模文本的检索又会存在精度不足、或者无法提炼总结跨文本概念等问题。为此，近两年的时间里，在无数技术人的共同努力下，RAG技术有了长足的成长和突破。

我们团队自研的开源Jupyter智能体助教MateGen Air，也提供了完整的公开课（部分）知识库问答功能：

5. RAG全栈技术体系介绍

但是，就像前文介绍的那样，RAG技术是一项应用面广、门槛很低、但同时上限也很高的一项技术。历经数年的技术发展，RAG技术的体系已经非常庞大，以下是RAG技术全栈技术框架概览：

GraphRAG

GraphRAG（Graph-enhanced Retrieval-Augmented Generation） 是一种在经典 RAG 基础上引入知识图谱/图结构的新型检索生成方法。其核心思想是通过将文档或数据转换成图的形式，从而捕捉实体与实体之间的语义关系，并在检索阶段利用图遍历、关系推理等机制来辅助上下文构建，这种结构化信息能够提升语义理解和多跳推理能力。

具体来说，GraphRAG 的流程包括：

图谱构建：将文本拆分为多个单元（TextUnit），提取实体与关系，构造知识图，并进行图社区检测与摘要；
混合检索：用户提问既可以进行向量检索定位实体，也可以通过图查询（如 Cypher/SPARQL）沿关系边扩展信息；
图增强生成：将检索到的节点、路径、社区摘要等信息拼接进 Prompt，引导 LLM 生成更准确、结构清晰、并基于事实推理的回答。

对比维度	传统 RAG	GraphRAG
检索方式	基于向量语义相似度	向量+知识图遍历/查询
关系理解能力	弱：只能匹配语义相近片段	强：能理解实体之间的多跳关系与结构
多跳推理支持	弱：难以综合跨文档信息	强：图结构天然支持推理路径遍历
语义上下文覆盖	依赖检索片段	可检索完整实体子图、社区摘要
可解释性	中：返回片段但缺关键信息结构	高：能显示实体关系路径及社区结构
性能/复杂度	低：直接使用向量库	高：需要图构建、遍历、摘要等pipeline

传统 RAG 主要是“先检索语义近似片段，再生成回答”，适合简单查询与短对话。但当问题需要“连接多个事实”“推理关系链”和“洞察上下文结构”时，传统 RAG 会显得力不从心，而 GraphRAG 正是为复杂推理场景设计的增强机制。

Agentic RAG

Agentic RAG（Agentic Retrieval-Augmented Generation） 是一种在传统 RAG 基础上进一步扩展的增强范式，它将检索增强生成与Agent（智能体）能力有机结合，使大模型不仅能够基于外部知识库进行回答，还能够通过一系列自主决策和工具调用来完成复杂任务。与经典 RAG 的“检索+拼接+生成”线性流程不同，Agentic RAG 将 LLM 视为一个具备推理、规划和操作能力的智能体，它在对话过程中可以根据问题拆解子任务，先后执行多轮检索、知识整合、函数调用甚至外部API请求，再将结果动态组合成最终的答案。

在这个模式下，大模型可以主动提出接下来的检索需求，或根据中间推理结果迭代获取更多信息，形成“循环式检索与生成”的闭环工作流。例如，当用户提出复杂查询时，Agentic RAG可以先调用检索工具定位候选内容，再使用工具对结果进行归纳或分类，必要时还会触发计算或外部查询操作，最后再汇总所有信息输出一个有依据的、分步骤的解答。

相比传统RAG，Agentic RAG不仅提升了回答准确性和透明度，也为多轮推理和跨知识库整合提供了更强的灵活性，是近年来大模型产品中非常重要的能力演进方向。

6. RAG热门开源项目&产品

而如果不打算自主开发，目前也有非常多RAG成熟的开源项目，可以直接作为RAG产品进行使用。

6.1 MaxKB

MaxKB（Max Knowledge Brain）是一款开源的、面向企业级应用的智能知识库助手，深度集成了 RAG（Retrieval‑Augmented Generation）管道和流程编排能力。它支持用户通过上传文档或自动爬取网页内容，系统会自动完成分段、向量化检索等流程，从而显著减少大模型回答时的“幻觉”风险，提升问答的准确性和可信度。

此外，MaxKB 配备了一个灵活的 Agentic Workflow 引擎和丰富的工具函数集，能够满足复杂业务场景下的智能流程编排需求。它支持与各类 LLM（如 OpenAI/Claude/Gemini、本地模型 Llama、Qwen）及第三方系统进行零代码集成，方便快速在企业内部构建智能客服、内部知识问答、学术研究助手等应用。

总之，MaxKB 提供了一个“开箱即用”的智能知识服务框架，技术中立且功能全面，适用于多样化企业场景，诸如客服、知识管理、教育及科研等。GPT 架构下通过 RAG 技术减少幻觉，并通过流程引擎强化业务能力，是一款值得企业部署的高效开源平台。但MaxKB只支持在线使用，数据隐私安全性难以得到保障，同时若想要创建更多知识库，还需要单独支付费用。

MaxKB项目主页：https://github.com/1Panel-dev/MaxKB

6.2 RAGFlow

RAGFlow 是一款功能全面且高可配置的开源 RAG 引擎，专注于“深度文档理解”（Deep Document Understanding），旨在帮助企业和开发者高效构建以文档为基础的智能问答系统。

它支持多种文档格式（如 PDF、Word、PPT、Excel、扫描件等），并以复杂布局识别和OCR 分块模板为核心，对文档进行结构化拆分，以生成适合检索的知识单元。

在检索阶段，RAGFlow 提供多路召回策略（包括向量检索和混合重排序），并生成可追溯的引用，能够显著减少模型幻觉，提高答案可信度。

在生成环节，它具备内置的流程引擎（Agentic Workflow），结合 LLM 能够执行自动化推理任务（如代码执行、SQL 查询）。

技术架构上，RAGFlow 提供 Docker + Helm 快速部署能力，支持 x86 & GPU 加速；并兼容主流 LLM 提供商与自部署选项，包括 OpenAI、Anthropic、Ollama、本地模型等。此外，它还配备交互式 Web UI 和低代码 Agent 搭建界面，用户可零代码创建知识库、上传文档、并生成可引用的对话助手或检索系统。

RAGFlow项目主页：https://github.com/infiniflow/ragflow/

6.3 LangChain-ChatChat

LangChain‑Chatchat（原名 LangChain‑ChatGLM）是一款基于 LangChain 框架构建的开源、本地部署知识库问答与 Agent 应用平台，致力于在中文场景和开源大模型上提供流畅、可脱机运行的智能对话体验。它融合向量检索与生成式大模型，实现了完整的 RAG 问答流程，包括文档读取、内容分段、向量化检索、Top‑k 匹配，以及将检索出的内容与用户问题一起拼入 Prompt，驱动 LLM 生成答案。该项目已支持主流开源 LLM（如 ChatGLM‑6B、GLM‑4-Chat、Qwen2‑Instruct、LLaMA 等）及 Embedding 模型，同时兼容多个本地推理框架（如 Xinference、Ollama、FastChat），也支持通过 OpenAI API 调用 GPT 模型。无论是在线还是离线环境，用户都能通过命令行或 Docker 快速部署，并自定义知识库路径和模型配置。

在功能方面，LangChain‑Chatchat 提供：

一站式知识库问答接口，支持文件、数据库、图片等多源输入；
可控的 Agent 能力，支持工具调用与流程执行；
丰富的 WebUI 与低代码交互方式，便于管理会话、系统提示词、检索配置等。

LangChain-chatchat官网：https://github.com/chatchat-space/Langchain-Chatchat

7. RAG系统开发框架

最佳RAG系统开发开源框架：LangChain&LangGraph

在当前的大模型应用开发生态中，LangChain 已经成为构建RAG（Retrieval-Augmented Generation）系统最受欢迎的框架之一。LangChain 不仅提供了面向开发者的高层API，还整合了文档加载、文本分块、向量检索、上下文拼接、输出解析等全流程工具，极大降低了RAG应用的开发门槛。在检索阶段，LangChain 提供了多种Document Loaders（如PDF、Markdown、网页、数据库加载器），并内置了RecursiveCharacterTextSplitter、MarkdownHeaderTextSplitter等分块工具，方便将原始文本转化为高质量的检索单元。向量化方面，LangChain兼容主流Embedding模型（OpenAI Embedding、Hugging Face模型、Cohere等），并支持Chroma、FAISS、Weaviate、Pinecone等多种向量数据库无缝集成。

在生成与问答环节，LangChain封装了RetrievalQA、ConversationalRetrievalChain、MultiQueryRetriever等常用组件，能够快速搭建基于单轮或多轮对话的检索增强问答系统。对于更高阶的能力，LangChain还支持LLM Chain与Agent模式，开发者可以通过工具调用和多步骤推理，构建具备复杂交互逻辑的Agentic RAG系统。总体来看，LangChain为RAG开发提供了丰富的工具集和模块化能力，使构建一个可扩展的知识检索与生成系统从“几周工程”缩短为“几天内可原型验证”。

新一代Agents SDK、ADK内置的在线RAG服务

在最新的大模型技术体系中，OpenAI Agent SDK和谷歌 Agent Development Kit（ADK）*分别代表了两大云平台对*检索增强生成（RAG）能力的官方支持路径，两者虽然同属“Agent+RAG”范式，但在功能侧重点和生态整合方面各有特色。

OpenAI Agent SDK通过原生File Search机制，为开发者提供了极简化的RAG接入方式。用户仅需在Assistant配置中启用文件检索工具，便可实现自动分块、向量化与高效召回，整个过程在OpenAI云端一体化托管，无需额外配置数据库或索引管理。该模式支持多轮对话的上下文跟踪和结果拼接，能够与Function Calling无缝结合，实现“先检索后调用工具”的闭环逻辑，尤其适合对系统稳定性和开发便捷性要求较高的场景。

相比之下，谷歌ADK则在多模态检索与推理流水线方面提供了更强的灵活性。其核心能力之一“Grounding”不仅支持文本向量检索，还能原生处理PDF、表格、扫描件等多模态数据，并提供自动可追溯引用功能，使答案生成过程更加透明可信。ADK允许开发者通过流水线（Pipeline）将检索、摘要、分类等步骤串联组合，构建复杂的多步推理流程，并支持与谷歌云生态（Drive、Gmail、Cloud Storage）深度集成。

总体而言，OpenAI Agent SDK更加专注于“一体化、低门槛的RAG体验”，而谷歌ADK则以“多模态、可编排、高可扩展性”为核心定位。两者均标志着RAG技术从最初的工程框架（如LangChain、LlamaIndex）走向平台原生支持，也体现了未来智能体开发将更加重视知识检索、自动推理和可追溯性等能力的趋势。

公开课课件领取：

二、从零到一手动搭建RAG系统

import os
from openai import OpenAI
import matplotlib.pyplot as plt
import numpy as np
from typing import Dict, List, Optional, Tuple, Union

import PyPDF2
import markdown
import html2text
import json
from tqdm import tqdm
import tiktoken
import re
from bs4 import BeautifulSoup
from IPython.display import display, Code, Markdown

api_key = 'your-openai-api-key'

# 实例化客户端
client = OpenAI(api_key=api_key,
                base_url="https://ai.devtool.tech/proxy/v1")

# 临时设置环境变量
os.environ["OPENAI_API_KEY"] = 'your-openai-api-key'
os.environ["OPENAI_BASE_URL"] = "https://ai.devtool.tech/proxy/v1"

正如此前所说，在自然语言处理和机器学习领域中，Embedding 是将文本转化为数值向量的常用方法。通过这种方式，模型可以衡量不同文本之间的相似性，进而应用于如搜索、分类、推荐等多个领域。OpenAI 刚刚发布了其第三代 Embedding 模型，这些模型具有更高的性能、更低的成本，且在多语言处理上表现出色。以下是关于 OpenAI 新一代 Embedding 模型的详细介绍和调用方法。

1.OpenAI Embedding 模型简介

Embedding 是将文本字符串表示为向量（浮点数列表），通过计算向量之间的距离来衡量文本之间的相关性。向量距离越小，表示文本之间的相关性越高；距离越大，相关性越低。常见的 Embedding 应用包括：

搜索：根据文本查询的相关性对结果进行排序
聚类：根据文本相似性将其分组
推荐：根据相关文本字符串推荐项目
异常检测：识别与其他内容相关性较低的异常点
多样性测量：分析相似性分布
分类：将文本字符串根据其最相似的标签进行分类

OpenAI 最新的 Embedding 模型包括 text-embedding-3-small 和 text-embedding-3-large，它们比以往的模型具有更高的性能，且支持更多语言。这两个模型分别生成长度为 1536 和 3072 的向量。此外，用户可以通过设置维度参数来减少向量的维度，而不损失其表示概念的能力。

OpenAI Embedding模型获取方法

要获取文本的 Embedding 向量，可以将文本字符串发送到 OpenAI 的 Embedding API 端点，并指定所使用的模型（例如 text-embedding-3-small）。响应结果将包含 Embedding 向量，以及一些额外的元数据信息。

截止目前，OpenAI 提供了两个第三代 Embedding 模型，定价基于输入的 token 数量。以下是关于模型性能与定价的概览：

模型	每美元支持的页面数量	在 MTEB 测试中的表现	最大输入 token 数量
text-embedding-3-small	62,500	62.3%	8191
text-embedding-3-large	9,615	64.6%	8191
text-embedding-ada-002	12,500	61.0%	8191

# 调用 embedding API 获取文本的向量表示
response = client.embeddings.create(
    input="测试文本",  # 输入文本
    model="text-embedding-3-small"  # 选择 Embedding 模型
)

# 打印返回的 embedding 向量
print(response.data[0].embedding)

len(response.data[0].embedding)

返回的 Embedding 向量可以直接用于多种应用场景，例如存储在向量数据库中，进行文本相似度搜索等。默认情况下，text-embedding-3-small 生成的向量长度为 1536，text-embedding-3-large 的向量长度为 3072。

余弦相似度介绍与效果简介
欧氏距离与余弦相似度计算公式

假设现有a、b两个向量：

$$\vec{a} = [a_1, a_2, a_3, ...]$$

$$\vec{b} = [b_1, b_2, b_3, ...]$$

余弦相似度计算公式为：

$$ \text{Cosine Similarity} (\vec{a}, \vec{b}) = \frac{\vec{a} \cdot \vec{b}}{|\vec{a}| |\vec{b}|} $$

其中：

$\vec{a} \cdot \vec{b}$ 表示向量 $\vec{a}$ 和向量 $\vec{b}$ 的点积。
$|\vec{a}|$ 和 $|\vec{b}|$ 分别是向量 $\vec{a}$ 和 $\vec{b}$ 的模（长度）。

点积 (Dot Product) 定义为： $$ \vec{a} \cdot \vec{b} = a_1b_1 + a_2b_2 + \ldots + a_nb_n $$

向量的模 (Magnitude) 定义为： $$ |\vec{a}| = \sqrt{a_1^2 + a_2^2 + \ldots + a_n^2} $$ $$ |\vec{b}| = \sqrt{b_1^2 + b_2^2 + \ldots + b_n^2} $$

例如，余弦相似度可以通过如下方式进行计算和呈现：

import matplotlib.pyplot as plt
import numpy as np

# 创建两个向量
a = np.array([0, 1])
b = np.array([1, 1])

# 计算两个向量的余弦相似度
cosine_similarity = np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 绘制向量
plt.quiver(0, 0, a[0], a[1], angles='xy', scale_units='xy', scale=1, color='r')
plt.quiver(0, 0, b[0], b[1], angles='xy', scale_units='xy', scale=1, color='b')

# 设置图表属性
plt.xlim(-0.5, 1.5)
plt.ylim(-0.5, 1.5)
plt.grid()
plt.title(f'Cosine Similarity: {cosine_similarity:.2f}')
plt.xlabel('X axis')
plt.ylabel('Y axis')

# 添加图例
plt.legend(['Vector a', 'Vector b'])

# 显示图表
plt.show()

这幅图展示了二维空间中两个向量的方向。红色向量代表$\vec{a}$ ，蓝色向量代表 $\vec{b}$ 。它们之间的夹角表示了两个向量的余弦相似度。余弦相似度是通过计算两个向量的点积并除以它们各自的范数（即长度）来得到的。在这个示例中，这两个向量的余弦相似度大约为 0.71，意味着它们在方向上有一定程度的相似性。这个值越接近 1，表示两个向量的方向越相似。

余弦相似度计算函数

def cosine_sim(vector1: List[float], vector2: List[float]) -> float:
    """
    计算两个向量之间的余弦相似度
    """
    dot_product = np.dot(vector1, vector2)
    magnitude = np.linalg.norm(vector1) * np.linalg.norm(vector2)
    if not magnitude:
        return 0
    return dot_product / magnitude

text1 = '我喜欢吃苹果'
text2 = "苹果是我最喜欢吃的水果"
text3 = "我喜欢用苹果手机"

vector1 = client.embeddings.create(
    input=text1,  
    model="text-embedding-3-large"  
).data[0].embedding

vector2 = client.embeddings.create(
    input=text2,  
    model="text-embedding-3-large"  
).data[0].embedding

vector3 = client.embeddings.create(
    input=text3,  
    model="text-embedding-3-large"  
).data[0].embedding

cosine_sim(vector1, vector2)

cosine_sim(vector1, vector3)

cosine_sim(vector2, vector3)

为了实现 RAG 模型的功能，我们首先需要一个向量化（Embedding）模块。向量化是 RAG 的基础，它的作用是将文档片段转化为向量表示，便于后续的检索操作。在这个过程中，我们将实现一个向量化类，用来将文本片段映射成向量。

为了便于扩展和未来可能使用不同的模型，我们首先编写一个 Embedding 基类。该基类定义了获取文本向量表示的方法，同时包含一个计算两个向量之间余弦相似度的功能。这样，如果我们未来使用不同的向量化模型，只需继承该基类并重写向量获取的逻辑，而不需要重复编写相似度计算部分。

class BaseEmbeddings:
    """
    向量化的基类，用于将文本转换为向量表示。不同的子类可以实现不同的向量获取方法。
    """
    def __init__(self, path: str, is_api: bool) -> None:
        """
        初始化基类。
        
        参数：
        path (str) - 如果是本地模型，path 表示模型路径；如果是API模式，path可以为空
        is_api (bool) - 表示是否使用API调用，如果为True表示通过API获取Embedding
        """
        self.path = path
        self.is_api = is_api
    
    def get_embedding(self, text: str, model: str) -> List[float]:
        """
        抽象方法，用于获取文本的向量表示，具体实现需要在子类中定义。
        
        参数：
        text (str) - 需要转换为向量的文本
        model (str) - 所使用的模型名称
        
        返回：
        list[float] - 文本的向量表示
        """
        raise NotImplementedError
    
    @classmethod
    def cosine_similarity(cls, vector1: List[float], vector2: List[float]) -> float:
        """
        计算两个向量之间的余弦相似度，用于衡量它们的相似程度。
        
        参数：
        vector1 (list[float]) - 第一个向量
        vector2 (list[float]) - 第二个向量
        
        返回：
        float - 余弦相似度值，范围从 -1 到 1，越接近 1 表示向量越相似
        """
        dot_product = np.dot(vector1, vector2)  # 向量点积
        magnitude = np.linalg.norm(vector1) * np.linalg.norm(vector2)  # 向量的模
        if not magnitude:
            return 0
        return dot_product / magnitude  # 计算余弦相似度

我们在这个基类基础上，可以通过继承它来实现具体的模型。例如，我们可以使用 OpenAI 的 API 来生成文本的向量表示，只需重写 get_embedding 方法即可。

class OpenAIEmbedding(BaseEmbeddings):
    """
    使用 OpenAI 的 Embedding API 来获取文本向量的类，继承自 BaseEmbeddings。
    """
    def __init__(self, path: str = '', is_api: bool = True) -> None:
        """
        初始化类，设置 OpenAI API 客户端，如果使用的是 API 调用。
        
        参数：
        path (str) - 本地模型的路径，使用API时可以为空
        is_api (bool) - 是否通过 API 获取 Embedding，默认为 True
        """
        super().__init__(path, is_api)
        if self.is_api:
            # 初始化 OpenAI API 客户端
            from openai import OpenAI
            self.client = OpenAI()
            self.client.api_key = os.getenv("OPENAI_API_KEY")  # 从环境变量中获取 API 密钥
            self.client.base_url = os.getenv("OPENAI_BASE_URL")  # 从环境变量中获取 API 基础URL
    
    def get_embedding(self, text: str, model: str = "text-embedding-3-large") -> List[float]:
        """
        使用 OpenAI 的 Embedding API 获取文本的向量表示。
        
        参数：
        text (str) - 需要转化为向量的文本
        model (str) - 使用的 Embedding 模型名称，默认为 'text-embedding-3-large'
        
        返回：
        list[float] - 文本的向量表示
        """
        if self.is_api:
            # 去掉文本中的换行符，保证输入格式规范
            text = text.replace("\n", " ")
            # 调用 OpenAI API 获取文本的向量表示
            return self.client.embeddings.create(input=[text], model=model).data[0].embedding
        else:
            raise NotImplementedError  # 如果不是 API 模式，这里未实现本地模型的处理

这样设计的结构让我们可以轻松替换或扩展向量化模型，而不需要改变整体框架。

使用示例

# 初始化 Embedding 模型
embedding_model = OpenAIEmbedding()

# 输入需要获取向量表示的文本
text = "这是一个示例文本，用于演示 OpenAI Embedding 的使用。"

# 获取文本的向量表示
embedding_vector = embedding_model.get_embedding(text, model="text-embedding-3-large")

print("文本的向量表示为：", embedding_vector)

vector1 = embedding_model.get_embedding(text1)
vector2 = embedding_model.get_embedding(text2)

similarity = BaseEmbeddings.cosine_similarity(vector1, vector2)

print(f"两段文本的余弦相似度为: {similarity}")

3. 文档加载与切分模块创建

在实现了向量化之后，我们接下来需要编写一个文档加载与切分模块，用于处理不同格式的文档并将其切分为小片段。为什么要进行切分呢？这是为了确保每个文档片段都尽量保持简短且信息集中，以便于后续的向量化和检索。

3.1 文档格式处理函数

我们的目标是支持多种格式的文档，例如 PDF、Markdown、TXT 等。每种文件格式都有不同的读取方式，下面我们展示一个支持多种格式的简单实现：

def read_file_content(cls, file_path: str):
    # 根据文件扩展名选择读取方法
    if file_path.endswith('.pdf'):
        return cls.read_pdf(file_path)
    elif file_path.endswith('.md'):
        return cls.read_markdown(file_path)
    elif file_path.endswith('.txt'):
        return cls.read_text(file_path)
    else:
        raise ValueError("Unsupported file type")

3.2 文档切分函数

这里我们考虑将文档按Token长度进行切分，设置一个最大的 Token 长度，然后按这个长度进行切分。在这个过程中，我们也会确保每个片段之间有一定的重叠，避免重要信息被切掉。

def get_chunk(cls, text: str, max_token_len: int = 600, cover_content: int = 150):
    chunk_text = []
    curr_len = 0
    curr_chunk = ''
    lines = text.split('\n')  # 以换行符为单位切分文本

    for line in lines:
        line = line.replace(' ', '')
        line_len = len(enc.encode(line))  # 计算当前行的 Token 长度
        if line_len > max_token_len:
            print('warning line_len = ', line_len)
        if curr_len + line_len <= max_token_len:
            curr_chunk += line + '\n'
            curr_len += line_len + 1
        else:
            chunk_text.append(curr_chunk)
            curr_chunk = curr_chunk[-cover_content:] + line
            curr_len = line_len + cover_content

    if curr_chunk:
        chunk_text.append(curr_chunk)

    return chunk_text

完整类编写如下：

enc = tiktoken.get_encoding("cl100k_base")

len(enc.encode("你好，好久不见！"))

class ReadFiles:
    """
    读取文件的类，用于从指定路径读取支持的文件类型（如 .txt、.md、.pdf）并进行内容分割。
    """

    def __init__(self, path: str) -> None:
        """
        初始化函数，设定要读取的文件路径，并获取该路径下所有符合要求的文件。
        :param path: 文件夹路径
        """
        self._path = path
        self.file_list = self.get_files()  # 获取文件列表

    def get_files(self):
        """
        遍历指定文件夹，获取支持的文件类型列表（txt, md, pdf）。
        :return: 文件路径列表
        """
        file_list = []
        for filepath, dirnames, filenames in os.walk(self._path):
            # os.walk 函数将递归遍历指定文件夹
            for filename in filenames:
                # 根据文件后缀筛选支持的文件类型
                if filename.endswith(".md"):
                    file_list.append(os.path.join(filepath, filename))
                elif filename.endswith(".txt"):
                    file_list.append(os.path.join(filepath, filename))
                elif filename.endswith(".pdf"):
                    file_list.append(os.path.join(filepath, filename))
        return file_list

    def get_content(self, max_token_len: int = 600, cover_content: int = 150):
        """
        读取文件内容并进行分割，将长文本切分为多个块。
        :param max_token_len: 每个文档片段的最大 Token 长度
        :param cover_content: 在每个片段之间重叠的 Token 长度
        :return: 切分后的文档片段列表
        """
        docs = []
        for file in self.file_list:
            content = self.read_file_content(file)  # 读取文件内容
            # 分割文档为多个小块
            chunk_content = self.get_chunk(content, max_token_len=max_token_len, cover_content=cover_content)
            docs.extend(chunk_content)
        return docs

    @classmethod
    def get_chunk(cls, text: str, max_token_len: int = 600, cover_content: int = 150):
        """
        将文档内容按最大 Token 长度进行切分。
        :param text: 文档内容
        :param max_token_len: 每个片段的最大 Token 长度
        :param cover_content: 重叠的内容长度
        :return: 切分后的文档片段列表
        """
        chunk_text = []
        curr_len = 0
        curr_chunk = ''
        token_len = max_token_len - cover_content
        lines = text.splitlines()  # 以换行符分割文本为行

        for line in lines:
            line = line.replace(' ', '')  # 去除空格
            line_len = len(enc.encode(line))  # 计算当前行的 Token 长度
            if line_len > max_token_len:
                # 如果单行长度超过限制，将其分割为多个片段
                num_chunks = (line_len + token_len - 1) // token_len
                for i in range(num_chunks):
                    start = i * token_len
                    end = start + token_len
                    # 防止跨单词分割
                    while not line[start:end].rstrip().isspace():
                        start += 1
                        end += 1
                        if start >= line_len:
                            break
                    curr_chunk = curr_chunk[-cover_content:] + line[start:end]
                    chunk_text.append(curr_chunk)
                start = (num_chunks - 1) * token_len
                curr_chunk = curr_chunk[-cover_content:] + line[start:end]
                chunk_text.append(curr_chunk)
            elif curr_len + line_len <= token_len:
                # 当前片段长度未超过限制时，继续累加
                curr_chunk += line + '\n'
                curr_len += line_len + 1
            else:
                chunk_text.append(curr_chunk)  # 保存当前片段
                curr_chunk = curr_chunk[-cover_content:] + line
                curr_len = line_len + cover_content

        if curr_chunk:
            chunk_text.append(curr_chunk)

        return chunk_text

    @classmethod
    def read_file_content(cls, file_path: str):
        """
        读取文件内容，根据文件类型选择不同的读取方式。
        :param file_path: 文件路径
        :return: 文件内容
        """
        if file_path.endswith('.pdf'):
            return cls.read_pdf(file_path)
        elif file_path.endswith('.md'):
            return cls.read_markdown(file_path)
        elif file_path.endswith('.txt'):
            return cls.read_text(file_path)
        else:
            raise ValueError("Unsupported file type")

    @classmethod
    def read_pdf(cls, file_path: str):
        """
        读取 PDF 文件内容。
        :param file_path: PDF 文件路径
        :return: PDF 文件中的文本内容
        """
        with open(file_path, 'rb') as file:
            reader = PyPDF2.PdfReader(file)
            text = ""
            for page_num in range(len(reader.pages)):
                text += reader.pages[page_num].extract_text()
            return text

    @classmethod
    def read_markdown(cls, file_path: str):
        """
        读取 Markdown 文件内容，并将其转换为纯文本。
        :param file_path: Markdown 文件路径
        :return: 纯文本内容
        """
        with open(file_path, 'r', encoding='utf-8') as file:
            md_text = file.read()
            html_text = markdown.markdown(md_text)
            # 使用 BeautifulSoup 从 HTML 中提取纯文本
            soup = BeautifulSoup(html_text, 'html.parser')
            plain_text = soup.get_text()
            # 使用正则表达式移除网址链接
            text = re.sub(r'http\S+', '', plain_text) 
            return text

    @classmethod
    def read_text(cls, file_path: str):
        """
        读取普通文本文件内容。
        :param file_path: 文本文件路径
        :return: 文件内容
        """
        with open(file_path, 'r', encoding='utf-8') as file:
            return file.read()

class Documents:
    """
    文档类，用于读取已分好类的 JSON 格式文档。
    """
    def __init__(self, path: str = '') -> None:
        self.path = path

    def get_content(self):
        """
        读取 JSON 格式的文档内容。
        :return: JSON 文档的内容
        """
        with open(self.path, mode='r', encoding='utf-8') as f:
            content = json.load(f)
        return content

# 初始化 ReadFiles 类，指定文件目录路径
file_reader = ReadFiles(path="./data")

# 获取目录下所有支持的文件类型
file_list = file_reader.get_files()
print("支持的文件列表：", file_list)

# 将文件内容读取并分块
document_chunks = file_reader.get_content(max_token_len=600, cover_content=150)
print("分块后的文档内容：", document_chunks)

document_chunks[0]

document_chunks[1]

4. 词向量数据库与向量检索模块

接下来我们将继续构建向量数据库以及检索模块，这是 RAG 模型中的核心功能之一。向量数据库用于存储文档片段及其对应的向量表示，而检索模块则根据用户提出的问题（Query）在数据库中检索相关的文档片段。通过这些功能，我们创建的简易 RAG 能够根据输入的查询快速找到最相关的文档片段。

为了构建这个向量数据库，我们需要以下几个关键功能：

持久化存储（persist）： 将数据库存储到本地，便于下次加载使用。
加载数据库（load_vector）： 从本地文件加载已经存储的向量和文档。
获取向量表示（get_vector）： 将文档转化为向量表示并存储。
检索（query）： 根据用户的 Query，检索数据库中的相关文档片段。

我们将基于这些功能来实现一个简单的 VectorStore 类。

首先，我们创建一个基础的 VectorStore 类，提供上述功能的框架。通过这个类，我们能够将文档片段转化为向量存储，加载本地数据库，进行检索。

class VectorStore:
    def __init__(self, document: List[str] = None) -> None:
        """
        初始化向量存储类，存储文档和对应的向量表示。
        :param document: 文档列表，默认为空。
        """
        if document is None:
            document = []
        self.document = document  # 存储文档内容
        self.vectors = []  # 存储文档的向量表示
    
    def get_vector(self, EmbeddingModel: BaseEmbeddings) -> List[List[float]]:
        """
        使用传入的 Embedding 模型将文档向量化。
        :param EmbeddingModel: 传入的用于生成向量的模型（需继承 BaseEmbeddings 类）。
        :return: 返回文档对应的向量列表。
        """
        # 遍历所有文档，获取每个文档的向量表示
        self.vectors = [EmbeddingModel.get_embedding(doc) for doc in self.document]
        return self.vectors
    
    def persist(self, path: str = 'storage'):
        """
        将文档和对应的向量表示持久化到本地目录中，以便后续加载使用。
        :param path: 存储路径，默认为 'storage'。
        """
        if not os.path.exists(path):
            os.makedirs(path)  # 如果路径不存在，创建路径
        # 保存向量为 numpy 文件
        np.save(os.path.join(path, 'vectors.npy'), self.vectors)
        # 将文档内容存储到文本文件中
        with open(os.path.join(path, 'documents.txt'), 'w') as f:
            for doc in self.document:
                f.write(f"{doc}\n")
    
    def load_vector(self, path: str = 'storage'):
        """
        从本地加载之前保存的文档和向量数据。
        :param path: 存储路径，默认为 'storage'。
        """
        # 加载保存的向量数据
        self.vectors = np.load(os.path.join(path, 'vectors.npy')).tolist()
        # 加载文档内容
        with open(os.path.join(path, 'documents.txt'), 'r') as f:
            self.document = [line.strip() for line in f.readlines()]

    def get_similarity(self, vector1: List[float], vector2: List[float]) -> float:
        """
        计算两个向量的余弦相似度。
        :param vector1: 第一个向量。
        :param vector2: 第二个向量。
        :return: 返回两个向量的余弦相似度，范围从 -1 到 1。
        """
        dot_product = np.dot(vector1, vector2)
        magnitude = np.linalg.norm(vector1) * np.linalg.norm(vector2)
        if not magnitude:
            return 0
        return dot_product / magnitude

    def query(self, query: str, EmbeddingModel: BaseEmbeddings, k: int = 1) -> List[str]:
        """
        根据用户的查询文本，检索最相关的文档片段。
        :param query: 用户的查询文本。
        :param EmbeddingModel: 用于将查询向量化的嵌入模型。
        :param k: 返回最相似的文档数量，默认为 1。
        :return: 返回最相似的文档列表。
        """
        # 将查询文本向量化
        query_vector = EmbeddingModel.get_embedding(query)
        # 计算查询向量与每个文档向量的相似度
        similarities = [self.get_similarity(query_vector, vector) for vector in self.vectors]
        # 获取相似度最高的 k 个文档索引
        top_k_indices = np.argsort(similarities)[-k:][::-1]
        # 返回对应的文档内容
        return [self.document[idx] for idx in top_k_indices]

上述代码解释如下：

get_vector 方法： 这个方法使用传入的 EmbeddingModel 对所有文档进行向量化，并将这些向量存储在 self.vectors 中。
persist 方法： 该方法将文档片段及其向量表示保存到本地文件系统，便于持久化存储。
load_vector 方法： 从本地文件系统加载已保存的文档片段和向量，供后续检索使用。
get_similarity 方法： 计算两个向量之间的余弦相似度，用于比较查询和文档向量的相似度。
query 方法： 接收用户输入的查询，通过向量化后在数据库中检索最相关的文档片段，并返回最匹配的文档。

假设我们已经有一组文档片段存储在 documents 中，并且使用 OpenAI 的 Embedding API 进行向量化处理，以下是一个简化的运行示例：

# 初始化文档列表
documents = [
    "机器学习是人工智能的一个分支。",
    "深度学习是一种特殊的机器学习方法。",
    "监督学习是一种训练模型的方式。",
    "强化学习通过奖励和惩罚进行学习。",
    "无监督学习不依赖标签数据。",
]

# 创建向量数据库
vector_store = VectorStore(document=documents)

# 使用 OpenAI Embedding 模型对文档进行向量化
embedding_model = OpenAIEmbedding()

# 获取文档向量并存储
vector_store.get_vector(embedding_model)

# 持久化存储到本地
vector_store.persist('storage')

# 模拟用户查询
query = "什么是深度学习？"
result = vector_store.query(query, embedding_model)

print("检索结果：", result)

在上面这段代码中，npy 文件是 NumPy 库用于存储数组数据的文件格式。npy 文件能够高效地保存和加载 NumPy 数组，并保留数组的形状、数据类型等信息。它是一种二进制文件格式，用于序列化 NumPy 数组，使得存储和读取过程更加快速和便捷。

loaded_array = np.load('./storage/vectors.npy')

print(loaded_array)

5. 大模型问答模块编写

现在我们已经构建了向量数据库和检索模块，接下来我们将实现大模型模块，用于根据检索到的相关文档片段生成对用户问题的回答。为了简化和便于扩展，我们会先实现一个基类 BaseModel，然后再以 GPT4o 模型为例，展示如何使用大语言模型来完成这个任务。

5.1 大模型模块的基类

这里我们先编写一个基类 BaseModel，它包含两个主要方法：

chat：负责处理用户的输入并生成回答。
load_model：如果是使用本地模型，这个方法负责加载模型。如果使用 API 模型（如 OpenAI），可以不用实现这个方法。

class BaseModel:
    """
    基础模型类，作为所有模型的基类。
    包含一些通用的接口，如加载模型、生成回答等。
    """
    def __init__(self, path: str = '') -> None:
        self.path = path  # 用于存储模型文件的路径，默认为空。

    def chat(self, prompt: str, history: List[dict], content: str) -> str:
        """
        使用模型生成回答的抽象方法。
        :param prompt: 用户的提问内容
        :param history: 之前的对话历史（字典列表）
        :param content: 提供的上下文内容
        :return: 模型生成的答案
        """
        pass  # 具体的实现由子类提供

    def load_model(self):
        """
        加载模型的方法，通常用于本地模型。
        """
        pass  # 如果是 API 模型，可能不需要实现

5.2 借助GPT4o模型进行对话

class GPT4oChat(BaseModel):
    """
    基于 GPT-4o 模型的对话类，继承自 BaseModel。
    主要用于通过 OpenAI API 来生成对话回答。
    """
    def __init__(self, api_key: str, base_url: str = "https://ai.devtool.tech/proxy/v1") -> None:
        """
        初始化 GPT-4o 模型。
        :param api_key: OpenAI API 的密钥
        :param base_url: 用于访问 OpenAI API 的基础 URL，默认为代理 URL
        """
        super().__init__()
        self.client = OpenAI(api_key=api_key, base_url=base_url)  # 初始化 OpenAI 客户端

    def chat(self, prompt: str, history: List = [], content: str = '') -> str:
        """
        使用 GPT-4o 生成回答。
        :param prompt: 用户的提问
        :param history: 之前的对话历史（可选）
        :param content: 可参考的上下文信息（可选）
        :return: 生成的回答
        """
        # 构建包含问题和上下文的完整提示
        full_prompt = PROMPT_TEMPLATE['GPT4o_PROMPT_TEMPLATE'].format(question=prompt, context=content)

        # 调用 GPT-4o 模型进行推理
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",  # 使用 GPT-4o 小型模型
            messages=[
                {"role": "user", "content": full_prompt}
            ]
        )

        # 返回模型生成的第一个回答
        return response.choices[0].message.content

5.3 提示模板

为了方便维护和复用提示语，可以使用一个字典来保存不同模型的提示模板。在这里我们为 GPT4o 定义了一个模板：

PROMPT_TEMPLATE = dict(
    GPT4o_PROMPT_TEMPLATE="""
    下面有一个或许与这个问题相关的参考段落，若你觉得参考段落能和问题相关，则先总结参考段落的内容。
    若你觉得参考段落和问题无关，则使用你自己的原始知识来回答用户的问题，并且总是使用中文来进行回答。
    问题: {question}
    可参考的上下文：
    ···
    {context}
    ···
    有用的回答:"""
)

6. 体验课RAG Demo完整流程演示

接下来，我们展示一个完整的 RAG Demo，结合我们前面实现的向量检索和大模型模块，展示如何在实际应用中使用 RAG 模型来回答问题。

# 加载并切分文档
docs = ReadFiles('./data').get_content(max_token_len=600, cover_content=150)
vector = VectorStore(docs)

# 使用 OpenAI Embedding 模型进行向量化
embedding = OpenAIEmbedding()
vector.get_vector(EmbeddingModel=embedding)

# 将向量和文档保存到本地
vector.persist(path='storage')

# 用户提出问题
question = '如何选择Jupyer的Kernel？'

# 在数据库中检索最相关的文档片段
content = vector.query(question, EmbeddingModel=embedding, k=1)[0]

content

# 使用 GPT4oChat 模型生成答案
chat = GPT4oChat(api_key = os.getenv("OPENAI_API_KEY"))  # 传入 OpenAI API 密钥
print(chat.chat(question, [], content))

完整函数封装

def run_mini_rag(question: str, knowledge_base_path: str, k: int = 1) -> str:
    """
    运行一个简化版的RAG项目。
    
    :param question: 用户提出的问题
    :param knowledge_base_path: 知识库的路径，包含文档的文件夹路径
    :param api_key: OpenAI API密钥，用于调用GPT-4o模型
    :param k: 返回与问题最相关的k个文档片段，默认为1
    :return: 返回GPT-4o模型生成的回答
    """
    api_key = os.getenv("OPENAI_API_KEY")
    # 1. 加载并切分文档
    docs = ReadFiles(knowledge_base_path).get_content(max_token_len=600, cover_content=150)
    vector = VectorStore(docs)

    # 2. 使用 OpenAI Embedding 模型进行向量化
    embedding = OpenAIEmbedding()
    vector.get_vector(EmbeddingModel=embedding)

    # 3. 将向量和文档保存到本地（可选）
    vector.persist(path='storage')

    # 4. 在数据库中检索最相关的文档片段
    content = vector.query(question, EmbeddingModel=embedding, k=k)[0]

    # 5. 使用 GPT-4o 生成答案
    chat = GPT4oChat(api_key=api_key)
    answer = chat.chat(question, [], content)
    
    return answer

question = '请问我应该如何在安装OpenAI库的时候设置代理呢？'

knowledge_base_path = './data'

answer = run_mini_rag(question, knowledge_base_path)

display(Markdown(answer))

二、基于LangChain的RAG系统Demo示例

理解了在LangChain中构建RAG的基本原理后，我们就可以开始动手实践了。接下来的案例中，我们通过 Streamlit 前端界面，结合 LangChain 框架与 DashScope 向量嵌入服务，实现了一个轻量化的 RAG（Retrieval-Augmented Generation） 智能问答系统，支持上传多个 PDF 文档，系统将自动完成文本提取、分块、向量化，并构建基于 FAISS 的检索数据库。用户随后可以在页面中输入任意问题，系统会调用大语言模型（如 DeepSeek-Chat）对 PDF 内容进行语义理解和回答生成。

其完整代码如下所示：

# ! pip install streamlit PyPDF2 dashscope faiss-cpu

FENCE0

基于此，我们能够实现：

LangChain 的多模块能力（向量搜索 + Agent工具）
Streamlit 前端交互
FAISS 向量数据库
DashScope Embedding + DeepSeek 模型接入
并完成了完整的 RAG（检索增强生成）流程

以下是各部分功能实现代码讲解：

🔧 1. 导入库 & 环境初始化

FENCE0

Streamlit 用于构建网页界面。
PyPDF2 用来读取 PDF 文本。
load_dotenv() 加载 .env 中的 API Key，例如：
```
DEEPSEEK_API_KEY=sk-xxx
DASHSCOPE_API_KEY=xxx
```

🔐 2. 加载 API 密钥与设置环境变量

FENCE1

从环境变量中读取 DashScope 和 DeepSeek API。
设置 KMP_DUPLICATE_LIB_OK 避免某些 MKL 多线程报错。

🧠 3. 初始化向量 Embedding 模型

FENCE2

用阿里云 DashScope 提供的 text-embedding-v1 将文本转为向量表示，用于相似度搜索。

📄 4. 处理 PDF 文本与向量化逻辑

FENCE3

pdf_read：逐页读取 PDF 内容并拼接。
get_chunks：将长文本切片为多个段落（chunk），每段 1000 字，重叠 200 字。
vector_store：用 FAISS 建立向量索引，并保存到本地 faiss_db/。

🔁 5. Agent对话链 + 工具调用（核心 RAG）

FENCE4

初始化 DeepSeek 模型为 Agent。
使用 LangChain 的 create_tool_calling_agent 构造 Agent，输入：
- prompt（你设定的系统角色）
- 工具（retriever 工具）
AgentExecutor.invoke：LangChain 自动判断是否调用工具，完成“读取上下文 → 查询 → 回答”流程。

🔍 6. 用户提问逻辑（调用 FAISS）

FENCE5

加载本地 FAISS 向量库；
将其转为 LangChain 的检索工具；
交由 Agent 调用完成回答。

🧠 7. 检查数据库是否存在

FENCE6

简单检查本地是否已有向量化数据。

🌐 8. 主界面逻辑（Streamlit）

FENCE7

页面标题与界面配置。
st.columns 分栏：左边显示提示，右边放置“清空数据库”按钮。
主输入框：st.text_input("请输入问题")
- 只有当数据库存在时才能提问。
侧边栏：
- PDF 上传器；
- 提交按钮（处理上传的 PDF → 分片 → 向量化 → 存储）。

🎯 9. 提交 PDF 后执行的逻辑

FENCE8

当点击“提交并处理”后：
1. 读取上传的 PDF；
2. 切片文本；
3. 向量化入库；
4. 弹出气球提示，并 st.rerun() 刷新页面状态。

📎 项目结构总结

模块	说明
🧾 PDF解析	读取用户上传的 PDF
✂️ 文本切片	按段落分割内容
📊 向量化	DashScope Embedding + FAISS 建库
🔁 查询接口	用户输入 → 召回相关 chunk
🤖 DeepSeek Agent	调用检索工具并给出回答
💻 UI层	Streamlit 实现全部交互

其中LangChain RAG核心功能相关代码如下：

Step 1：PDF 文件上传与文本提取

使用 st.file_uploader() 组件支持多文件上传，并通过 PyPDF2.PdfReader 对每页内容进行提取，组合为整体文本。

FENCE0

Step 2：文本分块与向量数据库构建

使用 RecursiveCharacterTextSplitter 将长文档切割为固定长度（1000字）+ 重叠（200字）的小块，将文本块通过 DashScopeEmbeddings 嵌入为向量，使用 FAISS 本地存储向量数据库。

FENCE0

Step 3：用户提问与语义检索

通过 Streamlit 获取用户输入问题，如果向量数据库存在，则加载 FAISS 检索器，使用 create_retriever_tool() 构建 LangChain 工具，交由 AgentExecutor 执行，自动调用检索器并生成答案。

FENCE0

完整的代码已经上传至百度网盘中的langchain_rag.py文件中，大家可以扫描下方二维码免费领取

项目运行效果如下所示：

from IPython.display import Video

Video("https://ml2022.oss-cn-hangzhou.aliyuncs.com/LangChain%20RAG.mp4", width=800, height=400)

课程说明：​

RAG技术实战公开课

Part 1.RAG快速入门

一、RAG（Retrieval-Augmented Generation，检索增强生成）技术综述​

1. RAG技术极简实现流程​

3. RAG技术核心应用场景：拓展模型知识边界与减少问答幻觉​

4. 问答机器人标配：RAG系统​

5. RAG全栈技术体系介绍​

6. RAG热门开源项目&产品​

6.1 MaxKB​

6.2 RAGFlow​

6.3 LangChain-ChatChat​

7. RAG系统开发框架​

二、从零到一手动搭建RAG系统​

1.OpenAI Embedding 模型简介​

OpenAI Embedding模型获取方法​

3. 文档加载与切分模块创建​

3.1 文档格式处理函数​

3.2 文档切分函数​

4. 词向量数据库与向量检索模块​

5. 大模型问答模块编写​

5.1 大模型模块的基类​

5.2 借助GPT4o模型进行对话​

5.3 提示模板​

6. 体验课RAG Demo完整流程演示​

二、基于LangChain的RAG系统Demo示例​

🔧 1. 导入库 & 环境初始化​

🔐 2. 加载 API 密钥与设置环境变量​

🧠 3. 初始化向量 Embedding 模型​

📄 4. 处理 PDF 文本与向量化逻辑​

🔁 5. Agent对话链 + 工具调用（核心 RAG）​

🔍 6. 用户提问逻辑（调用 FAISS）​

🧠 7. 检查数据库是否存在​

🌐 8. 主界面逻辑（Streamlit）​

🎯 9. 提交 PDF 后执行的逻辑​

📎 项目结构总结​

课程说明：

一、RAG（Retrieval-Augmented Generation，检索增强生成）技术综述

1. RAG技术极简实现流程

3. RAG技术核心应用场景：拓展模型知识边界与减少问答幻觉

4. 问答机器人标配：RAG系统

5. RAG全栈技术体系介绍

6. RAG热门开源项目&产品

6.1 MaxKB

6.2 RAGFlow

6.3 LangChain-ChatChat

7. RAG系统开发框架

二、从零到一手动搭建RAG系统

1.OpenAI Embedding 模型简介

OpenAI Embedding模型获取方法

3. 文档加载与切分模块创建

3.1 文档格式处理函数

3.2 文档切分函数

4. 词向量数据库与向量检索模块

5. 大模型问答模块编写

5.1 大模型模块的基类

5.2 借助GPT4o模型进行对话

5.3 提示模板

6. 体验课RAG Demo完整流程演示

二、基于LangChain的RAG系统Demo示例

🔧 1. 导入库 & 环境初始化

🔐 2. 加载 API 密钥与设置环境变量

🧠 3. 初始化向量 Embedding 模型

📄 4. 处理 PDF 文本与向量化逻辑

🔁 5. Agent对话链 + 工具调用（核心 RAG）

🔍 6. 用户提问逻辑（调用 FAISS）

🧠 7. 检查数据库是否存在

🌐 8. 主界面逻辑（Streamlit）

🎯 9. 提交 PDF 后执行的逻辑

📎 项目结构总结