Écrire des fonctions Python déployables

Exigences de la fonction de score

• Il existe deux façons d'ajouter l'objet score fonction :
  • Explicitement, par l'utilisateur
  • Implicitement, par la méthode utilisée pour sauvegarder la fonction Python en tant qu'actif dans le référentiel Watson Machine Learning
• La score fonction peut accepter un seul paramètre d'entrée JSON ou deux paramètres : la charge utile et le jeton porteur.
• La fonction score doit renvoyer un objet sérialisable JSON (par exemple, des dictionnaires ou des listes).

Évaluation des entrées d'évaluation

• Le contenu d'entrée de score doit inclure un tableau portant le nom values, comme indiqué dans cet exemple de schéma. Le input_data paramètre est obligatoire dans la charge utile. Le input_data paramètre peut également inclure des paires nom-valeur supplémentaires.

  {"input_data": [{
     "values": [["Hello world!"]]
                 }]
  }
  

• La charge utile de saisie du score doit être transmise en tant que valeur de paramètre d'entrée pour score. De cette manière, vous pouvez vous assurer que la valeur du paramètre score d'entrée est traitée correctement à l'intérieur du score.

• La charge d'entrée d'évaluation doit satisfaire les exigences d'entrée de la fonction Python concernée.

• La charge d'entrée de score doit inclure un tableau qui correspond à l'exemple de schéma de données d'entrée.

 {"input_data": [{
    "values": [["Hello, world!"]]
                }]
 }

Exemple Python de code (charge utile et jeton)

#wml_python_function
def my_deployable_function():

    def score(payload, token):

        message_from_input_payload = payload.get("input_data")[0].get("values")[0][0]
        response_message = "Received message - {0}".format(message_from_input_payload)

        # Score using the pre-defined model
        score_response = {
            'predictions': [{'fields': ['Response_message_field'],
                             'values': [[response_message]]
                            }]
        }
        return score_response

    return score

score = my_deployable_function()

Tester votre Python fonction

Voici comment vous pouvez tester votre Python fonction :

input_data = { "input_data": [{ "fields": [ "message" ],
                                "values": [[ "Hello, world!" ]]
                               }
                              ]
             }
function_result = score( input_data )
print( function_result )

Il renvoie le message « Hello, world! ».

Création de fonctions déployables via l'API REST

Pour les API REST, comme la Python fonction est téléchargée directement via un fichier, celui-ci doit déjà contenir la score fonction. Toute importation ponctuelle nécessaire pour une utilisation ultérieure dans la score fonction peut être effectuée dans la portée globale du fichier. Lorsque ce fichier est déployé en tant que fonction Python, les importations uniques disponibles dans la portée globale sont exécutées au cours du déploiement et sont simplement réutilisées avec chaque demande de prévision.

Exemple de fichier de fonctionscore :

Score function.py
---------------------
def score(input_data):
    return {'predictions': [{'values': [['Just a test']]}]}

Exemple de fonction score avec d'importation unique :

import subprocess
subprocess.check_output('pip install gensim --user', shell=True)
import gensim

def score(input_data):
    return {'predictions': [{'fields': ['gensim_version'], 'values': [[gensim.__version__]]}]}

Création de fonctions déployables via le client Python

Pour conserver une fonction Python en tant qu'actif, le client Python utilise la méthode wml_client.repository.store_function. Vous pouvez persister une Python fonction de deux façons :

• Grâce à un fichier contenant la Python fonction
• Grâce à l'objet fonction

def my_deployable_function():

    import subprocess
    subprocess.check_output('pip install gensim', shell=True)
    import gensim

    def score(input_data):
        import
        message_from_input_payload = payload.get("input_data")[0].get("values")[0][0]
        response_message = "Received message - {0}".format(message_from_input_payload)

        # Score using the pre-defined model
        score_response = {
            'predictions': [{'fields': ['Response_message_field', 'installed_lib_version'],
                             'values': [[response_message, gensim.__version__]]
                            }]
        }
        return score_response

    return score

function_meta = {
    client.repository.FunctionMetaNames.NAME:"test_function",
    client.repository.FunctionMetaNames.SOFTWARE_SPEC_ID: sw_spec_id
}
func_details = client.repository.store_function(my_deployable_function, function_meta)

score = my_deployable_function()

Utilisation de Vault pour déployer des fonctions déployables à exécution longue

Le jeton par défaut qu'une fonction Python déployable obtient au moment du déploiement expire (généralement après 1 heure) et ne peut pas être actualisé à l'intérieur score() sans informations d'identification durables.

