在当今数字化时代,数据驱动决策已经成为企业发展的关键。随着数据库技术的不断发展,查询数据库的方式也在不断演变。传统的数据库查询方式虽然功能强大,但有时显得复杂和冗长。而Function Calling(函数调用)提供了一种更加简洁、高效的数据库查询方式。本文将介绍Function Calling查询数据库的原理、优势以及实际应用案例。
一、什么是Function Calling?
Function Calling,顾名思义,就是通过调用函数的方式来完成特定任务。在数据库查询中,Function Calling指的是将查询操作封装成函数,通过调用这些函数来完成数据库查询。与传统的SQL查询语句不同,Function Calling具有更高的可读性和复用性。
二、Function Calling查询数据库的优势
-
简洁明了:Function Calling将复杂的查询操作封装在函数内部,调用时只需简短的函数名称和参数,极大简化了查询语句。
-
高效复用:常用的查询操作可以封装成函数,方便在不同的项目或模块中复用,减少重复代码,提高开发效率。
-
易于维护:由于查询逻辑集中在函数内部,当需要修改查询逻辑时,只需更新函数定义,无需逐个修改所有使用到该查询的地方。
-
安全性高:通过预定义的函数进行数据库操作,可以有效防止SQL注入等安全问题,增强系统的安全性。
三、Function Calling查询数据库的实现
import openai
import os
from math import *
from icecream import ic
import json
from math import *
import requests
import logging
import sqlite3
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())
openai.api_key = os.getenv('OPENAI_API_KEY')
openai.api_base = os.getenv('OPENAI_API_URL')
model = os.getenv('MODEL')
amap_key = os.getenv('GAODE_MAP_API_KEY')
database_schema_string = """
CREATE TABLE customers (
id INT PRIMARY KEY NOT NULL, -- 主键,不允许为空
customer_name VARCHAR(255) NOT NULL, -- 客户名,不允许为空
email VARCHAR(255) UNIQUE, -- 邮箱,唯一
register_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP -- 注册时间,默认为当前时间
);
CREATE TABLE products (
id INT PRIMARY KEY NOT NULL, -- 主键,不允许为空
product_name VARCHAR(255) NOT NULL, -- 产品名称,不允许为空
price DECIMAL(10,2) NOT NULL -- 价格,不允许为空
);
CREATE TABLE orders (
id INT PRIMARY KEY NOT NULL, -- 主键,不允许为空
customer_id INT NOT NULL, -- 客户ID,不允许为空
product_id INT NOT NULL, -- 产品ID,不允许为空
price DECIMAL(10,2) NOT NULL, -- 价格,不允许为空
status INT NOT NULL, -- 订单状态,整数类型,不允许为空。0代表待支付,1代表已支付,2代表已退款
create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, -- 创建时间,默认为当前时间
pay_time TIMESTAMP -- 支付时间,可以为空
);
"""
def get_sql_completion(messages, model=model):
"""
使用 OpenAI 的 Chat API 来生成完成的 SQL 查询语句。
参数:
- messages: 一个包含用户消息和系统消息的列表,用于上下文对话。
- model: 使用的 OpenAI 模型的名称,默认为环境变量中定义的模型。
返回:
- 生成的 SQL 查询语句。
"""
response = openai.ChatCompletion.create(
model=model,
messages=messages,
temperature=0,
tools=[{
"type": "function",
"function": {
"name": "ask_database",
"description": "Use this function to answer user questions about business.
Output should be a fully formed SQL query.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": f"""
SQL query extracting info to answer the user's question.
SQL should be written using this database schema:
{database_schema_string}
The query should be returned in plain text, not in JSON.
The query should only contain grammars supported by SQLite.
""",
}
},
"required": ["query"],
}
}
}],
)
return response.choices[0].message
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
for statement in database_schema_string.split(';'):
cursor.execute(statement)
mock_data = [
(1, 1001, 'TSHIRT_1', 50.00, 0, '2023-10-12 10:00:00', None),
(2, 1001, 'TSHIRT_2', 75.50, 1, '2023-10-16 11:00:00', '2023-08-16 12:00:00'),
(3, 1002, 'SHOES_X2', 25.25, 2, '2023-10-17 12:30:00', '2023-08-17 13:00:00'),
(4, 1003, 'HAT_Z112', 60.75, 1, '2023-10-20 14:00:00', '2023-08-20 15:00:00'),
(5, 1002, 'WATCH_X001', 90.00, 0, '2023-10-28 16:00:00', None)
]
for record in mock_data:
cursor.execute('''
INSERT INTO orders (id, customer_id, product_id, price, status, create_time, pay_time)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', record)
conn.commit()
def ask_database(query):
"""
执行 SQL 查询并返回结果。
参数:
- query: 要执行的 SQL 查询语句。
返回:
- 查询结果。
"""
cursor.execute(query)
records = cursor.fetchall()
return records
def test_promopt():
"""
测试通过自然语言询问来生成并执行 SQL 查询的功能。
"""
prompt = "统计每月每件商品的销售额"
messages = [
{"role": "system", "content": "基于 order 表回答用户问题"},
{"role": "user", "content": prompt}
]
response = get_sql_completion(messages)
print(response)
conn.close()
def test_promopt_result():
"""
测试自然语言询问的更复杂场景,包括对数据库操作的调用和日志记录。
"""
prompt = "统计每月每件商品的销售额"
messages = [
{"role": "system", "content": "基于 order 表回答用户问题"},
{"role": "user", "content": prompt}
]
response = get_sql_completion(messages)
if response.content is None:
response.content = "null"
messages.append(response)
logging.info("====Function Calling====")
logging.info(response)
if response.tool_calls is not None:
tool_call = response.tool_calls[0]
if tool_call.function.name == "ask_database":
arguments = tool_call.function.arguments
args = json.loads(arguments)
logging.info("====SQL====")
logging.info(args["query"])
result = ask_database(args["query"])
logging.info("====DB Records====")
logging.info(result)
messages.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": "ask_database",
"content": str(result)
})
response = get_sql_completion(messages)
logging.info("====最终回复====")
logging.info(response.content)
conn.close()
if __name__ == '__main__':
test_promopt_result()
2024-05-31 23:12:38,331 - INFO - ====Function Calling====
2024-05-31 23:12:38,332 - INFO - {
"role": "assistant",
"content": "null",
"tool_calls": [
{
"id": "call_zhket5tQwLB9jGukvB2GpfMX",
"type": "function",
"function": {
"name": "ask_database",
"arguments": "{"query":"SELECT strftime('%Y-%m', create_time) AS month, product_id, SUM(price) AS total_sales FROM orders WHERE status = 1 GROUP BY month, product_id;"}"
}
}
]
}
2024-05-31 23:12:38,334 - INFO - ====SQL====
2024-05-31 23:12:38,334 - INFO - SELECT strftime('%Y-%m', create_time) AS month, product_id, SUM(price) AS total_sales FROM orders WHERE status = 1 GROUP BY month, product_id;
2024-05-31 23:12:38,335 - INFO - ====DB Records====
2024-05-31 23:12:38,335 - INFO - [('2023-10', 'HAT_Z112', 60.75), ('2023-10', 'TSHIRT_2', 75.5)]
2024-05-31 23:12:41,284 - INFO - ====最终回复====
2024-05-31 23:12:41,285 - INFO - 每月每件商品的销售额统计如下:
- 2023年10月:
- 商品ID:HAT_Z112,销售额:60.75
- 商品ID:TSHIRT_2,销售额:75.5
在以上示例中,通过调用函数实现了对数据库的查询。这样做不仅代码更加简洁易读,还方便了查询逻辑的复用和维护。
四、实际应用案例
-
电商平台的商品查询:在电商平台中,商品查询是非常频繁的操作。通过Function Calling,可以将不同条件的商品查询封装成函数,例如按价格区间、按分类、按销量排序等,大大提高了查询的灵活性和效率。
-
用户管理系统:在用户管理系统中,可以将用户的增删改查操作封装成函数,例如根据用户ID查询、按注册时间排序、批量删除用户等,使系统更易于维护和扩展。
-
数据分析系统:在数据分析系统中,常常需要对大量数据进行复杂查询和统计。通过Function Calling,可以将常用的统计分析操作封装成函数,例如按时间段统计销售额、按地区统计用户数量等,提高了数据处理的效率和准确性。
五、总结
Function Calling查询数据库作为一种新兴的技术手段,正在逐渐受到开发者的青睐。它不仅简化了数据库查询操作,提高了代码的可读性和复用性,还增强了系统的安全性和维护性。在未来,随着数据库技术的不断发展,相信Function Calling将会在更多领域发挥重要作用,助力企业实现更高效的数据处理和决策支持。
希望本文能够为大家提供一些有益的参考,让我们共同探索Function Calling的更多可能性,推动技术的进步和应用的发展。