当前位置：首页 > news >正文

构建AI智能体：三十四、LangChain SQLDatabaseToolkit终极指南：架构、优势与最佳实践

news 2025/9/16 9:16:34

一、初识SQLDatabaseToolkit

SQLDatabaseToolkit 是 LangChain 框架中的一个核心组件，它不属于一个独立的软件，而是一个工具箱或工具集。它的核心目的是为大语言模型提供与 SQL 数据库进行交互的能力，将大模型的自然语言理解能力与数据库的精准数据存储和检索能力结合起来。

二、什么是SQLDatabaseToolkit

1. 基础理解

SQLDatabaseToolkit 是一个封装好的工具集合，它包含了让大模型能够安全、有效地对 SQL 数据库执行查询、获取 schema 信息、检查查询结果等一系列操作所必需的工具。

我们可以把它想象成一个给大模型准备的“数据库操作瑞士军刀”，大模型在 LangChain 的协调下可以根据用户输入的提示请求，从这套工具箱里选择合适的工具来使用，最终完成用户的数据查询或操作任务。它的主要作用是作为大模型和 SQL 数据库之间的桥梁。

2. 使用说明

2.1 导入必要的库

from langchain.agents import create_sql_agent, AgentExecutor
from langchain.agents.agent_toolkits import SQLDatabaseToolkit
from langchain.sql_database import SQLDatabase
from langchain.llms.openai import OpenAI
from langchain.chat_models import ChatOpenAI
import os

create_sql_agent: 创建专门用于SQL数据库交互的代理
AgentExecutor: 代理执行器，负责运行代理
SQLDatabaseToolkit: 提供与SQL数据库交互的工具集
SQLDatabase: 表示SQL数据库连接
ChatOpenAI: OpenAI聊天模型类（这里用于兼容模式调用其他API）
os: 用于访问环境变量

2.2 连接到数据库

db_user = "root"
db_password = "Aa123456!"
db_host = "localhost"
db_name = "ecommerce_db"
db = SQLDatabase.from_uri(f"mysql+pymysql://{db_user}:{db_password}@{db_host}/{db_name}")
# 或者直接使用
# db = SQLDatabase.from_uri(f"mysql+pymysql://root:Aa123456!@localhost/ecommerce_db")

使用SQLAlchemy格式的URI连接MySQL数据库
格式: mysql+pymysql://用户名:密码@主机/数据库名
这里直接使用了硬编码的凭据，实际项目中应使用环境变量或配置管理

2.3 获取API密钥

api_key = os.environ.get('DASHSCOPE_API_KEY')

从环境变量中获取DashScope API密钥，避免将密钥硬编码在代码中

2.4 初始化LLM模型

llm = ChatOpenAI(temperature=0.01,model="deepseek-v3",  openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",openai_api_key=api_key
)

使用ChatOpenAI类，但配置为调用DashScope的兼容模式API
temperature=0.01: 低温度值使输出更加确定性和一致性
model="deepseek-v3": 指定使用的模型，也可以使用其他的，如qwen-max或qwen-turbo等
openai_api_base: 设置API端点为DashScope的兼容模式
openai_api_key: 提供API密钥进行认证

2.5 创建SQL工具箱

toolkit = SQLDatabaseToolkit(db=db, llm=llm)

创建SQLDatabaseToolkit实例，将数据库连接和LLM模型关联起来
工具箱会自动创建以下工具:
- ListSQLDatabaseTool: 列出数据库中的表
- InfoSQLDatabaseTool: 获取表结构信息
- QuerySQLDataBaseTool: 执行SQL查询

2.6 创建SQL代理执行器

agent_executor = create_sql_agent(llm=llm,toolkit=toolkit,verbose=False # 打印详细执行过程
)

创建一个专门用于SQL任务的代理
llm=llm: 指定使用的语言模型
toolkit=toolkit: 提供SQL工具箱
verbose=False: 不显示详细的执行过程，设置为True可用于调试

