Implementierbare Python -Funktionen schreiben

Anforderungen an die Funktion 'score'

• Es gibt zwei Möglichkeiten, das score Funktionsobjekt hinzuzufügen:
  • explizit, nach Benutzer
  • implizit durch die Methode, die zum Speichern der Python-Funktion als Asset im Watson Machine Learning-Repository verwendet wird
• Die score Funktion kann einen einzelnen JSON-Eingabeparameter oder zwei Parameter akzeptieren: Nutzlast und Bearer-Token.
• Die Funktion score muss ein serialisierbares JSON-Objekt (z. B. Wörterverzeichnisse oder Listen) zurückgeben.

Anforderungen an Scoring-Eingabe

• Die Scoring-Eingabenutzdaten müssen ein Array mit dem Namen values enthalten, wie in diesem Beispielschema gezeigt. Der input_data Parameter ist in der Nutzlast obligatorisch. Der input_data Parameter kann auch zusätzliche Name-Wert-Paare enthalten.

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

• Die Scoring-Eingabedaten müssen als Eingabeparameterwert für scoreübergeben werden. Auf diese Weise können Sie sicherstellen, dass der Wert des score Eingabeparameters innerhalb der entsprechend verarbeitet score wird.

• Die Scoring-Eingabenutzdaten müssen den Anforderungen für die Eingabe der betreffenden Python-Funktion entsprechen.

• Die Scoring-Eingabenutzdaten müssen ein Array enthalten, das dem Datenschema der Beispieleingabe entspricht.

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

Beispielcode Python (Nutzlast und 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()

Testen Sie Ihre Python Funktion

So können Sie Ihre Python Funktion testen:

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

Es gibt die Meldung „Hallo, Welt!“ zurück.

Bereitstellbare Funktionen über REST-API erstellen

Bei REST-APIs muss die score Funktion bereits in der Datei enthalten sein, da die Python Funktion direkt über eine Datei hochgeladen wird. Jeder einmalige Import, der für die spätere Verwendung innerhalb der score Funktion erforderlich ist, kann im globalen Bereich der Datei vorgenommen werden. Wenn diese Datei als Python-Funktion bereitgestellt wird, werden die im globalen Bereich verfügbaren einmaligen Importe während der Bereitstellung ausgeführt und später bei jeder Vorhersageanforderung einfach erneut verwendet.

Beispielfunktionsdatei score:

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

Beispielfunktion score mit einmaligen Importen:

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

Bereitstellbare Funktionen über den Python-Client erstellen

Um eine Python-Funktion als Asset als persistent zu definieren, verwendet der Python-Client die Methode wml_client.repository.store_function. Sie können eine Python Funktion auf zwei Arten beibehalten:

• Durch eine Datei, die die Python Funktion enthält
• Durch das Funktionsobjekt

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

Verwendung von Vault zum Bereitstellen von lang laufenden Funktionen

Das Standard-Token, das eine bereitstellbare Python Funktion zum Zeitpunkt der Bereitstellung erhält, läuft ab (in der Regel nach einer Stunde) und kann ohne dauerhafte score() Anmeldedaten nicht innerhalb aktualisiert werden.

Vault stellt langlebige Anmeldedaten bereit. Durch das Speichern einer API, eines Benutzernamens oder eines Passworts in einem Vault-Geheimnis können Sie:

• Rufen Sie diese Anmeldedaten während des Bereitstellungsablaufs sicher ab (verwenden Sie das kurzlebige Token nur einmal).
• Verwenden Sie sie, um bei Bedarf einen score() Client innerhalb zu initialisieren. Diese Anmeldedaten verfallen nicht wie ein Token, sodass sich der Client jederzeit erneut authentifizieren kann.

Voraussetzungen

Sie müssen eine Tresorintegration und einen Tresorgeheimnis in erstellen, um die Tresorfunktionalität für IBMSoftware Hub die Bereitstellung von Python Funktionen nutzen zu können. Weitere Informationen finden Sie unter Verwalten von Geheimnissen und Tresoren in der IBMSoftware Hub Dokumentation.

 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

Bereitstellbaren Funktionen in JupyterLab erstellen

Wenn Sie Ihre Python-Funktion in JupyterLab erstellen, muss Ihr Code den Kommentar #wml_python_function enthalten. Wenn der Kommentar fehlt, wird die Funktion nicht als Python-Funktionsasset, sondern als Script-Asset in Watson Studio importiert.

Sehen Sie sich dieses Beispiel an:

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

Initialisierung im Bereitstellungsbereich

Im Bereitstellungsbereich initialisieren ibm-watson-studio-lib Sie ohne zusätzliche Parameter zu übergeben. Siehe Beispielcode:

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

Initialisierung innerhalb der Score-Funktion

In der Score-Funktion müssen Sie Ihr Bearer-Token als Parameter übergeben. Siehe Beispielcode:

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

Zugriff auf Daten aus der Score-Funktion

Möglicherweise möchten Sie innerhalb der Score-Funktion auf Datenbestände zugreifen. Beispiel:

• Auf Remote-Daten muss mit den Anmeldedaten des Benutzers zugegriffen werden, der die Score-Funktion aufruft.
• In Ihrem speziellen Anwendungsfall können Sie weder noch den JDBCWatson Machine LearningPython Client verwenden.

Wenn Sie in solchen Situationen einen Code-Schnipsel für die Datenerfassung aus Ihrem Notizbuch heraus generieren, schlägt der Code fehl.

Siehe diese Codebeispiele als Referenz für den Zugriff auf Daten aus der Score-Funktion heraus:

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