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

SQLDatabaseToolkit 是一个封装好的工具集合,它包含了让大模型能够安全、有效地对 SQL 数据库执行查询、获取 schema 信息、检查查询结果等一系列操作所必需的工具。
我们可以把它想象成一个给大模型准备的“数据库操作瑞士军刀”,大模型在 LangChain 的协调下可以根据用户输入的提示请求,从这套工具箱里选择合适的工具来使用,最终完成用户的数据查询或操作任务。它的主要作用是作为大模型和 SQL 数据库之间的桥梁。
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 os2.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")2.3 获取API密钥
api_key = os.environ.get('DASHSCOPE_API_KEY')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
)2.5 创建SQL工具箱
toolkit = SQLDatabaseToolkit(db=db, llm=llm)2.6 创建SQL代理执行器
agent_executor = create_sql_agent(
llm=llm,
toolkit=toolkit,
verbose=False # 打印详细执行过程
)# 定义中文提示词!
# 重写原有的英文提示词前缀和后缀
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 # 注入我们的中文后缀
}
)2.7 执行查询并获取结果
result = agent_executor.run("获取所有客户的姓名和联系电话")
print(result)SQLDatabaseToolkit 并不是一个单一的工具,它通常包含以下几个关键工具:
运行示例:
# 示例: 使用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. 邮箱格式符合常见电子邮件格式
运行示例:
# 示例: 使用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: 客户创建时间,时间戳类型,默认当前时间
运行示例:
# 示例: 使用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: 存储订单项信息,连接订单和产品
运行示例:
# 示例: 使用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. 邮箱格式符合常见电子邮件格式
基础的工作流程图,展示了其内部组件如何协作以及与外部组件如大模型、数据库的交互:

工作流步骤:

内部流程执行步骤:
当使用基于 SQLDatabaseToolkit 构建的 Agent智能体时,其内部的工作流程通常如下:
测试用例:查询销售额最高的产品

主要异常类型及处理策略:
在 SQLDatabaseToolkit 出现之前,让 LLM 操作数据库通常需要手动完成一系列繁琐步骤:
SQLDatabaseToolkit 的出现,正是为了将这些碎片化的步骤标准化、自动化和工具化。它的核心意义在于:
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)代码中我们将执行器的 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.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
(注:数据库中未找到联系电话字段,故提供邮箱作为联系方式)
以上分析过程,关注两点:
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. 以下是部分客户的姓名:张三、李四、王五、赵六、钱七。请注意,数据库中未找到与这些客户关联的联系电话信息。如果需要联系电 话,请进一步确认是否有其他相关表或字段存在。
分析过程是中文的描述,并且输出结果没有使用邮箱代替电话;
SQLDatabaseToolkit 是一个功能强大且实用的工具集,它极大地降低了通过自然语言访问和操作 SQL 数据库的门槛,是构建基于 LLM 的数据驱动应用的关键组件之一。使用 SQLDatabaseToolkit 自动生成数据库文档具有以下优势:
这种方法特别适用于大型项目、敏捷开发环境以及需要频繁更新数据库结构的场景,能够显著提高团队的生产力和文档质量。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。