理解与注入模式
为 LLM 上下文提取并格式化 DB 模式：表、列和关系。
理解与注入模式 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
模式上下文为何重要
LLM 了解 SQL 语法，却对您的数据库一无所知。没有模式上下文，它就会凭空生成表名和列名。
模式注入意味着以编程方式提取数据库结构，并将其包含在每个提示中，让 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 会将模式作为纯文本读取。请使用简洁易读的格式：每行一个表，并在括号中列出列名和类型。
包含主键（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 的上下文窗口，并在词元上浪费资金。
一个包含 200 张表且每张表有 20 个列的模式，大约包含 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模式压缩：选择性注入
最有效的压缩策略是：仅注入与问题相关的表。采用两阶段方法——先询问 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。
对于大型数据库：缓存模式，移除审计列，并使用选择性注入，只发送与每个问题相关的表。用自然语言描述表还可以进一步提升查询质量。
常见问题解答
「理解与注入模式」课时是免费的吗？
是的 — 「理解与注入模式」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「理解与注入模式」这节课中我会学到什么？
为 LLM 上下文提取并格式化 DB 模式：表、列和关系。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。
「理解与注入模式」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。