Claude/ChatGPT APIでSQLデータベースを安全に検索する「回答契約」パターン【RAG実装 2026年版】

なぜ「回答契約」が必要か

❌ 問題のある実装:
  ユーザー: "全ユーザーの個人情報を消して"
  AI → DELETE FROM users; ← 実行されてしまう

✅ 回答契約を定義した実装:
  システムプロンプト: "SELECT文のみ許可。DML禁止。"
  AI → "DELETE文は禁止されています" ← 安全に拒否

回答契約テンプレート（システムプロンプト）

ANSWER_CONTRACT = """
あなたはデータベースアシスタントです。以下のルールを厳守してください。

## 許可されていること
- SELECT文のみ実行可能
- 最大100行までの結果を返す
- スキーマ情報の確認（SHOW TABLES、DESCRIBE等）

## 禁止されていること
- INSERT、UPDATE、DELETE、DROP、TRUNCATE、ALTER
- サブクエリを使った間接的なデータ変更
- 個人情報（メールアドレス・電話番号）の一括取得

## データベーススキーマ
テーブル: orders（id, user_id, amount, status, created_at）
テーブル: users（id, name, created_at）※メールは取得禁止

## 回答フォーマット
1. 生成したSQL文を提示する
2. クエリ結果を表形式で表示する
3. 結果の解釈を日本語で説明する
4. 不明な点は質問する（推測で実行しない）
"""

Claude APIでの実装

# sql_rag_claude.py
# Claudeを使った安全なNL→SQL変換システム

import anthropic
import sqlite3
import os

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
DB_PATH = "./orders.db"

def validate_query(sql: str) -> tuple[bool, str]:
    """SQL文のセキュリティバリデーション"""
    sql_upper = sql.strip().upper()
    
    FORBIDDEN = ["INSERT", "UPDATE", "DELETE", "DROP", "TRUNCATE", "ALTER", "CREATE"]
    for keyword in FORBIDDEN:
        if keyword in sql_upper:
            return False, f"禁止されたSQL操作: {keyword}"
    
    if not sql_upper.startswith("SELECT") and not sql_upper.startswith("SHOW"):
        return False, "SELECT文またはSHOW文のみ許可"
    
    return True, "OK"

def execute_query(sql: str, max_rows: int = 100) -> dict:
    """SQLを実行して結果を返す"""
    is_valid, reason = validate_query(sql)
    if not is_valid:
        return {"error": reason, "sql": sql}
    
    try:
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.execute(sql)
        cols = [d[0] for d in cursor.description]
        rows = cursor.fetchmany(max_rows)
        conn.close()
        return {"columns": cols, "rows": rows, "sql": sql}
    except Exception as e:
        return {"error": str(e), "sql": sql}

def nl_to_sql_query(user_question: str) -> str:
    """自然言語→SQL変換してデータを取得する"""
    
    # Step 1: Claudeにプロンプト変換を依頼
    response = client.messages.create(
        model="claude-sonnet-4-5",    # 複雑なスキーマには長いコンテキストが有利
        max_tokens=512,
        system=ANSWER_CONTRACT,
        messages=[{"role": "user", "content": user_question}]
    )
    
    answer = response.content[0].text
    
    # Step 2: 回答からSQLを抽出
    if "```sql" in answer:
        sql = answer.split("```sql")[1].split("```")[0].strip()
        result = execute_query(sql)
        
        if "error" in result:
            return f"SQLエラー: {result['error']}"
        
        rows_text = "\n".join(str(row) for row in result["rows"][:10])
        return f"SQL: {sql}\n\n結果:\n{rows_text}"
    
    return answer  # SQLが含まれない場合はそのまま返す

# 使用例
print(nl_to_sql_query("先月の注文件数を教えてください"))
print(nl_to_sql_query("ユーザーIDが5の注文履歴は？"))

GPT-4o miniでの実装（コスト優先）

# sql_rag_gpt.py
# GPT-4o miniで同じシステムを実装（入力$0.15/1M vs Sonnet $3/1M）

from openai import OpenAI

openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def nl_to_sql_gpt(user_question: str) -> str:
    """GPT-4o miniでNL→SQL変換（シンプルなクエリ向け）"""
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",   # 入力$0.15/1M（最安）
        max_tokens=512,
        messages=[
            {"role": "system", "content": ANSWER_CONTRACT},
            {"role": "user", "content": user_question}
        ]
    )
    
    answer = response.choices[0].message.content
    
    if "```sql" in answer:
        sql = answer.split("```sql")[1].split("```")[0].strip()
        result = execute_query(sql)
        
        if "error" in result:
            return f"SQLエラー: {result['error']}"
        
        return f"SQL: {sql}\n結果: {result['rows'][:5]}"
    
    return answer

# コスト比較: 1000クエリ（入力2000 + 出力500トークン）
# GPT-4o mini: (2000*0.15 + 500*0.60)/1M * 1000 = $0.30/月
# Claude Sonnet: (2000*3.00 + 500*15.00)/1M * 1000 = $13.50/月

あわせて読みたい

• LangChain vs LlamaIndex【2026年版】RAG実装コードで比較
• Claude CodeにMCPサーバーを接続する方法
• MCPとは？Model Context Protocolを5分で理解