10 节 / 共 12

4.1 系统架构与数据准备

📌 本节目标

学完本节,你将能够绘制一个简单的 RAG 系统架构图,并理解每个组件的作用。同时,你将学会如何收集和预处理一个小型文档集,为后续的向量化与检索环节打下坚实基础。

1. 核心概念讲解

在搭建一个 RAG 系统之前,我们需要先搞清楚它的“骨架”——也就是系统架构。RAG 的核心理念是“检索 + 生成”,因此它的架构也围绕这两个阶段展开。

1.1 什么是 RAG 系统架构?

简单来说,RAG 系统架构描述的是数据如何流动、各个模块如何协作。一个标准的 RAG 系统包含以下几个核心模块:

  • 数据源:你的知识库所在地,比如本地文件夹、数据库或网页。
  • 文档加载器:负责读取不同格式(如 PDF、TXT、Markdown)的文档。
  • 文本分割器:将长文档切分成更小的“块”(Chunks),方便后续检索。
  • 嵌入模型:将文本块转换成计算机能理解的数值向量(Embeddings)。
  • 向量数据库:专门存储和检索这些向量的数据库。
  • 大语言模型(LLM):根据检索到的上下文生成最终答案。

比如:想象你有一个巨大的图书馆(数据源)。你想知道“如何做番茄炒蛋”。如果直接问图书管理员(LLM),他可能记不住所有书的内容。RAG 的做法是:先让一个助手(检索器)去图书馆快速找到关于“番茄炒蛋”的几页书(文档块),然后把这些内容递给管理员,管理员基于这些内容给出精准回答。

1.2 数据准备的核心步骤

数据准备是 RAG 系统中最基础也最容易被忽视的一环。它主要包括三个步骤:

  1. 收集:从各种来源获取原始文档。
  2. 清洗:去除无关信息,比如 HTML 标签、多余的空格、页眉页脚。
  3. 分割:将长文本切分成语义完整的短文本块。

比如:你从网上下载了一篇 50 页的 PDF 技术手册。直接把它扔给 LLM 是不现实的,因为 LLM 的上下文窗口有限。你需要先提取出纯文本,然后按章节或段落切分成每块 500 字左右的片段。

1.3 代码示例:使用 Python 进行基础文本分割

下面的代码展示了如何使用 LangChain 库中的 RecursiveCharacterTextSplitter 进行文本分割。这是目前最常用且效果较好的分割方式。

# 首先,确保你已经安装了 langchain 库
# pip install langchain

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 1. 准备一段示例文本
sample_text = """
人工智能(Artificial Intelligence,简称 AI)是计算机科学的一个分支。
它企图了解智能的实质,并生产出一种新的能以人类智能相似的方式做出反应的智能机器。
该领域的研究包括机器人、语言识别、图像识别、自然语言处理和专家系统等。

自从人工智能诞生以来,理论和技术日益成熟,应用领域也不断扩大。
可以设想,未来人工智能带来的科技产品,将会是人类智慧的“容器”。
人工智能可以对人的意识、思维的信息过程的模拟。
人工智能不是人的智能,但能像人那样思考、也可能超过人的智能。
"""

# 2. 初始化分割器
# chunk_size: 每个文本块的最大字符数
# chunk_overlap: 相邻块之间的重叠字符数,用于保持上下文连贯
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=100,      # 每个块最多100个字符
    chunk_overlap=20,    # 块与块之间重叠20个字符
    length_function=len, # 使用 len 函数计算字符长度
    separators=["\n\n", "\n", "。", "!", "?", ","] # 优先按段落、句子分割
)

# 3. 执行分割
chunks = text_splitter.split_text(sample_text)

# 4. 打印分割结果
print(f"原始文本长度: {len(sample_text)} 字符")
print(f"分割后得到 {len(chunks)} 个文本块\n")
for i, chunk in enumerate(chunks):
    print(f"--- 块 {i+1} (长度: {len(chunk)} 字符) ---")
    print(chunk)
    print()

代码解释

  • RecursiveCharacterTextSplitter 会递归地尝试用 separators 列表中的分隔符进行分割,优先保证语义完整性。
  • chunk_overlap 参数非常关键,它能防止关键信息正好被切分在边界上而丢失。

2. 深入理解

2.1 为什么需要精心准备数据?

RAG 系统的效果上限,很大程度上取决于数据准备的质量。一个经典的说法是:“垃圾进,垃圾出”(Garbage In, Garbage Out)。如果数据本身包含大量噪声或分割不合理,即使最强的 LLM 也无法给出好结果。

  • 分割粒度的权衡
    • 块太大:可能包含过多无关信息,降低检索精度,且浪费 LLM 的上下文窗口。
    • 块太小:可能丢失上下文语义,导致检索到的信息不完整。
  • 常见的误区
    • 忽视清洗:直接使用含有乱码、标记符号的文本。
    • 统一分割:所有文档都用相同的 chunk_size,没考虑文档结构差异(如代码 vs. 散文)。
    • 忽略重叠:没有设置 chunk_overlap,导致跨段落的逻辑被切断。

2.2 不同分割策略对比

| 策略 | 优点 | 缺点 | 适用场景 | | :--- | :--- | :--- | :--- | | 固定长度分割 | 实现简单,速度极快 | 严重破坏语义,容易切在句子中间 | 对语义要求极低的快速原型 | | 递归字符分割 | 平衡了速度和语义,能按自然边界切分 | 对非结构化文本效果一般 | 通用场景,最推荐 | | 语义分割 | 基于句子嵌入相似度,语义完整性最好 | 速度慢,需要额外模型,实现复杂 | 高质量问答系统,长文档处理 |

