RAG 数据摄取流水线

RAG Data Ingestion Pipeline

引言

检索增强生成（Retrieval Augmented Generation，RAG）系统依赖外部文档知识库，为大语言模型的回答提供依据。本课讨论如何组织适合检索的知识库，并介绍 RAG 流水线所需的数据准备与预处理：清理原始数据，删除模板文字、导航链接等噪声；将文档分成有用的文本块；为这些文本块建立索引，以便高效开展语义搜索。

为什么数据准备对 RAG 至关重要？

RAG 系统的质量取决于知识库的质量。网站或数据转储中的原始数据经常包含大量噪声，例如 HTML 标签、页眉、页脚、重复导航菜单和广告。建立索引前，必须清理文档并整理格式，删除模板文字、元数据、页眉、页脚及其他信息价值较低的内容。例如，抓取网页时可能同时取得页首菜单和页脚版权声明，而这些信息与正文无关。清理工作直接影响检索质量。数据越干净、内容越集中，检索器越容易将用户问题与相关信息匹配起来。提取结果中的广告、导航链接和模板文字，都应在预处理阶段删除。

RAG 数据摄取流水线

图 1 · RAG数据准备流程
下载 SVG下载 SVG
RAG数据准备流程
原始数据依次经过摄取、清洗、分块、嵌入和索引，形成可供检索的数据。

本课重点介绍前两个关键步骤：数据摄取与数据清理。检索系统使用的文档通常分为结构化、半结构化和非结构化三类。理解三者之间的区别，是有效处理数据的基础。

1. 结构化文档

定义：按照预先定义的格式组织的数据，通常采用表格形式，便于直接查询和处理。

示例：

  • 数据库，例如 SQL 表
  • 电子表格，例如 Excel 文件、CSV 文件
  • 具有明确 schema 定义的 JSON 和 XML 文件

2. 半结构化文档

定义：包含部分结构化元素，但没有严格组织形式或完整 schema 的文档。

示例：

  • HTML 网页
  • 没有严格 schema 的 JSON 文档
  • 结构可变的 XML 文档

3. 非结构化文档

定义：没有预先定义结构的数据，通常难以自动处理，需要更先进的提取方法。

示例：

  • PDF 文档，例如研究论文和报告
  • DOCX 文件，例如信件和文章
  • 需要 OCR 的图像和扫描文档
  • 电子邮件和纯文本文档

文档类型直接影响 RAG 系统处理多模态数据的方式。结构化文档通常更容易直接进行 embedding 和检索；半结构化及非结构化文档往往需要提取文本、视觉或音频等多模态内容，并采用专门的方法处理，例如对图像使用光学字符识别（OCR）、对音频进行转写、为非结构化文本生成语义 embedding。正确识别文档类型，有助于选择合适的多模态 embedding 模型和处理工具，提高检索流水线的整体准确性与效果。

数据摄取

数据摄取是为检索系统准备数据的关键起点，包括收集、解析、组织来自不同来源的原始数据，并进行初步处理。

数据摄取的主要目标包括：

  • 高效收集多种格式、多个来源的原始数据。
  • 将收集的数据组织成适合处理的一致格式。
  • 可靠处理错误，确保数据完整性。

批量摄取与流式摄取：选择批量、流式或二者结合的方式，是重要的设计决策。批量摄取按计划周期收集和传输数据，例如每小时或每天执行 ETL 作业。在无需实时更新时，这种方式较简单，也适合高效处理大量数据，例如每晚汇总销售数据。常见工具包括调度器，以及 Apache Airflow、AWS Glue 等 ETL 框架。流式摄取以接近实时的方式持续接收数据，适合欺诈检测、用户分析数据流等对时效敏感的应用。流式流水线利用 Apache Kafka、Pulsar、Amazon Kinesis 或 Apache Flink 等工具，以较低延迟处理事件流。现代数据架构经常结合两种方式，例如通过 Kafka 流摄取用户交互事件，同时每天批量导入 CRM 数据。选择取决于数据变化速度，以及使用方对数据新鲜度的要求。

下图展示数据摄取各步骤。

图 2 · 可靠的数据摄取
下载 SVG下载 SVG
可靠的数据摄取
数据摄取结合内容解析、重试、幂等处理、容错、元数据和检查点，并由日志与监控反馈运行情况。

1. 解析与结构化

定义：将 PDF、HTML、图像、JSON 等原始数据源转换成有意义的结构化元素，例如文段、表格、标题、文本块和元数据。

