skillhub-137-rag-architect/references/vector-databases.md

向量数据库

---

数据库对比矩阵

特性	Pinecone	Weaviate	Qdrant	Chroma	pgvector
托管方式	仅托管	托管 + 自托管	托管 + 自托管	自托管（云版内测中）	自托管
混合搜索	支持（稀疏-稠密）	支持（BM25 + 向量）	支持（稀疏向量）	有限	手动（+ pg_trgm）
过滤能力	优秀	优秀	优秀	基础	SQL 原生
最大维度	20,000	无限	65,535	无限	2,000
定价模式	按向量/查询计费	按节点计费	按节点计费	免费（开源）	免费（扩展）
多租户	命名空间	多租户类	集合 + 负载	集合	Schema/RLS
最适合	企业级 SaaS	语义应用	高性能场景	原型开发	PostgreSQL 用户

何时使用各方案

Pinecone

最适合：
- 有严格 SLA 的企业级 RAG
- 希望零基础设施管理的团队
- 需要稀疏-稠密混合搜索的应用
- 成本可预测的高并发生产环境

应避免：
- 成本敏感型项目（大规模时价格昂贵）
- 需要自托管或数据本地化
- 超出元数据的复杂过滤需求
- 希望避免厂商锁定

Weaviate

最适合：
- 带内置向量化的语义搜索
- 多模态（文本、图像）应用
- GraphQL 原生团队
- 需要 BM25 + 向量混合搜索

应避免：
- 仅需简单嵌入存储
- 内存受限的环境
- 不熟悉 GraphQL 的团队

Qdrant

最适合：
- 高性能、低延迟需求
- 带负载索引的复杂过滤
- Rust/性能优先团队
- 自托管，完全控制

应避免：
- 希望完全托管便捷性的团队
- 偏好 GraphQL（仅支持 REST/gRPC）

Chroma

最适合：
- 本地开发与原型验证
- LangChain/LlamaIndex 集成
- 简单 RAG 概念验证
- 教学项目

应避免：
- 大规模生产负载
- 多租户应用
- 高可用性需求

pgvector

最适合：
- 已有 PostgreSQL 基础设施
- 事务 + 向量在同一数据库
- SQL 原生团队
- 成本优化（无需新增基础设施）

应避免：
- 向量维度超过 2000
- 数十亿级向量（扩展性限制）
- 亚毫秒级延迟需求

---

Pinecone 设置

from pinecone import Pinecone, ServerlessSpec

# 初始化客户端
pc = Pinecone(api_key="your-api-key")

# 创建无服务器索引
pc.create_index(
    name="rag-index",
    dimension=1536,  # OpenAI ada-002
    metric="cosine",
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"
    )
)

# 获取索引引用
index = pc.Index("rag-index")

# 上传向量及其元数据
index.upsert(
    vectors=[
        {
            "id": "doc-1",
            "values": embedding_vector,
            "metadata": {
                "source": "manual.pdf",
                "page": 42,
                "section": "installation",
                "tenant_id": "acme-corp"
            }
        }
    ],
    namespace="production"
)

# 带元数据过滤的查询
results = index.query(
    vector=query_embedding,
    top_k=10,
    include_metadata=True,
    namespace="production",
    filter={
        "tenant_id": {"$eq": "acme-corp"},
        "section": {"$in": ["installation", "setup"]}
    }
)

# 混合搜索（稀疏-稠密）
from pinecone_text.sparse import BM25Encoder

bm25 = BM25Encoder()
bm25.fit(corpus)  # 在文档集上拟合

results = index.query(
    vector=dense_embedding,
    sparse_vector=bm25.encode_queries(query_text),
    top_k=10,
    alpha=0.5  # 平衡稠密与稀疏权重
)

---

Weaviate 设置

import weaviate
from weaviate.classes.config import Configure, Property, DataType

# 连接到 Weaviate Cloud
client = weaviate.connect_to_weaviate_cloud(
    cluster_url="https://your-cluster.weaviate.network",
    auth_credentials=weaviate.auth.AuthApiKey("your-api-key")
)

# 或自托管
client = weaviate.connect_to_local(
    host="localhost",
    port=8080
)

# 创建带向量化器的集合
client.collections.create(
    name="Document",
    vectorizer_config=Configure.Vectorizer.text2vec_openai(
        model="text-embedding-3-small"
    ),
    properties=[
        Property(name="content", data_type=DataType.TEXT),
        Property(name="source", data_type=DataType.TEXT),
        Property(name="page", data_type=DataType.INT),
        Property(name="tenant_id", data_type=DataType.TEXT, index_filterable=True)
    ]
)

