0Pricing
AI Agents · 课时

生成并验证 SQL 查询

安全 SQL 的提示词模式：仅 SELECT 模式和参数化查询。

生成并验证 SQL 查询 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

SQL 生成目标

生成 SQL 查询只是工作的一半。在针对真实数据库执行查询之前，您需要验证查询是安全的、语法正确的，并且确实实现了用户的意图。

本课程涵盖仅 SELECT 模式强制、解析、安全执行和 EXPLAIN 计划验证。

强制仅 SELECT 模式

NL-to-SQL 智能体最危险的行为是执行破坏性语句。无论 LLM 返回什么内容，都必须始终强制使用仅 SELECT 模式。

简单的字符串检查并不充分，请使用正规的 SQL 解析器。

import sqlparse

def is_select_only(sql):
    parsed = sqlparse.parse(sql)
    if not parsed:
        return False
    for statement in parsed:
        stmt_type = statement.get_type()
        if stmt_type != 'SELECT':
            print(f'Blocked statement type: {stmt_type}')
            return False
    return True

# Test
print(is_select_only('SELECT * FROM users'))  # True
print(is_select_only('DROP TABLE users'))      # False — Blocked

作为纵深防御的关键字阻止列表

即使使用了 sqlparse，也应添加关键字阻止列表作为第二道防线。某些 SQL 注入可能会绕过解析器。在执行前检查危险关键字，可以增加额外的安全层。

DANGEROUS_KEYWORDS = [
    'INSERT', 'UPDATE', 'DELETE', 'DROP', 'CREATE',
    'ALTER', 'TRUNCATE', 'GRANT', 'REVOKE', 'EXEC',
    'EXECUTE', 'CALL', 'MERGE'
]

def passes_blocklist(sql):
    sql_upper = sql.upper()
    for keyword in DANGEROUS_KEYWORDS:
        # Check as whole word to avoid false positives like 'CREATED_AT'
        import re
        if re.search(r'\b' + keyword + r'\b', sql_upper):
            raise ValueError(f'Blocked keyword detected: {keyword}')
    return True

def validate_sql(sql):
    if not is_select_only(sql):
        raise ValueError('Only SELECT statements are allowed')
    passes_blocklist(sql)
    return True

使用 sqlparse 解析 SQL

sqlparse 会对 SQL 字符串进行标记化和解析，但不会执行它们。您可以检查查询结构、提取表名并检查语法问题。

使用 pip install sqlparse 安装。

import sqlparse
from sqlparse.sql import IdentifierList, Identifier
from sqlparse.tokens import Keyword, DML

def extract_table_names(sql):
    parsed = sqlparse.parse(sql)[0]
    tables = []
    from_seen = False
    for token in parsed.tokens:
        if token.ttype is DML and token.value.upper() == 'SELECT':
            continue
        if token.ttype is Keyword and token.value.upper() in ('FROM', 'JOIN'):
            from_seen = True
            continue
        if from_seen:
            if isinstance(token, Identifier):
                tables.append(token.get_name())
            elif isinstance(token, IdentifierList):
                for item in token.get_identifiers():
                    tables.append(item.get_name())
            from_seen = False
    return tables

print(extract_table_names('SELECT u.name FROM users u JOIN orders o ON u.id = o.user_id'))
# ['users', 'orders']

验证表是否存在于模式中

从生成的 SQL 中提取表名后，请将其与已知模式进行交叉核对。如果 LLM 臆造了一个表名，应在执行前拒绝该查询，而不是等到数据库返回晦涩难懂的错误。

def validate_tables_exist(sql, known_tables):
    used_tables = extract_table_names(sql)
    invalid = [t for t in used_tables if t and t not in known_tables]
    if invalid:
        raise ValueError(
            f'Query references non-existent tables: {invalid}. '
            f'Available tables: {list(known_tables)[:10]}...'
        )
    return True

# Usage
known = set(build_schema_dict(conn).keys())
try:
    validate_tables_exist(generated_sql, known)
except ValueError as e:
    # Send error back to LLM for correction
    corrected_sql = llm_fix_sql(generated_sql, str(e))
    print('Corrected SQL:', corrected_sql)

参数化执行

绝不要使用字符串格式化将用户提供的值注入 SQL。即使查询由 LLM 生成，任何用户提供的筛选值也应作为参数传入，以防止 SQL 注入。

import sqlite3

conn = sqlite3.connect(':memory:')
conn.execute('CREATE TABLE orders (status TEXT, user_id INTEGER)')
conn.execute("INSERT INTO orders VALUES ('pending', 42)")

def safe_execute(conn, sql_template, params=()):
    """Execute with parameterized values."""
    cur = conn.cursor()
    cur.execute(sql_template, params)  # driver handles escaping
    columns = [d[0] for d in cur.description]
    rows = cur.fetchmany(200)
    return {'columns': columns, 'rows': rows}

sql = 'SELECT * FROM orders WHERE status = ? AND user_id = ?'
result = safe_execute(conn, sql, params=('pending', 42))
print(result)

执行前检查 EXPLAIN 计划

对于针对大型表的高开销查询，请先运行 EXPLAIN，再执行实际查询。如果计划器显示正在对包含百万行的表进行全表扫描，请向用户发出警告或拒绝该查询。

def check_explain_plan(conn, sql):
    explain_sql = f'EXPLAIN {sql}'
    with conn.cursor() as cur:
        cur.execute(explain_sql)
        plan = '\n'.join(row[0] for row in cur.fetchall())

    # Check for sequential scans on large tables
    if 'Seq Scan' in plan:
        print('WARNING: Query involves a sequential scan')
        print(plan)
        return {'safe': False, 'plan': plan, 'warning': 'Sequential scan detected'}

    return {'safe': True, 'plan': plan}

