如何作為 SQL 問答環節的一部分進行查詢驗證

在任何 SQL 鏈或代理器中，最容易出錯的部分可能就是編寫有效且安全的 SQL 查詢。在本指南中，我們將介紹一些驗證查詢和處理無效查詢的策略。

我們將涵蓋

1. 在查詢生成中附加「查詢驗證器」步驟；
2. 提示工程，以減少錯誤的發生率。

設定

首先，取得所需的套件並設定環境變數

%pip install --upgrade --quiet  langchain langchain-community langchain-openai

# Uncomment the below to use LangSmith. Not required.
# import os
# os.environ["LANGSMITH_API_KEY"] = getpass.getpass()
# os.environ["LANGSMITH_TRACING"] = "true"

以下範例將使用與 Chinook 資料庫的 SQLite 連線。請按照這些安裝步驟在與此筆記本相同的目錄中建立 Chinook.db

• 將此檔案另存為 Chinook_Sqlite.sql
• 執行 sqlite3 Chinook.db
• 執行 .read Chinook_Sqlite.sql
• 測試 SELECT * FROM Artist LIMIT 10;

現在，Chinook.db 在我們的目錄中，我們可以使用 SQLAlchemy 驅動的 SQLDatabase 類別與其介接

from langchain_community.utilities import SQLDatabase

db = SQLDatabase.from_uri("sqlite:///Chinook.db")
print(db.dialect)
print(db.get_usable_table_names())
print(db.run("SELECT * FROM Artist LIMIT 10;"))

API 參考：SQLDatabase

sqlite
['Album', 'Artist', 'Customer', 'Employee', 'Genre', 'Invoice', 'InvoiceLine', 'MediaType', 'Playlist', 'PlaylistTrack', 'Track']
[(1, 'AC/DC'), (2, 'Accept'), (3, 'Aerosmith'), (4, 'Alanis Morissette'), (5, 'Alice In Chains'), (6, 'Antônio Carlos Jobim'), (7, 'Apocalyptica'), (8, 'Audioslave'), (9, 'BackBeat'), (10, 'Billy Cobham')]

查詢檢查器

最簡單的策略可能是要求模型本身檢查原始查詢中的常見錯誤。假設我們有以下 SQL 查詢鏈

選擇 聊天模型

OpenAI

Anthropic

Azure

Google Vertex

AWS

Groq

Cohere

NVIDIA

Fireworks AI

Mistral AI

Together AI

IBM watsonx

Databricks

xAI

pip install -qU "langchain[openai]"

import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain.chat_models import init_chat_model

llm = init_chat_model("gpt-4o-mini", model_provider="openai")

from langchain.chains import create_sql_query_chain

chain = create_sql_query_chain(llm, db)

API 參考：create_sql_query_chain

我們想要驗證其輸出。我們可以透過使用第二個提示和模型呼叫來擴展鏈來做到這一點

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

system = """Double check the user's {dialect} query for common mistakes, including:
- Using NOT IN with NULL values
- Using UNION when UNION ALL should have been used
- Using BETWEEN for exclusive ranges
- Data type mismatch in predicates
- Properly quoting identifiers
- Using the correct number of arguments for functions
- Casting to the correct data type
- Using the proper columns for joins

If there are any of the above mistakes, rewrite the query.
If there are no mistakes, just reproduce the original query with no further commentary.

Output the final SQL query only."""
prompt = ChatPromptTemplate.from_messages(
    [("system", system), ("human", "{query}")]
).partial(dialect=db.dialect)
validation_chain = prompt | llm | StrOutputParser()

full_chain = {"query": chain} | validation_chain

API 參考：StrOutputParser | ChatPromptTemplate

query = full_chain.invoke(
    {
        "question": "What's the average Invoice from an American customer whose Fax is missing since 2003 but before 2010"
    }
)
print(query)

SELECT AVG(i.Total) AS AverageInvoice
FROM Invoice i
JOIN Customer c ON i.CustomerId = c.CustomerId
WHERE c.Country = 'USA'
AND c.Fax IS NULL
AND i.InvoiceDate >= '2003-01-01' 
AND i.InvoiceDate < '2010-01-01'