# 自动向量化插入
documents = client.collections.get("Document")
documents.data.insert(
    properties={
        "content": "Installation guide content...",
        "source": "manual.pdf",
        "page": 42,
        "tenant_id": "acme-corp"
    }
)

# 或使用预计算向量
documents.data.insert(
    properties={"content": "...", "source": "..."},
    vector=precomputed_embedding
)

# 混合搜索（BM25 + 向量）
from weaviate.classes.query import MetadataQuery, Filter

results = documents.query.hybrid(
    query="how to install",
    alpha=0.5,  # 0=仅 BM25，1=仅向量
    limit=10,
    filters=Filter.by_property("tenant_id").equal("acme-corp"),
    return_metadata=MetadataQuery(score=True, explain_score=True)
)

for obj in results.objects:
    print(f"Score: {obj.metadata.score}, Content: {obj.properties['content'][:100]}")

client.close()

---

Qdrant 设置

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue,
    PayloadSchemaType
)

# 连接到 Qdrant Cloud
client = QdrantClient(
    url="https://your-cluster.qdrant.io",
    api_key="your-api-key"
)

# 或本地连接
client = QdrantClient(host="localhost", port=6333)

# 创建集合
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,
        distance=Distance.COSINE
    )
)

# 创建负载索引以实现快速过滤
client.create_payload_index(
    collection_name="documents",
    field_name="tenant_id",
    field_schema=PayloadSchemaType.KEYWORD
)

# 上传数据点
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id="doc-1",
            vector=embedding_vector,
            payload={
                "content": "Installation guide...",
                "source": "manual.pdf",
                "page": 42,
                "tenant_id": "acme-corp"
            }
        )
    ]
)

# 带过滤条件的搜索
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    limit=10,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="tenant_id",
                match=MatchValue(value="acme-corp")
            )
        ]
    ),
    with_payload=True
)

# 大批量数据上传
from qdrant_client.models import Batch

client.upsert(
    collection_name="documents",
    points=Batch(
        ids=ids_list,
        vectors=vectors_list,
        payloads=payloads_list
    )
)

---

Chroma 设置

import chromadb
from chromadb.config import Settings

# 持久化本地存储
client = chromadb.PersistentClient(
    path="./chroma_data",
    settings=Settings(anonymized_telemetry=False)
)

# 使用自定义嵌入函数创建集合
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction

embedding_fn = OpenAIEmbeddingFunction(
    api_key="your-openai-key",
    model_name="text-embedding-3-small"
)

collection = client.get_or_create_collection(
    name="documents",
    embedding_function=embedding_fn,
    metadata={"hnsw:space": "cosine"}
)

# 添加文档（自动嵌入）
collection.add(
    ids=["doc-1", "doc-2"],
    documents=["Installation guide...", "Configuration steps..."],
    metadatas=[
        {"source": "manual.pdf", "page": 42},
        {"source": "manual.pdf", "page": 43}
    ]
)

# 或使用预计算嵌入
collection.add(
    ids=["doc-3"],
    embeddings=[precomputed_vector],
    metadatas=[{"source": "guide.pdf"}],
    documents=["Original text for reference"]
)

# 查询
results = collection.query(
    query_texts=["how to install"],
    n_results=10,
    where={"source": "manual.pdf"},
    include=["documents", "metadatas", "distances"]
)

# 更新已有文档
collection.update(
    ids=["doc-1"],
    documents=["Updated installation guide..."],
    metadatas=[{"source": "manual_v2.pdf", "page": 42}]
)

---

pgvector 设置

-- 启用扩展
CREATE EXTENSION IF NOT EXISTS vector;

-- 创建含向量列的表
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    embedding vector(1536),  -- OpenAI 维度
    source VARCHAR(255),
    page INTEGER,
    tenant_id VARCHAR(100),
    created_at TIMESTAMP DEFAULT NOW()
);

-- 创建 HNSW 索引（大多数情况下推荐）
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- 或针对超大数据集使用 IVFFlat
CREATE INDEX ON documents
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

-- 在过滤列上创建索引
CREATE INDEX ON documents (tenant_id);

import psycopg2
from pgvector.psycopg2 import register_vector

conn = psycopg2.connect("postgresql://localhost/ragdb")
register_vector(conn)