# 定义中文提示词！
# 重写原有的英文提示词前缀和后缀
ZH_SQL_PREFIX = """
你是一个专业的SQL数据分析师。你需要根据用户的问题，与一个SQL数据库进行交互，并返回答案。
请全程使用中文与用户交流和进行内部思考。
请严格按照以下步骤执行：
1. 首先，查看数据库中有哪些表可用。
2. 然后，根据需要查询相关表的schema信息，了解字段含义。
3. 根据schema生成一条正确的SQL查询语句。
4. 执行查询并分析结果。
5. 最后，用清晰、友好的中文回复用户的问题。请务必确保生成的SQL语句语法正确且符合数据库的schema。注意：如果你无法根据提供的信息得到答案，请如实告知用户，而不要编造答案。开始吧！
"""
# 后缀中需要保留`{tool_names}`、`{agent_scratchpad}`等变量，这是Agent工作所必需的。
ZH_SQL_SUFFIX = """
问题：{input}
思考过程：我需要一步步思考。首先，{agent_scratchpad}
"""# 5. 创建使用中文提示词的代理执行器
agent_executor = create_sql_agent(llm=llm,toolkit=toolkit,verbose=True,  # 打印详细执行过程，此时会是中文agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION, # 使用这种类型以便自定义提示词agent_executor_kwargs={"prefix": ZH_SQL_PREFIX,    # 注入我们的中文前缀"suffix": ZH_SQL_SUFFIX     # 注入我们的中文后缀}
)

注意这部分的差异，可以使用极简的模式，也可以自定义提示词前缀和后缀
LangChain 的 create_sql_agent 函数内部已经预设了一套提示词来指导Agent的行为。要切换语言，最有效的方法是创建一个自定义的代理（Agent），并为其提供中文的提示词模板。
使用 AgentType.ZERO_SHOT_REACT_DESCRIPTION 并自定义 prefix 和 suffix 部分，明确要求LLM使用中文进行思考和输出。

2.7 执行查询并获取结果

result = agent_executor.run("获取所有客户的姓名和联系电话")
print(result)

使用代理执行器运行用户输入的关键词
代理会分析查询意图、查看数据库中有哪些表、获取相关表的结构信息、生成合适的SQL查询、执行查询并返回结果、将结果进行整理后输出响应

三、SQLDatabaseToolkit的核心组成

SQLDatabaseToolkit 并不是一个单一的工具，它通常包含以下几个关键工具：

1. QuerySQLDataBaseTool

功能：这是最重要的工具，用于执行 SQL 查询语句并返回结果。
工作流程：大模型根据用户的问题和理解的数据表结构，生成一条 SQL 查询语句，然后通过这个工具来执行它，并将结果返回给用户。

运行示例：

# 示例: 使用QuerySQLDataBaseTool执行查询
def example_query_data():"""使用QuerySQLDataBaseTool执行查询"""print("=== 示例: 使用QuerySQLDataBaseTool执行查询 ===")    # 构建一个查询query = "SELECT * FROM customers LIMIT 5"result = query_tool.run(query)    # 使用Qwen API分析查询结果prompt = f"""数据库查询结果: {result}    请用中文分析这些数据，提供一些业务洞察。例如：1. 数据的基本统计信息2. 可能的业务含义3. 数据质量观察"""    analysis = call_qwen_api(prompt)print(f"查询结果: {result}")print(f"Qwen分析: {analysis}")print()

输出结果：

=== 示例: 使用QuerySQLDataBaseTool执行查询 ===
查询结果: [(1, '张三', 'zhangsan@example.com', '2023-01-15 10:30:45'), (2, '李四', 'lisi@example.com', '2023-02-20 14:25:30')]
Qwen分析: 查询返回了前5位客户的信息。从数据可以看出：
1. 系统至少有2位客户
2. 客户信息包含ID、姓名、邮箱和创建时间
3. 客户创建时间分布在2023年1月和2月
4. 邮箱格式符合常见电子邮件格式

2. InfoSQLDatabaseTool

