LLMにDBを直接操作させてText-to-SQLを検証してみる

はじめに

「日本語で質問するだけでSQLを自動生成してくれる」機能は、最近いろんなツールやサービスで見かけるようになりました。

実務でText-to-SQLを検討するなら、もう一歩進んでAIがDBに実際に接続し、実行結果を見ながら自分で修正していくエージェント的な使い方のほうが実用に近いはずです。

ただ、AIにDBを直接触らせるのは、率直に言って怖いです。誤ったSQLでデータを壊されるリスクはもちろん、SELECTだけに絞ったとしても、もっともらしい誤答を自信満々に返してくる可能性は十分にあります。

そこで、「どこまで安全に運用できるか、どこで間違えるのかを実際に検証してから判断する」というスタンスでこの記事を書くことにしました。read-only接続やcross_checkのような安全策を組み込んだ最小限のツールセットを実装し、実際にLLM APIを繋いで動かしてみた検証結果をまとめてみました。

結果、事前には想定していなかった「言語の壁」という具体的な盲点が見つかりました。

実装環境

項目	内容
使用モデル	gpt-5.6-luna（OpenAI互換API経由）
実行環境	Python 3 ＋ sqlite3 標準ライブラリ、API呼び出しに urllib
DB	SQLite（shop.db、read-only接続）
実行モード	対話入力/引数指定 → LLMがツールを自律的に呼び出すエージェントループ

サンプルDBは顧客・商品・注文の3テーブル構成です。ここが今回の発見に直結するのですが、顧客の都道府県（prefecture）や商品カテゴリ（category）のデータは英語表記（Tokyo、electronics など）で格納されています。

テーブル構成とデータ

setup_db.py で作成しているスキーマです。

CREATE TABLE customers (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    prefecture TEXT,
    joined_at TEXT NOT NULL
);
CREATE TABLE products (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    category TEXT NOT NULL,
    price INTEGER NOT NULL
);
CREATE TABLE orders (
    id INTEGER PRIMARY KEY,
    customer_id INTEGER,
    product_id INTEGER NOT NULL,
    quantity INTEGER NOT NULL,
    ordered_at TEXT NOT NULL,
    FOREIGN KEY(customer_id) REFERENCES customers(id),
    FOREIGN KEY(product_id) REFERENCES products(id)
);

投入している実データです。prefecture と category が英語表記になっている点、customers の4番（Diana）は prefecture がNULL、orders の10番は customer_id がNULLになっている点が、今回の発見に関わってきます。

customers

id	name	prefecture	joined_at
1	Alice	Tokyo	2023-01-10
2	Bob	Osaka	2023-02-14
3	Carol	Hokkaido	2023-03-20
4	Diana	(NULL)	2023-04-05
5	Evan	Tokyo	2023-05-01
6	Frank	Fukuoka	2023-06-11

products

id	name	category	price
1	Laptop	electronics	120000
2	Mouse	electronics	3000
3	Chair	furniture	15000
4	Desk	furniture	40000
5	Bottle	lifestyle	2000

orders

id	customer_id	product_id	quantity	ordered_at
1	1	1	1	2024-03-02
2	1	2	2	2024-03-10
3	2	3	1	2024-03-15
4	2	2	1	2024-04-01
5	3	5	3	2024-03-22
6	3	2	1	2024-04-11
7	5	4	1	2024-02-18
8	6	3	1	2024-03-05
9	6	3	2	2024-05-05
10	(NULL)	1	1	2024-03-12
11	1	5	4	2024-06-01
12	2	4	1	2024-03-29

実装：最小限のツールセット

DB操作の土台になる SQLAgent クラスです。read-only接続にしている点がポイントで、「SELECTだけ許可する」という制約を文字列パターンではなくDB接続そのものの権限に委ねています。

import sqlite3
from typing import Any


