Utilizzare route personalizzate arbitrarie in un singolo deployment

Nel servizio LLM moderno, i server dei modelli implementano e supportano più percorsi di inferenza per scopi diversi. Per questi casi d'uso, Vertex Inference consiglia di utilizzare il metodo invoke per accedere a più route in un singolo deployment.

Il metodo invoke può essere attivato durante il caricamento di un Model impostando invokeRoutePrefix come "/*". Una volta eseguito il deployment del modello in un endpoint, qualsiasi route non root sul server del modello sarà accessibile con la chiamata HTTP invoke. Ad esempio, "/invoke/foo/bar" verrà inoltrato come "/foo/bar" al server del modello.

Questa funzionalità è in fase di anteprima pubblica e presenta le seguenti limitazioni:

  • I modelli abilitati per l'invocazione possono essere sottoposti a deployment solo su un endpoint dedicato.
  • Per i modelli abilitati per l'invocazione sono supportate solo le chiamate HTTP e non RPC.
  • Quando carichi un modello, puoi impostare solo uno tra predictRoute o invokeRoutePrefix. Il valore predefinito è predictRoute. Se il campo invokeRoutePrefix è impostato su un modello, tutte le altre route Vertex, ad eccezione di invoke (ad es. :predict, :rawPredict ..), verranno disattivate dopo il deployment.
  • "/*" è l'unico valore consentito per invokeRoutePrefix ed espone tutti i percorsi non root. È consigliabile una gestione cauta per le route che non vuoi esporre.

Caricamento di un modello con invocazione abilitata

from google.cloud import aiplatform

invoke_enabled_model = aiplatform.Model.upload(
    display_name="invoke-enabled-model",
    serving_container_image_uri=IMAGE_URI,
    serving_container_invoke_route_prefix="/*",
    serving_container_health_route=HEALTH_ROUTE,
    serving_container_environment_variables={"KEY": "VALUE"},
    serving_container_args=[],
    sync=True,
)

Deployment di un modello abilitato all'invocazione

dedicated_endpoint = aiplatform.Endpoint.create(
    display_name="dedicated-endpoint-for-invoke-enabled-model",
    dedicated_endpoint_enabled=True,
    sync=True,
)

dedicated_endpoint.deploy(
    model=invoke_enabled_model,
    traffic_percentage=100,
    machine_type=MACHINE_TYPE,
    accelerator_type=ACCELERATOR_TYPE,
    accelerator_count=1,
    max_replica_count=1,
)

Effettuare una richiesta di inferenza su una route personalizzata arbitraria

La route di invocazione consente l'accesso a tutti i percorsi di richiesta non root nel deployment. Ad esempio, /invoke/foo/bar verrà inoltrato come /foo/bar al server del modello. Esistono due modi per accedere all'itinerario.

Richiesta di route personalizzata a un endpoint dedicato

La richiesta di chiamata a un endpoint dedicato verrà indirizzata a uno dei modelli di cui è stato eseguito il deployment in base alla configurazione di suddivisione del traffico.

def invoke_tabular_sample(
    project: str,
    location: str,
    endpoint_id: str,
    request_path: str,
    http_request_body: Dict[str, Any],
    stream: bool = False,
):
    aiplatform.init(project=project, location=location)

    dedicated_endpoint = aiplatform.Endpoint(endpoint_id)
    if stream:
        for chunk in dedicated_endpoint.invoke(
            request_path=request_path,
            body=json.dumps(http_request_body).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            stream=True,
        ):
            print(chunk)
    else:
        response = dedicated_endpoint.invoke(
            request_path=request_path,
            body=json.dumps(http_request_body).encode("utf-8"),
            headers={"Content-Type": "application/json"},
        )
        print(response)

Richiesta di route personalizzata a un modello di cui è stato eseguito il deployment

La richiesta di chiamata può essere effettuata per scegliere come target un modello specifico di cui è stato eseguito il deployment. Può essere utile per test e debug.

def invoke_direct_deployed_model_inference_tabular_sample(
    project: str,
    location: str,
    endpoint_id: str,
    request_path: str,
    http_request_body: Dict[str, Any],
    deployed_model_id: str,
    stream: bool = False,
):
    aiplatform.init(project=project, location=location)

    dedicated_endpoint = aiplatform.Endpoint(endpoint_id)
    if stream:
        for chunk in dedicated_endpoint.invoke(
            request_path=request_path,
            body=json.dumps(http_request_body).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            deployed_model_id=deployed_model_id,
            stream=True,
        ):
            print(chunk)
    else:
        response = dedicated_endpoint.invoke(
            request_path=request_path,
            body=json.dumps(http_request_body).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            deployed_model_id=deployed_model_id,
        )
        print(response)