Разработка RAG с векторной базой данных Milvus

Разработка RAG с векторной базой данных Milvus

Milvus — открытая векторная база данных, разработанная для production-масштаба: миллиарды векторов, высокая пропускная способность, горизонтальное масштабирование. Поддерживает HNSW, IVF_FLAT, IVF_SQ8, DISKANN индексы и гибридный поиск (sparse + dense). Хорошо подходит для enterprise-систем с большими объёмами данных.

Установка и подключение

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility

# Подключение к Milvus
connections.connect(
    alias="default",
    host="localhost",
    port="19530"
)

# Или через URI (Milvus Lite для локальной разработки)
from pymilvus import MilvusClient
client = MilvusClient("./milvus_local.db")  # SQLite-подобный файл

Создание коллекции со схемой

fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=4096),
    FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=512),
    FieldSchema(name="doc_type", dtype=DataType.VARCHAR, max_length=64),
    FieldSchema(name="page", dtype=DataType.INT32),
    FieldSchema(
        name="dense_vector",
        dtype=DataType.FLOAT_VECTOR,
        dim=1536  # text-embedding-3-small
    ),
    FieldSchema(
        name="sparse_vector",
        dtype=DataType.SPARSE_FLOAT_VECTOR  # BM25
    ),
]

schema = CollectionSchema(fields=fields, description="Corporate Knowledge Base")
collection = Collection(name="knowledge_base", schema=schema)

# Индексы для векторных полей
collection.create_index(
    field_name="dense_vector",
    index_params={"metric_type": "COSINE", "index_type": "HNSW", "params": {"M": 16, "efConstruction": 200}}
)
collection.create_index(
    field_name="sparse_vector",
    index_params={"metric_type": "IP", "index_type": "SPARSE_INVERTED_INDEX"}
)

collection.load()

Hybrid Search с RRF

from pymilvus import AnnSearchRequest, RRFRanker

def milvus_hybrid_search(query: str, top_k: int = 5) -> list:
    # Dense вектор
    dense_vec = dense_embedder.embed_query(query)

    # Sparse вектор (через встроенный BM25Encoder)
    sparse_vec = sparse_encoder.encode_queries([query])

    # Два запроса для RRF
    dense_req = AnnSearchRequest(
        data=[dense_vec],
        anns_field="dense_vector",
        param={"metric_type": "COSINE", "params": {"ef": 100}},
        limit=30,
    )

    sparse_req = AnnSearchRequest(
        data=sparse_vec,
        anns_field="sparse_vector",
        param={"metric_type": "IP"},
        limit=30,
    )

    # RRF fusion
    results = collection.hybrid_search(
        reqs=[dense_req, sparse_req],
        rerank=RRFRanker(k=60),
        limit=top_k,
        output_fields=["text", "source", "doc_type"],
    )

    return results

Практический кейс: поиск по продуктовой документации

Масштаб: 2.5M чанков (техническая документация на 6 языках, включая русский).

Индекс: HNSW (M=32, efConstruction=400) для dense; SPARSE_INVERTED_INDEX для sparse.

Throughput при hybrid search: 850 QPS при latency P99 < 400мс на кластере из 3 узлов (8 vCPU, 32GB RAM каждый).

Сравнение с Pinecone Serverless при том же объёме: Milvus self-hosted дешевле в ~8× при высоком QPS, но требует DevOps-поддержки.

Партиционирование для мультитенантности

# Создание партиций для изоляции клиентов
collection.create_partition(partition_name="client_001")
collection.create_partition(partition_name="client_002")

# Поиск только в партиции клиента
results = collection.search(
    data=[query_vector],
    anns_field="dense_vector",
    partition_names=["client_001"],
    limit=5,
)

Сроки

• Настройка Milvus кластера + схема: 3–5 дней
• Ingestion pipeline с hybrid indexing: 5–10 дней
• RAG-пайплайн и оценка: 1–2 недели
• Итого: 3–5 недель