对于入门阶段,递归字符分割(如 LangChain 的 RecursiveCharacterTextSplitter)是最佳选择——它足够简单,效果也足够好。

3. 实战演示

现在,我们从头开始,构建一个完整的数据准备流程。我们将使用一个虚构的小型公司“明远科技”的员工手册作为数据源。

3.1 步骤一:收集原始数据

假设我们有一个名为 employee_handbook.txt 的文本文件,内容如下:

明远科技员工手册

第一章 总则
第一条 本手册旨在规范员工行为,维护公司正常运营秩序。
第二条 本手册适用于明远科技全体正式员工。

第二章 考勤制度
第三条 工作时间为周一至周五,上午9:00至下午18:00。
第四条 员工需通过公司OA系统进行打卡签到。
第五条 迟到或早退超过30分钟,需提交书面说明。

第三章 假期管理
第六条 年假:入职满一年后,每年享有5天带薪年假。
第七条 病假:需提供二级以上医院开具的证明。
第八条 事假:需提前一天向部门主管申请。

第四章 薪酬福利
第九条 每月15日发放上月工资。
第十条 公司为员工缴纳五险一金。

3.2 步骤二:编写数据加载与清洗函数

我们创建一个 Python 脚本 prepare_data.py

# prepare_data.py
import re

def load_document(file_path):
    """从文本文件加载文档"""
    with open(file_path, 'r', encoding='utf-8') as file:
        text = file.read()
    return text

def clean_text(text):
    """清洗文本:去除多余空白和特殊字符"""
    # 将多个换行符替换为单个换行符
    text = re.sub(r'\n+', '\n', text)
    # 去除行首行尾的空格
    text = '\n'.join([line.strip() for line in text.split('\n')])
    return text

def split_into_chapters(text):
    """按章节分割文档(这是一个简单示例,实际可用更复杂的逻辑)"""
    # 使用正则表达式匹配 "第X章" 作为分割点
    chapters = re.split(r'(第[一二三四五六七八九十]+章\s*\n)', text)
    # 重组:将标题和内容配对
    result = []
    for i in range(1, len(chapters), 2):
        title = chapters[i].strip()
        content = chapters[i+1].strip()
        result.append(f"{title}\n{content}")
    return result

# 主流程
if __name__ == "__main__":
    # 1. 加载
    raw_text = load_document("employee_handbook.txt")
    print("原始文本加载完成。")
    
    # 2. 清洗
    cleaned_text = clean_text(raw_text)
    print("文本清洗完成。")
    
    # 3. 按章节分割
    chapter_chunks = split_into_chapters(cleaned_text)
    
    print(f"共分割出 {len(chapter_chunks)} 个章节块:\n")
    for idx, chunk in enumerate(chapter_chunks, 1):
        print(f"--- 块 {idx} ---")
        print(chunk)
        print()

运行结果

原始文本加载完成。
文本清洗完成。
共分割出 4 个章节块:

--- 块 1 ---
第一章 总则
第一条 本手册旨在规范员工行为,维护公司正常运营秩序。
第二条 本手册适用于明远科技全体正式员工。

--- 块 2 ---
第二章 考勤制度
第三条 工作时间为周一至周五,上午9:00至下午18:00。
第四条 员工需通过公司OA系统进行打卡签到。
第五条 迟到或早退超过30分钟,需提交书面说明。

--- 块 3 ---
第三章 假期管理
第六条 年假:入职满一年后,每年享有5天带薪年假。
第七条 病假:需提供二级以上医院开具的证明。
第八条 事假:需提前一天向部门主管申请。

--- 块 4 ---
第四章 薪酬福利
第九条 每月15日发放上月工资。
第十条 公司为员工缴纳五险一金。

至此,我们已经成功地将一本员工手册按章节拆分成了 4 个语义完整的文本块,每个块都对应一个独立的知识主题。

4. 关键要点

  • 架构先行:在写任何代码之前,先在纸上画出 RAG 系统的数据流图,明确各模块职责。
  • 数据质量决定上限:花 70% 的时间在数据准备上,包括收集、清洗和分割。
  • 分割不是越细越好:根据文档类型(手册、代码、对话)选择合适的 chunk_sizechunk_overlap,一般 chunk_size 在 200-1000 字符间调整。
  • 善用现成工具RecursiveCharacterTextSplitter 是入门首选,它能在简单和效果之间取得良好平衡。
  • 保持迭代:没有完美的分割策略。先跑通流程,再根据检索效果反向调整参数。

✏️ 练习任务

练习方向:收集并预处理一个小型文档集

  1. 基础任务

    • 输入:从网络上下载或自己编写 3 篇不同主题的短文(每篇约 500-1000 字),主题可以是“Python 基础”、“健康饮食”和“旅游攻略”。
    • 操作
      1. 使用 Python 读取这 3 个文本文件。
      2. 编写一个简单的清洗函数,去除文档中的空行和特殊符号(如 #*)。
      3. 使用 RecursiveCharacterTextSplitter 将每篇文档分割成 chunk_size=300chunk_overlap=50 的文本块。
    • 期望输出:打印出所有分割后的文本块,并标注每个块来自哪篇原始文档(例如:来源:Python基础.txt - 块1)。
  2. 进阶挑战(选做)

    • 任务:你的文档集中有一篇是 Markdown 格式的笔记,含有标题(###)、列表(-)和代码块(```` ```)。请设计一个分割策略,优先按##` 标题进行分割,并且确保代码块不被截断。
    • 提示:你可以自定义 separators 参数,或者在分割后编写一个后处理函数来合并被截断的代码块。

💡 下一节我们将学习如何将准备好的文本块转换成向量,并存储到向量数据库中,从而实现真正的“检索”功能。敬请期待!


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