AIアプリケーションは、多くの場合、コンピューティング、データ、コストの点で小規模に始めることができます。ユーザーエンゲージメントの増加に合わせて本番アプリケーションがスケールするにつれて、主な要素、例えば大量のデータの保存・取得に関連するコストなどの最適化が非常に重要になっていきます。こうした課題は、次の点に焦点を当てることで対応できます。
効率的なベクトル検索アルゴリズム
自動量子化プロセス
最適化された埋め込み戦略
検索拡張生成 （RAG）システムとエージェントベースのシステムはどちらも、意味的類似の検索を実行するためにベクターデータ（画像、動画、テキストなどのデータオブジェクトの数値表現）を利用します。RAGまたはエージェント駆動型ワークフローを使用するシステムは、応答時間を短くして検索レイテンシを最小限に抑え、インフラストラクチャのコストをコントロールできるよう、大規模かつ高次元のデータセットを効率的に取り扱う必要があります。
チュートリアルについて
このチュートリアルでは、高度なAIワークロードを大規模に設計、配置、管理し、最適なパフォーマンスとコスト効率を確保するために必要な手法を習得できます。
具体的には、このチュートリアルでは次の方法を学びます。
量子化対応の汎用多言語埋め込みモデルである Voyage AI の
voyage-3-largeを使用して埋め込みを生成し、MongoDB データベースに取り込みます。埋め込みを低精度のデータ型へと自動的に量子化し、メモリ使用量とクエリのレイテンシの両方を最適化します。
float32、int8、バイナリ埋め込みを比較するクエリを実行し、データ型の精度を効率および検索精度と比較する。
量子化された埋め込みのリコール（リテンションとも呼ばれる）を測定します。これは、量子化された近似最近傍探索（ANN）検索が、フル精度の厳密最近傍探索（ENN）検索と同じドキュメントをどれだけ効果的に取得できるかを評価するものです。
注意
バイナリ量子化は、精度の低下に対処するために再スコアリングパスが必要になる場合がありますが、リソース消費量の削減が必要なシナリオに最適です。
スカラー量子化は実用的な妥協点で、性能と精度のバランスを取る必要があるユースケースのほとんどに適しています。
Float32は最大の忠実度を保証しますが、パフォーマンスとメモリのオーバーヘッドが最も高いため、大規模なシステムやレイテンシの影響が大きいシステムには適していません。
前提条件
Atlas の サンプル データ セット からの映画データを含むコレクションを使用します。
High-CPU
S20以上の検索階層を使用する 2 以上の検索ノードを持つM20以上のクラスター有料階層Voyage AI APIキー。無料階層キーでは、このチュートリアルに必要なリクエストボリュームはサポートされていません。
また、次のことも推奨します。
- A Hugging Face access token. 必須ではありませんが、Hugging Face アクセストークンを使用すると、このチュートリアルで使用される Wikipedia データセットのダウンロード速度が向上します。
手順
必要なライブラリをインポートし、環境変数を設定してください。
.ipynb拡張機能を持つファイルを保存して、インタラクティブな Python ノートブックを作成します。ライブラリをインストールしてください。
このチュートリアルでは、次のライブラリをインポートする必要があります。
PyMongo
voyageai
Voyage AI Python クライアントは、データの埋め込みを生成するために使用します。
pandas
データセット
Hugging Faceライブラリ既成のデータセットへのアクセスを提供します。
matplotlib
ライブラリをインストールするには、次のコマンドを実行します。
pip install --quiet -U pymongo voyageai pandas datasets matplotlib環境変数を安全に取得し、設定します。
次の
set_env_securelyヘルパー関数は、環境変数を安全に取得し、設定します。次のコードをコピーして貼り付け、実行し、プロンプトが表示されたら、Voyage AI API キーやクラスター接続文字列などの秘密の値を設定してください。また、任意の Hugging Face アクセストークンの入力を求められます。トークンを提供すると、レート制限が緩和され、データセットのダウンロードが高速化されます。スキップするにはEnterを押します。import getpassimport osimport voyageaiimport huggingface_hub # Function to securely get and set environment variablesdef set_env_securely(var_name, prompt): value = getpass.getpass(prompt) os.environ[var_name] = value # Environment Variablesset_env_securely("VOYAGE_API_KEY", "Enter your Voyage API Key: ")set_env_securely("MONGO_URI", "Enter your MongoDB URI: ")MONGO_URI = os.environ.get("MONGO_URI")if not MONGO_URI: raise ValueError("MONGO_URI not set in environment variables.") # Voyage Clientvoyage_client = voyageai.Client()# Hugging Face Token (optional)# Providing a token increases rate limits and speeds up dataset downloads.# Press Enter to skip.hf_token = getpass.getpass("Enter your Hugging Face token (optional, press Enter to skip): ")if hf_token: os.environ["HF_TOKEN"] = hf_token huggingface_hub.login(hf_token) print("Hugging Face token set.")else: print("No Hugging Face token provided. You may experience slower download speeds.")
データをクラスターに取り込みます。
このステップでは、次のデータセットから最大 250000 件のドキュメントを読み込みます。
データをクラスターにロードするための関数を定義します。
次のコードをコピーしてノートブックに貼り付け、実行します。サンプルコードは、次の関数を定義します。
generate_bson_vectorデータセットの埋め込みをBSONバイナリベクトルに変換して、ベクトルの効率的なストレージとプロセシングを実現します。get_mongo_clientクラスター接続文字列を取得します。insert_dataframe_into_collectionクラスターにデータを取り込むため。
import pandas as pdfrom datasets import load_datasetfrom bson.binary import Binary, BinaryVectorDtypeimport pymongo # Connect to Clusterdef get_mongo_client(uri): """Connect to MongoDB and confirm the connection.""" client = pymongo.MongoClient(uri) if client.admin.command("ping").get("ok") == 1.0: print("Connected to MongoDB successfully.") return client print("Failed to connect to MongoDB.") return None # Generate BSON Vectordef generate_bson_vector(array, data_type): """Convert an array to BSON vector format.""" array = [float(val) for val in eval(array)] return Binary.from_vector(array, BinaryVectorDtype(data_type)) # Load Datasetsdef load_and_prepare_data(dataset_name, amount): """Load and prepare streaming datasets for DataFrame.""" print(f"Loading {amount} records from {dataset_name}...") data = load_dataset(dataset_name, streaming=True, split="train").take(amount) return pd.DataFrame(data) # Insert datasets into MongoDB Collectiondef insert_dataframe_into_collection(df, collection): """Insert Dataset records into MongoDB collection.""" collection.insert_many(df.to_dict("records")) print(f"Inserted {len(df)} records into '{collection.name}' collection.")データをクラスターにロードします。
ノート PC 内の次のコードをコピーして貼り付け、実行して、データセットをクラスターにロードします。このコードは、次のアクションを実行します。
データセットを取得します。
埋め込みをBSON形式に変換します。
クラスターにコレクションを作成し、データを挿入します。
import pandas as pdfrom bson.binary import Binary, BinaryVectorDtypefrom pymongo.errors import CollectionInvalid wikipedia_data_df = load_and_prepare_data("MongoDB/wikipedia-22-12-en-voyage-embed", amount=250000)wikipedia_annotation_data_df = load_and_prepare_data("MongoDB/wikipedia-22-12-en-annotation", amount=250000)wikipedia_annotation_data_df.drop(columns=["_id"], inplace=True) # Convert embeddings to BSON formatwikipedia_data_df["embedding"] = wikipedia_data_df["embedding"].apply( lambda x: generate_bson_vector(x, BinaryVectorDtype.FLOAT32)) # MongoDB Setupmongo_client = get_mongo_client(MONGO_URI)DB_NAME = "testing_datasets"db = mongo_client[DB_NAME] collections = { "wikipedia-22-12-en": wikipedia_data_df, "wikipedia-22-12-en-annotation": wikipedia_annotation_data_df,} # Create Collections and Insert Datafor collection_name, df in collections.items(): if collection_name not in db.list_collection_names(): try: db.create_collection(collection_name) print(f"Collection '{collection_name}' created successfully.") except CollectionInvalid: print(f"Error creating collection '{collection_name}'.") else: print(f"Collection '{collection_name}' already exists.") # Clear collection and insert fresh data collection = db[collection_name] collection.delete_many({}) insert_dataframe_into_collection(df, collection)重要: 埋め込みをBSONベクトルに変換し、データセットをクラスターに取り込むには、時間がかかる場合があります。
データセットが正常にロードされたことを確認するには、クラスターにログインし、Data Explorerでコレクションを視覚的に検査します。
コレクションにMongoDB ベクトル検索インデックスを作成します。
このステップでは、embeddingフィールドに次の3つのインデックスを作成します。
スカラー量子化インデックス | スカラー量子化法を用いて埋め込みを量子化します。 |
バイナリ量子化されたインデックス | バイナリ量子化法を使用して埋め込みを量子化します。 |
Float32 ANN Index | float32 近似最近傍探索 メソッドを使用して埋め込みを量子化します。 |
MongoDB ベクトル検索インデックスを作成するための関数を定義します。
以下をコピーしてノートブックに貼り付け、実行します。
import timefrom pymongo.operations import SearchIndexModel def setup_vector_search_index(collection, index_definition, index_name="vector_index"): new_vector_search_index_model = SearchIndexModel( definition=index_definition, name=index_name, type="vectorSearch" ) # Create the new index try: result = collection.create_search_index(model=new_vector_search_index_model) print(f"Creating index '{index_name}'...") # Wait for initial sync to complete print("Polling to check if the index is ready. This might take a few minutes.") predicate=None if predicate is None: predicate = lambda index: index.get("queryable") is True while True: indices = list(collection.list_search_indexes(result)) if len(indices) and predicate(indices[0]): break time.sleep(5) print(f"Index '{index_name}' is ready for querying.") return result except Exception as e: print(f"Error creating new vector search index '{index_name}': {e!s}") return Noneインデックスを定義します。
次のインデックス構成は、さまざまな量子化戦略を実装します。
vector_index_definition_scalar_quantizedこの構成では、スカラー量子化（int8）を使用します。
各ベクトル次元を32ビットの浮動小数点数から8ビットの整数に変換します。
精度とメモリ効率のバランスを良好に保ちます
メモリの最適化が必要なほとんどの本番環境のユースケースに適しています
vector_index_definition_binary_quantizedこの構成では、バイナリ量子化（int1）を使用します。
各ベクトル次元を1ビットまで縮小します。
最大限のメモリ効率を提供します
メモリ制約が重要となる非常に大規模な配置に最適です
これらのインデックスが作成されると、自動量化が透過的に行われ、 MongoDB ベクトル検索はインデックスの作成および検索操作中に、float32 から指定された量子化形式への変換を処理します。
vector_index_definition_float32_annインデックス構成は、cosine類似度関数を使用して、1024ディメンションの完全忠実度ベクトルをインデックス化します。# Scalar Quantizationvector_index_definition_scalar_quantized = { "fields": [ { "type": "vector", "path": "embedding", "quantization": "scalar", "numDimensions": 1024, "similarity": "cosine", } ]}# Binary Quantizationvector_index_definition_binary_quantized = { "fields": [ { "type": "vector", "path": "embedding", "quantization": "binary", "numDimensions": 1024, "similarity": "cosine", } ]}# Float32 Embeddingsvector_index_definition_float32_ann = { "fields": [ { "type": "vector", "path": "embedding", "numDimensions": 1024, "similarity": "cosine", } ]}setup_vector_search_index関数を使用して、スカラー、バイナリ、float32 インデックスを作成します。インデックスのコレクション名とインデックス名を設定します。
wiki_data_collection = db["wikipedia-22-12-en"]wiki_annotation_data_collection = db["wikipedia-22-12-en-annotation"]vector_search_scalar_quantized_index_name = "vector_index_scalar_quantized"vector_search_binary_quantized_index_name = "vector_index_binary_quantized"vector_search_float32_ann_index_name = "vector_index_float32_ann"MongoDB ベクトル検索インデックスを作成します。
setup_vector_search_index( wiki_data_collection, vector_index_definition_scalar_quantized, vector_search_scalar_quantized_index_name,)setup_vector_search_index( wiki_data_collection, vector_index_definition_binary_quantized, vector_search_binary_quantized_index_name,)setup_vector_search_index( wiki_data_collection, vector_index_definition_float32_ann, vector_search_float32_ann_index_name,)重要：操作が完了するまでに数分かかる可能性があります。インデックスをクエリで使用するには、インデックスが Ready の状態である必要があります。
クラスターにログインし、Search & Vector Search ページのインデックスを視覚的に検査して、インデックスの作成が成功したことを確認します。
MongoDB ベクトル検索インデックスを使用して埋め込みを生成し、コレクションをクエリするための関数を定義します。
このコードは、次の関数を定義します。
get_embedding()関数は、Voyage AIのvoyage-3-large埋め込みモデルを使用して、指定されたテキストの1024ディメンションの埋め込みを生成します。custom_vector_search関数は次の入力パラメータを受け取り、ベクトル検索操作の結果を返します。user_query埋め込みを生成するためのクエリテキスト文字列。
collection検索するMongoDBコレクション。
embedding_path埋め込みを含むコレクション内のフィールド。
vector_search_index_nameクエリで使用するインデックスの名前。
top_k結果として返す上位ドキュメントの数。
num_candidates検討する候補の数。
use_full_precisionFalseの場合は近似最近傍探索を、Trueの場合は厳密最近傍探索を実行するためのフラグです。Ann
use_full_precisionFalse検索の場合、 値はデフォルトで に設定されます。use_full_precisionTrueEXN 検索を実行するには、 の値を に設定します。具体的には、この関数は次のアクションを実行します。
クエリテキストの埋め込みを生成します。
$vectorSearchステージを構築します検索方法を設定する
返すフィールドをコレクション内で指定します
パフォーマンス統計を収集した後、パイプラインを実行します。
結果を返します
def get_embedding(text, task_prefix="document"): """Fetch embedding for a given text using Voyage AI.""" if not text.strip(): print("Empty text provided for embedding.") return [] result = voyage_client.embed([text], model="voyage-3-large", input_type=task_prefix) return result.embeddings[0]
def custom_vector_search( user_query, collection, embedding_path, vector_search_index_name="vector_index", top_k=5, num_candidates=25, use_full_precision=False,): """Perform vector search on a MongoDB collection using specified index.""" # Generate embedding for the user query query_embedding = get_embedding(user_query, task_prefix="query")
if query_embedding is None or not query_embedding: return { "error": "Invalid query or embedding generation failed.", "execution_time_ms": None, "results": [], }
# Define the vector search stage vector_search_stage = { "$vectorSearch": { "index": vector_search_index_name, "queryVector": query_embedding, "path": embedding_path, "limit": top_k, } }
# Configure search precision approach if not use_full_precision: # For approximate nearest neighbor (ANN) search vector_search_stage["$vectorSearch"]["numCandidates"] = num_candidates else: # For exact nearest neighbor (ENN) search vector_search_stage["$vectorSearch"]["exact"] = True
# Project stage to fetch desired fields and vector search score project_stage = { "$project": { "_id": 0, "title": 1, "text": 1, "wiki_id": 1, "url": 1, "score": {"$meta": "vectorSearchScore"}, } }
# Define the aggregate pipeline pipeline = [vector_search_stage, project_stage]
try: # Execute the explain command to measure latency explain_result = collection.database.command( "explain", {"aggregate": collection.name, "pipeline": pipeline, "cursor": {}}, verbosity="executionStats", )
# Extract the execution time vector_search_explain = explain_result["stages"][0]["$vectorSearch"] execution_time_ms = vector_search_explain["explain"]["query"]["stats"]["context"]["millisElapsed"]
# Execute the actual aggregate query results = list(collection.aggregate(pipeline))
return { "results": results, "execution_time_ms": execution_time_ms, } except Exception as e: print(f"Error during vector search: {e}") return { "error": str(e), "execution_time_ms": None, "results": [], }MongoDB ベクトル検索クエリを実行して、検索パフォーマンスを評価します。
次のクエリは、さまざまな量子化戦略にわたってベクトル検索を実行し、スカラー量子化、バイナリ量子化、および全精度（float32）ベクトルのパフォーマンスメトリクスを測定し、各精度レベルでのレイテンシー測定を取得し、分析比較のために結果形式を標準化します。クエリ文字列には、Voyage AIを使用して生成された埋め込みを使用しています "How do I increase my productivity for maximum output"（最大限の出力を得るために生産性を上げるにはどうすればよいか）。
クエリは、精度レベル（スカラー、バイナリ、float32）、結果セットのサイズ（top_k）、ミリ秒単位でのクエリのレイテンシ、検索したドキュメントの内容など、主なパフォーマンス指標をresults変数に保存し、さまざまな量子化戦略の検索パフォーマンスを横断的に評価するための包括的なメトリクスを提供します。
vector_search_indices = [ vector_search_float32_ann_index_name, vector_search_scalar_quantized_index_name, vector_search_binary_quantized_index_name,]
# Random queryuser_query = "How do I increase my productivity for maximum output"test_top_k = 5test_num_candidates = 25
# Result is a list of dictionaries with the following headings: precision, top_k, latency_ms, resultsresults = []
for vector_search_index in vector_search_indices: # Conduct a vector search operation using scalar quantized vector_search_results = custom_vector_search( user_query, wiki_data_collection, embedding_path="embedding", vector_search_index_name=vector_search_index, top_k=test_top_k, num_candidates=test_num_candidates, use_full_precision=False, ) # Include the precision in the results precision = vector_search_index.split("vector_index")[1] precision = precision.replace("quantized", "").capitalize()
results.append( { "precision": precision, "top_k": test_top_k, "num_candidates": test_num_candidates, "latency_ms": vector_search_results["execution_time_ms"], "results": vector_search_results["results"][0], # Just taking the first result, modify this to include more results if needed } )
# Conduct a vector search operation using full precisionprecision = "Float32_ENN"vector_search_results = custom_vector_search( user_query, wiki_data_collection, embedding_path="embedding", vector_search_index_name="vector_index_scalar_quantized", top_k=test_top_k, num_candidates=test_num_candidates, use_full_precision=True,)
results.append( { "precision": precision, "top_k": test_top_k, "num_candidates": test_num_candidates, "latency_ms": vector_search_results["execution_time_ms"], "results": vector_search_results["results"][0], # Just taking the first result, modify this to include more results if needed })
# Convert the results to a pandas DataFrame with the headings: precision, top_k, latency_msresults_df = pd.DataFrame(results)results_df.columns = ["precision", "top_k", "num_candidates", "latency_ms", "results"]
# To display the results:results_df.head()結果のパフォーマンスメトリクスは、精度レベル間のレイテンシの違いを示しています。ここからは、量子化によってパフォーマンスが大幅に向上する一方、精度と検索速度の間に明確なトレードオフが存在し、フル精度の float32 操作は、量子化された操作と比較して計算時間が著しく長くなることがわかります。
レイテンシを測定するには、top-k および num_candidates の値を変更します。
次のクエリは、異なる精度レベルや取得規模にわたってベクトル検索のパフォーマンスを評価する、体系的なレイテンシ測定フレームワークを紹介します。パラメーター top-k は、返される結果の数を決定するだけでなく、MongoDB の HNSW グラフ検索における numCandidates パラメーターも設定します。
numCandidatesの値は、DNS 検索中にMongoDB ベクトル検索 が探索する HNSWグラフ内のノードの数に影響します。ここで、値が大きいほど、真の最近傍が見つかる可能性は高くなりますが、計算時間がより長く必要になります。
latency_msを人間が読める形式に変換する関数を定義します。from datetime import timedelta def format_time(ms): """Convert milliseconds to a human-readable format""" delta = timedelta(milliseconds=ms) # Extract minutes, seconds, and milliseconds with more precision minutes = delta.seconds // 60 seconds = delta.seconds % 60 milliseconds = round(ms % 1000, 3) # Keep 3 decimal places for milliseconds # Format based on duration if minutes > 0: return f"{minutes}m {seconds}.{milliseconds:03.0f}s" elif seconds > 0: return f"{seconds}.{milliseconds:03.0f}s" else: return f"{milliseconds:.3f}ms"ベクトル検索クエリのレイテンシを測定する関数を定義します。
次の関数は、
user_query、collection、vector_search_index_name、use_full_precision値、top_k_values値、およびnum_candidates_values値を入力として受け取り、ベクトル検索の結果を返します。ここで次の点をメモしておきます。top_kとnum_candidatesの値が増加すると、レイテンシが増加します。これは、ベクトル検索操作でより多くのドキュメントが使用され、検索に時間がかかるためです。完全忠実度検索 (
use_full_precision=True) のレイテンシは、近似検索 (use_full_precision=False) よりも高くなります。これは、完全忠実度検索がデータセット全体を検索し、フル精度の float32 ベクトルを使用するため、近似検索よりも時間がかかるからです。量子化検索では近似検索と量子化ベクトルを使用するため、レイテンシは完全忠実度検索よりも低くなります。
def measure_latency_with_varying_topk( user_query, collection, vector_search_index_name="vector_index_scalar_quantized", use_full_precision=False, top_k_values=[5, 10, 100], num_candidates_values=[25, 50, 100, 200, 500, 1000, 2000, 5000, 10000],): results_data = [] # Conduct vector search operation for each (top_k, num_candidates) combination for top_k in top_k_values: for num_candidates in num_candidates_values: # Skip scenarios where num_candidates < top_k if num_candidates < top_k: continue # Construct the precision name precision_name = vector_search_index_name.split("vector_index")[1] precision_name = precision_name.replace("quantized", "").capitalize() # If use_full_precision is true, then the precision name is "_float32_" if use_full_precision: precision_name = "_float32_ENN" # Perform the vector search vector_search_results = custom_vector_search( user_query=user_query, collection=collection, embedding_path="embedding", vector_search_index_name=vector_search_index_name, top_k=top_k, num_candidates=num_candidates, use_full_precision=use_full_precision, ) # Extract the execution time (latency) latency_ms = vector_search_results["execution_time_ms"] # Append the results to the list results_data.append( { "precision": precision_name, "top_k": top_k, "num_candidates": num_candidates, "latency_ms": latency_ms, } ) print(f"Top-K: {top_k}, NumCandidates: {num_candidates}, " f"Latency: {format_time(latency_ms)}, Precision: {precision_name}") return results_dataMongoDB ベクトル検索クエリを実行して、レイテンシを測定します。
レイテンシ評価操作は、すべての量子化戦略にわたって検索を実行し、複数の結果セットサイズをテストし、標準化されたパフォーマンスメトリクスを取得し、結果を集計して比較分析を行うことで、包括的なパフォーマンス分析を実施します。これにより、さまざまな構成や検索負荷におけるベクトル検索の動作を詳細に評価することができます。
# Run the measurementsuser_query = "How do I increase my productivity for maximum output"top_k_values = [5, 10, 50, 100]num_candidates_values = [25, 50, 100, 200, 500, 1000, 2000, 5000, 10000] latency_results = [] for vector_search_index in vector_search_indices: latency_results.append( measure_latency_with_varying_topk( user_query, wiki_data_collection, vector_search_index_name=vector_search_index, use_full_precision=False, top_k_values=top_k_values, num_candidates_values=num_candidates_values, ) ) # Conduct vector search operation using full precisionlatency_results.append( measure_latency_with_varying_topk( user_query, wiki_data_collection, vector_search_index_name="vector_index_scalar_quantized", use_full_precision=True, top_k_values=top_k_values, num_candidates_values=num_candidates_values, ))レイテンシの測定結果は、精度タイプ間の明確なパフォーマンス階層を示しています。バイナリ量子化が最も高速な検索時間を示し、スカラー量子化がそれに続きます。フル精度の float32 近似最近傍探索操作は、レイテンシが著しく高くなります。Top-K の値が大きくなるにつれて、量子化検索とフル精度検索間のパフォーマンスの差がより顕著になります。float32 厳密最近傍探索操作は最も時間がかかりますが、最も精度の高い結果を提供します。
さまざまなトップk値に対する検索レイテンシをプロットします。
import matplotlib.pyplot as plt # Map your precision field to the labels and colors you want in the legendprecision_label_map = { "_scalar_": "scalar", "_binary_": "binary", "_float32_ann": "float32_ann", "_float32_ENN": "float32_ENN",} precision_color_map = { "_scalar_": "orange", "_binary_": "red", "_float32_ann": "blue", "_float32_ENN": "purple",} # Flatten all measurements and find the unique top_k valuesall_measurements = [m for precision_list in latency_results for m in precision_list]unique_topk = sorted(set(m["top_k"] for m in all_measurements)) # For each top_k, create a separate plotfor k in unique_topk: plt.figure(figsize=(10, 6)) # For each precision type, filter out measurements for the current top_k value for measurements in latency_results: # Filter measurements with top_k equal to the current k filtered = [m for m in measurements if m["top_k"] == k] if not filtered: continue # Extract x (num_candidates) and y (latency) values x = [m["num_candidates"] for m in filtered] y = [m["latency_ms"] for m in filtered] # Determine the precision, label, and color from the first measurement in this filtered list precision = filtered[0]["precision"] label = precision_label_map.get(precision, precision) color = precision_color_map.get(precision, "blue") # Plot the line for this precision type plt.plot(x, y, marker="o", color=color, label=label) # Label axes and add title including the top_k value plt.xlabel("Number of Candidates") plt.ylabel("Latency (ms)") plt.title(f"Search Latency vs Num Candidates for Top-K = {k}") # Add a legend and grid, then show the plot plt.legend() plt.grid(True) plt.show()このコードは次のレイテンシチャートを返します。これは、
top-k（取得される結果の数）が増加するにつれて、バイナリ、スカラー、float32などのさまざまな埋め込み精度タイプでベクトル検索ドキュメントの検索がどのように実行されるかを示しています。
表現キャパシティーとリテンションを測定します。
次のクエリは、 MongoDB ベクトル検索 が基礎データセットから関連ドキュメントをどの程度効果的に検索するかを測定します。これは、フィールド 真実の関連ドキュメントの総数 (Found/ Total) に対する関連ドキュメントの総数に対する比率として計算されます。例、クエリにフィールド 真実の関連ドキュメントが 5 あり、 MongoDB ベクトル検索 がそれらのドキュメントの 4 を見つけた場合、呼び出しは 0.8 になりますまたは 80%。
ベクトル検索操作の表現能力と保持力を測定する関数を定義します。この関数は次の処理を行います。
フル精度の float32 ベクトルと 厳密最近傍探索 を使用してベースライン検索を作成します。
量子化ベクトルと近似最近傍探索を使用して量子化検索を作成します。
量子化検索の保持率をベースライン検索と比較して計算します。
量子化検索のために、保持は合理的な範囲内で維持されなければなりません。表現能力が低い場合、ベクトル検索操作がクエリのセマンティックな意味を捉えられず、結果が正確でない可能性があります。これは、量子化が効果的ではなく、使用された初期の埋め込みモデルが量子化プロセスに適していないことを示しています。量子化に対応した埋め込みモデルの利用をお勧めします。これは、訓練プロセス中に、量子化後もセマンティックな特性を維持するようにモデルが特別に最適化されていることを意味します。
def measure_representational_capacity_retention_against_float_enn( ground_truth_collection, collection, quantized_index_name, # This is used for both the quantized search and (with use_full_precision=True) for the baseline. top_k_values, # List/array of top-k values to test. num_candidates_values, # List/array of num_candidates values to test. num_queries_to_test=1,): retention_results = {"per_query_retention": {}} overall_retention = {} # overall_retention[top_k][num_candidates] = [list of retention values] # Initialize overall retention structure for top_k in top_k_values: overall_retention[top_k] = {} for num_candidates in num_candidates_values: if num_candidates < top_k: continue overall_retention[top_k][num_candidates] = [] # Extract and store the precision name from the quantized index name. precision_name = quantized_index_name.split("vector_index")[1] precision_name = precision_name.replace("quantized", "").capitalize() retention_results["precision_name"] = precision_name retention_results["top_k_values"] = top_k_values retention_results["num_candidates_values"] = num_candidates_values # Load ground truth annotations ground_truth_annotations = list( ground_truth_collection.find().limit(num_queries_to_test) ) print(f"Loaded {len(ground_truth_annotations)} ground truth annotations") # Process each ground truth annotation for annotation in ground_truth_annotations: # Use the ground truth wiki_id from the annotation. ground_truth_wiki_id = annotation["wiki_id"] # Process only queries that are questions. for query_type, queries in annotation["queries"].items(): if query_type.lower() not in ["question", "questions"]: continue for query in queries: # Prepare nested dict for this query if query not in retention_results["per_query_retention"]: retention_results["per_query_retention"][query] = {} # For each valid combination of top_k and num_candidates for top_k in top_k_values: if top_k not in retention_results["per_query_retention"][query]: retention_results["per_query_retention"][query][top_k] = {} for num_candidates in num_candidates_values: if num_candidates < top_k: continue # Baseline search: full precision using ENN (Float32) baseline_result = custom_vector_search( user_query=query, collection=collection, embedding_path="embedding", vector_search_index_name=quantized_index_name, top_k=top_k, num_candidates=num_candidates, use_full_precision=True, ) baseline_ids = { res["wiki_id"] for res in baseline_result["results"] } # Quantized search: quantized_result = custom_vector_search( user_query=query, collection=collection, embedding_path="embedding", vector_search_index_name=quantized_index_name, top_k=top_k, num_candidates=num_candidates, use_full_precision=False, ) quantized_ids = { res["wiki_id"] for res in quantized_result["results"] } # Compute retention for this combination if baseline_ids: retention = len( baseline_ids.intersection(quantized_ids) ) / len(baseline_ids) else: retention = 0 # Store the results per query retention_results["per_query_retention"][query].setdefault( top_k, {} )[num_candidates] = { "ground_truth_wiki_id": ground_truth_wiki_id, "baseline_ids": sorted(baseline_ids), "quantized_ids": sorted(quantized_ids), "retention": retention, } overall_retention[top_k][num_candidates].append(retention) print( f"Query: '{query}' | top_k: {top_k}, num_candidates: {num_candidates}" ) print(f" Ground Truth wiki_id: {ground_truth_wiki_id}") print(f" Baseline IDs (Float32): {sorted(baseline_ids)}") print( f" Quantized IDs: {precision_name}: {sorted(quantized_ids)}" ) print(f" Retention: {retention:.4f}\n") # Compute overall average retention per combination avg_overall_retention = {} for top_k, cand_dict in overall_retention.items(): avg_overall_retention[top_k] = {} for num_candidates, retentions in cand_dict.items(): if retentions: avg = sum(retentions) / len(retentions) else: avg = 0 avg_overall_retention[top_k][num_candidates] = avg print( f"Overall Average Retention for top_k {top_k}, num_candidates {num_candidates}: {avg:.4f}" ) retention_results["average_retention"] = avg_overall_retention return retention_resultsMongoDB ベクトル検索インデックスのパフォーマンスを評価して比較します。
# Access the databaseDB_NAME = "testing_datasets"db = mongo_client[DB_NAME] # Access collectionswiki_data_collection = db["wikipedia-22-12-en"]ground_truth_collection = db["wikipedia-22-12-en-annotation"] overall_recall_results = []top_k_values = [5, 10, 50, 100]num_candidates_values = [25, 50, 100, 200, 500, 1000, 5000]num_queries_to_test = 1 for vector_search_index in vector_search_indices: overall_recall_results.append( measure_representational_capacity_retention_against_float_enn( ground_truth_collection=ground_truth_collection, collection=wiki_data_collection, quantized_index_name=vector_search_index, top_k_values=top_k_values, num_candidates_values=num_candidates_values, num_queries_to_test=num_queries_to_test, ) )出力には、グラウンド トゥルース データセット内の各クエリの保持結果が表示されます。保持は 0 から 1 までの小数で表されます。1.0 はグラウンド トゥルース ID が保持されることを意味し、0.25 はグラウンド トゥルース ID の 25% のみが保持されることを意味します。
さまざまな精度タイプの保持機能をプロットしてください。
import matplotlib.pyplot as plt # Define colors and labels for each precision typeprecision_colors = {"_scalar_": "orange", "_binary_": "red", "_float32_": "green"} if overall_recall_results: # Determine unique top_k values from the first result's average_retention keys unique_topk = sorted(list(overall_recall_results[0]["average_retention"].keys())) for k in unique_topk: plt.figure(figsize=(10, 6)) # For each precision type, plot retention vs. number of candidates at this top_k for result in overall_recall_results: precision_name = result.get("precision_name", "unknown") color = precision_colors.get(precision_name, "blue") # Get candidate values from the average_retention dictionary for top_k k candidate_values = sorted(result["average_retention"][k].keys()) retention_values = [ result["average_retention"][k][nc] for nc in candidate_values ] plt.plot( candidate_values, retention_values, marker="o", label=precision_name.strip("_"), color=color, ) plt.xlabel("Number of Candidates") plt.ylabel("Retention Score") plt.title(f"Retention vs Number of Candidates for Top-K = {k}") plt.legend() plt.grid(True) plt.show() # Print detailed average retention results print("\nDetailed Average Retention Results:") for result in overall_recall_results: precision_name = result.get("precision_name", "unknown") print(f"\n{precision_name} Embedding:") for k in sorted(result["average_retention"].keys()): print(f"\nTop-K: {k}") for nc in sorted(result["average_retention"][k].keys()): ret = result["average_retention"][k][nc] print(f" NumCandidates: {nc}, Retention: {ret:.4f}")このコードは、次のリテンション チャートを返します。
float32_ann、scalar、binaryの埋め込みに対して、コードは以下のような詳細な平均保持結果も返します。Detailed Average Retention Results: _float32_ann Embedding: Top-K: 5 NumCandidates: 25, Retention: 1.0000 NumCandidates: 50, Retention: 1.0000 NumCandidates: 100, Retention: 1.0000 NumCandidates: 200, Retention: 1.0000 NumCandidates: 500, Retention: 1.0000 NumCandidates: 1000, Retention: 1.0000 NumCandidates: 5000, Retention: 1.0000 Top-K: 10 NumCandidates: 25, Retention: 1.0000 NumCandidates: 50, Retention: 1.0000 NumCandidates: 100, Retention: 1.0000 NumCandidates: 200, Retention: 1.0000 NumCandidates: 500, Retention: 1.0000 NumCandidates: 1000, Retention: 1.0000 NumCandidates: 5000, Retention: 1.0000 Top-K: 50 NumCandidates: 50, Retention: 0.8125 NumCandidates: 100, Retention: 0.8750 NumCandidates: 200, Retention: 0.8750 NumCandidates: 500, Retention: 1.0000 NumCandidates: 1000, Retention: 1.0000 NumCandidates: 5000, Retention: 1.0000 Top-K: 100 NumCandidates: 100, Retention: 0.7586 NumCandidates: 200, Retention: 0.7241 NumCandidates: 500, Retention: 0.9655 NumCandidates: 1000, Retention: 0.9655 NumCandidates: 5000, Retention: 1.0000 _scalar_ Embedding: Top-K: 5 NumCandidates: 25, Retention: 0.2500 NumCandidates: 50, Retention: 0.5000 NumCandidates: 100, Retention: 0.7500 NumCandidates: 200, Retention: 1.0000 NumCandidates: 500, Retention: 1.0000 NumCandidates: 1000, Retention: 1.0000 NumCandidates: 5000, Retention: 1.0000 Top-K: 10 NumCandidates: 25, Retention: 0.4000 NumCandidates: 50, Retention: 0.6000 NumCandidates: 100, Retention: 0.8000 NumCandidates: 200, Retention: 1.0000 NumCandidates: 500, Retention: 1.0000 NumCandidates: 1000, Retention: 1.0000 NumCandidates: 5000, Retention: 1.0000 Top-K: 50 NumCandidates: 50, Retention: 0.7500 NumCandidates: 100, Retention: 0.8125 NumCandidates: 200, Retention: 0.8750 NumCandidates: 500, Retention: 0.9375 NumCandidates: 1000, Retention: 0.9375 NumCandidates: 5000, Retention: 1.0000 Top-K: 100 NumCandidates: 100, Retention: 0.8276 NumCandidates: 200, Retention: 0.8276 NumCandidates: 500, Retention: 0.8621 NumCandidates: 1000, Retention: 0.8966 NumCandidates: 5000, Retention: 0.9310 _binary_ Embedding: Top-K: 5 NumCandidates: 25, Retention: 0.2500 NumCandidates: 50, Retention: 0.2500 NumCandidates: 100, Retention: 0.7500 NumCandidates: 200, Retention: 0.7500 NumCandidates: 500, Retention: 1.0000 NumCandidates: 1000, Retention: 1.0000 NumCandidates: 5000, Retention: 1.0000 Top-K: 10 NumCandidates: 25, Retention: 0.2000 NumCandidates: 50, Retention: 0.2000 NumCandidates: 100, Retention: 0.8000 NumCandidates: 200, Retention: 0.8000 NumCandidates: 500, Retention: 1.0000 NumCandidates: 1000, Retention: 1.0000 NumCandidates: 5000, Retention: 1.0000 Top-K: 50 NumCandidates: 50, Retention: 0.2500 NumCandidates: 100, Retention: 0.5625 NumCandidates: 200, Retention: 0.6250 NumCandidates: 500, Retention: 0.7500 NumCandidates: 1000, Retention: 0.8125 NumCandidates: 5000, Retention: 1.0000 Top-K: 100 NumCandidates: 100, Retention: 0.4483 NumCandidates: 200, Retention: 0.5517 NumCandidates: 500, Retention: 0.7586 NumCandidates: 1000, Retention: 0.8621 NumCandidates: 5000, Retention: 1.0000リコール結果は、3 つの埋め込みタイプ間で異なるパフォーマンス パターンを示しています。
スカラー量子化は着実な改善を示し、より高い K 値で強力な検索精度を示しています。2 進量子化は、最初は低いものの、Top-K 50 と Top-K 100 で改善し、計算効率とリコール性能のトレードオフを示唆しています。float32 の埋め込みは、最も強力な初期パフォーマンスを示し、Top-K 50 および Top-K 100 でスカラー量子化と同じ最大リコールに達します。
これは、float32 が低い Top-K 値でより高い再現率を提供する一方で、スカラー量子化は計算効率を向上させつつ、高い Top-K 値でも同等のパフォーマンスを達成できることを示唆しています。2 進量子化は、リコールの上限が低いにもかかわらず、メモリおよび計算の制約が最大限のリコール精度の必要性を上回るシナリオでは依然として価値があるかもしれません。
