概述
本文整理Function Calling方法。
代码一
import openai
import os
import numpy as np
import pandas as pd
import json
import io
from openai import OpenAI
import inspect
openai.api_key = os.getenv("OPENAI_API_KEY")
openai.api_base="https://www.openai.com"
client = OpenAI(api_key=openai.api_key ,base_url=openai.api_base)
def sunwukong_function(data):
"""
孙悟空算法函数，该函数定义了数据集计算过程
:param data: 必要参数，表示带入计算的数据表，用字符串进行表示
:return：sunwukong_function函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
"""
data = io.StringIO(data)
df_new = pd.read_csv(data, sep='\s+', index_col=0)
res = df_new * 10
return json.dumps(res.to_string())
df = pd.DataFrame({'x1':[1, 2], 'x2':[3, 4]})
df_str = df.to_string()
result_json=sunwukong_function(df_str)
sunwukong={
"type": "function",
"function": {"name": "sunwukong_function",
"description": "用于执行孙悟空算法函数，定义了一种特殊的数据集计算过程",
"parameters": {"type": "object",
"properties": {"data": {"type": "string",
"description": "执行孙悟空算法的数据集"},
},
"required": ["data"],
},
}
}
tools = [sunwukong]
messages=[
{"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
{"role": "user", "content": "请在数据集data上执行孙悟空算法"}
]
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages
)
response.choices[0].message
代码二
messages=[
{"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
{"role": "user", "content": "请在数据集data上执行孙悟空算法"}
]
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
tools=tools,
tool_choice="auto",
)
first_response = response.choices[0].message
available_tools = {
"sunwukong_function": sunwukong_function,
}
tool_calls = response.choices[0].message.tool_calls
for tool_call in tool_calls:
function_name = tool_call.function.name
function_to_call = available_tools[function_name]
function_args = json.loads(tool_call.function.arguments)
function_response = function_to_call(**function_args)
print(function_name)
print(function_args)
print(function_response)
function_response = function_to_call(**function_args)
messages.append(first_response)
# 追加function返回消息
messages.append(
{
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": function_response,
}
)
second_response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
)
print(second_response.choices[0].message.content)
代码三
from openai import OpenAI
import json
openai.api_key = os.getenv("OPENAI_API_KEY")
openai.api_base="https://www.openai.com"
client = OpenAI(api_key=openai.api_key ,base_url=openai.api_base)
# Example dummy function hard coded to return the same weather
# In production, this could be your backend API or an external API
def sunwukong_function(data):
"""
孙悟空算法函数，该函数定义了数据集计算过程
:param data: 必要参数，表示带入计算的数据表，用字符串进行表示
:return：sunwukong_function函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
"""
data = io.StringIO(data)
df_new = pd.read_csv(data, sep='\s+', index_col=0)
res = df_new['x1'] * 10
return json.dumps(res.to_string())
df_str=pd.DataFrame({'x1':[1, 2], 'x2':[3, 4]}).to_string
def run_conversation():
# Step 1: send the conversation and available functions to the model
messages=[
{"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
{"role": "user", "content": "请在数据集data上执行孙悟空算法"}
]
tools = [
{
"type": "function",
"function": {"name": "sunwukong_function",
"description": "用于执行孙悟空算法函数，定义了一种特殊的数据集计算过程",
"parameters": {"type": "object",
"properties": {"data": {"type": "string",
"description": "执行孙悟空算法的数据集"},
},
"required": ["data"],
},
}
}]
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
tools=tools,
tool_choice="auto", # auto is default, but we'll be explicit
)
response_message = response.choices[0].message
tool_calls = response_message.tool_calls
# Step 2: check if the model wanted to call a function
if tool_calls:
# Step 3: call the function
# Note: the JSON response may not always be valid; be sure to handle errors
available_functions = {
"sunwukong_function": sunwukong_function,
} # only one function in this example, but you can have multiple
messages.append(response_message) # extend conversation with assistant's reply
# Step 4: send the info for each function call and function response to the model
for tool_call in tool_calls:
function_name = tool_call.function.name
function_to_call = available_functions[function_name]
function_args = json.loads(tool_call.function.arguments)
function_response = function_to_call(**function_args)
messages.append(
{
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": function_response,
}
) # extend conversation with function response
second_response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
) # get a new response from the model where it can see the function response
return second_response
result=run_conversation()
# 使用StringIO将字符串转换为文件对象
df_str='\\n x1\\n0 10\\n1 20\\n.'
data = io.StringIO(df_str)
# 使用read_csv()函数读取数据，并设置第一列为索引
df_new = pd.read_csv(data, sep='\s+', index_col=0)
# # 2. Function Calling函数封装
import openai
import os
import numpy as np
import pandas as pd
import json
import io
from openai import OpenAI
openai.api_key = os.getenv("OPENAI_API_KEY")
openai.api_base="https://newone.nxykj.tech/v1"
client = OpenAI(api_key=openai.api_key ,base_url=openai.api_base)
# In[177]:
response = client.chat.completions.create(
model="gpt-4-0613",
messages=[
{"role": "user", "content": "什么是JSON Schema？"}
]
)
response.choices[0].message.content
# In[31]:
def sunwukong_function(data):
"""
孙悟空算法函数，该函数定义了数据集计算过程
:param data: 必要参数，表示带入计算的数据表，用字符串进行表示
:return：sunwukong_function函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
"""
data = io.StringIO(data)
df_new = pd.read_csv(data, sep='\s+', index_col=0)
res = df_new * 10
return json.dumps(res.to_string())
# In[32]:
# 创建一个DataFrame
df = pd.DataFrame({'x1':[1, 2], 'x2':[3, 4]})
df_str = df.to_string()
data = io.StringIO(df_str)
df_new = pd.read_csv(data, sep='\s+', index_col=0)
# In[33]:
sunwukong={
"type": "function",
"function": {"name": "sunwukong_function",
"description": "用于执行孙悟空算法函数，定义了一种特殊的数据集计算过程",
"parameters": {"type": "object",
"properties": {"data": {"type": "string",
"description": "执行孙悟空算法的数据集"},
},
"required": ["data"],
},
}
}
# In[34]:
tools = [sunwukong]
# In[35]:
available_tools = {
"sunwukong_function": sunwukong_function,
}
# In[36]:
import inspect
print(inspect.getdoc(sunwukong_function))
# In[37]:
function_description = inspect.getdoc(sunwukong_function)
# In[38]:
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "以下是孙悟空函数的函数说明：%s" % function_description},
{"role": "user", "content": "请帮我编写一个JSON Schema对象，用于说明孙悟空函数的参数输入规范。输出结果要求是JSON Schema格式的JONS类型对象，不需要任何前后修饰语句。"}
]
)
response.choices[0].message.content
# In[53]:
response.choices[0].message.content
# In[40]:
r=response.choices[0].message.content.replace("```","").replace("json","")
# In[202]:
s='{\n "$schema": "http://json-schema.org/draft-07/schema#",\n "type": "object",\n "properties": {\n "data": {\n "type": "string"\n }\n },\n "required": ["data"]\n}'
# In[41]:
json.loads(r)
# In[ ]:
# In[42]:
sunwukong
# In[43]:
sunwukong['function']['parameters']
# In[44]:
system_prompt = '以下是某的函数说明：%s' % function_description
user_prompt = '根据这个函数的函数说明，请帮我创建一个JSON格式的字典，这个字典有如下5点要求：\
1.字典总共有三个键值对；\
2.第一个键值对的Key是字符串name，value是该函数的名字：%s，也是字符串；\
3.第二个键值对的Key是字符串description，value是该函数的函数的功能说明，也是字符串；\
4.第三个键值对的Key是字符串parameters，value是一个JSON Schema对象，用于说明该函数的参数输入规范。\
5.输出结果必须是一个JSON格式的字典，且不需要任何前后修饰语句' % function_name
# In[45]:
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
]
)
response.choices[0].message.content
# In[46]:
json_function_description=json.loads(response.choices[0].message.content.replace("```","").replace("json",""))
# In[47]:
json_function_description
# In[48]:
sunwukong
# In[49]:
json_str={"type": "function","function":json_function_description}
json_str
# In[65]:
sunwukong
# ### 定义自动输出function 参数的函数
# In[50]:
def auto_functions(functions_list):
"""
Chat模型的functions参数编写函数
:param functions_list: 包含一个或者多个函数对象的列表；
:return：满足Chat模型functions参数要求的functions对象
"""
def functions_generate(functions_list):
# 创建空列表，用于保存每个函数的描述字典
functions = []
# 对每个外部函数进行循环
for function in functions_list:
# 读取函数对象的函数说明
function_description = inspect.getdoc(function)
# 读取函数的函数名字符串
function_name = function.__name__
system_prompt = '以下是某的函数说明：%s' % function_description
user_prompt = '根据这个函数的函数说明，请帮我创建一个JSON格式的字典，这个字典有如下5点要求：\
1.字典总共有三个键值对；\
2.第一个键值对的Key是字符串name，value是该函数的名字：%s，也是字符串；\
3.第二个键值对的Key是字符串description，value是该函数的函数的功能说明，也是字符串；\
4.第三个键值对的Key是字符串parameters，value是一个JSON Schema对象，用于说明该函数的参数输入规范。\
5.输出结果必须是一个JSON格式的字典，只输出这个字典即可，前后不需要任何前后修饰或说明的语句' % function_name
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
]
)
json_function_description=json.loads(response.choices[0].message.content.replace("```","").replace("json",""))
json_str={"type": "function","function":json_function_description}
functions.append(json_str)
return functions
## 最大可以尝试4次
max_attempts = 4
attempts = 0
while attempts < max_attempts:
try:
functions = functions_generate(functions_list)
break # 如果代码成功执行，跳出循环
except Exception as e:
attempts += 1 # 增加尝试次数
print("发生错误：", e)
if attempts == max_attempts:
print("已达到最大尝试次数，程序终止。")
raise # 重新引发最后一个异常
else:
print("正在重新运行...")
return functions
# In[53]:
functions_list = [sunwukong_function]
tools = auto_functions(functions_list)
# In[54]:
tools
# In[55]:
df_str = pd.DataFrame({'x1':[1, 2], 'x2':[3, 4]}).to_string()
df_str
# In[56]:
messages=[
{"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
{"role": "user", "content": "请在数据集data上执行孙悟空算法"}
]
# In[57]:
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
tools=tools,
tool_choice="auto",
)
response.choices[0].message
# In[59]:
def tangseng_function(data):
"""
唐僧算法函数，该函数定义了数据集计算过程
:param data: 必要参数，表示带入计算的数据表，用字符串进行表示
:return：tangseng_function函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
"""
data = io.StringIO(data)
df_new = pd.read_csv(data, sep='\s+', index_col=0)
res = df_new * 1000000
return json.dumps(res.to_string())
# In[80]:
functions_list=[sunwukong_function,tangseng_function]
# In[98]:
tools = auto_functions(functions_list)
tools
# In[62]:
messages=[
{"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
{"role": "user", "content": "请在数据集data上执行唐僧算法函数"}
]
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
tools=tools,
tool_choice="auto",
)
response.choices[0].message
# # 3. 封装调用2轮response的函数
# In[63]:
functions_list
# In[103]:
def run_conversation(messages, functions_list=None, model="gpt-3.5-turbo"):
"""
能够自动执行外部函数调用的对话模型
:param messages: 必要参数，字典类型，输入到Chat模型的messages参数对象
:param functions_list: 可选参数，默认为None，可以设置为包含全部外部函数的列表对象
:param model: Chat模型，可选参数，默认模型为gpt-3.5-turbo
:return：Chat模型输出结果
"""
# 如果没有外部函数库，则执行普通的对话任务
if functions_list == None:
response = client.chat.completions.create(
model=model,
messages=messages,
)
response_message = response.choices[0].message
final_response = response_message.content
# 若存在外部函数库，则需要灵活选取外部函数并进行回答
else:
# 创建functions对象
tools = auto_functions(functions_list)
# 创建外部函数库字典
available_functions = {func.__name__: func for func in functions_list}
# 第一次调用大模型
response = client.chat.completions.create(
model=model,
messages=messages,
tools=tools,
tool_choice="auto", )
response_message = response.choices[0].message
tool_calls = response_message.tool_calls
if tool_calls:
messages.append(response_message)
for tool_call in tool_calls:
function_name = tool_call.function.name
function_to_call = available_functions[function_name]
function_args = json.loads(tool_call.function.arguments)
## 真正执行外部函数的就是这儿的代码
function_response = function_to_call(**function_args)
messages.append(
{
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": function_response,
}
)
## 第二次调用模型
second_response = client.chat.completions.create(
model=model,
messages=messages,
)
# 获取最终结果
final_response = second_response.choices[0].message.content
else:
final_response = response_message.content
return final_response
# In[118]:
df_str = pd.DataFrame({'x1':[1, 2], 'x2':[3, 4]}).to_string()
df_str
# In[127]:
## 测试一下函数（1）
messages = [
{"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
{"role": "user", "content": '请在data上执行唐僧算法函数'}]
# In[131]:
run_conversation(messages = messages, functions_list = functions_list)
# In[80]:
## 测试一下函数（2）
messages = [
{"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
{"role": "user", "content": '请在data上执行孙悟空算法函数'}]
# In[81]:
run_conversation(messages = messages, functions_list = functions_list)
# '已在数据集上成功执行了孙悟空算法函数，修改后的数据集为：\n\n```\n x1 x2\n0 10 30\n1 20 40\n```'
# In[82]:
## 测试一下函数（3）
messages = [
{"role": "system", "content": "数据集data：%s，数据集以字符串形式呈现" % df_str},
{"role": "user", "content": '请解释一下data数据集'}]
# In[83]:
run_conversation(messages = messages, functions_list = functions_list)
# # 增加多轮对话的效果（附加代码练习）
# In[1]:
import openai
import os
import numpy as np
import pandas as pd
import json
import io
from openai import OpenAI
import inspect
openai.api_key = os.getenv("OPENAI_API_KEY")
openai.api_base="https://newone.nxykj.tech/v1"
client = OpenAI(api_key=openai.api_key ,base_url=openai.api_base)
# In[9]:
def sunwukong_function(data):
"""
孙悟空算法函数，该函数定义了数据集计算过程
:param data: 必要参数，表示带入计算的数据表，用字符串进行表示
:return：sunwukong_function函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
"""
data = io.StringIO(data)
df_new = pd.read_csv(data, sep='\s+', index_col=0)
res = df_new * 10
return json.dumps(res.to_string())
# In[3]:
def tangseng_function(data):
"""
唐僧算法函数，该函数定义了数据集计算过程
:param data: 必要参数，表示带入计算的数据表，用字符串进行表示
:return：tangseng_function函数计算后的结果，返回结果为表示为JSON格式的Dataframe类型对象
"""
data = io.StringIO(data)
df_new = pd.read_csv(data, sep='\s+', index_col=0)
res = df_new * 1000000
return json.dumps(res.to_string())
# In[5]:
def auto_functions(functions_list):
"""
Chat模型的functions参数编写函数
:param functions_list: 包含一个或者多个函数对象的列表；
:return：满足Chat模型functions参数要求的functions对象
"""
def functions_generate(functions_list):
# 创建空列表，用于保存每个函数的描述字典
functions = []
# 对每个外部函数进行循环
for function in functions_list:
# 读取函数对象的函数说明
function_description = inspect.getdoc(function)
# 读取函数的函数名字符串
function_name = function.__name__
system_prompt = '以下是某的函数说明：%s' % function_description
user_prompt = '根据这个函数的函数说明，请帮我创建一个JSON格式的字典，这个字典有如下5点要求：\
1.字典总共有三个键值对；\
2.第一个键值对的Key是字符串name，value是该函数的名字：%s，也是字符串；\
3.第二个键值对的Key是字符串description，value是该函数的函数的功能说明，也是字符串；\
4.第三个键值对的Key是字符串parameters，value是一个JSON Schema对象，用于说明该函数的参数输入规范。\
5.输出结果必须是一个JSON格式的字典，只输出这个字典即可，前后不需要任何前后修饰或说明的语句' % function_name
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
]
)
json_function_description=json.loads(response.choices[0].message.content.replace("```","").replace("json",""))
json_str={"type": "function","function":json_function_description}
functions.append(json_str)
return functions
## 最大可以尝试4次
max_attempts = 4
attempts = 0
while attempts < max_attempts:
try:
functions = functions_generate(functions_list)
break # 如果代码成功执行，跳出循环
except Exception as e:
attempts += 1 # 增加尝试次数
print("发生错误：", e)
if attempts == max_attempts:
print("已达到最大尝试次数，程序终止。")
raise # 重新引发最后一个异常
else:
print("正在重新运行...")
return functions
# In[6]:
def run_conversation(messages, functions_list=None, model="gpt-3.5-turbo"):
"""
能够自动执行外部函数调用的对话模型
:param messages: 必要参数，字典类型，输入到Chat模型的messages参数对象
:param functions_list: 可选参数，默认为None，可以设置为包含全部外部函数的列表对象
:param model: Chat模型，可选参数，默认模型为gpt-3.5-turbo
:return：Chat模型输出结果
"""
# 如果没有外部函数库，则执行普通的对话任务
if functions_list == None:
response = client.chat.completions.create(
model=model,
messages=messages,
)
response_message = response.choices[0].message
final_response = response_message.content
# 若存在外部函数库，则需要灵活选取外部函数并进行回答
else:
# 创建functions对象
tools = auto_functions(functions_list)
# 创建外部函数库字典
available_functions = {func.__name__: func for func in functions_list}
# 第一次调用大模型
response = client.chat.completions.create(
model=model,
messages=messages,
tools=tools,
tool_choice="auto", )
response_message = response.choices[0].message
tool_calls = response_message.tool_calls
if tool_calls:
messages.append(response_message)
for tool_call in tool_calls:
function_name = tool_call.function.name
function_to_call = available_functions[function_name]
function_args = json.loads(tool_call.function.arguments)
## 真正执行外部函数的就是这儿的代码
function_response = function_to_call(**function_args)
messages.append(
{
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": function_response,
}
)
## 第二次调用模型
second_response = client.chat.completions.create(
model=model,
messages=messages,
)
# 获取最终结果
final_response = second_response.choices[0].message.content
else:
final_response = response_message.content
return final_response
# In[7]:
def chat_with_model(functions_list=None,
prompt="你好",
model="gpt-3.5-turbo",
system_message=[{"role": "system", "content": "你是小智助手。"}]):
messages = system_message
messages.append({"role": "user", "content": prompt})
while True:
answer = run_conversation(messages=messages,
functions_list=functions_list,
model=model)
print(f"智能助手回答: {answer}")
# 询问用户是否还有其他问题
user_input = input("您还有其他问题吗？(输入退出以结束对话): ")
if user_input == "退出":
break
# 记录用户回答
messages.append({"role": "user", "content": user_input})
# In[10]:
functions_list=[sunwukong_function,tangseng_function]
# In[12]:
chat_with_model(functions_list,prompt="你好")
# In[1]:
user_input = input("您还有其他问题吗？(输入退出以结束对话): ")
