Vector DB实战对比——5大主流DB写入和查询100K vectors性能基准

日期: 2026-09-15 方向: AI系统工程 / RAG 阶段: Phase 3 - RAG高级模式 (Day 135-148) 标签: #VectorDB #Pinecone #Weaviate #Qdrant #pgvector #Chroma #HNSW

---

今日目标

核心结论预告：Qdrant在single-node性能/成本是最佳；Pinecone serverless是创业公司零运维首选；pgvector够用且经济；Chroma只适合<1M原型；Weaviate hybrid search原生但性能略低于Qdrant。

---

一、核心概念：向量索引算法

1.1 三大主流索引算法

1.2 HNSW原理（最重要）

HNSW = 多层小世界图：

Layer 2 (sparse):     A ────────────── F
                      │                │
Layer 1 (medium):     A ──── C ──── F ──── J
                      │      │      │      │
Layer 0 (full):  A─B─C─D─E─F─G─H─I─J─K─L─M─N
                  ↑
                查询从顶层开始greedy search,
                逐层下降refine

• M: 每个node的邻居数（典型16-64）。M大→召回高，内存大
• ef_construction: 建图时的搜索宽度（典型128-512）。大→图质量高，建图慢
• ef_search: 查询时的搜索宽度（典型32-256）。大→召回高，查询慢

复杂度：O(log N)查询，远优于brute force的O(N)。

1.3 IVF原理

把向量空间用k-means聚成 nlist 个 cells (Voronoi)：

查询时：
1. 先和nlist个centroid比较（快）
2. 选最近的nprobe个cells（典型8-32）
3. 在这nprobe个cells内brute force

• nlist: 通常 = 4 * sqrt(N)
• nprobe: 越大召回越高但越慢
• 建议：N > 1M时IVF开始优于HNSW（HNSW内存压力大）

1.4 距离度量的"陷阱"

不同DB对同样的"cosine"含义不同：

生产坑：从一个DB迁移到另一个时，必须检查并反转排序逻辑。

---

二、5大Vector DB速览

2.1 Pinecone

• 架构：完全SaaS，单region多replica
• 索引类型：内部HNSW（serverless）/ p1 (HNSW) / s1 (IVF disk-based, 便宜)
• 价格：serverless按storage + reads + writes计费；s1 $70/mo起
• 优势：零运维，scaling自动，filter性能好
• 劣势：vendor lock-in，私有部署only enterprise

2.2 Weaviate

• 架构：开源（Go），可自部署或Weaviate Cloud
• 索引：HNSW + 原生支持BM25 → hybrid search原生
• 价格：自部署免费；Cloud $25/mo起
• 优势：multi-tenancy原生、GraphQL、hybrid search
• 劣势：性能略低于Qdrant

2.3 Qdrant

• 架构：开源（Rust），单binary
• 索引：HNSW + payload index（filter优化）
• 价格：自部署免费；Qdrant Cloud $25/mo起
• 优势：性能最强、payload filter优化、量化（int8/binary）支持好
• 劣势：相对年轻，生态不如Pinecone

2.4 pgvector

• 架构：Postgres扩展
• 索引：HNSW (since 0.5.0) / IVFFlat
• 价格：已有PG的话$0
• 优势：和existing data同库、SQL JOIN方便、ACID
• 劣势：性能一般、scaling困难、>10M vectors掉链子

2.5 Chroma

• 架构：Python优先，可local/server
• 索引：HNSW (hnswlib)
• 价格：开源免费；Chroma Cloud beta
• 优势：简单到极致，原型快
• 劣势：production scaling有限

---

三、Benchmark代码：bench_runner.py

"""
bench_runner.py — 在5个Vector DB上benchmark性能
依赖：
  pip install pinecone-client weaviate-client qdrant-client \
              psycopg2-binary chromadb numpy tqdm pandas

环境变量：
  PINECONE_API_KEY=...
  WEAVIATE_URL=http://localhost:8080
  QDRANT_URL=http://localhost:6333
  POSTGRES_URL=postgresql://user:pass@localhost:5432/db
"""
import os
import time
import json
from dataclasses import dataclass
from typing import List, Dict
import numpy as np
from tqdm import tqdm
import pandas as pd

