LangChain之数据库操作：通过链Chain和代理Agent查询数据库信息_langchain 数据库

Chain和Agent查询数据库

在LangChain，其提供了SQL Chain链和SQL Agent代理，他们支持基于自然语言提示构建和运行SQL查询，以此来操作数据库，并且与SQLAlchemy支持的任何SQL版本兼容（例如，MySQL，PostgreSQL，Oracle SQL，Databricks，SQLite）。

可以用来完成以下任务：

基于自然语言问题生成能运行的SQL查询

创建可以基于数据库数据回答问题的聊天机器人

根据用户想要分析的见解构建自定义仪表板
1
2
3
4
5

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

create_sql_query_chain：基于用户自然语言问题构建SQL查询

SQLDatabaseChain：使用链进行查询、创建和执行来查询SQL数据库

create_sql_agent：使用代理进行健壮和灵活的与SQL数据库交互
1
2
3
4
5

准备数据库、表

使用LangChain中的SQLDatabase工具来与SQL数据库进行交互，首先准备一些测试数据。

基于SQLite数据库

基于SQLite数据库进行基本操作：创建表、插入测试数据以及查询插入数据

# 导入sqlite3库，一个Python内置的轻量级数据库
import sqlite3

print("------------------------连接数据库------------------------")
# 连接数据库
conn = sqlite3.connect('demo.db')
cursor = conn.cursor()

print("------------------------数据表创建------------------------")

# 执行SQL命令 创建User表
cursor.execute('''
        CREATE TABLE User (
            ID INTEGER PRIMARY KEY, 
            Name TEXT NOT NULL, 
            Age INT,
            Money REAL,
            CreateDate DATE DEFAULT CURRENT_DATE 
        );
    ''')

print("------------------------数据插入------------------------")
# 插入数据
users = [
    ('Java', 10, 20.5),
    ('Python', 20, 0),
    ('Vue', 33, 55.0),
    ('Go', 5, 100.55),
]

for user in users:
    cursor.execute('''
        INSERT INTO User (Name, Age, Money) 
        VALUES (?, ?, ?);
    ''', user)

print("------------------------查询数据------------------------")
users = cursor.execute("SELECT *  from User")
for row in users:
    print(row)

# 提交更改
conn.commit()

# 关闭数据库连接
conn.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

基于MySQL数据库

创建数据库：demo

CREATE DATABASE demo
  CHARACTER SET utf8mb4
  COLLATE utf8mb4_general_ci;
1
2
3

创建数据表：tb_users

CREATE TABLE `tb_users` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(255) NOT NULL,
  `age` int(11) DEFAULT NULL,
  `money` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=12 DEFAULT CHARSET=utf8
1
2
3
4
5
6
7

插入几条测试数据

INSERT INTO `demo`.`tb_users` (`id`, `name`, `age`, `money`) VALUES (1, 'Java', 22, '100');
INSERT INTO `demo`.`tb_users` (`id`, `name`, `age`, `money`) VALUES (2, 'Python', 30, '200');
INSERT INTO `demo`.`tb_users` (`id`, `name`, `age`, `money`) VALUES (3, 'Goi', 11, '300');
1
2
3

SQLDatabase工具

在 LangChain 中,SQLDatabase工具可以用来与SQL数据库进行交互。SQLDatabase是数据库连接的包装器，为了与SQL数据库通信，它使用SQLAlchemy Core API 。

初始化数据库

首先连接到sqlite数据库，进行数据库初始化

from langchain_community.utilities import SQLDatabase
import sqlalchemy as sa

# 连接到demo数据库
db = SQLDatabase.from_uri("sqlite:///demo.db")
1
2
3
4
5

光标查询​

查询模式:cursor，将结果作为SQLAlchemy的CursorResult实例返回

result = db.run("SELECT * FROM User LIMIT 10;", fetch="cursor")
print(type(result))
print(list(result.mappings()))
1
2
3

