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

通过 ChatGPT 的 Function Call 查询数据库

news 2025/8/2 21:47:10

用 Function Calling 的方式实现手机流量包智能客服的例子。

def get_sql_completion(messages, model="gpt-3.5-turbo"):response = client.chat.completions.create(model=model,messages=messages,temperature=0,tools=[{  # 摘自 OpenAI 官方示例 https://github.com/openai/openai-cookbook/blob/main/examples/How_to_call_functions_with_chat_models.ipynb"type": "function","function": {"name": "ask_database","description": "Use this function to answer user questions about packages. \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

#  描述数据库表结构
database_schema_string = """
CREATE TABLE packages (id INT PRIMARY KEY NOT NULL, -- 主键，不允许为空package_name STR NOT NULL, -- 套餐名称，不允许为空monthly_fee INT NOT NULL, -- 月费，单位元，不允许为空flow_size INT NOT NULL, -- 流量大小，单位G，不允许为空condition STR -- 购买的限制条件，允许为空
);
"""

import sqlite3# 创建数据库连接
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()# 创建orders表
cursor.execute(database_schema_string)# 插入4条明确的模拟记录
mock_data = [(1, '经济套餐', 50, 10, None),(2, '畅游套餐', 180, 100, None),(3, '无限套餐', 300, 1000, None),(4, '校园套餐', 150, 200, '仅限在校生'),
]for record in mock_data:cursor.execute('''INSERT INTO packages (id, package_name, monthly_fee, flow_size, condition)VALUES (?, ?, ?, ?, ?)''', record)# 提交事务
conn.commit()

def ask_database(query):cursor.execute(query)records = cursor.fetchall()return recordsprompt = "请问流量最大的套餐是哪个？"
# prompt = "统计每月每件商品的销售额"
# prompt = "哪个用户消费最高？消费多少？"messages = [{"role": "system", "content": "基于 packages 表回答用户问题"},{"role": "user", "content": prompt}
]
print("====messages====")
print_json(messages)response = get_sql_completion(messages)
# print("====first Function Calling====")
# print_json(response)if response.content is None:response.content = ""
messages.append(response)
print("====Function Calling====")
print_json(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)print("====SQL====")print(args["query"])result = ask_database(args["query"])print("====DB Records====")print(result)messages.append({"tool_call_id": tool_call.id,"role": "tool","name": "ask_database","content": str(result)})response = get_sql_completion(messages)print("====最终回复====")print(response.content)