# ============================================================
# 1. 测试数据生成
# ============================================================
N_VECTORS = 100_000
DIM = 1024
N_QUERIES = 1000

print("Generating test vectors...")
np.random.seed(42)
all_vectors = np.random.randn(N_VECTORS, DIM).astype(np.float32)
all_vectors /= np.linalg.norm(all_vectors, axis=1, keepdims=True)
all_metadata = [
    {
        "id": f"doc_{i}",
        "category": ["finance", "tech", "energy", "healthcare"][i % 4],
        "year": 2020 + (i % 5),
        "score": float(np.random.rand()),
    }
    for i in range(N_VECTORS)
]
query_vectors = all_vectors[:N_QUERIES]


# ============================================================
# 2. Adapter for each DB
# ============================================================
class VectorDBAdapter:
    name = "base"
    def setup(self): ...
    def insert_batch(self, vectors, ids, metadatas): ...
    def query(self, vector, top_k=10, filter=None): ...
    def teardown(self): ...


# --- 2.1 Pinecone ---
class PineconeAdapter(VectorDBAdapter):
    name = "pinecone"
    def setup(self):
        from pinecone import Pinecone, ServerlessSpec
        self.pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
        self.index_name = "bench-test"
        if self.index_name not in self.pc.list_indexes().names():
            self.pc.create_index(
                name=self.index_name, dimension=DIM, metric="cosine",
                spec=ServerlessSpec(cloud="aws", region="us-east-1"),
            )
        self.index = self.pc.Index(self.index_name)
        time.sleep(5)  # 等就绪

    def insert_batch(self, vectors, ids, metadatas):
        items = [
            (ids[i], vectors[i].tolist(), metadatas[i])
            for i in range(len(vectors))
        ]
        self.index.upsert(vectors=items)

    def query(self, vector, top_k=10, filter=None):
        return self.index.query(
            vector=vector.tolist(),
            top_k=top_k,
            filter=filter,
        )

    def teardown(self):
        self.pc.delete_index(self.index_name)


# --- 2.2 Qdrant ---
class QdrantAdapter(VectorDBAdapter):
    name = "qdrant"
    def setup(self):
        from qdrant_client import QdrantClient
        from qdrant_client.models import Distance, VectorParams
        self.client = QdrantClient(url=os.environ["QDRANT_URL"])
        self.collection = "bench_test"
        self.client.recreate_collection(
            collection_name=self.collection,
            vectors_config=VectorParams(size=DIM, distance=Distance.COSINE),
        )

    def insert_batch(self, vectors, ids, metadatas):
        from qdrant_client.models import PointStruct
        points = [
            PointStruct(id=i, vector=vectors[i].tolist(), payload=metadatas[i])
            for i in range(len(vectors))
        ]
        self.client.upsert(collection_name=self.collection, points=points)

    def query(self, vector, top_k=10, filter=None):
        from qdrant_client.models import Filter, FieldCondition, MatchValue
        qf = None
        if filter:
            conditions = []
            for k, v in filter.items():
                conditions.append(
                    FieldCondition(key=k, match=MatchValue(value=v))
                )
            qf = Filter(must=conditions)
        return self.client.search(
            collection_name=self.collection,
            query_vector=vector.tolist(),
            limit=top_k, query_filter=qf,
        )


# --- 2.3 Weaviate ---
class WeaviateAdapter(VectorDBAdapter):
    name = "weaviate"
    def setup(self):
        import weaviate
        self.client = weaviate.Client(url=os.environ["WEAVIATE_URL"])
        self.cls = "BenchTest"
        if self.client.schema.exists(self.cls):
            self.client.schema.delete_class(self.cls)
        self.client.schema.create_class({
            "class": self.cls, "vectorizer": "none",
            "properties": [
                {"name": "category", "dataType": ["text"]},
                {"name": "year", "dataType": ["int"]},
                {"name": "score", "dataType": ["number"]},
            ],
        })

    def insert_batch(self, vectors, ids, metadatas):
        with self.client.batch as batch:
            for i in range(len(vectors)):
                batch.add_data_object(
                    data_object=metadatas[i],
                    class_name=self.cls,
                    vector=vectors[i].tolist(),
                )

    def query(self, vector, top_k=10, filter=None):
        q = self.client.query.get(self.cls, ["category", "year"]) \
            .with_near_vector({"vector": vector.tolist()}) \
            .with_limit(top_k)
        if filter:
            where = {"path": list(filter.keys())[0],
                     "operator": "Equal",
                     "valueText": list(filter.values())[0]}
            q = q.with_where(where)
        return q.do()


