notes / saas / 向量数据库选型

🧮 向量数据库选型

LanceDB · ChromaDB · Pinecone · Weaviate · Qdrant · Milvus · pgvector · Vald
从嵌入式到云原生，选对向量数据库是 RAG 和 Agent 系统的地基

01 为什么需要向量数据库？

传统数据库用精确匹配和关键词搜索——你搜 "apple"，它只返回包含 "apple" 的行。但语言不是这样工作的：

什么时候 不需要 向量数据库？

• 数据量 < 10 万条，内存里 numpy 暴力搜索就够了
• 只需要精确匹配（用户 ID、订单号、SKU）
• 全文搜索已满足需求（Meilisearch / Typesense）
• 不需要语义理解，只需要前缀匹配和 faceted 过滤

02 向量搜索基础：Embedding、索引与距离度量

Embedding：数据的数字指纹

Embedding 模型将非结构化数据转化为固定维度的浮点向量。常见模型：

距离度量

cos(A, B) = (A · B) / (||A|| × ||B||)

L2(A, B) = sqrt(Σ(Ai - Bi)²)

实践建议：大多数场景用余弦相似度。如果 Embedding 模型已做归一化（OpenAI 的就是），内积 = 余弦，直接用 dot product 更快。

ANN 索引：速度的魔法

精确 KNN 搜索是 O(n)，百万级以上不可接受。ANN (Approximate Nearest Neighbor) 用索引换取速度：

03 LanceDB — 嵌入式 Rust 向量数据库

LanceDB 是一个基于 Lance 列式格式的嵌入式向量数据库，用 Rust 编写。它不运行服务器——你的应用直接读写磁盘上的文件。类似于 SQLite 之于传统数据库，LanceDB 是向量数据库的 "SQLite"。

核心架构

优势

• 零运维：无服务器进程，无需部署、配置、监控
• 极低延迟：嵌入式，无网络开销，本地读取 <1ms
• 云存储原生：直接读写 S3/GCS/Azure，无需下载到本地
• 版本化：每次写入创建新版本，支持时间旅行查询
• Lance 格式：列式存储，零拷贝读取，自动碎片整理
• 多语言 SDK：Python、JavaScript/TypeScript、Rust

劣势

• 无内置 RBAC：安全依赖应用层或存储层（S3 IAM 等）
• 并发写入受限：嵌入式架构，同一表同一时刻只有一个写入者
• 生态较新：2023 年开源，社区和工具链还在成长
• 大规模分布式：不如 Milvus 原生支持百亿级分布式

代码示例

# pip install lancedb
import lancedb
import numpy as np

# 连接（本地目录，自动创建）
db = lancedb.connect("./my_lance_db")

# 创建表
data = [
    {"text": "向量数据库是 AI 应用的基础设施", "vector": np.random.rand(768).tolist(), "source": "doc1"},
    {"text": "LanceDB 用 Rust 编写，嵌入式架构", "vector": np.random.rand(768).tolist(), "source": "doc2"},
    {"text": "HNSW 索引在百万级数据上仅需 1ms", "vector": np.random.rand(768).tolist(), "source": "doc3"},
]
table = db.create_table("documents", data)

# 向量搜索
query_vector = np.random.rand(768).tolist()
results = table.search(query_vector).limit(5).to_pandas()
print(results[["text", "source", "_distance"]])

# 创建 IVF-PQ 索引加速（数据量大时）
table.create_index(
    num_partitions=256,   # IVF 聚类数
    num_sub_vectors=96,   # PQ 量化子向量数
)

# 过滤搜索：元数据过滤 + 向量搜索
results = (
    table.search(query_vector)
    .where("source = 'doc1'", prefilter=True)
    .limit(5)
    .to_pandas()
)

# 时间旅行：查看上一版本的数据
old_version = db.open_table("documents", version=0)
print(f"当前版本: {table.version}, 旧版本: 0")

