RAG项目开发全流程解析

2401_86951043

490人浏览 · 2026-04-06 17:24:25

2401_86951043 · 2026-04-06 17:24:25 发布

本篇文章主要讲述，在搭建一个RAG项目时，所需要处理的逻辑流程，以及对每一个所用到的知识的大致讲述，适合大模型应用开发的新手思路搭建与流程规划

一：基础名词讲解

LLM

LLM（Large Language Model）：大预言模型

LLM 的核心能力就 4 样：

理解人类语言你说 “帮我解释一下 RAG”，它能听懂你要什么。
生成人类语言写文案、写邮件、写代码、写故事、聊天。
逻辑推理简单数学、总结内容、提取重点、分析问题。
按照指令做事你让它 “简洁回答”“专业一点”“用表格整理”，它都能照做。

它是怎么学会的？

不需要人教它具体知识，只做两步：

喂给它海量文本网页、书籍、文章、代码、百科……
训练它预测下一个字比如：“今天天气很__”LLM 学会填：“好”“热”“冷” 等。

通过万亿次这种预测，它慢慢就懂了语法、逻辑、常识、世界知识。、

训练方式有：

ZeroShot（不用给任何案例，靠模型本身能力，简单任务能用，复杂任务容易错）

FewShot（给 3～5 个例子，模型立刻学会格式、规则，效果远强于 ZeroShot）

大模型的训练涉及机器学习、深度学习等方面的知识，也需要一定的数学基础，但是在大模型应用开发初期，对于模型的训练，我们并不过多关注，我们只需要知道，它是干什么的就行

RAG

RAG（Retrieval-Augmented Generation）：检索增强生成

让大模型（LLM）先去查外部资料，再根据资料回答，而不是只靠自己瞎编

上边我们说到，大模型是基于数据训练出来的，但是大模型训练出来之后，一些比较新的信息和数据是它所不知道的，比如说，如果一个大模型是2025.5训练出来的，你问它2025.5之后的信息，它肯定是不知道的，就会瞎编

LLM具有两个致命缺点：

知识过时（比如不知道你公司最新规则、最新文档）
幻觉（喜欢胡说八道、乱编答案）

RAG 就是专门解决这两个问题的

有了RAG，如果你的问题大模型不知道，就可以从外部的文档等知识去获取

Embedding

把文字变成一串数字（向量）让计算机能读懂语义、计算相似度。

计算机看不懂文字，但计算机能算两个数字数组的距离

距离越近 → 语义越相似

比如：

“苹果” 和 “水果” 距离近
“苹果” 和 “汽车” 距离远

这就是语义检索的基础

有了这三个基础概念，我们就可以来总结一下RAG项目开发的流程：

1：用户准备一些资料，把这些资料通过Embedding转成向量组，然后存入向量数据库（构建RAG）

2：用户输入问题，并痛过Embeddng转成向量

3：计算问题向量与向量数据库中哪些向量离的更近（计算余弦相似度），离的更近的就是有关问题的资料

4：把找到的资料与问题一块丢给大模型，得到最后的结果

二：真实开发

现在常用的开发技术有很多，最基础的就是通过HTTP或HTTPS请求，在代码中直接通过请求来响应大模型，同样也有一些开发框架，Python的LangChain，Java的LangChain4j，两者的开发RAG流程都相同，我们在这里主要讲解思路，因此我们使用LangChain里的一些方法，但是学会了这个流程之后，使用LangChain4j也是手到擒来

注意：对于下边的一些类名和方法名，我会进行简单的叙述，不了解的不需要过多去关注，只需要知道它的功能即可

除此之外，我们配置的是一个聊天型的配置RAG的LLM，如果要配置其他类型的，流程相同，但方法有些不同

我们按照真实开发流程进行讲解

1：上传资料并存储

这里就是IO操作

拿到上传的文件之后，我们首先将文件中的内容，转换为md5格式（向量数据库中存储的都是md5格式的数据，因为md5转换后都是32位十六进制的字符串，方便我们后续判断），然后检查上传的文件是否在向量库中已经存在（建立一个文件，每上传一个新的向量，记录一下这个向量的内容的md5的值），就是比较转换后的md5字符串是否在向量库中存在，如果存在，那就不用重新上传了；如果不存在则进行下一步

现在文件已经上传，并且已经判断出向量库中没有存储（md5仅仅判断文件中的内容是否已经在向量库中），现在我们要对文件进行拆分，因为LLM 一次看不完特别长的文章，向量库检索也需要短文本，太长的内容会丢失精度，我们此时可以使用RecursiveCharacterTextSplitter进行文本划分

