赞
踩
tools
是聊天完成API中的可选参数,可用于提供函数规范。其目的是使模型能够生成符合提供规范的函数参数。请注意,API不会实际执行任何函数调用。开发人员需要使用模型输出来执行函数调用。
在tools
参数中,如果提供了functions
参数,则模型将决定何时适合使用其中一个函数。可以通过将tool_choice
参数设置为{"name": "<insert-function-name>"}
来强制API使用特定的函数。还可以通过将tool_choice
参数设置为"none"
来强制API不使用任何函数。如果使用了函数,则输出中的响应将包含"finish_reason": "function_call"
,以及一个具有函数名称和生成的函数参数的tool_choice
对象。
本笔记本包含以下2个部分:
!pip install scipy
!pip install tenacity
!pip install tiktoken
!pip install termcolor
!pip install openai
!pip install requests
import json
import openai
import requests
from tenacity import retry, wait_random_exponential, stop_after_attempt
from termcolor import colored
GPT_MODEL = "gpt-3.5-turbo-0613"
首先,让我们定义一些实用工具,用于调用聊天完成 API,并维护和跟踪对话状态。
# 定义一个装饰器函数,用于在请求失败时进行重试 @retry(wait=wait_random_exponential(multiplier=1, max=40), stop=stop_after_attempt(3)) def chat_completion_request(messages, tools=None, tool_choice=None, model=GPT_MODEL): # 定义请求头 headers = { "Content-Type": "application/json", "Authorization": "Bearer " + openai.api_key, } # 定义请求体 json_data = {"model": model, "messages": messages} # 如果有工具,则将工具添加到请求体中 if tools is not None: json_data.update({"tools": tools}) # 如果有工具选择,则将工具选择添加到请求体中 if tool_choice is not None: json_data.update({"tool_choice": tool_choice}) try: # 发送POST请求 response = requests.post( "https://api.openai.com/v1/chat/completions", headers=headers, json=json_data, ) # 返回响应 return response except Exception as e: # 请求失败时打印错误信息,并返回错误对象 print("Unable to generate ChatCompletion response") print(f"Exception: {e}") return e
# 定义一个函数,用于打印对话信息 def pretty_print_conversation(messages): # 定义一个字典,用于存储角色和对应的颜色 role_to_color = { "system": "red", "user": "green", "assistant": "blue", "tool": "magenta", } # 遍历对话信息列表 for message in messages: # 如果是系统角色,则打印系统信息,并使用红色字体 if message["role"] == "system": print(colored(f"system: {message['content']}\n", role_to_color[message["role"]])) # 如果是用户角色,则打印用户信息,并使用绿色字体 elif message["role"] == "user": print(colored(f"user: {message['content']}\n", role_to_color[message["role"]])) # 如果是助手角色,并且有函数调用信息,则打印函数调用信息,并使用蓝色字体 elif message["role"] == "assistant" and message.get("function_call"): print(colored(f"assistant: {message['function_call']}\n", role_to_color[message["role"]])) # 如果是助手角色,并且没有函数调用信息,则打印助手信息,并使用蓝色字体 elif message["role"] == "assistant" and not message.get("function_call"): print(colored(f"assistant: {message['content']}\n", role_to_color[message["role"]])) # 如果是工具角色,则打印工具信息,并使用洋红色字体 elif message["role"] == "tool": print(colored(f"function ({message['name']}): {message['content']}\n", role_to_color[message["role"]]))
让我们创建一些函数规范,以便与一个假想的天气API进行接口。我们将把这些函数规范传递给聊天完成API,以生成符合规范的函数参数。
# 定义了一个名为tools的列表,其中包含两个字典元素 # 第一个字典元素表示一个函数工具,用于获取当前天气信息 # 第二个字典元素表示一个函数工具,用于获取N天的天气预报信息 tools = [ # 第一个函数工具:获取当前天气信息 # 函数名为get_current_weather,描述为"Get the current weather" # 参数为一个对象,包含两个属性:location和format # location属性表示城市和州,类型为字符串,描述为"The city and state, e.g. San Francisco, CA" # format属性表示温度单位,类型为字符串,取值为"celsius"或"fahrenheit",描述为"The temperature unit to use. Infer this from the users location." # 参数中location和format属性为必填项 { "type": "function", "function": { "name": "get_current_weather", "description": "Get the current weather", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "The city and state, e.g. San Francisco, CA", }, "format": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "The temperature unit to use. Infer this from the users location.", }, }, "required": ["location", "format"], }, } }, # 第二个函数工具:获取N天的天气预报信息 # 函数名为get_n_day_weather_forecast,描述为"Get an N-day weather forecast" # 参数为一个对象,包含三个属性:location、format和num_days # location属性表示城市和州,类型为字符串,描述为"The city and state, e.g. San Francisco, CA" # format属性表示温度单位,类型为字符串,取值为"celsius"或"fahrenheit",描述为"The temperature unit to use. Infer this from the users location." # num_days属性表示预报天数,类型为整数,描述为"The number of days to forecast" # 参数中location、format和num_days属性为必填项 { "type": "function", "function": { "name": "get_n_day_weather_forecast", "description": "Get an N-day weather forecast", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "The city and state, e.g. San Francisco, CA", }, "format": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "The temperature unit to use. Infer this from the users location.", }, "num_days": { "type": "integer", "description": "The number of days to forecast", } }, "required": ["location", "format", "num_days"] }, } }, ]
如果我们询问模型当前的天气情况,它会回答一些澄清问题。
# 定义一个空列表用于存储聊天信息 messages = [] # 向列表中添加一条系统信息,包括角色和内容 messages.append({"role": "system", "content": "不要假设函数中需要的值,如果用户请求不明确,应该要求澄清。"}) # 向列表中添加一条用户信息,包括角色和内容 messages.append({"role": "user", "content": "今天天气怎么样?"}) # 发送聊天请求并获取回复信息 chat_response = chat_completion_request(messages, tools=tools) # 从回复信息中获取助手的回复信息 assistant_message = chat_response.json()["choices"][0]["message"] # 将助手的回复信息添加到聊天信息列表中 messages.append(assistant_message) # 输出助手的回复信息 assistant_message
{'role': 'assistant',
'content': 'Sure, I can help you with that. Could you please tell me the city and state you are in or the location you want to know the weather for?'}
一旦我们提供了缺失的信息,它将为我们生成适当的函数参数。
# 创建一个空列表,用于存储对话中的消息 messages = [] # 向消息列表中添加一条用户发送的消息,包括角色和内容 messages.append({"role": "user", "content": "I'm in Glasgow, Scotland."}) # 调用chat_completion_request函数,传入消息列表和工具参数,返回聊天机器人的响应 chat_response = chat_completion_request( messages, tools=tools ) # 从聊天机器人的响应中获取第一条回复消息 assistant_message = chat_response.json()["choices"][0]["message"] # 将聊天机器人的回复消息添加到消息列表中 messages.append(assistant_message) # 输出聊天机器人的回复消息 assistant_message
{'role': 'assistant',
'content': None,
'tool_calls': [{'id': 'call_o7uyztQLeVIoRdjcDkDJY3ni',
'type': 'function',
'function': {'name': 'get_current_weather',
'arguments': '{\n "location": "Glasgow, Scotland",\n "format": "celsius"\n}'}}]}
通过不同的提示方式,我们可以让它针对我们告诉它的另一个函数。
# 创建一个空列表messages用于存储对话消息 messages = [] # 向messages列表中添加一条系统消息,包含角色和内容 messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."}) # 向messages列表中添加一条用户消息,包含角色和内容 messages.append({"role": "user", "content": "what is the weather going to be like in Glasgow, Scotland over the next x days"}) # 调用chat_completion_request函数,传入messages列表和tools参数,并将返回结果赋值给chat_response变量 chat_response = chat_completion_request( messages, tools=tools ) # 从chat_response的json数据中获取助手的回复消息,并将其赋值给assistant_message变量 assistant_message = chat_response.json()["choices"][0]["message"] # 将助手的回复消息添加到messages列表中 messages.append(assistant_message) # 输出助手的回复消息 assistant_message
{'role': 'assistant',
'content': 'Sure, I can help you with that. Please let me know the value for x.'}
再次,模型要求我们澄清,因为它还没有足够的信息。在这种情况下,它已经知道了预测的位置,但它需要知道预测需要多少天。
# 向messages列表中添加一条消息,该消息由一个字典组成,包含两个键值对
# 键"role"表示消息的角色,值为"user"表示该消息是用户发送的
# 键"content"表示消息的内容,值为"5 days"表示用户发送的消息内容是"5 days"
messages.append({"role": "user", "content": "5 days"})
# 调用chat_completion_request函数,向聊天模型发送对话消息,并将返回的响应存储在chat_response变量中
# chat_completion_request函数的参数包括messages列表和tools变量
chat_response = chat_completion_request(
messages, tools=tools
)
# 从chat_response的JSON响应中获取"choices"键对应的值,并取第一个元素
chat_response.json()["choices"][0]
{'index': 0,
'message': {'role': 'assistant',
'content': None,
'tool_calls': [{'id': 'call_drz2YpGPWEMVySzYgsWYY249',
'type': 'function',
'function': {'name': 'get_n_day_weather_forecast',
'arguments': '{\n "location": "Glasgow, Scotland",\n "format": "celsius",\n "num_days": 5\n}'}}]},
'finish_reason': 'tool_calls'}
我们可以通过使用function_call参数来强制模型使用特定的函数,例如get_n_day_weather_forecast。通过这样做,我们强制模型对如何使用它进行假设。
# 定义一个空列表,用于存储对话信息
messages = []
# 向列表中添加一个字典,表示系统的回复
messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})
# 向列表中添加一个字典,表示用户的请求
messages.append({"role": "user", "content": "Give me a weather report for Toronto, Canada."})
# 调用 chat_completion_request 函数,传入对话信息和工具参数,指定使用 get_n_day_weather_forecast 函数
chat_response = chat_completion_request(
messages, tools=tools, tool_choice={"type": "function", "function": {"name": "get_n_day_weather_forecast"}}
)
# 从响应中获取第一个选择的消息,并返回该消息的内容
chat_response.json()["choices"][0]["message"]
{'role': 'assistant',
'content': None,
'tool_calls': [{'id': 'call_jdmoJQ4lqsu4mBWcVBYtt5cU',
'type': 'function',
'function': {'name': 'get_n_day_weather_forecast',
'arguments': '{\n "location": "Toronto, Canada",\n "format": "celsius",\n "num_days": 1\n}'}}]}
# 创建一个空列表,用于存储消息 messages = [] # 向消息列表中添加一个字典,表示系统角色的消息,提醒用户在请求不明确时不要做出假设,应该询问澄清 messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."}) # 向消息列表中添加一个字典,表示用户角色的消息,请求获取加拿大多伦多的天气报告 messages.append({"role": "user", "content": "Give me a weather report for Toronto, Canada."}) # 调用chat_completion_request函数,传入消息列表和工具参数,获取聊天响应 chat_response = chat_completion_request( messages, tools=tools ) # 从聊天响应的json数据中获取第一个选择的消息的内容 chat_response.json()["choices"][0]["message"]
{'role': 'assistant',
'content': None,
'tool_calls': [{'id': 'call_RYXaDjxpUCfWmpXU7BZEYVqS',
'type': 'function',
'function': {'name': 'get_current_weather',
'arguments': '{\n "location": "Toronto, Canada",\n "format": "celsius"\n}'}}]}
我们还可以强制模型不使用任何函数。这样做可以防止它生成一个正确的函数调用。
# 定义一个空列表messages,用于存储聊天消息 messages = [] # 向messages列表中添加一条系统消息,包含角色和内容 messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."}) # 向messages列表中添加一条用户消息,包含角色和内容 messages.append({"role": "user", "content": "Give me the current weather (use Celcius) for Toronto, Canada."}) # 调用chat_completion_request函数,传入messages列表、tools和tool_choice参数,并将返回结果赋值给chat_response变量 chat_response = chat_completion_request( messages, tools=tools, tool_choice="none" ) # 从chat_response的JSON响应中获取第一条回复消息的内容,并返回 chat_response.json()["choices"][0]["message"]
{'role': 'assistant',
'content': '{ "location": "Toronto, Canada", "format": "celsius" }'}
像gpt-4-1106-preview或gpt-3.5-turbo-1106这样的新型模型可以在一次操作中调用多个函数。
# 创建一个空列表 messages messages = [] # 向 messages 列表中添加一个字典,字典包含两个键值对,分别是 "role" 和 "content",表示消息的角色和内容 messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."}) # 向 messages 列表中再添加一个字典,字典包含两个键值对,分别是 "role" 和 "content",表示消息的角色和内容 messages.append({"role": "user", "content": "what is the weather going to be like in San Francisco and Glasgow over the next 4 days"}) # 调用 chat_completion_request 函数,传入 messages 列表、tools 和 model 参数,返回一个响应对象 chat_response = chat_completion_request( messages, tools=tools, model='gpt-3.5-turbo-1106' ) # 从响应对象中获取 JSON 格式的数据,并返回其中 "choices" 列表中第一个元素的 "message" 字典中的 "tool_calls" 值 assistant_message = chat_response.json()["choices"][0]["message"]['tool_calls']
[{'id': 'call_fLsKR5vGllhbWxvpqsDT3jBj',
'type': 'function',
'function': {'name': 'get_n_day_weather_forecast',
'arguments': '{"location": "San Francisco, CA", "format": "celsius", "num_days": 4}'}},
{'id': 'call_CchlsGE8OE03QmeyFbg7pkDz',
'type': 'function',
'function': {'name': 'get_n_day_weather_forecast',
'arguments': '{"location": "Glasgow", "format": "celsius", "num_days": 4}'}}]
在下一个示例中,我们将演示如何执行输入为模型生成的函数,并使用它来实现一个代理,可以为我们回答有关数据库的问题。为简单起见,我们将使用Chinook示例数据库。
*注意:*在生产环境中,SQL生成可能存在高风险,因为模型不能完全可靠地生成正确的SQL。
首先,让我们定义一些有用的实用程序函数,以从SQLite数据库中提取数据。
# 连接到 SQLite 数据库文件 "data/Chinook.db"
conn = sqlite3.connect("data/Chinook.db")
# 打印连接成功的提示信息
print("Opened database successfully")
Opened database successfully
# 获取表名函数 def get_table_names(conn): """返回一个表名列表。""" table_names = [] # 存储表名的列表 tables = conn.execute("SELECT name FROM sqlite_master WHERE type='table';") # 查询数据库中的表 for table in tables.fetchall(): # 遍历查询结果 table_names.append(table[0]) # 将表名添加到列表中 return table_names # 获取列名函数 def get_column_names(conn, table_name): """返回一个列名列表。""" column_names = [] # 存储列名的列表 columns = conn.execute(f"PRAGMA table_info('{table_name}');").fetchall() # 查询表的列信息 for col in columns: # 遍历查询结果 column_names.append(col[1]) # 将列名添加到列表中 return column_names # 获取数据库信息函数 def get_database_info(conn): """返回一个包含每个表的表名和列的字典列表。""" table_dicts = [] # 存储表名和列的字典的列表 for table_name in get_table_names(conn): # 遍历数据库中的每个表名 columns_names = get_column_names(conn, table_name) # 获取表的列名列表 table_dicts.append({"table_name": table_name, "column_names": columns_names}) # 将表名和列名添加到字典中,并添加到列表中 return table_dicts # 返回表名和列的字典列表
现在可以使用这些实用函数来提取数据库模式的表示。
# 获取数据库信息的函数
database_schema_dict = get_database_info(conn)
# 将数据库信息转换为字符串形式
database_schema_string = "\n".join(
[
f"Table: {table['table_name']}\nColumns: {', '.join(table['column_names'])}"
for table in database_schema_dict
]
)
与之前一样,我们将为我们希望API生成参数的函数定义一个函数规范。请注意,我们将数据库模式插入到函数规范中。这对于模型来说非常重要。
# 定义了一个列表变量 tools,其中包含一个字典元素 tools = [ { "type": "function", # 元素中包含一个键值对,键为 type,值为字符串 "function" "function": { # 元素中包含一个键值对,键为 function,值为字典 "name": "ask_database", # 字典中包含一个键值对,键为 name,值为字符串 "ask_database" "description": "Use this function to answer user questions about music. Input should be a fully formed SQL query.", # 字典中包含一个键值对,键为 description,值为字符串 "parameters": { # 字典中包含一个键值对,键为 parameters,值为字典 "type": "object", # 字典中包含一个键值对,键为 type,值为字符串 "object" "properties": { # 字典中包含一个键值对,键为 properties,值为字典 "query": { # 字典中包含一个键值对,键为 query,值为字典 "type": "string", # 字典中包含一个键值对,键为 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. """, # 字典中包含一个键值对,键为 description,值为字符串,其中包含一个 f-string,用于格式化字符串 } }, "required": ["query"], # 字典中包含一个键值对,键为 required,值为列表,其中包含一个字符串 "query" }, } } ]
现在让我们实现实际针对数据库执行查询的函数。
# 代码注释 # 定义一个函数,用于使用提供的SQL查询语句查询SQLite数据库 def ask_database(conn, query): try: # 执行SQL查询并获取结果 results = str(conn.execute(query).fetchall()) except Exception as e: # 如果查询失败,返回错误信息 results = f"查询失败,错误信息:{e}" # 返回查询结果 return results # 定义一个函数,用于执行函数调用 def execute_function_call(message): # 判断函数调用是否为ask_database函数 if message["tool_calls"][0]["function"]["name"] == "ask_database": # 获取查询语句 query = json.loads(message["tool_calls"][0]["function"]["arguments"])["query"] # 调用ask_database函数执行查询 results = ask_database(conn, query) else: # 如果函数调用不是ask_database函数,返回错误信息 results = f"错误:函数 {message['tool_calls'][0]['function']['name']} 不存在" # 返回查询结果 return results
# 创建一个空列表来存储对话消息 messages = [] # 添加系统角色的消息,内容是回答用户问题的目的,通过对Chinook音乐数据库生成SQL查询来实现 messages.append({"role": "system", "content": "Answer user questions by generating SQL queries against the Chinook Music Database."}) # 添加用户角色的消息,内容是询问前5位以曲目数量计算的艺术家是谁 messages.append({"role": "user", "content": "Hi, who are the top 5 artists by number of tracks?"}) # 发送聊天完成请求,并将返回的响应存储在chat_response变量中 chat_response = chat_completion_request(messages, tools) # 从响应中获取助手的消息 assistant_message = chat_response.json()["choices"][0]["message"] # 将助手消息的内容转换为字符串,并将其赋值给assistant_message的content键 assistant_message['content'] = str(assistant_message["tool_calls"][0]["function"]) # 将助手的消息添加到消息列表中 messages.append(assistant_message) # 检查助手消息中是否存在tool_calls键 if assistant_message.get("tool_calls"): # 执行助手消息中的函数调用,并将结果存储在results变量中 results = execute_function_call(assistant_message) # 将工具角色的消息添加到消息列表中,包括工具调用的id、名称和内容 messages.append({"role": "tool", "tool_call_id": assistant_message["tool_calls"][0]['id'], "name": assistant_message["tool_calls"][0]["function"]["name"], "content": results}) # 打印格式化后的对话消息 pretty_print_conversation(messages)
[31msystem: Answer user questions by generating SQL queries against the Chinook Music Database.
[0m
[32muser: Hi, who are the top 5 artists by number of tracks?
[0m
[34massistant: {'name': 'ask_database', 'arguments': '{\n "query": "SELECT Artist.Name, COUNT(Track.TrackId) AS TrackCount FROM Artist JOIN Album ON Artist.ArtistId = Album.ArtistId JOIN Track ON Album.AlbumId = Track.AlbumId GROUP BY Artist.Name ORDER BY TrackCount DESC LIMIT 5"\n}'}
[0m
[35mfunction (ask_database): [('Iron Maiden', 213), ('U2', 135), ('Led Zeppelin', 114), ('Metallica', 112), ('Lost', 92)]
[0m
# 定义一个空列表messages messages = [] # 向messages列表中添加一个字典,字典中有两个键值对,分别为"role"和"content",对应的值为"user"和"What is the name of the album with the most tracks?" messages.append({"role": "user", "content": "What is the name of the album with the most tracks?"}) # 调用chat_completion_request函数,传入messages列表和tools参数,将返回值赋值给chat_response chat_response = chat_completion_request(messages, tools) # 从chat_response的json格式中获取"choices"列表中第一个元素的"message"键对应的值,赋值给assistant_message assistant_message = chat_response.json()["choices"][0]["message"] # 将assistant_message字典中的"content"键对应的值转换为字符串类型,并赋值给"content"键 assistant_message['content'] = str(assistant_message["tool_calls"][0]["function"]) # 将assistant_message字典添加到messages列表中 messages.append(assistant_message) # 如果assistant_message字典中有"tool_calls"键,则执行execute_function_call函数,并将返回值添加到messages列表中 if assistant_message.get("tool_calls"): results = execute_function_call(assistant_message) messages.append({"role": "tool", "tool_call_id": assistant_message["tool_calls"][0]['id'], "name": assistant_message["tool_calls"][0]["function"]["name"], "content": results}) # 打印出messages列表中的内容,以便查看对话过程 pretty_print_conversation(messages)
[31msystem: Answer user questions by generating SQL queries against the Chinook Music Database.
[0m
[32muser: Hi, who are the top 5 artists by number of tracks?
[0m
[34massistant: {'name': 'ask_database', 'arguments': '{\n "query": "SELECT Artist.Name, COUNT(Track.TrackId) AS TrackCount FROM Artist JOIN Album ON Artist.ArtistId = Album.ArtistId JOIN Track ON Album.AlbumId = Track.AlbumId GROUP BY Artist.Name ORDER BY TrackCount DESC LIMIT 5"\n}'}
[0m
[35mfunction (ask_database): [('Iron Maiden', 213), ('U2', 135), ('Led Zeppelin', 114), ('Metallica', 112), ('Lost', 92)]
[0m
[32muser: What is the name of the album with the most tracks?
[0m
[34massistant: {'name': 'ask_database', 'arguments': '{\n "query": "SELECT Album.Title, COUNT(Track.TrackId) AS TrackCount FROM Album JOIN Track ON Album.AlbumId = Track.AlbumId GROUP BY Album.Title ORDER BY TrackCount DESC LIMIT 1"\n}'}
[0m
[35mfunction (ask_database): [('Greatest Hits', 57)]
[0m
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。