功能：用于获取数据库的 Schema 信息，即表结构、列名、列数据类型等信息。
工作流程：在生成 SQL 语句之前，大模型通常需要知道数据库里有哪些表，每个表有哪些字段。这个工具可以查询指定表的 Schema（例如，通过 SELECT * FROM table_name LIMIT 0; 或类似的机制），为大模型生成准确的 SQL 提供上下文。它通常会限制返回的行数以避免信息过载。

运行示例：

# 示例: 使用InfoSQLDatabaseTool获取表结构信息
def example_table_info():"""使用InfoSQLDatabaseTool获取表结构"""print("=== 示例: 使用InfoSQLDatabaseTool获取表结构 ===")    # 假设我们有一个名为'customers'的表table_name = "customers"result = info_tool.run(table_name)    # 使用Qwen API解释表结构prompt = f"""表 '{table_name}' 的结构信息: {result}    请用中文解释这个表的结构，包括每个字段的用途和可能的数据关系。"""    explanation = call_qwen_api(prompt)print(f"表结构: {result}")print(f"Qwen分析: {explanation}")print()

输出结果：

=== 示例: 使用InfoSQLDatabaseTool获取表结构 ===
表结构: [('id', 'int', 'NO', 'PRI', None, 'auto_increment'), ('name', 'varchar(100)', 'YES', '', None, ''), ('email', 'varchar(100)', 'YES', '', None, ''), ('created_at', 'timestamp', 'YES', '', 'CURRENT_TIMESTAMP', '')]
Qwen分析: customers表用于存储客户基本信息，包含以下字段：
- id: 主键，自增整数，唯一标识每个客户
- name: 客户姓名，可变长度字符串
- email: 客户邮箱，可变长度字符串
- created_at: 客户创建时间，时间戳类型，默认当前时间

3. ListSQLDatabaseTool

功能：用于列出数据库中的所有表。
工作流程：当数据库很大时，大模型首先需要知道有哪些表可供查询。这个工具会返回一个表名的列表，如通过 SHOW TABLES;或查询 information_schema。

运行示例：

# 示例: 使用ListSQLDatabaseTool获取所有表名
def example_list_tables():"""使用ListSQLDatabaseTool获取所有表"""print("=== 示例1: 使用ListSQLDatabaseTool获取所有表 ===")    # 直接使用工具result = list_tool.run("")    # 使用Qwen API解释结果prompt = f"""数据库表列表: {result}    请用中文简要解释这个数据库可能包含哪些业务数据，并推测每个表的用途。"""    explanation = call_qwen_api(prompt)print(f"表列表: {result}")print(f"Qwen分析: {explanation}")print()

输出结果：

=== 示例: 使用ListSQLDatabaseTool获取所有表 ===
表列表: ['customers', 'orders', 'products', 'order_items']
Qwen分析: 这个数据库似乎是一个电子商务系统。包含以下表：
- customers: 存储客户信息
- orders: 存储订单信息
- products: 存储产品信息
- order_items: 存储订单项信息，连接订单和产品

4. ExecuteSQLDatabaseTool

功能：一个更通用的工具，可以执行任何 SQL 语句，包括 INSERT, UPDATE, DELETE 等写操作。
注意：出于安全考虑，这个工具通常默认不启用或需要显式配置。在绝大多数只读查询场景下，只会使用 QuerySQLDataBaseTool。

运行示例：

# 示例: 使用QuerySQLDataBaseTool执行查询
def example_query_data():"""使用QuerySQLDataBaseTool执行查询"""print("=== 示例: 使用QuerySQLDataBaseTool执行查询 ===")    # 构建一个查询query = "SELECT * FROM customers LIMIT 5"result = query_tool.run(query)    # 使用Qwen API分析查询结果prompt = f"""数据库查询结果: {result}    请用中文分析这些数据，提供一些业务洞察。例如：1. 数据的基本统计信息2. 可能的业务含义3. 数据质量观察"""    analysis = call_qwen_api(prompt)print(f"查询结果: {result}")print(f"Qwen分析: {analysis}")print()

输出结果：

四、SQLDatabaseToolkit的工作流程

