ChatGPTのチャット画面から日本語で指示してBigQueryのデータを更新しながら分析してみた話

こんにちは。那須野です。

ChatGPTは順調に進化してきて、いろいろなデータ分析に使えるようになってきていますよね。

前回はBigQueryのデータにアクセスしてデータ分析、つまりreadをさせてみましたが、今回はデータの更新、つまりwriteもさせてみようと思います。

readを比べるとwriteはリスクが高めで、思わぬテーブルやレコードをdeleteされたりupdateされたりということが怖いので、権限を細かめに設定することの重要性が増してきます。

そのあたりを頭に入れておきつつ、実際に試してみましょう。

※この記事は、前回の内容を取り組んだ方向けの内容になっています。アーキテクチャや手順など詳細は前回の方に書いてあるので、まだ見ていない方は先に読んで適宜取り組んでから以下をお読みください。

ChatGPTによるBigQueryデータ更新とは？

ChatGPTによるBigQueryデータ更新とは、ChatGPT画面で自然言語（日本語）でチャットで指示して、ChatGPTにBigQueryのデータを更新させていくことを指します。例えば以下のようなイメージです。

「売上テーブルを以下のロジックで集計して、会員テーブルのrankを更新して。」

「BIツールで参照しやすいようにGA4、サーチコンソール、Google広告の実績データを日次で集計してdigital_marketing_daily_indexテーブルとして保存して。」

「都道府県ごとの今日の天気予報を収集してweather_forecastテーブルにinsertして。」

いろいろできそうですよね。

ChatGPTによるBigQueryデータ更新のメリット

BigQueryデータ更新で何よりも重要なのは、プロセスを保存→検証して品質保証できるようになる点でしょう。

というのも、ChatGPTの前回の分析の欠点は、分析した結果がChatGPTの個人チャット上にしかない点です。結果として、裏でどういう分析をしたのか分かりづらく、スレッドが流れてしまうと本人も思い出せなくなります。加えて、分析プロセスを第三者チェックするのが極めて大変です。つまるところ、ビジネスにおいて死活問題となる「組織的なデータ分析運用」に対してサポートがほとんどなされないという点です。

この点について、BigQueryを更新できるようにすることで様々な改善が見えてきます。例を挙げてみましょう。

• モニタリング - いつ、だれが、どんな目的で、どんなSQLを実行したか、そしてその結果としてどんなデータがSELECTされたかをBigQueryに記録し、管理側で参照できるようにする。
• レビュー運用 - ChatGPTの分析をBigQueryに保存し、他のAI（ChatGPTの別人格でもよい）にレビューさせることが容易になる。

それ以外にも、アクション経由では授受できないような大きなアウトプットをBigQueryに保存することで、多段階の分析をより安全・確実に実行できる点も見逃せないですね。

ChatGPTによるBigQueryデータ更新の注意点

しかし、生成AIは人間と同じようにミスをしうるものです。BigQueryの閲覧だけであればリスクは少ないですが、編集の権限を与えた場合はトンデモナイことをやらかす可能性を否定できなくなります。例えばAIがうっかりデータセットごとDROPするような展開…を考慮しないといけないのです。

そのため、駆け出しエンジニアが初めての手作業をするときにベテランエンジニアが目を凝らして見守っていたのと同じように、ChatGPTを見守る仕組みが必要です。具体的には、

• うっかりミスを未然に防げる仕組み
• 何か起きたときに状況を把握できる仕組み
• 素早い応急処置で運用への影響を最小限にする仕組み
• その後に可及的速やかに恒久対応ができる仕組み

の4点を抑えておく必要があります。いくつか方法はあるのですが、今回は簡易的かつ効果的な方法としてサービスアカウントを3つ立てる方法を取ってみようと思います。エンドポイント実行用のrunner、BigQuery閲覧用のreader、BigQuery更新用のwriterで、それぞれに適切な権限を付与する形です。

これにより、閲覧系と更新系がサービスアカウントレベルで分離するため、それぞれの系統に対してBigQuery権限を細かに付与すること・・・例えば「このテーブルは閲覧のみ、このテーブルは更新もできる」などの設定が容易になり、実行履歴で明快に区別できるようになり、またコードに触らずとも編集系だけを即座に切るなどの対応も可能になります。

また、何かのはずみで編集権限が変なところについて変に更新できてしまうと困るので、プロンプトレベルで制御するだけでなく、サービスアカウントのデータ編集権限による制御も加えて二重のロックをかけることで、うっかりミスを極力減らすことができるはずです。

