G-gen の神谷です。本記事では、BigQuery の機能を使って、商品を意味&ランキング検索できる ChatBot を作ってみたので、そのご紹介ができればと思います。

アプリの概要

このアプリは Google Chat を使って、ユーザーが入力した商品検索クエリに基づいて関連する商品を見つけ出します。

BigQuery の機能を用いて、単純なベクトル（≒意味）検索ではなく、事前に集計された売れ行きランキングデータを統合することで、ユーザーのニーズに合った人気商品を提案します。

検索の流れは以下の5つのステップで構成されています。

検索: ユーザーがキーワードを入力して商品を検索
テキストをベクトル変換: 入力されたテキストが Text Embeddings API を使ってベクトル表現に変換される
類似ベクトルを検索・抽出: BigQuery を使って、商品の説明文のベクトルデータから、入力テキストのベクトルに類似したものを検索・抽出
検索結果をプロンプトに組み込み、生成 AI に渡す: 抽出された類似商品の情報が、質問応答用のプロンプトに埋め込まれ、Gemini/PaLM2 API に渡される
検索結果表示: 生成 AI が生成した自然言語の回答が、ユーザーに返される

商品検索チャットボット

ユースケース

このアプリケーションは、ファッション用品を扱う小売店での活用を想定しています。店舗のスタッフが接客中に、お客様の要望を聞きながらチャットボットを使って、お客様のニーズに最も合致する商品を検索することができます。

このチャットボットは、自然言語処理技術を用いて、ユーザーの質問の意図を理解し、最も関連性の高い商品情報を提示します。

背景とメリット

近年、リテール業界では、店舗とECサイトを組み合わせたハイブリッド型の販売スタイルが主流となりつつあります。店舗販売の強みは、販売員が直接お客様と対面で接客し、人間の言葉で商品の魅力を伝えられる点にあります。

このような環境下では、販売員一人一人が幅広い商品カテゴリの知識を持ち、お客様の曖昧な要望から適切な商品を見つけ出せることが重要になります。加えて、それらの商品の特徴や売れ行き状況も把握しておく必要があります。

本アプリケーションを活用することで、以下のようなメリットが得られます。

ベテラン販売員しか持ち得なかった高度な商品知識を、新入社員でも手軽に利用できるようになる
組織内の知識共有が効率的かつリアルタイムに行われ、「データの民主化」が促進される
お客様の要望に素早く的確に応えられるようになり、顧客満足度の向上につながる

つまり、生成 AI を使ってこれまで属人化していた知識やノウハウを組織全体で共有・活用することが、本アプリケーションの目的だと言えます。

アーキテクチャ

システムアーキテクチャと、RAG テーブル設計は以下の通りです。

システムアーキテクチャ

ユーザインターフェースとして Google Chat を使っていますが、すでに Google Workspace を導入しているユーザであれば、Google Cloud とのインテグレーションがしやすく、さっと生成 AI アプリケーションを作るのに適しています。

システムアーキテクチャ

このアーキテクチャ図は、Google Chat をユーザインターフェースとし、Cloud Functions で BigQuery へのクエリ発行と検索・レスポンス生成処理を行い、Google Cloud のサービスを活用する構成を示しています。

RAG テーブル設計

RAG テーブル設計には、商品名、ブランド、商品説明、ベクトル表現、類似度（ベクトル距離）、人気ランキングなどのカラムが含まれています。

特に重要なのは、商品説明のテキストデータが Text Embeddings API でベクトル化され、類似度検索に使われている点です。これによって、単なるキーワード検索ではなく、商品特徴を踏まえた意味検索が可能になります。

検索処理の詳細

本アプリケーションの検索処理では、以下の2つのステップを踏んでいます。

意味的に類似する商品を上位10件抽出
抽出した10件の中から売れ行きの良い順に並べ替え

まず、ユーザーが入力した検索クエリのテキストを Text Embeddings API でベクトル化します。次に、そのベクトルと RAG テーブル内の各商品の説明文ベクトルとのコサイン類似度を計算し、類似度が高い上位10件を抽出します。

続いて、抽出した10件の商品について、RAG テーブル内の売れ行きランキングデータを参照し、売れ行きが良い順にソートします。

以上の処理により、検索クエリに意味的に近く、かつ人気のある商品を優先的にユーザーに提示することができます。

使っている技術と実装例

このシステムでは主に以下の技術を使っています。

