LLMによるText to SQLの原理は3つの段階で説明できます。まず、モデルにデータベースのスキーマ（関連するテーブルのDDL）を説明します。次に、ユーザーの質問と厳格な指示（ターゲットのダイアレクトに対してSQLクエリのみを生成すること）を伝えます。最後に、クエリを取得し、検証して、読み取り専用で実行します。
import psycopg2
def get_schema(conn):
"""Retourne le schéma sous forme de CREATE TABLE simplifiés."""
query = """
SELECT table_name, column_name, data_type
FROM information_schema.columns
WHERE table_schema = 'public'
ORDER BY table_name, ordinal_position;
"""
tables = {}
with conn.cursor() as cur:
cur.execute(query)
for table, col, dtype in cur.fetchall():
tables.setdefault(table, []).append(f"{col} {dtype}")
lines = []
for table, cols in tables.items():
cols_str = ", ".join(cols)
lines.append(f"TABLE {table} ({cols_str});")
return "\n".join(lines)
→
業務上の意味を説明するコメントを追加してください
「ca_ht」という列名だけでは、モデルにとって意味が曖昧です。スキーマに「ca_ht（税抜売上高、ユーロ単位）」のような注釈を追加してください。この数語だけで、列の選択ミスを大幅に減らせます。PostgreSQLでは、COMMENT ON COLUMNで設定したコメントをinformation_schemaやpg_descriptionから取得できます。
import re
import ollama
import psycopg2
import sqlparse
MODEL = "qwen3-coder:30b"
SYSTEM_PROMPT = """Tu es un expert PostgreSQL. Génère UNE seule requête SQL
qui répond à la question de l'utilisateur, en respectant ces règles :
- Uniquement des requêtes SELECT (jamais INSERT/UPDATE/DELETE/DROP).
- Utilise exactement les noms de tables et colonnes du schéma fourni.
- Ajoute toujours LIMIT 100 si la question ne précise pas de limite.
- Réponds UNIQUEMENT avec le SQL, sans explication ni balise Markdown.
Schéma de la base :
{schema}"""
def generate_sql(question, schema):
resp = ollama.chat(
model=MODEL,
messages=[
{"role": "system", "content": SYSTEM_PROMPT.format(schema=schema)},
{"role": "user", "content": question},
],
options={"temperature": 0}, # déterminisme : crucial pour du SQL
)
return clean_sql(resp["message"]["content"])
def clean_sql(raw):
# Retire les fences Markdown ```sql ... ``` si le modèle en ajoute
raw = re.sub(r"```(?:sql)?", "", raw).strip()
return raw.rstrip(";") + ";"
def is_read_only(sql):
statements = sqlparse.parse(sql)
if len(statements) != 1:
return False # une seule requête, pas d'empilement
stmt = statements[0]
if stmt.get_type() != "SELECT":
return False
forbidden = ("insert", "update", "delete", "drop",
"alter", "truncate", "grant", "create")
lowered = sql.lower()
return not any(kw in lowered for kw in forbidden)
def run_query(sql):
if not is_read_only(sql):
raise ValueError(f"Requête refusée (non lecture seule) : {sql}")
# Rôle 'readonly' : ne dispose QUE du privilège SELECT côté base
conn = psycopg2.connect(
dbname="analytics", user="readonly",
password="...", host="localhost",
)
with conn.cursor() as cur:
cur.execute("SET statement_timeout = '5s';") # anti-requête folle
cur.execute(sql)
cols = [d[0] for d in cur.description]
rows = cur.fetchall()
conn.close()
return cols, rows
if __name__ == "__main__":
from schema import get_schema
ro = psycopg2.connect(dbname="analytics", user="readonly",
password="...", host="localhost")
schema = get_schema(ro)
question = "Combien de commandes par mois en 2025 ?"
sql = generate_sql(question, schema)
print("SQL généré :", sql)
cols, rows = run_query(sql)
print(cols)
for r in rows:
print(r)
SELECT権限だけを持つSQLロールを作成してください。モデルがDROP TABLEを生成しても、データベースは拒否します。これが唯一、本当に信頼できる安全策です。「GRANT SELECT ON ALL TABLES IN SCHEMA public TO readonly;」だけを実行し、それ以外の権限は付与しないでください。
-- À exécuter une fois par un admin
CREATE ROLE readonly WITH LOGIN PASSWORD '...';
GRANT CONNECT ON DATABASE analytics TO readonly;
GRANT USAGE ON SCHEMA public TO readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO readonly;
-- Les tables créées plus tard héritent aussi du SELECT
ALTER DEFAULT PRIVILEGES IN SCHEMA public
GRANT SELECT ON TABLES TO readonly;