なお、ここまでやってもモノによっては復旧が困難になる可能性があるため、やはりBigQueryのタイムトラベル機能の有効化したうえで復元訓練を積んでおくまでは必須でしょう。

それでは次は、設定手順を具体的に見ていきます。

ChatGPTによるBigQueryデータ更新の設定手順

API有効化や作業ユーザーのロール設定は前回と同様なので割愛します。

今回はReadもWriteもできるということで、rwをキーワードとして足した名前で作っていこうと思います。それでは具体的な手順を見ていきましょう。

①認証情報の作成

1. Cloud Run経由でのアクセスに最低限のセキュリティ対策を施すため、APIとサービスの認証情報を開き、画面上部の「認証情報を作成」ボタンを押し、「APIキー」を選びます。
2. 名前＝「chatgpt-bg-rw-connector-api-key」などのように付けつつ、アプリケーションの制限＝「ウェブサイト」にしてウェブサイトに「https://chatgpt.com/」を追加し、APIの制限＝「キーを制限」を選んでAPIのプルダウンから「Cloud Run Admin API」を選んで「保存」ボタンを押し、APIキーをコピペします。
3. Secret Managerを開き、「シークレットを作成」ボタンを押し、名前＝「chatgpt-bg-rw-connector-secret」などのように付けつつ、APIキーをシークレットの値として入力し、最後に一番下の「シークレットを作成」ボタンを押します。

②サービスアカウントの作成

サービスアカウントの画面を開き、サービスアカウントを3つ作成してそれぞれロールを付与します。また、Read用、Write用についてはBigQueryスタジオ画面から対象となるデータセットの権限を付与します。

③Cloud Run関数の作成

