传统RAG系统通常基于静态或不常更新的文档集运行。然而，许多应用需要与持续变化的信息源进行交互。新闻订阅、金融市场数据、社交媒体更新、物联网传感器数据以及不断演进的知识库，都要求RAG系统能够摄取、处理和获取反映最新状态的信息。整合此类动态性带来了独特的架构和操作难题，这些超出了标准RAG实现的范围。设计RAG系统以有效处理高动态和流式数据源的策略将得到详细说明，从而确保及时和相关的信息获取与生成。
动态RAG中信息时效性的必要性
当数据源快速变化时，主要问题是数据过时。RAG系统获取的信息即使只过时几分钟或几小时，也可能导致大型语言模型（LLM）给出不准确、误导性或不相关的回应。例如，一个用于金融新闻分析的RAG系统；处理市场动态新闻的延迟可能使其输出无用甚至有害。
应对此情况需要从批处理转向支持持续摄取和近实时更新的架构。受影响的主要组件是：
- 数据摄取：必须处理高速、高容量的数据流。
- 数据处理与嵌入 (embedding)：分块、元数据提取和嵌入生成需以最小延迟完成。
- 向量 (vector)索引：向量数据库必须支持频繁的更新、添加和删除，而不会显著降低性能或消耗过多资源。
- 检索机制：可能需要考虑时间敏感性或优先处理最新信息。
用于动态数据摄取的架构模式
可以采用几种架构模式来构建适用于动态数据的RAG系统。选择通常取决于传入数据的具体速度、容量和多样性，以及所需的时效性。
近实时（NRT）索引管道
处理动态数据的基础是能够快速更新检索索引。NRT索引旨在最大程度地缩短新信息的“存活时间”，使其在到达后几乎立即变得可搜索。
- 增量更新：现代向量 (vector)数据库（例如Milvus, Weaviate, Pinecone）提供API，可有效添加、更新或删除单个向量或小批量数据。这避免了每次更新都需对整个数据集进行完全重新索引。从架构上讲，这意味着您的数据管道必须能够识别变化并将其作为细粒度更新传播到向量存储。
- 分层索引：一种做法是维护多个索引分片或段。新数据流入一个较小、高度动态的“热”层，该层针对频繁写入进行了优化。热层中的数据会定期合并到更大、更静态的“温”或“冷”层，后者可能针对读取性能进行了优化。查询随后会在相关层之间进行搜索。
- 向量搜索中的日志结构合并（LSM）树：一些向量数据库正在采用与LSM树类似的原理，这在NoSQL数据库中很常见。新数据写入内存结构（memtable），并定期作为不可变段刷新到磁盘。这些段随后被压缩。这种架构天生适合高写入吞吐量 (throughput)。
流处理集成
对于真正的流式数据源，与流处理框架集成非常必要。像Apache Kafka、Apache Flink和Spark Streaming这样的系统提供了管理连续数据流的基础支撑。
动态RAG系统的典型流式摄取管道。数据从源头流经消息队列，由流处理器实时处理、嵌入 (embedding)，然后索引到近实时向量数据库中。
在此设置中：
- 数据摄取：动态数据（例如推文、新闻文章、日志条目）被推送到像Kafka这样的消息队列中。这解耦了数据生产者和消费者，并提供了缓冲。
- 流处理：Flink或Spark Streaming作业从Kafka消费数据。此作业执行以下操作：
- 预处理：清洗、规范化、元数据提取。
- 分块：将文档划分为易于嵌入的片段。
- 嵌入生成：调用嵌入模型服务（该服务可独立扩展）将文本块转换为向量。
- NRT索引：嵌入向量及其对应的文本和元数据被写入配置为NRT更新的向量数据库中。
调整Lambda和Kappa架构
尽管完整的Lambda或Kappa架构对于某些RAG用例可能过于详细，但其原理可以提供参考。
- RAG的Lambda架构：您可以有一个批处理层，以较低频率处理和索引大量历史语料。一个速度层，类似于上述的流处理设置，负责处理传入的实时数据，并将其索引到一个独立、更小或更易变的索引中。RAG系统的检索组件随后会查询这两个层并合并结果，可能会优先考虑速度层中的信息。
- RAG的Kappa架构：这种更简洁的方法完全依赖于流处理。所有数据，无论是历史数据还是新数据，都被视为一个流。重新处理或重新索引涉及重放该流。这要求高效的流处理基础设施和向量存储中高效的NRT索引能力。
变更数据捕获（CDC）
对于频繁更新的传统数据库（SQL或NoSQL）中的数据，CDC是一种高效技术。CDC系统（例如Debezium）捕获源数据库中的行级变化（插入、更新、删除），并将这些变化作为事件流传输。这些事件随后可以被送入Kafka主题，由流处理器处理，并用于近实时地更新RAG系统的向量索引和知识库。这保证了RAG系统与操作数据存储保持同步。
时间敏感信息的检索策略
仅仅快速索引数据是不够的；检索机制还必须了解语料库的动态特性。
- 时间加权评分：检索分数可根据信息的时效性进行调整。例如，可以将指数衰减函数应用于旧文档的相关性分数。如果Srel是原始相关性分数，tdoc是文档的“年龄”，则时间调整后的分数Sfinal可能是：
Sfinal=Srel⋅e−λ⋅tdoc
其中 λ 是一个衰减常数，控制旧文档被惩罚的速度。
- 时间范围过滤器：向量 (vector)数据库通常支持元数据过滤。查询可以通过时间范围过滤器进行增强，以明确限制搜索结果为在特定时间窗口内创建或更新的文档（例如，“过去24小时”、“本周”）。这在用户明确查找最新信息时特别有用。
- 混合索引查询：如果采用分层索引方法（热/温层），检索服务可以首先查询“热”索引或为其结果分配更高的优先级。这确保了最新数据得到优先处理。
- 文档版本控制与失效：当文档更新或删除时，必须管理其在向量索引中对应的嵌入 (embedding)向量。这包括：
- 软删除：将向量标记 (token)为已删除但不立即移除，这对于快速失效很有用。实际移除可以在后续的压缩阶段进行。
- 覆盖/更新：如果向量数据库支持基于文档ID的高效原地更新，则用新版本替换现有向量。
- 墓碑管理：跟踪已删除的文档ID，以防止在旧数据流被重新处理时它们再次出现。
LLM与动态数据的交互及上下文 (context)管理
LLM是获取到的动态信息的最终使用者。
- 动态提示：可以动态构建提示，以告知LLM所提供上下文的时效性。例如，“根据过去一小时内获取到的以下信息...”
- 上下文刷新：对于长时间运行的对话式RAG应用，如果已知底层数据高度易变，系统可能需要定期重新获取上下文。
- 检测过时信号：RAG系统可以尝试检测所获取的上下文，即使是最近的，是否仍可能被尚未索引的更新信息取代。这是一个进阶领域，可能涉及预测模型或基于数据源更新频率的启发式方法。
动态RAG系统的操作考量
在动态数据上运行RAG系统会带来特定的操作复杂性：
-
监控数据管道延迟：监控数据从源头出现到在RAG系统中可查询的端到端延迟非常重要。为数据时效性定义服务水平目标（SLO）。
监控摄取延迟与既定服务水平目标（SLO）的示例。峰值表示流处理管道中可能存在问题。
-
可伸缩性与弹性：摄取管道、嵌入 (embedding)服务和向量 (vector)数据库必须能够独立扩展，以处理数据量和速度的波动。云原生架构和基于Kubernetes的容器化在此处很有帮助。
-
成本管理：持续流处理和NRT索引可能资源密集。可通过以下方式进行优化：
- 使用高效的嵌入模型。
- 在适当情况下对向量数据库进行微批次更新。
- 选择平衡写入性能、查询性能和成本的向量数据库配置。
- 可能会对不那么重要的数据流进行降采样或过滤。
-
更新的幂等性：确保如果数据更新被处理多次（例如，由于流处理器中的重试），它不会导致向量索引中出现重复条目或不正确状态。在upsert操作中使用唯一的文档ID是常见的做法。
-
错误处理与死信队列（DLQ）：整个流处理管道中都需要错误处理。无法成功处理的数据应路由到死信队列（DLQ），以便后续检查和重新处理，而不是停止整个流。
使RAG系统适应高动态和流式数据源，是构建更灵敏、更具上下文 (context)感知能力的AI应用的重要一步。这需要细致的架构设计、选择适合流处理和NRT索引的技术，以及持续监控以确保数据时效性和系统性能。尽管存在挑战，但能让LLM回应基于最实时可用信息的能力，开辟了新一类应用，涵盖从实时事件监控与分析到高度个性化和自适应的用户体验。