# S3 远程存储
db_remote = lancedb.connect("s3://my-bucket/lancedb")

何时选 LanceDB

• ✅ 个人项目、原型、MVP——不想部署任何服务
• ✅ Edge / Serverless 场景——Lambda、Vercel Edge Function
• ✅ 数据量 < 1000 万条，无需分布式
• ✅ 需要版本化和时间旅行
• ✅ 已有 S3/GCS 存储基础设施，想直接利用

04 ChromaDB — AI 原生的 Python 向量数据库

ChromaDB（简称 Chroma）是为 AI 应用设计的开源向量数据库，最大特点是 内置 Embedding 函数——你不需要自己调 OpenAI API 做向量化，Chroma 自动帮你做。它是 LangChain 生态最紧密集成的向量数据库。

核心特性

• 内置 Embedding：默认用 all-MiniLM-L6-v2，一行代码配置 OpenAI/Cohere
• 双模式：嵌入式（PersistentClient）和 Client-Server（HttpClient）
• 元数据过滤：支持 $eq, $ne, $gt, $gte, $in, $nin 等操作符
• Collection 抽象：数据按 Collection 隔离，类似数据库的表

优势

• 最简单的上手体验：pip install chromadb → 3 行代码开始用
• 内置 Embedding：省去自己管理 Embedding 管道
• LangChain 集成最深：官方维护，文档示例最丰富
• Python 原生：API 设计非常 Pythonic

劣势

• 性能天花板：Python 实现内核，大规模下性能不如 Rust/Go/C++ 方案
• 缺乏高级索引：默认 HNSW，没有 IVF-PQ 等量化索引选项
• 生产就绪度：Server 模式在高并发下不够稳定，适合中小规模
• 嵌入式写入瓶颈：SQLite 后端，大吞吐写入受限

代码示例

# pip install chromadb
import chromadb

# 嵌入式模式（数据持久化到磁盘）
client = chromadb.PersistentClient(path="./chroma_db")

# 创建 Collection（自动使用默认 Embedding 函数）
collection = client.get_or_create_collection(
    name="knowledge_base",
    metadata={"hnsw:space": "cosine"}  # 距离度量
)

# 添加文档——Chroma 自动做 Embedding！
collection.add(
    documents=[
        "向量数据库是 AI 时代的存储基础设施",
        "ChromaDB 让你 3 行代码开始用向量搜索",
        "RAG 系统的核心是检索质量",
    ],
    metadatas=[
        {"source": "intro", "page": 1},
        {"source": "quickstart", "page": 1},
        {"source": "rag-guide", "page": 3},
    ],
    ids=["doc1", "doc2", "doc3"]
)

# 语义搜索——直接传文本，Chroma 自动 Embedding
results = collection.query(
    query_texts=["什么是向量数据库？"],
    n_results=3,
    where={"source": "intro"},  # 元数据过滤
)
print(results["documents"])

# 也可以手动传向量（用自己的 Embedding 模型）
import openai
resp = openai.embeddings.create(
    model="text-embedding-3-small",
    input="什么是向量数据库？"
)
query_vector = resp.data[0].embedding

results = collection.query(
    query_embeddings=[query_vector],
    n_results=3,
)

何时选 ChromaDB

• ✅ 快速原型、Jupyter 实验、Hackathon
• ✅ 数据量 < 100 万条
• ✅ 已在用 LangChain / LlamaIndex，想要最省心的集成
• ✅ 团队纯 Python 栈，不想碰 Rust/Go 生态

05 Pinecone — 全托管云向量数据库

Pinecone 是向量数据库领域的先行者和商业化最成功的案例。你不需要管理任何基础设施——注册账号、创建 Index、开始写入和查询。2024 年推出的 Serverless 模式更是零固定成本起步。

架构特点

