Usar rutas personalizadas arbitrarias en una sola implementación

En el servicio de LLMs moderno, los servidores de modelos implementan y admiten varias rutas de inferencia para diferentes propósitos. En esos casos prácticos, 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 al subir un Model si se define invokeRoutePrefix como "/*". Una vez que el modelo se haya desplegado en un endpoint, se podrá acceder a cualquier ruta que no sea raíz en el servidor del modelo mediante una llamada HTTP invoke. Por ejemplo, "/invoke/foo/bar" se reenviaría como "/foo/bar" al servidor del modelo.

Esta función está en fase de vista previa pública y tiene las siguientes restricciones:

  • Los modelos habilitados para invocar solo se pueden desplegar en un endpoint específico.
  • Solo se admiten llamadas HTTP para los modelos con invocación habilitada y no se admite RPC.
  • Al subir un modelo, solo se puede definir uno de los valores predictRoute o invokeRoutePrefix. El valor predeterminado es predictRoute. Si el campo invokeRoutePrefix se define en un modelo, todas las demás rutas de Vertex, excepto invoke (por ejemplo, :predict, :rawPredict, etc.), se inhabilitarán una vez desplegado.
  • "/*" es el único valor permitido para invokeRoutePrefix y expone todas las rutas que no son raíz. Se recomienda tener cuidado con las rutas que no quieras exponer.

Subir un modelo con invocación habilitada

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

Desplegar un modelo con invocación habilitada

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

Hacer una solicitud de inferencia en una ruta personalizada arbitraria

La ruta de invocación permite acceder a todas las rutas de solicitud que no sean de 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 endpoint específico

Las solicitudes a un endpoint específico se dirigirán a uno de los modelos desplegados en función de 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 desplegado

Se puede enviar una solicitud de invocación para orientarla a un modelo implementado específico. Puede ser útil para hacer pruebas y depurar errores.

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)