推荐做法：

  • 保留来源 URL、时间戳、页码等元数据。
  • 保留标题、表格、文段和项目列表等语义结构。
  • 让不同格式的数据采用一致结构。

常见问题：

  • 只提取原始文本，丢失结构和上下文。
  • 未保留原始元数据，增加后续检索的困难。

推荐工具：

  • LlamaParse：输出结构化 Markdown，适合处理表格
  • Trafilatura：结构化 HTML 提取
  • PyMuPDF（fitz）：快速解析 PDF
  • Grobid：解析学术和科学论文 PDF

2. 重试机制

定义：数据获取任务因短暂故障失败后自动重试，例如网络波动或数据源暂时不可用。

推荐做法：

  • 采用指数退避，逐步增加重试间隔。
  • 限制重试次数，通常为 3～5 次，防止负载过高。
  • 清楚记录每次重试，便于了解运行情况和排查问题。

常见问题：

  • 没有重试逻辑，轻微故障便中断流水线。
  • 重试过多，导致数据源过载或摄取延迟增加。

3. 幂等处理

定义：同一摄取任务重复执行多次，仍然得到相同结果，防止产生重复数据。

推荐做法：

  • 通过 SHA-256、MD5 等内容哈希生成唯一标识符。
  • 再次处理文档前，始终使用其唯一哈希检查文档是否已处理。

常见问题：

  • 仅依赖时间戳或来源系统生成的 ID，而它们可能变化或重复。
  • 完全忽视幂等性，导致重复条目。

推荐工具：Python 内置的 hashlib。

import hashlib

def generate_hash(content):
    return hashlib.sha256(content.encode('utf-8')).hexdigest()

4. 容错与错误处理（DLQ）

定义：隔离有问题的数据，妥善处理摄取错误，使整个流水线能够继续运行。

推荐做法：

  • 设置 Dead-Letter Queue（DLQ），供人工检查有问题的文档。
  • 定期监控 DLQ 中重复出现的问题，持续改进解析和摄取逻辑。

常见问题：

  • 完全忽视 DLQ，损坏数据使流水线停止运行。
  • 没有定期检查 DLQ，错过持续改进的机会。

推荐工具：Kafka Streams、Redis Streams、简单的 Python 队列或数据库。

其概念可以用以下 Python 代码表示：

dead_letter_queue = []

try:
    process_document(document)
except Exception as e:
    dead_letter_queue.append({"document": document, "error": str(e)})

5. 元数据提取与检查点

定义：提取有意义的元数据，并持久记录摄取进度，以保证处理连续性。

推荐做法：

  • 记录来源 URL、文档时间戳、文档 ID 等关键元数据。
  • 定期保存摄取进度检查点，便于故障或中断后恢复。

常见问题：

  • 未定期保存检查点，重复处理已经摄取的文档。
  • 元数据不一致，影响下游检索准确性。

推荐工具：简单的检查点文件、Redis、PostgreSQL。

以下 Python 示例展示一种简单实现：

def save_checkpoint(last_document_id):
    with open("checkpoint.txt", "w") as f:
        f.write(last_document_id)

def load_checkpoint():
    try:
        with open("checkpoint.txt", "r") as f:
            return f.read().strip()
    except FileNotFoundError:
        return None

6. 日志与监控

定义：通过结构化日志记录流水线事件、错误和性能指标，并集中汇总，以便实时监控和告警。

推荐做法：

  • 使用 JSON 等结构化日志，记录时间戳、事件类型、文档 ID 和错误。
  • 采用 ELK Stack、Datadog、Splunk 等集中式日志管理系统。
  • 对严重问题和异常设置实时告警。

常见问题：

  • 日志缺少结构，增加错误分析和监控的复杂程度。
  • 日志没有集中管理，难以关联不同摄取事件。

推荐工具：

  • Python 内置 logging
  • ELK Stack、Datadog、Splunk：集中管理日志
  • Prometheus、Grafana：监控与告警

反馈循环

1. 日志与监控 → 重试机制

出现指定告警或达到错误阈值时，日志系统自动触发重试。

2. DLQ → 日志与监控

记录并密切监控 DLQ 事件，识别反复出现的摄取问题，主动触发人工或自动干预。

数据摄取的推荐做法

设计良好的摄取流水线具有可扩展性和可靠性，并从起点维护数据质量。

1. 选择合适的摄取模式：根据应用对延迟和数据量的要求，选择批量、流式或混合摄取。批量摄取适合周期性加载，结构较简单、成本较低；流式摄取支持实时场景，同时增加复杂性。二者结合通常可以兼顾不同需求。

