OllamaとText-to-SQLでPostgreSQLを自然言語で操作する方法｜PythonとローカルLLMでSQLを自動生成して業務データを安全に検索する手順

この記事のポイント

・OllamaのPOST /api/generateに「CREATE TABLEスキーマ＋質問」を渡すとSQL文が得られる
・information_schemaでスキーマを動的取得すれば、テーブル追加後もプロンプト更新不要になる
・生成SQLは読み取り専用ユーザー＋SELECTのみ許可フィルタで安全に実行できる
・few-shot例をプロンプトに追加するとJOINや集計クエリの生成精度が大幅に向上する

# Ollamaのサービス状態を確認
$ systemctl status ollama

# llama3.3またはmistralのモデルがあるか確認
$ ollama list
NAME                              ID              SIZE    MODIFIED
llama3.3:70b-instruct-q4_0        ab12cd34ef56    43 GB   3 days ago
mistral:7b-instruct-q4_0          ff11aa22bb33    4.1 GB  5 days ago

# postgres ユーザーでpsqlを起動
$ sudo -u postgres psql

-- データベースとロールの作成
postgres=# CREATE DATABASE bizdb;
postgres=# CREATE USER bizadmin WITH PASSWORD 'admin_pw';
postgres=# GRANT ALL PRIVILEGES ON DATABASE bizdb TO bizadmin;
postgres=# CREATE USER bizreader WITH PASSWORD 'reader_pw';

postgres=# \c bizdb

bizdb=# GRANT CONNECT ON DATABASE bizdb TO bizreader;

-- サンプルテーブルの作成
bizdb=# CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    prefecture VARCHAR(20),
    created_at DATE DEFAULT CURRENT_DATE
);

bizdb=# CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    customer_id INTEGER REFERENCES customers(id),
    product_name VARCHAR(100) NOT NULL,
    amount INTEGER NOT NULL,
    order_date DATE NOT NULL
);

-- サンプルデータの投入
bizdb=# INSERT INTO customers (name, prefecture) VALUES
    ('株式会社アルファ', '東京都'),
    ('ベータ工業', '大阪府'),
    ('ガンマ商事', '愛知県');

bizdb=# INSERT INTO orders (customer_id, product_name, amount, order_date) VALUES
    (1, 'Linux研修パッケージ', 198000, '2026-07-15'),
    (2, 'サーバー保守契約', 480000, '2026-07-20'),
    (1, 'クラウド移行支援', 320000, '2026-08-01'),
    (3, 'Linux研修パッケージ', 198000, '2026-08-10');

bizdb=# \q

$ sudo -u postgres psql -d bizdb

bizdb=# GRANT USAGE ON SCHEMA public TO bizreader;
bizdb=# GRANT SELECT ON ALL TABLES IN SCHEMA public TO bizreader;
bizdb=# ALTER DEFAULT PRIVILEGES IN SCHEMA public
        GRANT SELECT ON TABLES TO bizreader;
bizdb=# \q

# 接続確認（bizreaderでログイン）
$ psql -U bizreader -d bizdb -h localhost -c "SELECT count(*) FROM orders;"
 count
-------
     4
(1 row)

import psycopg2

def get_schema_ddl(conn, schema_name="public"):
    cur = conn.cursor()
    cur.execute(
        """
        SELECT c.table_name, c.column_name,
               c.data_type, c.is_nullable
        FROM information_schema.columns c
        WHERE c.table_schema = %s
        ORDER BY c.table_name, c.ordinal_position
        """,
        (schema_name,)
    )
    rows = cur.fetchall()
    tables = {}
    for table_name, col_name, data_type, nullable in rows:
        if table_name not in tables:
            tables[table_name] = []
        col_def = f"  {col_name} {data_type.upper()}"
        if nullable == "NO":
            col_def += " NOT NULL"
        tables[table_name].append(col_def)
    ddl = ""
    for table_name, cols in tables.items():
        ddl += f"CREATE TABLE {table_name} (\n"
        ddl += ",\n".join(cols)
        ddl += "\n);\n\n"
    return ddl.strip()


# 動作確認
conn = psycopg2.connect(
    dbname="bizdb", user="bizreader",
    password="reader_pw", host="localhost"
)
print(get_schema_ddl(conn))

def build_prompt(schema_ddl, question):
    return f"""あなたはPostgreSQLの専門エンジニアです。
以下のデータベーススキーマを参照して、質問に対応するSQLクエリを生成してください。

制約:
- SELECT文のみ生成する。INSERT/UPDATE/DELETE/DROPは使わない
- コードブロック記号を含めず、SQLのみを返す
- 結果が多い場合はLIMIT 20を付ける

スキーマ:
{schema_ddl}

質問: {question}

SQL:"""

import requests

