向量数据库选型实录：Milvus与Qdrant在图像检索场景下的性能对决

# docker-compose-milvus.yml
version: '3.5'
services:
  etcd:
    image: quay.io/coreos/etcd:v3.5.5
    environment:
      - ETCD_AUTO_COMPACTION_MODE=revision
      - ETCD_AUTO_COMPACTION_RETENTION=1000
      - ETCD_QUOTA_BACKEND_BYTES=4294967296
      - ETCD_SNAPSHOT_COUNT=50000
    volumes:
      - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcd
    command: etcd -advertise-client-urls=http://etcd:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd

  minio:
    image: minio/minio:RELEASE.2023-03-20T20-16-18Z
    environment:
      MINIO_ACCESS_KEY: minioadmin
      MINIO_SECRET_KEY: minioadmin
    volumes:
      - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/minio:/minio_data
    command: minio server /minio_data
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]

  standalone:
    image: milvusdb/milvus:v2.3.3
    command: ["milvus", "run", "standalone"]
    ports:
      - "19530:19530"
    volumes:
      - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvus
    depends_on:
      - etcd
      - minio

# docker-compose-qdrant.yml
version: '3.5'
services:
  qdrant:
    image: qdrant/qdrant:v1.7.3
    ports:
      - "6333:6333"
      - "6334:6334"  # gRPC端口
    volumes:
      - ./qdrant_storage:/qdrant/storage
    command: ["qdrant", "--config-path", "/qdrant/config/production.yaml"]

# milvus_test.py
from pymilvus import (Collection, CollectionSchema, DataType, FieldSchema,
                      connections, utility)
import numpy as np
import time

# 连接Milvus
connections.connect(host='localhost', port='19530')

# 定义schema，注意标量字段要加index才能做过滤
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=128),
    FieldSchema(name="color", dtype=DataType.VARCHAR, max_length=32),
    FieldSchema(name="price", dtype=DataType.FLOAT),
]
schema = CollectionSchema(fields=fields, description="product search")
collection = Collection(name="products", schema=schema)

# 创建索引：IVF_SQ8在精度和内存间平衡，nlist=1024
collection.create_index(field_name="embedding",
                        index_params={"index_type": "IVF_SQ8", "metric_type": "L2", "params": {"nlist": 1024}})

# 写入500万条数据
BATCH_SIZE = 10000
for i in range(0, 5000000, BATCH_SIZE):
    vectors = np.random.randn(BATCH_SIZE, 128).astype(np.float32)
    colors = np.random.choice(['red', 'blue', 'green', 'black', 'white'], BATCH_SIZE)
    prices = np.random.uniform(10, 1000, BATCH_SIZE).astype(np.float32)
    data = [vectors, colors, prices]
    collection.insert(data)
    if i % 100000 == 0:
        print(f"inserted {i} rows")

collection.flush()
print(f"total rows: {collection.num_entities}")

# 带标量过滤的混合查询
collection.load()
query_vector = np.random.randn(128).astype(np.float32)
start = time.time()
results = collection.search(data=[query_vector], anns_field="embedding",
                            param={"metric_type": "L2", "params": {"nprobe": 32}},
                            limit=10,
                            expr='color == "red" and price < 500',
                            output_fields=["color", "price"])
print(f"query latency: {(time.time()-start)*1000:.1f} ms")
print(f"result count: {len(results[0])}")

# qdrant_test.py
from qdrant_client import QdrantClient
from qdrant_client.models import (Distance, VectorParams, PointStruct,
                                  Filter, FieldCondition, MatchValue, Range)
import numpy as np
import time

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

# 创建collection，指定向量维度和距离函数
client.create_collection(
    collection_name="products",
    vectors_config=VectorParams(size=128, distance=Distance.L2),
)

# 创建标量索引，关键步骤！
client.create_payload_index(collection_name="products", field_name="color", field_schema="keyword")
client.create_payload_index(collection_name="products", field_name="price", field_schema="float")

# 批量写入，注意Qdrant的upsert接口
BATCH_SIZE = 10000
for i in range(0, 5000000, BATCH_SIZE):
    vectors = np.random.randn(BATCH_SIZE, 128).astype(np.float32)
    colors = np.random.choice(['red', 'blue', 'green', 'black', 'white'], BATCH_SIZE)
    prices = np.random.uniform(10, 1000, BATCH_SIZE).astype(np.float32)
    points = []
    for j in range(BATCH_SIZE):
        points.append(PointStruct(
            id=i+j,
            vector=vectors[j].tolist(),
            payload={"color": colors[j], "price": prices[j]}
        ))
    client.upsert(collection_name="products", points=points)
    if i % 100000 == 0:
        print(f"upserted {i} rows")

# 带标量过滤的查询
query_vector = np.random.randn(128).astype(np.float32)
query_filter = Filter(
    must=[
        FieldCondition(key="color", match=MatchValue(value="red")),
        FieldCondition(key="price", range=Range(lt=500))
    ]
)
start = time.time()
results = client.search(
    collection_name="products",
    query_vector=query_vector,
    query_filter=query_filter,
    limit=10,
    with_payload=True
)
print(f"query latency: {(time.time()-start)*1000:.1f} ms")
print(f"result count: {len(results)}")