# --- 2.4 pgvector ---
class PgvectorAdapter(VectorDBAdapter):
    name = "pgvector"
    def setup(self):
        import psycopg2
        self.conn = psycopg2.connect(os.environ["POSTGRES_URL"])
        cur = self.conn.cursor()
        cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
        cur.execute("DROP TABLE IF EXISTS bench_test")
        cur.execute(f"""
            CREATE TABLE bench_test (
                id SERIAL PRIMARY KEY,
                embedding vector({DIM}),
                category TEXT, year INT, score FLOAT
            )
        """)
        self.conn.commit()

    def insert_batch(self, vectors, ids, metadatas):
        cur = self.conn.cursor()
        from psycopg2.extras import execute_values
        values = [
            (vectors[i].tolist(), metadatas[i]["category"],
             metadatas[i]["year"], metadatas[i]["score"])
            for i in range(len(vectors))
        ]
        execute_values(
            cur,
            "INSERT INTO bench_test (embedding, category, year, score) VALUES %s",
            values, template="(%s::vector, %s, %s, %s)"
        )
        self.conn.commit()

    def build_index(self):
        cur = self.conn.cursor()
        cur.execute("""
            CREATE INDEX ON bench_test USING hnsw (embedding vector_cosine_ops)
            WITH (m = 16, ef_construction = 200)
        """)
        self.conn.commit()

    def query(self, vector, top_k=10, filter=None):
        cur = self.conn.cursor()
        where = ""
        if filter:
            where = "WHERE " + " AND ".join(f"{k}=%s" for k in filter)
        params = [vector.tolist()] + (list(filter.values()) if filter else [])
        cur.execute(f"""
            SELECT id, embedding <=> %s::vector AS dist
            FROM bench_test {where}
            ORDER BY dist LIMIT {top_k}
        """, params)
        return cur.fetchall()


# --- 2.5 Chroma ---
class ChromaAdapter(VectorDBAdapter):
    name = "chroma"
    def setup(self):
        import chromadb
        self.client = chromadb.PersistentClient(path="./chroma_bench")
        try:
            self.client.delete_collection("bench_test")
        except: pass
        self.coll = self.client.create_collection(
            name="bench_test", metadata={"hnsw:space": "cosine"}
        )

    def insert_batch(self, vectors, ids, metadatas):
        self.coll.add(
            ids=[str(x) for x in ids],
            embeddings=[v.tolist() for v in vectors],
            metadatas=metadatas,
        )

    def query(self, vector, top_k=10, filter=None):
        return self.coll.query(
            query_embeddings=[vector.tolist()],
            n_results=top_k,
            where=filter,
        )


