纯手撸一个RAG
- RAG基本流程
- 第一阶段:数据预处理(索引) - 构建知识库
- 第二阶段:查询与生成(推理) - 回答问题
- 总结
- Chunk介绍
- Chunk框架的介绍
- Chunk核心概念
- 选择分块策略和框架
- 如何选择分块框架
- Python代码实现
- 第一步:文档准备
- 第二步:chunk代码
- 第二步:Embedding--我这里选择的千问Embedding模型
- 第三步:将Embedding之后的数据存入向量数据库中--这里选择的向量数据库为 chromadb
- 第四步:执行一次搜索【孙悟空的金箍棒是从哪里得来的?】--这里的LLM模型选择的也是千问模型
- embed.py文件的完整代码
- 第五步:可视化查看向量数据库
- 结束语
RAG基本流程
整个流程可以解读为:先用知识库训练一个“超级大脑”,当用户提问时,这个大脑会先从自己的知识库里精准地找到相关答案片段,然后再组织语言给出最终回答。
下面我将为您详细解释图中每个步骤的含义。
第一阶段:数据预处理(索引) - 构建知识库
这个阶段是“备课”的过程,通常离线进行,目的是将您的知识文档(如公司内部文档、产品手册等)处理成系统可以快速查询的格式。
- Long Text (长文本)
- 含义:这就是您的原始知识库来源,可以是PDF、Word、TXT、网页等任何格式的长篇文档。
- 类比:就像一本完整的教科书。
- Chunk (分块)
- 含义:由于LLM的上下文窗口有限,并且为了提高检索精度,需要将长文本切割成更小的、语义相对完整的片段。这个过程就是我们之前讨论的 Text Chunking。
- 方法:可以使用固定大小、按句子/段落,或更高级的语义分块等方法。
- 类比:把教科书中的每一章、每一节分解成一个个重点明确的段落或知识点卡片。
- Text Pieces (文本片段)
- 含义:分块后产生的具体文本片段,也就是一个个“Chunk”。
- 类比:上一步制作好的知识点卡片。
- Embedding (嵌入)
- 含义:使用嵌入模型将这些文本片段转换成向量。向量是一串数字,在高维空间中表示文本的语义信息。语义相近的文本,其向量在空间中的距离也更近。
- 目的:将文本转换为计算机可以理解和计算的数学形式。
- 类比:为每一张知识点卡片生成一个独一无二的“数字指纹”,这个指纹编码了卡片上的所有知识。
- Vector DB (向量数据库)
- 含义:将这些生成的向量(以及对应的原始文本片段)存储到一个专门的数据库中,这种数据库擅长高效地存储和检索向量数据。
- 目的:为后续的相似性搜索做好准备。
- 类比:将一个装满所有“数字指纹”和对应“知识点卡片”的智能档案柜归档完毕。
第二阶段:查询与生成(推理) - 回答问题
这个阶段是“考试”或“答疑”的过程,在线实时进行。当用户提出一个问题时,系统执行以下步骤来生成答案。
-
Query / Prompt (用户查询/问题)
- 含义:用户输入的问题或指令。例如:“我们公司的休假政策是怎样的?”
- 类比:学生提出的一个问题。
-
Embedding (嵌入查询)
- 含义:使用同一个嵌入模型将用户的问题也转换为一个向量。
- 目的:让计算机能够理解用户问题的“数字指纹”是什么样的。
- 类比:为学生的这个问题也生成一个“数字指纹”。
-
Context (Vector) (检索上下文)
- 含义:系统拿着用户问题的“向量”,去向量数据库中进行相似性搜索。它会寻找与问题向量最相近的那些向量(即语义最相关的文本片段)。
- 结果:系统检索出最相关的几个“文本片段”。
- 类比:拿着问题的“指纹”去智能档案柜里查找,找出指纹最匹配的几张“知识点卡片”。
-
Prompt + Context (组合提示词)
-
含义:将用户原始的问题和从向量数据库中检索到的相关上下文组合成一个新的、更丰富的提示词。
-
模板示例:
请根据以下背景信息回答问题。
背景信息:[这里插入检索到的相关文本片段]
问题:[这里插入用户的原问题]
回答: -
类比:学生拿到了与问题最相关的几张知识点卡片作为参考,然后开始组织答案。
-
-
LLM (大语言模型)
- 含义:将组合好的新提示词发送给大语言模型(如GPT-4)。
- 过程:LLM的核心任务不再是依靠自己的内部知识,而是基于提供的“背景信息” 来生成一个精准、可靠的答案。
- 优点:避免了LLM的幻觉问题,答案更具事实性,并且可以引用最新的、私有的知识。
- 类比:学生参考着找到的知识点卡片,写出一份准确、完整的答案。
-
Response / Answer (最终响应)
- 含义:LLM生成的最终答案返回给用户。这个答案既结合了LLM强大的语言组织和推理能力,又扎根于您提供的真实数据。
- 类比:学生提交了最终答卷。
总结
这张图清晰地展示了RAG如何将信息检索 和文本生成 完美结合:
- 左边(预处理):是知识灌输的过程,把外部知识结构化地“教”给系统。
- 右边(查询):是知识应用的过程,系统利用学到的知识来精准地回答用户问题。
Chunk介绍
RAG(Retrieval-Augmented Generation)系统中的 chunk(文本分块/片段)是构建高效检索的基础。它将大型文档分解为更小、更易管理的语义单元,以便后续的向量化、索引和检索。合理的分块策略能显著影响知识检索的准确性和生成答案的质量。
Chunk框架的介绍
框架名称 | 主要特点 | 适用场景 | 支持的分块策略 | 轻量/易用性 |
---|---|---|---|---|
LangChain | 生态成熟,功能全面,社区活跃 | 通用RAG应用,快速原型开发 | 固定大小、递归字符、语义1 | 中等 |
Chonkie | 极速、轻量(核心仅9.7MB)2 | 高性能要求,轻量化部署 | Token、句子、语义、SDPM8 | ✅ |
RAGFlow | 深度文档理解,智能解析表格/公式4 | 企业级知识库,复杂格式文档 | 智能布局分析(含多模态)4 | 需Docker |
LightRAG | 模块化设计,支持自定义分块策略10 | 研究、定制化场景 | 可按字符、语义或混合策略扩展10 | 高灵活性 |
Chunk核心概念
在RAG中,Chunk 是指将长文档切割成的、带有一定语义完整性且大小可控的文本片段。这个过程之所以关键,是因为它直接 bridge 了原始文档和LLM的有效处理能力。
- 为什么需要分块?
- 适应模型上下文窗口:LLM有token处理上限,必须将长文本切块。
- 提升检索精度:过大的块会包含无关信息(噪声),降低检索准确性;过小的块可能上下文不足,影响LLM理解。合适的块能让检索系统更精准地定位到与问题最相关的信息。
- 平衡效率与成本:小块文本的向量化计算和索引更高效,存储和检索成本也更低。
- 分块的核心原则
- 保持语义完整性:理想的分块应尽可能保持一个完整的语义单元(如一个概念、一段论证),避免在句子中间或意群中断开。常用的分隔符包括段落(
\n\n
)、换行符(\n
)、句号(。
)等。 - 设置重叠区域:相邻的块之间保留一部分重叠文本(例如100-300字符),有助于保持上下文的连续性,防止关键信息因被切割在两个块的边界而丢失。
- 选择合适的大小:没有 universally 的最佳大小,需根据文档类型、模型能力和具体任务试验。一个常见的起始参考范围是 256~768个tokens。
- 保持语义完整性:理想的分块应尽可能保持一个完整的语义单元(如一个概念、一段论证),避免在句子中间或意群中断开。常用的分隔符包括段落(
选择分块策略和框架
你可以根据文档特点和项目需求,参考以下策略进行选择。
- 固定大小分块 (Fixed-size Chunking):
最简单直接的方法,按固定字符数或token数切割。优点是简单快速;缺点是可能粗暴地破坏文本语义结构。
# 示例:LangChain 的固定大小分块from langchain.text_splitter import CharacterTextSplittertext_splitter = CharacterTextSplitter(chunk_size=1000,chunk_overlap=200,separator="\n")docs = text_splitter.split_text(your_text)
-
递归字符分块 (Recursive Character Text Splitting):
LangChain 默认推荐的方法。它优先尝试用一组分隔符(如["\n\n", "\n", "。", " ", ""]
)递归地分割文本,直到块的大小符合要求。这种方法比固定分块更能尊重文本的天然结构。 -
语义分块 (Semantic Chunking):
利用嵌入模型计算句子或段落间的语义相似度,在语义变化边界处进行切割。优点是能更好地保持语义连贯性;缺点是计算开销稍大。
# 示例:使用 Chonkie 进行语义分块from chonkie import SemanticChunkerchunker = SemanticChunker(embedding_model="all-minilm-l6-v2", max_chunk_size=512, similarity_threshold=0.7)chunks = chunker.chunk(your_text)
-
专用分块 (Specialized Chunking):
对于代码、论文、PPT等高度结构化的文档,通用分块策略往往效果不佳。- RAGFlow 这类框架的价值在此凸显,它能深度解析文档布局,对表格、公式、多栏排版进行结构化提取和智能分块,最大限度保留原始信息的完整性。
如何选择分块框架
选择时,可以重点考虑以下几点:
- 文档类型与复杂度:如果主要处理纯文本文档(TXT, MD),LangChain 或 Chonkie 足够。如果需要处理PDF、PPT、Excel等复杂格式,并需要处理其中的表格、图片,RAGFlow 的深度解析能力更为合适。
- 性能与资源要求:若对速度和资源占用非常敏感(如服务器less环境),Chonkie 以其轻量和高效著称28。若需要高度定制化的分块逻辑(例如为特定领域优化),LightRAG 的模块化设计提供了更大灵活性10。
- 开发与部署成本:LangChain 生态系统庞大,社区支持好,适合快速开发和验证想法。RAGFlow 提供开箱即用的体验,包括可视化界面,更适合部署企业级应用4。
💡 实践建议
- 没有银弹:最佳分块策略高度依赖于你的具体数据、查询类型和模型。强烈建议进行实验和评估。
- 重叠是关键:不要忽视
chunk_overlap
参数。适当的重叠(通常为块大小的10%-20%)能有效防止边界效应。 - 评估维度:可以从检索精度(Recall@K, MRR)、生成答案质量(人工评估或LLM评估)以及系统延迟等维度综合评估分块效果。
Python代码实现
废话不多说,直接上代码
第一步:文档准备
西游记全文:此处省略N个字
📖保存为:data.md文件
第二步:chunk代码
from langchain.text_splitter import RecursiveCharacterTextSplitter
def read_data() -> str:with open("data.md", "r", encoding="utf-8") as f:return f.read()
def get_chunks() -> list[str]:# 1️⃣ 定义示例文本(可替换为你自己的内容)# text = """# RAG(Retrieval-Augmented Generation)是将外部知识与大语言模型结合的一种技术方式,# 通过“先检索、再生成”的流程,让模型能结合知识库回答问题。# 而文本切分,就是其中的关键第一步。# """text = read_data()# 2️⃣ 初始化分块器(推荐配置)text_splitter = RecursiveCharacterTextSplitter(separators=["\n\n", "\n", "。"], # 语义感知分段,自定义分割符chunk_size=1000, ## 最大长度,每段最大长度(字符数)chunk_overlap=200 ## 重叠长度,相邻 chunk 的重叠长度)# 3️⃣ 执行分块chunks = text_splitter.split_text(text)result = []# 4️⃣ 输出查看:前几个 chunk 结果# print(f"总共分成 {len(chunks)} 块:\n")for i, chunk in enumerate(chunks):result.append(f"{chunk}")# print(f"第 {i+1} 块内容:\n{chunk}\n{'-'*30}")return resultif __name__ == '__main__':chunks = get_chunks()for c in chunks:print(c)print("--------------")
📖 保存为:chunkLangChain.py文件
第二步:Embedding–我这里选择的千问Embedding模型
from openai import OpenAIdef embed(text: str) -> list[float]:client = OpenAI(api_key="API Key", # 如果您没有配置环境变量,请在此处用您的API Key进行替换base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # 百炼服务的base_url)completion = client.embeddings.create(model="text-embedding-v4",input=text,dimensions=1024,# 指定向量维度(仅 text-embedding-v3及 text-embedding-v4支持该参数)encoding_format="float")assert completion.data[0]assert completion.data[0].embeddingreturn completion.data[0].embedding
📖 保存为embed.py文件
第三步:将Embedding之后的数据存入向量数据库中–这里选择的向量数据库为 chromadb
## 代码写在embed.py文件中
import chunkLangChain
import chromadbchromadb_client = chromadb.PersistentClient("./chromaDb")
chromadb_collection = chromadb_client.get_or_create_collection("joker")
def create_db() -> None:for idx, c in enumerate(chunkLangChain.get_chunks()):print(f"Process: {c}")embedding = embed(c)chromadb_collection.upsert(ids=str(idx),documents=c,embeddings=embedding)
if __name__ == '__main__':create_db() # 只需要处理一次
第四步:执行一次搜索【孙悟空的金箍棒是从哪里得来的?】–这里的LLM模型选择的也是千问模型
def query_db(question: str) -> list[str]:question_embedding = embed(question)result = chromadb_collection.query(query_embeddings=question_embedding,n_results=5)assert result["documents"]return result["documents"][0]def llm_query(content:str):client = OpenAI(api_key="API Key", # 如果您没有配置环境变量,请在此处用您的API Key进行替换base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # 百炼服务的base_url)completion = client.chat.completions.create(model="qwen-plus",messages=[{"role": "system", "content": "你是一个全能的助手!"},{"role": "user", "content": content},],# Qwen3模型通过enable_thinking参数控制思考过程(开源版默认True,商业版默认False)# 使用Qwen3开源版模型时,若未启用流式输出,请将下行取消注释,否则会报错# extra_body={"enable_thinking": False},)assert completion.choicesassert completion.choices[0]assert completion.choices[0].messageassert completion.choices[0].message.contentreturn completion.choices[0].message.contentif __name__ == '__main__':# create_db() # 只需要处理一次question = "孙悟空的如意金箍棒是从哪里得来的?"chunks = query_db(question)prompt = "请根据上下文回答用户的问题\n"prompt += f"Question: {question}\n"prompt += "Context:\n"for c in chunks:prompt += f"{c}\n"print(prompt)# 将向量数据库中找到的片段统一丢给LLM;之后统一返回前端print(llm_query(prompt))
embed.py文件的完整代码
from openai import OpenAI
import chunkLangChain
import chromadbchromadb_client = chromadb.PersistentClient("./chromaDb")
chromadb_collection = chromadb_client.get_or_create_collection("joker")def embed(text: str) -> list[float]:client = OpenAI(api_key="API Key", # 如果您没有配置环境变量,请在此处用您的API Key进行替换base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # 百炼服务的base_url)completion = client.embeddings.create(model="text-embedding-v4",input=text,dimensions=1024,# 指定向量维度(仅 text-embedding-v3及 text-embedding-v4支持该参数)encoding_format="float")assert completion.data[0]assert completion.data[0].embeddingreturn completion.data[0].embeddingdef create_db() -> None:for idx, c in enumerate(chunkLangChain.get_chunks()):print(f"Process: {c}")embedding = embed(c)chromadb_collection.upsert(ids=str(idx),documents=c,embeddings=embedding)def query_db(question: str) -> list[str]:question_embedding = embed(question)result = chromadb_collection.query(query_embeddings=question_embedding,n_results=5)assert result["documents"]return result["documents"][0]def llm_query(content:str):client = OpenAI(api_key="API Key", # 如果您没有配置环境变量,请在此处用您的API Key进行替换base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # 百炼服务的base_url)completion = client.chat.completions.create(model="qwen-plus",messages=[{"role": "system", "content": "你是一个全能的助手!"},{"role": "user", "content": content},],# Qwen3模型通过enable_thinking参数控制思考过程(开源版默认True,商业版默认False)# 使用Qwen3开源版模型时,若未启用流式输出,请将下行取消注释,否则会报错# extra_body={"enable_thinking": False},)assert completion.choicesassert completion.choices[0]assert completion.choices[0].messageassert completion.choices[0].message.contentreturn completion.choices[0].message.contentif __name__ == '__main__':# create_db() # 只需要处理一次question = "孙悟空的如意金箍棒是从哪里得来的??"chunks = query_db(question)prompt = "请根据上下文回答用户的问题\n"prompt += f"Question: {question}\n"prompt += "Context:\n"for c in chunks:prompt += f"{c}\n"print(prompt)# 将向量数据库中找到的片段统一丢给LLM;之后统一返回前端print(llm_query(prompt))
第五步:可视化查看向量数据库
import chromadb
import umap
import plotly.express as px
import pandas as pd
import numpy as np# 连接到Chroma数据库
client = chromadb.PersistentClient(path="./chromaDb") # 替换为您的数据库路径# 获取集合列表
collections = client.list_collections()
print("可用集合:", [col.name for col in collections])# 选择要可视化的集合
collection_name = "joker2" # 替换为您的集合名称
collection = client.get_collection(collection_name)# 获取所有向量和对应的元数据
results = collection.get(include=["embeddings", "metadatas", "documents"])
# 提取向量和元数据
embeddings = results["embeddings"]
metadatas = results["metadatas"]
documents = results["documents"]print(f"检索到 {len(embeddings)} 个向量")# 使用UMAP进行降维
reducer = umap.UMAP(n_components=3, random_state=42)
embedding_3d = reducer.fit_transform(embeddings)# 准备可视化数据
df = pd.DataFrame({'x': embedding_3d[:, 0],'y': embedding_3d[:, 1],'z': embedding_3d[:, 2],
})
# print(metadatas)
# 添加元数据(如果有) 因为我们在存储时,这部分信息没有存储,这部分可以注释掉
# if metadatas and len(metadatas) > 0:
# print(metadatas[0])
# for key in metadatas[0].keys():
# df[key] = [meta.get(key, "") for meta in metadatas]# 添加文档文本(如果有)
if documents and len(documents) > 0:df['document'] = documents# 创建3D散点图
fig = px.scatter_3d(df,x='x',y='y',z='z',title=f"Chroma集合 '{collection_name}' 的向量可视化",hover_data=df.columns.tolist() # 悬停时显示所有信息
)# 更新布局
fig.update_layout(scene=dict(xaxis_title='UMAP 1',yaxis_title='UMAP 2',zaxis_title='UMAP 3')
)# 显示图形
fig.show()
结束语
- 结合前端,可以将文件做成可视化上传;
- 结合不同用户,设置不同的数据库,将不同用户上传的文件进行embedding向量化存储;
- 根据不同的用户或不同的数据库,设置知识库管理系统;
- 市面上可供使用的轮子:RAGFlow等。