class SQLAgent:
    """スキーマ確認とSQL実行を行う最小ツールセット。

    read_only=True (既定) の場合、SQLiteのURI接続で読み取り専用モードを
    指定する。「SELECT/CTEのみ許可」を文字列パターンで判定するのは
    抜け漏れが起きやすい(例: `WITH x AS (SELECT 1) DELETE FROM t;` は
    "with "始まりの文字列チェックだけでは弾けない)。そのため一次防御は
    「文字列の形」ではなく「DB接続そのものの権限」に置く。read_only接続に
    書き込み系SQLを投げると sqlite3.OperationalError
    ("attempt to write a readonly database") で確実に拒否される。
    """

    def __init__(self, db_path: str, read_only: bool = True):
        if read_only:
            uri = f"file:{db_path}?mode=ro"
            self.conn = sqlite3.connect(uri, uri=True)
        else:
            self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()

    def list_tables(self) -> list[tuple[str, str]]:
        """sqlite_master からテーブル名とDDLを取得する。"""
        self.cursor.execute("SELECT name, sql FROM sqlite_master WHERE type='table';")
        return self.cursor.fetchall()

    def execute(self, sql: str) -> dict[str, Any]:
        """SQLを実行し、成功/失敗を辞書形式で返す。

        sqlite3.Cursor.execute() は元々単一ステートメントしか受け付けない
        ため、セミコロン区切りの複数文注入(`SELECT 1; DROP TABLE x;`)も
        ここで自然にエラーになる。
        """
        try:
            self.cursor.execute(sql)
            if self.cursor.description is None:
                return {"ok": True, "columns": [], "rows": []}
            columns = [d[0] for d in self.cursor.description]
            rows = self.cursor.fetchall()
            return {"ok": True, "columns": columns, "rows": rows}
        except sqlite3.Error as e:
            return {"ok": False, "error": str(e)}

    def close(self) -> None:
        self.conn.close()

cross_check は、実装方針の異なる2つのSQLを実行して結果行を突き合わせるだけの、シンプルな関数です。「不一致」がバグなのか単なる解釈違いなのかまでは判定せず、判断は呼び出し側（今回はLLM自身）に委ねています。

def cross_check(agent: SQLAgent, sql_a: str, sql_b: str, label: str = "task") -> dict[str, Any]:
    """方針の異なる2つのSQLを実行し、結果行の一致を判定する。

    「不一致」は、実装のバグと単なる解釈違いのどちらの場合もあり得る。
    このため最終判断は人間、もしくは呼び出し側の追加ロジックに委ねる。
    """
    result_a = agent.execute(sql_a)
    result_b = agent.execute(sql_b)
    if not result_a["ok"] or not result_b["ok"]:
        return {"label": label, "match": False, "result_a": result_a, "result_b": result_b}
    match = result_a["rows"] == result_b["rows"]
    return {"label": label, "match": match, "result_a": result_a, "result_b": result_b}