# Use before executing
plan_result = check_explain_plan(conn, generated_sql)
if not plan_result['safe']:
    print(f'Optimization hint: {plan_result["warning"]}')

强制行数限制

LLM 可能会生成没有 LIMIT 的 SELECT * FROM logs，从而潜在地返回数百万行。请始终强制执行最大行数限制：可以将 LIMIT 添加到查询中，也可以获取有界的结果集。

import re

MAX_ROWS = 500

def enforce_row_limit(sql, max_rows=MAX_ROWS):
    sql_upper = sql.upper().rstrip().rstrip(';')

    # Check if LIMIT already present
    if re.search(r'\bLIMIT\b', sql_upper):
        # Extract current limit and enforce maximum
        match = re.search(r'LIMIT\s+(\d+)', sql_upper)
        if match:
            current = int(match.group(1))
            if current > max_rows:
                sql = re.sub(r'LIMIT\s+\d+', f'LIMIT {max_rows}', sql, flags=re.IGNORECASE)
    else:
        sql = sql.rstrip(';') + f' LIMIT {max_rows}'

    return sql

print(enforce_row_limit('SELECT * FROM users'))
# SELECT * FROM users LIMIT 500

从 LLM 输出中提取干净的 SQL

LLM 经常会将 SQL 包裹在 Markdown 代码块（```sql ... ```）中，或在 SQL 附带解释性文字。您需要先提取原始 SQL，然后再进行解析或执行。

import re

CODE_FENCE = chr(96) * 3  # three backticks, built at runtime to avoid template issues

def extract_sql(llm_response):
    # Remove markdown code blocks ('''sql ... ''' or ''' ... ''')
    pattern = CODE_FENCE + r'(?:sql)?\s*([\s\S]+?)' + CODE_FENCE
    match = re.search(pattern, llm_response, re.IGNORECASE)
    if match:
        return match.group(1).strip()

    # If no code block, look for SELECT statement
    match = re.search(r'(SELECT\s+[\s\S]+?;)', llm_response, re.IGNORECASE)
    if match:
        return match.group(1).strip()

    # Fallback: strip common preamble phrases
    cleaned = re.sub(r'^(Here is|The SQL query is|Query:)[^\n]*\n', '',
                     llm_response, flags=re.IGNORECASE).strip()
    return cleaned

if __name__ == '__main__':
    demo_response = 'Here is the SQL query:\n' + CODE_FENCE + 'sql\nSELECT * FROM users;\n' + CODE_FENCE
    print(extract_sql(demo_response))

完整验证流程

将所有验证步骤串联到一个函数中：该函数接收原始 LLM 输出，并返回安全、可执行的 SQL 字符串；如果无法通过验证，则抛出带有描述性消息的错误，以便进行恢复。

def validate_and_prepare_sql(llm_output, known_tables, max_rows=500):
    # Step 1: extract raw SQL
    sql = extract_sql(llm_output)
    if not sql:
        raise ValueError('No SQL found in LLM response')

    # Step 2: type check
    if not is_select_only(sql):
        raise ValueError('Only SELECT queries allowed')

    # Step 3: keyword blocklist
    passes_blocklist(sql)

    # Step 4: table existence check
    validate_tables_exist(sql, known_tables)

    # Step 5: row limit
    sql = enforce_row_limit(sql, max_rows)

    return sql

# Full flow
try:
    safe_sql = validate_and_prepare_sql(llm_output, known_tables)
    result = safe_execute(conn, safe_sql)
except ValueError as e:
    corrected = llm_fix_sql(llm_output, str(e))
    safe_sql = validate_and_prepare_sql(corrected, known_tables)
    result = safe_execute(conn, safe_sql)

只读数据库用户

代码层面的验证很重要，但还不够。作为最后一道防线，请使用只有 SELECT 权限的只读用户账户连接数据库。即使恶意查询绕过了所有检查，数据库也会拒绝执行它。

# Create read-only user in PostgreSQL:
# CREATE USER nl_to_sql_reader WITH PASSWORD 'secure_password';
# GRANT CONNECT ON DATABASE yourdb TO nl_to_sql_reader;
# GRANT USAGE ON SCHEMA public TO nl_to_sql_reader;
# GRANT SELECT ON ALL TABLES IN SCHEMA public TO nl_to_sql_reader;

import os
import psycopg2

def get_readonly_connection():
    return psycopg2.connect(
        host=os.getenv('DB_HOST'),
        database=os.getenv('DB_NAME'),
        user='nl_to_sql_reader',       # read-only account
        password=os.getenv('DB_READER_PASS')
    )

知识检查

在 NL-to-SQL 智能体中，针对 SQL 验证的正确纵深防御方法是什么？

回顾：生成与验证 SQL

安全的 SQL 生成需要完整的验证流程：从 LLM 输出中提取干净的 SQL，使用 sqlparse 强制仅 SELECT，应用关键字阻止列表，根据真实模式验证表名，强制执行行数限制，并将只读数据库用户作为最后的保护措施。

涉及用户提供值时，参数化查询可以防止注入。EXPLAIN 计划检查可以避免在生产数据上运行意外的高开销查询。

常见问题解答

「生成并验证 SQL 查询」课时是免费的吗？

是的 — 「生成并验证 SQL 查询」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「生成并验证 SQL 查询」这节课中我会学到什么？

安全 SQL 的提示词模式：仅 SELECT 模式和参数化查询。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。

「生成并验证 SQL 查询」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. NL 到 SQL 代理的工作原理
  2. 理解与注入模式
  3. 生成并验证 SQL 查询
  4. 处理含糊的数据库问题
← 返回 AI Agents