OceanBaseはpyobvectorを使用したベクトルストレージおよび検索をサポートします。本記事では、クイックスタートの方法を紹介します。pyobvectorはOceanBaseベクトルストレージのPython SDKであり、SQLAlchemyをベースにしており、Milvus APIと基本的には互換性があります。
前提条件
OceanBaseクラスタをデプロイ済みであり、MySQLテナントを作成済みであること。
Python 3.9およびそれ以降のバージョンをインストール済み環境であること。
クイックスタート
pyobvectorのインストール
まず、pyobvectorをローカル環境にインストールする必要があります。以下に示すコマンドをご参照ください：
pip install -U pyobvector
-U パラメータは、環境にpyobvectorが既にインストールされている場合は最新バージョンに自動的にアップグレードし、インストールされていない場合は最新バージョンを直接インストールすることを意味します。
使用方法
pyobvectorは2つのモードをサポートしています：
Milvus互換モード：MilvusLikeClientクラスで提供されるMilvus APIに似た方法でベクトルストレージを使用します。
SQLAlchemyハイブリッドモード：ObVecClientクラスが提供するベクトルストレージ機能を使用して、SQLAlchemyライブラリでリレーショナルデータベースのステートメントを実行します。このモードでは、pyobvectorはSQLAlchemyの拡張機能と見なすことができます。
Milvus互換モード
クライアント接続の確立
pyobvectorではMilvusLikeClientを提供しており、ユーザーはMilvus互換の方法でOceanBaseのベクトルストレージおよび検索機能を利用できます。以下のステートメントを使用して、クライアントオブジェクトを作成できます：
from pyobvector import *
# 下記のデータベース接続情報パラメータを、ご自身のデータベースインスタンスの情報に置き換えてください
client = MilvusLikeClient(uri="127.0.0.1:2881", user="root@test", db_name="test")
ベクトルインデックス付きコレクションの作成
MilvusのAPIとの互換性を維持するため、pyobvectorのMilvusClientも create_collection などのメソッドを提供しています。メソッド名は「コレクション作成」となっていますが、OceanBase上では実際にはテーブルを作成する操作になります。以下の例では、id、embedding、metadata の3つの列を持つテーブルを作成し、embedding 列に対してHNSWタイプのベクトルインデックスを作成できます。embedding 列は64次元のベクトルタイプです。
fields = [
FieldSchema(
name="id",
dtype=DataType.INT64,
is_primary=True,
auto_id=True,
),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=64),
FieldSchema(name="metadata", dtype=DataType.JSON),
]
index_params = MilvusLikeClient.prepare_index_params()
index_params.add_index(
field_name="embedding",
index_name="embedding_idx",
index_type=VecIndexType.HNSW,
distance="l2",
m=16,
ef_construction=256,
)
schema = CollectionSchema(fields)
table_name = "vector_search"
client.create_collection(table_name, schema=schema, index_params=index_params)
データの構築と書き込み
大量のベクトルデータにおける検索のシナリオをシミュレートするために、このステップではまずいくつかのベクトルデータを作成します。方法は、Pythonのrandomモジュールを使用して、ランダムな浮動小数点数のリストを作成することです。
import random
random.seed(20241023)
batch_size = 100
batch = []
for i in range(1000):
batch.append(
{
"embedding": [random.uniform(-1, 1) for _ in range(64)],
"metadata": {"idx": i},
}
)
if len(batch) == batch_size:
client.insert(collection_name=table_name, data=batch)
batch = []
if len(batch) > 0:
client.insert(collection_name=table_name, data=batch)
類似ベクトル検索の実行
random.uniformを使用してターゲットとなるベクトルデータを入力として作成し、先ほどデータを挿入したコレクションでベクトル検索を実行します。
target_data = [random.uniform(-1, 1) for _ in range(64)]
res = client.search(
collection_name=table_name,
data=target_data,
anns_field="embedding",
limit=5,
output_fields=["id", "metadata"],
)
print(res)
# 期待される実行結果は次のとおりです
# [{'id': 63, 'metadata': {'idx': 62}}, {'id': 796, 'metadata': {'idx': 795}}, {'id': 187, 'metadata': {'idx': 186}}, {'id': 784, 'metadata': {'idx': 783}}, {'id': 880, 'metadata': {'idx': 879}}]
SQLAlchemyハイブリッドモード
クライアント接続の確立
pyobvectorは、OceanBaseのベクトルストレージと検索機能をSQLAlchemyのハイブリッドモードで利用できるObVecClientを提供します。以下のステートメントでクライアントオブジェクトを作成できます：
from pyobvector import *
# 下記のデータベース接続情報パラメータを、ご自身のデータベースインスタンスの情報に置き換えてください
client = ObVecClient(uri="127.0.0.1:2881", user="root@test", db_name="test")
テーブルとベクトルインデックスの作成
以下の例では、id、embedding、metadata の3つの列を持つテーブルを作成し、embedding 列に対してHNSWタイプのベクトルインデックスを作成できます。embedding 列は64次元のベクトル型です。
from sqlalchemy import Column, Integer, JSON
from sqlalchemy import func
cols = [
Column("id", Integer, primary_key=True, autoincrement=True),
Column("embedding", VECTOR(64)),
Column("metadata", JSON),
]
table_name = "vector_test3"
client.create_table(table_name, columns=cols)
print(f"Table {table_name} created")
client.create_index(
table_name,
is_vec_index=True,
index_name="embedding_idx",
column_names=["embedding"],
vidx_params="distance=l2, type=hnsw, lib=vsag", # m=16, ef_construction=256
)
print(f"Index {table_name}.embedding_idx created")
データの構築と書き込み
大量のベクトルデータにおける検索のシナリオをシミュレートするために、このステップではまずいくつかのベクトルデータを作成します。方法は、Pythonのrandomモジュールを使用して、ランダムな浮動小数点数のリストを作成することです。
import random
random.seed(20241023)
batch_size = 100
batch = []
for i in range(1000):
batch.append(
{
"embedding": [random.uniform(-1, 1) for _ in range(64)],
"metadata": {"idx": i},
}
)
if len(batch) == batch_size:
client.insert(table_name, data=batch)
batch = []
if len(batch) > 0:
client.insert(table_name, data=batch)
類似ベクトル検索の実行
random.uniformを使用してターゲットとなるベクトルデータを入力として作成し、先ほどデータを挿入したコレクションでベクトル検索を実行します。
target_data = [random.uniform(-1, 1) for _ in range(64)]
res = client.ann_search(
table_name,
vec_data=target_data,
vec_column_name="embedding",
distance_func=func.l2_distance,
topk=5,
output_column_names=["id", "metadata"],
)
for r in res:
print(r)
# 期待される実行結果は次のとおりです
# (63, '{"idx": 62}')
# (796, '{"idx": 795}')
# (187, '{"idx": 186}')
# (784, '{"idx": 783}')
# (880, '{"idx": 879}')
関連ドキュメント
ベクトルデータの詳細については、ベクトルデータを参照してください。
テーブル作成後のベクトルインデックスの作成、削除方法については、ベクトルインデックスを参照してください。