Vault fournit des identifiants à longue durée de vie. En stockant une API, un nom d'utilisateur ou un mot de passe dans un secret Vault, vous pouvez :

• Récupérez ces informations d'identification en toute sécurité pendant le processus de déploiement (en utilisant le jeton à durée de vie limitée une seule fois).
• Utilisez-les pour initialiser un client à l'intérieur score() chaque fois que nécessaire. Ces informations d'identification n'expirent pas comme un jeton, ce qui permet au client de toujours se réauthentifier.

Prérequis

Vous devez créer une intégration de coffre-fort et un secret de coffre-fort dans IBMSoftware Hub pour utiliser la fonctionnalité de coffre-fort pour le déploiement Python de fonctions. Pour plus d'informations, consultez la section Gestion des secrets et des coffres-forts dans la IBMSoftware Hub documentation.

 def my_deployable_fun_with_secret(space_id=space_id, secret_name=secret_name):

     from ibm_watsonx_ai import APIClient, Credentials
     import requests
     import os

     # initialize a client, This will pick up token from the backend available in deploy flow.
     # This way of initializing will not work in score()
     host = os.environ.get("RUNTIME_ENV_APSX_URL")
     credentials = Credentials(
         instance_id="openshift",
         url=host,
         version="5.1",
     )
     wx_client = APIClient(credentials)
     first_token = wx_client._get_icptoken()

     # The user stores their username / password (could be apikey also)
     # in the secret that is identified by the "secret_name" in the code.
     # We use this first_token to fetch this information
     # The reason is: first_token will expire after some time, and we need
     # credentials that will not expire if we need to use it with a client
     # for the deployment lifetime.
     def get_secret_urn(secret_name):
         nonlocal host
         url = host + "/zen-data/v2/secrets"
         headers = {
             "Content-Type": "application/json",
             "Authorization": "Bearer " + token
         }
         params = {"secret_name": secret_name}

         response = requests.get(url, headers=headers, params=params, verify=False)
         secrets_list = response.json()["secrets"]
         return secrets_list[0]["secret_urn"]

     def get_secret_details(secret_urn):
         nonlocal host
         url = host + "/zen-data/v2/secrets/" + secret_urn
         headers = {
             "Content-Type": "application/json",
             "Authorization": "Bearer " + token
         }

         response = requests.get(url, headers=headers, verify=False)
         secret_details = response.json()
         return secret_details['data']['secret']['credentials']

     # we get the secret information, with the helper functions
     # get_secret_urn, get_secret_details
     secret_urn = get_secret_urn(secret_name)
     client_credentials = get_secret_details(secret_urn)

     vault_username = client_credentials['username']
     vault_password = client_credentials['password']

     # We now initialize wx_score_client a client object which
     # does not use token , but username and password
     from ibm_watsonx_ai import APIClient, Credentials
     creds = Credentials(
         url=os.environ.get("RUNTIME_ENV_APSX_URL"),
         username=vault_username,
         password=vault_password,
         instance_id="openshift",
         version="5.1"
     )
     wx_score_client = APIClient(credentials=creds)
     wx_score_client.set.default_space(space_id)

     def score(payload):
         # wx_score_client can be used inside score() function
         # and will work as long as the username / password it was used to initialize with
         # is not revoked
         use_score_token = wx_score_client.repository.list()["ID"].tolist()

         score_response = {
             "predictions": [
                 {
                     "fields": [
                         "use_score_token"
                     ],
                     "values": [
                         use_score_token
                     ]
                 }
             ]
         }
         return score_response

     return score

Création de fonctions déployables dans JupyterLab

Si vous créez votre fonction Python dans JupyterLab, votre code doit contenir un commentaire#wml_python_function. Si le commentaire est manquant, la fonction sera importée dans Watson Studio comme un actif de script plutôt qu'un actif de fonction Python.

Reportez-vous à cet exemple :

from tornado.escape import json_encode, json_decode, url_escape

# Define scoring function
def callModel(payload_scoring):

    print(json.dumps(payload_scoring))
    predictions =[]
    for value in payload_scoring:
        sums = []
        for value in payload_scoring["input_data"][0]["values"]:
            first = value[0]
            second = value[1]
            sums.append([first, second, first + second])
        predictions.append({"fields": ["FIRST", "SECOND", "SUM"],"values": sums})

    return {"predictions": predictions}


