慢查询诊断工具的调用边界设计

继续商行

378人浏览 · 2026-08-25 16:39:16

继续商行 · 2026-08-25 16:39:16 发布

慢查询诊断工具的调用边界设计

线上慢查询 SQL 治理一直是一件费时费力的体力活。每次收到 MySQL 慢日志告警，DBA 和开发人员都要登录跳板机，手动执行 EXPLAIN、分析索引选择度、查看 SHOW PROFILE，最后再给出加索引或改写 SQL 的建议。

我们团队尝试用大模型 Agent 结合 Tool Calling（工具调用）构建一套自动诊断系统。只要把 Slow Log 投递给 Agent，它就能自动调用数据库诊断工具并给出重构方案。

然而在初测阶段，Agent 踩到了致命的工程盲区：遇到一条极其复杂的多表 Join 慢 SQL 时，LLM 生成的 Tool Calling 陷入了无限死循环，不停重复调用 SHOW FULL PROCESSLIST；更有甚者，大模型在尝试“验证优化效果”时，竟然企图调用 ALTER TABLE 直接去线上建索引。

这篇文章介绍我们如何用确定性的状态机（FSM）、SQL 语法树（AST）只读安全防线以及主从只读拓扑，彻底收口 DB 诊断 Agent 的不确定性行为。

1. 为什么智能 Agent 的 Tool Calling 必须硬性约束？

大模型本身是不具备“安全意识”的，它对于工具调用的选择完全依赖概率推导。

如果直接把数据库连接池暴露给 Agent 的 Tool Calling 接口，会引发三大线上隐患：

幻觉写操作风险：模型可能为了“解决问题”而输出 UPDATE、DELETE 或 ALTER TABLE 语句，直接破坏生产数据。
工具调用无限死循环：当 EXPLAIN 返回的字段无法满足模型预期时， Agent 可能会陷入循环重试，把 DB 连接池耗尽。
主库 CPU 拖垮风险：在主库上直接运行 EXPLAIN ANALYZE 复杂的慢 SQL，会真实触发语句执行，导致主库 CPU 瞬间打满。

针对这些问题，我们设计的确定性 Tool Calling 状态机与生产隔离拓扑如下：

2. 基于 AST 语法树校验与有限状态机 (FSM) 的代码防线

为了防止 Agent 执行非预期指令，我们使用 Python 编写了一个带 FSM 轮次限制和 sqlglot 语法树解析的安全工具包装器（Tool Wrapper）。

代码绝不信任 LLM 传入的任何 SQL 字符串，在提交数据库前强制做 AST 校验与白名单拦截：

import sqlglot
from sqlglot import exp
from enum import Enum, auto
from typing import Dict, Any, List

class AgentState(Enum):
    IDLE = auto()
    WAITING_FOR_TOOL = auto()
    EXECUTING_TOOL = auto()
    COMPLETED = auto()
    FAILED = auto()

class UnsafeSQLError(Exception):
    """当 Agent 试图执行写操作或违规语句时抛出"""
    pass

