企业官网建设流程全解析

数据库 Agent 安全执行：SQL 白名单、审批与只读账号

数据库 Agent 不应拿到可随意写入的高权限连接。解析 SQL、限制语句类型、人工审批高风险动作，并用只读账号执行查询，才能把模型错误限制在可控范围。

1. 把危险 SQL 当成必须覆盖的测试输入

SQL 优化 Agent 可以调用EXPLAIN读取执行计划并生成候选索引，但它不应直接在目标数据库执行ALTER TABLE。收益要在隔离数据副本上比较执行时间、扫描行数和写入开销，再由人工审批变更。

安全测试应显式构造包含多语句、DROP TABLE、无WHERE的DELETE和注释绕过等输入，例如DROP TABLE order_legacy; ALTER TABLE order_master ADD INDEX ...。断言重点不是模型会不会生成它，而是解析器、权限和审批层能否稳定拒绝。

模型输出具有随机性，提示词约束不能替代执行层校验。只要工具具备写权限，就应假定参数可能非法或越权。

要将 Agent 引入数据库治理这种核心场景，不宜依赖模型的“自觉性”，需要在 Agent 与数据库之间建立由确定性软件工程构筑的防护沙箱与拦截闸门。

2. LLM Tool Calling 的非确定性陷阱与沙箱隔离架构

为什么单纯依靠 Prompt 无法阻断 Agent 生成危险指令？

因为 LLM 生成工具调用参数本质上是基于概率分布的 Token 预测。当慢查询上下文里包含了“清理废弃字段”、“重建临时表”等敏感词汇时，模型的注意力机制非常容易被诱导，输出包含DROP、TRUNCATE或无WHERE条件的DELETE指令。

为了解决 Tool Calling 的非确定性风险，可以设计一套确定性 SQL 校验沙箱架构。这套架构的核心是将 Agent 的“决策权”与最终“执行权”物理解耦：

1. 意图路由与 AST 静态解析：Agent 生成的任何 SQL 工具调用指令，需要先经过 AST（抽象语法树）解析器。纯静态代码强制剥离所有的 DDL 删除操作与危险 DML。
2. 影子数据库（Shadow DB）沙箱演练：Agent 生成的ALTER TABLE语句不能直接在生产或测试主库上执行，需要先在基于 Docker 容器秒级克隆的影子数据库沙箱中演练。如果在沙箱中执行报错，或者导致锁表超时，直接拦截并向 Agent 返回错误反馈。
3. 确定性参数 Hash 与幂等屏障：防止 Agent 在遇到网络抖动或解析超时时，重复下发相同的索引创建请求，避免数据库产生索引碎片或锁竞争。

下面是 Agent SQL 工具调用的确定性防护与沙箱演练流程：

通过这套控制体系，可以把非确定性的 Agent 行为限制在了安全的规则框架内。

3. 确定性 SQL 解析与 Agent 拦截器脚手架代码

在实现上，可以使用 Python 编写了一个 Agent 工具调用的安全拦截器。该拦截器结合了 SQL 语法树分析、只读权限强制校验以及工具调用的超时退避机制。

代码拒绝简单的正则匹配，而是基于标准的 SQL 解析逻辑来识别操作类型：

