Gemini API × PostgreSQL 実装ガイド — AI駆動型データベース最適化システムを本番環境に構築する

取り組みの背景 — なぜ AI × PostgreSQL なのか

データベース最適化は、すべてのバックエンドエンジニアが避けて通れない課題です。クエリが遅い、インデックスが効いていない、スキーマ設計が最適でない——そういった問題の調査と解決には、深い専門知識と多くの時間が必要でしました。

Gemini 2.5 Pro の登場によって、この状況は大きく変わりつつあります。長大なシステムプロンプトにスキーマ情報を埋め込み、自然言語でSQLを生成させ、EXPLAINプランを解析させ、インデックス戦略を提案させる——これらすべてを一つのAPIで実現できます。

対象読者はPythonとPostgreSQLの基本を理解しているバックエンドエンジニア・SREの方々です。

1. システム全体アーキテクチャ

本システムは以下のコンポーネントで構成されます。

コアコンポーネント:

Schema Introspector: PostgreSQLのシステムカタログ（pg_catalog）からスキーマ情報を自動収集
Gemini Query Agent: スキーマをコンテキストに自然言語→SQL変換を実行
Query Analyzer: EXPLAINプランをGeminiに渡してボトルネックを特定
Index Advisor: テーブル統計とクエリパターンからインデックス戦略を提案
Schema Reviewer: DDL変更に対してGeminiがレビューコメントを返す
FastAPI サーバー: 上記機能をREST APIとして提供

# システム構成図（擬似コード）
#
# ┌─────────────────────────────────────────────┐
# │              FastAPI Application             │
# │                                             │
# │  /sql/generate  ← Text-to-SQL              │
# │  /query/analyze ← EXPLAIN Plan Analysis    │
# │  /index/advise  ← Index Recommendations    │
# │  /schema/review ← DDL Review               │
# └──────────────┬──────────────────────────────┘
#                │
#     ┌──────────▼──────────┐
#     │   Gemini 2.5 Pro    │
#     │   (google-genai)    │
#     └──────────┬──────────┘
#                │
#     ┌──────────▼──────────┐
#     │   PostgreSQL 16     │
#     │   (asyncpg)         │
#     └─────────────────────┘

2. 環境準備とライブラリインストール

必要なライブラリ

pip install google-genai asyncpg fastapi uvicorn pydantic python-dotenv

`.env` ファイルの設定

# .env
GEMINI_API_KEY=YOUR_GEMINI_API_KEY
DATABASE_URL=postgresql://user:password@localhost:5432/yourdb
 
# オプション: コスト管理用
GEMINI_MAX_TOKENS_PER_REQUEST=8192
GEMINI_DAILY_TOKEN_BUDGET=1000000

プロジェクト構成

db_ai_optimizer/
├── main.py              # FastAPI アプリケーション
├── schema_introspector.py  # スキーマ収集
├── gemini_agent.py      # Gemini API クライアント
├── query_analyzer.py    # クエリ解析
├── index_advisor.py     # インデックス推薦
├── schema_reviewer.py   # スキーマレビュー
├── models.py            # Pydantic モデル
└── .env

3. PostgreSQL スキーマの自動収集

AIにSQLを生成させるためには、まずデータベースのスキーマ情報を正確にコンテキストとして渡す必要があります。schema_introspector.py でその役割を担います。

# schema_introspector.py
import asyncpg
import json
from typing import Optional
 
