Let's try working with BigQuery in Gemini (1) - Gemini による BigQuery 操作の基本
Gemini で BigQuery の操作を行うためのクエリを生成し、ユーザからの問い合わせの回答を生成します
Table of contents
author: ryutah
はじめに
こんにちは、技術推進グループグループマネージャのryutahです。
近年の生成AI技術、特に大規模言語モデル (LLM) の急速な発展は、私たちの生活や働き方に大きな変革をもたらしています。ChatGPTをはじめとする様々な生成AIサービスが次々と登場し、私自身も業務効率化のためにこれらのサービスを積極的に活用しています。
しかし、システム開発者としては、既存の生成AIサービスを利用するだけでなく、その技術を応用して新たな価値を生み出すことが求められています。そこで今回は、よくある業務データに対して生成AIを適用し、どのように活用できるかを具体的に探っていきます。
今回は第一歩として、Googleが提供する生成AIであるGeminiを用いて適切にクエリを実行するための手順を紹介します。
TL;DR
検証に利用したサンプルコードとその実行結果を以下に示します。
サンプルコード
ryutah / gemini-bigquery-analysis
実行結果の比較
Function Callingを利用した場合と利用しない場合の実行結果を比較はこちらです。
| 検証 | 結果 |
|---|---|
| Function Calling なし | 申し訳ありませんが、私はデータベースにアクセスしたり、クエリを実行したりすることはできません。そのため、今日の日付を取得したり、今日が何日か教えてあげたりすることはできません。 今日の日付を知りたい場合は、お使いのデバイスの日付と時刻設定を確認してください。 もし、特定のデータベースで今日の日付を取得する方法を知りたい場合は、データベースの種類と使用しているプログラミング言語を教えてください。そうすれば、適切なクエリを提供できます。 |
| Function Calling あり | 今日は7月22日です。 |
生成 AI を用いて BigQuery でクエリを実行させるためのステップ
それでは、生成AIを用いてBigQuery上のデータをクエリするための基本的なステップを見ていきましょう。
生成AIと外部サービスを組み合わせるための仕組みとして、 Function Calling というものがあります。Function Callingは、生成AIと外部サービスを連携させるための機能で、ユーザーの質問に対して適切な関数を呼び出し、その結果を基に回答を生成します。
参考: https://cloud.google.com/vertex-ai/generative-ai/docs/multimodal/function-calling
上記フロー図が示すように、Function Callingを利用することで、ユーザーの質問内容に応じて適切な関数を呼び出し、必要なパラメータを生成できます。これにより、BigQueryのような外部サービスともシームレスに連携し、複雑なデータ処理が可能になります。
今回はこのFunction Callingを活用し、Geminiに適切なSQLクエリを生成させ、BigQueryで実行する流れを構築していきます。
上のシーケンス図は、Function Callingを伴う問い合わせの流れを示しています。Function Callingの部分を詳しく見てみると、以下のステップで処理が行われます。
- 関数呼び出しのパラメータ生成
- Geminiが、ユーザーの質問に基づいて、BigQueryでクエリを実行するための関数 (run_query) とそのパラメータ (SQLクエリ) を生成します。
- クエリ実行のための関数実行
- アプリケーション側で、生成されたパラメータ (run_queryとSQLクエリ) を受け取り、BigQueryでクエリを実行するための関数を呼び出します。
- クエリ実行
- BigQueryでSQLクエリが実行され、結果が取得されます。
- クエリ実行結果を元に回答生成
- 実行結果をGeminiに渡し、最終的な回答を生成します。
ここからは、それぞれのステップについて詳しく見ていきましょう。
1. 関数呼び出しのパラメータ生成
Geminiから関数呼び出しのパラメータを受け取る方法を見ていきましょう。
基本的な Gemini の使い方
基本的なGeminiの問い合わせは、以下のように行います。
model = GenerativeModel("gemini-1.5-flash-001")
response = model.generate_content([
Content(
role="user",
parts=[
Part.from_text("こんにちは"),
],
),
])
しかし、このままだとGeminiは呼び出し可能な関数を認識せず、適切なパラメータを生成できません。そこで、Geminiに関数の定義を伝える必要があります。
関数の定義を Gemini に伝える
関数の定義を伝えるには、モデルの生成時に tools パラメータを設定します。今回は、BigQueryでクエリを実行するための関数run_queryを定義し、Geminiに認識させます。
def run_query(query: str):
"""
Get information from data in BigQuery using SQL queries
Args:
query (str): SQL query on a single line that will help give quantitative answers to the user's question when run on a BigQuery dataset and table. In the SQL query, always use the fully qualified dataset and table names.
"""
pass
この run_query 関数をGeminiに認識させるために、モデル生成時の設定を以下のように変更します。
model = GenerativeModel(
"gemini-1.5-flash-001",
tools=[
Tool(
[
FunctionDeclaration(
name="run_query",
description="Get information from data in BigQuery using SQL queries",
# Parameters は JSON Schema Object format の形式で定義する
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "SQL query on a single line that will help give quantitative answers to the user's question when run on a BigQuery dataset and table. In the SQL query, always use the fully qualified dataset and table names.",
}
},
"required": [
"query",
],
},
),
]
)
],
)
これで、Geminiは run_query 関数を認識し、必要に応じてFunction Callingを行うようになります。
Function Calling のテスト
実際にクエリの実行を伴う問い合わせをしてみましょう。
response = model.generate_content([
Content(
role="user",
parts=[
Part.from_text(
"今日の日付を取得するクエリを実行し、今日が何日か教えて下さい。"
)
],
)
])
# 結果の出力
print(json.dumps(response.to_dict(), indent=2))
上記コードを実行すると、以下のような結果が得られます。
{
"candidates": [
{
"content": {
"role": "model",
"parts": [
// 関数呼び出しのためのパラメータが生成されている
{
"function_call": {
"name": "run_query",
"args": {
"query": "SELECT CURRENT_DATE()"
}
}
}
]
},
// ...
],
// ...
}
Geminiがrun_query関数を呼び出し、引数としてSQLクエリ ( SELECT CURRENT_DATE() ) を生成していることが確認できます。
2. クエリ実行のための関数実行
次に、生成されたrun_query関数呼び出しのパラメータを受け取り、実際にBigQueryでクエリを実行する処理を実装します。
パラメータの抽出と関数の実行
1. 関数呼び出しのパラメータ生成 に記載されているGeminiの生成結果からパラメータを抽出する処理を定義します。
response = model.generate_content([
Content(
role="user",
parts=[
Part.from_text(
"今日の日付を取得するクエリを実行し、今日が何日か教えて下さい。"
)
],
)
])
if len(response.candidates[0].function_calls) > 0:
function_call = response.candidates[0].function_calls[0]
run_query(function_call.args["query"])
このコードでは、response.candidates[0].function_calls から run_query 関数呼び出しのパラメータを抽出し、run_query 関数に渡しています。
注意: ここでは、定義されている関数が
run_queryのみであるため、厳密なチェックは省略しています。複数の関数を定義する場合は、関数名に応じて処理を分岐させる必要があります。if len(response.candidates[0].function_calls) > 0: for fc in response.candidates[0].function_calls: # 関数名によって処理を分岐させる
クエリ実行の確認
run_query 関数を以下のように変更し、実行してみましょう。
def run_query(query: str):
print(f"Query: {query}")
# Output:
# Query: SELECT CURRENT_DATE()
実行結果として、Geminiによって生成されたクエリ ( SELECT CURRENT_DATE() ) が出力されます。
3. クエリ実行
いよいよ、BigQueryで実際にクエリを実行する処理を実装します。
BigQuery でのクエリ実行
def run_query(query: str):
client = bigquery.Client(
project="[PROJECT_ID]", location="[LOCATION]"
)
resp = client.query_and_wait(query)
print(resp.to_dataframe().to_json(orient="records"))
# Output:
# [{ "f0_": 1721260800000 }]
このコードでは、BigQueryクライアントを初期化し、query_and_wait メソッドでクエリを実行しています。実行結果はがわかるように、今回はJSON形式で出力しています。
注意: 本来は、入力値のチェックやエラーハンドリングの考慮が必要になるところになってきますが、今回は与えられクエリを実行するだけのシンプルな処理を定義します。
4. クエリ実行結果を元に回答生成
最後に、クエリの実行結果をGeminiに渡し、最終的な回答を生成する処理を実装します。
run_query 関数の戻り値変更
まずは、run_query関数がクエリの実行結果 (DataFrame) を返すように変更します。
def run_query(query: str) -> pandas.DataFrame:
client = bigquery.Client(
project="[PROJECT_ID]", location="[LOCATION]"
)
resp = client.query_and_wait(query)
return resp.to_dataframe()
回答生成
関数の結果を組み込みながら回答をさせるためには、ユーザからの問い合わせにより生成されたFunction Callingのパラメータと、実際に関数を実行した結果を一連の問い合わせとして設定しつつ、 Geminiに問い合わせる必要があります。
# ユーザからの問い合わせの実行
user_content = Content(
role="user",
parts=[
Part.from_text(
"今日の日付を取得するクエリを実行し、今日が何日か教えて下さい。"
)
],
)
response = model.generate_content(
[user_content],
generation_config=GenerationConfig(
temperature=0,
),
)
# Function Calling の結果をもとに、関数を実行
if len(response.candidates[0].function_calls) > 0:
function_call = response.candidates[0].function_calls[0]
fc_responses = []
# 関数の実行
if function_call.name == "run_query":
query = str(function_call.args["query"])
results = run_query(query)
fc_responses = results.to_dict(orient="records")
# ユーザからの問い合わせと Function Calling の結果をもとに回答を生成
response_with_fc = model.generate_content(
[
user_content,
response.candidates[0].content,
Content(
parts=[
Part.from_function_response(
name="run_query",
response={
"contents": fc_responses,
},
),
],
),
],
)
print(final_response.candidates[0].content.parts[0].text)
型変換エラーの対処
上記コードを実行するだけでは、以下のようなエラーが発生してしまいます。
ValueError: Unable to coerce value: datetime.date(2024, 7, 22)
Geminiにリクエストを送信する際に、パラメータをシリアライズしているのですが、その際にエラーが発生しています。
シリアライズは、以下の処理で行われますので、こちらに併せて型変換します。
def to_proto(self, value) -> struct_pb2.Value:
"""Return a protobuf Value object representing this value."""
if isinstance(value, struct_pb2.Value):
return value
if value is None:
return struct_pb2.Value(null_value=0)
if isinstance(value, bool):
return struct_pb2.Value(bool_value=value)
if isinstance(value, (int, float)):
return struct_pb2.Value(number_value=float(value))
if isinstance(value, str):
return struct_pb2.Value(string_value=value)
if isinstance(value, collections.abc.Sequence):
return struct_pb2.Value(
list_value=self._marshal.to_proto(struct_pb2.ListValue, value),
)
if isinstance(value, collections.abc.Mapping):
return struct_pb2.Value(
struct_value=self._marshal.to_proto(struct_pb2.Struct, value),
)
raise ValueError("Unable to coerce value: %r" % value)
from: googleapis/proto-plus-python - proto/marshal/rules/struct.py
if function_call.name == "run_query":
query = str(function_call.args["query"])
results = run_query(query)
fc_responses = results.to_dict(orient="records")
for col in results.columns:
if results[col].dtype == "dbdate":
# ISO 8601 形式に変換
results[col] = results[col].map(
lambda x: x.strftime("%Y-%m-%dT%H:%M:%S%Z")
)
上記のようにすることで、実行結果は以下のようになります。
今日は7月22日です。
単純な結果ですが、 Function Callingを行わずに同様の問い合わせをした場合の結果は以下のようになるため、関数の実行結果を正しく利用出ていることがわかります。
申し訳ありませんが、私はデータベースにアクセスしたり、クエリを実行したりすることはできません。そのため、今日の日付を取得したり、今日が何日か教えてあげたりすることはできません。
今日の日付を知りたい場合は、お使いのデバイスの日付と時刻設定を確認してください。
もし、特定のデータベースで今日の日付を取得する方法を知りたい場合は、データベースの種類と使用しているプログラミング言語を教えてください。そうすれば、適切なクエリを提供できます。
まとめ
今回は、Geminiでデータ分析をするための前段階として、BigQueryでクエリを実行し、その結果をGeminiに返すことでユーザーの質問に回答する方法を紹介しました。
次回は、BigQueryに保存されている具体的なデータに対して、Geminiを活用した実践的な分析例を紹介していきます。より複雑なクエリの実行や、分析結果の可視化など、実務で役立つテクニックを解説する予定です。どうぞお楽しみに！
参考サイト
※本記事は、ジーアイクラウド株式会社の見解を述べたものであり、必要な調査・検討は行っているものの必ずしもその正確性や真実性を保証するものではありません。
※リンクを利用する際には、必ず出典がGIC dryaki-blogであることを明記してください。
リンクの利用によりトラブルが発生した場合、リンクを設置した方ご自身の責任で対応してください。
ジーアイクラウド株式会社はユーザーによるリンクの利用につき、如何なる責任を負うものではありません。