Qwen

Qwenは、Alibaba Cloudが開発した大規模言語モデルであり、ユーザーの入力を理解および分析するために使用されます。Alibaba CloudのModel StudioでQwenモデルのAPIサービスを利用できます。

OceanBaseデータベースは、ベクトル型データの格納、ベクトルインデックス、そしてembeddingベクトル検索機能を提供しています。QwenのAPIを利用して、ベクトル化したデータをOceanBaseデータベースに保存し、その後OceanBaseのベクトル検索機能を用いて関連するデータを検索することが可能です。

前提条件

• OceanBaseデータベースV4.4.0以降をデプロイし、MySQLテナントを作成していること。テナントの作成後、以下の手順を参照して操作してください。

• 環境に使用可能なMySQLテナント、MySQLデータベース、およびデータベースアカウントが既に存在し、データベースアカウントに読み書き権限が付与されていること。

• Python 3.9以降のバージョンと対応するpipをインストールしていること。

• poetry、pyobvector、およびDashScope SDKをインストールしていること。

  pip install poetry
  pip install pyobvector
  pip install dashscope
  

• Qwen APIキーを準備しておくこと。

• ベクトル検索機能を有効にするために、テナントでob_vector_memory_limit_percentageパラメータを設定していることを確認してください。V4.4.1以前のバージョンでは値を30に設定することを推奨しますが、V4.4.1以降のバージョンではデフォルト値の0のままにしておくことを推奨します。このパラメータをより正確に設定する必要がある場合は、ob_vector_memory_limit_percentageを参照してこの値を計算してください。

ステップ1：OceanBaseデータベースの接続文字列を取得する

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ベースのシステム(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データベースに保存する

1. テストデータの準備 事前にベクトル化されたデータを含む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を行ごとに呼び出します。
    df["embedding"] = df.combined.apply(generate_embeddings)
    output_datapath = './fine_food_reviews_self_embeddings.csv'
    df.to_csv(output_datapath)
   

2. 以下のスクリプトを実行し、テストデータをOceanBaseデータベースに挿入します。スクリプトのディレクトリは、テストデータのディレクトリと同じである必要があります。

   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]}")
   

ステップ4：OceanBaseデータベースのデータをクエリする

1. 以下の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)
   

2. 質問を入力し、関連する回答を出力します。

   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.