class SchemaIntrospector:
    """PostgreSQL システムカタログからスキーマ情報を収集するクラス"""
 
    def __init__(self, dsn: str):
        self.dsn = dsn
        self._pool: Optional[asyncpg.Pool] = None
 
    async def init_pool(self):
        self._pool = await asyncpg.create_pool(
            self.dsn,
            min_size=2,
            max_size=10,
            command_timeout=30
        )
 
    async def get_schema_context(self, schema_name: str = "public") -> str:
        """
        指定スキーマのテーブル・カラム・インデックス・外部キー情報を
        Gemini に渡しやすい形式の文字列で返す
        """
        async with self._pool.acquire() as conn:
            # テーブルとカラム情報の取得
            tables = await conn.fetch("""
                SELECT
                    t.table_name,
                    c.column_name,
                    c.data_type,
                    c.is_nullable,
                    c.column_default,
                    col_description(
                        (t.table_schema || '.' || t.table_name)::regclass::oid,
                        c.ordinal_position
                    ) AS column_comment
                FROM information_schema.tables t
                JOIN information_schema.columns c
                    ON t.table_name = c.table_name
                    AND t.table_schema = c.table_schema
                WHERE t.table_schema = $1
                    AND t.table_type = 'BASE TABLE'
                ORDER BY t.table_name, c.ordinal_position
            """, schema_name)
 
            # インデックス情報の取得
            indexes = await conn.fetch("""
                SELECT
                    tablename,
                    indexname,
                    indexdef
                FROM pg_indexes
                WHERE schemaname = $1
                ORDER BY tablename, indexname
            """, schema_name)
 
            # 外部キー制約の取得
            fkeys = await conn.fetch("""
                SELECT
                    tc.table_name,
                    kcu.column_name,
                    ccu.table_name AS foreign_table_name,
                    ccu.column_name AS foreign_column_name
                FROM information_schema.table_constraints AS tc
                JOIN information_schema.key_column_usage AS kcu
                    ON tc.constraint_name = kcu.constraint_name
                    AND tc.table_schema = kcu.table_schema
                JOIN information_schema.constraint_column_usage AS ccu
                    ON ccu.constraint_name = tc.constraint_name
                WHERE tc.constraint_type = 'FOREIGN KEY'
                    AND tc.table_schema = $1
            """, schema_name)
 
            # テーブルごとの行数推定（統計情報）
            row_counts = await conn.fetch("""
                SELECT
                    relname AS table_name,
                    n_live_tup AS estimated_rows
                FROM pg_stat_user_tables
                WHERE schemaname = $1
            """, schema_name)
 
        return self._format_schema_context(
            tables, indexes, fkeys, row_counts
        )
 
    def _format_schema_context(self, tables, indexes, fkeys, row_counts) -> str:
        """スキーマ情報を自然言語に近い形式でフォーマット"""
        row_count_map = {r["table_name"]: r["estimated_rows"] for r in row_counts}
 
        # テーブルごとにカラムをグループ化
        table_cols: dict[str, list] = {}
        for row in tables:
            tname = row["table_name"]
            if tname not in table_cols:
                table_cols[tname] = []
            table_cols[tname].append(row)
 
        lines = ["## Database Schema\n"]
        for tname, cols in table_cols.items():
            est_rows = row_count_map.get(tname, "unknown")
            lines.append(f"### Table: {tname} (estimated rows: {est_rows})")
            for col in cols:
                nullable = "NULL" if col["is_nullable"] == "YES" else "NOT NULL"
                default = f" DEFAULT {col['column_default']}" if col["column_default"] else ""
                comment = f" -- {col['column_comment']}" if col["column_comment"] else ""
                lines.append(
                    f"  - {col['column_name']}: {col['data_type']} {nullable}{default}{comment}"
                )
 
        lines.append("\n## Indexes")
        for idx in indexes:
            lines.append(f"  - [{idx['tablename']}] {idx['indexname']}: {idx['indexdef']}")
 
        lines.append("\n## Foreign Keys")
        for fk in fkeys:
            lines.append(
                f"  - {fk['table_name']}.{fk['column_name']} → "
                f"{fk['foreign_table_name']}.{fk['foreign_column_name']}"
            )
 
        return "\n".join(lines)
 
    async def close(self):
        if self._pool:
            await self._pool.close()

4. Gemini Query Agent — Text-to-SQL の中核実装

スキーマ情報をシステムプロンプトに埋め込み、自然言語でSQLを生成させるエージェントです。セキュリティ上、生成されたSQLは 読み取り専用 のデータベース接続で実行することが鉄則です。

# gemini_agent.py
import re
from google import genai
from google.genai import types
 
