在当今数字化时代，数据驱动决策已经成为企业发展的关键。随着数据库技术的不断发展，查询数据库的方式也在不断演变。传统的数据库查询方式虽然功能强大，但有时显得复杂和冗长。而Function Calling（函数调用）提供了一种更加简洁、高效的数据库查询方式。本文将介绍Function Calling查询数据库的原理、优势以及实际应用案例。
一、什么是Function Calling？
Function Calling，顾名思义，就是通过调用函数的方式来完成特定任务。在数据库查询中，Function Calling指的是将查询操作封装成函数，通过调用这些函数来完成数据库查询。与传统的SQL查询语句不同，Function Calling具有更高的可读性和复用性。
二、Function Calling查询数据库的优势
-
简洁明了：Function Calling将复杂的查询操作封装在函数内部，调用时只需简短的函数名称和参数，极大简化了查询语句。
-
高效复用：常用的查询操作可以封装成函数，方便在不同的项目或模块中复用，减少重复代码，提高开发效率。
-
易于维护：由于查询逻辑集中在函数内部，当需要修改查询逻辑时，只需更新函数定义，无需逐个修改所有使用到该查询的地方。
-
安全性高：通过预定义的函数进行数据库操作，可以有效防止SQL注入等安全问题，增强系统的安全性。
三、Function Calling查询数据库的实现
import openaiimport osfrom math import *from icecream import icimport jsonfrom math import *import requestsimport loggingimport 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.argumentsargs = 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的更多可能性，推动技术的进步和应用的发展。
