Gravando funções Python implementáveis

Requisitos de função de pontuação

• Existem duas maneiras de adicionar o objeto score de função:
  • explicitamente, pelo usuário
  • implicitamente, pelo método que é usado 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: carga útil e token de portador.
• A função score deve retornar um objeto serializável JSON (por exemplo: dicionários ou listas).

Requisitos de entrada de pontuação

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

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

• A carga útil de entrada da pontuação deve ser passada como um valor de parâmetro de entrada para score. Dessa forma, você pode garantir que o valor do parâmetro score de entrada 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 de pontuação deve incluir uma matriz que corresponde ao Esquema de dados de entrada de exemplo.

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

Código Python de exemplo (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 Python função

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

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

Retorna a mensagem "Olá, mundo!".

Criando funções implementáveis por meio da API de REST

Para APIs REST, como a Python função é carregada diretamente por meio de um arquivo, o arquivo já deve conter a score função. Qualquer importação única que precise ser feita para ser usada posteriormente dentro da score função pode ser feita dentro do escopo global do arquivo. Quando este arquivo é implementado como uma função Python, as importações pontuais disponíveis no escopo global sejam executadas durante a implementação e posteriormente simplesmente reutilizadas com cada solicitação de predição.

Amostra do arquivo de função score:

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

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

Criando funções implementáveis por meio do cliente Python

Para persistir uma função Python como um ativo, o cliente Python usa o método wml_client.repository.store_function. Você pode persistir uma Python função de duas maneiras:

• Através de um arquivo que contém a Python função
• Através 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()

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

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

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

• Obtenha 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 dentro score() sempre que necessário. Essas credenciais não expiram como um token, portanto, o cliente sempre pode se autenticar novamente.

Pré-requisitos

Você deve criar uma integração de cofre e um segredo de cofre em IBMSoftware Hub para usar a funcionalidade de cofre para implantar Python funções. Para obter mais informações, consulte Gerenciando segredos e cofres na IBMSoftware Hub documentação.

 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

Criando funções implementáveis no JupyterLab

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

Consulte 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 da implantação

No escopo de 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 dentro da função de pontuação

Na função de pontuação, você deve passar seu token de portador como um 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

Você pode querer acessar os ativos de dados a partir da função de pontuação. Por Exemplo:

• Os dados remotos devem ser acessados usando 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 Watson Machine LearningPython cliente JDBC ou o cliente.

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

Veja estes exemplos de código como referência para acessar dados a partir da função de pontuação:

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