模型每次升级都在变好，但喂给它的向量库可能还停在上一个坐标系——数据 pipeline 才是 RAG 不随时间退化的真正基础设施。
大多数 RAG demo 死在「跑通即终点」：一次性把文档全量嵌入、塞进向量库、上线。真正的痛苦在第二天才开始——文档改了一行，你是重嵌整库还是只补那一块？删掉的页面还会不会被召回？换个更好的 embedding model，旧向量怎么办？你给 reranker 喂的「用户近 7 天点击」特征，训练和线上算的是不是同一个值？这些都不是模型问题，是 数据 pipeline 工程问题，且每一个都会悄悄腐蚀线上质量而不报错。这一期把 AI 数据流水线拆成四个最容易塌方的承重点：嵌入 ETL 的成本结构、增量索引的 diff 与删除、嵌入模型升级的索引迁移、特征存储的离线/在线一致性。每一点都给可直接抄的实现骨架与失败模式。
一条 AI 数据流水线本质是 extract → chunk → embed → load。前三步里，extract 和 chunk 几乎免费（CPU、毫秒级），embed 才是瓶颈：它要走网络、按 token 计费、有 rate limit。所以工程上的第一原则是——把 embed(chunk) 当成纯函数：相同文本必出相同向量，于是可以按 hash(model_id + chunk_text) 做缓存键。一旦同一块内容出现在多个文档、或同一文档被反复 ingest，缓存命中就把成本与延迟砍掉一个数量级。第二原则是批处理：embedding API 支持一次传一批，单条调用 100 个 chunk 的吞吐远高于 100 次单条。第三是 chunk 与 embed 解耦——chunk 策略会反复调，但只要文本没变就不该重新付费嵌入。
# 嵌入即纯函数：内容哈希做缓存键 + 批处理
import hashlib, json
MODEL = "voyage-3" # 把 model_id 写进 key，换模型自动 miss
def key(text): return hashlib.sha256(f"{MODEL}:{text}".encode()).hexdigest()
def embed_batch(chunks, cache):
todo = [c for c in chunks if key(c) not in cache]
if todo: # 只为未命中的付费
vecs = client.embed(todo, model=MODEL).embeddings
for c, v in zip(todo, vecs): cache[key(c)] = v
return [cache[key(c)] for c in chunks]把 model_id 编进缓存键这一行，是为下一节「换模型」埋的伏笔——换 model 会让所有 key 自动失效，逼你重嵌，而不是悄悄混用两个坐标系的向量。
文档库是活的：新增、修改、删除每天都在发生。全量重建在文档少时能忍，过万就成了每日数小时 + 成倍 API 账单的负担。增量索引的核心是维护一张「块指纹表」（chunk_id → content_hash），每次 ingest 与上一次对比，得出三类操作：新增（新 hash）、变更（同 id 不同 hash → 重嵌 + upsert）、删除（上次有这次没有 → 从向量库删）。前两类大多数人会做，删除几乎总被漏掉——因为「不报错」。漏删的后果是幽灵召回：文档早已下架，它的向量还躺在库里被检索到，模型据此生成已经过时甚至错误的答案。删除要走tombstone（墓碑标记）流程，确保向量库里那条真的消失。
# diff 三态：新增 / 变更 / 删除（删除最容易漏）
def reindex(new_chunks, fingerprint_store, vdb):
old = fingerprint_store.load() # {chunk_id: hash}
new = {c.id: c.hash for c in new_chunks}
changed = [c for c in new_chunks
if old.get(c.id) != c.hash] # 新增 + 变更
deleted = old.keys() - new.keys() # ← 关键：上次有这次无
if changed: vdb.upsert(embed_batch([c.text for c in changed]))
if deleted: vdb.delete(ids=list(deleted)) # tombstone，杜绝幽灵召回
fingerprint_store.save(new)指纹表本身就是一份廉价的「数据版本快照」——出问题时可以拿它和向量库对账，找出索引与源数据漂移的那几条。
这是最隐蔽、代价最大的一类事故。embedding model 每次升级（甚至同名模型的小版本），向量空间的几何都变了：同一句话被 model A 和 model B 嵌出的两个向量，余弦相似度毫无意义。如果你「为了省钱」只重嵌新文档、旧文档保留旧向量，整个索引就混进了两套不可比的坐标系，召回从此随机退化——而且不报错。正确做法借用部署里的蓝绿（blue-green）思路：新模型在一个独立的新索引里全量重建，期间双写、灰度对比召回质量，确认不退化后再原子切换流量，旧索引留作回滚。关键纪律：每条向量的元数据里写死它的 embedding_model 版本，让混用在查询层就能被断言拦下。
# 向量元数据绑定模型版本，查询层断言同源
vdb.upsert(id=cid, vector=v, metadata={
"embedding_model": "voyage-3", # ← 版本指纹
"indexed_at": ts, "doc_id": doc})
def search(q, vdb, model):
assert vdb.meta("embedding_model") == model, \
"query model != index model — 坐标系不一致！"
return vdb.query(embed(q, model=model), top_k=10)text-embedding-3 和某开源模型都输出 1024 维，shape 对得上、代码不报错——但语义空间完全不同，召回纯随机。维度兼容 ≠ 空间兼容。换模型没有「增量升级」，只有全量重建。
RAG 不只有向量。reranker 排序、个性化召回、过滤规则往往要用结构化特征：用户近 7 天点击、文档热度、作者权威度。这里有两个经典陷阱。其一 training-serving skew：训练用离线批处理 SQL 算特征，线上用另一套实时代码算——两套逻辑必然漂移，于是离线评估涨、线上掉。其二 数据泄漏：构造训练样本时，特征值要按当时那个预测时点（point-in-time）取，而不是「现在」——否则等于让模型在训练时偷看了未来，离线分虚高、上线打回原形。特征存储（feature store，如 Feast）就是为这两件事而生：离线 store 供训练、在线 store 供低延迟服务，同一份特征定义只写一次，并在离线侧实现 point-in-time 正确 join。
# point-in-time join：按预测时点取特征，杜绝未来泄漏
# 训练样本 = (entity, 预测发生的 event_timestamp, label)
training_df = store.get_historical_features(
entity_df=labels, # 含 event_timestamp 列
features=["user:clicks_7d", "doc:popularity"],
).to_df() # Feast 只取 ≤ event_timestamp 的特征值，不取"现在"
# 线上服务：同一份特征定义，低延迟在线 store
feats = store.get_online_features(
features=["user:clicks_7d", "doc:popularity"],
entity_rows=[{"user_id": u, "doc_id": d}],
).to_dict() # 训练/服务复用同一定义 → 无 skewSELECT ... WHERE now() 给历史样本回填特征。这等于给三个月前的样本贴上今天的点击数——典型 point-in-time 泄漏，离线 AUC 漂亮得离谱，上线归零。任何带时间属性的特征，回填时都要问：这个值在预测发生那一刻真的存在吗？
把四点串成一次重构：让你那个「一次性脚本」的 RAG 进化成能日更、能换模型、不退化的流水线。
hash(model_id + chunk_text) 缓存。把今天的全量 ingest 跑两遍，确认第二遍命中率接近 100%、几乎零成本。chunk_id → content_hash 表，ingest 改成三态 diff。专门测一次删除：删掉一篇文档，确认它的向量真的从库里消失、不再被召回。embedding_model，查询层加断言。模拟一次换模型：建 green 索引全量重嵌，用一个固定 eval 集对比 blue/green 召回，确认不退化再切。做完这四步，你的 RAG 从「demo 能跑」升级到「线上能养」。以后看任何向量检索系统，你会本能地问四个问题：嵌入有没有缓存、增量删没删干净、换模型怎么迁、特征会不会泄漏——而不是只盯着 top_k 调参。
model_id 编进键是同一个思路：任何会改变最终向量的输入，都必须进缓存键。