Créer des services d'IA

Définir un service d'IA avec la bibliothèque client Python

Copy link to section

Pour définir un service d'IA dans un notebook en utilisant la bibliothèque client Python watsonx.ai, suivez les étapes suivantes :

1. Pour travailler avec le service d'IA en Python, installez le SDK Python " ibm-watsonx-ai:

   pip install ibm-watsonx-ai
   Copy to clipboardCopié dans le presse-papiers

2. Après avoir installé la bibliothèque client Python, initialisez le client et définissez l'espace de déploiement par défaut :

   from ibm_watsonx_ai import APIClient
   from ibm_watsonx_ai import Credentials
   
   credentials = Credentials(
       url=url, api_key=apikey
   )
   
   client = APIClient(credentials)
   client.set.default_space(space_id=space_id)
   Copy to clipboardCopié dans le presse-papiers

3. Définissez votre service d'IA en Python en utilisant le schéma suivant :

   def basic_generate_demo(context, model="google/flan-t5-xl", **parameters):
    # "parameters" is a reserved argument and will be enabled in future
   
    # generate token from task credentials api
    task_token = context.generate_token()
   
    def generate(context):
        user_token = context.get_token()  # extract token from header
        user_headers = context.get_headers()
        json_body = context.get_json()
   
        # example 1: json
        return {
            "headers": {
                "Content-Type": "application/json",
                "user-custom-header": "my-header-x1",
            },
            "body": {
                "model": model
            },
        }
   
    def generate_stream(context):
        user_token = context.get_token()  # extract token from header
        user_headers = context.get_headers()
        json_body = context.get_json()
   
        # return a generator
        data_to_stream = json_body.get("sse", "Default message!")
        for x in data_to_stream:
            yield x
   
    def generate_batch(input_data_references, output_data_reference):
        # generate token from task credentials api
        task_token = context.generate_token()
        # do something.
        # ...
   
    return generate, generate_stream, generate_batch
   Copy to clipboardCopié dans le presse-papiersShow more

   Remarque :

   En fonction de votre cas d'utilisation, vous devez inclure au moins une de ces fonctions en tant que fonction imbriquée :

   • generate()
   • generate_stream()
   • generate_batch()

   Pour plus d'informations, voir Conditions requises pour la création d'un service AI.

Exigences relatives à la définition d'un service d'IA

Copy link to section

Le service d'IA capture la logique de votre cas d'utilisation d'IA générative (par exemple, une application de génération augmentée par récupération) et gère l'appel API REST au point de terminaison de déploiement " /ml/v4/deployments.

Suivez les lignes directrices suivantes pour définir un service d'IA :

• Créer une fonction Python. Vous pouvez donner n'importe quel nom à votre fonction. Pour en savoir plus sur les paramètres de la fonction, consultez la documentation de l'API REST dewatsonx.ai

• En fonction de votre cas d'utilisation, la fonction Python que vous souhaitez déployer doit inclure au moins une de ces fonctions en tant que fonction imbriquée dans son champ d'application :

  • generate()
  • generate_stream()
  • generate_batch()

• Lorsque vous utilisez la bibliothèque client Python watsonx.ai pour enregistrer la fonction Python qui contient une référence à une fonction externe, seul le code dans la portée de la fonction externe (y compris ses fonctions imbriquées) est enregistré. Par conséquent, le code en dehors de la portée de la fonction externe ne sera pas sauvegardé et ne sera donc pas disponible lorsque vous déployez la fonction.

def generate(context):
    user_token = context.get_token()
    headers = context.get_headers()
    json_body = context.get_json()

    return {
        "headers": {
             "Content-Type": "text/plain"
        },
        "body": "Hello WatsonX"
    }
Copy to clipboardCopié dans le presse-papiers

def generate_stream(context):
    user_token = context.get_token()
    headers = context.get_headers()
    json_body = context.get_json()

    for x in ["Hello", "WatsonX", "!"]:
        yield x
Copy to clipboardCopié dans le presse-papiers

def generate_batch(input_data_references: list[dict], output_data_reference: dict):
    # context from outer function is visible
    batch_token = context.generate_token()
    print(f"batch_token: {batch_token[-5:]}", flush=True)
    print(
        f"generate_batch:\n{input_data_references=}\n{output_data_reference=}",
        flush=True,
    )
Copy to clipboardCopié dans le presse-papiers

Exemple de code pour créer un service d'IA

Copy link to section

L'exemple de code définit un service AI " deployable_ai_service_f1. Lorsqu'une demande API REST est envoyée au point de terminaison " /ml/v4/deployments, " deployable_ai_service_f1 est invoqué. La fonction prend une charge d'entrée JSON et inclut les fonctions imbriquées suivantes dans son champ d'application :

• generate(): Effectue un appel à l'API REST vers le point de terminaison '/ml/v4/deployments/{id_or_name}/ai_service Il prend en charge un objet de contexte, extrait le jeton, les en-têtes et le corps JSON, et renvoie une réponse basée sur la clé de mode dans le corps JSON. Le format de la réponse peut être JSON, octets ou chaîne, avec des en-têtes personnalisés optionnels.
• generate_stream(): Appelle l'API REST au point de terminaison d'inférence SSE (Server-Sent Events) '/ml/v4/deployments/{id_or_name}/ai_service_stream. Il prend en charge un objet de contexte, extrait le jeton, les en-têtes et le corps JSON, et renvoie un flux d'événements ESS marqués par " eos (End of Stream).
• generate_batch(): Effectue un appel API REST vers le point de terminaison des emplois '/ml/v4/deployments_jobs. Il récupère " input_data_references et " output_data_reference du corps JSON de la demande, génère un jeton de lot et enregistre les références des données d'entrée et de sortie.