import json import time from typing import Dict, Any, Tuple import sqlglot from sqlglot import exp class DangerousSQLError(Exception): """当 Agent 试图调用破坏性 SQL 时抛出的确定性异常""" pass class DatabaseAgentSandbox: def __init__(self, shadow_db_uri: str, max_execution_sec: float = 2.0): self.shadow_db_uri = shadow_db_uri self.max_execution_sec = max_execution_sec # 允许的 DDL/DML 白名单操作类型 self.allowed_expressions = (exp.Select, exp.Explain, exp.AlterTable) def intercept_and_execute(self, tool_call_payload: str) -> str: """ Agent 工具调用的统一入口。 执行确定性拦截、语法树解析与沙箱演练。 """ try: payload = json.loads(tool_call_payload) raw_sql = payload.get("sql", "").strip() tool_name = payload.get("tool_name", "") if not raw_sql: return json.dumps({"status": "error", "message": "SQL 参数为空"}) # 1. 静态 AST 语法树解析与危险指令过滤 self._verify_sql_safety(raw_sql) # 2. 工具调用的路由逻辑 if tool_name == "query_explain": return self._run_explain(raw_sql) elif tool_name == "apply_index_recommendation": return self._run_in_shadow_sandbox(raw_sql) else: return json.dumps({"status": "error", "message": f"未知的工具名称: {tool_name}"}) except DangerousSQLError as e: # 安全红线拦截，直接返回错误信息给 Agent，引导模型纠正 CoT return json.dumps({ "status": "security_blocked", "message": f"【安全防护拦截】检测到高危 SQL 指令，已被沙箱拒绝: {str(e)}" }) except Exception as e: return json.dumps({"status": "error", "message": f"系统拦截器处理异常: {str(e)}"}) def _verify_sql_safety(self, sql: str): """基于 SQLGlot 抽象语法树执行受限的语句类型检查""" try: parsed = sqlglot.parse_one(sql) except Exception as err: raise DangerousSQLError(f"SQL 语法无法解析: {err}") # 检查是否包含任何 DROP, TRUNCATE, DELETE, UPDATE 节点 for node in parsed.find_all(exp.Drop, exp.Truncate, exp.Delete, exp.Update): raise DangerousSQLError(f"禁止执行破坏性 SQL 操作: {node.key.upper()}") # 如果包含 ALTER TABLE，需要严格限制只能是 ADD INDEX if isinstance(parsed, exp.AlterTable): alter_sql_upper = sql.upper() if "DROP COLUMN" in alter_sql_upper or "DROP INDEX" in alter_sql_upper: raise DangerousSQLError("ALTER TABLE 中禁止包含 DROP 相关子句") if "ADD INDEX" not in alter_sql_upper and "ADD KEY" not in alter_sql_upper: raise DangerousSQLError("ALTER TABLE 仅允许 ADD INDEX 操作") def _run_explain(self, sql: str) -> str: """只读执行 EXPLAIN 分析""" # 强制拼接 EXPLAIN 关键字 explain_sql = f"EXPLAIN {sql}" if not sql.upper().startswith("EXPLAIN") else sql # 模拟只读 DB 执行 return json.dumps({ "status": "success", "type": "explain_result", "data": [{"id": 1, "select_type": "SIMPLE", "table": "orders", "type": "ALL", "possible_keys": None, "rows": None}] }) def _run_shadow_sandbox(self, sql: str) -> str: """在 Shadow DB 沙箱中演练索引创建""" start_time = time.time() # 模拟在影子容器中运行 DDL time.sleep(0.1) # 模拟沙箱耗时 if (time.time() - start_time) > self.max_execution_sec: return json.dumps({"status": "sandbox_timeout", "message": "沙箱中索引创建耗时过长，可能导致线上锁表！"}) return json.dumps({ "status": "success", "type": "sandbox_verified", "message": "影子沙箱已返回 EXPLAIN，请与基线计划和扫描行数比较。" }) # 验证 Agent 工具调用拦截防线 if __name__ == "__main__": sandbox = DatabaseAgentSandbox(shadow_db_uri="mysql://shadow:3306/test_db") # 1. 测试 Agent 输出危险 DROP 指令 dangerous_payload = json.dumps({ "tool_name": "apply_index_recommendation", "sql": "DROP TABLE legacy_orders; ALTER TABLE orders ADD INDEX idx_created (created_at)" }) res1 = sandbox.intercept_and_execute(dangerous_payload) print("高危指令测试结果:\n", res1) # 2. 测试 Agent 输出正常优化指令 safe_payload = json.dumps({ "tool_name": "apply_index_recommendation", "sql": "ALTER TABLE orders ADD INDEX idx_user_created (user_id, created_at)" }) res2 = sandbox.intercept_and_execute(safe_payload) print("\n安全指令测试结果:\n", res2)

代码的阻断能力非常干净利落。任何试图绕过校验规则包含DROP或TRUNCATE的 SQL 参数，都会被_verify_sql_safety明确斩断，并以结构化 JSON 形式向 Agent 返回security_blocked提示，重新引导 Agent 修正推理策略。

4. 验证口径与记录方法

为了让开发者能够在本地秒级复现 Agent 索引优化效果，可以搭建一套极轻量级的 Docker Compose 演练脚手架。

这套脚手架包含三个组件：

1. 主测试数据库（MySQL 8.0）：使用可重复生成的数据集，并保存行数、分布和初始化脚本。
2. 影子演练数据库（Shadow MySQL）：基于临时 tmpfs 挂载，数据秒级重置，供 Agent 演练 DDL。
3. Agent 工作流 Runner：集成上述 Python 拦截沙箱，提供 HTTP 调试接口。

docker-compose.yml配置文件定义如下：

version: '3.8' services: main-db: image: mysql:8.0 container_name: agent_main_db environment: MYSQL_ROOT_PASSWORD: rootpassword MYSQL_DATABASE: shop_order ports: - "3306:3306" command: --default-authentication-plugin=mysql_native_password --slow_query_log=1 --long_query_time=0.5 shadow-db: image: mysql:8.0 container_name: agent_shadow_db tmpfs: - /var/lib/mysql:rw,noexec,nosuid,size=1g environment: MYSQL_ROOT_PASSWORD: rootpassword MYSQL_DATABASE: shop_order ports: - "3307:3306"

5. Agent 自动化索引优化的执行边界

在大模型与 Agent 快速接入企业生产基础设施的今天，数据库索引优化的自动化改造带来了明显的效能红线。

Agent 调用数据库工具时，可以先落实三项不可省略的约束：

1. Agent 不持有生产 DDL 权限，只生成候选方案并在隔离副本验证。变更由 DBA 审批后选择原生 Online DDL、pt-online-schema-change或gh-ost；这些工具仍需评估 MDL、复制延迟和回滚风险。
2. SQL 校验优先使用语法树：正则容易被换行、注释或嵌套查询绕过。AST 也不是完整安全边界，还要配合数据库只读权限、语句白名单和执行超时。
3. 沙箱演练需要带有时空配额限制：在影子数据库中演练 DDL 时，需要限制语句最大耗时（CPU time）与内存分配（tmpfs limit）。防止 Agent 生成产生 Cartesian Product（笛卡尔积）的坏 SQL，反向拖垮本地演练宿主机。

收尾