M2 - 智能体式人工智能 - 用反思改进 SQL 生成

在这个实验中，你将探索反思模式如何改进一个把自然语言问题转化为 SQL 数据库查询的智能体工作流。你将看到智能体如何发现自身输出中的问题、加以改进，并在给出最终答案之前提升响应质量。

你将练习应用反思模式，来提升智能体工作流编写 SQL 查询的能力。为此，你将在工作流中编写一个反思步骤，让智能体：

• 审阅自己的中间结果（例如草稿 SQL 或工具输出）。
• 识别错误或缺漏。
• 检查自己的响应和工具使用情况。
• 在提交最终答案之前对输出进行改进。

在这一步，你将准备好工作环境，以便立即开始编码。你将：

1. 导入核心 Python 库

   • json 用于处理结构化数据。
   • pandas 用于处理表格数据。
   • dotenv 用于加载环境变量（例如 API 密钥）。

2. 加载环境变量 这能确保你的工作环境正确配置了所需的密钥和设置。

3. 导入 utils.py 模块 该文件包含一些辅助函数，你将用它们来格式化输出，并支持工作流中的后续步骤。

注意： 如果你想查看 utils.py 的内容，请前往顶部菜单，选择 File > Open。

import json
import utils
import pandas as pd
from dotenv import load_dotenv

_ = load_dotenv()

本课程的实验将使用一个名为 aisuite 的软件包（aisuite 仓库），它让调用不同模型提供方托管的大模型变得很简单。Andrew 会在模块 3 中进一步讨论 aisuite 的细节。现在，先初始化 aisuite client。这个客户端为你提供了一种统一的方式来连接和使用不同的大模型——这样你就不必操心每个模型各自的设置差异。

import aisuite as ai

client = ai.Client()

在这一步，你将创建一个名为 products.db 的本地 SQLite 数据库。 该数据库将自动填充随机生成的产品数据。

你稍后会在本实验中使用这份数据来练习和测试你的 SQL 查询。

utils.create_transactions_db()

你可以通过运行下方单元格来查看表结构（schema）。

utils.print_html(utils.get_schema('products.db'))

在这张表中，每一行代表一个事件（插入、补货、销售或价格更新）。库存水平、销量或价格趋势等分析结果都是从这些事件中衍生出来的。

有了这份 schema，你就可以通过对事件历史进行聚合，随时重建出当前状态（库存、最新价格、总销量）。

在这一步，你将使用一个函数，把你的自然语言问题转化为 SQL 查询。

你提供你的问题和数据库 schema 作为输入。大模型随后会生成能够回答该问题的 SQL 查询。

这样一来，你就可以专注于提问，而由模型负责编写查询语句。

def generate_sql(question: str, schema: str, model: str) -> str:
    prompt = f"""
    You are a SQL assistant. Given the schema and the user's question, write a SQL query for SQLite.

    Schema:
    {schema}

    User question:
    {question}

    Respond with the SQL only.
    """
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    return response.choices[0].message.content.strip()

运行下方单元格，看看 generate_sql 是如何从一个纯英文的问题出发，为 transactions 表创建出 SQL 查询的第一版（V1）的。 你只需提供问题、schema 和模型名称，函数就会返回初始的查询草稿。

# Example usage of generate_sql

# We provide the schema as a string
schema = """
Table name: transactions
id (INTEGER)
product_id (INTEGER)
product_name (TEXT)
brand (TEXT)
category (TEXT)
color (TEXT)
action (TEXT)
qty_delta (INTEGER)
unit_price (REAL)
notes (TEXT)
ts (DATETIME)
"""

# We ask a question about the data in natural language
question = "Which color of product has the highest total sales?"

utils.print_html(question, title="User Question")

# Generate the SQL query using the specified model
sql_V1 = generate_sql(question, schema, model="openai:gpt-4.1")

# Display the generated SQL query
utils.print_html(sql_V1, title="SQL Query V1")

现在你将执行第一版（V1）SQL 查询并查看其结果。这一步很重要，因为它能让你验证大模型生成的查询是否真的从数据库中取回了你所期望的信息。

• utils.execute_sql(...)：针对 products.db 数据库运行生成的 SQL 查询（V1），并以 pandas DataFrame 的形式返回输出。使用 DataFrame 能更方便地检查、分析结果，并将其传递给工作流的后续步骤。

• utils.print_html(...)：接收 DataFrame，并将其渲染为笔记本中一个格式整齐的 HTML 表格。这让原始输出更易读，也方便你快速判断查询结果是否与用户的问题相符。

# Execute the generated SQL query (sql_V1) against the products.db database.
# The result is returned as a pandas DataFrame.
df_sql_V1 = utils.execute_sql(sql_V1, db_path='products.db')

