Building an AI Agent That Optimizes Your Database Performance

Building an AI Agent That Optimizes Your Database Performance

Database performance degrades over time. As your data grows, queries that were fast at launch become slow, indexes that were effective become fragmented, and the database server becomes overloaded. Manual database optimization is a specialized skill that most developers do not have, which means performance issues go unaddressed until users complain. What you need is an AI agent that analyzes your database's performance, identifies slow queries, recommends indexes, and optimizes the database automatically. With the Deployxa MCP server and an LLM, this is straightforward to build. Here is how.

The direct answer is that a database optimization agent is a script that runs on a schedule (e.g., weekly), analyzes your database's slow query log, identifies the slowest queries (via pg_stat_statements for Postgres), uses an LLM to recommend indexes and query optimizations, applies the recommendations (via SQL commands), and verifies the improvements (by re-running the slow queries and comparing the execution time). The agent uses the Deployxa MCP server to coordinate the optimization process and to notify you of improvements. For more on database performance, see our article on database connection pooling across blue/green deployments.

Why Manual Database Optimization Is Hard

Three problems make manual database optimization hard. First, it requires specialized knowledge. Identifying slow queries, understanding execution plans, and choosing the right indexes is a specialized skill that most developers do not have. Second, it is time-consuming. Analyzing slow queries, testing index changes, and verifying improvements takes hours, which means it gets postponed. Third, it is ongoing. Database performance degrades continuously as data grows, which means optimization needs to be ongoing, not one-time. An AI agent solves all three problems: it uses an LLM to apply specialized knowledge, it runs automatically (on a schedule), and it is ongoing. For more on agentic patterns, see our article on building an AI agent that auto-scales your apps.

What the Optimization Agent Does

The database optimization agent performs the following tasks:

1. Collect slow queries. The agent queries pg_stat_statements (for Postgres) to identify the slowest queries, ranked by total execution time.

1. Analyze execution plans. For each slow query, the agent runs EXPLAIN ANALYZE to get the execution plan, which shows how the database executes the query.

1. Recommend optimizations. The agent uses an LLM to analyze the execution plan and recommend optimizations (e.g., "add an index on the users.email column", "rewrite the query to use a JOIN instead of a subquery").

1. Apply optimizations. The agent applies the recommendations (e.g., creates the index, rewrites the query).

1. Verify improvements. The agent re-runs the slow query and compares the execution time before and after the optimization, to verify the improvement.

1. Report. The agent sends a Slack notification with the optimizations applied and the performance improvements.

Step-by-Step: Building the Optimization Agent

Here is how to build the database optimization agent in Python, using an LLM and Postgres.

Step 1: Install dependencies

pip install openai psycopg2-binary requests schedule
npm install -g @deployxa/mcp-server
deployxa-mcp login

Step 2: Create the agent script

# db_optimizer.py
import openai
import psycopg2
import requests
import schedule
import time
import json
from datetime import datetime

# Configuration
DATABASE_URL = os.environ.get("DATABASE_URL")
SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL")
DEPLOYXA_MCP_URL = "http://localhost:3000/mcp"
OPTIMIZATION_INTERVAL_DAYS = 7
SLOW_QUERY_THRESHOLD_MS = 100  # Queries slower than 100ms are considered slow

def send_slack_notification(message):
    requests.post(SLACK_WEBHOOK_URL, json={"text": f"🗄️ {message}"})

def get_slow_queries():
    conn = psycopg2.connect(DATABASE_URL)
    cur = conn.cursor()
    
    # Get the slowest queries from pg_stat_statements
    cur.execute("""
        SELECT query, total_exec_time, calls, mean_exec_time
        FROM pg_stat_statements
        WHERE mean_exec_time > %s
        ORDER BY total_exec_time DESC
        LIMIT 10
    """, (SLOW_QUERY_THRESHOLD_MS,))
    
    slow_queries = []
    for row in cur.fetchall():
        slow_queries.append({
            "query": row[0],
            "total_exec_time_ms": row[1],
            "calls": row[2],
            "mean_exec_time_ms": row[3],
        })
    
    cur.close()
    conn.close()
    return slow_queries

def get_execution_plan(query):
    conn = psycopg2.connect(DATABASE_URL)
    cur = conn.cursor()
    
    cur.execute(f"EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) {query}")
    plan = "\n".join(row[0] for row in cur.fetchall())
    
    cur.close()
    conn.close()
    return plan

def analyze_with_llm(query, plan):
    client = openai.OpenAI()
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """You are a database optimization agent. Analyze the slow query and its execution plan, and recommend optimizations.

Return a JSON object with:
- "analysis": a plain-English explanation of why the query is slow
- "recommendations": a list of recommendations, each with:
  - "type": "index" or "query_rewrite" or "configuration"
  - "sql": the SQL to apply (e.g., CREATE INDEX ...)
  - "explanation": why this will help
- "estimated_improvement": estimated percentage improvement

Guidelines:
- Recommend indexes for columns that are frequently filtered or joined
- Recommend query rewrites for inefficient SQL (e.g., subqueries -> JOINs)
- Be conservative: do not recommend changes that might break the app
- Consider the trade-off between read performance and write performance
"""
            },
            {
                "role": "user",
                "content": f"""Slow query:
{query}

Mean execution time: {query['mean_exec_time_ms']:.1f}ms
Total execution time: {query['total_exec_time_ms']:.1f}ms
Calls: {query['calls']}

Execution plan:
{plan}
"""
            }
        ]
    )
    
    return json.loads(response.choices[0].message.content)