class GeminiQueryAgent:
    """
    Gemini 2.5 Pro を使って自然言語 → SQL 変換を行うエージェント
    スキーマコンテキストをシステムプロンプトに埋め込む
    """
 
    MODEL_ID = "gemini-2.5-pro"
 
    def __init__(self, api_key: str, schema_context: str):
        self.client = genai.Client(api_key=api_key)
        self.schema_context = schema_context
 
    def _build_system_prompt(self) -> str:
        return f"""あなたはPostgreSQLの専門家です。
以下のデータベーススキーマを理解した上で、ユーザーの自然言語の質問を
正確なPostgreSQLクエリに変換してください。
 
重要なルール:
1. 必ずSELECT文のみを生成すること（INSERT/UPDATE/DELETE/DROP等は絶対禁止）
2. クエリの末尾にセミコロンを付けること
3. カラム名やテーブル名は必ず二重引用符で囲むこと（例: "users"."email"）
4. LIMIT句を必ず付けること（デフォルト: LIMIT 100）
5. 生成したSQLだけを返し、余分な説明は不要
 
{self.schema_context}
"""
 
    async def generate_sql(self, natural_language_query: str) -> dict:
        """
        自然言語クエリをSQLに変換する
 
        Args:
            natural_language_query: 例 "先月の売上が10万円以上のユーザーを教えて"
 
        Returns:
            {"sql": "...", "explanation": "...", "confidence": 0.95}
        """
        response = self.client.models.generate_content(
            model=self.MODEL_ID,
            contents=[
                types.Content(
                    role="user",
                    parts=[types.Part(text=natural_language_query)]
                )
            ],
            config=types.GenerateContentConfig(
                system_instruction=self._build_system_prompt(),
                temperature=0.1,  # 低温度で決定論的に
                max_output_tokens=2048,
            )
        )
 
        raw_text = response.text.strip()
 
        # コードブロックからSQLを抽出
        sql = self._extract_sql(raw_text)
 
        # 安全性チェック: SELECT のみ許可
        if not self._is_safe_sql(sql):
            raise ValueError(f"安全でないSQL生成: {sql[:100]}...")
 
        # 追加の説明を生成
        explanation = await self._generate_explanation(sql)
 
        return {
            "sql": sql,
            "explanation": explanation,
            "raw_response": raw_text,
            "model": self.MODEL_ID,
            "usage": {
                "input_tokens": response.usage_metadata.prompt_token_count,
                "output_tokens": response.usage_metadata.candidates_token_count,
            }
        }
 
    def _extract_sql(self, text: str) -> str:
        """コードブロックまたは生テキストからSQL文を抽出"""
        # ```sql ... ``` ブロックを優先
        match = re.search(r"```(?:sql)?\s*(.*?)```", text, re.DOTALL | re.IGNORECASE)
        if match:
            return match.group(1).strip()
        # コードブロックなしの場合はそのまま
        return text.strip()
 
    def _is_safe_sql(self, sql: str) -> bool:
        """DDL・DML の混入チェック（SELECT のみ許可）"""
        sql_upper = sql.upper().strip()
        dangerous_keywords = [
            "INSERT", "UPDATE", "DELETE", "DROP", "CREATE",
            "ALTER", "TRUNCATE", "GRANT", "REVOKE", "EXECUTE"
        ]
        # SELECT から始まること
        if not sql_upper.startswith("SELECT") and not sql_upper.startswith("WITH"):
            return False
        # 危険キーワードが含まれていないこと
        for keyword in dangerous_keywords:
            if re.search(rf"\b{keyword}\b", sql_upper):
                return False
        return True
 
    async def _generate_explanation(self, sql: str) -> str:
        """生成されたSQLの日本語説明を生成"""
        response = self.client.models.generate_content(
            model="gemini-2.5-flash",  # 説明はFlashで十分かつコスト節約
            contents=[
                types.Content(
                    role="user",
                    parts=[types.Part(
                        text=f"以下のSQLを1〜2文で日本語で簡潔に説明してください:\n{sql}"
                    )]
                )
            ],
            config=types.GenerateContentConfig(
                temperature=0.3,
                max_output_tokens=256
            )
        )
        return response.text.strip()

5. クエリ性能分析 — EXPLAIN プランをAIが読む

EXPLAINプランの解読は、慣れていないエンジニアにとって難解です。Gemini にプランを渡すことで、ボトルネックの特定と改善提案を自動化できます。

# query_analyzer.py
import asyncpg
from google import genai
from google.genai import types
 