# Render the DataFrame as an HTML table in the notebook.
# This makes the query output easier to read and interpret.
utils.print_html(df_sql_V1, title="Output of SQL Query V1 - ❌ Does NOT fully answer the question")

❌ <strong>V1 输出存在的问题：</strong>查询结果并没有完全回答这个问题，因为总销售额是一个<em>负数</em>（-190571.46）。 <br><br> 在这份数据集中，销售（库存流出）时 <code>qty_delta</code> 被记录为负数，而退货或补货（库存增加）时则记录为正数。该查询使用了 <code>SUM(qty_delta)</code>，因此在对所有交易求和时，负数占据主导，导致总和为负。 <br> 这意味着这条 SQL 在语法上是有效的，但在语义上是错误的——总销售额不应该以负值来表示。 <br> ➡️ 这个问题恰恰体现了反思的必要性：模型必须改进其逻辑（例如，在计算销售额时把 <code>qty_delta</code> 乘以 <code>-1</code>，或使用 <code>ABS()</code>），从而让最终的查询能反映出真实的销售总额。

在本节中，你将学习如何用反思改进 SQL 查询。

首先，大模型可以只审查 SQL 文本本身，对照问题和 schema，并在需要时提出改进建议。 接着，大模型还可以结合实际的查询执行结果，从而发现诸如总额为负、缺少筛选条件或分组错误等更细微的问题。

这两种方法结合在一起，展示了反思如何让你的 SQL 工作流变得更可靠、更准确——先在纸面上检查逻辑，再用真实数据加以验证。

在这个函数中，你会要求大模型审阅一条 SQL 查询，对照原始问题和表 schema（例如 3.1 节中定义的那个）。模型会反思该查询是否完全回答了这个问题，如果没有，就提出一个改进版本。

• 输入：

  • 用户的问题
  • 原始 SQL 查询
  • 表 schema

• 输出：

  • feedback → 一段简短的评价（例如“有效，但缺少日期筛选条件”）
  • refined_sql → 最终的 SQL（如果正确则保持不变，如需改进则给出更新版本）

这个函数不会执行 SQL。它只是检查这条查询，并在逻辑与意图不完全匹配时提出改进建议。

def refine_sql(
    question: str,
    sql_query: str,
    schema: str,
    model: str,
) -> tuple[str, str]:
    """
    Reflect on whether a query's *shown output* answers the question,
    and propose an improved SQL if needed.
    Returns (feedback, refined_sql).
    """
    prompt = f"""
You are a SQL reviewer and refiner.

User asked:
{question}

Original SQL:
{sql_query}

Table Schema:
{schema}

Step 1: Briefly evaluate if the SQL OUTPUT fully answers the user's question.
Step 2: If improvement is needed, provide a refined SQL query for SQLite.
If the original SQL is already correct, return it unchanged.

Return STRICT JSON with two fields:
{{
  "feedback": "<1-3 sentences explaining the gap or confirming correctness>",
  "refined_sql": "<final SQL to run>"
}}
"""
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )

    content = response.choices[0].message.content
    try:
        obj = json.loads(content)
        feedback = str(obj.get("feedback", "")).strip()
        refined_sql = str(obj.get("refined_sql", sql_query)).strip()
        if not refined_sql:
            refined_sql = sql_query
    except Exception:
        # Fallback if model doesn't return valid JSON
        feedback = content.strip()
        refined_sql = sql_query

    return feedback, refined_sql

运行下方单元格，生成改进后的 SQL 查询（V2）。这一步将会：

• 展示 3.1 节针对问题 “Which color of product has the highest total sales?” 生成的初始 SQL 查询（V1）。
• 展示模型给出的反馈及其改进后的 SQL 提案（V2）。
• 对数据库执行原始 SQL（V1），并展示其实际输出，让你看到为什么需要改进。

# Example: refine the generated SQL (V1 → V2)

feedback, sql_V2 = refine_sql(
    question=question,
    sql_query=sql_V1,   # <- comes from generate_sql() (V1)
    schema=schema, # <- we reuse the schema from section 3.1
    model="openai:gpt-4.1"
)

# Display the original prompt
utils.print_html(question, title="User Question")

# --- V1 ---
utils.print_html(sql_V1, title="Generated SQL Query (V1)")

# Execute and show V1 output
df_sql_V1 = utils.execute_sql(sql_V1, db_path='products.db')
utils.print_html(df_sql_V1, title="SQL Output of V1 - ❌ Does NOT fully answer the question")

# --- Feedback + V2 ---
utils.print_html(feedback, title="Feedback on V1")
utils.print_html(sql_V2, title="Refined SQL Query (V2)")

