Developer's Catalog

データベース

MongoDB Atlas Vector Search

概要

MongoDB Atlas Vector Searchは、MongoDBの人気のあるドキュメント型NoSQLデータベースに統合されたベクトル検索機能です。MongoDB Atlas（クラウドマネージドサービス）の一部として提供され、既存のMongoDBインフラストラクチャでベクトル検索を実現できます。ドキュメント型データベースの柔軟性とベクトル検索を組み合わせることで、RAGアプリケーションや推薦システムの構築が容易になります。

詳細

MongoDBは2009年に10gen社（現MongoDB Inc.）によって開発され、現在では最も人気のあるNoSQLデータベースの一つとなっています。2023年にMongoDB Atlas Vector Searchが導入され、ベクトル埋め込みの保存と検索が可能になりました。Apache Luceneベースの検索エンジンを活用し、高度な類似性検索機能を提供しています。

MongoDB Atlas Vector Searchの主な特徴：

• ドキュメント型データベースとベクトル検索の統合
• Approximate Nearest Neighbor (ANN)アルゴリズム
• 高次元ベクトル（最大4096次元）のサポート
• ユークリッド距離、コサイン類似度、ドット積による類似性計算
• プリフィルタリングとポストフィルタリング
• 自動インデックス管理
• Atlas Searchとの統合による高度な検索機能
• グローバル分散クラスタでのベクトル検索
• マルチテナント対応
• エンタープライズグレードのセキュリティ

アーキテクチャの特徴

• Apache Luceneベースの検索エンジン
• 分散型レプリカセットアーキテクチャ
• 自動シャーディングとロードバランシング
• リアルタイムデータ同期

メリット・デメリット

メリット

• 統合ソリューション: 構造化データ、非構造化データ、ベクトルデータを一つのプラットフォームで管理
• 開発の容易さ: 既存のMongoDB APIとツールチェーンを活用
• 柔軟なデータモデル: ドキュメント型の柔軟性とベクトル検索の組み合わせ
• マネージドサービス: インフラストラクチャ管理が不要
• グローバルスケール: 世界中のリージョンでの展開が容易
• エンタープライズ機能: 包括的なセキュリティ、監査、コンプライアンス機能
• 豊富なエコシステム: 多くのフレームワークやツールとの統合

デメリット

• コスト: マネージドサービスのため運用コストが高い
• ベンダーロックイン: Atlas限定機能のため移行が困難
• パフォーマンス: 専用ベクトルデータベースと比較して劣る場合がある
• 機能制限: ベクトル検索専用データベースと比較して機能が限定的
• 学習曲線: MongoDBとベクトル検索両方の知識が必要

主要リンク

• 公式サイト
• ドキュメント
• GitHub
• MongoDB Atlas

書き方の例

セットアップとインデックス作成

// MongoDB接続（Node.js）
const { MongoClient } = require('mongodb');

const client = new MongoClient('mongodb+srv://username:[email protected]');

// ベクトル検索インデックスの定義
const vectorSearchIndex = {
  name: "vector_index",
  type: "vectorSearch",
  definition: {
    fields: [
      {
        type: "vector",
        path: "embedding",
        numDimensions: 768,
        similarity: "cosine"
      },
      {
        type: "filter",
        path: "category"
      },
      {
        type: "filter", 
        path: "metadata.year"
      }
    ]
  }
};

// インデックス作成（Atlas UIまたはAPIで実行）

ドキュメントの挿入

async function insertDocuments() {
  const db = client.db('vectordb');
  const collection = db.collection('documents');
  
  // ドキュメントの挿入
  const documents = [
    {
      title: "MongoDB Vector Search",
      content: "MongoDBでベクトル検索を実装する方法",
      embedding: Array(768).fill(0).map(() => Math.random()),
      category: "database",
      metadata: {
        year: 2024,
        author: "MongoDB Team"
      }
    }
  ];
  
  await collection.insertMany(documents);
}

ベクトル検索の実行

async function vectorSearch(queryVector) {
  const db = client.db('vectordb');
  const collection = db.collection('documents');
  
  // ベクトル検索パイプライン
  const pipeline = [
    {
      $vectorSearch: {
        index: "vector_index",
        path: "embedding",
        queryVector: queryVector,
        numCandidates: 100,
        limit: 10,
        filter: {
          category: "database"
        }
      }
    },
    {
      $project: {
        title: 1,
        content: 1,
        score: { $meta: "vectorSearchScore" }
      }
    }
  ];
  
  const results = await collection.aggregate(pipeline).toArray();
  return results;
}

Python実装例

from pymongo import MongoClient
import numpy as np
from datetime import datetime

# 接続
client = MongoClient("mongodb+srv://username:[email protected]")
db = client.vectordb
collection = db.documents

# ドキュメントの挿入
documents = [
    {
        "title": "MongoDB Atlas Vector Search",
        "content": "大規模なベクトル検索を実現",
        "embedding": np.random.rand(768).tolist(),
        "category": "database",
        "metadata": {
            "year": 2024,
            "tags": ["nosql", "vector", "search"]
        },
        "created_at": datetime.now()
    }
]

collection.insert_many(documents)

# ベクトル検索
def vector_search(query_vector, filters=None):
    pipeline = [
        {
            "$vectorSearch": {
                "index": "vector_index",
                "path": "embedding",
                "queryVector": query_vector,
                "numCandidates": 200,
                "limit": 10
            }
        }
    ]
    
    # フィルタの追加
    if filters:
        pipeline[0]["$vectorSearch"]["filter"] = filters
    
    # プロジェクション
    pipeline.append({
        "$project": {
            "title": 1,
            "content": 1,
            "category": 1,
            "score": {"$meta": "vectorSearchScore"}
        }
    })
    
    results = list(collection.aggregate(pipeline))
    return results

# 検索実行
query_embedding = np.random.rand(768).tolist()
results = vector_search(
    query_embedding,
    filters={"category": "database", "metadata.year": {"$gte": 2023}}
)

ハイブリッド検索

# テキスト検索とベクトル検索の組み合わせ
def hybrid_search(text_query, query_vector):
    pipeline = [
        {
            "$vectorSearch": {
                "index": "vector_index",
                "path": "embedding",
                "queryVector": query_vector,
                "numCandidates": 100,
                "limit": 20
            }
        },
        {
            "$match": {
                "$text": {"$search": text_query}
            }
        },
        {
            "$project": {
                "title": 1,
                "content": 1,
                "vectorScore": {"$meta": "vectorSearchScore"},
                "textScore": {"$meta": "textScore"},
                "combinedScore": {
                    "$add": [
                        {"$multiply": [{"$meta": "vectorSearchScore"}, 0.7]},
                        {"$multiply": [{"$meta": "textScore"}, 0.3]}
                    ]
                }
            }
        },
        {
            "$sort": {"combinedScore": -1}
        },
        {
            "$limit": 10
        }
    ]
    
    return list(collection.aggregate(pipeline))

バッチ処理と最適化

# バッチ埋め込み処理
def batch_embed_and_insert(texts, embeddings):
    documents = []
    for i, (text, embedding) in enumerate(zip(texts, embeddings)):
        doc = {
            "content": text,
            "embedding": embedding.tolist(),
            "metadata": {
                "batch_id": datetime.now().isoformat(),
                "index": i
            }
        }
        documents.append(doc)
    
    # バルク挿入
    if documents:
        collection.insert_many(documents, ordered=False)

# インデックス統計の取得
def get_index_stats():
    stats = db.command("collStats", "documents", indexDetails=True)
    return stats