2. 尽早定义数据接口约定与 schema：像对待 API 一样对待数据接口。建立 schema 和数据接口约定，明确输入数据的字段、类型与格式。摄取时通过 Avro、Protobuf 或 JSON Schema 等定义实施验证，拒绝或标记不符合要求的记录。这能防止 schema 漂移破坏下游流水线，也使数据要求更加明确。Confluent Schema Registry 等工具可以管理 schema 版本及兼容性。

3. 保证幂等性并删除重复事件：分布式系统不可避免地出现重复数据，例如重试消息。摄取过程应具有幂等性，使同一记录处理两次也不会产生不良影响。尽早实现去重逻辑，例如为事件配置唯一 ID 并忽略重复项，或维护记录的哈希签名以识别重复数据。流式处理中，可以利用基于事件时间窗口的去重方法识别近似重复项。这可以防止重复计数影响指标和下游分析。

4. 配置可靠的错误处理（Dead-Letter Queue）：系统偶尔会收到错误数据或格式异常的数据。流水线应妥善处理这些问题。设置结构化错误处理，将问题记录发送到 DLQ，供后续检查，并保留相应信息。例如，JSON 消息无法解析或违反 schema 时，将其转入 S3 bucket、独立 Kafka topic 等形式的 DLQ，同时记录错误。这样，有效数据仍可继续摄取，也便于检查、修复或重新处理问题记录。

5. 统一时间戳和时区：时间不一致是数据集成中的常见问题。数据进入系统时，将所有时间戳统一为 UTC 等标准；有需要时，另外记录摄取时间。采用 ISO 8601 等一致格式，并在可用时保留来源时区，可防止合并不同地区或设备的数据时产生混淆。这样能够防止下游按时间聚合时出现错误，并支持跨系统关联数据。

6. 为扩展能力与背压做好设计：摄取系统必须同时应对稳定数据流与偶发数据激增。设计背压机制，使下游处理滞后时，上游能够平稳减速。可采用消息队列缓冲，通过 Kafka、AWS SQS 等使生产者与消费者解耦；根据负载自动扩展摄取计算资源；对来源设置速率限制或限流。按键值或分片等方式分区数据，可以通过并行处理提高吞吐量。目标是在流量突增时保持数据完整、系统正常运行。当负载超过容量时，设计良好的流水线会缓冲数据并重试。

7. 实现可观测性与监控：看不见的问题难以修复。为摄取流水线配置完整的日志与指标。重点监控吞吐量（每秒事件数）、滞后或延迟（最新摄取数据距今多久）、错误率（失败记录占比）以及队列长度和积压量。使用 Logstash/Elasticsearch 或云日志服务等集中式日志工具记录事件详情。如果流水线包含多个微服务，还可使用 OpenTelemetry 进行分布式追踪。通过 Prometheus/Grafana、Datadog 等建立监控面板，并为输入数据突然减少、错误激增等异常配置告警。可观测性让流水线运行情况更透明，减少停机时间，加快问题排查。

8. 规划 schema 演进：新增字段、格式调整等变化不可避免。摄取设计应能够在这些变化下继续运行。使用带版本的 schema，并保持向后兼容，例如让新增字段成为可选字段。将 schema 版本与数据一同保存，或存入注册中心，让使用方知道如何解析每条记录。依据预期版本验证输入数据，并拒绝或转换确实不兼容的记录。提前考虑变化，可以防止上游新增列或调整数据类型时造成流水线故障。结合 schema 注册中心的 Avro、Protobuf 等工具，可以通过兼容性模式提供支持。

9. 在摄取阶段保护数据：摄取层通常是敏感数据进入平台的第一个入口，应从这里实施安全与隐私措施。所有数据传输均使用 TLS，包括私有网络内部传输。验证数据源身份并实施授权，在入口处采用掩码或 tokenization 等技术，防止保存原始身份标识。记录访问情况，为数据进入过程保留审计信息，包括谁在何时发送了什么。保护摄取入口有助于防止泄漏，并从一开始满足 GDPR、HIPAA 等法规要求。

10. 使用真实数据与边界情况测试：部署前，利用贴近生产环境的场景对摄取流水线进行压力测试。回放经过敏感信息匿名化处理的真实生产数据，观察流水线如何应对实际事件模式和数据量。主动加入格式错误记录、缺少必填字段、超大载荷、乱序事件等边界情况。模拟下游服务中断等故障，确认重试和背压逻辑正常工作。使用峰值数据量进行负载测试，可以发现性能瓶颈并调整自动扩展。将这些测试纳入 CI/CD，自动验证每次代码变更是否符合数据质量规则，例如没有新增重复项、没有违反 schema。严格测试有助于确认系统能够在生产环境扩展，并应对未曾预料的问题。