1. 基础流程

基础的工作流程图，展示了其内部组件如何协作以及与外部组件如大模型、数据库的交互：

工作流步骤：

输入处理：
- 用户输入自然语言查询，如"显示销售额最高的产品"
- LangChain Agent接收并解析查询
数据库探索阶段：
- Agent调用ListSQLDatabaseTool获取可用表列表
- 根据查询意图，Agent选择相关表并调用InfoSQLDatabaseTool获取表结构
SQL生成与执行：
- Agent结合L大模型（如Qwen API）生成适当的SQL查询
- 调用QuerySQLDataBaseTool执行查询或ExecuteSQLDatabaseTool执行写操作
- 数据库返回查询结果
结果处理与响应：
- Agent使用LLM解释查询结果
- 将结构化数据转换为自然语言响应
- 向用户返回最终答案

2. 与Qwen API模型的集成流程

内部流程执行步骤：

当使用基于 SQLDatabaseToolkit 构建的 Agent智能体时，其内部的工作流程通常如下：

用户提问：用户提出一个问题，例如：“我们公司上个月销售额最高的产品是什么？”
Agent 思考：LangChain Agent配备了LLM和SQLDatabaseToolkit开始工作。
获取 Schema 信息：Agent 可能会先调用 ListSQLDatabaseTool 来查看有哪些表，然后调用 InfoSQLDatabaseTool 来查看可能相关的表（如 sales, products 表）的结构，了解有哪些字段（如 sale_date, product_id, amount）。
生成 SQL：LLM 基于用户的提问和刚刚获取到的 Schema 信息，推理并生成一条准确的 SQL 查询语句。如：SELECT product_name, SUM(amount) AS total_sales FROM sales JOIN products ON sales.product_id = products.id WHERE sale_date >= '2023-10-01' GROUP BY product_name ORDER BY total_sales DESC LIMIT 1;
执行查询：Agent 调用 QuerySQLDataBaseTool，将生成的 SQL 语句作为参数传入，工具在数据库上执行该查询。
处理结果：数据库返回查询结果，例如一个包含产品名和销售额的元组，Toolkit 将这个结果返回给 Agent。
生成最终回答：Agent 将结构化的数据库结果再次交给大模型，由大模型将其组织成流畅的自然语言答案返回给用户。如：“根据记录，上个月销售额最高的产品是 '某某智能音箱'，总销售额为 150,000 元。”

测试用例：查询销售额最高的产品

用户输入："哪些产品销售额最高？"
Agent调用ListSQLDatabaseTool，返回表列表: [products, orders, order_items]
Agent调用InfoSQLDatabaseTool，获取products和order_items表结构
Agent请求Qwen API生成SQL，生成: SELECT p.name, SUM(oi.quantity * oi.price) as total_sales FROM products p JOIN order_items oi ON p.id = oi.product_id GROUP BY p.id ORDER BY total_sales DESC LIMIT 10
Agent调用QuerySQLDataBaseTool，执行上述SQL
数据库返回：[('产品A', 15000), ('产品B', 12000), ...]
Agent请求Qwen API解释结果，生成自然语言响应
Agent返回用户："销售额最高的产品是产品A，总销售额为15,000元，其次是产品B，销售额为12,000元..."

3. 异常处理流程

主要异常类型及处理策略：

数据库连接异常：记录详细错误日志，返回友好提示
SQL 语法或执行异常：提供错误信息作为上下文，尝试让 LLM 重新生成 SQL
API 调用异常：记录 API 错误详情，实现备用 API 回退机制
权限异常：明确提示权限限制
超时异常：设置查询超时时间，返回友好提示

五、SQLDatabaseToolkit的主要特点

自然语言交互：让不懂 SQL 的用户也能通过提问的方式访问数据库数据。
自动化：自动处理了从理解问题、查询 Schema 到生成 SQL 和执行的全流程。
安全性：通过工具的设计，可以严格控制大模型对数据库的访问权限，默认只允许只读操作，避免意外的数据修改。
灵活性：可以连接到任何支持 SQL 的数据库，如 MySQL, PostgreSQL, SQLite, Microsoft SQL Server 等。
集成性：作为 LangChain 生态系统的一部分，可以轻松与其他工具（如链 Chains、代理 Agents）和组件（如记忆 Memory）结合，构建更复杂的应用。