# 插入嵌入向量
cur = conn.cursor()
cur.execute(
    """
    INSERT INTO documents (content, embedding, source, page, tenant_id)
    VALUES (%s, %s, %s, %s, %s)
    RETURNING id
    """,
    ("Installation guide...", embedding_vector, "manual.pdf", 42, "acme-corp")
)

# 带过滤条件的相似度搜索
cur.execute(
    """
    SELECT id, content, source, page,
           1 - (embedding <=> %s) AS similarity
    FROM documents
    WHERE tenant_id = %s
    ORDER BY embedding <=> %s
    LIMIT 10
    """,
    (query_embedding, "acme-corp", query_embedding)
)

results = cur.fetchall()

# 使用 pg_trgm 进行混合搜索
cur.execute(
    """
    SELECT id, content,
           (0.5 * (1 - (embedding <=> %s))) +
           (0.5 * similarity(content, %s)) AS hybrid_score
    FROM documents
    WHERE tenant_id = %s
      AND content %% %s  -- Trigram 相似度阈值
    ORDER BY hybrid_score DESC
    LIMIT 10
    """,
    (query_embedding, query_text, "acme-corp", query_text)
)

---

索引调优指南

HNSW 参数

参数	描述	权衡
m	每个节点的连接数	越大 = 召回率越高，内存占用越大
ef_construction	构建时的搜索宽度	越大 = 索引质量越好，构建越慢
ef_search	查询时的搜索宽度	越大 = 召回率越高，查询越慢

# Qdrant HNSW 调优
client.update_collection(
    collection_name="documents",
    hnsw_config=HnswConfigDiff(
        m=16,                    # 默认值：16，提高召回率可增大
        ef_construct=100,        # 默认值：100，提高索引质量可增大
        full_scan_threshold=10000  # 低于此大小时使用暴力搜索
    )
)

# 查询时 ef 调整
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    limit=10,
    search_params=SearchParams(hnsw_ef=128)  # 越大召回率越高
)

量化扩展

# Qdrant 标量量化（内存减少 4 倍）
from qdrant_client.models import ScalarQuantization, ScalarQuantizationConfig

client.update_collection(
    collection_name="documents",
    quantization_config=ScalarQuantization(
        scalar=ScalarQuantizationConfig(
            type="int8",
            quantile=0.99,
            always_ram=True
        )
    )
)

---

多租户模式

命名空间隔离（Pinecone）

# 不同命名空间中的租户数据
index.upsert(vectors=[...], namespace="tenant-acme")
index.upsert(vectors=[...], namespace="tenant-globex")

# 在租户命名空间内查询
results = index.query(
    vector=query_embedding,
    namespace="tenant-acme",
    top_k=10
)

元数据过滤（Qdrant/Weaviate）

# 为所有文档添加 tenant_id
point = PointStruct(
    id="doc-1",
    vector=embedding,
    payload={"tenant_id": "acme", "content": "..."}
)

# 始终按租户过滤
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[FieldCondition(key="tenant_id", match=MatchValue(value="acme"))]
    )
)

每租户独立集合（高隔离性）

# 创建租户专属集合
client.create_collection(
    collection_name=f"docs_{tenant_id}",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)

---

决策流程图

开始
  │
  ├─ 需要托管服务且零运维？
  │   └─ 是 → Pinecone
  │
  ├─ 已有 PostgreSQL？
  │   └─ 是 → pgvector（若向量维度 < 2000）
  │
  ├─ 需要内置向量化？
  │   └─ 是 → Weaviate
  │
  ├─ 需要极致性能 + 自托管？
  │   └─ 是 → Qdrant
  │
  ├─ 原型开发 / 本地开发？
  │   └─ 是 → Chroma
  │
  └─ 默认推荐 → Qdrant（功能与性能的平衡）

---

快速参考

任务	Pinecone	Weaviate	Qdrant	pgvector
创建索引/集合	create_index()	collections.create()	create_collection()	CREATE TABLE
插入	upsert()	data.insert()	upsert()	INSERT
搜索	query()	query.near_vector()	search()	ORDER BY <=>
过滤	filter={}	Filter.by_property()	query_filter=Filter()	WHERE
删除	delete()	data.delete_by_id()	delete()	DELETE
混合搜索	sparse_vector 参数	query.hybrid()	sparse vectors	手动实现

相关技能

• 数据库优化器 - 索引调优与查询性能
• 云架构师 - 向量数据库托管的基础设施决策
• Python 专家 - 异步客户端的实现模式