OceanBaseはV4.3.3からベクトル型データの格納、ベクトルインデックス、そしてembeddingベクトル検索機能をサポートしています。これにより、ベクトル化したデータをOceanBaseに保存し、次の検索処理で利用できるようになりました。
Qwenは、Alibaba Cloudが開発した大規模言語モデルであり、ユーザーの入力を理解・分析するために使用されます。Alibaba Cloud Model Experience Centerでは、QwenモデルのAPIサービスを利用できます。
本記事では、Qwen APIを利用してベクトル化したデータをOceanBase Cloudに保存し、OceanBase Cloudのベクトル検索機能を用いて関連データを検索する方法について説明します。
前提条件
- 環境に利用可能なトランザクション型(MySQL)のクラスタインスタンスがあります。
テナントの作成を参照してテナントを作成した後、以下の手順を実行してください。
環境に使用可能なMySQL互換モードのテナント、MySQLデータベース、およびアカウントが既に存在し、データベースアカウントに読み書き権限が付与されていること。作成方法の詳細についてはアカウントの作成およびデータベースの作成(MySQL専用)をご参照ください。
プロジェクト管理者またはインスタンス管理者ロールを持ち、プロジェクト内のインスタンスに対する読み書き操作を実行できること。権限がない場合は、組織管理者に連絡して権限の追加を依頼してください。
Python 3.9以降のバージョンと対応するpipをインストールしています。マシンのPythonバージョンが低い場合は、Minicondaを使用して新しいPython 3.9以降の環境を作成できます。詳細については、Minicondaインストールガイドをご参照ください。
Poetry、Pyobvector、およびDashScopeをインストールします：
python3 -m pip install poetry python3 -m pip install pyobvector python3 -m pip install dashscope
ステップ1：データベース接続情報を取得する
ドロップダウンリストから、ID **クラスタインスタンスを選択**します。
**概要**ページに移動します。
接続をクリックし、**接続文字列を取得**を選択します。
ポップアップウィンドウで、**パブリックネットワークを使用**を選択します。
アクセスアドレスを取得し、**現在のブラウザのIPアドレスを追加**を選択します。
データベース関連情報を入力し、**接続文字列をコピー**します。
ステップ2：LLMプラットフォームアカウントを登録する
Alibaba Cloud Bailianアカウントに登録し、モデルサービスを有効化してAPIキーを取得します。
注意
Alibaba Cloud Bailian大規模言語モデルサービスを有効化するには、第三者プラットフォームに移動して完了する必要があります。この操作は第三者プラットフォームの料金規則に従い、該当する料金が発生する可能性があります。続行する前に、その公式ウェブサイトを訪問するか関連ドキュメントを確認し、料金体系を確認して同意してください。同意しない場合は、操作を続けないでください。
API Keyを環境変数に設定します：
- Unixベースのシステム(UbuntuやMacOSなど)の場合、端末で以下のコマンドを実行します：
export DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY"
- Windowsシステムの場合、コマンドプロンプトで以下のコマンドを実行します。
set DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY"
YOUR_DASHSCOPE_API_KEY を実際のQwen APIキーに置き換えてください。
ステップ3：ベクトルストレージとクエリ
OceanBase Cloudへのベクトルデータの保存
- テストデータの準備
事前にベクトル化されたデータを含むCSVファイルをダウンロードします。このファイルは、Alibaba Cloudが提供する1000件のグルメレビューを含む公開データセットで、最終列はベクトル化された値なので、再度ベクトルを計算する必要はありません。以下のコードを使用して、embedding列（つまりベクトル列）を再計算し、新しいCSVファイルを生成することもできます。
import dashscope
import pandas as pd
input_datapath = "./fine_food_reviews.csv"
# ここではtext_embedding_v1埋め込みモデルを使用しています。必要に応じて調整可能です。
def generate_embeddings(text):
rsp = dashscope.TextEmbedding.call(model=TextEmbedding.models.text_embedding_v1, input=text)
embeddings = [record['embedding'] for record in rsp.output['embeddings']]
return embeddings if isinstance(text, list) else embeddings[0]
df = pd.read_csv(input_datapath, index_col=0)
# 実際の生成には数分かかります。Qwen Embedding APIを行ごとに呼び出します。
df["embedding"] = df.combined.apply(generate_embeddings)
output_datapath = './fine_food_reviews_self_embeddings.csv'
df.to_csv(output_datapath)
- 以下のスクリプトを実行して、テストデータをOceanBase Cloudに挿入します。スクリプトがあるディレクトリは、テストデータがあるディレクトリと同じである必要があります。
import os
import sys
import csv
import json
from pyobvector import *
from sqlalchemy import Column, Integer, String
# pyobvectorを使用してOBに接続する場合、ユーザー名やパスワードに@が含まれる場合は、%40に置き換えてください。
client = ObVecClient(uri="host:port", user="username",password="****",db_name="test")
# 事前に準備されたテストデータセットは既にベクトル化されています。デフォルトではPythonスクリプトと同じディレクトリに配置されています。自分で再度ベクトル化した場合は、対応するファイルに置き換えてください。
file_name = "fine_food_reviews.csv"
file_path = os.path.join("./", file_name)
# 列を定義します。ベクトル化された列は最後のフィールドに配置されています。
cols = [
Column('id', Integer, primary_key=True, autoincrement=False),
Column('product_id', String(256), nullable=True),
Column('user_id', String(256), nullable=True),
Column('score', Integer, nullable=True),
Column('summary', String(2048), nullable=True),
Column('text', String(8192), nullable=True),
Column('combined', String(8192), nullable=True),
Column('n_tokens', Integer, nullable=True),
Column('embedding', VECTOR(1536))
]
# テーブル名
table_name = 'fine_food_reviews'
# テーブルが存在しない場合は作成します。
if not client.check_table_exists(table_name):
client.create_table(table_name,columns=cols)
# ベクトル列のインデックスを作成します。
client.create_index(
table_name=table_name,
is_vec_index=True,
index_name='vidx',
column_names=['embedding'],
vidx_params='distance=l2, type=hnsw, lib=vsag',
)
# CSVファイルを開いて読み込みます。
with open(file_name, mode='r', newline='', encoding='utf-8') as csvfile:
csvreader = csv.reader(csvfile)
# ヘッダー行を読み取ります。
headers = next(csvreader)
print("Headers:", headers)
batch = [] # データを格納し、10行ごとにデータベースに挿入します。
for i, row in enumerate(csvreader):
# CSVファイルには9つのフィールドがあります: id,product_id,user_id,score,summary,text,combined,n_tokens,embedding
if not row:
break
food_review_line= {'id':row[0],'product_id':row[1],'user_id':row[2],'score':row[3],'summary':row[4],'text':row[5],\
'combined':row[6],'n_tokens':row[7],'embedding':json.loads(row[8])}
batch.append(food_review_line)
# 10行ごとに挿入します。
if (i + 1) % 10 == 0:
client.insert(table_name,batch)
batch = [] # バッファをクリア
# 残りの行（ある場合）を挿入します。
if batch:
client.insert(table_name,batch)
# テーブル内のデータを確認し、すべてのデータが挿入されたことを保証します。
count_sql = f"select count(*) from {table_name};"
cursor = client.perform_raw_text_sql(count_sql)
result = cursor.fetchone()
print(f"インポートデータ総件数:{result[0]}")
ベクトルデータのクエリ
- 以下のPythonスクリプトを
query.pyという名前で保存します。
import os
import sys
import csv
import json
from pyobvector import *
from sqlalchemy import func
import dashscope
# コマンドライン引数を取得する
if len(sys.argv) != 2:
print("検索キーワードを入力してください。")
sys.exit()
queryStatement = sys.argv[1]
# pyobvectorを使用してOBに接続する際、ユーザー名やパスワードに@が含まれる場合は、%40に置き換えてください。
client = ObVecClient(uri="host:port", user="usename",password="****",db_name="test")
# テキストベクトルを生成する関数を定義する
def generate_embeddings(text):
rsp = dashscope.TextEmbedding.call(model=TextEmbedding.models.text_embedding_v1, input=text)
embeddings = [record['embedding'] for record in rsp.output['embeddings']]
return embeddings if isinstance(text, list) else embeddings[0]
def query_ob(query, tableName, vector_name="embedding", top_k=1):
embedding = generate_embeddings(query)
# 近似最近傍探索を実行する
res = client.ann_search(
table_name=tableName,
vec_data=embedding,
vec_column_name=vector_name,
distance_func=func.l2_distance,
topk=top_k,
output_column_names=['combined']
)
for row in res:
print(str(row[0]).replace("Title: ", "").replace("; Content: ", ": "))
# テーブル名
table_name = 'fine_food_reviews'
query_ob(queryStatement,table_name,'embedding',1)
- 質問を入力し、関連する回答を出力します。
python3 query.py 'pet food'
期待される結果は次のとおりです：
This is so good!: I purchased this after my sister sent a small bag to me in a gift box. I loved it so much I wanted to find it to buy for myself and keep it around. I always look on Amazon because you can find everything here and true enough, I found this wonderful candy. It is nice to keep in your purse for when you are out and about and get a dry throat or a tickle in the back of your throat. It is also nice to have in a candy dish at home for guests to try.