Gemini API × BigQuery で実現する AI データ分析 — 自然言語クエリからインサイト抽出まで

pip install google-cloud-bigquery google-genai db-dtypes pandas

import os
 
# Google AI Studio の API キーを使用する場合
os.environ["GOOGLE_API_KEY"] = "your-api-key-here"
 
# Vertex AI を使用する場合は、サービスアカウント認証を設定
# gcloud auth application-default login

from google.cloud import bigquery
 
client = bigquery.Client()
 
def get_table_schema(dataset_id: str, table_id: str) -> str:
    """テーブルスキーマを人間が読める形式で返す"""
    table_ref = f"{client.project}.{dataset_id}.{table_id}"
    table = client.get_table(table_ref)
 
    schema_lines = [f"テーブル: {table_ref}"]
    schema_lines.append(f"行数: 約{table.num_rows:,}行")
    schema_lines.append(f"カラム:")
 
    for field in table.schema:
        nullable = "NULL許可" if field.mode == "NULLABLE" else "必須"
        schema_lines.append(
            f"  - {field.name} ({field.field_type}, {nullable}): {field.description or '説明なし'}"
        )
 
    return "\n".join(schema_lines)
 
# 使用例
schema_info = get_table_schema(
    "google_analytics_sample",
    "ga_sessions_20170801"
)
print(schema_info)

from google import genai
 
# Gemini クライアントの初期化
genai_client = genai.Client()
 
def generate_sql(question: str, schema: str) -> str:
    """自然言語の質問から BigQuery SQL を生成する"""
 
    system_prompt = f"""あなたは BigQuery SQL のエキスパートです。
以下のテーブルスキーマに基づいて、ユーザーの質問に答える SQL クエリを生成してください。
 
{schema}
 
ルール:
- 有効な BigQuery Standard SQL のみを出力すること
- SQL クエリのみを返し、説明文は含めないこと
- LIMIT 句を適切に使い、大量のデータを返さないこと
- テーブル名にはプロジェクト名を含めること
- コメントは含めないこと
"""
 
    response = genai_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=question,
        config=genai.types.GenerateContentConfig(
            system_instruction=system_prompt,
            temperature=0.1,  # SQL生成は低温で正確性を重視
        ),
    )
 
    # コードブロックのマークダウンを除去
    sql = response.text.strip()
    sql = sql.replace("```sql", "").replace("```", "").strip()
 
    return sql
 
# 使用例
question = "チャネル別のセッション数トップ10を教えてください"
sql = generate_sql(question, schema_info)
print(sql)

SELECT
  channelGrouping,
  COUNT(*) AS session_count
FROM
  `bigquery-public-data.google_analytics_sample.ga_sessions_20170801`
GROUP BY
  channelGrouping
ORDER BY
  session_count DESC
LIMIT 10

from google.cloud.bigquery import QueryJobConfig
 
def validate_sql(sql: str) -> dict:
    """SQL をドライランで検証し、処理バイト数を返す"""
    job_config = QueryJobConfig(dry_run=True, use_query_cache=False)
 
    try:
        query_job = client.query(sql, job_config=job_config)
        bytes_processed = query_job.total_bytes_processed
        return {
            "valid": True,
            "estimated_bytes": bytes_processed,
            "estimated_gb": round(bytes_processed / (1024**3), 4),
        }
    except Exception as e:
        return {"valid": False, "error": str(e)}
 
# 検証実行
result = validate_sql(sql)
print(result)
# 出力: {'valid': True, 'estimated_bytes': 148567232, 'estimated_gb': 0.1384}

import pandas as pd
 
def execute_and_summarize(sql: str, question: str) -> str:
    """SQL を実行し、結果を Gemini で要約する"""
 
    # SQL 実行
    df = client.query(sql).to_dataframe()
 
    # データフレームを文字列に変換
    result_text = df.to_string(index=False, max_rows=50)
 
    # Gemini による要約
    summary_prompt = f"""以下のデータ分析結果を、ビジネスパーソンにも理解できる形で要約してください。
 
## 元の質問
{question}
 
## 実行した SQL
{sql}
 
## 結果データ
{result_text}
 
以下の観点で要約してください：
1. 主要な発見（最も重要なデータポイント）
2. 注目すべきトレンドやパターン
3. 次のアクションとして考えられること
"""
 
    response = genai_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=summary_prompt,
    )
 
    return response.text
 
# 実行例
summary = execute_and_summarize(sql, question)
print(summary)

class GeminiBigQueryAgent:
    """自然言語でBigQueryを操作する分析エージェント"""
 
    def __init__(self, dataset_id: str, table_ids: list[str]):
        self.bq_client = bigquery.Client()
        self.genai_client = genai.Client()
        self.schemas = self._load_schemas(dataset_id, table_ids)
        self.history = []
 
    def _load_schemas(self, dataset_id, table_ids) -> str:
        """複数テーブルのスキーマを一括取得"""
        all_schemas = []
        for tid in table_ids:
            schema = get_table_schema(dataset_id, tid)
            all_schemas.append(schema)
        return "\n\n".join(all_schemas)
 
    def ask(self, question: str) -> dict:
        """質問を受け取り、分析結果を返す"""
        # 1. SQL 生成
        sql = generate_sql(question, self.schemas)
 
        # 2. 検証
        validation = validate_sql(sql)
        if not validation["valid"]:
            return {
                "error": f"SQL検証エラー: {validation['error']}",
                "sql": sql,
            }
 
        # 3. コスト上限チェック（1GB以上は確認を求める）
        if validation["estimated_gb"] > 1.0:
            return {
                "warning": f"推定処理量が {validation['estimated_gb']}GB です。実行しますか？",
                "sql": sql,
            }
 
        # 4. 実行と要約
        summary = execute_and_summarize(sql, question)
 
        # 5. 履歴に追加
        self.history.append({
            "question": question,
            "sql": sql,
            "bytes": validation["estimated_bytes"],
        })
 
        return {
            "question": question,
            "sql": sql,
            "summary": summary,
            "estimated_cost_gb": validation["estimated_gb"],
        }
 
# 使用例
agent = GeminiBigQueryAgent(
    dataset_id="google_analytics_sample",
    table_ids=["ga_sessions_20170801"]
)
 
result = agent.ask("デバイス別のコンバージョン率を教えてください")
print(result["summary"])

import re
 
FORBIDDEN_PATTERNS = [
    r"\bDROP\b",
    r"\bDELETE\b",
    r"\bUPDATE\b",
    r"\bINSERT\b",
    r"\bCREATE\b",
    r"\bALTER\b",
    r"\bTRUNCATE\b",
]
 
def is_safe_sql(sql: str) -> bool:
    """読み取り専用の SQL のみ許可する"""
    upper_sql = sql.upper()
    for pattern in FORBIDDEN_PATTERNS:
        if re.search(pattern, upper_sql):
            return False
    return upper_sql.strip().startswith("SELECT") or upper_sql.strip().startswith("WITH")

# コスト制限付きクエリ実行
job_config = QueryJobConfig(
    maximum_bytes_billed=1_000_000_000  # 1GB上限
)
df = client.query(sql, job_config=job_config).to_dataframe()