Qwenは、Alibaba Cloudが開発した大規模言語モデルであり、ユーザー入力の理解と分析に使用されます。Alibaba Cloudのモデルエクスペリエンスセンターで、QwenモデルのAPIサービスを利用することができます。
OceanBaseデータベースは、バージョンV4.3.3からベクトル型データの格納、ベクトルインデックス、そして埋め込み(embedding)ベクトル検索機能をサポートしています。QwenのAPIを利用して、ベクトル化されたデータをOceanBaseデータベースに格納し、OceanBaseのベクトル検索機能を使用して関連するデータを検索します。
前提条件
OceanBaseデータベースV4.3.3以降のバージョンがデプロイ済みで、MySQLモードのテナントが作成されていること。テナントの作成の後、以下の手順を参照して操作します。
ご利用の環境には、既に利用可能なMySQLテナント、MySQLデータベース、およびアカウントが存在し、データベースアカウントには読み書き権限が付与されていること。
Python 3.9以降のバージョンと対応するpipをインストールします。
poetry、pyobvectorとDashScope SDKをインストールします。
pip install poetry pip install pyobvector pip install dashscopeQwen APIキーを準備します。
ベクトル検索機能を有効にするには、テナントで
ob_vector_memory_limit_percentage構成パラメータが設定されていることを確認してください。V4.3.5 BP3より前のバージョンでは、この値を30に設定することを推奨します。V4.3.5 BP3以降のバージョンでは、デフォルト値である0のままにしておくことを推奨します。より正確にこのパラメータを設定する必要がある場合は、ob_vector_memory_limit_percentageを参照してこの値を計算してください。
ステップ1：データベース接続情報を取得する
OceanBaseデータベースのデプロイ担当者または管理者に連絡し、対応するデータベース接続文字列を取得します。例：
obclient -h$host -P$port -u$user_name -p$password -D$database_name
パラメータの説明：
$host：OceanBaseデータベースへの接続IPアドレスです。OceanBaseデータベースプロキシ(OceanBase Database Proxy、ODP)経由で接続する場合はODPのアドレスを、直接接続する場合はOBServerノードのIPアドレスを使用します。$port：OceanBaseデータベースの接続ポートです。ODP経由の接続ではデフォルトで2883が使用されますが、ODPのデプロイ時にカスタマイズ可能です。直接接続の場合はデフォルトで2881が使用され、OceanBaseデータベースのデプロイ時にカスタマイズできます。$database_name：アクセスするデータベースの名前です。注意
テナントに接続するユーザーは、データベースに対する
CREATE、INSERT、DROP、およびSELECT権限が付与されていなければなりません。ユーザー権限の詳細については、MySQLモードの権限分類を参照してください。$user_name：テナントへの接続アカウントです。ODP経由で接続する場合の一般的な形式はユーザー名@テナント名#クラスタ名またはクラスタ名:テナント名:ユーザー名。直接接続の場合はユーザー名@テナント名となります。$password：アカウントのパスワードです。
接続文字列の詳細については、OBClientを使用してOceanBaseテナントに接続するを参照してください。
ステップ2：Qwen APIキーの環境変数を設定する
Unix系のOS(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データベースにベクトルデータを保存する
テストデータの準備 ベクトル化済みデータを含むテスト用のCSVファイルをダウンロードします。このCSVファイルには、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を1行ずつ呼び出します df["embedding"] = df.combined.apply(generate_embeddings) output_datapath = './fine_food_reviews_self_embeddings.csv' df.to_csv(output_datapath)以下のスクリプトを実行して、テストデータをOceanBaseデータベースに挿入します。このスクリプトは、上記で準備したCSVファイルと同じディレクトリに配置して実行してください。
import os import sys import csv import json from pyobvector import * from sqlalchemy import Column, Integer, String # pyobvectorを使用してOceanBaseに接続する際、ユーザー名やパスワードに「@」が含まれる場合は、「%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) # テーブルの列を定義します。ベクトルデータを含む'embedding'列は最後に配置します。 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 = [] # キャッシュをクリア ## 最後に、10行に満たない残りのデータを挿入します。 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]}")
ステップ4：OceanBaseデータベースのデータを検索する
以下の内容で
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を使用してOceanBaseに接続する際、ユーザー名やパスワードに「@」が含まれる場合は、「%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.