Utiliser des routes personnalisées arbitraires sur un seul déploiement

Dans la diffusion moderne de LLM, les serveurs de modèles implémentent et prennent en charge plusieurs routes d'inférence à des fins différentes. Pour ces cas d'utilisation, Vertex Inference recommande d'utiliser la méthode invoke pour accéder à plusieurs routes sur un seul déploiement.

La méthode invoke peut être activée lors de l'importation d'un Model en définissant invokeRoutePrefix sur "/*". Une fois le modèle déployé sur un point de terminaison, toute route non racine sur le serveur de modèles sera accessible avec l'appel HTTP d'invocation. Par exemple, "/invoke/foo/bar" sera transféré en tant que "/foo/bar" au serveur de modèles.

Cette fonctionnalité est en version Preview publique et présente les restrictions suivantes :

  • Les modèles compatibles avec l'invocation ne peuvent être déployés que sur un point de terminaison dédié.
  • Seuls les appels HTTP sont compatibles avec les modèles compatibles avec l'invocation, et RPC n'est pas compatible.
  • Lors de l'importation d'un modèle, vous ne pouvez définir qu'un seul des éléments suivants : predictRoute ou invokeRoutePrefix. La valeur par défaut est predictRoute. Si le champ invokeRoutePrefix est défini sur un modèle, toutes les autres routes Vertex, à l'exception de invoke (par exemple, :predict, :rawPredict, etc.), seront désactivées une fois déployées.
  • "/*" est la seule valeur autorisée pour invokeRoutePrefix, et elle expose tous les chemins non racine. Il est conseillé de gérer avec précaution les routes que vous ne souhaitez pas exposer.

Importer un modèle compatible avec l'invocation

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

Déployer un modèle compatible avec l'invocation

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

Envoyer une requête d'inférence sur une route personnalisée arbitraire

La route d'invocation permet d'accéder à tous les chemins de requête non racine dans le déploiement. Par exemple, /invoke/foo/bar sera transféré en tant que /foo/bar au serveur de modèles. Il existe deux façons d'accéder à la route.

Requête de route personnalisée vers un point de terminaison dédié

La requête d'invocation vers un point de terminaison dédié sera acheminée vers l'un des modèles déployés en fonction de la configuration de répartition du trafic.

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)

Requête de route personnalisée vers un modèle déployé

Une requête d'invocation peut être envoyée pour cibler un modèle déployé spécifique. Cela peut être utile pour les tests et le débogage.

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)