• Serverless 模式：按读写计费，空闲时成本为 $0
• Pod 模式：预留容量，适合稳定高吞吐场景
• 自动扩缩容：无需手动分片或调参
• Namespaces：Index 内的逻辑隔离，多租户利器
• Sparse-Dense 索引：原生支持混合搜索（关键词 + 语义）

优势

• 零运维：真正的全托管，不需要知道什么是 HNSW
• 企业级 SLA：99.9% 可用性承诺
• 混合搜索：Sparse-Dense 索引，BM25 + 向量联合
• 多租户：Namespace 原生隔离，查询时指定即可
• 生态集成：LangChain、LlamaIndex、Vercel AI SDK 等

劣势

• 闭源：数据锁定，无法自建或审计底层
• 成本：百万级以上数据量，月费显著高于自建
• 延迟：网络往返 ~20-50ms，比嵌入式方案慢 10-50x
• 定制受限：无法调整索引参数、无法做磁盘级优化

代码示例

# pip install pinecone
from pinecone import Pinecone

pc = Pinecone(api_key="your-api-key")

# 创建 Serverless Index
pc.create_index(
    name="knowledge",
    dimension=1536,
    metric="cosine",
    spec={
        "serverless": {
            "cloud": "aws",
            "region": "us-east-1"
        }
    }
)

index = pc.Index("knowledge")

# 批量 Upsert
index.upsert(
    vectors=[
        {"id": "doc1", "values": [0.1, 0.2, ...], "metadata": {"source": "wiki", "lang": "zh"}},
        {"id": "doc2", "values": [0.3, 0.4, ...], "metadata": {"source": "blog", "lang": "en"}},
    ],
    namespace="production"  # 多租户隔离
)

# 查询 + 元数据过滤
results = index.query(
    vector=[0.15, 0.25, ...],
    top_k=10,
    filter={"lang": {"$eq": "zh"}, "source": {"$in": ["wiki", "blog"]}},
    namespace="production",
    include_metadata=True,
)

# 混合搜索（Sparse-Dense Index）
results = index.query(
    vector={"values": [0.15, 0.25, ...], "sparse_values": {"indices": [2, 5, 10], "values": [0.3, 0.5, 0.2]}},
    top_k=10,
    include_metadata=True,
)

定价参考（2026）

何时选 Pinecone

• ✅ 团队没有基础设施经验，只想快速上线
• ✅ 需要混合搜索（关键词 + 语义）
• ✅ 企业级 SLA 要求
• ✅ 预算允许，数据量在百万级以下

06 Weaviate — 多模态知识图谱向量库

Weaviate 是一个将向量搜索与知识图谱融合的数据库。它最独特的卖点是 Modules 系统——内置多种 Embedding 模型的集成（OpenAI、Cohere、HuggingFace、多模态 CLIP 等），以及 Generative Search（直接在数据库层调用 LLM 做 RAG）。

核心特性

• Generative Search：查询结果直接送入 LLM，数据库层做 RAG
• Multi-tenancy：每个租户独立 Index 物理隔离
• Modules：可插拔的向量化模块，支持文本/图片/视频
• GraphQL API：查询语言是 GraphQL，表达力强
• Cross-refs：对象之间可以建交叉引用，形成知识图谱

优势

• 内置 RAG：with_generate() 一行做 RAG，无需自己编排
• 多模态原生：文本+图片+视频统一搜索
• 真正的多租户：物理隔离，不是逻辑隔离
• GraphQL：复杂查询表达力远超 REST

劣势

• Go 编写：性能不错但不如 Rust 方案极致
• 学习曲线：GraphQL + Modules + Schema 概念较多
• 内存消耗：默认全量内存，大数据集内存成本高
• Generative Search 耦合：RAG 逻辑绑在数据库层，灵活性受限

代码示例

# pip install weaviate-client
import weaviate
import weaviate.classes as wvc

# 连接本地 Weaviate
client = weaviate.connect_to_local()

