SQL（结构化查询语言）

使用案例 (Use case)

企业数据通常存储在SQL数据库中。(Enterprise data is often stored in SQL databases.)

LLMs使得使用自然语言与SQL数据库进行交互成为可能。(LLMs make it possible to interact with SQL databases using natural language.)

LangChain提供SQL Chains和Agents，用于基于自然语言提示构建和运行SQL查询。(LangChain offers SQL Chains and Agents to build and run SQL queries based on natural language prompts.)

这些与SQLAlchemy支持的任何SQL方言兼容 (例如MySQL、PostgreSQL、Oracle SQL、Databricks、SQLite)。(These are compatible with any SQL dialect supported by SQLAlchemy (e.g., MySQL, PostgreSQL, Oracle SQL, Databricks, SQLite).)

它们可以实现以下使用案例：(They enable use cases such as:)

根据自然语言问题生成将要运行的查询 (Generating queries that will be run based on natural language questions)
根据数据库数据回答问题的聊天机器人 (Creating chatbots that can answer questions based on database data)
基于用户想要分析的洞察力构建自定义仪表板 (Building custom dashboards based on insights a user wants to analyze)

概述 (Overview)

LangChain 提供与 SQL 数据库交互的工具:

基于自然语言用户问题构建 SQL 查询 (Build SQL queries)
使用链式查询创建和执行 SQL 数据库查询 (Query a SQL database)
使用代理进行强大而灵活的 SQL 数据库查询交互 (Interact with a SQL database)

快速入门 (Quickstart)

首先，获取所需的软件包并设置环境变量：

pip install langchain langchain-experimental openai

# 设置环境变量 OPENAI_API_KEY 或从 .env 文件加载
# import dotenv

# dotenv.load_env()

下面的示例将使用一个带有 Chinook 数据库的 SQLite 连接。

按照安装步骤在与此笔记本相同的目录中创建 Chinook.db：

将此文件保存到目录中，命名为 Chinook_Sqlite.sql
运行 sqlite3 Chinook.db
运行 .read Chinook_Sqlite.sql
测试 SELECT * FROM Artist LIMIT 10;

现在，Chinhook.db 在我们的目录中。

让我们创建一个 SQLDatabaseChain 来创建和执行 SQL 查询。

from langchain.utilities import SQLDatabase
from langchain.llms import OpenAI
from langchain_experimental.sql import SQLDatabaseChain

db = SQLDatabase.from_uri("sqlite:///Chinook.db")
llm = OpenAI(temperature=0, verbose=True)
db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True)

db_chain.run("有多少员工？")

    
    > 进入新的 SQLDatabaseChain 链...
    有多少员工？
    SQLQuery:SELECT COUNT(*) FROM "Employee";
    SQLResult: [(8,)]
    Answer:有 8 名员工。
    > 完成链。

    '有 8 名员工。'

请注意，这既创建了查询，也执行了查询。

在接下来的几节中，我们将介绍概述中提到的 3 种不同的用例。

深入了解

您可以从除了SQL数据库之外的其他来源加载表格数据。例如：

加载CSV文件
加载Pandas DataFrame 您可以在这里查看完整的文档加载器列表

案例1：文本到SQL查询

from langchain.chat_models import ChatOpenAI
from langchain.chains import create_sql_query_chain

让我们创建一个链来构建SQL查询：

chain = create_sql_query_chain(ChatOpenAI(temperature=0), db)
response = chain.invoke({"question":"有多少员工"})
print(response)

    SELECT COUNT(*) FROM Employee

在根据用户问题构建SQL查询之后，我们可以执行查询：

db.run(response)

    '[(8,)]'

正如我们所看到的，SQL查询构建器链仅仅创建了查询，而我们将查询执行分开处理。

深入了解

查看内部机制

我们可以查看LangSmith跟踪来解析这个问题：

一些论文报告了在以下提示下的良好性能：

每个表的CREATE TABLE描述，包括列名、类型等
随后是一个SELECT语句中的三个示例行

create_sql_query_chain采用了这个最佳实践（在这个博客中可以了解更多）。

改进

查询构建器可以通过多种方式进行改进，例如（但不限于）：

根据特定用例自定义数据库描述
在提示中硬编码一些问题及其相应的SQL查询示例
使用向量数据库包含与特定用户问题相关的动态示例