def deployable_ai_service_f1(context, params={"k1": "v1"}, **custom):
    """
    The outer function handles the REST call to the deployment endpoint
    POST /ml/v4/deployments

        context.generate_token() - generate a token from the task credentials

    To use `generate` and `generate_stream`, the deployment has to be ONLINE
    To use `generate_batch`, the deployment has to be BATCH
    """
    task_token = context.generate_token()
    print(f"outer function: {task_token[-5:]}", flush=True)

    def generate(context) -> dict:
        """
        The `generate` function handles the REST call to the inference endpoint
        POST /ml/v4/deployments/{id_or_name}/ai_service

            context.get_token()     - get the Bearer token from the header of the request
            context.get_json()      - get the body of the request
            context.get_headers()   - get the headers of the request

        The generate function should return a dict
        The following optional keys are supported currently
        - body
        - headers

        This particular example accepts a json body of the format:
        { "mode" : <value> }

        Depending on the <value> of the mode, it will return different response

        """
        user_token = context.get_token()
        headers = context.get_headers()
        json_body = context.get_json()
        print(f"my_generate: {user_token=}", flush=True)
        print(f"request headers: {headers=}", flush=True)
        print(f"json body: {json_body=}", flush=True)

        match json_body.get("mode", "no-match"):
            case "json":
                # response Content-Type is "application/json"
                return {
                    "headers": {
                        "Content-Type": "application/json",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": {
                        "user_token": user_token[-5:],
                        "task_token": task_token[-5:],
                        "json_body": json_body,
                        "params": params,
                        "custom": custom,
                    },
                }
            case "json-no-header":
                # response Content-Type is "application/json"
                return {
                    "body": {
                        "user_token": user_token[-5:],
                        "task_token": task_token[-5:],
                        "json_body": json_body,
                        "params": params,
                        "custom": custom,
                    },
                }
            case "json-custom-header":
                # response Content-Type is "text/plain; charset=utf-8; test-2"
                return {
                    "headers": {
                        "Content-Type": "text/plain; charset=utf-8; test-2",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": {
                        "user_token": user_token[-5:],
                        "task_token": task_token[-5:],
                        "json_body": json_body,
                        "params": params,
                        "custom": custom,
                    },
                }
            case "bytes":
                # response Content-Type is "application/octet-stream"
                return {
                    "headers": {
                        "Content-Type": "application/octet-stream",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": b"12345678910",
                }
            case "bytes-no-header":
                # response Content-Type is 'text/html; charset=utf-8'
                return {
                    "body": b"12345678910",
                }
            case "bytes-custom-header":
                # response Content-Type is "text/plain; charset=utf-8; test-2"
                return {
                    "headers": {
                        "Content-Type": "text/plain; charset=utf-8; test-2",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": b"12345678910",
                }
            case "str":
                # response Content-Type is "text/plain"
                return {
                    "headers": {
                        "Content-Type": "text/plain",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": f"Hello WatsonX: {json_body}",
                }
            case "str-no-header":
                # response Content-Type is "text/html; charset=utf-8"
                return {
                    "body": f"Hello WatsonX: {json_body}",
                }
            case "str-custom-header":
                # response Content-Type is "application/octet-stream; charset=utf-8; test-2"
                return {
                    "headers": {
                        "Content-Type": "application/octet-stream; charset=utf-8; test-2",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": f"Hello WatsonX: {json_body}",
                }
            case "negative-str-return":
                # Bad request
                return "Should give 400 bad request"
            case _:
                # response Content-Type is "text/html; charset=utf-8"
                return {"body": "No match"}

    def generate_stream(context):
        """
        The generate_stream function handles the REST call to the SSE inference endpoint
        POST /ml/v4/deployments/{id_or_name}/ai_service_stream

            context.get_token()     - get the Bearer token from the header of the request
            context.get_json()      - get the body of the request
            context.get_headers()   - get the headers of the request

        The generate_stream function be a python `generator` with yield
        The data in yield will the "data" for the SSE event

        Example: The following request json
            { "sse": ["Hello" , "", "WatsonX"," ", "!"]}
        will return the following stream of events
            --------------
            id: 1
            event: message
            data: Hello

            id: 2
            event: message
            data:

            id: 3
            event: message
            data: WatsonX

            id: 4
            event: message
            data:  

            id: 5
            event: message
            data: !

            id: 6
            event: eos
            ---------------
        The end of the stream will be marked by the event "eos"

        """
        user_token = context.get_token()
        headers = context.get_headers()
        json_body = context.get_json()
        print(f"generate_stream: {user_token=}", flush=True)
        print(f"generate_stream: {headers=}", flush=True)
        print(f"generate_stream: {json_body=}", flush=True)

        import time
        for x in json_body.get("sse", ["default", "message"]):
            time.sleep(1)
            yield x

    def generate_batch(input_data_references: list[dict], output_data_reference: dict) -> None:
        """
        The generate_batch function handles the REST jobs endpoint
        POST /ml/v4/deployments_jobs

            Arguments to the function are from the json body of the request to jobs
            - input_data_references : scoring.input_data_references
            - output_data_reference : scoring.output_data_reference

        context.generate_token() : can access context object
        from outer function scope if token is required
        """
        batch_token = context.generate_token()
        print(f"batch_token: {batch_token[-5:]}", flush=True)
        print(
            f"generate_batch:\n{input_data_references=}\n{output_data_reference=}",
            flush=True,
        )

    return generate, generate_stream, generate_batch
Copy to clipboardCopié dans le presse-papiersShow more

En savoir plus

Copy link to section

• Déployer des actifs de services d'IA

Sujet parent : Déployer des services d'IA avec le codage direct