def apply_recommendation(recommendation):
    conn = psycopg2.connect(DATABASE_URL)
    cur = conn.cursor()
    
    try:
        cur.execute(recommendation["sql"])
        conn.commit()
        return True
    except Exception as e:
        conn.rollback()
        print(f"  Failed to apply: {e}")
        return False
    finally:
        cur.close()
        conn.close()

def verify_improvement(query, old_time):
    conn = psycopg2.connect(DATABASE_URL)
    cur = conn.cursor()
    
    # Run the query multiple times to get an average
    times = []
    for _ in range(5):
        cur.execute(f"EXPLAIN (ANALYZE, FORMAT JSON) {query['query']}")
        result = cur.fetchone()
        times.append(result[0][0]["Execution Time"])
    
    avg_time = sum(times) / len(times)
    improvement = ((old_time - avg_time) / old_time) * 100
    
    cur.close()
    conn.close()
    return avg_time, improvement

def optimize_database():
    print(f"[{datetime.utcnow().isoformat()}] Running database optimization...")
    
    # Step 1: Get slow queries
    slow_queries = get_slow_queries()
    print(f"  Found {len(slow_queries)} slow queries")
    
    improvements = []
    
    for query in slow_queries[:5]:  # Optimize top 5 slowest
        print(f"\n  Analyzing: {query['query'][:80]}...")
        print(f"    Mean time: {query['mean_exec_time_ms']:.1f}ms")
        
        # Step 2: Get execution plan
        plan = get_execution_plan(query["query"])
        
        # Step 3: Analyze with LLM
        analysis = analyze_with_llm(query, plan)
        print(f"    Analysis: {analysis['analysis']}")
        
        # Step 4: Apply recommendations
        for rec in analysis.get("recommendations", []):
            print(f"    Applying: {rec['explanation']}")
            if apply_recommendation(rec):
                # Step 5: Verify improvement
                new_time, improvement = verify_improvement(query, query["mean_exec_time_ms"])
                improvements.append({
                    "query": query["query"][:80],
                    "old_time_ms": query["mean_exec_time_ms"],
                    "new_time_ms": new_time,
                    "improvement_percent": improvement,
                    "optimization": rec["explanation"],
                })
                print(f"    Improvement: {improvement:.1f}% ({query['mean_exec_time_ms']:.1f}ms -> {new_time:.1f}ms)")
    
    # Step 6: Report
    if improvements:
        message = "Database optimization complete!\n\n"
        for imp in improvements:
            message += f"  Query: {imp['query']}...\n"
            message += f"    {imp['old_time_ms']:.1f}ms -> {imp['new_time_ms']:.1f}ms ({imp['improvement_percent']:.1f}% faster)\n"
            message += f"    Optimization: {imp['optimization']}\n\n"
        send_slack_notification(message)
    else:
        print("  No improvements applied")

# Schedule the optimization
schedule.every(OPTIMIZATION_INTERVAL_DAYS).days.do(optimize_database)

# Run immediately, then on schedule
optimize_database()
while True:
    schedule.run_pending()
    time.sleep(1)

Step 3: Enable pg_stat_statements

Enable the pg_stat_statements extension in your Postgres database:

CREATE EXTENSION IF NOT EXISTS pg_stat_statements;

Add the following to your postgresql.conf:

shared_preload_libraries = 'pg_stat_statements'
pg_stat_statements.max = 10000
pg_stat_statements.track = all

Step 4: Run the agent

python db_optimizer.py

The agent runs immediately, then every 7 days. It identifies slow queries, recommends optimizations, applies them, and verifies improvements.

Common Pitfalls and Troubleshooting

The first pitfall is adding too many indexes. Each index speeds up reads but slows down writes, which means too many indexes can degrade write performance. The fix is to only add indexes that provide significant read improvement, and to remove indexes that are not used. The second pitfall is not testing optimizations in staging. An optimization that works in development (with small data) might not work in production (with large data), or might break the app. The fix is to test optimizations in staging (with a copy of the production data) before applying them to production. The third pitfall is not verifying improvements. An optimization that looks good on paper might not actually improve performance, which means you need to verify by comparing the execution time before and after. The fix is to always verify improvements with EXPLAIN ANALYZE. The fourth pitfall is not considering write performance. An index that speeds up reads might slow down writes, which means you need to consider the trade-off. The fix is to monitor write performance after adding an index, and to remove the index if write performance degrades significantly. The fifth pitfall is not maintaining indexes. Indexes can become fragmented over time, which degrades their performance. The fix is to periodically rebuild indexes (e.g., REINDEX INDEX index_name).

Conclusion: Let the Agent Optimize Your Database

Manual database optimization is hard, time-consuming, and ongoing. An AI agent that analyzes slow queries, recommends indexes, and verifies improvements is the sustainable solution. With the Deployxa MCP server, an LLM, and Postgres, you can build a database optimization agent that keeps your database fast as your data grows. Stop letting your database get slow and start optimizing automatically.

Ready to build your database optimization agent? Install the Deployxa MCP server with npm i -g @deployxa/mcp-server, run deployxa-mcp login, and start building. For more on agentic workflows, see our articles on the agentic database backup pipeline and building an AI agent that manages your secrets. Learn about the agentic log analysis pipeline and building an AI agent that manages custom domains in our companion articles. Explore our free developer tools to speed up your workflow.