# ============================================================
# 3. Run Benchmark
# ============================================================
def benchmark(adapter: VectorDBAdapter, batch_size: int = 200) -> Dict:
    print(f"\n=== {adapter.name} ===")
    adapter.setup()

    # 写入
    t0 = time.time()
    for i in tqdm(range(0, N_VECTORS, batch_size), desc="insert"):
        adapter.insert_batch(
            all_vectors[i:i + batch_size],
            list(range(i, i + batch_size)),
            all_metadata[i:i + batch_size],
        )
    insert_time = time.time() - t0

    # build index (pgvector需要显式)
    if hasattr(adapter, "build_index"):
        t0 = time.time()
        adapter.build_index()
        index_time = time.time() - t0
    else:
        index_time = 0

    # 等待索引就绪
    time.sleep(5)

    # Query (no filter)
    latencies = []
    for i in tqdm(range(N_QUERIES), desc="query"):
        t0 = time.time()
        adapter.query(query_vectors[i], top_k=10)
        latencies.append((time.time() - t0) * 1000)

    # Query with filter
    filter_latencies = []
    for i in tqdm(range(200), desc="filtered_query"):
        t0 = time.time()
        adapter.query(query_vectors[i], top_k=10,
                      filter={"category": "finance"})
        filter_latencies.append((time.time() - t0) * 1000)

    qps = N_QUERIES / (sum(latencies) / 1000)

    return {
        "db": adapter.name,
        "insert_time_s": round(insert_time, 1),
        "index_time_s": round(index_time, 1),
        "query_p50_ms": round(np.percentile(latencies, 50), 1),
        "query_p95_ms": round(np.percentile(latencies, 95), 1),
        "query_p99_ms": round(np.percentile(latencies, 99), 1),
        "filtered_query_p50_ms": round(np.percentile(filter_latencies, 50), 1),
        "qps_single_thread": round(qps, 1),
    }


def main():
    adapters = [
        PineconeAdapter(),
        QdrantAdapter(),
        WeaviateAdapter(),
        PgvectorAdapter(),
        ChromaAdapter(),
    ]
    results = []
    for a in adapters:
        try:
            r = benchmark(a)
            results.append(r)
            print(json.dumps(r, indent=2))
        except Exception as e:
            print(f"[ERROR] {a.name}: {e}")

    df = pd.DataFrame(results)
    df.to_csv("vdb_bench_results.csv", index=False)
    print("\n=== FINAL RESULTS ===")
    print(df.to_string(index=False))


if __name__ == "__main__":
    main()

---

四、实测结果（100K vectors × 1024 dim × 1000 queries）

测试环境：MacBook Pro M3 Pro 36GB / SaaS DB为各家免费层 / 单线程client

4.1 写入性能

观察：本地DB吞吐 = 网络DB的3-5倍。但生产场景下Pinecone可以并行写入（多worker）线性扩容。

4.2 查询性能（无filter, top_k=10）

重要：网络往返主导Pinecone延迟。同region部署的Pinecone serverless可以做到20-40ms p50。

4.3 Filter Query性能（按category="finance"）

关键洞察：Qdrant的"payload index"让filter几乎免费。pgvector的HNSW + WHERE子句性能掉得严重，是因为HNSW不能与WHERE pre-filter互通（post-filter丢失候选）。

4.4 内存与磁盘

---

五、金融领域应用

5.1 监管报告RAG的特殊需求

某保险公司监管报告RAG的真实需求：

{
    "query": "Solvency II minimum capital requirement",
    "filters": {
        "doc_type": "regulatory_filing",
        "jurisdiction": "EU",
        "year_range": [2022, 2024],
        "language": "en"
    }
}

复合filter下：

• Pinecone: 200ms（pre-filter在server端高效）
• Qdrant: 8ms（payload index完美匹配）
• pgvector: 350ms+（多WHERE让HNSW失效）

→ 监管类高filter场景，Qdrant是最优解。

5.2 多租户金融SaaS

为每个客户隔离数据：

推荐：B2B SaaS要支持几千租户时，Weaviate原生多租户最优；几十租户用Pinecone namespace。

5.3 高频更新场景

某做市商需要实时更新行情相关embedding（每秒数百次）：

• Pinecone: 建议批量update，但写入收费高
• Qdrant: 单点upsert极快（1-3ms）
• Weaviate: 批量更新性能尚可
• pgvector: 每次UPDATE触发HNSW重建，性能差

→ 高频更新选 Qdrant 或 Milvus。

---

六、生产经验：选型决策框架