六、SQLDatabaseToolkit的优势

在 SQLDatabaseToolkit 出现之前，让 LLM 操作数据库通常需要手动完成一系列繁琐步骤：

手动拼接提示词：开发者需要自己将数据库的 Schema（表名、列名、类型）信息拼接到给 LLM 的提示词中。
手动验证和执行 SQL：从 LLM 拿到生成的 SQL 后，需要自己编写代码连接数据库、执行查询、处理可能的错误（如语法错误、字段不存在）。
缺乏标准化流程：整个流程是分散且自定义的，难以复用、维护和优化。

SQLDatabaseToolkit 的出现，正是为了将这些碎片化的步骤标准化、自动化和工具化。它的核心意义在于：

降低门槛：让不熟悉编程和SQL的业务人员也能通过自然语言与数据库进行复杂交互，真正实现了“用说话的方式查数据”。
提升效率：为开发者和数据分析师提供了一个强大的脚手架，无需再从零开始构建数据库交互层，可以专注于更上层的业务逻辑和提示词工程。
推动AI应用落地：它将 LLM 的认知能力与数据库的海量结构化数据无缝衔接，是构建企业级智能数据分析、自动报表生成、客户洞察等应用的核心基础设施。

七、SQLDatabaseToolkit的使用场景

商业智能与数据分析：非技术人员（如产品经理、运营人员）直接通过对话查询业务数据，生成报表。
客户支持：客服人员快速查询用户订单状态、账户信息等。
内容生成：根据数据库中的事实数据自动生成报告、摘要或描述性文字。
企业知识库：将数据库中的结构化数据作为企业知识的一部分，供员工查询。

八、案例分析

1. 示例代码

from langchain.agents import create_sql_agent
from langchain.agents.agent_toolkits import SQLDatabaseToolkit
from langchain.sql_database import SQLDatabase
from langchain.llms.openai import OpenAI
from langchain.agents import AgentExecutor
from langchain.chat_models import ChatOpenAI
import os# 1. 连接到数据库
db = SQLDatabase.from_uri(f"mysql+pymysql://root:Aa123456!@localhost/ecommerce_db")# 从环境变量获取 dashscope 的 API Key
api_key = os.environ.get('DASHSCOPE_API_KEY')# 2. 选择LLM模型
llm = ChatOpenAI(temperature=0.01,model="deepseek-v3",  openai_api_base = "https://dashscope.aliyuncs.com/compatible-mode/v1",openai_api_key  = api_key
)
# 3. 创建SQL工具箱
toolkit = SQLDatabaseToolkit(db=db, llm=llm)# 4. 创建代理执行器
agent_executor = create_sql_agent(llm=llm,toolkit=toolkit,verbose=False # 打印详细执行过程
)# 5. 向代理提问！
result = agent_executor.run("获取所有客户的姓名和联系电话")
print(result)

2. 输出结果

代码中我们将执行器的 verbose=False 参数设置为false，没用输出过程：

result = agent_executor.run("获取所有客户的姓名和联系电话")
Here are the customer names and their contact emails (limited to 5 results):1. 张三 - zhangsan@example.com
2. 李四 - lisi@example.com
3. 王五 - wangwu@example.com
4. 赵六 - zhaoliu@example.com
5. 钱七 - qianqi@example.com

3. 执行过程

3.1 首先设置verbose = True，看看详细输出过程

result = agent_executor.run("获取所有客户的姓名和联系电话")
Error in StdOutCallbackHandler.on_chain_start callback: AttributeError("'NoneType' object has no attribute 'get'")
Action: sql_db_list_tables
Action Input: categories, order_items, orders, payments, products, user_addresses, usersThe most relevant tables for customer information would be "users" and possibly "user_addresses". I should check the schema of these tables to find
the columns for customer names and contact information.

