import sqlite3def create_sample_db(db_path: str = "ecommerce.db"): conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.executescript(""" CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, category TEXT NOT NULL, price REAL NOT NULL, stock INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY, customer_name TEXT NOT NULL, order_date DATE NOT NULL, total_amount REAL NOT NULL ); CREATE TABLE IF NOT EXISTS order_items ( id INTEGER PRIMARY KEY, order_id INTEGER REFERENCES orders(id), product_id INTEGER REFERENCES products(id), quantity INTEGER NOT NULL, unit_price REAL NOT NULL ); INSERT OR IGNORE INTO products VALUES (1, 'ワイヤレスイヤホン', 'オーディオ', 12800, 150), (2, 'USBハブ 7ポート', 'アクセサリ', 3980, 300), (3, 'メカニカルキーボード', '入力デバイス', 15800, 80), (4, 'ウェブカメラ 4K', 'カメラ', 8900, 120), (5, '外付けSSD 1TB', 'ストレージ', 11200, 200); """) conn.commit() conn.close()
import reBLOCKED_KEYWORDS = [ "DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE", "TRUNCATE", "EXEC",]def validate_sql(sql: str) -> bool: upper_sql = sql.upper().strip() if not upper_sql.startswith("SELECT"): return False for keyword in BLOCKED_KEYWORDS: if re.search(rf"\b{keyword}\b", upper_sql): return False return Truedef execute_query(db_path: str, sql: str) -> list: if not validate_sql(sql): raise ValueError( "安全でないクエリが検出されました" ) conn = sqlite3.connect(db_path) conn.execute("PRAGMA query_only = ON") cursor = conn.cursor() cursor.execute(sql) columns = [desc[0] for desc in cursor.description] results = cursor.fetchall() conn.close() return {"columns": columns, "rows": results}
PRAGMA query_only = ON により、データベースレベルでも書き込みを禁止しています。
ステップ5：エンドツーエンドのパイプライン
すべてのコンポーネントを組み合わせて、完全な Text-to-SQL パイプラインを構築します。
def ask_database(question: str, db_path: str = "ecommerce.db"): schema = extract_schema(db_path) sql = generate_sql(question, schema) print(f"生成されたSQL: {sql}") result = execute_query(db_path, sql) print(f"カラム: {result['columns']}") for row in result["rows"]: print(row) return result# 使用例ask_database("在庫が100個以上ある商品を価格が高い順に教えて")# → SELECT name, price, stock FROM products# WHERE stock >= 100 ORDER BY price DESCask_database("カテゴリごとの商品数を教えて")# → SELECT category, COUNT(*) as count# FROM products GROUP BY category
精度を向上させるテクニック
Few-Shot プロンプティング
システムプロンプトに質問と回答のペアを数例含めることで、生成精度が大幅に向上します。
FEW_SHOT_EXAMPLES = """質問: 最も高い商品は？SQL: SELECT name, price FROM products ORDER BY price DESC LIMIT 1質問: 2026年3月の注文数は？SQL: SELECT COUNT(*) FROM orders WHERE order_date BETWEEN '2026-03-01' AND '2026-03-31'質問: 注文金額の合計が最も多い顧客は？SQL: SELECT customer_name, SUM(total_amount) as total FROM orders GROUP BY customer_name ORDER BY total DESC LIMIT 1"""
スキーマにコメントを追加する
Gemini はカラムの説明があるとより正確なクエリを生成できます。
ANNOTATED_SCHEMA = """TABLE products ( id INTEGER PRIMARY KEY -- 商品ID name TEXT NOT NULL -- 商品名（日本語） category TEXT NOT NULL -- 商品カテゴリ price REAL NOT NULL -- 税込価格（円） stock INTEGER -- 現在の在庫数)"""