# Execute and show V2 output
df_sql_V2 = utils.execute_sql(sql_V2, db_path='products.db')
utils.print_html(df_sql_V2, title="SQL Output of V2 - ❌ Does NOT fully answer the question")

❌ 正如你所看到的……
尽管生成的 SQL 看起来是正确的，反馈也确认了这一点，但实际的查询结果却显示 total_sales 是一个负值。

正如前面提到的，这是因为该 SQL 在计算 qty_delta * unit_price 时，没有考虑到销售事件存储的是负数量（qty_delta < 0）。这种正负号（+ 与 –）的反转是一种细微的语义问题，仅通过审阅查询文本本身并不总能发现。

👉 这正是为什么智能体还必须对执行输出进行反思——从而发现诸如错误符号、缺失筛选条件或错误聚合等问题。来自查询结果的外部反馈，能让改进过程立足于真实情况，而不仅仅是 SQL 的结构。

与前一步只让模型审查 SQL 文本不同，这一次你将提供实际的查询执行结果作为外部反馈。 这份反馈来自针对数据库运行该 SQL 查询——就像 Andrew 视频中的示例一样——这样大模型就可以使用真实的输出来评估该查询是否真正回答了这个问题。

在这一步，你将获得：

• 基于真实执行输出对你的查询给出的简短评价。
• 具体的改进建议（例如缺少筛选条件、分组问题、符号错误）。
• 一条更贴合原始问题的改进后 SQL 语句。

def refine_sql_external_feedback(
    question: str,
    sql_query: str,
    df_feedback: pd.DataFrame,
    schema: str,
    model: str,
) -> tuple[str, str]:
    """
    Evaluate whether the SQL result answers the user's question and,
    if necessary, propose a refined version of the query.
    Returns (feedback, refined_sql).
    """
    prompt = f"""
    You are a SQL reviewer and refiner.

    User asked:
    {question}

    Original SQL:
    {sql_query}

    SQL Output:
    {df_feedback.to_markdown(index=False)}

    Table Schema:
    {schema}

    Step 1: Briefly evaluate if the SQL output answers the user's question.
    Step 2: If the SQL could be improved, provide a refined SQL query.
    If the original SQL is already correct, return it unchanged.

    Return a strict JSON object with two fields:
    - "feedback": brief evaluation and suggestions
    - "refined_sql": the final SQL to run
    """

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=1.0,
    )

    
    content = response.choices[0].message.content
    try:
        obj = json.loads(content)
        feedback = str(obj.get("feedback", "")).strip()
        refined_sql = str(obj.get("refined_sql", sql_query)).strip()
        if not refined_sql:
            refined_sql = sql_query
    except Exception:
        # Fallback if the model does not return valid JSON:
        # use the raw content as feedback and keep the original SQL
        feedback = content.strip()
        refined_sql = sql_query

    return feedback, refined_sql

运行下方单元格，看看查询结果的外部反馈是如何改进 SQL 修订过程的。

在这个示例中，你将：

• 展示基于你的问题生成的原始 SQL 查询（V1）。
• 展示 V1 的输出，说明为什么最初的尝试没有完全回答这个问题。
• 提供大模型基于该输出给出的反馈。
• 展示解决了该问题的改进后 SQL 查询（V2）。
• 执行 V2 并展示其输出，确认它现在✅完全回答了这个问题。

# Example: Refine SQL with External Feedback (V1 → V2)

# Execute the original SQL (V1)
df_sql_V1 = utils.execute_sql(sql_V1, db_path='products.db')

# Use external feedback to evaluate and refine
feedback, sql_V2 = refine_sql_external_feedback(
    question=question,
    sql_query=sql_V1,   # V1 query
    df_feedback=df_sql_V1,    # Output of V1
    schema=schema,
    model="openai:gpt-4.1"
)

# --- V1 ---
utils.print_html(question, title="User Question")
utils.print_html(sql_V1, title="Generated SQL Query (V1)")
utils.print_html(df_sql_V1, title="SQL Output of V1 - ❌ Does NOT fully answer the question")

# --- Feedback & V2 ---
utils.print_html(feedback, title="Feedback on V1")
utils.print_html(sql_V2, title="Refined SQL Query (V2)")

# Execute and display V2 results
df_sql_V2 = utils.execute_sql(sql_V2, db_path='products.db')
utils.print_html(df_sql_V2, title="SQL Output of V2 (with External Feedback) - ✅ Fully answers the question")

✅ 成功！
借助改进后的 SQL，负数 qty_delta 的问题已通过应用 ABS(qty_delta)（或等价地对销售取 -qty_delta）得到了修正。