然后我们就得到了划分之后的文本内容，内容处理完毕之后，我们还需要定义metadata（元数据，记录数据的操作者，操作时间或一些其他信息），然后我们就可以把划分后的文本内容和元数据都交给Chroma进行存储啦

Chroma（专门给 RAG 用的轻量级向量数据库），我们上边所提到的文本向量库，在Python中就可以使用Chroma来创建，因为在创建Chroma时，提供选项去选择Embedding模型，对的，你没看错，把文本转化为Embedding也需要一个专门的大模型去做，因此我们在创建Chroma时，指定了Embedding模型，我们直接把数据交给Chroma时，就完成了文本数据的向量化以及存储

2：配置查询

我们在上边已经建立好的向量数据库，在这里，我们将要配置对向量库的查询

我们会新建立一个Chroma（Chroma语法规定，如果在创建时，所规定的文件夹不存在，则就是创建；如果文件夹已经存在，就是建立查询），因此我们在这里创建查询，在后续，我们就可以通过这个新创建的Chroma去查询旧的Chroma中的内容

查询创立之后，我们再通过Chroma中的一个方法，获得retriever，retriever就是用户检索Chroma的入口，想要查询什么内容，就把这些内容传给retriever，就能够得到查询的结果

3：搭建链

在上边两步中，我们创建了文本向量库，并配置了对向量库的查询，我们下边将要配置大模型与提示词

对于大模型和提示词的配置，因为根据用途的不同，以及提示词方式的不同，有很多种大模型的配置方法和提示词的书写，因此我就不再说明具体的类，大家可以去学习一下，也不难

然后就是链的搭建了，我们叫做chain

比如现在chain = A | B | C

那么我通过chain可以传递一个参数进去，这个参数，经过A处理后的结果，作为B的输入进入B，再经过B处理之后，作为C的输入进入C被C处理

因此我们在一般这样调用：chain = prompt | model

我们把我们的问题，先传给prompt提示词，根据提示词封装信息模板，然后把这些信息给大模型处理

如果使用了RAG，那么链一般这样搭建：

chain = { RunnablePassthrough() | retriever } | prompt | model

RunnablePassthrough就是个 “传送带”，数据进来什么样，出去就什么样，原封不动传给下一环

retriever就是我们上边封装的对于rag的查询

然后我们将用户输入的问题+rag检索后的资料 -> prompt，然后再传给model

这样就完成了整条链的搭建

4：保存历史记录

在上边，我们搭建好了一个完整的链，用户能够问问题，rag也能通过用户的问题，在搭建好的文本向量库中搜索资料，并把用户的问题和资料结合起来，发给model，去得到最终的结果

但是还有一个很重要的问题，用户与model的聊天记录没有存储下来，他们的对话是"一次性的"，因此我们在这里解决历史记录的问题

历史记录可以保存在内存中，也可以保存在文件中

我们一般更推荐保存在文件中，因为内存有大小限制，而且关机就没了

如果需要用内存保存，就可以使用InMemoryChatMessageHistory类，进行不断存储就行

如果需要使用文件保存，那就需要自己定义一个类

def get_message(session_id):
    return FileChatMessageHistory(session_id,"./chat_history")
class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id, storage_path):
        self.session_id = session_id
        self.storage_path = storage_path
        self.file_path = os.path.join(self.storage_path, self.session_id)
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        all_messages = list(self.messages)
        all_messages.extend(messages)
        new_messages = [message_to_dict(message) for message in all_messages]
        with open(self.file_path, "w",encoding="utf-8") as f:
            json.dump(new_messages, f)

    @property
    def messages(self) -> list[BaseMessage]:
        try:
            with open(self.file_path, "r",encoding="utf-8") as f:
                messages_data = json.load(f)
                return messages_from_dict(messages_data)
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path, "w",encoding="utf-8") as f:
            json.dump([], f)

在这里我需要说明，所有的聊天对话以及记录都是BaseMessage类型的，因此在代码中做了很多类型的转换

5：带有历史记录的链

在完成了3、4这两步之后，我们将它们两个进行结合

        conversation_chain = RunnableWithMessageHistory(
            chain,
            get_message,
            input_messages_key="input",
            history_messages_key="history",
        )
        return conversation_chain

这样我们就得到了带有历史记录的chain

后续只需要通过chain去传输问题即可

三：总结

本篇文章主要讲了一个基础RAG项目的搭建流程，在其中，还存在很多的问题，比如，在链中传递参数时，参数每经过一次处理得到的数据的格式，可能与链的下一个处理的输入格式不同，在构建链时，经常需要去处理这些数据格式，除此之外，还有多用户的使用等等一些问题，不过这些问题都是围绕着这一套整体的流程去做的，你学会了整体的流程，其他的问题都是一块小的绊脚石，很轻易就能跨过去

希望这篇文章对你有所帮助