┌─────────────────── 数据规模 ────────────────────┐
│                                                  │
│   < 1M chunks ─────────────► Chroma / pgvector   │
│   1M - 10M ─────────────────► Qdrant / Weaviate  │
│   10M - 100M ────────────────► Pinecone / Qdrant │
│   > 100M ────────────────────► Milvus / Vespa    │
│                                                  │
├──────────────── 团队运维能力 ───────────────────┤
│                                                  │
│   零运维要求 ──────────────► Pinecone serverless │
│   一定运维能力 ────────────► Qdrant Cloud        │
│   有DevOps团队 ────────────► self-host Qdrant    │
│                                                  │
├──────────────── 已有技术栈 ────────────────────┤
│                                                  │
│   已用Postgres ────────────► pgvector (起步)     │
│   已用ES ──────────────────► Elasticsearch + dense_vector │
│   全新项目 ────────────────► Qdrant (性价比) 或 Pinecone (省心) │
│                                                  │
└──────────────────────────────────────────────────┘

6.1 8个生产坑

---

七、Cost & TCO分析

7.1 不同规模下年度TCO

生产真实账单：某B2B SaaS（2000租户、5M vectors）：Pinecone $400/mo vs Qdrant Cloud $200/mo vs self-host Qdrant $80/mo + 0.5 SRE = $80/mo + 部分人力。

7.2 隐藏成本

• Pinecone的reads费用：$0.40/M reads，10K QPS的应用一天就是 $345
• 写入费：$2/M writes，频繁更新场景成本暴涨
• Egress：跨region流量
• Backup：自动backup vs manual snapshot

---

八、关键速查表

8.1 Vector DB对比矩阵

8.2 HNSW参数调优指南

---

九、面试题

Q1: HNSW vs IVF，什么场景选哪个？

HNSW：召回高、查询快、内存大。适合中小规模(<10M)且对召回敏感（金融、法律）。IVF：内存友好、可disk-based。适合大规模（>10M）且能接受85-95%召回（推荐系统、广告）。实战：现代主流DB都HNSW默认，IVF用于Milvus等超大规模或DiskANN等disk-based方案。

Q2: 为什么pgvector的filter query性能差？

pgvector用HNSW做ANN，但HNSW图本身没有filter信息。两种执行计划：(a) post-filter：先ANN找top-N再过滤，可能N不够；(b) pre-filter：先WHERE再brute force，慢。Postgres planner不一定选最优。修复：(1) 用iterative search（HNSW继续往后找直到filter后够数）；(2) 创建partial index per filter value；(3) 用Qdrant的payload index原生集成filter的方案。

Q3: 你怎么决定一个RAG项目用Pinecone还是Qdrant？

关键问题：(1) 团队是否能self-host？不能选Pinecone; (2) 数据规模？<10M两个都行，>50M Qdrant更经济; (3) 预算？>$500/mo考虑Qdrant; (4) filter复杂度？复杂多条件Qdrant的payload index明显优势; (5) 延迟要求？<10ms p50必须self-host Qdrant。默认建议：原型Pinecone，scale后切Qdrant。

Q4: 解释Pinecone serverless和pod-based的区别？

Pod-based (s1/p1/p2)：固定容量vm，按pod-hour计费，不扩缩容。p1是HNSW、s1是IVF on disk。Serverless：on-demand，按storage + read units + write units计费，自动sharding，更适合spiky流量。取舍：稳定大量写入用pod (s1更便宜)，原型/低流量用serverless。

Q5: 面试官说"我们要存5亿vectors，你选什么vector DB"？

5亿到了"特种部队"领域。选项：(1) Milvus (开源，分布式架构成熟，支持多种索引，业界500亿+案例); (2) Vespa (Yahoo开源，原生支持disk + ANN); (3) Pinecone serverless（管理麻烦但可行，但月度成本上万）。自部署Milvus：3-shard cluster (32GB RAM each) + S3后端，月成本~$2000，QPS 1000+。避坑：Qdrant单节点不行，pgvector早就崩了，Weaviate sharding存在但运维难。

---

十、明日预告

Day 138: Hybrid Search——纯向量检索在金融场景有个致命弱点：找不到精确term（如股票代码"NVDA"、"10-K Item 7"）。明天我们实现 BM25 + dense vector 混合检索，用 Reciprocal Rank Fusion (RRF) 融合两者排名，在我们的金融benchmark上看准确率能否再提升5-10%。