<class 'sqlalchemy.engine.cursor.CursorResult'>
[{'ID': 1, 'Name': 'Java', 'Age': 10, 'Money': 20.5, 'CreateDate': '2024-04-14'},
{'ID': 2, 'Name': 'Python', 'Age': 20, 'Money': 0.0, 'CreateDate': '2024-04-14'}, 
{'ID': 3, 'Name': 'Vue', 'Age': 33, 'Money': 55.0, 'CreateDate': '2024-04-14'}, 
{'ID': 4, 'Name': 'Go', 'Age': 5, 'Money': 100.55, 'CreateDate': '2024-04-14'}]
1
2
3
4
5

字符串查询

查询模式：all 和one，以字符串格式返回结果。

result = db.run("SELECT * FROM User LIMIT 10;", fetch="all")
print(type(result))
print(result)
1
2
3

<class 'str'>
[(1, 'Java', 10, 20.5, '2024-04-14'), (2, 'Python', 20, 0.0, '2024-04-14'), (3, 'Vue', 33, 55.0, '2024-04-14'), (4, 'Go', 5, 100.55, '2024-04-14')]
1
2

带参数查询​

可以使用parameters可选参数，来绑定查询参数

result = db.run(
    "SELECT * FROM User WHERE Name LIKE :search;",
    parameters={"search": "P%"},
    fetch="cursor",
)
print(list(result.mappings()))
1
2
3
4
5
6

[{'ID': 2, 'Name': 'Python', 'Age': 20, 'Money': 0.0, 'CreateDate': '2024-04-14'}]
1

使用SQLAlchemy查询​

SQLDatabase工具除了使用纯文本SQL语句之外，其适配器还接受 SQLAlchemy 可选择项。

import sqlalchemy as sa
# 为了在sqlalchemy 的 Core API 上构建一个可选择的，需要一个表的定义。
metadata = sa.MetaData()
user = sa.Table(
    "User",
    metadata,
    sa.Column("Id", sa.INTEGER, primary_key=True),
    sa.Column("Name", sa.TEXT),
    sa.Column("Age", sa.INT),
    sa.Column("Money", sa.REAL),
)

# 构建一个查询语义
query = sa.select(user).where(user.c.Name.like("J%"))
# 执行查询
result = db.run(query, fetch="cursor")
print(list(result.mappings()))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

[{'Id': 1, 'Name': 'Java', 'Age': 10, 'Money': 20.5}]
1

create_sql_query_chain：查询数据库

create_sql_query_chain是一个根据自然语言创建生成SQL查询的链。

示例代码

from langchain_community.utilities import SQLDatabase
from langchain.chains.sql_database.query import create_sql_query_chain

# 连接 MySQL 数据库
db_user = "root"
db_password = "12345678"
db_host = "IP"
db_port = "3306"
db_name = "demo"
db = SQLDatabase.from_uri(f"mysql+pymysql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}")

print("数据库方言：", db.dialect)
print("获取数据表：", db.get_usable_table_names())

# 执行查询
res = db.run("SELECT count(*) FROM tb_users;")
print(type(res))
print("查询结果：", res)

from langchain_openai import ChatOpenAI

# 初始化大模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
chain = create_sql_query_chain(llm=llm, db=db)
response = chain.invoke({"question": "查询数据表tb_users中的用户,要求年龄大于20"})
print("Chain执行结果："+ response)

# 删除response无用部分
sql = response.replace("sql: ", "").replace("```sql", "").replace("```", "")
print("自然语言转SQL：" + sql)
res = db.run(sql)
print("查询结果：", res)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

执行测试

SQLDatabaseChain：查询数据库

langchain-experimental库包含实验性的LangChain新功能，这里使用该库的SQLDatabaseChain来实现查询数据库信息

需要安装langchain-experimental库

pip install langchain-experimental
1

示例代码

# 导入OpenAI信息
import os

os.environ["OPENAI_BASE_URL"] = "https://xxx.com/v1"
os.environ["OPENAI_API_KEY"] = "sk-fDqouTlU62yjkBhF46284543Dc8f42438a9529Df74B4Ce65"

# 导入langchain的实用工具和相关的模块
from langchain_community.utilities import SQLDatabase
from langchain_experimental.sql import SQLDatabaseChain

# 连接到demo数据库
db = SQLDatabase.from_uri("sqlite:///demo.db")