1. Cloud Runを開いて画面上の「関数を作成」をクリックし、例えばサービス名＝chatgpt-bq-rw-connector、リージョン＝asia-northeast1、ランタイム＝Python 3.13に設定します。
2. 続いて認証については「公開アクセスを許可する」のまま画面一番下の「作成」ボタンを押します。
3. エディタ画面が開かれたら、main.pyとrequirements.txtに以下2つのコードを貼り付け、関数のエントリポイントをを「run_bq」に書き換えます。前回と比べると、以下4点が主な更新点です。
   ①google.authライブラリを追加してimpersonateを実装
   ②サービスアカウントを環境変数から取得
   ③request bodyにmode (read/write) を追加（ない場合はreadを強制）
   ④jobを投げた後は、read/writeそれぞれで分岐処理
   
   main.py

   import json import os import traceback import functions_framework from google.cloud import bigquery from google.auth import default, impersonated_credentials API_KEY = os.getenv("EXPECTED_API_KEY") READER_SA = os.getenv("READER_SA") WRITER_SA = os.getenv("WRITER_SA") BQ_LOCATION = os.getenv("BQ_LOCATION") def _impersonated_client(target_principal: str) -> bigquery.Client: if not target_principal: raise RuntimeError("Service Account for impersonation is not set.") source_creds, _ = default(scopes=["https://www.googleapis.com/auth/cloud-platform"]) creds = impersonated_credentials.Credentials( source_credentials=source_creds, target_principal=target_principal, target_scopes=["https://www.googleapis.com/auth/bigquery"], lifetime=300, ) return bigquery.Client(credentials=creds, location=BQ_LOCATION) def get_bq_client(mode: str) -> bigquery.Client: m = (mode or "read").lower() if m == "write": return _impersonated_client(WRITER_SA) else: return _impersonated_client(READER_SA) @functions_framework.http def run_bq(request): if request.headers.get("x-api-key") != API_KEY: return ("Unauthorized", 401) try: body = request.get_json(silent=True) or {} mode = (body.get("mode") or "read").lower() # "read" or "write" sql = body.get("query") if not sql: return (json.dumps({"error": "'query' is required"}), 400, {"Content-Type": "application/json"}) client = get_bq_client(mode) job = client.query(sql) if mode == "read": try: import pandas as pd df = job.result().to_dataframe() df = df.where(df.notna(), None) # イレギュラーなNaNだけNone(null)に補正 df_json = df.to_json(orient="records", force_ascii=False) df_json_len = len(df_json) if df_json_len > MAX_JSON_CHARACTERS: payload = { "error": "response_records_exceeds_size_limit", "message": "responseのレコードサイズがChatGPTの受け取り上限を超えました。列を減らす、行を絞り込む、またはLIMITを設定してください。", "stats": { "rows": int(df.shape[0]), "cols": int(df.shape[1]), "characters": df_json_len, "max_characters": MAX_JSON_CHARACTERS } } return (json.dumps(payload, ensure_ascii=False), 413, {"Content-Type": "application/json"}) return (df_json, 200, {"Content-Type": "application/json"}) except Exception: rows = [dict(row) for row in job.result()] return (json.dumps(rows, ensure_ascii=False), 200, {"Content-Type":"application/json"}) else: # DML 実行結果を返す return ( json.dumps({ "job_id": job.job_id, "statement_type": getattr(job, "statement_type", None), "num_dml_affected_rows": job.num_dml_affected_rows, }), 200, {"Content-Type": "application/json"}, ) except Exception as e: # Cloud Loggingで確認できるようトレースバックを出力 print("Error during BigQuery execution:", str(e)) traceback.print_exc() return (json.dumps({"error": str(e)}), 500, {"Content-Type": "application/json"})

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90

   import json import os import traceback import functions_framework from google.cloud import bigquery from google.auth import default, impersonated_credentials   API_KEY = os.getenv("EXPECTED_API_KEY") READER_SA = os.getenv("READER_SA") WRITER_SA = os.getenv("WRITER_SA") BQ_LOCATION = os.getenv("BQ_LOCATION")   def _impersonated_client(target_principal: str) -> bigquery.Client:     if not target_principal:         raise RuntimeError("Service Account for impersonation is not set.")     source_creds, _ = default(scopes=["https://www.googleapis.com/auth/cloud-platform"])     creds = impersonated_credentials.Credentials(         source_credentials=source_creds,         target_principal=target_principal,         target_scopes=["https://www.googleapis.com/auth/bigquery"],         lifetime=300,     )     return bigquery.Client(credentials=creds, location=BQ_LOCATION)   def get_bq_client(mode: str) -> bigquery.Client:     m = (mode or "read").lower()     if m == "write":         return _impersonated_client(WRITER_SA)     else:         return _impersonated_client(READER_SA)   @functions_framework.http def run_bq(request):     if request.headers.get("x-api-key") != API_KEY:         return ("Unauthorized", 401)       try:         body = request.get_json(silent=True) or {}         mode = (body.get("mode") or "read").lower()  # "read" or "write"         sql = body.get("query")           if not sql:             return (json.dumps({"error": "'query' is required"}), 400, {"Content-Type": "application/json"})           client = get_bq_client(mode)         job = client.query(sql)           if mode == "read":             try:                 import pandas as pd                 df = job.result().to_dataframe()                 df = df.where(df.notna(), None) # イレギュラーなNaNだけNone(null)に補正                 df_json = df.to_json(orient="records", force_ascii=False)                 df_json_len = len(df_json)                                  if df_json_len > MAX_JSON_CHARACTERS:                     payload = {                         "error": "response_records_exceeds_size_limit",                         "message": "responseのレコードサイズがChatGPTの受け取り上限を超えました。列を減らす、行を絞り込む、またはLIMITを設定してください。",                         "stats": {                             "rows": int(df.shape[0]),                             "cols": int(df.shape[1]),                             "characters": df_json_len,                             "max_characters": MAX_JSON_CHARACTERS                         }                     }                     return (json.dumps(payload, ensure_ascii=False), 413, {"Content-Type": "application/json"})                                  return (df_json, 200, {"Content-Type": "application/json"})               except Exception:                 rows = [dict(row) for row in job.result()]                 return (json.dumps(rows, ensure_ascii=False), 200, {"Content-Type":"application/json"})         else:             # DML 実行結果を返す             return (                 json.dumps({                     "job_id": job.job_id,                     "statement_type": getattr(job, "statement_type", None),                     "num_dml_affected_rows": job.num_dml_affected_rows,                 }),                 200,                 {"Content-Type": "application/json"},             )       except Exception as e:         # Cloud Loggingで確認できるようトレースバックを出力         print("Error during BigQuery execution:", str(e))         traceback.print_exc()         return (json.dumps({"error": str(e)}), 500, {"Content-Type": "application/json"})

   
   
   requirements.txt

   google-cloud-bigquery>=3.20.0 google-auth>=2.20.0 pandas>=2.2.0 db-dtypes>=1.0.0

   1 2 3 4

   google-cloud-bigquery>=3.20.0 google-auth>=2.20.0 pandas>=2.2.0 db-dtypes>=1.0.0