Action: sql_db_schema
Action Input: users, user_addresses

... 忽略了schema的输出...

Action: sql_db_query_checker
Action Input: SELECT full_name, email FROM users LIMIT 10The original query is:

```sql
SELECT full_name, email FROM users LIMIT 10
```

After reviewing, I don't find any of the mentioned common mistakes in this simple query. The query is correctly formatted and doesn't involve any of the potential issues listed (NOT IN with NULL values, UNION vs UNION ALL, BETWEEN, data type mismatches, identifier quoting, function arguments, casting, or joins).

Here's the original query reproduced:

```sql
SELECT full_name, email FROM users LIMIT 10
```The query looks correct and safe to execute. I'll now run it to get the customer names and contact information.

Action: sql_db_query
Action Input: SELECT full_name, email FROM users LIMIT 10[('张三', 'zhangsan@example.com'), ('李四', 'lisi@example.com'), ('王五', 'wangwu@example.com'), ('赵六', 'zhaoliu@example.com'), ('钱七', 'qianqi@example.com')]I now have the list of customers with their names and contact information (email addresses).

Final Answer: 以下是客户的姓名和联系邮箱：
1. 张三 - zhangsan@example.com
2. 李四 - lisi@example.com
3. 王五 - wangwu@example.com
4. 赵六 - zhaoliu@example.com
5. 钱七 - qianqi@example.com

(注：数据库中未找到联系电话字段，故提供邮箱作为联系方式)

以上分析过程，关注两点：

1. 分析的过程都是英文的说明，这与使用的模型有关，有的模型是基于英文训练的，输出的内容多是英文，如果是支持中英文的，在提示词中要求使用中文分析也可以切换成中文输出
2. 我们要求输出的是名称和电话，输出的内容变成了名称和邮箱，也给了说明

3.2 我们换一个模型，改为Qwen-max，看看输出有什么不同

result = agent_executor.run("获取所有客户的姓名和联系电话")
Action: sql_db_list_tables
Action Input: categories, order_items, orders, payments, products, user_addresses, users从表名来看，`users` 表可能包含
客户的信息。我需要查看 `users` 表的结构来确定哪些列包含了客户的姓名和联系电话。
Action: sql_db_schema
Action Input: users

... 忽略了schema的输出...

*/从`users`表的结构来看，客户的姓名存储在`full_name`列中。但是，联系电话信息似乎没有直接出现在这个表里。考虑到联系信息
可能被单独存储，我需要检查一下`user_addresses`表是否包含电话号码。

Action: sql_db_schema
Action Input: user_addresses

... 忽略了schema的输出...

*/`user_addresses` 表中并没有直接包含电话号码的信息。考虑到联系信息可能存储在另一个表或者以某种形式存在于当前已知的表
中，但基于现有信息，我们只能获取客户的姓名。如果确实需要联系电话，可能需要进一步确认是否有其他相关表或字段存在。

不过，根据当前的需求，我可以先构建一个查询来获取所有客户的姓名。由于没有找到联系电话的具体位置，我将仅提供姓名信息，并
限制结果为10条记录。

Action: sql_db_query
Action Input: SELECT full_name FROM users LIMIT 10;[('张三',), ('李四',), ('王五',), ('赵六',), ('钱七',)]我已经从`users`表中获取了前10位客户的姓名。但是，由于数据库中没有直接提供联系电话的信息，我只能返回这些客户的姓名。

Final Answer: 以下是部分客户的姓名：张三、李四、王五、赵六、钱七。请注意，数据库中未找到与这些客户关联的联系电话信息。
如果需要联系电话，请进一步确认是否有其他相关表或字段存在。

> Finished chain.
以下是部分客户的姓名：张三、李四、王五、赵六、钱七。请注意，数据库中未找到与这些客户关联的联系电话信息。如果需要联系电
话，请进一步确认是否有其他相关表或字段存在。

分析过程是中文的描述，并且输出结果没有使用邮箱代替电话；