Developer's Catalog

データベース

Milvus

概要

Milvusは、大規模なベクトルデータの管理と類似性検索に特化したオープンソースのベクトルデータベースです。AIアプリケーションや機械学習システムで生成される非構造化データ（画像、音声、テキストの埋め込みベクトル）を効率的に処理し、高速な類似性検索を実現します。クラウドネイティブアーキテクチャを採用し、数十億規模のベクトルデータでもミリ秒単位での検索を可能にします。

詳細

Milvusは2019年にZilliz社によって開発され、現在はLF AI & Data Foundationのインキュベーションプロジェクトとして管理されています。分散アーキテクチャによりコンピューティングとストレージを分離し、高いスケーラビリティと可用性を実現しています。推薦システム、画像・動画検索、自然言語処理、異常検知など、ベクトル類似性検索が必要な幅広いAIアプリケーションで採用されています。

Milvusの主な特徴：

• クラウドネイティブ分散アーキテクチャ
• 複数のベクトルインデックスタイプサポート（HNSW、IVF、DiskANN等）
• ハードウェアアクセラレーション（GPU、SIMD）
• ハイブリッド検索（ベクトル検索 + スカラーフィルタリング）
• マルチテナント対応
• ホット/コールドストレージによるコスト最適化
• ACID特性サポート
• 豊富なSDK（Python、Go、Java、Node.js）
• RESTful/gRPC API
• Kubernetes対応

アーキテクチャ

Milvusは4層アーキテクチャを採用：

1. アクセス層: ステートレスなプロキシによる負荷分散
2. コーディネーター層: メタデータ管理とタスクスケジューリング
3. ワーカーノード層: データ処理とクエリ実行
4. ストレージ層: オブジェクトストレージとメッセージキュー

メリット・デメリット

メリット

• 高性能: ベクトル専用の最適化により超高速検索を実現
• スケーラビリティ: 水平スケーリングで数百億ベクトルに対応
• 高可用性: レプリケーションと自動フェイルオーバー
• 柔軟性: 10種類以上のインデックスタイプから選択可能
• コスト効率: ホット/コールドストレージで最適化
• エコシステム: LangChain、LlamaIndex等との統合
• マルチモーダル: 画像、テキスト、音声の統一検索

デメリット

• 学習コスト: ベクトルデータベースの概念理解が必要
• リソース消費: 大規模データでメモリ使用量が増大
• 複雑性: 分散システムの運用管理が必要
• 限定的なトランザクション: 完全なACIDサポートではない
• ベクトル専用: 従来のSQLクエリには不向き

主要リンク

• 公式サイト
• ドキュメント
• GitHub
• Zilliz Cloud

書き方の例

インストール・セットアップ

# Docker での実行（Milvus Standalone）
wget https://github.com/milvus-io/milvus/releases/download/v2.6.0/milvus-standalone-docker-compose.yml -O docker-compose.yml
docker compose up -d

# Kubernetes での実行（Milvus Distributed）
helm repo add milvus https://zilliztech.github.io/milvus-helm/
helm repo update
helm install my-milvus milvus/milvus

# Python SDK インストール
pip install pymilvus

# Milvus Lite（開発用）
pip install pymilvus
# Python内で直接使用可能

# Node.js SDK インストール
npm install @zilliz/milvus2-sdk-node

# Go SDK インストール
go get -u github.com/milvus-io/milvus-sdk-go/v2

基本操作（Python SDK）

from pymilvus import MilvusClient, DataType
import numpy as np

# 接続
client = MilvusClient(
    uri="http://localhost:19530",  # Milvus Standalone
    # uri="./milvus_demo.db"  # Milvus Lite
)

# コレクション作成
schema = MilvusClient.create_schema(
    auto_id=False,
    enable_dynamic_field=True,
)