4. 画面上にある「新しいリビジョンの編集とデプロイ」を押し、デプロイ画面を表示します。
5. 表示されたコンテナタブにて、まずはコンテナイメージURLを確認し、デプロイするコンテナが表示されていることを確認しましょう。（※もしここでデモコンテナが表示される場合、何かしらの作業ミスをしている可能性があります。そのまま作業を進めてもエラーが出かねないため、Cloud Runの関数を一度削除して再作成をした方が良い可能性があります。）
6. コンテナの設定で、メモリ＝1GiB、CPU＝1、リクエストのタイムアウト＝60秒、インスタンスあたりの同時リクエスト数＝50にしつつ、リビジョンスケーリングにてインスタンスの最小数＝0、最大数＝5に設定します。（分析するユーザー数やデータの重さによって微調整する）
7. 変数とシークレットタブにて、「環境変数」に「READER_SA」＝chatgpt-bg-reader@以下略、「WRITER_SA」＝chatgpt-bg-writer@以下略、「BQ_LOCATION」＝asia-northeast1を入力し、「環境変数として公開されるシークレット」として名前＝「EXPECTED_API_KEY」としつつシークレットには冒頭で作成したシークレットを宛がい、バージョン＝「latest」にします。
8. セキュリティタブにて、サービスアカウントにchatgpt-bq-runner@以下略を設定します。
9. 画面一番下の「デプロイ」ボタンを押すとデプロイが完了。改めてURL（サービスURL）が表示されているので、コピーしておきます。

④マニフェストファイルの公開