class HardenedDBToolRunner:
    """
    确定性数据库诊断工具运行器
    内置 AST 只读防线与 FSM 状态机
    """
    def __init__(self, db_readonly_connection: Any, max_rounds: int = 5):
        self.db_conn = db_readonly_connection
        self.max_rounds = max_rounds
        self.current_round = 0
        self.state = AgentState.IDLE
        # 工具调用白名单
        self.allowed_tools = {"explain_query", "show_indexes", "analyze_table_structure"}

    def execute_tool_call(self, tool_name: str, tool_args: Dict[str, Any]) -> str:
        """
        Agent 工具调用统一入口
        """
        # 1. 状态机轮次强约束：防 Tool Calling 无限死循环
        self.current_round += 1
        if self.current_round > self.max_rounds:
            self.state = AgentState.FAILED
            raise RuntimeError(f"FSM 安全防线拦截: Agent 工具调用轮次超过最大限制 ({self.max_rounds} 轮)，强制截断")

        # 2. 白名单工具校验
        if tool_name not in self.allowed_tools:
            raise UnsafeSQLError(f"未授权的工具调用: {tool_name}")

        self.state = AgentState.EXECUTING_TOOL

        # 3. 路由到具体工具处理逻辑
        if tool_name == "explain_query":
            return self._safe_explain(tool_args.get("sql", ""))
        elif tool_name == "show_indexes":
            return self._safe_show_indexes(tool_args.get("table_name", ""))
        else:
            raise NotImplementedError(f"工具 {tool_name} 暂未支持")

    def _safe_explain(self, target_sql: str) -> str:
        """
        使用 SQL AST 深度解析目标 SQL，严格杜绝一切写操作
        """
        if not target_sql:
            raise ValueError("SQL 不能为空")

        try:
            # 1. 解析 SQL 语法树 (支持 MySQL 语法)
            parsed_statements = sqlglot.parse(target_sql, read="mysql")
            if not parsed_statements:
                raise UnsafeSQLError("无法解析的 SQL 语法")

            for stmt in parsed_statements:
                # 2. 强行判定语句根节点：只允许 Select 表达式
                if not isinstance(stmt, exp.Select):
                    raise UnsafeSQLError(f"安全违规：非 SELECT 只读语句被拒绝: {type(stmt).__name__}")

                # 3. 递归遍历 AST 节点，严禁包含 Update/Delete/Drop/Alter 子树
                for node in stmt.walk():
                    if isinstance(node, (exp.Insert, exp.Update, exp.Delete, exp.Create, exp.Drop, exp.Alter)):
                        raise UnsafeSQLError(f"安全违规：在 AST 中检测到危险修改指令: {node}")

        except sqlglot.errors.ParseError as e:
            raise UnsafeSQLError(f"SQL 语法解析失败，禁止执行: {str(e)}") from e

        # 4. 组装安全的 EXPLAIN 语句，且只在只读副本沙箱上运行 (带 2 秒 Timeout 强截断)
        safe_explain_sql = f"EXPLAIN FORMAT=JSON {target_sql}"
        
        try:
            with self.db_conn.cursor() as cursor:
                # 设置当前 Session 级别的超时时间，防止超大 SQL 拖垮只读库
                cursor.execute("SET SESSION max_execution_time = 2000;") # 2000毫秒
                cursor.execute(safe_explain_sql)
                result = cursor.fetchone()
                return str(result)
        except Exception as e:
            return f"执行 EXPLAIN 异常 (已拦截风险): {str(e)}"

3. 生产部署拓扑与隔离治理

为了从物理层面上杜绝 Agent 对生产主库造成侵扰，我们在 K8s 集群部署拓扑中做出了严格的物理网络隔离与权限剥离。

具体的部署拓扑规则包含以下三条铁律：

# 生产部署隔离配置规范 (env.conf)
# 1. 账号权限收口：只赋予 SELECT 与 PROCESS 权限，严禁 GRANT ALL
DB_AGENT_USER="agent_diagnoser_ro"
DB_AGENT_PRIVILEGES="SELECT, PROCESS"

# 2. 物理网络隔离：Agent 服务只能连接 ReadOnly-Replica 从库 VIP
DB_HOST="10.0.96.45" # ReadOnly Replica VIP (严禁配置 Master VIP)
DB_PORT=3306

# 3. 实例级硬限制：限制该账号的最大连接数与 CPU 资源上限
DB_MAX_CONNECTIONS=5

AST 白名单、只读账号和状态机可以缩小 Agent 的操作面，但不能替代审计与回归测试。若要衡量诊断质量，应定义带标注的 SQL 样本、拒绝规则和失败分类，并记录误报、漏报与资源占用。

治理 AI Agent 的核心思路不是寄希望于大模型“变聪明”，而是用确定性的状态机和语法树把它牢牢圈在安全的沙箱里。只要防护边界足够硬， Agent 就是提升 DBA 效率的绝佳利器。

使用与验证

发布前把来源对齐

这篇讨论的是高并发与系统性能里的“慢查询诊断工具的调用边界设计”。判断不能只靠某一次顺利的结果，需要把请求队列、连接池、线程栈、慢查询和性能剖析放回同一段执行过程里看。每个配置都要能回答三个问题：它从哪里来、谁会读取、改错后怎样回退。把本地默认值、构建注入值和运行环境值放在同一张对照表里，部署前用实际制品跑一次检查，别依赖口头确认。

实际处理时，我会先选一个普通请求和一个边界请求，分别记下开始时间、关键输入与最终结果。若两者差异很大，就继续向下拆分，而不是马上把问题归因给某个工具。这里的目标不是把记录做得漂亮，而是让后来接手的人能够复走当时的路径。