#wml_python_function
def score(input):

    """AI function example.

    Example:
      {"input_data": [{"fields":["FIRST","SECOND","values":[[1,2]]}]}
    """

    # Score using the pre-defined model
    prediction = callModel(input);

    return prediction

Initialisation dans le cadre du déploiement

Dans le cadre du déploiement, vous initialisez ibm-watson-studio-lib sans passer de paramètres supplémentaires. Voir l'exemple de code :

def my_deployable_function():
    from ibm_watson_studio_lib import access_project_or_space
    wslib = access_project_or_space()
    token = wslib.auth.get_current_token()
    def score( payload ):
        message_from_input_payload = payload.get("input_data")[0].get("values")[0][0]
        response_message = "Received message - {0}".format(message_from_input_payload)

        score_response = {
            'predictions': [{'fields': ['Response_message_field'],
                             'values': [[response_message]]
                            }]
        }

    return score

Initialisation dans la fonction de score

Dans la fonction score, vous devez passer votre jeton porteur en tant que paramètre. Voir l'exemple de code :

def my_deployable_function():
    from ibm_watson_studio_lib import access_project_or_space
    def score(payload , token):
        wslib = access_project_or_space({"token": token})

        message_from_input_payload = payload.get("input_data")[0].get("values")[0][0]
        response_message = "Received message - {0}".format(message_from_input_payload)

        score_response = {
            'predictions': [{'fields': ['Response_message_field'],
                             'values': [[response_message]]
                            }]
        }
    return score

Accéder aux données depuis la fonction de score

Vous souhaiterez peut-être accéder aux ressources de données à partir de la fonction de score. Exemple :

• Les données à distance doivent être consultées à l'aide des identifiants de l'utilisateur qui appelle la fonction de score.
• Dans votre cas d'utilisation spécifique, vous ne pouvez pas utiliser JDBC ou le Watson Machine LearningPython client.

Dans ces situations, si vous générez un extrait de code d'ingestion de données à partir de votre bloc-notes, le code échouera.

Consultez ces exemples de code pour obtenir des informations sur l'accès aux données à partir de la fonction score :

def my_deployable_function():
    import itc_utils.flight_service as itcfs
    from ibm_watson_studio_lib import access_project_or_space

    def score(payload, token):
        # token is from the predictions header.
        # Use it to initialise a Flight client here under score.
        # This ensures multi tenancy of predictions endpoint
        user_wslib = access_project_or_space({"token": token})

        # read the table named IRIS from database with the connection provided
        # the connection should be promoted to space prior to deployment.
        # The connection with name `db2_conn1` is expected to be already present under `space`
        db_query = {
            'connection_name': 'db2_conn1',
            'interaction_properties': { 'table_name': 'IRIS'}
        }

        # Fetch data with Flight client
        flight_client = itcfs.get_flight_client(wslib=user_wslib)
        flight_info = itcfs.get_flight_info(flight_client, nb_data_request=db_query, wslib=user_wslib)
        df = itcfs.read_pandas_and_concat(flight_client, flight_info, timeout=240)

        # return the first 2 rows with the column names
        score_response = {
            "predictions": [{
                "fields": list(df.columns),
                "values": df.iloc[:2].values.tolist()
            }]
        }
        return score_response

    return score

def my_deployable_function():
    import itc_utils.flight_service as itcfs
    from ibm_watson_studio_lib import access_project_or_space

    def score(payload, token):
        # token is from the predictions header.
        # Use it to initialise a Flight client here under score.
        # This ensures multi tenancy of predictions endpoint
        user_wslib = access_project_or_space({"token": token})

        # read the table named IRIS from database with the connection provided
        # the connection should be promoted to space prior to deployment.
        # The connection with name `db2_conn1` is expected to be already present under `space`
        db_query = {
            'connection_name': 'db2_conn1',
            'interaction_properties': { 'table_name': 'IRIS'}
        }

        # Fetch data with Flight client
        flight_client = itcfs.get_flight_client(wslib=user_wslib)
        flight_descriptor = itcfs.get_flight_descriptor(nb_data_request=db_query, wslib=user_wslib)
        flight_info = flight_client.get_flight_info(flight_descriptor)
        df = itcfs.read_pandas_and_concat(flight_client, flight_info, timeout=240)

        # return the first 2 rows with the column names
        score_response = {
            "predictions": [{
                "fields": list(df.columns),
                "values": df.iloc[:2].values.tolist()
            }]
        }
        return score_response

    return score