1. 以下のようなマニフェストファイルを作成します。なお、servers[0].urlはさきほどコピーしたサービスURLに置き換えておきます。
   
   action_manifest.json

   { "openapi": "3.1.0", "info": { "title": "BigQuery Runner", "version": "1.2.0" }, "servers": [ { "url": "https://chatgpt-bq-rw-connector-<hash>.asia-northeast1.run.app" } ], "paths": { "/read": { "post": { "operationId": "bq_read", "summary": "Run a read-only SQL query (SELECT, etc.)", "x-openai-isConsequential": false, "requestBody": { "required": true, "content": { "application/json": { "schema": { "type": "object", "properties": { "mode": { "type": "string", "enum": ["read"], "description": "必ず read を指定" }, "query": { "type": "string", "description": "Standard SQL (SELECT ...)" } }, "required": ["mode", "query"] } } } }, "responses": { "200": { "description": "Rows as JSON", "content": { "application/json": { "schema": { "type": "array", "items": { "type": "object" } } } } }, "400": { "description": "Bad request (e.g., missing 'query')", "content": { "application/json": { "schema": { "type": "object", "properties": { "error": { "type": "string" } } } } } }, "401": { "description": "Unauthorized (invalid or missing API key)" }, "413": { "description": "Payload too large (response exceeds MAX_JSON_CHARACTERS)", "content": { "application/json": { "schema": { "type": "object", "properties": { "error": { "type": "string" }, "message": { "type": "string" }, "stats": { "type": "object", "properties": { "rows": { "type": "integer" }, "cols": { "type": "integer" }, "characters": { "type": "integer" }, "max_characters": { "type": "integer" } } } }, "required": ["error", "message", "stats"] } } } }, "500": { "description": "Internal server error during BigQuery execution", "content": { "application/json": { "schema": { "type": "object", "properties": { "error": { "type": "string" } } } } } } } } }, "/write": { "post": { "operationId": "bq_write", "summary": "Run a write (DML/DDL) statement", "x-openai-isConsequential": false, "requestBody": { "required": true, "content": { "application/json": { "schema": { "type": "object", "properties": { "mode": { "type": "string", "enum": ["write"], "description": "必ず write を指定" }, "query": { "type": "string", "description": "DML/DDL statement" } }, "required": ["mode", "query"] } } } }, "responses": { "200": { "description": "Job metadata", "content": { "application/json": { "schema": { "type": "object", "properties": { "job_id": { "type": "string" }, "statement_type": { "type": ["string", "null"] }, "num_dml_affected_rows": { "type": ["integer", "null"] } } } } } }, "400": { "description": "Bad request (e.g., missing 'query')", "content": { "application/json": { "schema": { "type": "object", "properties": { "error": { "type": "string" } } } } } }, "401": { "description": "Unauthorized (invalid or missing API key)" }, "500": { "description": "Internal server error during BigQuery execution", "content": { "application/json": { "schema": { "type": "object", "properties": { "error": { "type": "string" } } } } } } } } } } }

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165

   {   "openapi": "3.1.0",   "info": { "title": "BigQuery Runner", "version": "1.2.0" },   "servers": [     { "url": "https://chatgpt-bq-rw-connector-<hash>.asia-northeast1.run.app" }   ],   "paths": {     "/read": {       "post": {         "operationId": "bq_read",         "summary": "Run a read-only SQL query (SELECT, etc.)",         "x-openai-isConsequential": false,         "requestBody": {           "required": true,           "content": {             "application/json": {               "schema": {                 "type": "object",                 "properties": {                   "mode": { "type": "string", "enum": ["read"], "description": "必ず read を指定" },                   "query": { "type": "string", "description": "Standard SQL (SELECT ...)" }                 },                 "required": ["mode", "query"]               }             }           }         },         "responses": {           "200": {             "description": "Rows as JSON",             "content": {               "application/json": {                 "schema": {                   "type": "array",                   "items": { "type": "object" }                 }               }             }           },           "400": {             "description": "Bad request (e.g., missing 'query')",             "content": {               "application/json": {                 "schema": {                   "type": "object",                   "properties": {                     "error": { "type": "string" }                   }                 }               }             }           },           "401": {             "description": "Unauthorized (invalid or missing API key)"           },           "413": {             "description": "Payload too large (response exceeds MAX_JSON_CHARACTERS)",             "content": {               "application/json": {                 "schema": {                   "type": "object",                   "properties": {                     "error": { "type": "string" },                     "message": { "type": "string" },                     "stats": {                       "type": "object",                       "properties": {                         "rows": { "type": "integer" },                         "cols": { "type": "integer" },                         "characters": { "type": "integer" },                         "max_characters": { "type": "integer" }                       }                     }                   },                   "required": ["error", "message", "stats"]                 }               }             }           },           "500": {             "description": "Internal server error during BigQuery execution",             "content": {               "application/json": {                 "schema": {                   "type": "object",                   "properties": {                     "error": { "type": "string" }                   }                 }               }             }           }         }       }     },     "/write": {       "post": {         "operationId": "bq_write",         "summary": "Run a write (DML/DDL) statement",         "x-openai-isConsequential": false,         "requestBody": {           "required": true,           "content": {             "application/json": {               "schema": {                 "type": "object",                 "properties": {                   "mode": { "type": "string", "enum": ["write"], "description": "必ず write を指定" },                   "query": { "type": "string", "description": "DML/DDL statement" }                 },                 "required": ["mode", "query"]               }             }           }         },         "responses": {           "200": {             "description": "Job metadata",             "content": {               "application/json": {                 "schema": {                   "type": "object",                   "properties": {                     "job_id": { "type": "string" },                     "statement_type": { "type": ["string", "null"] },                     "num_dml_affected_rows": { "type": ["integer", "null"] }                   }                 }               }             }           },           "400": {             "description": "Bad request (e.g., missing 'query')",             "content": {               "application/json": {                 "schema": {                   "type": "object",                   "properties": {                     "error": { "type": "string" }                   }                 }               }             }           },           "401": {             "description": "Unauthorized (invalid or missing API key)"           },           "500": {             "description": "Internal server error during BigQuery execution",             "content": {               "application/json": {                 "schema": {                   "type": "object",                   "properties": {                     "error": { "type": "string" }                   }                 }               }             }           }         }       }     }   } }

2. Cloud Storageの公開バケットに上述のマニフェストファイルをアップロードし、ファイルの公開URLを取得します。
   • なお、適切な公開バケットがない場合は、バケットを東京リージョン（asia-northeast1）で「chatgpt-bq-connector-manifest」のような名前で作成し、権限タブにて「公開アクセス防止を削除」をクリックして「確認」を押し、画面下にある権限にて「Grant Access」ボタンから新しいプリンシパル＝「allUsers」、ロール＝「Storage オブジェクト閲覧者」を設定し、一番下の「保存」ボタンを押します。「このリソースを一般公開してもよろしいですか？」と聞かれるので、そのまま「一般公開アクセスを許可」を押して公開します。

⑤カスタムGPTの作成

