File size: 3,107 Bytes
703811e 8675aad 703811e 8675aad 703811e 8675aad 1977567 703811e 145adb7 765e3b9 59e7a3a 145adb7 703811e 145adb7 703811e 8675aad 145adb7 703811e 765e3b9 703811e 8675aad 703811e 8675aad 703811e 8675aad 703811e 8675aad 5005c12 8675aad 703811e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
import numpy as np
import time
import os, sys
from pathlib import Path
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from concrete.ml.deployment import FHEModelClient
import requests
def to_json(python_object):
if isinstance(python_object, bytes):
return {"__class__": "bytes", "__value__": list(python_object)}
raise TypeError(repr(python_object) + " is not JSON serializable")
def from_json(python_object):
if "__class__" in python_object:
return bytes(python_object["__value__"])
API_URL = "https://puqif7goarh132kl.us-east-1.aws.endpoints.huggingface.cloud"
headers = {
"Authorization": "Bearer " + os.environ.get("HF_TOKEN"),
"Content-Type": "application/json",
}
def query(payload):
response = requests.post(API_URL, headers=headers, json=payload)
if "error" in response:
assert False, f"Got an error: {response=}"
return response.json()
path_to_model = Path("compiled_model")
x, y = make_classification(n_samples=1000, class_sep=2, n_features=30, random_state=42)
_, X_test, _, y_test = train_test_split(x, y, test_size=0.2, random_state=42)
# Recover parameters for client side
fhemodel_client = FHEModelClient(path_to_model)
# Generate the keys
fhemodel_client.generate_private_and_evaluation_keys()
evaluation_keys = fhemodel_client.get_serialized_evaluation_keys()
# Save the key in the database
payload = {
"inputs": "fake",
"evaluation_keys": to_json(evaluation_keys),
"method": "save_key",
}
uid = query(payload)["uid"]
print(f"Storing the key in the database under {uid=}")
# Test the handler
nb_good = 0
nb_samples = len(X_test)
verbose = False
time_start = time.time()
duration = 0
for i in range(nb_samples):
# Quantize the input and encrypt it
encrypted_inputs = fhemodel_client.quantize_encrypt_serialize([X_test[i]])
# Prepare the payload
payload = {
"inputs": "fake",
"encrypted_inputs": to_json(encrypted_inputs),
"method": "inference",
"uid": uid,
}
if verbose or True:
print(f"Size of the payload: {sys.getsizeof(payload) / 1024} kilobytes")
# Run the inference on HF servers
duration -= time.time()
duration_inference = -time.time()
encrypted_prediction = query(payload)
duration += time.time()
duration_inference += time.time()
encrypted_prediction = from_json(encrypted_prediction)
# Decrypt the result and dequantize
prediction_proba = fhemodel_client.deserialize_decrypt_dequantize(encrypted_prediction)[0]
prediction = np.argmax(prediction_proba)
if verbose or True:
print(
f"for {i}-th input, {prediction=} with expected {y_test[i]} in {duration_inference:.3f} seconds"
)
# Measure accuracy
nb_good += y_test[i] == prediction
print(f"Accuracy on {nb_samples} samples is {nb_good * 1. / nb_samples}")
print(f"Total time: {time.time() - time_start} seconds")
print(f"Duration in inferences: {duration} seconds")
print(f"Duration per inference: {duration / nb_samples} seconds")
|