某电商平台订单库在促销期间频繁出现慢查询，平均响应时间从80毫秒拉长到2秒以上。团队最初靠人工翻慢日志、逐条执行EXPLAIN，一天只能处理十几个问题，效率很低。后来我们引入数据库性能优化Agent，将异常发现、根因分析、建议生成和审批执行串成自动化闭环。上线后单条慢SQL的平均诊断时间从45分钟压缩到3分钟以内，并且大部分低风险优化动作可以自动实施。本文就结合这个案例，拆解Agent的架构设计、核心代码与安全执行机制。
一、整体架构与数据采集
Agent整体分为采集层、分析层、决策层和执行层四个部分。采集层负责对接MySQL的慢查询日志和performance_schema。慢日志是离线数据，保留完整的SQL文本和执行时间，适合深度分析；performance_schema中的events_statements_summary_by_digest表提供实时聚合，可以快速发现突增的慢查询指纹。本案例采用双通道采集：慢日志用于事后归因，性能视图用于实时告警。这样可以兼顾准确性和响应速度。
采集器用Python编写，只读账号访问数据库，不在生产库执行复杂查询。慢日志文件可能被轮转，因此需要处理文件inode变化，这里使用logrotate的copytruncate模式，并定时检测文件大小重置游标。下面是一个简化版解析脚本，提取时间戳和查询耗时，只保留耗时超过1秒的记录。
import re
from datetime import datetime
def parse_slow_log_line(line):
time_match = re.search(r'# Time: (\d{6} \d{1,2}:\d{2}:\d{2})', line)
duration_match = re.search(r'Query_time: (\d+\.\d+)', line)
if time_match and duration_match:
timestamp = datetime.strptime(time_match.group(1), '%y%m%d %H:%M:%S')
duration = float(duration_match.group(1))
if duration > 1.0:
return timestamp, duration
return None
with open('/var/log/mysql/slow.log', 'r') as f:
for line in f:
result = parse_slow_log_line(line)
if result:
print(result)
性能视图的采集则通过一条聚合SQL完成，比如按照DIGEST_TEXT分组统计平均耗时、最大耗时和出现次数。为了避免频繁查询带来开销，Agent每30秒执行一次，只拉取最近5分钟内有更新的记录。采集的数据会写入本地消息队列，由分析层异步消费。
二、慢查询分析与根因定位
分析层的第一步是对慢SQL做归一化处理。把数值常量替换为占位符，生成稳定的SQL指纹。例如SELECT * FROM orders WHERE user_id = 10086归一化为SELECT * FROM orders WHERE user_id = ?。这样同一个业务查询即使参数不同，也只会生成一条分析任务。指纹库还会记录历史分析结果，如果表结构没有变化，直接复用缓存，避免重复调用大模型。
对于新指纹，Agent调用EXPLAIN FORMAT=JSON获取执行计划，重点解析access_type、key、rows和Extra字段。规则引擎先做一轮快速判断：当access_type为ALL且扫描行数超过设定阈值时，基本可以判定缺少有效索引；当Extra中包含Using filesort或Using temporary时，则需要检查排序字段和分组条件。下面代码演示如何获取执行计划并提取访问类型。
import pymysql
import json
conn = pymysql.connect(
host='127.0.0.1',
user='monitor',
password='secret',
database='orders'
)
cursor = conn.cursor()
cursor.execute('EXPLAIN FORMAT=JSON SELECT * FROM orders WHERE user_id = 10086 ORDER BY created_at DESC LIMIT 10')
plan = json.loads(cursor.fetchone()[0])
access_type = plan['query_block']['table']['access_type']
print(access_type)
cursor.close()
conn.close()
规则引擎只能覆盖约六成常见问题，剩下的需要大模型介入。我们把表结构、执行计划、索引信息和慢查询上下文拼接到提示词中，要求模型引用执行计划的具体指标给出判断依据。为了防止模型幻觉，提示词中明确要求只建议真实存在的字段，禁止虚构表结构。下面是一个简化的提示词构造示例，实际使用时会根据数据库版本和sql_mode动态调整。
prompt = f"""
你是MySQL性能优化专家，请根据以下信息定位慢查询原因。
表结构：
{table_schema}
执行计划：
{plan}
慢查询SQL指纹：
{指纹}
要求：
1. 引用执行计划中的access_type、key、rows字段给出判断依据。
2. 只建议表结构中存在的索引，禁止虚构字段。
3. 如果存在参数化改写空间，请给出改写后的SQL。
"""
print(prompt)
三、优化建议生成与安全执行
决策层生成的优化建议必须是结构化数据，而不是一段自然语言。每条建议包含建议类型、目标表、变更SQL、回滚SQL、风险等级和预期收益。常见建议类型有创建索引、删除冗余索引、改写SQL和调整参数。创建索引通常属于低风险，但在大表上执行时仍要考虑锁表时间和IO抖动；改写SQL属于中风险，因为语法变化可能影响结果集；调整参数和修改表结构属于高风险，必须走变更窗口。
安全执行是Agent能否落地的关键。执行器不会直接在线上数据库执行DDL，而是按照风险等级分流处理。低风险操作进入自动化队列，由调度模块在低峰期执行，并设置执行超时和失败告警；中高风险操作会创建审核任务，推送到企业IM或工单系统，DBA确认后才执行。下面是一个简化版执行器，展示风险分流逻辑。
class Executor:
def __init__(self, risk):
self.risk = risk
self.review_queue = []
def submit(self, sql, rollback_sql):
if self.risk == 'low':
self.auto_apply(sql, rollback_sql)
else:
self.review_queue.append({
'sql': sql,
'rollback': rollback_sql,
'risk': self.risk
})
def auto_apply(self, sql, rollback_sql):
print('执行:', sql)
print('回滚:', rollback_sql)
每个自动执行的任务在完成后都会进入验证阶段。Agent重新执行EXPLAIN对比执行计划，并观察实际响应时间。如果性能没有提升，甚至出现回退，会自动触发回滚脚本。对于创建索引，回滚脚本就是DROP INDEX；对于改写的SQL，则需要保留原SQL版本并支持快速切换。整个过程中全程记录审计日志，保证可追溯。
四、案例效果与迭代方向
这套Agent在测试环境稳定运行一个月后，慢查询平均耗时从1.96秒下降到0.63秒，降幅约68%。夜间慢查询告警数量从每天四十多条降到十条以内，大部分性能问题在用户反馈前就被自动处理。DBA团队不再需要半夜爬起来逐条翻日志，可以把精力放在数据模型设计和归档策略上。更重要的是，优化建议的沉淀形成了一个数据库调优知识库，新加入的工程师也能快速了解历史问题。
迭代过程中也踩过一些坑。大模型对MySQL 8.0的某些新特性不熟悉，曾经给出过不兼容的索引语法；还有一次规则引擎误判了覆盖索引的效果，生成的CREATE INDEX没有消除回表。后来我们在提示词中显式加入数据库版本和sql_mode信息，并强制所有建议在测试库验证，问题才得到解决。这也说明自动化Agent不能完全替代人工，需要保留审核和回滚机制。
下一步计划把Agent接入云数据库审计日志和应用APM链路追踪，从调用链角度还原慢查询发生的完整上下文。同时尝试引入强化学习，让Agent根据历史决策的执行结果调整策略权重，逐步降低对人工审核的依赖。对于数据量特别大的表，还可以结合分区、归档等方案，让优化建议更加立体。