大模型技术赋能 MySQL：从智能查询生成到向量化搜索

import openai
import pymysql
import json

openai.api_key = "your-api-key"

def generate_sql(question: str, schema: str) -> str:
    """根据自然语言问题和表结构生成SQL"""
    prompt = f"""
你是一个资深的MySQL DBA。请根据以下表结构，将用户的自然语言问题转换成正确的SQL查询。
表结构信息：
{schema}

注意事项：
1. 只返回SQL语句，不要包含任何解释
2. 使用标准MySQL语法
3. 如果问题不明确，请返回明确的错误提示

用户问题：{question}
SQL:
"""
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1
    )
    sql = response.choices[0].message.content.strip()
    # 去除可能的markdown代码块标记
    if sql.startswith("```sql"):
        sql = sql[6:-3]
    return sql

# 示例
schema_example = """
CREATE TABLE orders (
    id INT PRIMARY KEY AUTO_INCREMENT,
    user_id INT NOT NULL,
    product_name VARCHAR(100),
    amount DECIMAL(10,2),
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE users (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(50),
    email VARCHAR(100)
);
"""
question = "查询最近7天内每个用户的订单总金额"
sql = generate_sql(question, schema_example)
print(sql)
# 输出: SELECT u.name, SUM(o.amount) FROM orders o JOIN users u ON o.user_id=u.id WHERE o.created_at >= NOW() - INTERVAL 7 DAY GROUP BY u.id;

import subprocess
import re

def get_slow_queries(log_path="/var/log/mysql/mysql-slow.log", limit=5):
    """使用pt-query-digest或直接读取慢日志，简化版仅演示"""
    with open(log_path, "r") as f:
        content = f.read()
    # 简单正则提取查询（实际需要更精细解析）
    queries = re.findall(r"# Query_time: .*?\n(.*?);", content, re.DOTALL)
    return queries[:limit]

def optimize_suggestion(slow_queries: list, schema: str) -> str:
    prompt = f"""
以下是MySQL慢查询日志中提取的几条耗时SQL，以及表结构。请分析每条SQL的性能瓶颈，并给出具体的优化建议，包括索引添加、SQL重写或配置调整。

表结构：
{schema}

慢查询列表：
{chr(10).join(slow_queries)}

请以Markdown列表的形式输出，每条建议对应一个查询。
"""
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2
    )
    return response.choices[0].message.content

# 示例
slow = ["SELECT * FROM orders WHERE user_id=123 AND created_at > '2025-01-01'"]
schema = "orders表：id, user_id, product_name, amount, created_at；有索引(user_id)"
print(optimize_suggestion(slow, schema))

EXPLAIN FORMAT=JSON SELECT ...;

CREATE TABLE articles (
    id INT PRIMARY KEY AUTO_INCREMENT,
    title VARCHAR(200),
    content TEXT,
    embedding VECTOR(1536) NOT NULL
);

INSERT INTO articles (title, content, embedding) 
VALUES (
    'MySQL向量特性',
    'MySQL 8.0开始支持向量类型...',
    VECTOR('[0.12, -0.34, ..., 0.56]')   -- 实际需1536个浮点数
);

SELECT id, title, 
       1 - VECTOR_DISTANCE(embedding, VECTOR('[0.10, -0.30, ...]')) AS cosine_similarity
FROM articles
ORDER BY cosine_similarity DESC LIMIT 10;

from sentence_transformers import SentenceTransformer
import pymysql

model = SentenceTransformer('all-MiniLM-L6-v2')

def embed_text(text: str) -> list:
    return model.encode(text).tolist()

def search_similar(query: str, top_k=5):
    query_vec = embed_text(query)
    conn = pymysql.connect(host='localhost', user='root', password='...', database='test')
    cursor = conn.cursor()
    # 使用向量距离排序（需提前归一化）
    sql = """
    SELECT id, title, 
           1 - VECTOR_DISTANCE(embedding, %s) AS score
    FROM articles
    ORDER BY score DESC
    LIMIT %s
    """
    cursor.execute(sql, (f"[{','.join(map(str, query_vec))}]", top_k))
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return results

# 插入文档时
def insert_doc(title, content):
    embedding = embed_text(content)
    conn = pymysql.connect(...)
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO articles (title, content, embedding) VALUES (%s, %s, VECTOR(%s))",
        (title, content, f"[{','.join(map(str, embedding))}]")
    )
    conn.commit()

mysql-ai assist "查询本月新注册用户的消费总额"   # NL2SQL
mysql-ai tune /var/log/mysql-slow.log          # 慢查询分析
mysql-ai search "如何优化JOIN性能"            # 向量语义搜索文档

用户输入 → 意图识别（可用LLM分类） → 路由到对应处理器 → 调用LLM或向量检索 → 返回结果