請注意，我們如何在 Langsmith 追蹤中看到鏈的兩個步驟。

db.run(query)

'[(6.632999999999998,)]'

這種方法顯而易見的缺點是，我們需要進行兩次模型呼叫而不是一次來生成我們的查詢。為了繞過這個問題，我們可以嘗試在單個模型調用中執行查詢生成和查詢檢查

system = """You are a {dialect} expert. Given an input question, create a syntactically correct {dialect} query to run.
Unless the user specifies in the question a specific number of examples to obtain, query for at most {top_k} results using the LIMIT clause as per {dialect}. You can order the results to return the most informative data in the database.
Never query for all columns from a table. You must query only the columns that are needed to answer the question. Wrap each column name in double quotes (") to denote them as delimited identifiers.
Pay attention to use only the column names you can see in the tables below. Be careful to not query for columns that do not exist. Also, pay attention to which column is in which table.
Pay attention to use date('now') function to get the current date, if the question involves "today".

Only use the following tables:
{table_info}

Write an initial draft of the query. Then double check the {dialect} query for common mistakes, including:
- Using NOT IN with NULL values
- Using UNION when UNION ALL should have been used
- Using BETWEEN for exclusive ranges
- Data type mismatch in predicates
- Properly quoting identifiers
- Using the correct number of arguments for functions
- Casting to the correct data type
- Using the proper columns for joins

Use format:

First draft: <<FIRST_DRAFT_QUERY>>
Final answer: <<FINAL_ANSWER_QUERY>>
"""
prompt = ChatPromptTemplate.from_messages(
    [("system", system), ("human", "{input}")]
).partial(dialect=db.dialect)


def parse_final_answer(output: str) -> str:
    return output.split("Final answer: ")[1]


chain = create_sql_query_chain(llm, db, prompt=prompt) | parse_final_answer
prompt.pretty_print()

================================ System Message ================================

You are a {dialect} expert. Given an input question, create a syntactically correct {dialect} query to run.
Unless the user specifies in the question a specific number of examples to obtain, query for at most {top_k} results using the LIMIT clause as per {dialect}. You can order the results to return the most informative data in the database.
Never query for all columns from a table. You must query only the columns that are needed to answer the question. Wrap each column name in double quotes (") to denote them as delimited identifiers.
Pay attention to use only the column names you can see in the tables below. Be careful to not query for columns that do not exist. Also, pay attention to which column is in which table.
Pay attention to use date('now') function to get the current date, if the question involves "today".

Only use the following tables:
{table_info}

Write an initial draft of the query. Then double check the {dialect} query for common mistakes, including:
- Using NOT IN with NULL values
- Using UNION when UNION ALL should have been used
- Using BETWEEN for exclusive ranges
- Data type mismatch in predicates
- Properly quoting identifiers
- Using the correct number of arguments for functions
- Casting to the correct data type
- Using the proper columns for joins

Use format:

First draft: <<FIRST_DRAFT_QUERY>>
Final answer: <<FINAL_ANSWER_QUERY>>


================================ Human Message =================================

{input}

query = chain.invoke(
    {
        "question": "What's the average Invoice from an American customer whose Fax is missing since 2003 but before 2010"
    }
)
print(query)

SELECT AVG(i."Total") AS "AverageInvoice"
FROM "Invoice" i
JOIN "Customer" c ON i."CustomerId" = c."CustomerId"
WHERE c."Country" = 'USA'
AND c."Fax" IS NULL
AND i."InvoiceDate" BETWEEN '2003-01-01' AND '2010-01-01';

db.run(query)

'[(6.632999999999998,)]'

人為介入環節

在某些情況下，我們的資料非常敏感，我們永遠不希望在沒有人批准的情況下執行 SQL 查詢。請前往工具使用：人為介入環節頁面，了解如何為任何工具、鏈或代理器添加人為介入環節。

錯誤處理

在某些時候，模型會犯錯並製作無效的 SQL 查詢。或者我們的資料庫會出現問題。或者模型 API 會當機。我們希望為我們的鏈和代理器添加一些錯誤處理行為，以便在這些情況下優雅地失敗，甚至可能自動恢復。要了解有關工具的錯誤處理，請前往工具使用：錯誤處理頁面。