# フィールド定義
schema.add_field(
    field_name="id",
    datatype=DataType.INT64,
    is_primary=True,
)
schema.add_field(
    field_name="vector",
    datatype=DataType.FLOAT_VECTOR,
    dim=768,  # 埋め込みベクトルの次元
)
schema.add_field(
    field_name="text",
    datatype=DataType.VARCHAR,
    max_length=65535,
)

# インデックスパラメータ
index_params = client.prepare_index_params()
index_params.add_index(
    field_name="vector",
    index_type="HNSW",
    metric_type="L2",
    params={"M": 16, "efConstruction": 256}
)

# コレクション作成
client.create_collection(
    collection_name="documents",
    schema=schema,
    index_params=index_params
)

# データ挿入
docs = [
    {"id": 1, "vector": np.random.rand(768).tolist(), "text": "Milvusは高性能ベクトルデータベース"},
    {"id": 2, "vector": np.random.rand(768).tolist(), "text": "AIアプリケーションに最適"},
    {"id": 3, "vector": np.random.rand(768).tolist(), "text": "大規模な類似性検索を実現"},
]

client.insert(collection_name="documents", data=docs)

# ベクトル検索
query_vector = np.random.rand(768).tolist()

results = client.search(
    collection_name="documents",
    data=[query_vector],
    limit=3,
    output_fields=["id", "text"]
)

for hits in results:
    for hit in hits:
        print(f"ID: {hit['id']}, Distance: {hit['distance']}, Text: {hit['text']}")

高度な検索機能

# ハイブリッド検索（ベクトル + スカラーフィルタ）
results = client.search(
    collection_name="documents",
    data=[query_vector],
    filter="id > 1",  # スカラーフィルタリング
    limit=5,
    output_fields=["id", "text", "category"]
)

# 範囲検索
results = client.search(
    collection_name="documents",
    data=[query_vector],
    limit=10,
    search_params={"metric_type": "L2", "params": {"radius": 1.0}}
)

# 複数ベクトル検索（バッチ処理）
query_vectors = [np.random.rand(768).tolist() for _ in range(5)]
results = client.search(
    collection_name="documents",
    data=query_vectors,
    limit=3
)

# グループ化検索
results = client.search(
    collection_name="products",
    data=[query_vector],
    limit=10,
    group_by_field="category",
    output_fields=["id", "name", "category", "price"]
)

埋め込み関数との統合

from pymilvus import model

# OpenAI埋め込み
embedding_fn = model.dense.OpenAIEmbeddingFunction(
    model_name="text-embedding-3-small",
    api_key="your-api-key"
)

# Sentence Transformers
embedding_fn = model.dense.SentenceTransformerEmbeddingFunction(
    model_name="all-MiniLM-L6-v2"
)

# テキストから埋め込みベクトル生成
texts = ["Milvusの使い方", "ベクトル検索の実装"]
embeddings = embedding_fn.encode_documents(texts)

# 埋め込みを使用してデータ挿入
docs_with_embeddings = [
    {"id": i, "vector": embedding, "text": text}
    for i, (text, embedding) in enumerate(zip(texts, embeddings))
]

client.insert(collection_name="documents", data=docs_with_embeddings)

インデックス管理

# 利用可能なインデックスタイプ
# FLAT: 総当たり検索（精度100%）
# IVF_FLAT: 逆ファイルインデックス
# IVF_SQ8: 量子化によるメモリ削減
# HNSW: 階層的ナビゲート可能小世界グラフ
# DISKANN: ディスクベースのANNインデックス

# インデックス作成（IVF_FLAT）
index_params = {
    "metric_type": "L2",
    "index_type": "IVF_FLAT",
    "params": {"nlist": 1024}
}

client.create_index(
    collection_name="documents",
    field_name="vector",
    index_params=index_params
)

# インデックス情報取得
index_info = client.describe_index(
    collection_name="documents",
    field_name="vector"
)

# インデックス削除と再構築
client.drop_index(
    collection_name="documents",
    field_name="vector"
)

パーティション管理

# パーティション作成
client.create_partition(
    collection_name="documents",
    partition_name="2024_data"
)

