Prompt工程化调优实录：基于GPT-4o的SQL生成任务token成本削减43%

# eval_prompt.py — 基于OpenAI SDK的Prompt评测脚手架
import json, time, re
from openai import OpenAI
from sqlalchemy import create_engine, text

client = OpenAI(api_key="sk-xxx", timeout=30)
engine = create_engine("mysql+pymysql://root:pass@localhost:3306/insight")

def evaluate(prompt_template: str, test_cases: list) -> dict:
    total_cost = 0.0; correct_syntax = 0; correct_logic = 0; latency = []
    results = []

    for case in test_cases:
        question = case["question"]
        true_sql = case["sql"]
        prompt = prompt_template.format(question=question)

        start = time.perf_counter()
        resp = client.chat.completions.create(
            model="gpt-4o-2024-05-13",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.2,
            max_tokens=512
        )
        latency.append(time.perf_counter() - start)
        gen_sql = resp.choices[0].message.content.strip()

        # token统计：输入+输出
        usage = resp.usage
        total_tokens = usage.prompt_tokens + usage.completion_tokens
        total_cost += usage.prompt_tokens * 5e-6 + usage.completion_tokens * 15e-6

        # 语法检查：是否能执行
        try:
            with engine.connect() as conn:
                conn.execute(text(gen_sql))
            syntax_ok = True
        except Exception:
            syntax_ok = False

        # 逻辑检查：执行结果是否一致
        logic_ok = False
        if syntax_ok:
            with engine.connect() as conn:
                real = conn.execute(text(true_sql)).fetchall()
                pred = conn.execute(text(gen_sql)).fetchall()
            logic_ok = (real == pred)

        correct_syntax += syntax_ok
        correct_logic += logic_ok
        results.append({"q": question, "gen": gen_sql, "syntax": syntax_ok, "logic": logic_ok})

    return {
        "syntax_acc": correct_syntax / len(test_cases),
        "logic_acc": correct_logic / len(test_cases),
        "avg_tokens": total_tokens / len(test_cases),
        "avg_cost": total_cost / len(test_cases),
        "avg_latency": sum(latency) / len(latency)
    }

if __name__ == "__main__":
    with open("test_cases.json", "r", encoding="utf-8") as f:
        cases = json.load(f)
    baseline_prompt = "请把下面的问题转成SQL：{question}"
    stats = evaluate(baseline_prompt, cases)
    print(json.dumps(stats, indent=2, ensure_ascii=False))

你是一个资深MySQL数据分析师，负责将业务问题转换为精确的SQL查询。
数据库表结构如下：
- users(id INT PK, name VARCHAR(50), region VARCHAR(20), created_at DATETIME)
- orders(id INT PK, user_id INT FK, product_id INT FK, amount DECIMAL(10,2), order_date DATETIME)
- products(id INT PK, name VARCHAR(50), category VARCHAR(30), price DECIMAL(10,2))

业务规则：
1. 金额字段一律使用DECIMAL比较，不隐式转换
2. 日期过滤使用date()函数包裹字段
3. 涉及“最近”等时间词时，默认按order_date降序取前7天

请将下面的问题转换为SQL，只输出SQL语句本身：
{question}

示例1：问题：上个月每个品类的总销售额是多少？
SQL：SELECT p.category, SUM(o.amount) AS total_sales
FROM orders o JOIN products p ON o.product_id = p.id
WHERE o.order_date >= DATE_SUB(CURDATE(), INTERVAL 1 MONTH)
GROUP BY p.category;

示例2：问题：哪个用户的下单次数最多？
SQL：SELECT u.name, COUNT(o.id) AS order_count
FROM users u JOIN orders o ON u.id = o.user_id
GROUP BY u.id
ORDER BY order_count DESC LIMIT 1;

# prompt_v3.py — 带自校验链的Prompt模板
PROMPT_V3 = """
你是MySQL专家。根据表结构：{schema}
生成符合要求的SQL。要求：
- 必须处理NULL值，使用IFNULL或COALESCE
- 日期范围使用BETWEEN，避免开区间
- 只输出SQL，不要任何解释

任务：{question}

请先写出SQL，然后逐条检查以下项目：
1. 所有引用的表是否已JOIN？
2. 聚合函数是否遗漏GROUP BY？
3. 日期条件是否包含边界？
如果发现问题，重新生成正确的SQL。

最终输出格式（JSON）：
{{"sql": "你的SQL语句"}}
"""