到達点：LangGraphでNL→SQLフローを構築し、
- PostgreSQLチェックポイントで状態を保存
- エラー処理（SELECT限定・データセット固定・多文拒否）
- 自動再試行（エラーメッセージを使ってSQLをリペア）
までを実装します。
（Part 1/2のMVPを正式フローに差し替え）
追加依存関係の追加
backend/requirements.txt
langgraph>=0.1.0
langgraph-checkpoint-postgres>=0.1.0
psycopg[binary,pool]>=3.2
pydantic>=2.7
openai>=1.40.0
環境変数（例）：
.env
OPENAI_API_KEY=sk-***
OPENAI_MODEL=gpt-4o-mini # 任意のモデル名を指定
CHECKPOINT_NS=text2sql # 既出（Part 1/2と同じ命名でOK）
1. チェックポイント（PostgreSQL）を用意する
LangGraphは**状態（State）**をDBに保存できます。最初にテーブルを自動作成します。
langgraph/checkpointer.py
from __future__ import annotations
import os
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg import connect
from psycopg.rows import dict_row
DB_URI = os.getenv("DATABASE_URL") # postgresql://user:pass@host:5432/sqlbot
def build_checkpointer() -> PostgresSaver:
with connect(DB_URI, autocommit=True, row_factory=dict_row) as conn:
saver = PostgresSaver(conn)
saver.setup() # 初回テーブル作成
return saver
代表テーブル（バージョンで多少異なることあり）：checkpoint_migrations, checkpoint_writes, checkpoint_blobs など。
ここに スレッド（thread_id） ごとの進行状況が保存され、再実行・再試行が安定します。
2. スキーマキャッシュをDBに保存
NL→SQLの精度を上げるため、INFORMATION_SCHEMAから取得したテーブル/カラム情報をJSONBでキャッシュします（TTLつき）。
db/migrations/schema_cache.sql
CREATE TABLE IF NOT EXISTS bq_schema_cache (
project_id TEXT NOT NULL,
dataset TEXT NOT NULL,
table_name TEXT NOT NULL,
columns JSONB NOT NULL, -- [{name,type,description?}, ...]
fetched_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (project_id, dataset, table_name)
);
bq/schema_cache.py
from __future__ import annotations
import json, time
from sqlalchemy import text
from bq.client import run_query # ← 絶対インポートに統一
TTL_SEC = 6 * 3600 # 6時間
def get_table_schema(db, project: str, dataset: str, table: str):
row = db.execute(
text("""SELECT columns, fetched_at
FROM bq_schema_cache
WHERE project_id=:p AND dataset=:d AND table_name=:t"""),
{"p": project, "d": dataset, "t": table},
).fetchone()
if row and (time.time() - row.fetched_at.timestamp()) < TTL_SEC:
return row.columns
headers, rows = run_query(f"""
SELECT column_name, data_type
FROM `{project}.{dataset}.INFORMATION_SCHEMA.COLUMNS`
WHERE table_name = '{table}'
ORDER BY ordinal_position
""", project=project)
cols = [{"name": r[0], "type": r[1]} for r in rows]
db.execute(
text("""INSERT INTO bq_schema_cache(project_id,dataset,table_name,columns,fetched_at)
VALUES (:p,:d,:t,:c, now())
ON CONFLICT (project_id,dataset,table_name)
DO UPDATE SET columns=:c, fetched_at=now()"""),
{"p": project, "d": dataset, "t": table, "c": json.dumps(cols)},
)
db.commit()
return cols
最初は必要テーブルだけ（例：orders）で十分。後で自動収集・pgvector拡張も可能。
全テーブルを自動読み込みする場合はschema_loader.pyなどを実装して読み込み専用処理を定期実行するようにして対応する。
3. LLMクライアント（最小）
langgraph/llm.py
from __future__ import annotations
import os
from openai import OpenAI
_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
_client = OpenAI()
def complete_system_user(system: str, user: str) -> str:
resp = _client.chat.completions.create(
model=_MODEL,
messages=[{"role": "system", "content": system},
{"role": "user", "content": user}],
temperature=0.2,
)
return resp.choices[0].message.content or ""
4. エラー処理（ガード強化）
- SELECT限定
- 多文（;区切り）拒否
- データセット外アクセス拒否（project.dataset.table が許可DSかチェック）
- LIMIT補完（存在しなければ LIMIT 100 を末尾に付与）
langgraph/guards.py
from __future__ import annotations
import re
_FORBIDDEN = re.compile(r"\b(INSERT|UPDATE|DELETE|MERGE|TRUNCATE|DROP|ALTER|CREATE|GRANT|REVOKE)\b", re.I)
def enforce_sql_policy(sql: str, allowed_dataset: str) -> str:
s = sql.strip().rstrip(";")
if not re.match(r"(?s)^\s*SELECT\b", s, re.I):
raise ValueError("Only SELECT is allowed.")
if _FORBIDDEN.search(s):
raise ValueError("Dangerous SQL detected.")
if ";" in sql:
raise ValueError("Multiple statements are not allowed.")
def _qualify(m):
full = m.group(1)
parts = full.split(".")
if len(parts) == 1:
return m.group(0).replace(full, f"{allowed_dataset}.{full}")
ds = parts[-2] if len(parts) >= 2 else ""
if ds.lower() != allowed_dataset.lower():
raise ValueError(f"Access to dataset '{ds}' is not allowed.")
return m.group(0)
s = re.sub(r"\bFROM\s+([a-zA-Z_][\w\.]*)\b", _qualify, s, flags=re.I)
s = re.sub(r"\bJOIN\s+([a-zA-Z_][\w\.]*)\b", _qualify, s, flags=re.I)
if not re.search(r"\bLIMIT\s+\d+\b", s, re.I):
s = f"{s}\nLIMIT 100"
return s
5. グラフの状態とノードを定義する
5-1. State
langgraph/state.py
from __future__ import annotations
from pydantic import BaseModel, Field
from typing import List, Optional, Tuple
class NL2SQLState(BaseModel):
question: str
project_id: str
dataset: str
location: str
schema_hint: Optional[str] = None
sql: Optional[str] = None
headers: Optional[List[str]] = None
rows: Optional[List[Tuple]] = None
error: Optional[str] = None
retries: int = Field(default=0, ge=0)
5-2. プロンプト
langgraph/prompt.py
SYSTEM_PROMPT = """あなたはBigQuery専用のSQLアシスタントです。
- 返答はSQLのみ（説明文なし）
- DML/DDLは禁止（SELECTのみ）
- 必ず `{dataset}.table` の形式で参照
- 末尾に LIMIT を付ける（多すぎる結果を避ける）
- タイムゾーンはUTC基準で良い（必要ならDATE()で日付化）
"""
USER_TEMPLATE = """自然文:
{question}
利用可能なテーブルと主なカラム（参考）:
{schema_hint}
制約:
- dataset={dataset}
- SELECTのみ、1文のみ
- 結果はSQLだけを返す
"""
5-3. ノード（promptの {dataset} を format して渡す）
langgraph/nodes.py
from __future__ import annotations
from sqlalchemy.orm import Session
from .state import NL2SQLState
from .prompt import SYSTEM_PROMPT, USER_TEMPLATE
from .llm import complete_system_user
from bq.client import run_query
from .guards import enforce_sql_policy
from bq.schema_cache import get_table_schema
def build_schema_hint(db: Session, project: str, dataset: str, examples=("orders",)) -> str:
lines = []
for tbl in examples:
cols = get_table_schema(db, project, dataset, tbl)
if not cols:
continue
left = ", ".join(f"{c['name']}:{c['type']}" for c in cols[:8])
tail = " ..." if len(cols) > 8 else ""
lines.append(f"- {tbl}({left}{tail})")
return "\n".join(lines) if lines else "(スキーマ情報は最小です)"
def node_collect_schema(db: Session):
def _fn(state: NL2SQLState) -> NL2SQLState:
state.schema_hint = build_schema_hint(db, state.project_id, state.dataset, examples=("orders",))
return state
return _fn
def node_llm_generate_sql(state: NL2SQLState) -> NL2SQLState:
system = SYSTEM_PROMPT.format(dataset=state.dataset) # ← ここ重要
user = USER_TEMPLATE.format(
question=state.question,
schema_hint=state.schema_hint or "(なし)",
dataset=state.dataset
)
sql = complete_system_user(system, user).strip()
state.sql = sql
state.error = None
return state
def node_validate_and_execute(state: NL2SQLState) -> NL2SQLState:
try:
safe_sql = enforce_sql_policy(state.sql or "", state.dataset)
headers, rows = run_query(
safe_sql,
project=state.project_id,
location=state.location
)
state.headers, state.rows = headers, rows
state.error = None
state.sql = safe_sql
return state
except Exception as e:
state.error = str(e)
return state
def node_llm_repair_sql(state: NL2SQLState) -> NL2SQLState:
user = f"""次のSQLを修正して。エラー:\n{state.error}\n\nSQL:\n{state.sql}"""
fixed = complete_system_user(
"あなたはSQL修復アシスタントです。BigQuery方言で1文のSELECTだけを返す。",
user
).strip()
state.sql = fixed
return state
6. グラフ（DBセッション生成の簡素化）
langgraph/graph.py
from __future__ import annotations
from langgraph.graph import StateGraph, START, END
from .state import NL2SQLState
from .nodes import node_collect_schema, node_llm_generate_sql, node_validate_and_execute, node_llm_repair_sql
from .checkpointer import build_checkpointer
MAX_RETRIES = 2
def build_graph(db_session_factory):
builder = StateGraph(NL2SQLState)
def _collect(state: NL2SQLState):
with db_session_factory() as db:
return node_collect_schema(db)(state)
builder.add_node("collect_schema", _collect)
builder.add_node("draft_sql", node_llm_generate_sql)
builder.add_node("exec", node_validate_and_execute)
builder.add_node("repair", node_llm_repair_sql)
builder.add_edge(START, "collect_schema")
builder.add_edge("collect_schema", "draft_sql")
builder.add_edge("draft_sql", "exec")
def _route_on_error(state: NL2SQLState):
if state.error is None:
return END
if state.retries >= MAX_RETRIES:
return END
state.retries += 1
return "repair"
builder.add_conditional_edges("exec", _route_on_error)
builder.add_edge("repair", "exec")
checkpointer = build_checkpointer()
return builder.compile(checkpointer=checkpointer)
7. ハンドラから呼び出す（Part 2差し替え）
Part 2の handle_sql() を LangGraph呼び出しに置き換えます。
thread_id と checkpoint_ns を指定するのがポイント（Slackスレッド単位で状態が残る）。
app/handler_langgraph.py
from __future__ import annotations
import os, requests
from fastapi import BackgroundTasks
from .slack import parse_slash_payload
from .formatting import format_table
from ..langgraph.graph import build_graph
from .db.session import SessionLocal # ← ここを修正
CHECKPOINT_NS = os.getenv("CHECKPOINT_NS", "text2sql")
_graph = None
def get_graph():
global _graph
if _graph is None:
_graph = build_graph(SessionLocal)
return _graph
def post_to_slack(response_url: str, text: str, in_channel: bool = True):
payload = {"response_type": "in_channel" if in_channel else "ephemeral", "text": text}
requests.post(response_url, json=payload, timeout=10)
async def handle_sql_with_langgraph(body: bytes, bg: BackgroundTasks, tenant: dict):
p = parse_slash_payload(body)
question = p["text"]
response_url = p["response_url"]
state = {
"question": question,
"project_id": tenant["project_id"],
"dataset": tenant["dataset"],
"location": tenant["location"],
}
thread_id = f"{p['channel_id']}:{p['user_id']}"
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": CHECKPOINT_NS}}
def _run():
graph = get_graph()
out = graph.invoke(state, config=config)
if out.error:
return f":warning: 失敗（{out.retries}回試行）: {out.error}\n```\n{out.sql or ''}\n```"
table = format_table(out.headers or [], out.rows or [])
return f"*SQL*\n```\n{out.sql}\n```\n*RESULT*\n{table}"
inline = os.getenv("DEV_INLINE_REPLY", "false").lower() == "true" or not response_url
if inline:
return {"response_type": "ephemeral", "text": _run()}
bg.add_task(lambda: post_to_slack(response_url, _run(), in_channel=True))
return {"response_type": "ephemeral", "text": "受け付けました（LangGraph正式版）。結果は追って返信します。"}
main.py 側は、Part 2 の handle_sql をこちらに差し替えるだけです。
app/main.py
# from .handler import handle_sql # 旧
from .handler_langgraph import handle_sql_with_langgraph as handle_sql # 新
8. 動作確認（ローカル）
curl -X POST http://127.0.0.1:8000/slack/cmd \
-H "Content-Type: application/x-www-form-urlencoded" \
--data 'team_id=T09GABUDW2X&user_id=U1&channel_id=C1&response_url=&text=今月の売上を日次で、7日移動平均も'
- 1回目：スキーマ収集 → LLMがSQL生成 → 実行 → 表で返る
- 失敗時：エラーを食わせてリペア → 再実行（最大2回）
9. 実装の意図と運用Tips
チェックポイント：Thread単位で履歴が残るので、再送や途中失敗にも強い。
禁則処理：
- SELECT以外は即拒否
- 多文禁止（;）でインジェクション耐性
- データセット外拒否でテナント越境を防止
- LIMIT補完でコスト・UI崩れを防ぐ
再試行（リペア）： - Unrecognized name や Syntax error をLLMに渡して修正
- Not found: Table 系はschema_cacheの更新で解消されることが多い
拡張余地： - bq_schema_cache にdescription/サンプル行を入れる
- pgvectorで列説明の意味検索を追加
10. まとめ
• LangGraphで正式な自然言語問い合わせ→SQLのパイプラインを構築
• PostgreSQLチェックポイントで状態/再試行を堅牢化
• エラー処理で安全性を担保しつつ、自動リペアで実用性を底上げ
追記：Dockerでの実行（Part 3 版）
docker-compose.yml
version: "3.9"
services:
postgres:
image: postgres:15
environment:
POSTGRES_USER: sqlbot
POSTGRES_PASSWORD: sqlbot
POSTGRES_DB: sqlbot
ports: ["5432:5432"]
volumes: [pgdata:/var/lib/postgresql/data]
api:
build:
context: .
dockerfile: backend/Dockerfile
depends_on: [postgres]
ports: ["8000:8000"]
env_file:
- backend/.env
environment:
DATABASE_URL: postgresql+psycopg2://sqlbot:sqlbot@postgres:5432/sqlbot
OPENAI_API_KEY: ${OPENAI_API_KEY} # ← シェル or .env から渡す
# GOOGLE_APPLICATION_CREDENTIALS: /secrets/sa.json # ADCが使えない人向け（代替）
volumes:
- ./backend:/app
- ~/.config/gcloud:/root/.config/gcloud:ro # ADC（推奨）
# - ./secrets/sa.json:/secrets/sa.json:ro # 代替: SA JSON を使う場合
restart: unless-stopped
volumes:
pgdata:
# 起動
docker compose up --build -d
# マイグレーション（Part 2 + Part 3）
docker compose exec -T postgres psql -U sqlbot -d sqlbot < db/migrations/20250929_part3_schema_cache.sql
ランディングページのご案内
プロダクト概要・導入メリット・デモイメージは、こちらのLPにまとめています。
👉 ランディングページ
前回：Part 2 へ
Part 2：Slack多テナント用の参照系テーブルとルーティング実装
次回: Part 4：Cloud Run × BigQuery の最小権限とコスト設計を体系化し、運用で詰まりがちな領域（ロケーション不一致、定額/オンデマンド、結果キャッシュ）を開設する