# パーティション指定でデータ挿入
client.insert(
    collection_name="documents",
    data=docs,
    partition_name="2024_data"
)

# パーティション指定で検索
results = client.search(
    collection_name="documents",
    data=[query_vector],
    limit=5,
    partition_names=["2024_data"]
)

# パーティション一覧取得
partitions = client.list_partitions(collection_name="documents")

実用例：画像検索システム

from PIL import Image
import torch
from transformers import CLIPModel, CLIPProcessor

# CLIP モデルで画像埋め込み
model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

# コレクション作成
schema = MilvusClient.create_schema()
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("image_vector", DataType.FLOAT_VECTOR, dim=512)
schema.add_field("image_path", DataType.VARCHAR, max_length=500)
schema.add_field("metadata", DataType.JSON)

client.create_collection("image_search", schema=schema)

# 画像埋め込み生成
def get_image_embedding(image_path):
    image = Image.open(image_path)
    inputs = processor(images=image, return_tensors="pt")
    with torch.no_grad():
        image_features = model.get_image_features(**inputs)
    return image_features.numpy().flatten().tolist()

# 画像データ挿入
image_data = []
for i, img_path in enumerate(image_paths):
    embedding = get_image_embedding(img_path)
    image_data.append({
        "id": i,
        "image_vector": embedding,
        "image_path": img_path,
        "metadata": {"category": "product", "date": "2024-01-15"}
    })

client.insert("image_search", data=image_data)

# テキストクエリで画像検索
def search_images_by_text(text_query, limit=10):
    inputs = processor(text=[text_query], return_tensors="pt")
    with torch.no_grad():
        text_features = model.get_text_features(**inputs)
    query_vector = text_features.numpy().flatten().tolist()
    
    results = client.search(
        collection_name="image_search",
        data=[query_vector],
        limit=limit,
        output_fields=["image_path", "metadata"]
    )
    return results

# 検索実行
results = search_images_by_text("赤い車")

RAGアプリケーション統合

from langchain_milvus import Milvus
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter

# ドキュメント分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
documents = text_splitter.split_documents(raw_documents)

# Milvusベクトルストア
embeddings = OpenAIEmbeddings()
vectorstore = Milvus(
    embedding_function=embeddings,
    collection_name="rag_documents",
    connection_args={"uri": "http://localhost:19530"}
)

# ドキュメント追加
vectorstore.add_documents(documents)

# 類似性検索
query = "Milvusの主な機能は？"
results = vectorstore.similarity_search(query, k=5)

# RAGチェーン構築
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    return_source_documents=True
)

response = qa_chain({"query": query})
print(response["result"])

管理・監視

# コレクション統計情報
stats = client.get_collection_stats(collection_name="documents")
print(f"行数: {stats['row_count']}")

# メモリ使用量確認
client.get_loading_progress(collection_name="documents")

# コレクションのフラッシュ（永続化）
client.flush(collection_names=["documents"])

# バックアップ（データエクスポート）
client.query(
    collection_name="documents",
    expr="",  # 全データ
    output_fields=["id", "vector", "text"],
    limit=1000000
)

# コレクション削除
client.drop_collection(collection_name="documents")

# 接続を閉じる
client.close()

パフォーマンスチューニング

# 検索パラメータ最適化
search_params = {
    "metric_type": "L2",
    "params": {
        "nprobe": 16,  # IVFインデックス用
        "ef": 64,      # HNSWインデックス用
    }
}

# コレクションロード設定
client.load_collection(
    collection_name="documents",
    replica_number=2  # レプリカ数増加で高可用性
)

# メモリマップ有効化（大規模データ用）
schema.add_field(
    field_name="large_text",
    datatype=DataType.VARCHAR,
    max_length=65535,
    mmap_enabled=True  # メモリマップ有効化
)

# バッチ挿入最適化
batch_size = 1000
for i in range(0, len(all_data), batch_size):
    batch = all_data[i:i+batch_size]
    client.insert(collection_name="documents", data=batch)