AI 应用的数据架构：从 OLTP 到向量检索的统一

压测时 Agent 读到了三小时前的数据

7 月 18 日我们对知识问答 Agent 做季度压测。压到 1,200 QPS 的时候，测试同学反馈一个问题：刚在后台改的商品价格，问 Agent 还是旧值。

一开始我以为是缓存。查了一圈发现不是——向量库里的切片更新时间是三小时前，而业务库的变更早就提交了。

这篇记录我们把"业务数据 → 向量索引"这条链路重做的过程，以及多模数据库这个选项我们为什么选了一半。

先看清链路到底有多长

改造前的架构是典型的"业务库 + 独立向量库"双写模式：

业务服务 ──写──> PostgreSQL（订单/商品/用户）
   │
   └──MQ──> 同步服务 ──> 切分 ──> embedding ──> Qdrant
                                                    │
用户提问 ──> Agent ──────────────────────> 向量检索 ─┘
                    └────────────────────> 业务库查详情

这条链路有 5 个环节，每个都可能成为延迟来源。压测时的问题出在 MQ 积压——同步服务的消费速度跟不上，1,200 QPS 下积压了 47 万条，追平需要 3 小时。

更要命的是，这个问题在压测前一直存在，只是没人测过。平时 QPS 只有 80，同步延迟 P50 是 800ms，看起来"实时"，实际脆弱得不行。

我们把延迟分布拉出来看：

三个选项和它们的真实代价

问题清楚了，接下来是方案。我们评估了三条路。

方案一：继续双写，优化同步链路

最简单，把消费能力提上去、加并行、加批量。我们试了一版，把 embedding 调用改成批量（batch=32），同步服务消费能力从 180 QPS 提到 900 QPS。

但这条路有一个无解的硬伤：业务库和向量库是两个存储，无法在同一个事务里提交。任何时刻都可能存在"业务库有、向量库没有"的窗口。压到 1,200 QPS 还是会积压，只是阈值提高了。

而且双写带来一个隐性问题：两边的数据模型不一致。业务库里商品的"状态"是个枚举，向量库里是切分成的一段文本。改了状态字段的含义，两边要同步改，没人能保证。

方案二：CDC 替代双写

把 MQ 那一段换成 Debezium 抓 binlog。好处是业务代码零侵入，不会漏掉任何一条变更（包括 DBA 手工改的数据）。

@Component
public class ProductCdcListener {

    @KafkaListener(topics = "pg.public.product")
    public void onChange(ConsumerRecord<String, Envelope> rec) {
        Envelope env = rec.value();
        if (env.op().isDelete()) {
            vectorStore.deleteBySourceId(env.before().id());
            return;
        }
        Product p = env.after();
        // 关键：用 binlog 的 LSN 做版本，保证乱序到达时旧数据覆盖不了新数据
        vectorStore.upsert(toDocuments(p), Version.of(env.lsn()));
    }
}

Version.of(env.lsn()) 这行是必须的。CDC 消息不保证严格有序（我们按主键分区，同一主键有序，跨主键无序），如果商品 A 和 B 有关联内容（比如同一个类目的聚合切片），乱序到达会导致旧数据覆盖新的。用 LSN 做版本号，向量库侧做"只接受更新的版本"。

CDC 的另一个收益是能捕获删除。之前的双写逻辑里，删除操作经常漏——业务代码里 7 处删除商品的地方，只有 5 处发了 MQ 通知。这个问题我们查了两天。

但 CDC 没解决根本问题：仍然是两个存储，仍然有延迟窗口。

方案三：多模数据库，把向量放进业务库

这条路我们最后走了一半。

思路是：既然业务数据本来就在 PostgreSQL 里，那把 embedding 也存进去，用 pgvector。这样"改商品"和"改向量"就能在同一个事务里完成，一致性问题从根上消失。

ALTER TABLE product ADD COLUMN embedding vector(1024);

-- 业务字段和向量同一个事务更新
BEGIN;
  UPDATE product
     SET price = 12900,
         status = 'ON_SALE',
         updated_at = now(),
         embedding = '[0.021,-0.118,...]'::vector,
         embed_version = 'bge-m3-v2'
   WHERE id = 90211;
COMMIT;

-- 混合检索：结构化过滤 + 向量相似度，一条 SQL
SELECT id, name, price,
       1 - (embedding <=> $1::vector) AS score
  FROM product
 WHERE tenant_id = $2
   AND status = 'ON_SALE'
   AND price BETWEEN $3 AND $4
 ORDER BY embedding <=> $1::vector
 LIMIT 10;

这个 WHERE + 向量 ORDER BY 的组合，是独立向量库很难做好的事。Qdrant、Milvus 都支持 payload 过滤，但当过滤条件的选择性很高时（比如"某租户 + 某类目 + 价格区间"命中 0.1%），性能会明显下降，因为它们要先向量检索再过滤，或者维护额外的索引。

