不只是"自然语言转 SQL"：一个生产级数据库节点的技术亮点拆解

本文是「成都硅基边界」零代码构建平台旗下 Silicon-AI 智能体引擎实战复盘系列第六篇。前五篇聊了智能体、RAG、LangChain 教学、MCP 插件与工作流引擎，这一篇聚焦数据库：让 AI 直接对业务库做增删改查，到底有哪些坑，我们又是怎么一层层拦住的。

文中每个技术亮点都配了示意代码——均根据设计改写、非项目真实源码，只用于演示思路与 API 用法。

一、背景：RAG 解决不了"结构化查询"

企业最有价值的数据，大多躺在关系型数据库里：订单、客户、库存、账单。RAG 擅长从文档里找答案，但对"上个月华东区销售额 TOP5 的产品"这类问题无能为力——答案不是某段文字，而是需要聚合计算的结果。

所以"让 AI 直接写 SQL 查库"（NL2SQL）几乎是所有智能体平台的必备能力。但它也是风险最高的一环：

• 上下文爆炸：几十张表的完整 schema 塞进提示词，token 直接爆表，还会引入噪声；
• 危险的写操作：模型一句 DELETE FROM orders 就能删库，提示词里写"请不要删除数据"根本拦不住；
• 幻觉式回答：SQL 到底执行成功没有、影响了哪些行，模型可能凭空编造；
• 跨表关联无能：真实业务往往是多步操作（先查 A 表拿 ID，再写入 B 表）。

Silicon-AI 的数据库节点围绕这四个问题做了一整套设计，下面逐个拆解技术亮点。

二、亮点一：渐进式 Schema 裁剪，先筛表再取字段

最直觉的做法是把整个库的 schema 塞给模型，代价是上下文爆炸。我们改成两级渐进式检索：

flowchart TD
    A[用户问题] --> B[第一级：筛选相关表<br/>按问题语义从库中挑出候选表]
    B --> C[第二级：取表字段详情<br/>只展开被选中的表结构]
    C --> D[组装精简 Schema]
    D --> E[LLM 生成 SQL]
    E --> F{是否写操作}
    F -->|否| G[执行并返回结果]
    F -->|是| H[权限闸门校验]
    H -->|通过| I[执行 + 写后回查]
    H -->|拒绝| J[返回明确拒绝原因]
    I --> K[组织答案]

对应的示意实现：

# 示意代码（非项目源码）：两级 schema 裁剪，把全量 schema 变成按需加载
def build_prompt_schema(question: str, db) -> str:
    # 第一级：只给「表名 + 表描述」，让模型挑出相关表
    candidates = db.list_tables()                   # [(name, comment), ...]
    picked = llm_pick_tables(question, candidates)  # 模型返回相关表名列表

    # 第二级：只展开被选中表的字段（列名 / 类型 / 是否必填 / 业务含义）
    blocks = []
    for table in picked:
        cols = db.get_table_fields(table)
        lines = [f"- {c.name} {c.type}"
                 f"{'（必填）' if c.required else ''}：{c.comment}"
                 for c in cols]
        blocks.append(f"表 {table}：\n" + "\n".join(lines))
    return "\n\n".join(blocks)

schema 从"全量"变成按需加载，上下文成本和噪声同步下降——这是效果与成本双赢的第一步。

三、亮点二：权限闸门——不依赖"模型自觉"

这是安全上最关键的设计。平台给每个数据库配了三个独立开关：新增权限（add）、修改权限（edit）、删除权限（delete）。SQL 执行前先解析语句类型，命中对应开关：

# 示意代码（非项目源码）：权限闸门——安全策略必须是确定性代码
def detect_op(sql: str) -> str:
    """按语句前缀判定操作类型。"""
    head = sql.strip().upper()
    for op in ("INSERT", "UPDATE", "DELETE", "SELECT"):
        if head.startswith(op):
            return op
    return "UNKNOWN"

def run_sql(sql: str, fields=None) -> str:
    op = detect_op(sql)
    if op == "INSERT" and not config.add_authority:
        return "当前数据库未开启新增权限，操作已拒绝"
    if op == "UPDATE" and not config.edit_authority:
        return "当前数据库未开启修改权限，操作已拒绝"
    if op == "DELETE" and not config.delete_authority:
        return "当前数据库未开启删除权限，操作已拒绝"
    return execute(sql)      # 通过闸门才真正执行

核心思想：安全策略必须是确定性代码，不能是提示词里的"请小心"。提示词是软约束，模型有概率不遵守；代码是硬校验，百分百拦得住。我们采取"软硬双保险"——提示词里告知规则减少误生成，代码里再做一次拦截兜底。

四、亮点三：必填字段硬校验，拦截"半截 INSERT"

写库时另一类常见事故是"插入了不完整的记录"：schema 里 required=true 的列没给值，模型自己也没意识到。我们的做法是在执行前做必填字段硬校验：

# 示意代码（非项目源码）：解析 INSERT 目标表与列名，比对必填列
import re