これらのツールをLLMに渡す部分です。渡しているのは4つだけで、「次に何をするか」の判断はコード側では決め打ちにせず、すべてLLMに委ねています。

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "list_tables",
            "description": "利用可能なテーブルのDDL(スキーマ定義)を取得する。質問に答える前にスキーマを把握したい場合に呼ぶ。",
            "parameters": {"type": "object", "properties": {}, "required": []},
        },
    },
    {
        "type": "function",
        "function": {
            "name": "execute_sql",
            "description": "SQLiteに対してSELECT文(またはWITH句によるCTE)を実行し、結果を返す。",
            "parameters": {
                "type": "object",
                "properties": {"sql": {"type": "string", "description": "実行するSELECT文"}},
                "required": ["sql"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "cross_check",
            "description": (
                "同じ質問に対する実装方針の異なる2つのSELECT文を実行し、"
                "結果が一致するか確認する。NOT INのようなNULLが絡む否定条件や、"
                "集計の粒度・指標の定義に自信が持てない場合に使うと安全性が上がる。"
            ),
            "parameters": {
                "type": "object",
                "properties": {"sql_a": {"type": "string"}, "sql_b": {"type": "string"}},
                "required": ["sql_a", "sql_b"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "final_answer",
            "description": "十分な情報が集まったら、最終回答とその根拠SQLを提示して終了する。",
            "parameters": {
                "type": "object",
                "properties": {
                    "answer": {"type": "string", "description": "日本語での回答"},
                    "final_sql": {"type": "string", "description": "回答の根拠になった最終的なSELECT文"},
                },
                "required": ["answer", "final_sql"],
            },
        },
    },
]

SYSTEM_PROMPT = (
    "あなたはSQLiteのText-to-SQLエージェントです。"
    "ツール(list_tables, execute_sql, cross_check)を使ってユーザーの質問に答えてください。"
    "スキーマが分からない場合はまず list_tables を呼んでください。"
    "実行エラーが出た場合は、エラー内容を踏まえて execute_sql を再度呼び直してください。"
    "NOT INのようなNULLが絡みうる否定条件や、集計の粒度・指標の定義が曖昧な質問では、"
    "cross_check で2通りの実装を突き合わせてから回答してください。"
    "十分な確認が取れたら、必ず final_answer ツールを呼んで終了してください。"
    "final_answer 以外の形で自然文だけを返すことは禁止です。"
)

ツール呼び出しを実際に処理し、モデルとのやり取りをループさせる部分です。モデルが tool_calls を返す限りツールを実行して結果を返し続け、final_answer が呼ばれたら終了します。

def run(user_question: str, max_steps: int) -> int:
    client = OpenAICompatibleClient(api_key=api_key, model=model, base_url=base_url)
    agent = SQLAgent("shop.db", read_only=True)

    messages: list[dict] = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_question},
    ]

    for step in range(1, max_steps + 1):
        message = client.chat(messages, TOOLS)
        tool_calls = message.get("tool_calls")

        if not tool_calls:
            # ツールを呼ばず自然文だけ返してきた場合は、final_answer利用を再依頼
            messages.append({"role": "assistant", "content": message.get("content", "")})
            messages.append({"role": "user", "content": "必ず final_answer ツールを使って回答してください。"})
            continue

        messages.append({"role": "assistant", "content": message.get("content"), "tool_calls": tool_calls})

        for tool_call in tool_calls:
            name = tool_call["function"]["name"]
            args = json.loads(tool_call["function"]["arguments"] or "{}")

            if name == "final_answer":
                print(f"回答: {args.get('answer')}")
                print(f"根拠SQL: {args.get('final_sql')}")
                return 0

            result = call_tool(agent, name, args)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call["id"],
                "content": json.dumps(result, ensure_ascii=False),
            })

実際に動かした結果

12個のテストケースを実行しました。

ID	質問	結果概要	判定
TC01	東京都の顧客の名前を教えて	WHERE prefecture = '東京都' で0件ヒット→「見つかりませんでした」と回答	NG（誤答）
TC02	商品カテゴリごとの売上合計金額を教えて	electronics 252,000円 / furniture 140,000円 / lifestyle 14,000円	OK
TC03	2024年3月の注文件数を教えて	COUNT(*) = 7	OK
TC04	一度も注文していない顧客を教えて	NOT EXISTS と LEFT JOIN で cross_check、両方とも Diana のみ、match=True	OK
TC05	平均注文金額を教えて	2実装とも「注文ID単位の合計を平均」で 33,833.33円、match=True	OK（曖昧さは今回顕在化せず）
TC06	よく買われている都道府県を教えて	数量ベースで Tokyo(8) > Hokkaido(4) > Osaka(3) = Fukuoka(3)、match=True	OK
TC07	家電を2回以上買った東京都の顧客だけ教えて	prefecture='東京都' AND category='家電' で0件、match=Trueだが両方0件	NG（誤答疑い）
TC08	プロテインを買った人が他に何を買っているか	LIKE '%プロテイン%' で対象顧客なし、match=Trueで「該当データなし」	OK（実データに該当商品なしのため妥当）
TC09	各都道府県で一番人気の商品カテゴリを教えて	数量ベースと注文行数ベースで match=False。両論併記で回答	OK
TC10	都道府県ごとのリピート顧客の割合を教えて	2回以上注文の顧客割合を都道府県別に集計、match=True	OK
TC11（3回）	TC04を同一条件で3回実行	3回ともDianaのみで結果一致	OK

見つかった発見

発見1：日本語リテラル値と実データ表記の不一致

