# 向量数据库

---

## 数据库对比矩阵

| 特性 | 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 设置

```python
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 设置

```python
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 设置

```python
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 设置

```python
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 设置

```sql
-- 启用扩展
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);
```

```python
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` | 查询时的搜索宽度 | 越大 = 召回率越高，查询越慢 |

```python
# 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)  # 越大召回率越高
)
```

### 量化扩展

```python
# 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）
```python
# 不同命名空间中的租户数据
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）
```python
# 为所有文档添加 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"))]
    )
)
```

### 每租户独立集合（高隔离性）
```python
# 创建租户专属集合
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 专家** - 异步客户端的实现模式