はじめに

AIエンジニアリングを学習中のnoriです。

普段はSQLを使っているのですが、「AIの学習をしてみたい」と思い、何か作れそうなものはないかということで自然言語でSQLを生成・実行するAPIツールを作ってみました。

この記事は完璧な技術解説というより、学習記録 + 実装メモとして書いています。詰まったところや設計判断の理由も正直に書いているので、同じような学習をしている方の参考になれば嬉しいです。

作ったもの

こんな感じで動きます。

curl -X POST http://127.0.0.1:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "売上金額が一番高い商品は何ですか？"}'

{
  "question": "売上金額が一番高い商品は何ですか？",
  "sql": "SELECT p.name FROM products p JOIN sales s ON p.id = s.product_id GROUP BY p.id, p.name ORDER BY SUM(s.revenue) DESC LIMIT 1;",
  "query_result": "name\nぶどう\n",
  "answer": "売上金額が一番高い商品はぶどうです。"
}

自然言語の質問から自動でSQLを生成・実行して、結果を自然言語で回答してくれます。レスポンスに生成されたSQLも含めているので、「AIが何をしたか」を人間がレビューできる設計にしました。

技術スタック

役割	技術
ワークフロー制御	LangGraph
LLM	Claude（claude-haiku-4-5）
DB	SQLite
APIサーバー	FastAPI
環境管理	uv + Python 3.12

LLMにはAnthropicのClaude Haikuを使っています。安くて速いので開発中の試行錯誤に向いています。

アーキテクチャ

ユーザーの質問（自然言語）
        ↓
[generate_sql]  自然言語 → SQLを生成
        ↓
[execute_sql]   SQLを実行
        ↓
  SQLエラー？
  ↓ Yes（エラーメッセージをプロンプトに含めて再生成）
[generate_sql]  SQLを再生成（最大2回）
  ↓ No
[generate_answer]  結果を自然言語で回答
        ↓
    レスポンス

LangGraphを使ってワークフローをグラフとして定義しています。ポイントはSQLエラー時のリトライです。LLMが生成したSQLがエラーになった場合、エラーメッセージをプロンプトに含めて再生成することで自己修正できます。

実装解説

ディレクトリ構成

src/sql_assistant/
  database.py  → SQLiteのDB作成・クエリ実行
  graph.py     → LangGraphのワークフロー定義
  api.py       → FastAPIのエンドポイント
data/
  sales.db     → SQLiteのDBファイル

1. DBの準備（database.py）

SQLiteでproductsとsalesテーブルを作成してダミーデータを投入しています。

def get_schema() -> str:
    """テーブルのスキーマ情報を返す"""
    return """
テーブル: products
- id: INTEGER (主キー)
- name: TEXT (商品名)
- category: TEXT (カテゴリ: 果物/野菜)
- price: INTEGER (単価)

テーブル: sales
- id: INTEGER (主キー)
- product_id: INTEGER (外部キー -> products.id)
- quantity: INTEGER (販売数量)
- revenue: INTEGER (売上金額)
- sale_date: TEXT (販売日 YYYY-MM-DD形式)
"""

スキーマ情報をLLMに渡すことで、正しいカラム名でSQLを生成できるようにしています。

2. LangGraphのワークフロー（graph.py）

Stateの定義から始めます。グラフ全体で共有するデータです。

class SQLAssistantState(TypedDict):
    question: str
    sql: str
    query_result: str
    answer: str
    error_count: int

SQL生成ノードでは、前回のエラーメッセージもプロンプトに含めます。

def generate_sql_node(state: SQLAssistantState) -> SQLAssistantState:
    response = llm.invoke([
        HumanMessage(content=f"""以下のスキーマを参考にSQLを生成してください。
SQLのみ返してください。

スキーマ:
{get_schema()}

質問: {state['question']}
前回のエラー（あれば）: {state['query_result']}
""")
    ])
    sql = response.content.strip()
    sql = sql.replace("```sql", "").replace("```", "").strip()
    return {**state, "sql": sql}

replace("```sql", "") でLLMが返すコードブロックを除去しています。これをやらないとSQLとして実行できません。

条件分岐はこうなっています。

def should_retry(state: SQLAssistantState) -> str:
    if "SQLエラー" in state["query_result"] and state["error_count"] < 2:
        return "retry"
    return "answer"

3. 詰まったところ：LLMのSQL自己修正

実際に動かしてみると、最初はこんなSQLが生成されました。

-- 1回目（エラー）
SELECT p.name FROM products p
JOIN sales s ON p.id = s.products_id  -- ← products_id（存在しないカラム）
ORDER BY s.revenue DESC LIMIT 1;

products_id というカラムは存在しないのでエラーになります。でもエラーメッセージをプロンプトに含めてリトライすると：

-- 2回目（成功）
SELECT p.name FROM products p
JOIN sales s ON p.id = s.product_id  -- ← product_id（正しい）
ORDER BY s.revenue DESC LIMIT 1;

自己修正できました。LLMはエラーメッセージを見て何が間違っているかを判断して修正してくれます。これはLangGraphのリトライループの効果です。

4. FastAPIでAPI化（api.py）

@app.post("/query", response_model=QueryResponse)
def query(request: QueryRequest):
    result = graph.invoke({
        "question": request.question,
        "sql": "",
        "query_result": "",
        "answer": "",
        "error_count": 0,
    })
    return QueryResponse(
        question=result["question"],
        sql=result["sql"],
        query_result=result["query_result"],
        answer=result["answer"],
    )

シンプルです。graph.invoke() にStateの初期値を渡すだけで、あとはLangGraphが処理してくれます。

設計判断とトレードオフ

学習の一環として、設計判断の理由も記録しています。

なぜSQLiteを使ったか

本番ではPostgreSQLを使いますが、今回は学習目的なのでSQLiteにしました。ファイルベースなのでサーバー不要で手軽に始められます。sqlite3.connect() を psycopg2.connect() に変えるだけで本番DBに切り替えられます。

なぜレスポンスにSQLを含めたか

「AIが生成したSQLを人間がレビューできるようにする」という設計です。案件で「AIの判断を透明にしてほしい」という要件はよく来ます。SQLが見えることで、誤ったクエリが実行されていないか確認できます。

リトライは最大2回にした理由

LLMの生成は非決定的なので、同じエラーが繰り返されることがあります。無限ループを防ぐために上限を設けています。複雑なSQLは2回では足りないこともありますが、まず動くものを作ることを優先しました。

まとめ

LangGraph + FastAPIでSQL生成AIツールを作りました。

学んだことをまとめると：

LangGraphは「条件分岐・リトライ」のある複雑なワークフローを直感的に書ける
LLMへのプロンプトにスキーマ情報を含めることでSQL精度が上がる
エラーメッセージをプロンプトに含めてリトライすることでLLMが自己修正できる
FastAPIの /docs で自動生成されるAPIドキュメントが便利

今後はフロントエンド（React）を追加して、ブラウザから自然言語でBIクエリを投げられるツールにする予定です。

参考

この記事が同じような学習をしている方の参考になれば嬉しいです！