# 创建 Collection
collection = client.collections.create(
    name="Article",
    properties=[
        wvc.config.Property(name="title", data_type=wvc.config.DataType.TEXT),
        wvc.config.Property(name="content", data_type=wvc.config.DataType.TEXT),
        wvc.config.Property(name="category", data_type=wvc.config.DataType.TEXT),
    ],
    vectorizer_config=wvc.config.Configure.Vectorizer.text2vec_openai(
        model="text-embedding-3-small"
    ),
    # 多租户
    multi_tenancy_config=wvc.config.Configure.multi_tenancy(enabled=True),
)

# 插入数据（自动向量化）
collection.with_tenant("tenant_A").data.insert({
    "title": "向量数据库入门",
    "content": "向量数据库是 AI 应用的核心基础设施...",
    "category": "tutorial",
})

# 语义搜索
results = collection.with_tenant("tenant_A").query.near_text(
    query="如何选择向量数据库？",
    limit=5,
    filters=wvc.query.Filter.by_property("category").equal("tutorial"),
)

# Generative Search（数据库层 RAG！）
from weaviate.classes.query import MetadataQuery
results = collection.with_tenant("tenant_A").generate.near_text(
    query="总结这些文章的关键要点",
    limit=5,
    grouped_task="请用中文总结以上文章的三个核心观点",
)
print(results.generated)

何时选 Weaviate

• ✅ 需要多模态搜索（文本+图片+视频）
• ✅ 需要知识图谱交叉引用
• ✅ 多租户物理隔离是硬需求
• ✅ 想把 RAG 逻辑下沉到数据库层

07 Qdrant — 高性能 Rust 向量搜索引擎

Qdrant 是用 Rust 编写的高性能向量搜索引擎。它在内存效率和查询延迟方面表现优异，特别适合需要 复杂过滤 + 向量搜索 的场景——这是很多 SaaS 应用（电商、内容平台）的核心需求。

核心特性

• Filterable HNSW：在 HNSW 图遍历时直接跳过不满足过滤条件的节点，性能极高
• WAL 持久化：Write-Ahead Log 保证数据安全
• Payload 索引：对元数据建 B+ 树索引，过滤快如 SQL
• 量化支持：Scalar Quantization、Product Quantization、Binary Quantization
• 分片 & 分布式：支持水平扩展

优势

• Rust 性能：单机百万级 QPS，延迟 <1ms
• 过滤性能最强：Filterable HNSW 是同类最优实现
• 内存效率：量化后内存占用可降至 1/4
• gRPC + REST：双 API 风格满足不同需求
• 云托管：Qdrant Cloud 提供全托管服务

劣势

• 无内置 Embedding：需要自己管理向量化管道
• 分布式复杂：分片和再平衡需要手动管理
• 社区规模：比 Pinecone/Milvus 小

代码示例

# pip install qdrant-client
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue

client = QdrantClient(host="localhost", port=6333)

# 创建 Collection
client.create_collection(
    collection_name="products",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

# 插入数据
client.upsert(
    collection_name="products",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],
            payload={"name": "MacBook Pro", "category": "laptop", "price": 14999}
        ),
        PointStruct(
            id=2,
            vector=[0.3, 0.4, ...],
            payload={"name": "iPhone 18", "category": "phone", "price": 8999}
        ),
    ]
)

# 向量搜索 + 复杂过滤（Qdrant 的杀手锏！）
results = client.search(
    collection_name="products",
    query_vector=[0.15, 0.25, ...],
    query_filter=Filter(
        must=[
            FieldCondition(key="category", match=MatchValue(any=["laptop", "phone"])),
            FieldCondition(key="price", range={"lte": 15000}),
        ]
    ),
    limit=10,
)

# 启用量化节省内存
client.update_collection(
    collection_name="products",
    quantization_config=models.ScalarQuantization(
        scalar=models.ScalarQuantizationConfig(
            type=models.ScalarType.INT8,
            quantile=0.99,
            always_ram=True,
        ),
    ),
)