1. ChatGPTにアクセスし、画面左タブから「GPT」を選び、画面右上の「作成する」ボタンを押します。
2. 画面上の「構成」タブを押し、「名前」と「説明」と「指示」を適当に入力後、分析力を強化するために機能の「コードインタープリターとデータ分析」はONにしておきましょう。また、検索ができると変に外部に情報を求めてしまうので、機能で「ウェブ検索」はOFFにすることも検討しましょう。
3. 最後にアクションにある「新しいアクションを作成する」をクリックし、認証は認証タイプ＝「APIキー」を選びつつ取得していたAPIキーを転記し、2個目の認証タイプ＝「カスタム」を選んでカスタムヘッダーの名前＝「x-api-key」と入力して「保存する」ボタンを押します。
4. 続いてスキーマにて「URLからインポートする」ボタンを押して、先ほどコピーしていたマニフェストファイルの公開URLを入力して「インポートする」ボタンを押し、OpenAPIスキーマが正しく反映されていることを確認したうえで画面右上の「作成する」ボタンを押し、GPTを共有する先を「自分だけ」ないし「リンクを受け取った人」に変更して「保存する」ボタンを押したら完了です。
   • 分析の精度を高めるうえでは、カスタムGPTの指示において、アクセスできるデータセットの指定、データセットのスキーマとサンプルデータの提示、分析の際の注意事項などを記載するとよいです。特にデータ更新については、いつ、どういう目的で、どういう更新を行ってよいのかの制御を指示することが重要です。

⑥チャットでデータ分析の指示

1. ChatGPT画面で作成したGPTを選択し、好きに分析指示を出します。
2. 最初は「BigQueryデータ分析 が chatgpt-bq-rw-connector-{hash}.{region}.run.app に通信しようとしています」と確認が出るので「常に許可する」ボタンを押すと以降は確認がでなくなったうえで実行されて、アクセスできる範囲でいい感じにデータを見て集計して結果を返してくれます。

これら手順を踏むことで、readだけでなくwriteもできるカスタムGPTができあがり、プロンプトによる指示をしてより柔軟な分析をさせることができるでしょう。

BigQuery更新分析の具体的なプロンプトイメージ

これだけで結構実用的な分析が可能になる土台ができますが、実際のところは指示次第、プロンプト次第となります。

例えば組織的にログを取得していく上では、カスタムGPTレイヤーの指示によってログ記録を義務付けてもいいですし、場合によってはCloud Runの関数レイヤーでログ記録を実装することも可能です。

また、BigQuery上においては分析トランザクションの管理がより重要になっていくでしょう。例えばChatGPTの分析指示がいつ誰からなされ、分析Noとして何番が宛がわれ、どのユーザー指示を受けてどんなSQLをAIが書いたのかがBigQueryに保存され、それを別のエージェントが読み取って分析意図を読み解いてレビュー結果としてBigQueryに登録し、そのレビュー結果をまた別のエージェントがユーザー指示と照合して評価をし、及第点であればSQLとして実行されて集計結果が返ってきて、集計結果の解釈がBigQueryに保存されて、といったサイクルをユーザーに意識させずに回そうとしたとき、カスタムGPTやCloud Run関数、はたまたそれ以外のリソースをどう設計すべきなのか…

とりあえず全てをカスタムGPTレイヤーでどうにかしようとしたら、例えば以下のような指示が考えられます。書き方の工夫としては、以下を気を付けるとよいでしょう。

• 役割分担を定義し、レビューを義務付ける。
• 行程を明確に定義し、忘れず実行してほしい作業は行程として独立させる。またアウトプットを定義することで実行し忘れを極力減らさせる。（履歴保存など）
• データセットをプロンプトに定義することで、都度のBigQuery問い合わせを減らし、処理を高速化させる。（プロンプトに定義せずに都度調べさせるようプロンプトに定義することも可）

なかなか長いですが、これぐらい書けば割と汎用的に機能しそうですね。

まとめ

ということで、BigQueryを更新しながらのデータ分析についてまとめてみました。今回の制御はほぼ全てをフロントエンド（ChatGPTのプロンプト側）に寄せた実装なので、作るのは比較的簡単でしたね。

一方で、確実性やセキュリティを高めるには制御をバックエンド側（Cloud RunのPython側）に寄せる実装というのも選択肢に入ってくるでしょう。

どちらにしろ、こういうのができるようになってくるとワークフローの言語化スキルがより重要になってくるように感じますね。

以上、本日の備忘録でした。いつかどこかの誰かの参考になれば幸いです。