class QueryAnalyzer:
    """SQLクエリのEXPLAINプランをGeminiで解析するクラス"""
 
    def __init__(self, api_key: str, dsn: str):
        self.client = genai.Client(api_key=api_key)
        self.dsn = dsn
 
    async def analyze_query(self, sql: str) -> dict:
        """
        SQLのEXPLAINプランを取得し、Geminiで解析する
 
        Returns:
            {
                "explain_plan": "...",
                "analysis": "...",
                "bottlenecks": [...],
                "suggestions": [...],
                "estimated_cost": 123.45
            }
        """
        # EXPLAIN ANALYZE は本番DBへの負荷があるため、
        # まず EXPLAIN (FORMAT JSON) のみ実行
        explain_plan = await self._get_explain_plan(sql)
 
        # Geminiで解析
        analysis_result = await self._analyze_with_gemini(sql, explain_plan)
 
        return {
            "explain_plan": explain_plan,
            **analysis_result
        }
 
    async def _get_explain_plan(self, sql: str) -> str:
        """PostgreSQLからEXPLAINプランを取得（JSON形式）"""
        conn = await asyncpg.connect(self.dsn)
        try:
            # 読み取り専用トランザクションで実行
            async with conn.transaction():
                plan = await conn.fetchval(
                    f"EXPLAIN (FORMAT JSON, COSTS true, BUFFERS false) {sql}"
                )
            return str(plan)
        finally:
            await conn.close()
 
    async def _analyze_with_gemini(self, sql: str, explain_plan: str) -> dict:
        """Gemini 2.5 Pro でEXPLAINプランを解析"""
        prompt = f"""あなたはPostgreSQLのパフォーマンスチューニング専門家です。
以下のSQLとそのEXPLAINプランを分析し、JSON形式で回答してください。
 
## 対象SQL:
```sql
{sql}

EXPLAINプラン:

{explain_plan}

以下のJSON形式で回答してください（日本語）: {{ "summary": "パフォーマンス評価の一言サマリー（良好/要改善/危険）", "estimated_cost": 数値（total_costの値）, "bottlenecks": [ "特定されたボトルネック1", "特定されたボトルネック2" ], "suggestions": [ "具体的な改善提案1（例: users.email にインデックスを追加）", "具体的な改善提案2" ], "analysis": "詳細な解析コメント（200文字程度）" }} """

    response = self.client.models.generate_content(
        model="gemini-2.5-pro",
        contents=[types.Content(role="user", parts=[types.Part(text=prompt)])],
        config=types.GenerateContentConfig(
            temperature=0.2,
            max_output_tokens=1024,
            response_mime_type="application/json"  # JSON出力を強制
        )
    )

    import json
    result = json.loads(response.text)
    return result


---

## 6. インデックス推薦エンジン

クエリパターンとテーブル統計を組み合わせ、Gemini が最適なインデックス戦略を提案します。