BigQuery ML のテキストエンべディング関数
BigQuery ML の類似ベクトル検索関数
Cloud Functions による Google Chat アプリケーション構築

これらについて順番に詳しく説明していきます。

BigQuery ML のテキストエンべディング関数

テキストエンべディングとは、テキストデータを意味を踏まえた数値表現（ベクトル）にしたもの、またはそういったベクトルに変換する技術のことを指します。

BigQuery では、以下のようにテキストエンべディングを扱うことができます。

ベクトル列のカラムを ARRAY<FLOAT64> で定義する
ML.GENERATE_TEXT_EMBEDDING 関数でテキストからエンべディングベクトルを抽出する
日本語対応モデルは textembedding-gecko-multilingual

具体的な手順は以下のようになります。

BigQuery リモート関数用のコネクションオブジェクト作成

まず、Cloud Shell 上で以下のコマンドを実行し、BigQuery から Vertex AI API に接続するためのコネクションオブジェクトを作成します。

PROJECT_ID=your-project
REGION=us
CONNECTION_ID=connection-test
    
bq mk --connection --location=$REGION --project_id=$PROJECT_ID \
 --connection_type=CLOUD_RESOURCE $CONNECTION_ID

Vertex AI API を BigQuery のリモート関数として登録

次に、BigQuery 上で以下の SQL を実行し、日本語対応のエンベディングモデル textembedding-gecko-multilingual をリモート関数として登録します。

CREATE MODEL `your-project.sample-dataset.textembedding-gecko-multilingual`
REMOTE WITH CONNECTION `your-project.us.connection-test` 
OPTIONS(ENDPOINT = 'textembedding-gecko-multilingual@001')

テキストデータからエンベディングベクトルの抽出

これで BigQuery 上から ML.GENERATE_TEXT_EMBEDDING 関数を使ってエンベディングベクトルを抽出できるようになります。まず、下記のようなサンプルデータを用意します。

WITH sampleData AS (
  SELECT 'ランニングシューズ ハイパーX、初心者向け、赤、走りやすい' AS target_text, 1 AS sales_rank
  UNION ALL SELECT 'カーボン製テニスラケット、シニア向け、ブラック、ナイキ', 2
  UNION ALL SELECT '防水トレイルランニングバッグ、プーマ、キッズ、ブルー', 3
)
select * from sampleData

-- サンプルデータを定義するCTE（共通テーブル式）
WITH sampleData AS (
 SELECT 'ランニングシューズ ハイパーX、初心者向け、赤、走りやすい' AS target_text, 1 AS sales_rank
 UNION ALL SELECT 'カーボン製テニスラケット、シニア向け、ブラック、ナイキ', 2
 UNION ALL SELECT '防水トレイルランニングバッグ、プーマ、キッズ、ブルー', 3
),
-- テキストエンベディングを生成するCTE
text_embedding_table AS (
 SELECT *
 FROM ML.GENERATE_TEXT_EMBEDDING(
   MODEL `your-project.sample-dataset.textembedding-gecko-multilingual`, -- 使用するテキストエンベディングモデル
   (SELECT target_text AS content FROM sampleData), -- エンベディングを生成するテキストデータ
   STRUCT(TRUE AS flatten_json_output) -- 出力をフラットなJSONとして返す設定
 )  
)
-- メインのSELECT文
SELECT 
 main.*, -- sampleDataテーブルの全列を選択
 text_embedding_table.text_embedding AS target_text_embedding -- 生成されたテキストエンベディングを追加
FROM text_embedding_table -- テキストエンベディングテーブルを結合
INNER JOIN sampleData AS main -- sampleDataテーブルに別名"main"を付けて結合
 ON text_embedding_table.content = main.target_text -- エンベディングを生成したテキストデータで結合

このクエリでは、商品名などのテキストデータを持つソーステーブル sampleData に対して、ML.GENERATE_TEXT_EMBEDDING 関数を適用しエンベディングベクトルを抽出しています。

そして抽出したベクトルを元のソーステーブルと自己結合することで、ソーステーブルの各レコードにベクトル列 target_text_embedding を追加しています。

クエリ結果を product_ranking_emb_table という名前で永続化テーブルに保存します。

参考：ML.GENERATE_TEXT_EMBEDDING 関数を使用してテキストを埋め込む | BigQuery | Google Cloud

BigQuery ML の類似ベクトル検索関数

