スキーマの理解と注入
LLMのコンテキスト向けに、テーブル、カラム、リレーションなどのDBスキーマを抽出・整形します。
「スキーマの理解と注入」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
スキーマコンテキストが重要な理由
LLMはSQLの構文を知っていますが、あなたのデータベースについては何も知りません。スキーマコンテキストがなければ、テーブル名やカラム名を幻覚してしまいます。
スキーマ注入とは、DBの構造をプログラムで抽出し、すべてのプロンプトに含めることです。これにより、LLMは実際に存在するテーブル、カラム、型を把握できます。
INFORMATION_SCHEMAのクエリ
主要なリレーショナルデータベースはすべて、INFORMATION_SCHEMAを通じてメタデータを公開しています。これをクエリすれば、アプリケーションコードに触れることなく、すべてのテーブル、カラム名、データ型を取得できます。
PostgreSQL、MySQL、SQL Server、SQLiteで利用できます（一部に小さな違いがあります）。
import psycopg2
def get_schema(conn):
query = '''
SELECT table_name, column_name, data_type
FROM information_schema.columns
WHERE table_schema = 'public'
ORDER BY table_name, ordinal_position
'''
with conn.cursor() as cur:
cur.execute(query)
return cur.fetchall()テーブルごとのカラムのグループ化
INFORMATION_SCHEMAから取得した結果は、行が平坦に並んだリストです。テーブル名でグループ化すると、プロンプトに整形しやすい構造化表現を作成できます。
from collections import defaultdict
def build_schema_dict(conn):
rows = get_schema(conn)
schema = defaultdict(list)
for table_name, column_name, data_type in rows:
schema[table_name].append({
'name': column_name,
'type': data_type
})
return dict(schema)
# Result:
# {
# 'users': [{'name': 'id', 'type': 'integer'}, {'name': 'email', 'type': 'character varying'}],
# 'orders': [{'name': 'id', 'type': 'integer'}, {'name': 'user_id', 'type': 'integer'}]
# }LLMプロンプト向けのスキーマ整形
LLMはスキーマをプレーンテキストとして読み取ります。簡潔で読みやすい形式として、1行に1テーブルを記述し、括弧内にカラム名と型を記載します。
主キー（PK）と外部キー（FK）を含めると、LLMが正しいJOIN文を作成しやすくなります。
def format_schema_for_prompt(schema_dict, pk_info=None, fk_info=None):
lines = []
for table, columns in schema_dict.items():
col_parts = []
for col in columns:
label = col['name']
if pk_info and (table, col['name']) in pk_info:
label += ' PK'
if fk_info and (table, col['name']) in fk_info:
label += f' FK->{fk_info[(table, col["name"])]}'
col_parts.append(f"{label} ({col['type']})")
lines.append(f"Table {table}: {', '.join(col_parts)}")
return '\n'.join(lines)
# Output:
# Table users: id PK (integer), email (varchar), created_at (timestamp)
# Table orders: id PK (integer), user_id FK->users.id (integer), total (float)
if __name__ == '__main__':
demo_schema = {'users': [{'name': 'id', 'type': 'integer'}, {'name': 'email', 'type': 'varchar'}]}
demo_pk = {('users', 'id')}
print(format_schema_for_prompt(demo_schema, pk_info=demo_pk))
主キーと外部キーの追加
外部キーの関係は、スキーマコンテキストの中で最も重要な部分です。LLMにJOINの書き方を伝えるためです。information_schema.table_constraintsとkey_column_usageをクエリして抽出します。
def get_foreign_keys(conn):
query = '''
SELECT
kcu.table_name,
kcu.column_name,
ccu.table_name AS foreign_table,
ccu.column_name AS foreign_column
FROM information_schema.table_constraints AS tc
JOIN information_schema.key_column_usage AS kcu
ON tc.constraint_name = kcu.constraint_name
JOIN information_schema.constraint_column_usage AS ccu
ON ccu.constraint_name = tc.constraint_name
WHERE tc.constraint_type = 'FOREIGN KEY'
'''
with conn.cursor() as cur:
cur.execute(query)
return {
(row[0], row[1]): f'{row[2]}.{row[3]}'
for row in cur.fetchall()
}
if __name__ == '__main__':
class FakeCursor:
def __enter__(self): return self
def __exit__(self, *a): return False
def execute(self, query): pass
def fetchall(self):
return [('orders', 'user_id', 'users', 'id')]
class FakeConn:
def cursor(self): return FakeCursor()
fks = get_foreign_keys(FakeConn())
print('Foreign keys found:')
for (table, col), ref in fks.items():
print(f' {table}.{col} -> {ref}')
スキーマ圧縮：課題
実際のエンタープライズデータベースには、200以上のテーブルが存在することがあります。スキーマ全体を注入すると、GPT-4のコンテキストウィンドウを超え、トークンに無駄な費用がかかります。
各テーブルに20カラムある200テーブルのスキーマは、概算で40,000トークン以上になります。クエリのたびに送信するには高コストです。
def estimate_schema_tokens(schema_dict):
text = format_schema_for_prompt(schema_dict)
# Rough estimate: 1 token per 4 characters
estimated_tokens = len(text) // 4
print(f'Tables: {len(schema_dict)}')
print(f'Estimated schema tokens: {estimated_tokens}')
return estimated_tokens
# 200 tables * 15 columns * 25 chars/col = 75,000 chars = ~18,750 tokens
# Plus user question + system prompt = easily over context limitスキーマ圧縮：選択的な注入
最も効果的な圧縮戦略は、質問に関連するテーブルだけを注入することです。2段階のアプローチを使います。まず必要なテーブルをLLMに尋ね、次にそれらのスキーマだけを注入します。
def select_relevant_tables(question, all_table_names, n=5):
table_list = ', '.join(all_table_names)
prompt = f'''Database tables: {table_list}
Question: {question}
List the {n} most relevant table names as a JSON array.
Example: ["users", "orders", "products"]'''
response = llm_call(prompt)
import json
return json.loads(response)
def compressed_schema(question, conn):
all_tables = list(build_schema_dict(conn).keys())
relevant = select_relevant_tables(question, all_tables)
full_schema = build_schema_dict(conn)
return {t: full_schema[t] for t in relevant if t in full_schema}スキーマ圧縮：不要なカラムの除外
多くのテーブルには、created_at、updated_at、deleted_at、version、created_byのような監査用カラムがあります。これらは業務クエリに関係しないことが多いため、除外してトークン数を削減します。
AUDIT_COLUMNS = {
'created_at', 'updated_at', 'deleted_at', 'created_by',
'updated_by', 'version', 'is_deleted', 'modified_at'
}
def compress_schema(schema_dict, exclude_audit=True):
compressed = {}
for table, columns in schema_dict.items():
# Skip internal/system tables
if table.startswith('_') or table.startswith('pg_'):
continue
if exclude_audit:
columns = [c for c in columns if c['name'] not in AUDIT_COLUMNS]
if columns: # only include if columns remain
compressed[table] = columns
return compressed
if __name__ == '__main__':
demo_schema = {
'users': [{'name': 'id', 'type': 'INT'}, {'name': 'email', 'type': 'VARCHAR'}, {'name': 'created_at', 'type': 'TIMESTAMP'}],
'pg_stat': [{'name': 'x', 'type': 'INT'}],
}
compressed = compress_schema(demo_schema)
print('Tables kept:', list(compressed.keys()))
print('users columns after compression:', [c['name'] for c in compressed['users']])
テーブルの説明の追加
カラム名だけでは、意味を十分に理解できないことがあります。各テーブルが何を表すのかを自然言語で説明すると、SQL生成の品質が大幅に向上します。
説明は設定ファイル、またはPostgreSQLのテーブルコメントとして保存します。
TABLE_DESCRIPTIONS = {
'users': 'Registered app users with authentication info',
'orders': 'Customer purchase orders',
'order_items': 'Individual line items within an order',
'products': 'Product catalog with pricing',
'payments': 'Payment transactions linked to orders'
}
def format_schema_with_descriptions(schema_dict):
lines = []
for table, columns in schema_dict.items():
desc = TABLE_DESCRIPTIONS.get(table, '')
col_str = ', '.join(f"{c['name']} ({c['type']})" for c in columns)
if desc:
lines.append(f"Table {table} ({desc}): {col_str}")
else:
lines.append(f"Table {table}: {col_str}")
return '\n'.join(lines)
if __name__ == '__main__':
demo_schema = {'users': [{'name': 'id', 'type': 'INT'}], 'orders': [{'name': 'id', 'type': 'INT'}]}
print(format_schema_with_descriptions(demo_schema))
スキーマのキャッシュ
データベースのスキーマはめったに変更されません。クエリごとにINFORMATION_SCHEMAを取得すると、レイテンシーと負荷が増加します。整形済みのスキーマ文字列をキャッシュし、スキーマ変更イベントの発生時または時間ベースのTTLに達した時点で無効化してください。
import time
class SchemaCache:
def __init__(self, ttl_seconds=300):
self._cache = None
self._timestamp = 0
self.ttl = ttl_seconds
def get(self, conn):
now = time.time()
if self._cache is None or (now - self._timestamp) > self.ttl:
print('Refreshing schema cache...')
schema_dict = build_schema_dict(conn)
fk_info = get_foreign_keys(conn)
self._cache = format_schema_for_prompt(schema_dict, fk_info=fk_info)
self._timestamp = now
return self._cache
schema_cache = SchemaCache(ttl_seconds=300)スキーマ全体の注入フロー
すべての手法を組み合わせます。圧縮したスキーマをキャッシュし、システムプロンプトに注入し、大規模なデータベースでは選択的なテーブルフィルタリングを使用します。
def build_sql_agent_prompt(question, conn, large_db=False):
if large_db:
schema = compressed_schema(question, conn)
schema_text = format_schema_with_descriptions(schema)
else:
schema_text = schema_cache.get(conn)
system = f'''You are a PostgreSQL expert.
Return ONLY a valid SELECT query based on this schema:
{schema_text}
Rules:
- Use only SELECT statements
- Use table aliases for clarity
- Limit results to 100 rows unless asked for all
'''
return system理解度チェック
スキーマ全体を注入するのではなく、選択的なテーブル注入を使用するのはどのような場合ですか？
まとめ：スキーマの理解と注入
効果的なスキーマ注入は、信頼性の高いNL-to-SQLエージェントの基盤です。INFORMATION_SCHEMAから構造を抽出し、主キーと外部キーの関係を含め、LLM向けにコンパクトなテキストとして整形してください。
大規模なデータベースでは、スキーマをキャッシュし、監査用の列を除外し、各質問に関連するテーブルだけを送信する選択的な注入を使用してください。自然言語によるテーブルの説明を加えると、クエリの品質がさらに向上します。
よくある質問
「スキーマの理解と注入」レッスンは無料ですか？
はい。「スキーマの理解と注入」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「スキーマの理解と注入」で何を学びますか？
LLMのコンテキスト向けに、テーブル、カラム、リレーションなどのDBスキーマを抽出・整形します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。
「スキーマの理解と注入」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。