```python
# index_advisor.py
import asyncpg
from google import genai
from google.genai import types
import json

class IndexAdvisor:
    """テーブル統計とクエリパターンからインデックスを推薦するクラス"""

    def __init__(self, api_key: str, dsn: str):
        self.client = genai.Client(api_key=api_key)
        self.dsn = dsn

    async def advise(self, table_name: str, sample_queries: list[str]) -> dict:
        """
        指定テーブルに対するインデックス推薦を生成する

        Args:
            table_name: 対象テーブル名
            sample_queries: テーブルを使用する代表的なクエリリスト

        Returns:
            {"recommended_indexes": [...], "rationale": "...", "ddl_statements": [...]}
        """
        # テーブル統計の取得
        table_stats = await self._get_table_statistics(table_name)

        # 既存インデックスの取得
        existing_indexes = await self._get_existing_indexes(table_name)

        # Geminiで推薦を生成
        return await self._generate_recommendations(
            table_name, table_stats, existing_indexes, sample_queries
        )

    async def _get_table_statistics(self, table_name: str) -> dict:
        """pg_stats からカラム統計を取得"""
        conn = await asyncpg.connect(self.dsn)
        try:
            stats = await conn.fetch("""
                SELECT
                    attname AS column_name,
                    n_distinct,
                    correlation,
                    most_common_vals::text,
                    null_frac
                FROM pg_stats
                WHERE tablename = $1
                ORDER BY n_distinct DESC
            """, table_name)
            return [dict(s) for s in stats]
        finally:
            await conn.close()

    async def _get_existing_indexes(self, table_name: str) -> list[str]:
        """既存インデックス定義を取得"""
        conn = await asyncpg.connect(self.dsn)
        try:
            indexes = await conn.fetch("""
                SELECT indexdef FROM pg_indexes
                WHERE tablename = $1
            """, table_name)
            return [idx["indexdef"] for idx in indexes]
        finally:
            await conn.close()

    async def _generate_recommendations(
        self, table_name, stats, existing_indexes, sample_queries
    ) -> dict:
        prompt = f"""あなたはPostgreSQLのインデックス設計専門家です。

## テーブル名: {table_name}

## カラム統計情報:
{json.dumps(stats, ensure_ascii=False, indent=2)}

## 既存インデックス:
{chr(10).join(existing_indexes)}

## 代表的なクエリパターン:
{chr(10).join(f"- {q}" for q in sample_queries)}

上記の情報を元に、以下のJSON形式でインデックス推薦を返してください:
{{
  "recommended_indexes": [
    {{
      "columns": ["col1", "col2"],
      "type": "btree",
      "reason": "このインデックスが有効な理由",
      "priority": "high/medium/low"
    }}
  ],
  "ddl_statements": [
    "CREATE INDEX CONCURRENTLY idx_{table_name}_xxx ON {table_name}(col1, col2);"
  ],
  "existing_redundant_indexes": ["冗長な既存インデックスがあれば列挙"],
  "rationale": "推薦全体の根拠説明（200文字程度）"
}}
"""

        response = self.client.models.generate_content(
            model="gemini-2.5-pro",
            contents=[types.Content(role="user", parts=[types.Part(text=prompt)])],
            config=types.GenerateContentConfig(
                temperature=0.2,
                max_output_tokens=2048,
                response_mime_type="application/json"
            )
        )

        return json.loads(response.text)

7. スキーマレビューシステム

DDL変更（ALTER TABLE / CREATE TABLE）をGeminiにレビューさせることで、設計上の問題を早期発見できます。

# schema_reviewer.py
from google import genai
from google.genai import types
import json
 
class SchemaReviewer:
    """DDL変更をGeminiがレビューするクラス"""
 
    def __init__(self, api_key: str, schema_context: str):
        self.client = genai.Client(api_key=api_key)
        self.schema_context = schema_context
 
    async def review_ddl(self, proposed_ddl: str) -> dict:
        """
        提案されたDDLをレビューする
 
        Args:
            proposed_ddl: レビュー対象のDDL文
 
        Returns:
            {"score": 85, "issues": [...], "suggestions": [...], "approved": True}
        """
        prompt = f"""あなたはPostgreSQLのデータベース設計レビュアーです。
既存スキーマのコンテキストを踏まえ、提案されたDDL変更をレビューしてください。
 
## 既存スキーマ:
{self.schema_context}
 
## 提案されたDDL:
```sql
{proposed_ddl}

以下のJSON形式でレビュー結果を返してください: {{ "score": 0-100の評価点, "approved": true/false, "issues": [ {{ "severity": "critical/warning/info", "message": "問題の説明", "line": "問題のある行（わかる場合）" }} ], "suggestions": [ "改善提案1", "改善提案2" ], "performance_impact": "パフォーマンスへの影響評価", "migration_notes": "本番適用時の注意事項" }}

レビューの観点:

NOT NULL制約、DEFAULT値の適切さ
インデックス戦略（主キー・外部キー・検索用）
データ型の選択（UUID vs SERIAL vs BIGSERIAL）
命名規則の一貫性
既存データへの影響（LOCK、中断リスク）

