变更数据捕获用于实时RAG更新

对于大规模运行的检索增强生成 (RAG)系统，基础知识库的新鲜度不仅仅是“锦上添花”；它是回答质量和用户信任的基本决定因素。随着数据源的演变，新信息被添加，现有事实被更新，过时内容被移除，您的RAG系统必须及时反映这些变化。批量更新虽然实现起来更简单，但会引入明显延迟，导致基于陈旧或不完整信息的响应。变更数据捕获 (CDC) 为此提供了方案，使您的RAG系统能够准实时摄取和处理更新。

CDC 是一组软件设计模式，用于确定和追踪已更改的数据，以便可以利用这些更改的数据采取行动。CDC系统无需定期轮询整个数据集以检查修改，而是通常在底层监控源数据存储，捕获发生的单个更改事件（插入、更新、删除）。这种方法最大限度地减少了对源系统的影响，并允许下游消费者（例如您的RAG数据管道）以明显更低的延迟响应变化。

变更数据捕获的核心机制

实施CDC有多种方法，每种方法都有其权衡。对于大规模、对性能敏感的RAG系统，基于日志的CDC通常是首选方法。

1. 基于日志的CDC：大多数生产数据库维护事务日志（例如，PostgreSQL中的预写日志 (WAL)、MySQL中的二进制日志 (binlog) 或MongoDB中的oplog），用于复制、恢复和时间点还原。基于日志的CDC工具使用这些原生日志，非侵入性地读取已提交更改的流。此方法提供高保真度、对源数据库的开销低，并捕获所有更改，包括删除和模式修改。Debezium等工具在此方面很常见，为各种数据库提供连接器，可以将更改事件流式传输到Apache Kafka等平台。

2. 基于触发器的CDC：这种技术涉及在您希望监控的表上创建数据库触发器（例如，AFTER INSERT、AFTER UPDATE、AFTER DELETE）。这些触发器执行自定义代码，通常将有关更改的信息（例如，新行、旧行、操作类型）写入一个单独的“影子”或“审计”表。独立进程随后轮询此审计表。虽然对于简单场景而言，这很直接，但触发器可能会对源数据库造成明显的性能开销，尤其是在高写入负载下，因为它们与原始DML语句在同一事务中执行。

3. 基于查询的CDC（或基于时间戳的）：此方法依赖于源表中的列，这些列指示行上次修改时间（例如，last_updated_timestamp）或单调递增的版本号。CDC进程定期查询源表，查找自上次检查以来已更改的行。此方法通常对源数据库模式影响最小，但有几个缺点：它无法可靠地捕获删除（除非使用软删除），如果轮询不频繁，可能会遗漏中间更新，并给源系统带来重复的查询负载。对于要求低延迟和高准确性的系统，基于查询的CDC通常不足。

考虑到分布式RAG的可伸缩性、可靠性和准实时更新需求，基于日志的CDC（通常与流平台结合使用）提供了最坚实的基础。

为RAG更新管道设计CDC架构

将CDC集成到您的RAG数据管道中，涉及多个组件共同作用，以将更改从源系统传播到您的向量 (vector)数据库和文档存储。

在一个启用CDC的RAG更新管道中的数据流，从源数据库事务日志到向量数据库更新。

典型架构包括：

• 源数据库：这些是您的主要数据存储，包含需要输入RAG系统的信息。
• CDC代理/连接器：此组件（例如，在Kafka Connect中运行的Debezium连接器）追踪源数据库的事务日志，并将日志条目转换为结构化的更改事件。这些事件通常包含行的修改前和修改后状态（用于更新）、新状态（用于插入）或旧状态（用于删除），以及操作类型和模式等元数据。
• 消息代理/流处理平台：Apache Kafka是此处的常见选择。它作为更改事件的持久、可伸缩缓冲区。这使得CDC代理与下游消费者解耦，允许它们按自己的速度处理事件，并提供应对临时消费者不可用的弹性。
• 流处理器（可选但推荐）：Apache Flink、Spark Streaming或ksqlDB等系统可以消费Kafka中的事件。它们用于：
  • 过滤：忽略对不相关表或列的更改。
  • 转换：将事件负载转换为适合RAG管道的规范格式，或许与其他流连接进行丰富。
  • 路由：将不同类型的事件引导到不同的处理路径。
• RAG更新处理器：这是一个自定义应用程序或服务集，订阅（可能已处理的）更改事件流。其职责很主要：
  • 解析更改事件（例如，op: 'c'表示创建，op: 'u'表示更新，op: 'd'表示删除）。
  • 触发文档检索（如果只传递ID）或直接使用负载。
  • 调用新或更新内容的文档分块和预处理逻辑。
  • 协调使用您的嵌入 (embedding)模型对受影响文档块进行重新嵌入。
  • 向向量数据库发出适当的命令（插入、更新/upsert、删除）。
  • 如果您在向量索引旁边维护辅助文档存储，则可能更新该存储。

