Criação de serviços de IA

Definição de um serviço de IA com a biblioteca de clientes Python

Para definir um serviço de IA em um notebook usando a biblioteca cliente watsonx.ai Python, siga estas etapas:

1. Para trabalhar com o serviço de IA em Python, instale o SDK Python ' ibm-watsonx-ai:

   pip install ibm-watsonx-ai
   

2. Depois de instalar a biblioteca do cliente Python, inicialize o cliente e defina o espaço de implantação padrão:

   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)
   

3. Defina seu serviço de IA em Python usando o seguinte layout:

   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
   

   Nota:

   Dependendo do seu caso de uso, você deve incluir pelo menos uma dessas funções como uma função aninhada:

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

   Para obter mais informações, consulte Requisitos para criar um serviço de IA.

Requisitos para definir um serviço de IA

O serviço de IA captura a lógica de seu caso de uso de IA generativa (por exemplo, um aplicativo de geração aumentada por recuperação) e lida com a chamada da API REST para o ponto de extremidade de implantação ' /ml/v4/deployments.

Siga estas diretrizes para definir um serviço de IA:

• Crie uma função Python. Você pode especificar qualquer nome para sua função. Para saber mais sobre os parâmetros da função, consulte a documentação da API RESTwatsonx.ai.

• Dependendo do seu caso de uso, a função Python que você deseja implantar deve incluir pelo menos uma dessas funções como uma função aninhada em seu escopo:

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

• Quando você usa a biblioteca cliente watsonx.ai Python para salvar a função Python que contém uma referência a uma função externa, somente o código no escopo da função externa (incluindo suas funções aninhadas) é salvo. Portanto, o código fora do escopo da função externa não será salvo e, assim, não estará disponível quando você implementar a função.

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"
    }

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

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

Código de amostra para criar um serviço de IA

O código de exemplo define um serviço de IA " deployable_ai_service_f1. Quando uma solicitação da API REST é enviada para o endpoint " /ml/v4/deployments, o " deployable_ai_service_f1 é chamado. A função recebe uma carga útil de entrada JSON e inclui as seguintes funções aninhadas como parte de seu escopo:

• generate(): Faz uma chamada da API REST para o ponto de extremidade ' /ml/v4/deployments/{id_or_name}/ai_service. Ele recebe um objeto de contexto, extrai o token, os cabeçalhos e o corpo JSON e retorna uma resposta com base na chave de modo no corpo JSON. O formato da resposta pode ser JSON, bytes ou string, com cabeçalhos personalizados opcionais.
• generate_stream(): Faz uma chamada à API REST para o ponto de extremidade de inferência SSE (Server-Sent Events) ' /ml/v4/deployments/{id_or_name}/ai_service_stream. Ele recebe um objeto de contexto, extrai o token, os cabeçalhos e o corpo JSON e retorna um fluxo de eventos SSE marcados por ' eos (Fim do fluxo).
• generate_batch(): Faz uma chamada da API REST para o ponto de extremidade dos trabalhos ' /ml/v4/deployments_jobs. Ele recebe ' input_data_references e ' output_data_reference do corpo JSON da solicitação, gera um token de lote e registra as referências de dados de entrada e saída.

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

Saiba mais

• Implementação de ativos de serviço de IA

Tópico principal: Implementação de serviços de IA com codificação direta