Embedding 与向量数据库：模型选型、PGVector/Milvus/Chroma 对比

from openai import OpenAI

client = OpenAI()

response = client.embeddings.create(
    model="text-embedding-3-large",
    input="向量数据库选型需要综合考虑性能、成本和运维复杂度",
    dimensions=1024
)

vector = response.data[0].embedding
print(f"维度: {len(vector)}")  # 1024

import cohere

co = cohere.ClientV2(api_key="your-api-key")

response = co.embed(
    texts=["RAG 系统架构设计", "检索增强生成技术原理"],
    model="embed-v3",
    input_type="search_document",
    embedding_types=["float"],
    dimensions=1024
)

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-m3")

docs = [
    "Milvus 是一款开源的分布式向量数据库",
    "PGVector 是 PostgreSQL 的向量检索扩展",
]

dense_embeddings = model.encode(docs, normalize_embeddings=True)

output = model.encode(
    docs,
    pooling="cls",
    sparse_format="coo"
)
sparse_embeddings = output["lexical_weights"]

from FlagEmbedding import BGEM3FlagModel

model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)

docs = ["如何配置 HNSW 索引", "PGVector 安装与部署指南"]
query = "向量数据库索引配置方法"

doc_embeddings = model.encode(docs, batch_size=32)
query_embedding = model.encode([query])

import requests

def jina_embed(texts, task="retrieval"):
    url = "https://api.jina.ai/v1/embeddings"
    headers = {"Authorization": "Bearer your-jina-api-key"}
    data = {
        "model": "jina-embeddings-v3",
        "input": texts,
        "task": task,
        "dimensions": 1024
    }
    response = requests.post(url, json=data, headers=headers)
    return response.json()

import faiss

dimension = 1024
index = faiss.IndexHNSWFlat(dimension, 32)
index.hnsw.efConstruction = 200

import numpy as np
vectors = np.random.random((1_000_000, dimension)).astype("float32")
index.add(vectors)

index.hnsw.efSearch = 128
query = np.random.random((1, dimension)).astype("float32")
distances, indices = index.search(query, k=10)

100 万向量 → nlist = 1024 → nprobe = 32
搜索范围从 100 万缩小到约 3 万（3%），精度损失约 1-2%

1024 维向量 → 分为 32 个子空间 → 每个子空间 32 维
每个子空间训练 256 个聚类中心（8 bit 量化）
原始：1024 × 4 bytes = 4096 bytes/向量
压缩后：32 × 1 byte = 32 bytes/向量（压缩 128 倍）

import scann
import numpy as np

dataset = np.random.random((1_000_000, 128)).astype("float32")

searcher = (
    scann.builder_knn_btree(dataset, 10)
    .tree(
        num_leaves=1000,
        num_leaves_to_search=50,
        training_sample_size=100000
    )
    .score_ah(
        quantization_bits=2,
        rescoreidual_quantization_bits=2
    )
    .reorder(100)
    .build()
)

query = np.random.random((1, 128)).astype("float32")
neighbors, distances = searcher.search(query, k=10)

CREATE EXTENSION vector;

CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    embedding vector(1536) NOT NULL,
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 200);

CREATE INDEX ON documents USING ivfflat (embedding vector_l2_ops)
    WITH (lists = 100);

CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 200);

SET hnsw.ef_search = 128;

SELECT id, content, metadata,
       1 - (embedding <=> '[0.1, 0.2, ...]'::vector) AS similarity
FROM documents
WHERE metadata->>'category' = '技术文档'
  AND metadata->>'year' >= 2024
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;

-- 混合检索：向量 + 全文搜索的 SQL 实现
SELECT id, content,
       1 - (embedding <=> $1::vector) AS vector_score,
       ts_rank_cd(to_tsvector('chinese', content), plainto_tsquery('chinese', $2)) AS text_score,
       (1 - (embedding <=> $1::vector)) * 0.6 +
       ts_rank_cd(to_tsvector('chinese', content), plainto_tsquery('chinese', $2)) * 0.4 AS combined_score
FROM documents
ORDER BY combined_score DESC
LIMIT 10;

┌──────────────────────────────────────────────────────────┐
│                      Milvus Proxy                        │
│                    (负载均衡 / 路由)                        │
└────────────────────────┬─────────────────────────────────┘
                         │
         ┌───────────────┼───────────────┐
         ▼               ▼               ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Query Node   │ │ Data Node    │ │ Index Node   │
│ (查询服务)    │ │ (写入服务)    │ │ (索引构建)    │
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘
       │                │                │
       └────────────────┼────────────────┘
                        ▼
              ┌──────────────────┐
              │  Meta Store      │
              │  (etcd)          │
              └──────────────────┘
                        │
         ┌──────────────┼──────────────┐
         ▼              ▼              ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│  Pulsar/Kafka│ │  MinIO/S3    │ │  etcd        │
│  (日志流)     │  │  (对象存储)   │  │  (元数据)    │
└──────────────┘ └──────────────┘ └──────────────┘

from pymilvus import MilvusClient, DataType

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

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=1024)
schema.add_field("content", DataType.VARCHAR, max_length=2000)
schema.add_field("source", DataType.VARCHAR, max_length=200)
schema.add_field("category", DataType.VARCHAR, max_length=50)

