Usa rutas personalizadas arbitrarias en una sola implementación

En la entrega moderna de LLM, los servidores de modelos implementan y admiten varias rutas de inferencia para diferentes propósitos. Para esos casos de uso, Vertex Inference recomienda usar el método invoke para acceder a varias rutas en una sola implementación.

El método invoke se puede habilitar cuando se sube un Model configurando invokeRoutePrefix como "/*". Una vez que el modelo se implementa en un extremo, se podrá acceder a cualquier ruta no raíz en el servidor de modelos con la llamada HTTP de invocación. Por ejemplo, "/invoke/foo/bar" se reenviaría como "/foo/bar" al servidor de modelos.

Esta función se encuentra en la etapa de versión preliminar pública y tiene las siguientes restricciones:

  • Los modelos habilitados para la invocación solo se pueden implementar en un extremo dedicado.
  • Solo se admiten llamadas HTTP para los modelos habilitados para la invocación, y no se admite RPC.
  • Cuando se sube un modelo, solo se puede establecer uno de los parámetros predictRoute o invokeRoutePrefix. El valor predeterminado es predictRoute. Si el campo invokeRoutePrefix está configurado en un modelo, todas las demás rutas de Vertex, excepto invoke (p. ej., :predict, :rawPredict…), se inhabilitarán una vez que se implemente.
  • "/*" es el único valor permitido para invokeRoutePrefix y expone todas las rutas que no son raíz. Se recomienda un manejo cauteloso para las rutas que no deseas exponer.

Cómo subir un modelo habilitado para la invocación

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

Implementación de un modelo habilitado para la invocación

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

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

Cómo realizar una solicitud de inferencia en una ruta personalizada arbitraria

La ruta de invocación permite el acceso a todas las rutas de solicitud que no son raíz en la implementación. Por ejemplo, /invoke/foo/bar se reenviará como /foo/bar al servidor del modelo. Hay dos formas de acceder a la ruta.

Solicitud de ruta personalizada a un extremo dedicado

La solicitud de invocación a un extremo dedicado se enrutará a uno de los modelos implementados según la configuración de división del tráfico.

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)

Solicitud de ruta personalizada a un modelo implementado

Se puede realizar una solicitud de invocación para segmentar un modelo implementado específico. Puede ser útil para realizar pruebas y depurar.

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)