def parse_insert_target(sql: str):
    """从 INSERT 语句解析目标表名与显式列名列表。"""
    m = re.match(r"\s*INSERT\s+INTO\s+(?:[\w]+\.)?[`\"\[]?(\w+)[`\"\]]?"
                 r"(?:\s*\(([^)]*)\))?", sql, re.IGNORECASE)
    if not m:
        return None, []
    table = m.group(1)
    cols = [c.strip() for c in m.group(2).split(",")] if m.group(2) else []
    return table, cols

def check_required(sql: str, schema: dict):
    """required=true 的列必须显式出现，不依赖模型自觉。"""
    table, cols = parse_insert_target(sql)
    if not table or table not in schema:
        return
    missing = [c.name for c in schema[table].columns
               if c.required and c.name not in cols]
    if missing:
        raise ValueError(f"缺少必填字段：{', '.join(missing)}")

而且提示词里也会明确告知模型："若用户未提供必填字段值，不要生成 SQL，直接说明缺失哪些字段——即使生成也会被系统拦截"。告诉模型有拦截机制，它就不太会硬着头皮乱生成，这种"预告规则"比单纯禁止更有效。

五、亮点四：写操作回查——让模型看见"实际写了什么"

这是我认为最巧妙的一环。模型生成 UPDATE 或 INSERT 后，如果只返回"执行成功"，它很可能会凭想象向用户描述结果（"已为您新增了订单"——但订单号是数据库生成的，它其实不知道）。

我们的处理是：写操作执行后自动回查一次——

# 示意代码（非项目源码）：写后回查，让模型基于真实数据回答
def execute_write(sql: str, session, schema):
    op = detect_op(sql)

    if op == "INSERT":
        new_ids = session.execute(sql).inserted_primary_keys   # 拿到新行主键
        rows = session.execute(
            f"SELECT * FROM {parse_insert_target(sql)[0]} WHERE id IN :ids",
            {
   "ids": new_ids})
        return {
   "affected": len(new_ids), "rows": rows.fetchall()}

    if op == "UPDATE":
        table = re.match(r"\s*UPDATE\s+(?:[\w]+\.)?[`\"\[]?(\w+)", sql,
                         re.IGNORECASE).group(1)
        where = extract_where_clause(sql)          # 抽取原 WHERE 条件
        affected = session.execute(sql).rowcount
        rows = session.execute(f"SELECT * FROM {table} WHERE {where}")
        return {
   "affected": affected, "rows": rows.fetchall()}

• INSERT 执行完，按新行 ID 回查刚插入的那条记录；
• UPDATE 执行完，按原 WHERE 条件回查更新后的行。

回查结果再喂回模型，它组织出的回答就是基于真实数据的："订单已创建，订单号 SO20260827，金额 1,280 元"。让模型只描述它真正看到的东西，是消灭写操作幻觉最有效的手段。

六、亮点五：多步 SQL 编排，处理真实业务的"连环操作"

真实业务很少是一句 SQL 搞定的。我们把 SQL 执行封装成工具，让模型在循环里多次调用它，从而支持多步编排：

# 示意代码（非项目源码）：SQL 作为工具，模型循环调用完成多步操作
executed_steps = []

@tool
def sql_exec(sql: str, fields: list = None) -> str:
    """执行一条 SQL。SELECT 返回结果集；写操作返回影响行数与回查数据。"""
    run_sql_permission_gate(sql)          # 亮点二：权限闸门
    check_required(sql, schema_map)       # 亮点三：必填校验
    result = execute_write(sql, session, schema_map)   # 亮点四：写后回查
    executed_steps.append({
   "sql": sql, "affected": result["affected"]})
    return format_result(result)

@tool
def get_table_fields(table: str) -> str:
    """查询指定表的字段结构（用于中途补充上下文）。"""
    return db.describe(table)

# 模型在循环里多次调用：先查 A → 再查 B → 最后插入 C
agent = create_agent(
    model=llm,
    tools=[sql_exec, get_table_fields, search_tables],
    system_prompt="复杂需求请拆成多步，每步用工具执行，后续步骤依赖前一步的真实结果。",
)

• 关联新增：先 SELECT 客户表拿到客户 ID → 再 SELECT 产品表拿到产品 ID → 最后用两个 ID INSERT 订单表；
• 级联删除：删除主表记录前，先清理关联表中的依赖数据。

每一步执行都记进 executed_steps（执行轨迹），哪一步成功、哪一步失败、影响了多少行，一目了然——既是给模型看的上下文，也是给用户和运维看的审计日志。

七、亮点六：工程细节里的真功夫

# 示意代码（非项目源码）：SQL 消毒——模型输出常裹着 markdown 与分号
def sanitize_sql(raw: str) -> str:
    s = raw.strip()
    s = re.sub(r"^```(?:sql)?|```$", "", s, flags=re.MULTILINE).strip()  # 去代码块
    s = re.sub(r"--.*$", "", s, flags=re.MULTILINE)                      # 去行注释
    s = s.rstrip(";").strip()                                            # 去尾部分号
    return s

