AI 哨兵自动解析 PostgreSQL 慢查询

上周二，一个跑了几个月的查询突然开始占用大量 CPU。值班工程师花了四十分钟阅读 EXPLAIN 输出，才意识到是列类型变更导致索引失效。我也曾遇到过同样的问题，这也是我为什么要构建一个不需要等待询问的哨兵。这个工具不能替代 DBA，但它缩短了从发现慢查询到理解慢查询原因之间的 gap。

MonkeyCode 是一个开源 AI 编程助手，目前提供免费模型访问和免费服务器选项。声明：本文是作为 MonkeyCode 产品推广的一部分撰写的。免费服务器并非生产环境，但它是运行监控工具的理想场所——这类工具只需要读取查询统计信息。我下面描述的哨兵从 pg_stat_statements 中拉取最慢的查询，用执行计划格式化它们，然后请求免费模型用通俗语言解释根本原因。

自解释哨兵的架构

自解释哨兵由四部分组成：收集器、格式化器、模型调用器和存储层。收集器读取累积统计信息，因此不需要持续运行。格式化器将原始查询文本和计划 JSON 转换为 prompt，请求给出具体建议。模型调用器是唯一依赖 MonkeyCode 免费模型访问的部分，存储层保留每条建议供后续审查。

步骤 1：准备免费服务器数据库

在免费服务器上创建一个数据库并启用 pg_stat_statements。该扩展随标准 PostgreSQL 一起提供，因此只需调整 shared_preload_libraries 并重启实例。下面的代码片段还创建了一个存储建议的表，将每次模型调用的输出集中在一处。

CREATE EXTENSION IF NOT EXISTS pg_stat_statements;

CREATE TABLE IF NOT EXISTS query_advice (
    id BIGSERIAL PRIMARY KEY,
    query_id BIGINT,
    query_text TEXT,
    plan JSONB,
    advice TEXT,
    impact_estimate TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

步骤 2：编写收集器

收集器从 pg_stat_statements 中查询按总执行时间排列的最慢查询，然后为每个查询获取一个新的 EXPLAIN (FORMAT JSON)。它将原始数据写入暂存列表，以便模型调用器稍后处理。过滤掉系统查询很重要，因为你需要的是关于应用 SQL 的建议，而不是内部目录查询。

import json
import psycopg2

def collect_slow_queries(dsn: str, limit: int = 10) -> list[dict]:
    with psycopg2.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT queryid, query, calls, total_exec_time
                FROM pg_stat_statements
                WHERE query NOT LIKE '%%pg_%%'
                ORDER BY total_exec_time DESC
                LIMIT %s
            """, (limit,))
            rows = cur.fetchall()
            items = []
            for queryid, query, calls, total_ms in rows:
                cur.execute("EXPLAIN (FORMAT JSON) " + query)
                plan = cur.fetchone()[0]
                items.append({
                    'query_id': queryid,
                    'query': query,
                    'calls': calls,
                    'total_ms': total_ms,
                    'plan': plan,
                })
    return items

步骤 3：请求免费模型进行解释

这里是免费模型访问发挥作用的地方。下面的函数将查询文本和计划 JSON 发送到模型端点，请求返回一个结构化答案。端点和密钥是占位符，因为 MonkeyCode 的确切接口可能会更改；请根据项目文档进行调整。

import os
import requests

def ask_model(query: str, plan: dict) -> dict:
    prompt = f"""
    Explain why this SQL query is slow and suggest a concrete fix.
    Query: {query}
    Plan: {json.dumps(plan)}
    Respond with JSON: {{"advice": "...", "impact": "high|medium|low"}}
    """
    resp = requests.post(
        os.environ["MODEL_ENDPOINT"],
        headers={"Authorization": f"Bearer {os.environ['MODEL_KEY']}"},
        json={"messages": [{"role": "user", "content": prompt}]},
        timeout=30,
    )
    return resp.json()["choices"][0]["message"]["content"]

一个典型的响应是这样的：模型指出在大表上有一次顺序扫描，建议建立复合索引，并将影响标记为高。你不需要盲目信任它；它的价值在于为你自己的调查提供了一个起点。

步骤 4：存储并排序建议

解析模型响应并将其插入 query_advice 表。一个简单的 SELECT 然后按 impact 字段对建议进行排序，使最有价值的修复方案优先出现。这将一堆原始计划转化为一个优先处理的待办列表。

def store_advice(dsn: str, item: dict, advice: dict) -> None:
    with psycopg2.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO query_advice (query_id, query_text, plan, advice, impact_estimate)
                VALUES (%s, %s, %s, %s, %s)
            """, (item['query_id'], item['query'], json.dumps(item['plan']),
                  advice['advice'], advice['impact']))

SELECT impact_estimate, advice, created_at
FROM query_advice
ORDER BY CASE impact_estimate WHEN 'high' THEN 0 WHEN 'medium' THEN 1 ELSE 2 END;

步骤 5：定时执行与审查

通过 cron 作业每小时运行一次收集器和模型调用器。免费服务器可以轻松处理这个负载，免费令牌配额覆盖每小时少量查询。每天审查一次建议表，只应用那些经过你自己推理验证的更改。

*/60 * * * * cd /opt/sentinel && python3 run.py

不适合使用此方案的人群

不要将这个哨兵指向包含受监管个人数据的数据库，因为免费服务器并非生产环境。不要将模型建议当作真理，它是一个需要人工验证的假设。已经拥有专职 DBA 的团队可能会觉得建议过于通用，而完全无法容忍任何外部调用的团队则应该完全跳过模型调用步骤。

重申定位

零美元服务器不仅仅是测试沙箱，它可以运行一个安静的助手，解释你数据库中最糟糕的部分。其价值不在于免费令牌本身，而在于养成在重写查询之前先问「为什么这个查询慢」的习惯。如果你想要尝试这种模式，MonkeyCode 的免费服务器和免费模型访问是一个低成本起步的方式，开源项目也值得在信任它之前阅读一下。