一、面试题目请说明实时数据 RAG新闻、股价、业务订单类实时场景如何设计增量更新、数据同步、索引刷新、过期清理整体架构与落地方案二、知识储备1. 整体设计思路实时数据 RAG 和静态文档 RAG 最大区别数据持续新增、变更、过期不能全量重建索引。核心设计原则增量采集→增量分块→增量向量化→增量入库→定时过期清理→冷热数据分层做到不重建全量库、低延迟同步、不影响在线检索、可控资源开销适配新闻、股价、交易订单、实时业务流水等时序动态场景。2. 核心模块设计 原理、实现、优化1实时数据源接入层数据源类型新闻资讯 API、爬虫实时流、消息队列股价行情 WebSocket、金融实时接口订单业务 MQKafka/RabbitMQ、Binlog 数据订阅接入方式流式消费MQ 实时消费、WebSocket 长连接推送增量轮询定时拉取增量时间戳区间数据变更捕获MySQL Binlog/Canal 监听订单新增、修改、作废设计要点按时间戳、批次、业务分区打标签每条实时数据携带唯一 ID、时间、类型、状态、过期时效。2增量分块与结构化处理定义不做全量文档切块只对新增 / 变更单条实时数据做独立分块、结构化封装。实现新闻单条资讯作为最小切片保留发布时间、来源、摘要股价按分钟 / 小时粒度聚合切片保留时间序列字段订单单条订单结构化转文本 JSON 元数据保留订单号、状态、金额、时间优化实时数据不合并大文档一条事件一个向量切片便于单独更新、单独删除、单独过期。3增量向量化 索引写入核心原理静态 RAG 全量灌库实时 RAG 只做新增向量化、增量追加索引。流程新增实时数据 → 轻量化文本预处理 → 调用 Embedding 生成向量 →增量写入向量库不重建已有索引。关键机制写隔离增量写入走单独写入节点不影响在线检索查询节点幂等写入基于唯一业务 ID 做幂等避免重复向量化、重复入库元数据绑定向量附带时间、业务类型、过期时间、状态标签4数据更新策略新增、修改、删除新增数据增量向量化 → 追加写入向量库 元数据库实时可被检索。数据变更订单改状态、股价修正、新闻编辑旧版本向量标记失效不物理删除生成新内容向量增量新增入库检索时过滤已失效版本只召回最新有效数据数据作废 / 撤销通过元数据打禁用标签检索侧查询时添加过滤条件直接过滤作废数据。5定时过期清理 冷热分层定义新闻、股价、订单都有时效生命周期过期数据无需参与检索做自动下线清理。实现方式时效 TTL新闻 7 天、股价分时数据 30 天、订单流水 90 天配置 TTL定时任务按时间范围批量标记过期、归档至冷存储冷热分层近期热数据存高性能向量库历史冷数据归档至对象存储 / 离线库不占用在线检索资源好处控制向量库规模、降低检索开销、避免过时信息产生 RAG 幻觉。6检索侧适配实时 RAG时间维度过滤用户提问自动附带时间范围最新新闻、近 7 日股价、近 30 天订单检索时加元数据时间过滤只召回有效实时切片。实时兜底机制向量库未收录最新瞬间数据时触发实时 API 直查把接口结果临时拼接进上下文弥补向量库微小延迟。7全链路同步保障批次校验每批次增量数据做条数校验、落库对账故障重试向量化 / 入库失败进入死信队列自动重试版本基线每日凌晨做一次小范围全量比对修复增量漏同步数据3. 完整落地流程实时数据源通过 MQ/WebSocket/Binlog 增量接入单条事件独立分块、结构化封装携带元数据与时效标签仅对新增 / 变更数据增量向量化幂等写入向量库变更数据标记旧版本失效新增最新版本向量定时 TTL 任务做过期标记、冷热数据归档清理检索侧按时间、状态、类型过滤只召回有效实时数据极端实时场景兜底直调业务 API补齐向量库延迟每日基线对账保证增量同步不丢不漏4. 评估指标 合格基线数据同步延迟消息产生到可检索13s 内增量入库成功率99.9%重复入库率趋近 0过期数据清理覆盖率100%实时问答准确率基于最新实时数据回答正确率 90%三、破局之道面试高阶满分表述实时数据 RAG 不能套用传统静态文档全量索引模式核心是抛弃全量重建走增量流式架构。整体分为四层接入层用 MQ、WebSocket、Binlog 做实时增量捕获处理层对单条新闻、股价、订单做独立轻量化分块不做大文档合并索引层只对新增变更数据增量向量化、增量追加入库通过元数据版本标记实现修改和作废运维层配置 TTL 时效过期、冷热数据分层自动清理控制库规模与检索成本。同时检索侧增加时间、状态维度过滤极致实时场景兜底直调业务 API 补全延迟既保证数据秒级同步可查又避免全量重构开销、防止过时数据引发幻觉是新闻、股价、订单类实时 RAG 的标准工业设计方案。四、代码实现Python 实时 RAG 增量更新简易模拟import time class RealTimeRAGIncrement: def __init__(self): # 存储向量索引、失效ID、过期配置 self.vector_store dict() self.invalid_ids set() self.ttl_config { news: 7 * 86400, stock: 30 * 86400, order: 90 * 86400 } # 增量新增数据 def add_increment_data(self, biz_id: str, content: str, data_type: str, create_ts: int): # 模拟向量化 embedding [round(0.1 * i, 2) for i in range(10)] self.vector_store[biz_id] { content: content, embedding: embedding, type: data_type, ts: create_ts, valid: True } return 增量入库成功 # 数据更新标记旧失效新增新版本 def update_data(self, old_biz_id: str, new_biz_id: str, new_content: str, data_type: str): self.invalid_ids.add(old_biz_id) now_ts int(time.time()) return self.add_increment_data(new_biz_id, new_content, data_type, now_ts) # 过期数据清理 def clean_expired(self): now int(time.time()) to_clean [] for biz_id, info in self.vector_store.items(): ttl self.ttl_config.get(info[type], 30*86400) if now - info[ts] ttl: self.invalid_ids.add(biz_id) to_clean.append(biz_id) return f已清理过期数据{len(to_clean)} 条 # 检索时过滤失效与过期 def retrieve_valid(self): res [] for biz_id, info in self.vector_store.items(): if biz_id not in self.invalid_ids: res.append({biz_id: biz_id, content: info[content]}) return resJavaScript 版本class RealTimeRAGIncrement { constructor() { this.vectorStore new Map(); this.invalidIds new Set(); this.ttlConfig { news: 7 * 86400, stock: 30 * 86400, order: 90 * 86400 }; } addIncrementData(bizId, content, dataType, createTs) { const embedding Array.from({length:10}, (_,i) Number((0.1*i).toFixed(2))); this.vectorStore.set(bizId, { content, embedding, type: dataType, ts: createTs, valid: true }); return 增量入库成功; } updateData(oldBizId, newBizId, newContent, dataType) { this.invalidIds.add(oldBizId); const nowTs Math.floor(Date.now() / 1000); return this.addIncrementData(newBizId, newContent, dataType, nowTs); } cleanExpired() { const now Math.floor(Date.now() / 1000); const toClean []; for (let [bizId, info] of this.vectorStore) { const ttl this.ttlConfig[info.type] || 30*86400; if (now - info.ts ttl) { this.invalidIds.add(bizId); toClean.push(bizId); } } return 已清理过期数据${toClean.length} 条; } retrieveValid() { const res []; for (let [bizId, info] of this.vectorStore) { if (!this.invalidIds.has(bizId)) { res.push({bizId, content: info.content}); } } return res; } }