1. SQL 消毒：去掉 Markdown 代码块、尾部分号、行注释，避免"明明是对的却执行报错"；
2. 语句类型路由：SELECT 与写操作走不同返回策略——写操作只在真正影响数据时才返回结果，避免"执行了 0 行"却回答"已成功"；
3. PostgreSQL JSONB 的坑：对 jsonb 字段更新时，jsonb_set(data, ...) 直接写会报类型错误，必须显式转换：

-- 正确：先 ::jsonb 再 jsonb_set，最后 ::text 转回
UPDATE t SET data = jsonb_set(data::jsonb, '{key}', '"newValue"')::text WHERE id = 1;

-- 错误：缺少类型转换，PostgreSQL 会报函数不匹配
UPDATE t SET data = jsonb_set(data, '{key}', '"newValue"') WHERE id = 1;

1. 输出字段描述：要求模型在调用工具时说明每个输出字段的业务含义，让最终结果"可解释"，而不是一堆裸列名：

# 示意代码（非项目源码）：工具参数要求声明输出字段的业务含义
class SqlExecToolArgs(BaseModel):
    sql: str = Field(description="要执行的 SQL 语句")
    fields: List[FieldDesc] = Field(
        default=[],
        description="SELECT 输出字段描述列表；INSERT/UPDATE/DELETE 可为空")

八、亮点七：连接与会话治理（工程底座）

再聪明的查询也架不住连接池崩了。

# 示意代码（非项目源码）：连接串编码 + 连接池参数化
from urllib.parse import quote_plus

# 密码含 @ 必须百分号编码，否则 make_url 按第一个 @ 切分导致主机解析错误
url = (f"postgresql+psycopg://{quote_plus(user)}:{quote_plus(password)}"
       f"@{host}/{dbname}")

engine = create_engine(
    url,
    pool_pre_ping=True,          # 自动剔除失效连接
    pool_size=pool_size,         # 多副本部署时按 PG max_connections 反推
    max_overflow=max_overflow,
    pool_recycle=pool_recycle,
    pool_timeout=pool_timeout,
    connect_args={
   "connect_timeout": 10,
                  "keepalives": 1, "keepalives_idle": 60,   # 防中间设备静默断连
                  "application_name": "silicon-ai"},         # pg_stat_activity 可溯源
)

# 示意代码（非项目源码）：慢 SQL 监听，区分「SQL 慢」与「连接池等待」
@event.listens_for(engine, "before_cursor_execute")
def _start(conn, cursor, statement, params, context, executemany):
    context._t0 = time.monotonic()

@event.listens_for(engine, "after_cursor_execute")
def _log_slow(conn, cursor, statement, params, context, executemany):
    ms = (time.monotonic() - context._t0) * 1000
    if ms > 500:
        logger.warning("慢SQL %.0fms: %s", ms, " ".join(statement.split())[:200])

# 示意代码（非项目源码）：ASGI 中间件 + ContextVar，会话覆盖流式全生命周期
session_context: ContextVar[Session] = ContextVar("session_context")

class SessionMiddleware:
    async def __call__(self, scope, receive, send):
        if scope["type"] in ("http", "websocket"):
            with Session(engine) as session:
                token = session_context.set(session)
                try:
                    await self.app(scope, receive, send)   # SSE 流式期间 session 依然可用
                finally:
                    session_context.reset(token)

def get_current_session() -> Session:
    return session_context.get()      # 任何代码无需层层传参即可拿到会话

四个要点：密码百分号编码（生产库密码含 @ 会毁掉连接串）、连接池参数化 + pre_ping + keepalive + application_name、慢 SQL 监听（有慢日志 = SQL 慢；没慢日志但请求慢 = 卡在连接池等待）、会话覆盖 SSE 全生命周期（纯 ASGI 中间件 + ContextVar，流式响应内部也能拿到会话）。

九、两种模式：AI 生成 SQL 与手写 SQL

# 示意代码（非项目源码）：双模式共用同一条拦截管道
if db_config.create_sql == "ai":
    state = rag_database.query(question, llm)      # 模型生成 SQL 并执行
else:
    state = rag_database.exec_sql(db_config.sql)   # 直接执行用户手写 SQL

# 无论哪条路径，权限闸门、必填校验、写后回查全部一致生效

AI 模式适合业务人员（自然语言提问），自定义模式适合有确定逻辑的场景（不消耗生成 token、结果完全可控）。两种模式共用同一套权限校验与执行管道——上层怎么来不重要，底层拦截只有一条。

十、结语：让 AI 写 SQL 的三条铁律

回顾整个设计，可以浓缩成三句话：

1. 最小上下文——渐进式 schema 裁剪，只给模型它此刻需要知道的表结构；
2. 硬校验优先——权限、必填字段这类安全与完整性约束，必须是代码拦截，提示词只做辅助；
3. 写后必回查——让模型基于真实数据回答，而不是基于想象。

NL2SQL 的难点从来不是"让模型写出 SQL"，而是让它在可控、可审计、可解释的前提下写对 SQL。把这三点做到位，AI 查库才真正能进生产。