from langchain_openai import ChatOpenAI

# 创建LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, verbose=True)

# # 创建一个生成 SQL 查询的链
db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True)

# 运行查询问题
response = db_chain.invoke("总共有多少用户？")
print(response)

response = db_chain.invoke({"query": "哪个用户的年龄最大？"})
print(response)

response = db_chain.invoke({"query": "所有用户的金额合计是多少？"})
print(response)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

执行测试

执行日志如下，回答的非常好

> Entering new SQLDatabaseChain chain...
总共有多少用户？
SQLQuery:SELECT COUNT(*) FROM "User"
SQLResult: [(4,)]
Answer:总共有4个用户。
> Finished chain.
{'query': '总共有多少用户？', 'result': '总共有4个用户。'}


> Entering new SQLDatabaseChain chain...
哪个用户的年龄最大？
SQLQuery:SELECT "Name", MAX("Age") FROM "User"
SQLResult: [('Vue', 33)]
Answer:Vue
> Finished chain.
{'query': '哪个用户的年龄最大？', 'result': 'Vue'}


> Entering new SQLDatabaseChain chain...
所有用户的金额合计是多少？
SQLQuery:SELECT SUM("Money") FROM "User"
SQLResult: [(176.05,)]
Answer:所有用户的金额合计是176.05。
> Finished chain.
{'query': '所有用户的金额合计是多少？', 'result': '所有用户的金额合计是176.05。'}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

自定义prompt提示模板

可以自定义所使用的prompt提示模板，这是使用官方的一个prompt示例

# 导入langchain的实用工具和相关的模块
from langchain_community.utilities import SQLDatabase
from langchain_experimental.sql import SQLDatabaseChain

# 连接到demo数据库
db = SQLDatabase.from_uri("sqlite:///demo.db")

from langchain_openai import ChatOpenAI

# 创建LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, verbose=True)

from langchain_core.prompts import PromptTemplate

template = '''Given an input question, first create a syntactically correct {dialect} query to run, then look at the results of the query and return the answer.
Use the following format:

Question: "Question here"
SQLQuery: "SQL Query to run"
SQLResult: "Result of the SQLQuery"
Answer: "Final answer here"

Only use the following tables:

{table_info}.

Question: {input}'''

prompt = PromptTemplate(
    input_variables=["dialect", "table_info", "input"],
    template=template
)
# 创建一个生成 SQL 查询的链
db_chain = SQLDatabaseChain.from_llm(llm, db, prompt=prompt, verbose=True)

# 运行查询问题
response = db_chain.invoke("总共有多少用户？")
print(response)

response = db_chain.invoke({"query": "哪个用户的年龄最大？"})
print(response)

response = db_chain.invoke({"query": "所有用户的金额合计是多少？"})
print(response)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

> Entering new SQLDatabaseChain chain...
总共有多少用户？
SQLQuery:SELECT COUNT(*) FROM User
SQLResult: [(4,)]
Answer:Final answer here: 4
> Finished chain.
{'query': '总共有多少用户？', 'result': 'Final answer here: 4'}

> Entering new SQLDatabaseChain chain...
哪个用户的年龄最大？
SQLQuery:SELECT Name FROM User ORDER BY Age DESC LIMIT 1
SQLResult: [('Vue',)]
Answer:Final answer here: Vue
> Finished chain.
{'query': '哪个用户的年龄最大？', 'result': 'Final answer here: Vue'}


> Entering new SQLDatabaseChain chain...
所有用户的金额合计是多少？
SQLQuery:SELECT SUM(Money) FROM User
SQLResult: [(176.05,)]
Answer:Final answer here: 176.05
> Finished chain.
{'query': '所有用户的金额合计是多少？', 'result': 'Final answer here: 176.05'}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

使用查询检查器

有时语言模型可能生成带有小错误的无效SQL，这些错误可以尝试使用use_query_checker=True参数让LLM修复SQL。

# # 创建一个生成 SQL 查询的链
db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True, use_query_checker=True)
1
2

返回中间步骤