次に、ユーザから入力された検索クエリに対して、意味的に近い商品を見つけ出す必要があります。それには検索クエリのテキストもエンベディングベクトルに変換し、商品テーブルのエンベディングベクトルとの距離（類似度）を計算します。

入力されたテキストから類似アイテムを検索

BigQueryでは VECTOR_SEARCH 関数を使うことで、ベクトル同士の類似性（距離）を計算し、類似性の高いものから取得することができます。

具体的には以下のように書けます。

-- 検索クエリのテキストからテキストエンベディングを生成するCTE
WITH embedded_text AS (
 SELECT *
 FROM ML.GENERATE_TEXT_EMBEDDING(
   MODEL `your-project.sample-dataset.textembedding-gecko-multilingual`, -- 使用するテキストエンベディングモデル
   (SELECT "マラソンシューズ、女性向け、初心者、ピンク" AS content), -- 検索クエリのテキスト
   STRUCT(TRUE AS flatten_json_output) -- 出力をフラットなJSONとして返す設定
 )  
)
SELECT
 base.target_text, -- 商品のテキスト情報
 base.sales_rank, -- 商品の販売ランキング
 distance -- ベクトル間の距離（類似度）
FROM VECTOR_SEARCH( -- ベクトル検索関数
 TABLE `your-project.sample-dataset.product_ranking_emb_table`, -- 検索対象のテーブル
 'target_text_embedding', -- 検索対象のベクトル列
 (SELECT text_embedding FROM embedded_text), -- 検索クエリのテキストエンベディング
 top_k => 2, -- 上位2件の結果を返す
 distance_type => 'COSINE', -- コサイン類似度を使用
 OPTIONS => '{"use_brute_force":true}' -- ブルートフォース検索を使用
) 
ORDER BY base.sales_rank ASC -- 販売ランキングの昇順で並び替え

ここではまず、ユーザ入力の検索クエリ「マラソンシューズ、女性向け、初心者、ピンク」をエンべディングベクトルに変換しています。

そして商品テーブル product_ranking_emb_table の各レコードのエンベディングベクトル target_text_embedding との距離を計算し、距離が小さい（類似度が高い）上位2件を抽出しています。

最後に、商品の売れ行きランキングなどのカラム sales_rank で並び替えをすることで、検索クエリに意味的に近く、かつ人気のある商品を見つけ出すことができます。

検索高速化のためのベクトルインデックス登録

また、ベクトル検索のパフォーマンスを上げるために、エンべディングベクトル列にインデックスを作成しておくことをおすすめします。BigQuery では以下のように DDL 文でベクトルインデックスが作成できます。

CREATE OR REPLACE VECTOR INDEX test_index
ON your-project.sample-dataset.product_ranking_emb_table(target_text_embedding)
OPTIONS(index_type = "IVF",
  distance_type = "COSINE",
  ivf_options = '{"num_lists":5000}')

上記の処理では、IVF（Inverted File Index）アルゴリズムによって、num_lists で指定した数のグループに、距離が近いベクトル同士を分割します。検索時には、一旦入力したベクトルと最も距離が近いリストを特定し、そのリスト内で更に距離が近いベクトルを検索します。このベクトルインデックス化機能により、総当たり検索に比べて、高速化を実現しています。

注意点として、num_lists は5000以上の値でないと設定できないため、レコード件数が少ないと逆に検索効率が悪くなります。また、対応しているベクトルの次元数が1000以下のため、Vertex AI で提供しているマルチモーダルエンべディング API はデフォルトで1408次元のため利用不可です。こういったケースでは、ベクトルの次元数を落としてインデックスを作ることになるため、検索精度とのトレードオフを考慮することが重要です。

参考：

Cloud Functionsによる Google Chat アプリケーション構築

最後に、Cloud Fuctions と Google Chat を連携させて、チャットボットのインターフェースを作る方法について説明します。

Cloud Functions は、HTTP リクエストをトリガーにしてサーバーレスで処理を実行できる Google Cloud のサービスです。一方で Google Chat は、HTTP リクエストを送ることでボットとの対話が可能です。

したがって、Cloud Functions で HTTP リクエストを受け取る関数を作成し、その関数内で BigQuery にクエリを発行して検索・レスポンス生成処理を行い、レスポンスを Google Chat に返せば、チャットボットが実現できます。詳細な手順は以下のサイトを参照してください。