何时选 Qdrant

• ✅ 需要高性能 + 复杂过滤（电商、内容推荐）
• ✅ 对延迟敏感（<5ms 要求）
• ✅ 需要量化降低内存成本
• ✅ Rust 技术栈，想要可控的自建方案

08 Milvus — 云原生分布式向量数据库

Milvus 是面向超大规模场景的分布式向量数据库，背靠 Zilliz 公司。当数据量超过 10 亿条、需要多机并行搜索时，Milvus 是最成熟的选择。Zilliz Cloud 提供全托管版本。

架构：存算分离

优势

• 百亿级能力：存算分离架构，水平扩展无上限
• 索引最丰富：HNSW、IVF_FLAT、IVF_PQ、IVF_SQ8、SCANN、GPU 索引
• 多语言 SDK：Python、Java、Go、Node.js、C++
• 企业级特性：RBAC、多租户、资源组、动态扩缩
• Zilliz Cloud：全托管，起步价低

劣势

• 架构复杂：依赖 etcd + MinIO + Pulsar，部署和运维成本高
• 小题大做：百万级数据用 Milvus 是浪费
• 延迟较高：分布式架构下，单次查询 ~10-50ms
• 学习曲线：概念多（Collection → Segment → Channel → Partition）

代码示例

# pip install pymilvus
from pymilvus import MilvusClient, DataType

client = MilvusClient(uri="http://localhost:19530")

# 创建 Collection
client.create_collection(
    collection_name="articles",
    dimension=1536,
    metric_type="COSINE",
    auto_id=True,
)

# 也可以用 Schema 定义（支持标量字段 + 向量字段）
from pymilvus import CollectionSchema, FieldSchema, Collection, AnnIndex
schema = CollectionSchema(fields=[
    FieldSchema("id", DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema("title", DataType.VARCHAR, max_length=512),
    FieldSchema("category", DataType.VARCHAR, max_length=64),
    FieldSchema("embedding", DataType.FLOAT_VECTOR, dim=1536),
])
collection = Collection("articles_v2", schema)

# 创建索引
index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 16, "efConstruction": 256}
)
client.create_index("articles_v2", index_params)

# 插入 + 搜索
data = [
    {"title": "向量数据库选型指南", "category": "tutorial", "embedding": [0.1, 0.2, ...]},
]
client.insert("articles_v2", data)

results = client.search(
    collection_name="articles_v2",
    data=[[0.15, 0.25, ...]],
    limit=10,
    filter='category == "tutorial"',
    output_fields=["title", "category"],
)

何时选 Milvus

• ✅ 数据量 10 亿+，必须分布式
• ✅ 团队有 K8s 运维能力
• ✅ 需要多种索引类型（GPU 加速、PQ 量化）
• ✅ 企业级需求：RBAC、资源隔离、审计

09 PostgreSQL + pgvector — 不加新组件的向量搜索

pgvector 是 PostgreSQL 的向量搜索扩展。如果你已经在用 PostgreSQL，它是加入向量搜索 成本最低 的方式——不需要新服务、新运维、新 SDK。

核心特性

• SQL 原生：SELECT * FROM docs ORDER BY embedding <=> '[0.1,0.2,...]' LIMIT 10
• ACID 事务：向量数据和业务数据在同一个事务里
• JOIN 查询：向量搜索结果直接 JOIN 业务表
• HNSW + IVFFlat：两种索引可选
• pgvectorscale：Timescale 的扩展，增加 DiskANN 索引和 Streaming Filtering

优势

• 零额外运维：用现有的 PostgreSQL 即可
• 事务一致性：向量写入和业务写入原子完成
• SQL 灵活性：复杂的过滤、聚合、JOIN 都可以做
• 生态成熟：所有 ORM（Prisma、SQLAlchemy、Drizzle）都支持

