实时RAG：流式Embedding+增量索引，让新数据5秒内可检索

你刚发布的新闻，5分钟后才能搜到——这就是传统RAG的"实时"水平

某新闻聚合平台的RAG系统，用户经常抱怨"刚发布的新闻搜不到"。排查发现：新文档从发布到可检索，延迟是5-15分钟。因为传统RAG是"批量处理"的——每15分钟拉取一批新文档，然后批量Embedding，批量导入向量数据库。

2026年，实时RAG正在成为越来越多场景的刚需：新闻、股票、舆情、客服、IoT数据。以下是一个完整的实时RAG架构。

实时RAG的技术挑战

1. 流式Embedding：数据到达即Embedding，而不是等待批量
2. 增量索引更新：新向量插入后立即可检索，而不是等待索引重建
3. 一致性保证：新数据和旧数据的一致性，不能有"空窗期"
4. 性能平衡：实时写入和实时查询的并发竞争

实时RAG架构

数据源（新闻API、数据库CDC、Kafka）
    ↓
消息队列（Kafka/Pulsar）
    ↓
流式处理（Flink/Spark Streaming）
    ↓
Embedding服务（批量API调用/自部署模型）
    ↓
向量数据库（增量写入+实时索引）
    ↓
RAG查询服务

关键组件

消息队列：Kafka

选择Kafka而不是批处理，是因为Kafka提供了"事件驱动"的架构。新数据到达后，立即作为事件推送到下游。

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'new_documents',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='latest',
    enable_auto_commit=True,
)

for message in consumer:
    document = json.loads(message.value)
    # 实时处理：Embedding → 写入向量数据库
    process_document(document)

流式Embedding

批量Embedding API（如OpenAI）的延迟是200ms/条。实时场景下，需要优化：

1. 微批处理：不是逐条Embedding，而是攒够16条再批量调用API（减少网络往返）
2. 自部署Embedding模型：用BGE-M3自部署，延迟降到10ms/条
3. 异步处理：Embedding和写入异步进行，不阻塞消息消费

实测：自部署BGE-M3，微批处理16条，单条Embedding延迟从200ms降到15ms。

增量索引

Milvus的增量写入：新数据写入后，在segment被flush（默认1秒）后即可检索。但索引构建需要时间——HNSW索引构建约1-5秒（取决于数据量）。

优化：使用Milvus的"streaming"模式。新数据先写入一个"streaming segment"，使用暴力搜索（无索引），查询延迟约100ms。同时后台构建HNSW索引，构建完成后切换到索引搜索。

金句：实时RAG的"实时"是分阶段的。1秒内可检索（暴力搜索），5秒内可检索（索引搜索），30秒后性能最优（索引优化完成）。

实测效果

金句：实时RAG不是"更快"，而是"另一种架构"。它需要Kafka、流式处理、增量索引——这些基础设施的成本不低，但带来的"实时性"价值远超成本。**