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

Dans le service LLM moderne, 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 même 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, tout itinéraire non racine sur le serveur de modèles sera accessible avec un appel HTTP d'invocation. Par exemple, "/invoke/foo/bar" serait 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 activés pour l'invocation ne peuvent être déployés que sur un point de terminaison dédié.
  • Seuls les appels HTTP sont acceptés pour les modèles activés par l'appel, et RPC n'est pas accepté.
  • Lorsque vous importez un modèle, vous ne pouvez définir qu'une seule des valeurs 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 le modèle déployé.
  • "/*" est la seule valeur autorisée pour invokeRoutePrefix. Elle expose tous les chemins non racine. Il est conseillé de manipuler avec précaution les routes que vous ne souhaitez pas exposer.

Importer un modèle avec l'appel activé

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 avec appel activé

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

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

La route d'appel permet d'accéder à tous les chemins de requête non racine du déploiement. Par exemple, /invoke/foo/bar sera transmis en tant que /foo/bar au serveur de modèle. Il existe deux façons d'accéder à l'itinéraire.

Router une requête personnalisée vers un point de terminaison dédié

Les requêtes d'appel à un point de terminaison dédié seront acheminées 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)

Envoyer une requête d'itinéraire personnalisé à un modèle déployé

Les requêtes d'appel peuvent être envoyées 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)