劣势

• 性能上限：百万级以上 HNSW 延迟 >10ms，不如专用向量库
• 内存压力：HNSW 索引必须常驻内存，大数据集吃内存
• 功能有限：没有混合搜索、没有多模态、没有自动向量化
• 写放大：频繁更新向量时，HNSW 索引重建开销大

代码示例

-- 安装扩展
CREATE EXTENSION IF NOT EXISTS vector;

-- 创建表
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT,
    source VARCHAR(255),
    embedding vector(1536)
);

-- 插入数据
INSERT INTO documents (content, source, embedding)
VALUES ('向量数据库入门', 'tutorial', '[0.1, 0.2, ...]'::vector);

-- 创建 HNSW 索引
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- 向量搜索 + 过滤 + JOIN！
SELECT d.content, d.source, 1 - (d.embedding <=> '[0.15, 0.25, ...]'::vector) AS similarity
FROM documents d
JOIN user_permissions up ON d.source = up.allowed_source
WHERE up.user_id = 42
  AND d.source != 'internal'
ORDER BY d.embedding <=> '[0.15, 0.25, ...]'::vector
LIMIT 10;

-- 设置搜索参数
SET hnsw.ef_search = 100;  -- 越高召回率越好，越慢

# Python (SQLAlchemy)
from pgvector.sqlalchemy import Vector
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Document(Base):
    __tablename__ = "documents"
    id = Column(Integer, primary_key=True)
    content = Column(String)
    source = Column(String)
    embedding = Column(Vector(1536))

# 查询
from sqlalchemy import select
stmt = (
    select(Document, Document.embedding.cosine_distance(query_vec).label("distance"))
    .order_by(Document.embedding.cosine_distance(query_vec))
    .limit(10)
)
results = session.execute(stmt).all()

何时选 pgvector

• ✅ 已有 PostgreSQL，不想引入新组件
• ✅ 数据量 < 500 万条
• ✅ 需要向量搜索和业务查询在同一个事务里
• ✅ 过滤逻辑复杂，需要 JOIN 多张表

10 全面对比表

11 决策框架：我该选什么？

场景速查

12 实战代码示例：统一的向量搜索服务

以下是一个生产级的向量搜索服务，抽象了底层引擎，支持运行时切换：

# vector_service.py — 统一向量搜索服务
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Dict, Optional
import numpy as np

@dataclass
class VectorDocument:
    id: str
    text: str
    vector: List[float]
    metadata: Dict

@dataclass
class SearchResult:
    id: str
    text: str
    score: float
    metadata: Dict

class VectorStore(ABC):
    """向量存储抽象层"""
    @abstractmethod
    def upsert(self, docs: List[VectorDocument]) -> int: ...
    @abstractmethod
    def search(self, query_vector: List[float], top_k: int = 10,
               filters: Optional[Dict] = None) -> List[SearchResult]: ...
    @abstractmethod
    def delete(self, ids: List[str]) -> int: ...


class QdrantStore(VectorStore):
    """Qdrant 实现 — 适合生产环境"""
    def __init__(self, collection_name: str, dim: int = 1536):
        from qdrant_client import QdrantClient
        from qdrant_client.models import Distance, VectorParams
        self.client = QdrantClient(host="localhost", port=6333)
        self.collection = collection_name
        # 自动创建 Collection
        try:
            self.client.create_collection(
                collection_name=collection_name,
                vectors_config=VectorParams(size=dim, distance=Distance.COSINE),
            )
        except Exception:
            pass  # Collection 已存在

    def upsert(self, docs: List[VectorDocument]) -> int:
        from qdrant_client.models import PointStruct
        points = [
            PointStruct(id=hash(d.id) % (2**63), vector=d.vector, payload={
                "text": d.text, **d.metadata
            })
            for d in docs
        ]
        self.client.upsert(self.collection, points)
        return len(points)

    def search(self, query_vector, top_k=10, filters=None) -> List[SearchResult]:
        from qdrant_client.models import Filter, FieldCondition, MatchValue
        qdrant_filter = None
        if filters:
            conditions = [
                FieldCondition(key=k, match=MatchValue(value=v))
                for k, v in filters.items()
            ]
            qdrant_filter = Filter(must=conditions)

        results = self.client.search(
            collection_name=self.collection,
            query_vector=query_vector,
            limit=top_k,
            query_filter=qdrant_filter,
        )
        return [
            SearchResult(id=str(r.id), text=r.payload.get("text", ""),
                        score=r.score, metadata=r.payload)
            for r in results
        ]

    def delete(self, ids: List[str]) -> int:
        self.client.delete(
            collection_name=self.collection,
            points_selector=[hash(id_) % (2**63) for id_ in ids],
        )
        return len(ids)