TC01・TC07で、質問文中の日本語（東京都、家電）をそのままSQLのリテラル値として使い、実データが英語表記（Tokyo、electronics）であることに気づかず、「0件＝該当なし」と誤答しました。
まさかこんな初歩的なところで躓くとは…

-- TC01で生成されたSQL(誤り)
SELECT name FROM customers WHERE prefecture = '東京都';
-- 実データは 'Tokyo' なので常に0件

これは、SQLのロジック自体は正しく書けているのに、「質問文の言語」と「DBの値の言語」がそもそも噛み合っていないという、SQL構文の外側にある問題です。

さらに厄介なのは、execute_sql が0件を返してもエラーにはならないため、エージェントは値の不一致にまったく気づけないという点です。「実行が成功した＝正しい」という前提そのものが崩れるケースだと言えます。

発見2：cross_checkは「共有された誤った前提」には無力

TC07では cross_check を使って2つの実装方針を突き合わせていますが、両方とも同じ日本語リテラル（'東京都'、'家電'）を前提にしていたため、両方とも0件を返し、match=True（一致）と判定されました。

「一致した」という結果は、あくまで「2つの実装が同じ判断をした」ことの証明であって、「その判断が正しい」ことの証明にはならない、という点がここで明確になりました。実装を2通り作って突き合わせる仕組みは、2つの実装が独立に間違いを犯した場合には有効ですが、同じ思い込みを共有している場合はすり抜けます。

発見3：曖昧さの検出はエージェントの「気まぐれ」に依存する

TC09（「人気」の定義：数量ベースか注文件数ベースか）では、2つの実装が異なる集計軸を選んだ結果 match=False となり、エージェントは断定を避けて両方の定義とその結果を回答に含めました。これは望ましい挙動です。

一方でTC05（平均注文金額）は、「注文明細1行あたりの平均」なのか「顧客1人あたりの合計購入額の平均」なのかで本来は解釈が割れうる質問でしたが、今回の実行では2つの実装がたまたま同じ粒度（注文ID単位の合計の平均）を選んでしまい、曖昧さが顕在化しませんでした。

つまり cross_check による曖昧さの検出は、エージェントが実際に異なる解釈を選んでくれるかどうかという偶然性に左右されることがわかりました。

発見4：非決定性は今回は小さかった

TC04を同一条件で3回実行したところ、生成されたSQL文面には多少の揺らぎがあったものの、最終的な回答内容（該当顧客・件数）は3回とも一致しました。今回の範囲では、非決定性による結果のブレは大きな問題にはなりませんでした。

発見5：自己修正ループは発火しなかった

今回の実行ログでは、実行時エラー自体が一度も発生しませんでした。「存在しないカラム名を書く→エラー→修正」のような自己修正の流れが実際に働く場面を観測できなかったのは、裏を返せば今回のスキーマ・質問セットの範囲では、モデルが構文レベルのミスをほぼ起こさなかったということでもあります。

わかったこと

• 実際にLLM APIを繋いで動かしてみると、「質問文の言語」と「DBの値表現」の不一致という、事前に想定していなかった種類のバグが最も目立つ問題として見つかった
• cross_check は意味的バグの検知に有効だが、2つの実装が同じ誤った前提を共有していると意味が無い。これは今回、日本語/英語の表記ゆれという具体的な形で実証された
• 曖昧さの検出も cross_check に頼る場合、エージェントが実際に異なる解釈を選ぶかどうかに依存するため、確実に機能するとは限らない
• 対策として、execute_sql が0件を返した際に SELECT DISTINCT <カラム> のような形で実際の値を確認させる仕組みを追加する必要がありそう

おわりに

LLMにDBを直接操作させるエージェントを組んでみると、「SQLの構文として正しいかどうか」だけでは見えてこない問題が浮かび上がってきます。今回一番大きかったのは、質問の言葉とデータの言葉がそもそも噛み合っていないという、SQLのロジック以前の問題でした。cross_check のような検証の仕組みを用意しても、2つの実装が同じ誤った前提の上に立っていれば意味がない、という点も実感できました。

次のステップとしては、0件ヒット時に実データの値を確認しにいく仕組みを execute_sql の周りに追加し、同じテストケースで効果があるか検証してみたいと思います。