数据清理

完成数据摄取后，下一个关键步骤是清理：修复或删除错误、不一致和噪声，提高数据质量。研究表明，数据科学家的大部分时间用于数据准备，包括清理与预处理。这有充分理由：干净的数据能够带来更准确的模型和分析结果。例如，重复记录可能误导算法，异常值可能影响数据分布。下图介绍数据清理过程；具体流程会随数据类型、数据量及其他条件变化。

图 3 · 清洗与质量检查流程
下载 SVG下载 SVG
清洗与质量检查流程
清洗流程结合文本、结构、OCR、敏感信息和跨模态检查，监控结果返回前面的质量处理步骤。

1. 初步质量评估

定义：尽早发现结构问题、噪声和低质量内容。

推荐做法：

  • 实施启发式检查和随机抽样。
  • 利用 LLM 快速评估，自动进行质量检查。

常见问题：

  • 省略或低估早期质量检查。
  • 仅依靠人工检查，造成延迟和错误。

推荐工具：自定义抽样脚本、基于 LLM 的自动评估。

2. 文本规范化

定义：统一和清理文本，提高可读性与一致性。

推荐做法：

  • 统一空白字符，删除 emoji，修复 Unicode 问题。
  • 兼顾文本可读性与内容上下文的保留。

常见问题：

  • 过度规范化导致重要上下文丢失。
  • 忽略不易察觉的 Unicode 问题。

推荐工具：regex、ftfy。

3. 结构整理

定义：删除页眉、页脚、模板文字等无关结构元素。

推荐做法：

  • 保持标题、文段等文档结构的完整性。
  • 对不同数据源一致应用处理规则。

常见问题：

  • 删除必要的上下文信息，使内容难以理解。
  • 结构处理方式不一致。

推荐工具：Trafilatura、BeautifulSoup。

4. OCR 噪声削减与纠正

定义：减少 OCR 错误，并过滤低置信度文本，提高 OCR 提取文本的质量。

推荐做法：

  • 采用基于置信度的过滤和拼写纠正。
  • 对于复杂文本问题，结合 LLM 辅助纠正。

常见问题：

  • 让噪声较多或不准确的 OCR 内容进入数据集。
  • 过滤过度，丢失有效内容。

推荐工具：SymSpell、LLM API。

5. 实体与敏感数据识别及掩码处理

定义：识别并遮蔽敏感个人信息，保证隐私与合规。

推荐做法：

  • 定期更新实体检测和掩码规则。
  • 持续验证掩码效果，防止隐私泄漏。

常见问题：

  • 遗漏敏感信息。
  • 掩码处理过度，导致数据丢失。

推荐工具：spaCy、Presidio、regex。

6. 跨模态一致性

定义：确保文本内容与关联的图像、表格等多模态元素相互对应、保持一致。

推荐做法：

  • 使用 CLIP 等多模态 embedding 验证准确性。
  • 定期验证跨模态一致性。

常见问题：

  • 忽略多模态一致性检查。
  • 没有验证便假定内容已自动对应。

推荐工具：CLIP、启发式验证脚本。

7. 去重与过滤

定义：删除重复或无关内容，维护数据集完整性。

推荐做法：

  • 结合精确匹配与语义去重方法。
  • 根据反馈循环持续改进过滤标准。

常见问题：

  • 仅依靠精确匹配进行去重。
  • 对重复内容缺少充分的语义分析。

推荐工具：hashlib、SentenceTransformers、LLM API。

8. 日志与监控

定义：系统追踪并记录数据清理过程，提高透明度与可靠性。

推荐做法：

  • 实施结构化日志，便于了解运行情况和排查问题。
  • 集中管理日志，支持有效的实时监控与告警。

常见问题：

  • 日志结构不佳，难以监控。
  • 日志分散，降低故障排查效率。

推荐工具：Python logging、ELK Stack、Datadog。

反馈循环

日志与监控 → 初步质量评估：持续监控能够迅速向初步质量检查提供反馈，根据观察到的流水线错误或质量下降调整标准。

日志与监控 → OCR 噪声削减：监控发现 OCR 准确率明显下降时，可以立即向 OCR 纠正环节反馈。

日志与监控 → 结构整理：及时发现并记录结构提取错误，提供反馈，改进解析工具和结构整理规则。

搜索课程

正在加载课程目录…