参考：Cloud Functions を使用して HTTP Google Chat アプリを作成する | Google for Developers

当記事では、main コードのイメージを記載します。

pip install Flask==2.2.5 google-cloud-aiplatform==1.42.1 google-cloud-bigquery==3.12.0 pandas==1.5.3 pandas-gbq==0.19.2 requests==2.31.0 tenacity==8.2.3 gunicorn==21.2.0 "functions-framework>=3.*"

import flask
import functions_framework
from google.cloud import bigquery
from vertexai.preview.generative_models import GenerativeModel
from typing import Any, Mapping
import os
  
# 環境変数からプロジェクトID、ロケーション、モデル名、テキストエンベディングモデル、商品ランキングテーブルを取得
PROJECT_ID = os.environ.get("PROJECT_ID", "your-project")
LOCATION = os.environ.get("LOCATION", "us-central1")
MODEL_NAME = os.environ.get("MODEL_NAME", "gemini-1.0-pro-001")
TEXT_EMBEDDING_MODEL = os.environ.get("TEXT_EMBEDDING_MODEL", "your-project.sample-dataset.textembedding-gecko-multilingual")
PRODUCT_RANKING_TABLE = os.environ.get("PRODUCT_RANKING_TABLE", "your-project.sample-dataset.product_ranking_emb_table")
  
# 生成モデルの設定を定義
generation_config = {
   "temperature": 0.1,
   "top_p": 0.95,
   "top_k": 40,
   "candidate_count": 1,
   "max_output_tokens": 8192,
}
  
# 生成モデルをロード
model = GenerativeModel(MODEL_NAME, generation_config=generation_config)
print(f"Model loaded: {model}")
  
@functions_framework.http
def chat_bot(request: flask.Request) -> Mapping[str, Any]:
   # リクエストからユーザーメッセージを取得
   request_json = request.get_json(silent=True)
   user_message = request_json["message"]["text"]
   print(f"user_message: {user_message}")
  
   if user_message.startswith("/search"):
       # ユーザーメッセージから検索クエリを抽出
       query = user_message.replace("/search", "").strip()
       # 商品を検索
       results = search_products(query)
       # 検索結果からコンテキストを生成
       context = generate_product_context(results)
       # プロンプトを作成
       prompt = f"あなたは商品検索アシスタントです。以下の商品情報を参考に、ユーザーの質問「{query}」に近い商品をオススメしてください。商品の魅力が十分伝わるような文章にしてください。\n\n{context}"
       # 生成モデルを使用して応答を生成
       response_text = model.generate_content(prompt, stream=False).text
   else:
       # 検索以外のメッセージに対する応答
       response_text = (
           "商品検索を行うには、'/search 検索クエリ' のように入力してください。"
       )
  
   print(f"response_text: {response_text}")
   response = {"text": response_text}
   return flask.jsonify(response)
  
def search_products(query):
   # BigQueryクライアントを作成
   client = bigquery.Client(project=PROJECT_ID)
  
   # 商品検索のクエリを定義
   query_job = client.query(
       f"""
       WITH embedded_text AS (
         SELECT *
         FROM ML.GENERATE_TEXT_EMBEDDING(
           MODEL `{TEXT_EMBEDDING_MODEL}`,
           (SELECT "{query}" AS content),
           STRUCT(TRUE AS flatten_json_output)
         )
       )
       SELECT
         base.target_text,
         base.sales_rank,
         distance
       FROM VECTOR_SEARCH(
         TABLE `{PRODUCT_RANKING_TABLE}`,
         'target_text_embedding',
         (SELECT text_embedding FROM embedded_text),
         top_k => 2,
         distance_type => 'COSINE',
         OPTIONS => '{{"use_brute_force":true}}'
       )
       ORDER BY base.sales_rank ASC
       """
   )
  
   print(f"query_job: {query_job}")
   # クエリを実行し、結果をデータフレームに変換
   results = query_job.to_dataframe().to_dict("records")
   print(f"results: {results}")
   return results
  
def generate_product_context(results):
   product_info = []
  
   # 検索結果から商品情報を抽出
   for row in results:
       print(f"row: {row}")
       product_info.append(
           f"商品情報: {row['target_text']}, 販売ランキング: {row['sales_rank']}"
       )
  
   # 商品情報を改行で結合
   context = "\n".join(product_info)
   print(f"context: {context}")
   return context