index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 16, "efConstruction": 256}
)
index_params.add_index(
    field_name="source",
    index_type="TRIE"
)

client.create_collection(
    collection_name="knowledge_base",
    schema=schema,
    index_params=index_params
)

client.insert(
    collection_name="knowledge_base",
    data=[
        {"embedding": [0.1, 0.2, ...], "content": "...", "source": "产品手册", "category": "技术"},
    ]
)

# Milvus Lite：零配置启动
from pymilvus import MilvusClient

client = MilvusClient("milvus_demo.db")
client.create_collection(
    collection_name="test",
    dimension=128
)

client.insert("test", [{"id": i, "vector": [0.1]*128} for i in range(100)])
results = client.search("test", [[0.1]*128], limit=5)

import chromadb

client = chromadb.Client()

collection = client.create_collection(
    name="knowledge_base",
    metadata={"hnsw:space": "cosine"}
)

collection.add(
    documents=[
        "PGVector 是 PostgreSQL 的向量检索扩展",
        "Milvus 是一款开源分布式向量数据库",
        "Chroma 专注于简洁易用的向量存储",
    ],
    ids=["doc1", "doc2", "doc3"],
    metadatas=[
        {"source": "技术文档", "year": 2024},
        {"source": "技术文档", "year": 2024},
        {"source": "技术博客", "year": 2023},
    ]
)

results = collection.query(
    query_texts=["向量数据库选型对比"],
    n_results=2,
    where={"year": {"$gte": 2024}}
)

# 启动 Chroma 服务
chroma run --path /data/chroma

import chromadb

client = chromadb.HttpClient(host="localhost", port=8000)
collection = client.get_or_create_collection("knowledge_base")

# 默认使用 Sentence Transformers 的 all-MiniLM-L6-v2
collection = client.create_collection(
    name="auto_embed",
    metadata={"hnsw:space": "cosine"}
)
# 直接传入文本，Chroma 自动向量化
collection.add(documents=["Hello World", "Chroma is simple"])
results = collection.query(query_texts=["greeting"], n_results=1)

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="your-api-key")
pc.create_index(
    name="knowledge-base",
    dimension=1024,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)

index = pc.Index("knowledge-base")
index.upsert(vectors=[
    {"id": "doc1", "values": [0.1]*1024, "metadata": {"source": "产品手册"}}
])

results = index.query(vector=[0.1]*1024, top_k=5, include_metadata=True)

import weaviate

client = weaviate.connect_to_local()

client.collections.create(
    name="KnowledgeBase",
    vectorizer_config=weaviate.classes.config.Configure.Vectorizer.text2vec_openai(),
    properties=[
        weaviate.classes.config.Property(name="content", data_type=weaviate.classes.config.DataType.TEXT),
        weaviate.classes.config.Property(name="source", data_type=weaviate.classes.config.DataType.TEXT),
    ]
)

collection = client.collections.get("KnowledgeBase")
collection.data.insert({"content": "Weaviate 内置向量化支持", "source": "官方文档"})

results = collection.query.near_text(
    query="向量化数据库",
    limit=5,
    target_vector="default"
)

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

client = QdrantClient(url="http://localhost:6333")

client.create_collection(
    collection_name="knowledge_base",
    vectors_config=VectorParams(size=1024, distance=Distance.COSINE),
)

client.upsert(
    collection_name="knowledge_base",
    points=[
        PointStruct(id=1, vector=[0.1]*1024, payload={"source": "文档", "category": "技术"}),
    ]
)

results = client.query_points(
    collection_name="knowledge_base",
    query=[0.1]*1024,
    query_filter=Filter(must=[
        FieldCondition(key="category", match=MatchValue(value="技术"))
    ]),
    limit=5
)

Phase 1: Chroma（本地原型）
    ↓ 验证方案可行性
Phase 2: PGVector（轻量生产）
    ↓ 数据量 < 500 万，复用 PG
Phase 3: Milvus/Qdrant（规模增长）
    ↓ 数据量 > 1000 万，需要扩展
Phase 4: Zilliz Cloud/Pinecone（全面托管）
    ↓ 团队转向业务，减少运维

ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON documents
    USING (tenant_id = current_setting('app.current_tenant')::int);

CREATE POLICY read_access ON documents
    FOR SELECT
    USING (
        access_level IN ('public', 'internal')
        OR current_user IN (
            SELECT user_name FROM document_permissions
            WHERE document_id = documents.id AND permission = 'read'
        )
    );

SET app.current_tenant = '42';
SELECT * FROM documents;
-- 自动只返回 tenant_id=42 的数据

from pymilvus import MilvusClient

client = MilvusClient(uri="http://localhost:19530", token="root:Milvus")

client.create_role("reader")
client.grant_privilege(
    role_name="reader",
    object_type="Collection",
    object_name="knowledge_base",
    privilege="Search"
)

client.create_user("analyst", "secure-password")
client.assign_role("analyst", "reader")

CREATE EXTENSION pgaudit;

SET pgaudit.log = 'read, write, ddl';
SET pgaudit.log_catalog = on;
SET pgaudit.log_level = 'log';