所有这些示例都涉及自定义链的提示。

例如，我们可以在提示中包含一些示例，如下所示：

from langchain.prompts import PromptTemplate

TEMPLATE = """给定一个输入问题，首先创建一个语法正确的{dialect}查询来运行，然后查看查询的结果并返回答案。
使用以下格式：

问题："问题在这里"
SQL查询："要运行的SQL查询"
SQL结果："SQL查询的结果"
答案："最终答案在这里"

只使用以下表格：

{table_info}。

一些与问题对应的SQL查询示例：

{few_shot_examples}

问题：{input}"""

CUSTOM_PROMPT = PromptTemplate(
    input_variables=["input", "few_shot_examples", "table_info", "dialect"], template=TEMPLATE
)

案例2：文本到SQL查询和执行 (Case 2: Text-to-SQL query and execution)

我们可以使用langchain_experimental中的SQLDatabaseChain来创建和运行SQL查询。

from langchain.llms import OpenAI
from langchain_experimental.sql import SQLDatabaseChain

llm = OpenAI(temperature=0, verbose=True)
db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True)

db_chain.run("有多少名员工？")

    
    > 进入新的SQLDatabaseChain链...
    有多少名员工？
    SQL查询：SELECT COUNT(*) FROM "Employee";
    SQL结果：[(8,)]
    答案：有8名员工。
    > 完成链的执行。

    '有8名员工。'

正如我们所看到的，我们得到了与前一个案例相同的结果。

在这里，该链还处理了查询的执行，并根据用户的问题和查询结果提供了最终答案。

在使用这种方法时要小心，因为它容易受到SQL注入的攻击：

该链执行由LLM创建的查询，并且没有经过验证
例如，记录可能会被意外创建、修改或删除

这就是为什么我们看到SQLDatabaseChain位于langchain_experimental中的原因。

深入了解

查看内部情况

我们可以使用LangSmith跟踪来查看内部情况：

如上所述，首先我们创建查询：

text: ' SELECT COUNT(*) FROM "Employee";'

然后，它执行查询并将结果传递给LLM进行合成。

改进

SQLDatabaseChain的性能可以通过以下几种方式进行提升：

添加示例行
指定自定义表信息
使用查询检查器使用参数use_query_checker=True自动纠正无效的SQL
自定义LLM提示使用参数prompt=CUSTOM_PROMPT包含特定的指令或相关信息
获取中间步骤使用参数return_intermediate_steps=True访问SQL语句以及最终结果
限制行数使用参数top_k=5限制查询返回的行数

对于数据库中表的数量较多的情况，您可能会发现SQLDatabaseSequentialChain很有用。

这个Sequential Chain处理的过程包括：

根据用户的问题确定要使用的表
仅使用相关的表调用正常的SQL数据库链

添加示例行

当数据格式不明显时，提供示例数据可以帮助LLM构建正确的查询。

例如，我们可以告诉LLM艺术家的全名保存在Track表中，通过提供两行数据。

db = SQLDatabase.from_uri(
    "sqlite:///Chinook.db",
    include_tables=['Track'], # 我们只包含一个表以在提示中保存令牌 :)
    sample_rows_in_table_info=2)

示例行将添加到每个相应表的列信息之后。

我们可以使用db.table_info检查包含了哪些示例行：

print(db.table_info)

    
    CREATE TABLE "Track" (
        "TrackId" INTEGER NOT NULL, 
        "Name" NVARCHAR(200) NOT NULL, 
        "AlbumId" INTEGER, 
        "MediaTypeId" INTEGER NOT NULL, 
        "GenreId" INTEGER, 
        "Composer" NVARCHAR(220), 
        "Milliseconds" INTEGER NOT NULL, 
        "Bytes" INTEGER, 
        "UnitPrice" NUMERIC(10, 2) NOT NULL, 
        PRIMARY KEY ("TrackId"), 
        FOREIGN KEY("MediaTypeId") REFERENCES "MediaType" ("MediaTypeId"), 
        FOREIGN KEY("GenreId") REFERENCES "Genre" ("GenreId"), 
        FOREIGN KEY("AlbumId") REFERENCES "Album" ("AlbumId")
    )
    
    /*
    2 rows from Track table:
    TrackId Name    AlbumId MediaTypeId GenreId Composer    Milliseconds    Bytes   UnitPrice
    1   For Those About To Rock (We Salute You) 1   1   1   Angus Young, Malcolm Young, Brian Johnson   343719  11170334    0.99
    2   Balls to the Wall   2   2   1   None    342562  5510424 0.99
    */