可以使用return_intermediate_steps=True参数返回 SQLDatabaseChain 的中间步骤，这允许访问生成的 SQL 语句以及针对 SQL 数据库运行该语句的结果。

# 创建一个生成 SQL 查询的链
db_chain = SQLDatabaseChain.from_llm(llm, db, prompt=prompt, return_intermediate_steps=True)

# 运行查询问题
response = db_chain.invoke({"query": "总共有多少用户？"})
print(response["intermediate_steps"])
1
2
3
4
5
6

[{'input': '总共有多少用户？\nSQLQuery:', 'top_k': '5', 'dialect': 'sqlite', 'table_info': '\nCREATE TABLE "User" (\n\t"ID" INTEGER, \n\t"Name" TEXT NOT NULL, \n\t"Age" INTEGER, \n\t"Money" REAL, \n\t"CreateDate" DATE DEFAULT CURRENT_DATE, \n\tPRIMARY KEY ("ID")\n)\n\n/*\n3 rows from User table:\nID\tName\tAge\tMoney\tCreateDate\n1\tJava\t10\t20.5\t2024-04-14\n2\tPython\t20\t0.0\t2024-04-14\n3\tVue\t33\t55.0\t2024-04-14\n*/', 'stop': ['\nSQLResult:']}, 'SELECT COUNT(*) FROM User', {'sql_cmd': 'SELECT COUNT(*) FROM User'}, '[(4,)]', {'input': '总共有多少用户？\nSQLQuery:SELECT COUNT(*) FROM User\nSQLResult: [(4,)]\nAnswer:', 'top_k': '5', 'dialect': 'sqlite', 'table_info': '\nCREATE TABLE "User" (\n\t"ID" INTEGER, \n\t"Name" TEXT NOT NULL, \n\t"Age" INTEGER, \n\t"Money" REAL, \n\t"CreateDate" DATE DEFAULT CURRENT_DATE, \n\tPRIMARY KEY ("ID")\n)\n\n/*\n3 rows from User table:\nID\tName\tAge\tMoney\tCreateDate\n1\tJava\t10\t20.5\t2024-04-14\n2\tPython\t20\t0.0\t2024-04-14\n3\tVue\t33\t55.0\t2024-04-14\n*/', 'stop': ['\nSQLResult:']}, 'Final answer here: 4']
1

限制返回的行数

可以使用top_k参数来选择查询表中要获取的最大结果数（默认为10），这可以避免查询结果超出提示最大长度或不必要地消耗token。

db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True, use_query_checker=True, top_k=2)

1
2

create_sql_agent：查询数据库

概述

LangChain的SQL代理，它提供了比SQLDatabaseChain更灵活的与SQL数据库交互方式。相比SQLDatabaseChain，SQL代理有以下优点：

根据数据库架构和内容回答问题，例如检索特定表的描述。

具有纠错能力，能够捕获执行生成的查询时遇到的错误，并正确重新生成并执行新的查询。
1
2
3

通过create_sql_agent函数初始化代理，创建的SQL Agent。需使用到SQLDatabaseToolkit工具箱，其中包含以下工具：

创建并执行查询

检查查询语法

检索数据表描述

借助这些工具，代理可以借助LLM完成SQL查询任务
1
2
3
4
5
6
7

代理使用ReAct风格的提示，通过思考确定下一个action，并逐层前进，直到得到答案。

示例代码

from langchain_community.utilities import SQLDatabase
from langchain_openai import OpenAI
from langchain.agents import create_sql_agent
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain.agents.agent_types import AgentType

# 连接到FlowerShop数据库
db = SQLDatabase.from_uri("sqlite:///demo.db")
llm = OpenAI(temperature=0, verbose=True)

# 创建SQL Agent
agent_executor = create_sql_agent(
    llm=llm,
    toolkit=SQLDatabaseToolkit(db=db, llm=llm),
    verbose=True,
    agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
)

# 使用Agent执行SQL查询
questions = [
    "总共有多少用户？",
    "哪个用户的年龄最大？",
    "所有用户的金额合计是多少？",
]

# 循环提问
for question in questions:
    response = agent_executor.invoke(question)
    print(response)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

执行测试