现在输出结果是正数且有意义的，展示出了总销售额最高的正确颜色。这说明了结合反思与外部反馈的重要性：

• 单看 SQL 文本本身，一切似乎都没问题。
• 而实际的执行输出揭示了一个符号错误。
• 改进后的查询修正了逻辑，得出了正确的答案。

在这一步，你将使用一个函数来自动化整个使用大模型创建、运行和改进 SQL 查询的工作流。

该工作流按以下关键步骤运行：

1. 提取数据库 schema
2. 根据你的自然语言问题生成初始（V1）SQL 查询
3. 结合执行反馈对 V1 进行反思——审查实际的查询结果，并在需要时改进 SQL
4. 执行改进后的（V2）SQL 查询，确保它完全回答了你的问题

最后，你将看到：

• 初始查询和改进后的查询
• 它们各自的输出结果
• 大模型对改进过程的说明性反馈

这让你能够以更流畅、更准确、完全自动化的方式使用 SQL 查询。

def run_sql_workflow(
    db_path: str,
    question: str,
    model_generation: str = "openai:gpt-4.1",
    model_evaluation: str = "openai:gpt-4.1",
):
    """
    End-to-end workflow to generate, execute, evaluate, and refine SQL queries.

    Steps:
      1) Extract database schema
      2) Generate SQL (V1)
      3) Execute V1 → show output
      4) Reflect on V1 with execution feedback → propose refined SQL (V2)
      5) Execute V2 → show final answer
    """

    # 1) Schema
    schema = utils.get_schema(db_path)
    utils.print_html(
        schema,
        title="📘 Step 1 — Extract Database Schema"
    )

    # 2) Generate SQL (V1)
    sql_v1 = generate_sql(question, schema, model_generation)
    utils.print_html(
        sql_v1,
        title="🧠 Step 2 — Generate SQL (V1)"
    )

    # 3) Execute V1
    df_v1 = utils.execute_sql(sql_v1, db_path)
    utils.print_html(
        df_v1,
        title="🧪 Step 3 — Execute V1 (SQL Output)"
    )

    # 4) Reflect on V1 with execution feedback → refine to V2
    feedback, sql_v2 = refine_sql_external_feedback(
        question=question,
        sql_query=sql_v1,
        df_feedback=df_v1,          # external feedback: real output of V1
        schema=schema,
        model=model_evaluation,
    )
    utils.print_html(
        feedback,
        title="🧭 Step 4 — Reflect on V1 (Feedback)"
    )
    utils.print_html(
        sql_v2,
        title="🔁 Step 4 — Refined SQL (V2)"
    )

    # 5) Execute V2
    df_v2 = utils.execute_sql(sql_v2, db_path)
    utils.print_html(
        df_v2,
        title="✅ Step 5 — Execute V2 (Final Answer)"
    )

现在轮到你执行完整的 SQL 处理流水线了。你可以尝试以下几种 OpenAI 模型的不同组合，它们各自具有不同的能力和表现：

• openai:gpt-4o
• openai:gpt-4.1
• openai:gpt-4.1-mini
• openai:gpt-3.5-turbo

💡 在这个工作流中，openai:gpt-4.1 通常在自我反思任务上表现最好。

重要提示： 由于大语言模型（LLM）具有随机性，每次运行都可能返回略有不同的结果。 建议你尝试不同的模型及其组合，找出最适合你的方案。

run_sql_workflow(
    "products.db", 
    "Which color of product has the highest total sales?",
    model_generation="openai:gpt-4.1",
    model_evaluation="openai:gpt-4.1"
)

通过完成这个实验，你学会了如何：

• 使用大模型把自然语言问题转化为 SQL 查询。
• 应用反思模式（结合或不结合外部反馈）来改进生成的 SQL。
• 自动化一个完整的 SQL 工作流，从 schema 提取到查询执行与改进。
• 尝试不同的大模型，比较它们的表现和准确性。

核心洞见在于：反思让你的智能体更可靠——智能体不会止步于第一次尝试，而是能够审阅、改进，并交付更符合你意图的结果。

🎉 <strong>恭喜！</strong>

你已经完成了构建**智能体式 SQL 工作流（V1 → V2）**的实验。

在此过程中，你练习了反思、执行与验证是如何共同构成一个可靠流水线的。 你也看到了外部反馈的重要性：有时单看 SQL 文本似乎是正确的，但实际执行输出却揭示了隐藏的问题（例如总额为负、缺少筛选条件或分组错误）。

有了这些技能，你现在已经准备好设计属于你自己的智能体流水线了——这些工作流能够：

• 生成初始查询（V1）。
• 结合执行反馈对其进行反思和改进（V2）。
• 交付更安全、更透明、也更值得信赖的结果。🌟