情况3：SQL代理

LangChain拥有一个SQL代理，它提供了一种比SQLDatabaseChain更灵活的与SQL数据库交互的方式。

使用SQL代理的主要优势包括：

它可以根据数据库的模式和内容（例如描述特定表）回答问题
它可以通过运行生成的查询、捕获回溯并正确地重新生成来从错误中恢复

要初始化代理，我们使用create_sql_agent函数。

该代理包含SQLDatabaseToolkit，其中包含以下工具：

创建和执行查询
检查查询语法
检索表描述
...等等

from langchain.agents import create_sql_agent
from langchain.agents.agent_toolkits import SQLDatabaseToolkit
# from langchain.agents import AgentExecutor
from langchain.agents.agent_types import AgentType

db = SQLDatabase.from_uri("sqlite:///Chinook.db")
llm = OpenAI(temperature=0, verbose=True)

agent_executor = create_sql_agent(
    llm=OpenAI(temperature=0),
    toolkit=SQLDatabaseToolkit(db=db, llm=OpenAI(temperature=0)),
    verbose=True,
    agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
)

from langchain.agents import create_sql_agent
from langchain.agents.agent_toolkits import SQLDatabaseToolkit
# from langchain.agents import AgentExecutor
from langchain.agents.agent_types import AgentType

db = SQLDatabase.from_uri("sqlite:///Chinook.db")
llm = OpenAI(temperature=0, verbose=True)

agent_executor = create_sql_agent(
    llm=OpenAI(temperature=0),
    toolkit=SQLDatabaseToolkit(db=db, llm=OpenAI(temperature=0)),
    verbose=True,
    agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
)

