DB向けSQLアシスタント

プロジェクトの目標

アーキテクチャ

Step 1: Schema Tools

import psycopg

conn = psycopg.connect(DATABASE_URL)

def list_tables():
    with conn.cursor() as cur:
        cur.execute("SELECT table_name FROM information_schema.tables WHERE table_schema='public'")
        return [r[0] for r in cur.fetchall()]

def describe_table(name):
    with conn.cursor() as cur:
        cur.execute('''
            SELECT column_name, data_type FROM information_schema.columns
            WHERE table_name = %s
        ''', (name,))
        return cur.fetchall()

ステップ2：安全なrun_sqlツール

import re

def run_sql(query: str, limit: int = 100):
    q = query.strip().rstrip(';').lower()
    if not q.startswith('select'):
        return {'error': 'Only SELECT statements are allowed.'}
    if any(bad in q for bad in [' drop ', ' delete ', ' update ', ' insert ', ' alter ', ' truncate ']):
        return {'error': 'Statement contains a disallowed keyword.'}
    with conn.cursor() as cur:
        cur.execute(f'SET statement_timeout = 5000')   # 5 seconds
        cur.execute(f'SELECT * FROM ({query}) sub LIMIT {limit}')
        cols = [c.name for c in cur.description]
        rows = cur.fetchall()
        return {'columns': cols, 'rows': rows}

Tool Definitions

tools = [
    {'type': 'function', 'function': {'name': 'list_tables', 'description': 'List tables in the database', 'parameters': {'type': 'object', 'properties': {}}}},
    {'type': 'function', 'function': {'name': 'describe_table', 'description': 'Get columns of a table', 'parameters': {'type': 'object', 'properties': {'name': {'type': 'string'}}, 'required': ['name']}}},
    {'type': 'function', 'function': {'name': 'run_sql', 'description': 'Execute a SELECT query (read-only, max 100 rows, 5s timeout)', 'parameters': {'type': 'object', 'properties': {'query': {'type': 'string'}}, 'required': ['query']}}}
]
import json
print(json.dumps(tools, indent=2))

System Prompt

system = '''
You are a SQL analyst assistant for a Postgres database.
First use list_tables and describe_table to learn the schema.
Then write a single SELECT query to answer the user.
Never modify data.
After receiving results, explain them in plain language.
'''
print(system.strip())

システムプロンプトにスキーマを渡す

schema = ''
for t in list_tables():
    cols = describe_table(t)
    schema += f'{t}: {cols}\n'
system = system + f'\nSchema:\n{schema}'

読み取り専用のデータベースユーザー

PIIのマスキング

PII_COLS = {'email', 'phone'}
for row in rows:
    for i, col in enumerate(cols):
        if col in PII_COLS:
            row[i] = '[REDACTED]'

クエリコストの見積もり

会話の例

出力チャート

適切に失敗させる

すべてのクエリを監査する

SELECTだけに制限する理由

まとめ