我们实测：在 420 万条商品数据上，tenant_id + status 过滤后命中 3,800 条的场景，pgvector 的 HNSW 索引 + 部分索引组合耗时 8.4ms；Qdrant 用 payload 索引耗时 31ms。

事务一致性是更大的收益。改造后我们在 ISO 层面根本不存在"业务库改了向量库没改"这个状态。

为什么只走了一半

pgvector 不是万能的，它有几个硬约束。

规模上限。我们的知识库有 41 万切片，全塞进 PostgreSQL 之后，HNSW 索引占 7.2GB，加上原表，单表 12GB。这个量级 PostgreSQL 扛得住，但如果涨到千万级，HNSW 索引的内存占用会失控。我们算了一下，1,000 万条 1024 维向量的 HNSW 索引约 45GB，必须全在内存里才有好性能，这机器成本不划算。

文档类内容和业务数据的访问模式完全不同。商品数据读写都频繁，知识库（PDF、工单、IM 记录）基本是写多读少、批量倒入。把它们放一个库里，备份策略、扩容节奏、慢查询治理全都互相干扰。

所以最后的分工是：

这个划分的依据不是技术，是业务对一致性的要求。商品改了价格，Agent 必须立刻说新价格；知识文档改了条款，5 分钟内生效就行。把它统一了，反而是过度设计。

一致性保障：延迟要可观测

接受了"部分最终一致"之后，剩下的工作就是把它变得可观测。我们的做法是一个持续运行的校验任务：

@Scheduled(fixedDelay = 10_000)
public void probeFreshness() {
    // 写入一个探针记录，带时间戳
    long probeId = probeDao.insert(Instant.now());

    // 轮询向量库，看多久能查到
    Instant start = Instant.now();
    while (Duration.between(start, Instant.now()).toSeconds() < 60) {
        if (vectorStore.exists("probe", probeId)) {
            long lagMs = Duration.between(start, Instant.now()).toMillis();
            metrics.timer("rag.sync.lag").record(lagMs, MILLISECONDS);
            probeDao.delete(probeId);
            return;
        }
        sleep(200);
    }
    metrics.counter("rag.sync.timeout").increment();   // 60 秒还没同步，告警
    alerting.page("向量同步超过 60s");
}

这个探针每 10 秒跑一次，成本可以忽略，但它把"同步延迟"从一个模糊的概念变成了一条可以告警的曲线。

还有一个我们觉得更有用的指标：答案新鲜度。做法是对一小部分（1%）真实请求做双路查询，一路走向量检索，一路直接查业务库拿权威值，比对答案里的关键字段是否一致。

// 抽样比对：向量召回的 price 和业务库的 price 是否一致
if (sampler.shouldSample(0.01)) {
    BigDecimal fromVector = extractPrice(agentAnswer);
    BigDecimal fromSource = productDao.priceOf(orderNo);
    if (fromVector.compareTo(fromSource) != 0) {
        metrics.counter("rag.stale_answer").increment();
        log.warn("stale: vector={} source={} traceId={}", fromVector, fromSource, traceId);
    }
}

7 月 25 日这个指标报了一次警：rag.stale_answer 在 20 分钟内从 0.02% 涨到 1.8%。查下来是某个商品的批量导入任务绕过了 CDC，直接写了业务库但没触发 binlog（用了 COPY）。如果没有这个指标，我们可能要等用户投诉。

改造后的数据

成本这块要说清楚：省下的钱不多（¥1,500/月），主要收益是一致性。别指望靠这个省钱。

还没统一的三块

坦白讲，"从 OLTP 到向量检索的统一"这个说法在业界喊了一年多，我目前看到的落地都是局部的。我们这边至少三块还没想清楚。

图谱数据。GraphRAG 我们做了个原型，实体关系存在 Neo4j 里，又是第三个存储。多模库号称支持图，但性能和查询表达力跟专业图库差得远。短期看不到统一的可能。

跨库事务。虽然业务实体和向量同库了，但一次 Agent 请求往往还要读知识库（Qdrant）和记忆（Redis）。跨这三个存储的一致性只能靠应用层，我们目前的做法是"以业务库为准，其余作为补充"，出错时优先信业务库。

向量索引的在线重建。换 embedding 模型要重建全部向量。pgvector 上重建 420 万条需要 4 小时，期间索引不可用。我们试过并发建索引（CREATE INDEX CONCURRENTLY），但 HNSW 的并发构建内存占用会翻倍。现在的做法是双索引切换，成本是多一倍存储。

留个问题

关于《AI 应用的数据架构：从 OLTP 到向量检索的统一》里这个坑，你当时是怎么处理的？欢迎在评论区聊聊你踩过的类似情况。

参考

• 官方文档：database-sql
• Baeldung 上的相关实战
• StackOverflow 上的相关讨论