代理任务示例＃1 - 运行查询 (Agent task example #1 - Running queries)

agent_executor.run(
    "列出每个国家的总销售额。哪个国家的客户花费最多？"
)

    
    
    > 进入新的AgentExecutor链...
    动作：sql_db_list_tables
    动作输入： 
    观察结果：Album, Artist, Customer, Employee, Genre, Invoice, InvoiceLine, MediaType, Playlist, PlaylistTrack, Track
    思考：我应该查询Invoice和Customer表的模式。
    动作：sql_db_schema
    动作输入：Invoice, Customer
    观察结果： 
    CREATE TABLE "Customer" (
        "CustomerId" INTEGER NOT NULL, 
        "FirstName" NVARCHAR(40) NOT NULL, 
        "LastName" NVARCHAR(20) NOT NULL, 
        "Company" NVARCHAR(80), 
        "Address" NVARCHAR(70), 
        "City" NVARCHAR(40), 
        "State" NVARCHAR(40), 
        "Country" NVARCHAR(40), 
        "PostalCode" NVARCHAR(10), 
        "Phone" NVARCHAR(24), 
        "Fax" NVARCHAR(24), 
        "Email" NVARCHAR(60) NOT NULL, 
        "SupportRepId" INTEGER, 
        PRIMARY KEY ("CustomerId"), 
        FOREIGN KEY("SupportRepId") REFERENCES "Employee" ("EmployeeId")
    )
    
    /*
    Customer表的3行数据:
    CustomerId  FirstName   LastName    Company Address City    State   Country PostalCode  Phone   Fax Email   SupportRepId
    1   Luís    Gonçalves   Embraer - Empresa Brasileira de Aeronáutica S.A.    Av. Brigadeiro Faria Lima, 2170 São José dos Campos SP  Brazil  12227-000   +55 (12) 3923-5555  +55 (12) 3923-5566  luisg@embraer.com.br    3
    2   Leonie  Köhler  None    Theodor-Heuss-Straße 34 Stuttgart   None    Germany 70174   +49 0711 2842222    None    leonekohler@surfeu.de   5
    3   François    Tremblay    None    1498 rue Bélanger   Montréal    QC  Canada  H2G 1A7 +1 (514) 721-4711   None    ftremblay@gmail.com 3
    */
    
    
    CREATE TABLE "Invoice" (
        "InvoiceId" INTEGER NOT NULL, 
        "CustomerId" INTEGER NOT NULL, 
        "InvoiceDate" DATETIME NOT NULL, 
        "BillingAddress" NVARCHAR(70), 
        "BillingCity" NVARCHAR(40), 
        "BillingState" NVARCHAR(40), 
        "BillingCountry" NVARCHAR(40), 
        "BillingPostalCode" NVARCHAR(10), 
        "Total" NUMERIC(10, 2) NOT NULL, 
        PRIMARY KEY ("InvoiceId"), 
        FOREIGN KEY("CustomerId") REFERENCES "Customer" ("CustomerId")
    )
    
    /*
    Invoice表的3行数据:
    InvoiceId   CustomerId  InvoiceDate BillingAddress  BillingCity BillingState    BillingCountry  BillingPostalCode   Total
    1   2   2009-01-01 00:00:00 Theodor-Heuss-Straße 34 Stuttgart   None    Germany 70174   1.98
    2   4   2009-01-02 00:00:00 Ullevålsveien 14    Oslo    None    Norway  0171    3.96
    3   8   2009-01-03 00:00:00 Grétrystraat 63 Brussels    None    Belgium 1000    5.94
    */
    思考：我应该查询每个国家的总销售额。
    动作：sql_db_query
    动作输入：SELECT Country, SUM(Total) AS TotalSales FROM Invoice INNER JOIN Customer ON Invoice.CustomerId = Customer.CustomerId GROUP BY Country ORDER BY TotalSales DESC LIMIT 10
    观察结果：[('USA', 523.0600000000003), ('Canada', 303.9599999999999), ('France', 195.09999999999994), ('Brazil', 190.09999999999997), ('Germany', 156.48), ('United Kingdom', 112.85999999999999), ('Czech Republic', 90.24000000000001), ('Portugal', 77.23999999999998), ('India', 75.25999999999999), ('Chile', 46.62)]
    思考：我现在知道最终答案了
    最终答案：销售额最高的国家是美国，总销售额为523.06美元。
    
    > 链结束。





    '销售额最高的国家是美国，总销售额为523.06美元。'

根据LangSmith跟踪，我们可以看到：

代理使用了ReAct风格的提示
首先，它会查看表格：动作：sql_db_list_tables 使用工具 sql_db_list_tables
给出表格作为观察结果，它进行了思考，然后确定下一个动作：

观察结果：Album, Artist, Customer, Employee, Genre, Invoice, InvoiceLine, MediaType, Playlist, PlaylistTrack, Track
思考：我应该查询Invoice和Customer表的模式。
动作：sql_db_schema
动作输入：Invoice, Customer

然后，它使用来自工具 sql_db_schema 的模式生成查询：

思考：我应该查询每个国家的总销售额。
动作：sql_db_query
动作输入：SELECT Country, SUM(Total) AS TotalSales FROM Invoice INNER JOIN Customer ON Invoice.CustomerId = Customer.CustomerId GROUP BY Country ORDER BY TotalSales DESC LIMIT 10

最后，它使用工具 sql_db_query 执行生成的查询

代理任务示例 #2 - 描述一个表 (Agent task example #2 - Describing a Table)

agent_executor.run("描述 playlisttrack 表")

    
    
    > 进入新的 AgentExecutor 链...
    动作: sql_db_list_tables
    动作输入: 
    观察结果: Album, Artist, Customer, Employee, Genre, Invoice, InvoiceLine, MediaType, Playlist, PlaylistTrack, Track
    思考: PlaylistTrack 表与问题最相关。
    动作: sql_db_schema
    动作输入: PlaylistTrack
    观察结果: 
    CREATE TABLE "PlaylistTrack" (
        "PlaylistId" INTEGER NOT NULL, 
        "TrackId" INTEGER NOT NULL, 
        PRIMARY KEY ("PlaylistId", "TrackId"), 
        FOREIGN KEY("TrackId") REFERENCES "Track" ("TrackId"), 
        FOREIGN KEY("PlaylistId") REFERENCES "Playlist" ("PlaylistId")
    )
    
    /*
    从 PlaylistTrack 表中获取 3 行数据:
    PlaylistId  TrackId
    1   3402
    1   3389
    1   3390
    */
    思考: 我现在知道最终答案了
    最终答案: PlaylistTrack 表包含两列，PlaylistId 和 TrackId，它们都是整数并且组成了一个主键。它还有两个外键，一个指向 Track 表，一个指向 Playlist 表。
    
    > 链结束。





    'PlaylistTrack 表包含两列，PlaylistId 和 TrackId，它们都是整数并且组成了一个主键。它还有两个外键，一个指向 Track 表，一个指向 Playlist 表。'

深入了解

要了解更多关于SQL Agent及其工作原理的信息，请参阅SQL Agent Toolkit文档。

您还可以查看其他文档类型的代理程序：

与Elastic Search集成

超越上述用例，还可以与其他数据库进行集成。

例如，我们可以与 Elasticsearch 分析数据库进行交互。

该链通过 Elasticsearch DSL API（过滤器和聚合）构建搜索查询。

Elasticsearch 客户端必须具有索引列表、映射描述和搜索查询的权限。

请参阅此处以了解如何在本地运行 Elasticsearch 的说明。

在此之前，请确保安装 Elasticsearch Python 客户端：

pip install elasticsearch

from elasticsearch import Elasticsearch
from langchain.chat_models import ChatOpenAI
from langchain.chains.elasticsearch_database import ElasticsearchDatabaseChain

# 初始化 Elasticsearch Python 客户端。
# 请参阅 https://elasticsearch-py.readthedocs.io/en/v8.8.2/api.html#elasticsearch.Elasticsearch
ELASTIC_SEARCH_SERVER = "https://elastic:pass@localhost:9200"
db = Elasticsearch(ELASTIC_SEARCH_SERVER)

取消注释下一个单元格以初始化填充数据库。

# customers = [
#     {"firstname": "Jennifer", "lastname": "Walters"},
#     {"firstname": "Monica","lastname":"Rambeau"},
#     {"firstname": "Carol","lastname":"Danvers"},
#     {"firstname": "Wanda","lastname":"Maximoff"},
#     {"firstname": "Jennifer","lastname":"Takeda"},
# ]
# for i, customer in enumerate(customers):
#     db.create(index="customers", document=customer, id=i)

llm = ChatOpenAI(model_name="gpt-4", temperature=0)
chain = ElasticsearchDatabaseChain.from_llm(llm=llm, database=db, verbose=True)

question = "What are the first names of all the customers?"
chain.run(question)

我们可以自定义提示。

from langchain.chains.elasticsearch_database.prompts import DEFAULT_DSL_TEMPLATE
from langchain.prompts.prompt import PromptTemplate

PROMPT_TEMPLATE = """给定一个输入问题，创建一个语法正确的 Elasticsearch 查询来运行。除非用户在问题中指定了他们希望获得的特定数量的示例，否则将查询限制为最多 {top_k} 个结果。您可以按照相关列对结果进行排序，以返回数据库中最有趣的示例。

除非另有指示，否则不要查询特定索引中的所有列，只查询与问题相关的少数几个列。

注意只使用映射描述中可见的列名。注意不要查询不存在的列。还要注意哪个列位于哪个索引中。将查询返回为有效的 JSON。

使用以下格式：

问题：在此处提出问题
ESQuery：Elasticsearch 查询，格式为 JSON
"""

PROMPT = PromptTemplate.from_template(
    PROMPT_TEMPLATE,
)
chain = ElasticsearchDatabaseChain.from_llm(llm=llm, database=db, query_prompt=PROMPT)

SQL（结构化查询语言）

使用案例 (Use case)​

概述 (Overview)​

快速入门 (Quickstart)​

深入了解​

案例1：文本到SQL查询​

深入了解​

案例2：文本到SQL查询和执行 (Case 2: Text-to-SQL query and execution)​

深入了解​

情况3：SQL代理​

代理任务示例＃1 - 运行查询 (Agent task example #1 - Running queries)​

代理任务示例 #2 - 描述一个表 (Agent task example #2 - Describing a Table)​

深入了解​

与Elastic Search集成​

使用案例 (Use case)

概述 (Overview)

快速入门 (Quickstart)

深入了解

案例1：文本到SQL查询

深入了解

案例2：文本到SQL查询和执行 (Case 2: Text-to-SQL query and execution)

深入了解

情况3：SQL代理

代理任务示例＃1 - 运行查询 (Agent task example #1 - Running queries)

代理任务示例 #2 - 描述一个表 (Agent task example #2 - Describing a Table)

深入了解

与Elastic Search集成