Criação de funções implantáveis do Python

Requisitos da função de pontuação

• Duas maneiras de adicionar o objeto de função score existem:
  • explicitamente, por usuário
  • implicitamente, pelo método utilizado para salvar a função Python como um ativo no repositório Watson Machine Learning
• A score função pode aceitar um único parâmetro de entrada JSON ou dois parâmetros: payload e token de portador.
• A função score deve retornar um objeto serializável JSON (por exemplo: dicionários ou listas).

Assinaturas de funções de pontuação compatíveis

score(payload): A autenticação (por exemplo, usando wslib ou ibm-watsonx-ai) pode ser inicializada uma única vez no momento da implantação, fora do fluxo de pontuação, e reutilizada em todas as solicitações de previsão.

score(payload, token): A autenticação pode ser inicializada dentro da função de pontuação durante a fase de previsão, permitindo o acesso baseado em tokens, com possível sobrecarga por solicitação

Requisitos para a pontuação

• A carga útil de entrada da pontuação deve incluir um array com o nome values, conforme mostrado neste esquema de exemplo. O parâmetro input_data é obrigatório na carga útil.. O parâmetro input_data também pode incluir pares nome-valor adicionais.

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

• A carga útil de pontuação deve ser passada como um valor de parâmetro de entrada para score. Desta forma é possível garantir que o valor do parâmetro de entrada score seja tratado adequadamente dentro do score.

• A carga útil de entrada de pontuação deve corresponder aos requisitos de entrada para a Função de Python em questão.

• A carga útil de entrada para pontuação deve incluir um array que corresponda ao esquema de dados de entrada do Exemplo.

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

Exemplo de código do ` Python ` (carga útil e token)

#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()

Testando sua função ` Python `

Veja como você pode testar sua função ` Python `:

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

Ele exibe a mensagem "Olá, mundo!".

Criação de funções implantáveis por meio da API REST

Para APIs REST, porque a função Python é carregada diretamente por meio de um arquivo, o arquivo já deve conter a função score . Qualquer uma única vez de importação que precise ser feita para ser usada posteriormente dentro da função score pode ser feita dentro do escopo global do arquivo. Quando esse arquivo é implantado como uma função do Python, as importações únicas disponíveis no escopo global são executadas durante a implantação e, posteriormente, simplesmente reutilizadas em cada solicitação de previsão.

Exemplo score de arquivo de função:

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

Exemplo score de função com importações únicas:

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__]]}]}

Criação de funções implantáveis por meio do cliente do Python

Para persistir uma função do Python como um ativo, o cliente Python utiliza o wml_client.repository.store_function método. É possível salvar uma função do ` Python ` de duas maneiras:

• Por meio de um arquivo que contém a função ` Python `
• Por meio do objeto de função

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()

Utilização de funções de IA assíncronas

async def score_generator():
    """
    Async AI Function example
    """
    import os
    import sys
    import pandas as pd
    import importlib

    async def score(payload):
        module, attribute = payload["input_data"][0]["values"][0]
        m = importlib.import_module(module)
        attr_value = eval(attribute, m.__dict__)

        return {
            "predictions": [
                {
                    "values": [
                        {module: attr_value},
                        {"pandas": pd.__version__},
                        {"sys": sys.executable}
                    ]
                }
            ]
        }

    return score

score = score_generator()

Usando o Vault para implantar funções implantáveis de longa duração

O token padrão que uma função do Python implantável recebe no momento da implantação expira (geralmente após 1 hora) e não pode ser atualizado internamente score() sem credenciais duradouras.

O Vault fornece credenciais de longa duração. Ao armazenar uma API, um nome de usuário ou uma senha em um segredo do Vault, você pode:

• Recupere essas credenciais com segurança durante o fluxo de implantação (usando o token de curta duração apenas uma vez).
• Use-os para inicializar um cliente sempre score() que necessário. Essas credenciais não expiram como um token, portanto, o cliente sempre pode se autenticar novamente.

Pré-requisitos

É necessário criar uma integração com o Vault e um segredo do Vault em IBMSoftware Hub para utilizar a funcionalidade do Vault na implantação de funções do Python. Para obter mais informações, consulte a seção “Gerenciamento de segredos e cofres” na documentação do IBM Software Hub.

 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

Criação de funções implantáveis no JupyterLab

Se você criar sua função ` Python ` em ` JupyterLab, `, seu código deve conter um #wml_python_function comentário. Se o comentário estiver faltando, a função será importada para Watson Studio como um recurso de script, em vez de um recurso de função do Python.

Veja este exemplo:

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

Inicialização no escopo de implantação

No escopo da implantação, você inicializa ibm-watson-studio-lib sem passar nenhum parâmetro adicional. Veja o código de exemplo:

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

Inicialização na função de pontuação

Na função de pontuação, você deve passar seu token de portador como parâmetro. Veja o código de exemplo:

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

Acessando dados a partir da função de pontuação

Talvez você queira acessar os ativos de dados a partir da função de pontuação. Por exemplo:

• O acesso aos dados remotos deve ser feito utilizando as credenciais do usuário que está chamando a função de pontuação.
• No seu caso específico, você não pode usar o JDBC nem o cliente Watson Machine Learning Python.

Nessas situações, se você gerar um trecho de código de ingestão de dados a partir do seu notebook, o código não funcionará.

Veja estes exemplos de código como referência para acessar dados a partir da função 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