11c - 向量数据库对比

from pinecone import Pinecone
 
pc = Pinecone(api_key="your-api-key")
 
# 创建 Serverless 索引
pc.create_index(
    name="rag-knowledge-base",
    dimension=1536,        # OpenAI text-embedding-3-small
    metric="cosine",
    spec={"serverless": {"cloud": "aws", "region": "us-east-1"}}
)

index = pc.Index("rag-knowledge-base")
 
# 写入向量（含元数据）
index.upsert(vectors=[
    {"id": "doc-001", "values": embedding, "metadata": {"source": "wiki", "topic": "RAG"}}
])
 
# 语义查询 + 元数据过滤
results = index.query(
    vector=query_embedding,
    top_k=10,
    include_metadata=True,
    filter={"topic": {"$eq": "RAG"}}
)

import weaviate
 
client = weaviate.Client(
    url="http://localhost:8080",
    additional_headers={"X-OpenAI-Api-Key": "your-key"}
)
 
# 混合搜索：alpha 控制向量 vs 关键词权重
result = (
    client.query
    .get("Document", ["title", "content", "source"])
    .with_hybrid(
        query="RAG 系统的检索策略",
        alpha=0.5,  # 0=纯关键词, 1=纯向量, 0.5=均衡
        fusion_type="relativeScoreFusion"
    )
    .with_where({
        "path": ["source"],
        "operator": "Equal",
        "valueString": "technical-docs"
    })
    .with_limit(10)
    .do()
)

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue, Range
)
 
client = QdrantClient("localhost", port=6333)
 
# 创建集合
client.create_collection(
    collection_name="products",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE)
)
 
# 复杂过滤搜索：语义相似 + 价格范围 + 类别匹配
results = client.search(
    collection_name="products",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(key="category", match=MatchValue(value="electronics")),
            FieldCondition(key="price", range=Range(gte=100, lte=500)),
            FieldCondition(key="in_stock", match=MatchValue(value=True))
        ]
    ),
    limit=10,
    with_payload=True
)

import chromadb
 
# 嵌入式模式 — 无需启动服务器
client = chromadb.PersistentClient(path="./chroma_db")
 
# 创建集合
collection = client.create_collection(
    name="knowledge_base",
    metadata={"hnsw:space": "cosine"}
)
 
# 直接添加文档（自动嵌入）
collection.add(
    documents=["RAG 系统通过检索增强生成质量", "向量数据库存储高维嵌入"],
    metadatas=[{"source": "tutorial"}, {"source": "docs"}],
    ids=["doc1", "doc2"]
)
 
# 语义查询
results = collection.query(
    query_texts=["如何提升 RAG 检索质量"],
    n_results=5,
    where={"source": "tutorial"}
)

-- 启用扩展
CREATE EXTENSION IF NOT EXISTS vector;
 
-- 创建带向量列的表
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    title TEXT NOT NULL,
    content TEXT,
    source VARCHAR(100),
    embedding vector(1536)  -- OpenAI embedding 维度
);
 
-- 创建 HNSW 索引（推荐生产使用）
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);

-- 语义搜索 + 传统 SQL 过滤，一条查询搞定
SELECT id, title, content,
       1 - (embedding <=> '[0.1, 0.2, ...]'::vector) AS similarity
FROM documents
WHERE source = 'technical-docs'
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
 
connections.connect("default", host="localhost", port="19530")
 
# 定义 Schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="text_embedding", dtype=DataType.FLOAT_VECTOR, dim=1536),
    FieldSchema(name="image_embedding", dtype=DataType.FLOAT_VECTOR, dim=512),
    FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=200)
]
schema = CollectionSchema(fields, description="多模态文档嵌入")
 
# 创建集合并配置 HNSW 索引
collection = Collection("multimodal_docs", schema)
collection.create_index(
    field_name="text_embedding",
    index_params={"metric_type": "IP", "index_type": "HNSW", "params": {"M": 16, "efConstruction": 256}}
)

你是一位向量数据库选型顾问。请根据以下项目需求推荐最合适的向量数据库：

项目需求：
- 向量规模：[预计向量数量，如 100 万 / 1 亿 / 10 亿]
- 查询模式：[纯语义搜索 / 混合搜索（语义+关键词） / 带复杂过滤的搜索]
- 延迟要求：[实时 <50ms / 准实时 <200ms / 批量处理]
- 团队能力：[有 DevOps 团队 / 无运维经验 / 有 PostgreSQL 经验]
- 预算范围：[月预算上限]
- 部署要求：[全托管 / 自托管 / 混合]
- 现有技术栈：[已有数据库、云平台、编程语言]

请从 Pinecone、Weaviate、Qdrant、ChromaDB、pgvector、Milvus 中推荐 1-2 个最佳选择，并说明理由。

原型/MVP 阶段
  └─ 首选 ChromaDB（嵌入式，零配置）
  └─ 备选 pgvector（如已有 PostgreSQL）

早期生产（<1000 万向量）
  ├─ 零运维需求 → Pinecone Serverless
  ├─ 需要混合搜索 → Weaviate Cloud
  ├─ 性能优先 + 成本敏感 → Qdrant Cloud
  └─ 已有 PostgreSQL → pgvector

规模化生产（1 亿+向量）
  ├─ 企业级 + 有 K8s 团队 → Milvus / Zilliz Cloud
  ├─ 高性能过滤搜索 → Qdrant（自托管集群）
  └─ 全托管 + 预算充足 → Pinecone Enterprise

# 原型阶段：ChromaDB 嵌入式，零配置
import chromadb
client = chromadb.PersistentClient(path="./prototype_db")
collection = client.create_collection("company_docs")

# 生产阶段：Qdrant Cloud，带过滤的多租户搜索
from qdrant_client import QdrantClient
client = QdrantClient(url="https://xxx.qdrant.io", api_key="...")
 
# 多租户过滤：每个企业客户只搜索自己的文档
results = client.search(
    collection_name="enterprise_docs",
    query_vector=query_embedding,
    query_filter=Filter(must=[
        FieldCondition(key="tenant_id", match=MatchValue(value="company-abc"))
    ]),
    limit=10
)

from abc import ABC, abstractmethod
from typing import List, Dict, Optional
 
class VectorStore(ABC):
    """向量数据库抽象层，便于后续迁移"""
 
    @abstractmethod
    def upsert(self, vectors: List[Dict]) -> None:
        pass
 
    @abstractmethod
    def search(self, vector: List[float], top_k: int,
               filters: Optional[Dict] = None) -> List[Dict]:
        pass
 
# 开发环境用 ChromaDB
class ChromaStore(VectorStore):
    def __init__(self, collection_name: str):
        self.client = chromadb.PersistentClient(path="./dev_db")
        self.collection = self.client.get_or_create_collection(collection_name)
 
    def upsert(self, vectors):
        # ChromaDB 实现
        ...
 
    def search(self, vector, top_k, filters=None):
        # ChromaDB 实现
        ...
 
# 生产环境用 Qdrant
class QdrantStore(VectorStore):
    def __init__(self, collection_name: str, url: str, api_key: str):
        self.client = QdrantClient(url=url, api_key=api_key)
        self.collection_name = collection_name
 
    def upsert(self, vectors):
        # Qdrant 实现
        ...
 
    def search(self, vector, top_k, filters=None):
        # Qdrant 实现
        ...