Momento Japan Momentoを使用すると、わずか5回のAPIコールで、Pythonアプリケーション用の強力で完全なベクトルインデックスを作成できます。
ベクトルインデックスとは、高次元ベクトルの格納と検索に最適化された特殊なデータストアのことです。リレーショナルデータベースのように行や列指向のデータを検索するのではなく、ベクトルインデックスはベクトル間の類似性を計算するように設計されています。テキスト検索、画像検索、レコメンデーション、画像認識、音声認識などのタスクに特に有用です。ベクトル検索は、データがベクトルとして表現でき、類似性検索が重要なあらゆるアプリケーションで活躍します。ベクトル・インデックスのより詳細な紹介は、Kirk Kirnellのベクトル・インデックスについての記事を参照ください。ベクトル・インデックスがどのように機能し、どのように機械学習ワークフローに組み込むかについてのより詳細な情報は、ベクトル検索に関するMichael Landis氏の講演を参照ください。
Momento Vector Indexは、ベクトル・インデックスを素早く簡単に始める方法を提供します。この記事では、ベクトルインデックスを設定し、データを追加し、そのデータを検索する簡単なPythonプログラムを作成します。
前提条件
Momento APIキー
Momento Vector Indexを使用するには、AWS us-west-2リージョンのスーパーユーザーキーが必要です。Momento Consoleにアクセスし、指示に従ってメールアドレス、Googleアカウント、またはGitHubアカウントでログインしてください。
コンソールで、APIキーメニューオプションを選択します。
APIキーのページで、キャッシュの保存場所に一致する情報を選択します：
1.クラウド・プロバイダー クラウドプロバイダー: AWS
2.リージョン: us-west-2
3.キーの種類：Super User
4.(オプション）有効期限
Momento Python SDK
プログラムで使用するにはMomento SDKパッケージをインストールする必要があります。pypiで入手可能です: https://pypi.org/project/momento/. pipを使用している場合は、pip install momentoを実行してください。
最初のMomento Vector Indexプログラムを書く
import asyncio
from momento import (
CredentialProvider,
PreviewVectorIndexClientAsync,
VectorIndexConfigurations,
)
from momento.requests.vector_index import ALL_METADATA, Item, SimilarityMetric
from momento.responses.vector_index import (
CreateIndex,
DeleteIndex,
ListIndexes,
Search,
UpsertItemBatch,
)
async def main() -> None:
configuration = VectorIndexConfigurations.Default.latest()
auth_provider = CredentialProvider.from_environment_variable("MOMENTO_API_KEY")
async with PreviewVectorIndexClientAsync(configuration, auth_provider) as client:
# create a momento vector index
index_name = "getting_started_index"
create_index_response = await client.create_index(index_name, num_dimensions=2,
similarity_metric=SimilarityMetric.COSINE_SIMILARITY)
match create_index_response:
case CreateIndex.Success():
print(f"Index with name {index_name} successfully created!")
case CreateIndex.IndexAlreadyExists():
print(f"Index with name {index_name} already exists")
case CreateIndex.Error() as create_index_error:
print(f"Error while creating index {create_index_error.message}")
# list all indexes
list_indexes_response = await client.list_indexes()
match list_indexes_response:
case ListIndexes.Success() as success:
for index in success.indexes:
print(f"- {index.name!r}")
case ListIndexes.Error() as list_indexes_error:
print(f"Error while listing indexes {list_indexes_error.message}")
# upsert data into the index
items = [
Item(id="item_1", vector=[0.0, 1.0], metadata={"key1": "value1"}),
Item(id="item_2", vector=[1.0, 0.0], metadata={"key2": 12345, "key3": 678.9}),
Item(id="item_3", vector=[-1.0, 0.0], metadata={"key1": ["value2", "value3"]}),
Item(id="item_4", vector=[0.5, 0.5], metadata={"key4": True})
]
upsert_response = await client.upsert_item_batch(
index_name,
items=items,
)
match upsert_response:
case UpsertItemBatch.Success():
print("Successfully added items")
case UpsertItemBatch.Error() as upsert_error:
print(f"Error while adding items to index {index_name}: "
f"{upsert_error.message}")
# wait a short time to ensure the vectors are uploaded
await asyncio.sleep(1)
# search the index
search_response = await client.search(
index_name, query_vector=[1.0, 0.0], top_k=4, metadata_fields=ALL_METADATA
)
match search_response:
case Search.Success() as success:
print(f"Search succeeded with {len(success.hits)} matches:")
print(success.hits)
case Search.Error() as search_error:
print(f"Error while searching on index {index_name}: {search_error.message}")
# delete the index
delete_response = await client.delete_index(index_name)
match delete_response:
case DeleteIndex.Success():
print(f"Index {index_name} deleted successfully!")
case DeleteIndex.Error() as delete_error:
print(f"Error while deleting index {index_name}: {delete_error.message}")
if __name__ == "__main__":
asyncio.run(main())
ここに、これから説明する基本的なMomento Vector Indexの例を示します。インデックスを作成し、すべてのインデックスを一覧表示し、データをアップロードし、そのデータを検索し、最後にインデックスを削除します。
Momentoの依存関係をインストールした後、プログラムを実行すると、次のような出力が得られます：
`% python getting_started.py`
`Index with name getting_started_index already exists`
`- 'getting_started_index'
Successfully added items
Search succeeded with 4 matches:
[SearchHit(id='item_2', score=1.0, metadata={'key3': 678.9, 'key2': 12345}), SearchHit(id='item_4', score=0.7071067690849304, metadata={'key4': True}), SearchHit(id='item_1', score=0.0, metadata={'key1': 'value1'}), SearchHit(id='item_3', score=-1.0, metadata={'key1': ['value2', 'value3']})]
Index getting_started_index deleted successfully!`
次のセクションでは、この出力がどのように作られたかを説明しましょう。
create_client
最初にしなければならないのは、ベクター・インデックス・クライアントを作ることです：
configuration = VectorIndexConfigurations.Default.latest()
auth_provider = CredentialProvider.from_environment_variable("MOMENTO_API_KEY")
async with PreviewVectorIndexClientAsync(configuration, auth_provider) as client:
ベクターインデックスクライアントは、他の Momento クライアントと同様に、設定オブジェクトと認証プロバイダを必要とします。認証プロバイダは Momento API キーを読み込み、解析します。環境変数から読み込むことも、文字列から直接読み込むこともできます。
コンフィギュレーションには、基礎となるgRPCクライアントの設定と、顧客が設定可能な機能が含まれます。毎回作成する手間を省くため、VectorIndexConfigurations.Default.latest() はデフォルト設定の最新版です。後方互換性のためにコンフィギュレーションをバージョンアップしているので、コンフィギュレーションを変更しても顧客のデプロイメントには影響しません。latestは常に最新バージョンを指します。
クライアントは非同期のコンテキスト・マネージャーなので、async withを使えば、コンテキストの外に出たときにクライアントを自動的にクリーンアップできます。
create_index
クライアントができたので、インデックスを作成しましょう：
index_name = "getting_started_index"
create_index_response = await client.create_index(index_name, num_dimensions=2,
similarity_metric=SimilarityMetric.COSINE_SIMILARITY)
match create_index_response:
case CreateIndex.Success():
print(f"Index with name {index_name} successfully created!")
case CreateIndex.IndexAlreadyExists():
print(f"Index with name {index_name} already exists")
case CreateIndex.Error() as create_index_error:
print(f"Error while creating index {create_index_error.message}")
create_index関数は3つの引数を取ります：
・index_name: インデックスの名前
・num_dimensions: インデックスの次元数。
・Smilarity_metric: 検索でベクトルを比較する際に使用するメトリック
この例では、2次元のインデックスを作成します。次元は複雑なデータの特徴や属性を表すので、実際のインデックスには何百もある可能性があります。このようにするのは、検索時にインデックスがベクトルをどのように比較するかを視覚化しやすくするためです。また、類似度の指標として余弦類似度を使用しています。これはベクトル間の角度を比較するもので、-1 から 1 の間で正規化されます。
この関数は、Momento API が使用するエラー処理のパターンを示しています。クライアントメソッドは決して例外を投げてはいけません。その代わりに、さまざまなタイプのコール結果を表すレスポンスオブジェクトを返します。インデックスが作成された場合は Success、 その名前のインデックスが既に存在する場合は IndexAlreadyExists、 呼び出しに失敗した場合は Error となります。すべての Momento 呼び出しは、特定の失敗についての詳細を含むエラーオブジェクトを返すことができます。
list_indexes
インデックスを作成したので、すべてのインデックスを一覧表示することでインデックスを確認できます：
list_indexes_response = await client.list_indexes()
match list_indexes_response:
case ListIndexes.Success() as success:
for index in success.indexes:
print(f"- {index.name!r}")
case ListIndexes.Error() as list_indexes_error:
print(f"Error while listing indexes {list_indexes_error.message}")list_indexes関数は引数をとらず、あなたのアカウントにある地域のすべてのインデックス名のリストを持つSuccessオブジェクトを返します。この関数は、コードがハードコードされたインデックスを使用しておらず、インデックスを調べる必要がある場合や、インデックスを作成しすぎていないことを確認するためにインデックス数を追跡する必要がある場合に便利です。
upsert_item_batch
これで新しいインデックスにベクトルを追加できます：
items = [
Item(id="item_1", vector=[0.0, 1.0], metadata={"key1": "value1"}),
Item(id="item_2", vector=[1.0, 0.0], metadata={"key2": 12345, "key3": 678.9}),
Item(id="item_3", vector=[-1.0, 0.0], metadata={"key1": ["value2", "value3"]}),
Item(id="item_4", vector=[0.5, 0.5], metadata={"key4": True})
]
upsert_response = await client.upsert_item_batch(
index_name,
items=items,
)
match upsert_response:
case UpsertItemBatch.Success():
print("Successfully added items")
case UpsertItemBatch.Error() as upsert_error:
print(f"Error while adding items to index {index_name}: "
f"{upsert_error.message}")
upsert_item_batch関数は、インデックス名とベクトルを表す項目のリストを受け取り、それらをインデックスに挿入し、IDが一致する既存のベクトルを置き換えます。アイテムには、一意のID、インデックスの次元数に一致するベクトル、そしてオプションのメタデータが含まれます。メタデータのキーは文字列でなければなりませんが、値は文字列、int、float、boolean、または文字列のリストにすることができます。
ベクトル[-1.0, 0.0]、[0.0, 1.0]、[0.5, 0.5]、[1.0, 0.0]をアップロードしました。これらは2次元なので、可視化することができます：
角度の違いによって、クエリーベクトルをこれらのベクトルと比較できることがおわかりいただけるだでしょう。
search
インデックスにデータが格納されたので、それを検索することができます：
search_response = await client.search(
index_name, query_vector=[1.0, 0.0], top_k=4, metadata_fields=ALL_METADATA
)
match search_response:
case Search.Success() as success:
print(f"Search succeeded with {len(success.hits)} matches:")
print(success.hits)
case Search.Error() as search_error:
print(f"Error while searching on index {index_name}: {search_error.message}")
検索関数は、インデックス名、インデックスの次元数に一致するクエリ・ベクトル、返す結果の数を表すオプションのtop_k引数、返したいメタデータを表すオプションのmetadata_fields引数を取ります。ここではALL_METADATAのセンチネル値を使用しており、これはすべてのメタデータを返すことを意味しています。例：metadata_fields=["key1", "key3"]を指定すると、これらのフィールド名に一致するメタデータのみを返すように指定できます。metadata_fieldsが指定されない場合、メタデータは返されません。
検索成功のレスポンスには、検索ヒットのリストが含まれます。それぞれのヒットには、ベクトルのID、スコア、つまりそのベクトルと検索ベクトルの類似度（コサイン類似度で-1から1まで）、そしてリクエストされたメタデータが含まれます。以下はプログラムの出力例です：
[SearchHit(id='item_2', score=1.0, metadata={'key3': 678.9, 'key2': 12345}), SearchHit(id='item_4', score=0.7071067690849304, metadata={'key4': True}), SearchHit(id='item_1', score=0.0, metadata={'key1': 'value1'}), SearchHit(id='item_3', score=-1.0, metadata={'key1': ['value2', 'value3']})]
マッチはスコア順に返されます。ベクトル[1.0, 0.0]を検索したので、そのベクトルと完全に一致するitem_2のスコアは1.0です。ベクトル[-1.0, 0.0]を持つitem_3は正反対で、スコアは-1.0です。検索ベクトルと直交するベクトルを持つitem_1のスコアは0.0です。より高次元のベクトルは簡単に視覚化できませんが、パターンは同じです：ベクトルが検索ベクトルに近いほど、そのスコアは1.0に近くなります。ベクトルが検索ベクトルの反対側に一致すればするほど、-1.0に近づきます。
delete_index
最後に、このサンプルの後始末をするためにインデックスを削除します：
delete_response = await client.delete_index(index_name)
match delete_response:
case DeleteIndex.Success():
print(f"Index {index_name} deleted successfully!")
case DeleteIndex.Error() as delete_error:
print(f"Error while deleting index {index_name}: {delete_error.message}")
delete_index関数はインデックス名を受け取り、そのインデックスとその中のすべてのデータを削除します。削除に成功するか、削除するインデックスがない場合は成功を返します。
はじめる準備はできましたか？
Momentoでは、シンプルさを念頭に置いてサービスに取り組んでいます。私たちは、あなたが実際に関心のある問題の解決に集中できるよう、市場で最も速く、最も簡単な開発者体験を提供することを目指しています。Momento Vector Indexは完全にサーバーレスのベクトルインデックスであり、わずか5回のAPIコールで最大限に活用できます。
Momento Vector Indexについてもっと知りたい方は、こちらの開発ドキュメントをお読みください。また、LangChainとの統合についてはこちらのブログをご覧ください。