def generate_sql(prompt, model="llama3.3:70b-instruct-q4_0"):
    resp = requests.post(
        "http://localhost:11434/api/generate",
        json={
            "model": model,
            "prompt": prompt,
            "stream": False,
            "options": {
                "temperature": 0.1,  # 決定論的に生成したいので低めに設定
                "num_ctx": 4096
            }
        },
        timeout=120
    )
    resp.raise_for_status()
    sql = resp.json()["response"].strip()
    # コードブロック記号が含まれていた場合は除去
    if "```" in sql:
        lines = sql.split("\n")
        sql = "\n".join(
            l for l in lines
            if not l.strip().startswith("```")
        )
    return sql.strip()

import re

DANGEROUS = re.compile(
    r"\b(INSERT|UPDATE|DELETE|DROP|CREATE|ALTER|TRUNCATE|"
    r"GRANT|REVOKE|EXECUTE|COPY|VACUUM)\b",
    re.IGNORECASE
)

def validate_sql(sql):
    if not sql.strip().upper().startswith("SELECT"):
        return False, "SELECT文で始まっていません"
    if DANGEROUS.search(sql):
        m = DANGEROUS.search(sql)
        return False, f"危険なキーワードが含まれています: {m.group()}"
    # 末尾以外のセミコロン（複数クエリ）を検出
    if ";" in sql.rstrip(";"):
        return False, "複数のクエリが含まれています"
    return True, "OK"

def run_text_to_sql(question):
    conn = psycopg2.connect(
        dbname="bizdb", user="bizreader",
        password="reader_pw", host="localhost"
    )
    schema_ddl = get_schema_ddl(conn)
    prompt = build_prompt(schema_ddl, question)
    sql = generate_sql(prompt)

    ok, reason = validate_sql(sql)
    if not ok:
        conn.close()
        return {"error": reason, "sql": sql, "rows": []}

    cur = conn.cursor()
    try:
        cur.execute(sql)
        columns = [desc[0] for desc in cur.description]
        rows = cur.fetchall()
    except psycopg2.Error as e:
        conn.rollback()
        conn.close()
        return {"error": str(e.pgerror), "sql": sql, "rows": []}

    conn.close()
    return {"sql": sql, "columns": columns, "rows": rows}


# 実行例
result = run_text_to_sql(
    "東京都の顧客の注文を合計金額が高い順に見せてください"
)
print("生成SQL:", result["sql"])
for row in result.get("rows", []):
    print(row)

# 実行例の出力
生成SQL: SELECT c.name, o.product_name, o.amount
         FROM customers c
         JOIN orders o ON c.id = o.customer_id
         WHERE c.prefecture = '東京都'
         ORDER BY o.amount DESC LIMIT 20;

('株式会社アルファ', 'クラウド移行支援', 320000)
('株式会社アルファ', 'Linux研修パッケージ', 198000)

FEW_SHOT = """
質問: 全顧客の一覧を見せてください
SQL: SELECT id, name, prefecture, created_at FROM customers ORDER BY id;

質問: 2026年8月の注文件数を教えてください
SQL: SELECT count(*) AS order_count FROM orders
     WHERE order_date >= '2026-08-01' AND order_date < '2026-09-01';

質問: 顧客ごとの合計注文金額を高い順に見せてください
SQL: SELECT c.name, sum(o.amount) AS total_amount
     FROM customers c
     JOIN orders o ON c.id = o.customer_id
     GROUP BY c.name ORDER BY total_amount DESC;
"""

def build_prompt(schema_ddl, question):
    return f"""（上記プロンプトのスキーマ部分は同じ）

参考例:
{FEW_SHOT}

質問: {question}
SQL:"""

$ sudo -u postgres psql -d bizdb

bizdb=# COMMENT ON COLUMN orders.amount IS '注文金額（税込、円単位）';
bizdb=# COMMENT ON COLUMN customers.prefecture IS '都道府県名（例: 東京都、大阪府）';
bizdb=# \q

try:
    cur.execute(sql)
    rows = cur.fetchall()
except psycopg2.Error as e:
    conn.rollback()
    return {"error": f"SQL実行エラー: {e.pgerror}", "sql": sql, "rows": []}

# systemdのoverride.confでOllama環境変数を設定
$ sudo systemctl edit ollama

# 追加するセクション
[Service]
Environment="OLLAMA_NUM_PARALLEL=1"
Environment="OLLAMA_MAX_LOADED_MODELS=1"

$ sudo systemctl daemon-reload && sudo systemctl restart ollama

Text-to-SQLとローカルLLM活用を2日間のハンズオンで体験する

OllamaによるText-to-SQL構築から、pgvectorを使ったRAGパイプラインの実装まで、本記事で解説した内容を実機のGPUサーバーで試したい方向けに、「ローカルAIマスターセミナー」を開催しています。
少人数（最大8名）ZOOMハンズオン形式で実施しています。