CONCURRENT オプションの使用推奨 """

  response = self.client.models.generate_content(
      model="gemini-2.5-pro",
      contents=[types.Content(role="user", parts=[types.Part(text=prompt)])],
      config=types.GenerateContentConfig(
          temperature=0.1,
          max_output_tokens=2048,
          response_mime_type="application/json"
      )
  )

  return json.loads(response.text)


---

## 8. FastAPI でAPIサーバーとして提供

すべてのコンポーネントをまとめてREST APIとして公開します。

```python
# main.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from dotenv import load_dotenv

from schema_introspector import SchemaIntrospector
from gemini_agent import GeminiQueryAgent
from query_analyzer import QueryAnalyzer
from index_advisor import IndexAdvisor
from schema_reviewer import SchemaReviewer

load_dotenv()

GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
DATABASE_URL = os.getenv("DATABASE_URL")
# 読み取り専用DBユーザーのURL（Text-to-SQL実行用）
DATABASE_URL_READONLY = os.getenv("DATABASE_URL_READONLY", DATABASE_URL)

# グローバルインスタンス
introspector: SchemaIntrospector = None
query_agent: GeminiQueryAgent = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """アプリ起動時にスキーマを読み込む"""
    global introspector, query_agent
    introspector = SchemaIntrospector(DATABASE_URL)
    await introspector.init_pool()

    # スキーマコンテキストを一度取得してキャッシュ
    schema_context = await introspector.get_schema_context()
    query_agent = GeminiQueryAgent(GEMINI_API_KEY, schema_context)

    print("✅ スキーマコンテキスト読み込み完了")
    yield

    await introspector.close()

app = FastAPI(
    title="DB AI Optimizer API",
    description="Gemini 2.5 Pro × PostgreSQL最適化システム",
    lifespan=lifespan
)

class SQLGenerateRequest(BaseModel):
    query: str  # 自然言語のクエリ

class QueryAnalyzeRequest(BaseModel):
    sql: str

class IndexAdviseRequest(BaseModel):
    table_name: str
    sample_queries: list[str]

class SchemaReviewRequest(BaseModel):
    ddl: str

@app.post("/sql/generate")
async def generate_sql(req: SQLGenerateRequest):
    """自然言語 → SQL変換エンドポイント"""
    try:
        result = await query_agent.generate_sql(req.query)
        return result
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"SQL生成エラー: {str(e)}")

@app.post("/query/analyze")
async def analyze_query(req: QueryAnalyzeRequest):
    """クエリパフォーマンス解析エンドポイント"""
    analyzer = QueryAnalyzer(GEMINI_API_KEY, DATABASE_URL)
    try:
        result = await analyzer.analyze_query(req.sql)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"解析エラー: {str(e)}")

@app.post("/index/advise")
async def advise_index(req: IndexAdviseRequest):
    """インデックス推薦エンドポイント"""
    advisor = IndexAdvisor(GEMINI_API_KEY, DATABASE_URL)
    try:
        result = await advisor.advise(req.table_name, req.sample_queries)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"推薦エラー: {str(e)}")

@app.post("/schema/review")
async def review_schema(req: SchemaReviewRequest):
    """スキーマDDLレビューエンドポイント"""
    schema_context = await introspector.get_schema_context()
    reviewer = SchemaReviewer(GEMINI_API_KEY, schema_context)
    try:
        result = await reviewer.review_ddl(req.ddl)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"レビューエラー: {str(e)}")

@app.get("/health")
async def health_check():
    return {"status": "ok", "model": "gemini-2.5-pro"}

サーバーの起動

uvicorn main:app --host 0.0.0.0 --port 8000 --reload
 
# 本番環境では workers を増やす
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

9. 本番運用のセキュリティ設計

読み取り専用ロールの作成

Text-to-SQLで生成されたクエリは必ず読み取り専用ユーザーで実行する点が肝心です。

-- 読み取り専用ユーザーの作成
CREATE ROLE ai_optimizer_readonly;
GRANT CONNECT ON DATABASE yourdb TO ai_optimizer_readonly;
GRANT USAGE ON SCHEMA public TO ai_optimizer_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ai_optimizer_readonly;
 
-- デフォルト権限（将来のテーブルにも適用）
ALTER DEFAULT PRIVILEGES IN SCHEMA public
  GRANT SELECT ON TABLES TO ai_optimizer_readonly;
 
CREATE USER ai_optimizer_user WITH PASSWORD 'strong_password'
  IN ROLE ai_optimizer_readonly;

レートリミットとトークンバジェット管理

# token_budget.py
import asyncio
from collections import defaultdict
from datetime import datetime, date
 
class TokenBudgetManager:
    """
    Gemini API のトークン使用量を管理し、
    日次上限を超えないよう制御するクラス
    """
 
    def __init__(self, daily_budget: int = 1_000_000):
        self.daily_budget = daily_budget
        self._usage: dict[str, int] = defaultdict(int)
        self._lock = asyncio.Lock()
 
    async def consume(self, tokens: int) -> bool:
        """トークンを消費。上限超過なら False を返す"""
        async with self._lock:
            today = str(date.today())
            if self._usage[today] + tokens > self.daily_budget:
                return False
            self._usage[today] += tokens
            return True
 
    def get_today_usage(self) -> int:
        return self._usage.get(str(date.today()), 0)

10. コスト管理とモニタリング

Gemini APIのコストは、使用するモデルとトークン数に依存します。本システムで想定されるコスト感を把握しておきましょう。

トークンコスト概算（2026年4月時点 Gemini 2.5 Pro）:

Input: $1.25 / 1M tokens（128k以下の場合）
Output: $10.00 / 1M tokens

1リクエストあたりの目安:

Text-to-SQL: 入力 2,000〜5,000 tokens、出力 200〜500 tokens
EXPLAIN分析: 入力 3,000〜8,000 tokens、出力 500〜1,000 tokens
インデックス推薦: 入力 2,000〜4,000 tokens、出力 1,000〜2,000 tokens

1日100リクエストで概算しても、$0.50〜$2.00程度に収まります。

スキーマコンテキストを毎回送信するのではなく、Gemini のコンテキストキャッシュ（cachedContent）機能を活用すれば、さらにコストを削減できます。詳しくはコンテキストキャッシュ活用ガイドをご参照ください。

まとめ

ここではGemini 2.5 Pro と PostgreSQL を組み合わせた AI駆動型データベース最適化システムを完全実装しました。

実装したのは以下の4つの主要機能です。

Text-to-SQL: 自然言語からSQLを生成し、安全に実行
クエリ分析: EXPLAINプランをAIが解読し、ボトルネックと改善策を提示
インデックス推薦: テーブル統計とクエリパターンから最適なインデックス戦略を自動提案
スキーマレビュー: DDL変更の問題点を事前にAIがレビュー

特に重要なポイントとして、読み取り専用ロールによるセキュリティ設計と、コンテキストキャッシュを活用したコスト最適化を忘れずに実装してください。このシステムを土台に、Slack Bot連携やCI/CDパイプラインへの組み込みなど、さらなる拡張も容易に実現できます。

データベース最適化の実践知識

Gemini API × PostgreSQL 実装ガイド — AI駆動型データベース最適化システムを本番環境に構築する

取り組みの背景 — なぜ AI × PostgreSQL なのか

1. システム全体アーキテクチャ

2. 環境準備とライブラリインストール

必要なライブラリ

`.env` ファイルの設定

プロジェクト構成

3. PostgreSQL スキーマの自動収集

4. Gemini Query Agent — Text-to-SQL の中核実装

5. クエリ性能分析 — EXPLAIN プランをAIが読む

EXPLAINプラン:

7. スキーマレビューシステム

サーバーの起動

9. 本番運用のセキュリティ設計

読み取り専用ロールの作成

レートリミットとトークンバジェット管理

10. コスト管理とモニタリング

まとめ

お読みいただきありがとうございます

関連記事

Gemini API × PostgreSQL 実装ガイド — AI駆動型データベース最適化システムを本番環境に構築する

取り組みの背景 — なぜ AI × PostgreSQL なのか

1. システム全体アーキテクチャ

2. 環境準備とライブラリインストール

必要なライブラリ

.env ファイルの設定

プロジェクト構成

3. PostgreSQL スキーマの自動収集

4. Gemini Query Agent — Text-to-SQL の中核実装

5. クエリ性能分析 — EXPLAIN プランをAIが読む

EXPLAINプラン:

7. スキーマレビューシステム

サーバーの起動

9. 本番運用のセキュリティ設計

読み取り専用ロールの作成

レートリミットとトークンバジェット管理

10. コスト管理とモニタリング

まとめ

お読みいただきありがとうございます

関連記事

`.env` ファイルの設定