将更改传播到RAG组件

来自CDC管道的更改事件启动一系列操作，以更新您的RAG系统：

1. 文档预处理：

   • 创建 (c)：从事件负载中获取新文档数据（如果未完全包含），并通过您的标准分块和元数据提取管道进行处理。
   • 更新 (u)：RAG更新处理器必须确定更新的范围。如果一个小字段更改，它可能只影响一个块。如果文档的很大一部分被修改，则可能需要使多个现有块失效并生成新块。这通常需要检索完整文档，重新分块，然后与现有块进行比较，以在块级别识别修改、添加或删除。
   • 删除 (d)：必须移除与已删除文档ID相关联的所有块和相应的嵌入 (embedding)。

2. 嵌入生成：

   • 对于在预处理步骤中识别的新或修改过的块，必须生成嵌入。这涉及调用您的嵌入模型服务。
   • 对于已删除的块，需要识别其相应的向量 (vector)ID，以便从向量数据库中移除。

3. 向量数据库更新：这是更改在检索组件中体现的地方。

   • 新块：嵌入被插入到向量数据库中，通常与它们的文档ID、块ID以及任何相关元数据一起。
   • 更新的块：大多数向量数据库支持“upsert”操作（如果存在则更新，如果不存在则插入），这是很理想的。如果不支持，这可能需要先删除旧向量，然后插入新向量。在更新过程中对文档块使用一致的ID是很重要的。
   • 已删除的块/文档：与已删除内容对应的向量必须从索引中移除。在某些向量数据库实现中，高效删除可能是一个挑战，可能需要定期压缩或重新索引段以回收空间并保持性能。

4. 辅助文档存储（如果适用）：如果您的RAG系统单独存储块或文档的完整文本（例如，在S3存储桶、Elasticsearch或关系数据库中），以便稍后检索显示给用户或作为LLM的最终上下文 (context)，则此存储也必须与CDC事件同步更新。

RAG中CDC的重要考量

为实时RAG更新实施CDC涉及处理多项复杂性：

• 模式演变：源数据库模式会更改。您的CDC管道和RAG更新处理器必须处理这些更改。一些CDC工具（如Debezium）可以传播模式更改事件，允许下游系统进行调整。否则，需要仔细的协调和部署策略。
• 数据转换和序列化：更改事件通常以JSON或Avro等格式生成。确保您的流处理器和RAG更新处理器能够正确反序列化和解析这些事件，并将它们转换为您的文档处理和嵌入 (embedding)管道所需的精确结构。
• 幂等性：分布式系统中的消息传递语义（特别是重试时）可能导致重复事件处理。您的RAG更新处理器（特别是那些写入向量 (vector)数据库和文档存储的）必须设计为幂等的。例如，对已存在的块重新处理插入事件不应创建重复条目或报错。Upsert操作本身就是幂等的。
• 顺序：虽然Kafka等平台保证分区内的顺序，但跨多个分区或由多个消费者实例处理可能导致事件乱序。对于RAG，如果文档被更新然后迅速删除，在更新之前处理删除事件可能导致不一致状态。策略包括：
  • 按文档ID对Kafka主题进行分区，以确保给定文档的所有更改由单个消费者实例按顺序处理。
  • 在您的更新逻辑中使用事件时间戳和乐观锁或版本控制，尽管这会增加复杂性。
• 错误处理和死信队列 (DLQs)：如果处理更改事件反复失败（例如，嵌入模型故障，格式错误的事件）会怎样？失败的事件应路由到DLQ，以便稍后检查和重新处理，而不是暂停整个管道。
• 源系统影响：尽管基于日志的CDC影响小，但请确保其配置正确。配置错误或过度激进的日志读取仍可能影响源数据库性能。密切监控您的源系统。
• 端到端延迟：“准实时”是相对的。测量从源数据库中提交更改到其在RAG系统中反映的延迟。这涉及日志写入、CDC代理轮询、Kafka传播、流处理以及最终RAG组件更新中的延迟。目标延迟将取决于您的应用程序要求。典型延迟范围从几秒到几分钟。
• 初始数据加载（回填）：CDC处理持续性更改。初次设置系统或添加新数据源时，您需要一种策略来执行现有数据的初始批量加载到您的RAG管道和向量数据库中。一些CDC工具提供快照功能以方便此操作。此初始加载必须与快照过程中发生的更改进行协调，以避免数据丢失或不一致。
• 向量数据库效率：频繁更新，尤其是删除，可能导致某些向量数据库中的索引碎片化或效率低下。理解您所选向量数据库在更新/删除性能方面的特性以及可能需要的任何维护操作（如压缩或重新索引）。

通过周全地处理这些考量，您可以构建一个CDC管道，使您的大规模分布式RAG系统与数据源保持同步，确保其检索的信息和生成的回答尽可能地最新和准确。这种新鲜数据的持续流动，将静态知识库转换为动态、响应迅速的信息资源。