class LanceDBStore(VectorStore):
    """LanceDB 实现 — 适合开发/Edge"""
    def __init__(self, db_path: str = "./vectordb", dim: int = 1536):
        import lancedb
        self.db = lancedb.connect(db_path)
        self.dim = dim

    def upsert(self, docs: List[VectorDocument], table_name: str = "default") -> int:
        import pandas as pd
        df = pd.DataFrame([{
            "id": d.id, "text": d.text, "vector": d.vector, **d.metadata
        } for d in docs])
        try:
            table = self.db.open_table(table_name)
            table.add(df)
        except Exception:
            self.db.create_table(table_name, df)
        return len(docs)

    def search(self, query_vector, top_k=10, filters=None, table_name="default") -> List[SearchResult]:
        table = self.db.open_table(table_name)
        query = table.search(query_vector).limit(top_k)
        if filters:
            for k, v in filters.items():
                query = query.where(f"{k} = '{v}'")
        results = query.to_pandas()
        return [
            SearchResult(id=str(r["id"]), text=r["text"],
                        score=1 - r["_distance"], metadata={})
            for _, r in results.iterrows()
        ]

    def delete(self, ids: List[str], table_name: str = "default") -> int:
        # LanceDB 通过 SQL 删除
        table = self.db.open_table(table_name)
        ids_str = ", ".join(f"'{i}'" for i in ids)
        table.delete(f"id IN ({ids_str})")
        return len(ids)


# 使用示例
def main():
    # 开发环境用 LanceDB
    store = LanceDBStore() if os.getenv("ENV") == "dev" else QdrantStore("production")

    docs = [
        VectorDocument("1", "RAG 系统需要向量数据库", [0.1]*1536, {"source": "tutorial"}),
        VectorDocument("2", "LanceDB 是嵌入式向量数据库", [0.2]*1536, {"source": "docs"}),
    ]
    store.upsert(docs)

    results = store.search([0.15]*1536, top_k=5, filters={"source": "tutorial"})
    for r in results:
        print(f"[{r.score:.3f}] {r.text}")

13 进阶主题：混合搜索、多租户、过滤

混合搜索：关键词 + 语义的强强联合

纯向量搜索擅长语义匹配，但精确关键词（产品名、SKU、缩写）可能丢失。混合搜索是 2026 年的标准做法：

多租户隔离

过滤性能：Post-filter vs Pre-filter

• Pre-filter：先过滤元数据，再在结果集上做向量搜索。适合过滤条件能大幅缩小结果集的场景。
• Post-filter：先做向量搜索取 Top-K*2，再过滤。适合过滤条件宽松的场景。
• Filterable HNSW（Qdrant 独有）：在 HNSW 图遍历时直接跳过不满足条件的节点，兼顾速度和召回率。

14 与 Agent 研究的关联

向量数据库是 Agent 系统的"记忆层"。在 Agent 架构研究 中，多个项目展示了不同的向量存储策略：

Agent 场景下的向量数据库选型建议

15 关联模块