In this blog post, I will share my weekly report on the Implementation of Quantum Neural Network as a part of QCourse-551 under the supervision of mentors from QWorld and Classiq. Classical neural networks have brought advanced capabilities to solve problems we couldn't before. I will be developing this project and publishing research work in the upcoming four months starting today! I will also be participating in the Classiq Bootcamp and Hackathon in October. It's going to be a lot of fun!
Project Detail
Quantum Neural Networks involve combining classical neural networks with the advantage of quantum information to create more efficient algorithms. The goal of this project is to understand how we can utilize the quantum capabilities to create quantum layers. We will create a hybrid network of classical and quantum layers to classify the MNIST dataset.
Mentor: Tal Michaeli
Team Members: Ashmit JaiSarita Gupta (Me), Asif Saad, Roman Ledenov
GitHub Repository of my work: devilkiller-ag/QNN-MNIST-Classification (github.com)
Presentation Slides: QNN-MNIST-Classification/PresentationSlides at main · devilkiller-ag/QNN-MNIST-Classification (github.com)
Workflow
[🗸] Understanding the basics of Quantum Neural Networks and Classiq.
[🗸] Decide the Quantum Neural Network Architecture. (VQC?)
[🗸] Preprocess the MNIST dataset for compatibility with our quantum algorithm. Convert pixel values to a format suitable for quantum circuits. (Angle Encoding)
[🗸] Develop a quantum encoding scheme to represent MNIST digits using qubits. This step involves mapping classical data to quantum states.
[🗸] Create Quantum Neural Network.
[...] Experimentation: Training & Testing
Week 1
Installed the Classiq SDK with the QML extension.
Organized the first group meeting (with Asif Asad) and decided to go through the Classiq QNN Documentation and the paper on Quantum On Chip Training.
Read the Paper on Quantum On Chip Training with Parameter Shift and Gradient Pruning.
Exploring the Torch Quantum Python Library (Website | Repository) and the implementation of a simple QNN for MNIST Training.
Attend the two lessons of the Classiq Bootcamp.
I connected to Amirali Malekani Nezhad and Roman Ledenov today.
For Week 2, we plan to discuss the project in detail, create a project timeline, divide tasks among us, and start implementing it.
I am participating in the Quantum Games Hackathon (30th Sept. - 8th Oct. 2023) so maybe my output will be a little low in Week 2 but I will try to cover everything in Week 3.
Here are my findings from the paper I read:
Parameterized Quantum Circuit (PQC) Gradient Can be obtained by parameter shift whose cost scales linearly with the number of qubits. Quantum On Chip QOC) PQC Training with parameter shift gives gradient however gradients obtained from naïve parameter shift have low fidelity and thus degrading the training accuracy. We can use probabilistic gradient pruning to identify gradients with potentially large errors and then remove them. Specifically, small gradients have larger relative errors than large ones, thus having a higher probability of being pruned.
Result: The results demonstrate that our on-chip training achieves over 90% and 60% accuracy for 2-class and 4-class image classification tasks. The probabilistic gradient pruning brings up to 7% PQC accuracy improvements over no pruning.
Methodology: To enable PQC on-chip learning, we first use an in-situ quantum gradient computation via parameter shift and its real QC implementation. A probabilistic gradient pruning method is then used to save the gradient computation cost with enhanced noise-robustness and training efficiency.
Week 2
I wasn't able to work much in Week 2 because I was participating in a Quantum Games Hackathon 2023 where I developed QuantaVania an action-adventure 2D platformer game with the potential to evolve into an open-world sandbox game in which players can learn quantum computing from the ground up while playing, design their game level and share it with others in the quantum community via our web platform, and mine qubits, quantum gates and power-ups. Our game will not only allow them to run the game on their local device but also real quantum computers and simulators from various quantum computing providers like IBM Quantum, IONQ, Rigetti Computing, etc.
We intend to teach the players quantum computing as they go through the levels. We'll expose them to qubits in the first level, and then they'll have to find the X-Gate hiding behind any box or monster. As the levels progress, the player will discover new gates that he may use in the gun circuit. And, at the end of each level, we will introduce to the user each quantum algorithm, from basic to advanced, in the form of a game problem.
We won the Special Category Prize by Snarto/Onyx for demonstrating Logistics Optimization using Quantum Computing through our game.
Week 3
This week was very difficult for our team. Most of our mentors and organizers of the QCourse 551-1 are based in Israel. Unfortunately, Hamas attacked Israel which led to War in the region. Due to this and lack of communication in the last week, our team came into silent mode. We weren't able to decide on common meeting timing and communication media. We almost lost our hope of being able to initiate the project.
Week 4
Thanks to our mentor Tal, he didn't lose faith in us and motivated us to start from zero. He organized a team meeting guided us in detail on how to start and gave us an introduction to the Classiq Documentation on Quantum Neural Networks. We restarted our work and formed a WhatsApp discussion group. All three of us are from different backgrounds and we knew little about Quantum Neural Network. Our plan for this week was to introduce ourselves to the unfamiliar concepts we will be using in this project.
Here is the list of things I did this week:
They studied Tensors and PyTorch Basics from this tutorial on YouTube by Mr. P. Solver.
I have studied the implementation of a Basic Neural Network by following this tutorial.
Studied Cross-Entropy Loss Function (tutorial).
Studied Sequential Neural Network (tutorial).
A general "sequential" neural network can be expressed as
$$f(x) = \underset{i=1}{\overset{n}{\Huge{\kappa}}} R_i(A_ix+b_i)$$
where
$$\underset{i=1}{\overset{n}{\Huge{\kappa}}}f_i(x) = f_n \circ f_{n-1} \circ \ldots \circ f_1(x)$$
and the A_i are matrices and the b_i are bias vectors. Typically the R_i are the same for all the layers (typically ReLU) except for the last layer, where R_i is just is just the identity function. In clever architectures, like convolutional neural networks, the A_i's become sparse matrices (most of their parameters are fixed to equal zero).
Implemented Sequential Neural Network for MNIST Dataset using PyTorch by following this tutorial. Here is the notebook for my implementation.
Studied Quantum Neural Networks, Quantum Layer, and Datasets provided by Classiq through Classiq Documentation on QNN.
Implemented Quantum Neural Network using Classiq and PyTorch to determine the correct angle for Rx Gate for performing a "NOT" Gate. Here is the notebook for my implementation.
For implementing QNN for the MNIST Dataset, the first thing we need to do is encode images into Qubits. The MNIST dataset contains 28x28 px images which we want to encode in less than 6 qubits/image. We can do this by using algorithms like Quantum Probability Image Encoding (Resources: Qiskit Textbook and Medium article on QPIE and QHED by Jimin Lee) or Flexible Representation of Quantum Images or Novel Enhanced Quantum Representation of Images. We need to research which option will be best for our purpose.
Code for Implemented Sequential Neural Network for MNIST Dataset using PyTorch:
# Imports
import torch
import torch.nn as nn
from torch.optim import SGD
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
import torchvision
import numpy as np
import matplotlib.pyplot as plt
class CTDataset(Dataset):
def __init__(self, filepath):
self.x, self.y = torch.load(filepath)
self.x = self.x / 255.
self.y = F.one_hot(self.y, num_classes=10).to(float)
def __len__(self):
return self.x.shape[0]
def __getitem__(self, ix):
return self.x[ix], self.y[ix]
class MyNeuralNet(nn.Module):
def __init__(self):
super().__init__()
self.Matrix1 = nn.Linear(28**2,100)
self.Matrix2 = nn.Linear(100,50)
self.Matrix3 = nn.Linear(50,10)
self.R = nn.ReLU()
def forward(self,x):
x = x.view(-1,28**2)
x = self.R(self.Matrix1(x))
x = self.R(self.Matrix2(x))
x = self.Matrix3(x)
return x.squeeze()
# Dataset and DataLoader
train_ds = CTDataset('MNIST/processed/training.pt')
test_ds = CTDataset('MNIST/processed/test.pt')
train_dl = DataLoader(train_ds, batch_size=5)
# Loss Function
L = nn.CrossEntropyLoss()
# Network
f = MyNeuralNet()
# Training
def train_model(dl, f, n_epochs=20):
# Optimization
opt = SGD(f.parameters(), lr=0.01)
L = nn.CrossEntropyLoss()
# Train model
losses = []
epochs = []
for epoch in range(n_epochs):
print(f'Epoch {epoch}')
N = len(dl)
for i, (x, y) in enumerate(dl):
# Update the weights of the network
opt.zero_grad()
loss_value = L(f(x), y)
loss_value.backward()
opt.step()
# Store training data
epochs.append(epoch+i/N)
losses.append(loss_value.item())
return np.array(epochs), np.array(losses)
epoch_data, loss_data = train_model(train_dl, f)
# Plot of cross entropy averaged per epoch
epoch_data_avgd = epoch_data.reshape(20,-1).mean(axis=1)
loss_data_avgd = loss_data.reshape(20,-1).mean(axis=1)
plt.plot(epoch_data_avgd, loss_data_avgd, 'o--')
plt.xlabel('Epoch Number')
plt.ylabel('Cross Entropy')
plt.title('Cross Entropy (avgd per epoch)')
# Plotting 40 training results
xs, ys = train_ds[0:2000]
yhats = f(xs).argmax(axis=1)
fig, ax = plt.subplots(10,4,figsize=(10,15))
for i in range(40):
plt.subplot(10,4,i+1)
plt.imshow(xs[i])
plt.title(f'Predicted Digit: {yhats[i]}')
fig.tight_layout()
plt.show()
# Testing and plotting 40 predictions
xs, ys = test_ds[:2000]
yhats = f(xs).argmax(axis=1)
fig, ax = plt.subplots(10,4,figsize=(10,15))
for i in range(40):
plt.subplot(10,4,i+1)
plt.imshow(xs[i])
plt.title(f'Predicted Digit: {yhats[i]}')
fig.tight_layout()
plt.show()
Code for Implementing Quantum Neural Network using Classiq and PyTorch to determine the angle of Rx Gate to make it an X Gate
# Imports
from typing import Dict
from classiq import Model ,synthesize
from classiq.builtin_functions import HardwareEfficientAnsatz
from classiq import QReg
from classiq.applications.qnn import QLayer
from classiq.applications.qnn.datasets import DATALOADER_NOT
from classiq.applications.qnn.types import (
MultipleArguments,
SavedResult,
ResultsCollection
)
from classiq.execution import execute_qnn
from classiq.synthesis import SerializedQuantumProgram
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader
# Step 1: Creating Quantum Layer
## Step 1.1: Create Parametric Quantum Circuit (PQC)
_NUM_QUBITS = 1
_REPS = 1
_CONNECTIVITY_MAP = "circular"
def add_rx(md: Model, prefix: str, in_wire=None) -> Dict[str, QReg]:
if in_wire is not None:
kwargs = { "in_wires": { "IN": in_wire["OUT"] } }
else:
kwargs = {}
hwea_params = HardwareEfficientAnsatz(
num_qubits=_NUM_QUBITS,
connectivity_map=_CONNECTIVITY_MAP,
reps=_REPS,
one_qubit_gates="rx",
two_qubit_gates=[],
parameter_prefix=prefix,
)
return md.HardwareEfficientAnsatz(hwea_params, **kwargs)
model = Model()
output_1 = add_rx(model, "input_")
output_2 = add_rx(model, "weight_", output_1)
quantum_program = synthesize(model.get_model())
## Step 1.2: Create the execution and post-processing
def execute(quantum_program: SerializedQuantumProgram, arguments:MultipleArguments) -> ResultsCollection:
return execute_qnn(quantum_program, arguments)
def post_process(result: SavedResult) -> torch.Tensor:
"""
Take in a `SavedResult` with `ExecutionDetails` value type, and return the
probability of measuring |0> which equals the amount of `|0>` measurements
divided by the total amount of measurements.
"""
counts: dict = result.value.counts
# The probability of measuring |0>
p_zero: float = counts.get("0", 0.0) / sum(counts.values())
return torch.tensor(p_zero)
## Step 1.3: Create a network
class QNet(torch.nn.Module):
def __init__(self, *args, **kwargs) -> None:
super().__init__()
self.qlayer = QLayer(
quantum_program,
execute,
post_process,
*args,
**kwargs,
)
def forward(self, x: torch.Tensor) -> torch.Tensor:
x = self.qlayer(x)
return x
model = QNet()
# Step 2: Choose a Dataset, Loss Function, and Optimizer
_LEARNING_RATE = 1.0
data_loader = DATALOADER_NOT
loss_function = nn.L1Loss()
optimizer = optim.SGD(model.parameters(), lr=_LEARNING_RATE)
# Step 3: Training
def train(model: nn.Module, data_loader: DataLoader, loss_function: nn.modules.loss._Loss, optimizer: optim.Optimizer, epoch: int = 20) -> None:
for index in range(epoch):
print(index, model.qlayer.weight)
for data, label in data_loader:
optimizer.zero_grad()
output = model(data)
loss = loss_function(output, label)
loss.backward()
optimizer.step()
train(model, data_loader, loss_function, optimizer)
# Step 4: Testing
def check_accuracy(model: nn.Module, data_loader: DataLoader, atol=1e-4) -> float:
num_correct = 0
total = 0
model.eval()
with torch.no_grad():
for data, labels in data_loader:
predictions = model(data)
is_prediction_correct = predictions.isclose(labels, atol=atol)
print(f"data: {data}\n labels: {labels}\n is_prediction_correct: {is_prediction_correct}\n sum: {is_prediction_correct.sum()}\n item: {is_prediction_correct.sum().item()}")
num_correct += is_prediction_correct.sum().item()
total += labels.size(0)
accuracy = float(num_correct) / float(total)
print(f"Test Accuracy of the model: {accuracy*100:.2f}")
return accuracy
check_accuracy(model, data_loader)
# The results show that the accuracy is 1, meaning a 100% success rate
# at performing the required transformation (i.e. the network learned to
# perform a X-gate). We may further test it by printing the value of
# model.qlayer.weight, which is a tensor of shape (1,1), which should,
# after training, be close to pi=3.1416.
A sneak peek at Quantum Circuit Game Engine I published this week
This week, I transformed a segment of my previous quantum game projects into a Python package. This move aims to facilitate quantum game developers in seamlessly incorporating quantum circuits into their Quantum games built on the Pygame platform.
The features I have included are:
Modular and Abstract Code.
All configurations are in one place in the
config.pyfile.Developers can create a Quantum Circuit for any number of qubit/wires and circuit width (max. number of gates which can be applied in a wire) of their choice.
Easy to change UI by replacing color configs and graphics for gates with those of your choice.
Easy to change the size of the Quantum Circuit by adjusting
QUANTUM_CIRCUIT_TILE_SIZE,GATE_TILE_WIDTH, andGATE_TILE_HIEGHTin theconfig.pyfile.Easily change controls by changing keys in the
handle_input()method of theQuantumCircuitGridclass.
If this project is helpful for you or you liked my work, consider supporting me through Ko.fi🍵. Also, kindly consider giving a star to this repository.😁
You can install the QCGE python package using pip install qcge. Explore more on the PyPI page of this package and the GitHub repository.
Week 5
This week I mainly researched the flow we have to take for building our Quantum Neural Network and several methods to encode dataset images into a Quantum Circuit. I am adding the flow I have decided at the top of the blog for easy access and check-marked the steps that are completed.
Initially, I explored image processing encoding methods like FRQI, NERQ, and QPIXL++. However, from my research, I concluded that these techniques are primarily designed for quantum image processing tasks rather than traditional machine learning datasets like MNIST. Our MNIST Dataset contains images of 28px x 28px size. QPIXL++ is among the latest and most efficient methods to encode images into quantum circuits but it is implemented in C++ and this too is suitable for quantum image processing. I found the following methods good for our dataset images but these require too many qubits.
Quantum Encoding Techniques:
Amplitude Encoding:
- Amplitude encoding represents pixel values as probability amplitudes of quantum states. This will require around 8 to 10 qubits per pixel. For a 28x28 color image, a rough estimate suggests the need for approximately (28 x 28 x 8) to (28 x 28 x 10) qubits. Precision in representing pixel intensities plays a key role in determining the qubit requirements.
Angle Encoding:
- Angle encoding involves mapping pixel values to angles in quantum states. Similar to amplitude encoding, the qubit requirement is (28 x 28 x 8) to (28 x 28 x 10) qubits, depending on the precision needed for the angles.
Binary Encoding:
- Binary encoding represents each pixel with a binary string. For colorful images using 8 bits per pixel, the qubit requirement is (28 x 28 x 8) qubits. This method offers simplicity in representation.
Quantum Feature Maps:
- Quantum feature maps transform classical data into quantum states using quantum gates. The qubit requirement is comparable to amplitude and angle encoding, ranging from (28 x 28 x 8) to (28 x 28 x 10) qubits.
Quantum Circuit Encoding:
- Quantum circuit encoding represents the entire image with a quantum circuit. The qubit requirements depend on the circuit's depth and complexity, potentially exceeding amplitude and angle encoding.
Quantum Convolutional Networks (QCN):
- QCNs simulate classical convolutional layers using quantum gates. Qubit requirements can be high, depending on the network's architecture and design choices.
In our case, I have used 1 qubit instead of 8-10 qubits per pixel, so my implementation uses 28x28 qubits for image encoding. Initial Tal suggested using Angle Encoding for our dataset, but I am not sure if using these many qubits is practical for our approach or not. We need to discuss this in detail with him.
I have written code for loading images and encoding them using Amplitude and Angle Encoding Techniques. The notebook for it is available in the GitHub repository I mentioned at the start of the blog.
Imports
import numpy as np
import math
import matplotlib.pyplot as plt
from matplotlib import style
from PIL import Image
style.use('default')
from qiskit import QuantumCircuit
Image Loading
# Load Image
def load_image(img_path, image_size):
# Load the image from filesystem
image_raw = np.array(Image.open(img_path))
print('Raw Image Info: ', image_raw.shape)
print('Raw Image Datatype: ', image_raw.dtype)
# Convert the RBG component of the image to B&W image, as a numpy (uint8) array
image = []
for i in range(image_size):
image.append([])
for j in range(image_size):
image[i].append(image_raw[i][j][0] / 256)
image = np.array(image)
print('Image shape (numpy array): ', image.shape)
# Display the image
plt.title('Big Image')
plt.xticks(range(0, image.shape[0]+1, 32))
plt.yticks(range(0, image.shape[1]+1, 32))
plt.imshow(image, extent=[0, image.shape[0],
image.shape[1], 0], cmap='viridis')
plt.show()
return image
Plot Image
def plot_image(img, title: str):
plt.title(title)
plt.xticks(range(img.shape[0]+1, 32))
plt.yticks(range(img.shape[1]+1, 32))
plt.imshow(img, extent=[0, img.shape[0], img.shape[1], 0], cmap='viridis')
plt.show()
Amplitude Encoding
def amplitude_encode_image(image, size):
# Create a quantum circuit
num_qubits = size * size
qc = QuantumCircuit(num_qubits, num_qubits)
# Amplitude encode each pixel value into the quantum state
for i in range(size):
for j in range(size):
intensity = image[i, j]
amplitude = np.sqrt(intensity / 256) # Map intensity to amplitude
qc.ry(2 * np.arcsin(amplitude), i * size + j)
return qc
## USAGE ##
image_path = './assets/mnist-0-28.png'
image_size = 28 # Original Image-Width
image = load_image(image_path, image_size)
amplitude_encoded_circuit = amplitude_encode_image(image, size=image_size)
print(amplitude_encoded_circuit)
Angle Encoding
def angle_encode_image(image, size):
# Create a quantum circuit
num_qubits = size * size
qc = QuantumCircuit(num_qubits, num_qubits)
# Angle encode each pixel value into the quantum state
for i in range(size):
for j in range(size):
intensity = image[i, j]
angle = intensity * (2 * np.pi / 256) # Map intensity to angle
qc.ry(angle, i * size + j)
return qc
## USAGE ##
image_path = './assets/mnist-0-28.png'
image_size = 28 # Original Image-Width
image = load_image(image_path, image_size)
angle_encoded_circuit = angle_encode_image(image, size=image_size)
print(angle_encoded_circuit)
Both techniques use 28x28 = 784 qubits for encoding the image into a quantum circuit. I have saved the output circuit drawing in Txt format in the following files located in the same directory: amplitude_encoding_output.txt and angle_encoding_output.txt.
Week 6
My end-semester exams have started this week (November 6th, 2023 to November 27th, 2023). After Discussing the above encoding schemes in the bi-weekly meeting with our mentor (Tal Michaeli) We decided on the Quantum Neural Network Architecture for our purpose as given below. We also planned to go ahead with the following steps as a classical pre-processing layer:
Compress/Shrink the dataset images using classical methods to bring down the number of pixels in the image and consequently the number of qubits required.
Use angle encoding to encode two pixels per qubit by using two rotation gates (Rx and Ry) on each qubit.
Quantum Neural Network Architecture:
This Hybrid QNN architecture combines classical and quantum processing for efficient MNIST digit classification. Classical data preprocessing is followed by quantum encoding, where classical information is translated into quantum states.
The Quantum Layer utilizes quantum gates to perform computations, followed by Classical Layer(s) for further processing. The Quantum/Classical interfacing or pooling layer integrates quantum and classical information, leading to classical output post-processing. Finally, the classification step provides the ultimate prediction.
This hybrid approach leverages the strengths of quantum computing while maintaining compatibility with classical methods, offering a promising paradigm for enhanced machine learning tasks on quantum devices.
Step 1: Classical Data Pre-Processing: Shrinking the Image:
We are using PyTorch transforms to resize our dataset images of 28 x 28 px to shrink them into 10 x 10 px images. Shrinking the images more is resulting in loss of data which will increase the error rate.
def resize_image(image, resize_value=4):
resize_transform = transforms.Compose([
transforms.ToPILImage(),
transforms.Resize((resize_value, resize_value)),
transforms.ToTensor()
])
re_image = resize_transform(image)
re_image = re_image.squeeze()
return re_image
def plot_numpy_image(image, title='image'):
plt.imshow(image, cmap='viridis')
plt.title(title)
plt.colorbar()
plt.show()
This is the plot of the first 4 data images without compression:
fig, ax = plt.subplots(1, 4, figsize=(10, 15))
for i in range(4):
plt.subplot(1,4,i+1)
plt.imshow(x[i])
plt.title(f'Digit is: {y[i]}')
fig.tight_layout()
plt.savefig(f"original.png")
plt.show()
After Resizing this the output images shrinked to 10x10px:
resize_value = 10
fig, ax = plt.subplots(1, 4, figsize=(10, 15))
for i in range(4):
plt.subplot(1,4,i+1)
image = x[i].numpy()
re_image = resize_image(image, resize_value)
plt.imshow(re_image)
plt.title(f'Digit is: {y[i]}')
fig.tight_layout()
plt.savefig(f"compressed_to_{resize_value}_px.png")
plt.show()
Step 2: Quantum Encoding: Angle Encoding
We used angle encoding to encode two pixels per qubit by using two rotation gates (Rx and Ry) on each qubit. The angle of the rotation rate is decided according to the intensity of pixels.
def angle_encode_image(image, size):
# Create a quantum circuit
num_qubits = size * size // 2
qc = QuantumCircuit(num_qubits, num_qubits)
# Angle encode pairs of pixel values into the quantum state
for i in range(0, size, 2):
for j in range(0, size, 2):
intensity1 = image[i, j]
intensity2 = image[i, j + 1]
angle_rx = intensity1 * (2 * np.pi / 256) # Map intensity to Rx angle
angle_ry = intensity2 * (2 * np.pi / 256) # Map intensity to Ry angle
qubit_index = (i // 2) * (size // 2) + (j // 2)
qc.rx(angle_rx, qubit_index)
qc.ry(angle_ry, qubit_index)
return qc
Week 7
(I was having my exam this week.) This Tuesday, I presented our work progress in front of all teams. This week was a little stressful for me due submission deadline for recording my three talks at the Data Science Conference 2023 and my End-Semester Exams. Asif and Roman were not able to present the work so I had to give the presentation alone. However, they helped me a lot in creating the slides for the presentation. Here is my presentation slide.
I haven't worked much after this due to two exams this week. However, I had a meeting with Asif where we discussed our next step and we found an interesting paper on Efficient Learning for Deep Quantum Neural Networks which Asif will explore before our next meet. We haven't been able to contact Roman for the last week so we are unaware of his status on the project.
Week 8
(I was having my exam this week.) This week, I tried to write code for every individual part of the circuit. I was a little confused about the API of Classiq and how to use them. I tried to write this with the help of internet searches and classic documentation.
This is the implementation I followed to build Hybrid QNN as specified in this paper:
- Classical Compression Layer
Center Crop the 28x28 image to 24x24 image
Down Sample to 4x4 image
Flatten the Image
- Quantum Encoding Layer
- Encode the flattened image into a 4-qubit circuit using RX, RY, RZ, and RY gates as discussed earlier.
- Quantum Entanglement Layer
- This itself contains four entangling layers as described in the paper: " (i) RZZ layer: add RZZ gates to all logical adjacent wires and the logical farthest wires to form a ring connection, for example, an RZZ layer in a 4-qubit circuit contains 4 RZZ gates which lie on wires 1 and 2, 2 and 3, 3 and 4, 4 and 1; (ii) RXX layer: same structure as in RZZ layer; (iii) RZX layer: same structure as in RZZ layer; (i𝑣) CZ layer: add CZ gates to all logical adjacent wires."
- Hybrid Neural Network
- To combine all these layers.
import torch
import torchvision.transforms as transforms
from torch import nn
import matplotlib.pyplot as plt
from PIL import Image
import numpy as np
from typing import Dict
import classiq
from classiq import Model, QReg, RX, RY, RZ, synthesize
from classiq.builtin_functions import HardwareEfficientAnsatz
classiq.authenticate()
_NUM_QUBITS = 4
_CONNECTIVITY_MAP = "circular"
# Classical Layer for Image Commpression:The input MNIST images are all 28 × 28.
# This Classical Layer will firstly center-crop them to 24 × 24 and then down-sample them to 4 × 4 for MNIST.
class ClassicalCompressionLayer(nn.Module):
def __init__(self):
super(ClassicalCompressionLayer, self).__init__()
self.center_crop = transforms.CenterCrop((24, 24))
self.down_sample = transforms.Resize((4, 4))
self.flatten = nn.Flatten()
def forward(self, x):
x = self.center_crop(x)
x = self.down_sample(x)
x = self.flatten(x)
return x
# Quantum Layer for Encoding: The output of Classical Compression Layer is
# encoded by this quantum layer into a quantum circuit. We use Angle encoding to encode 4 pixels per qubit using RX, RY, RZ, and RX gate on each qubit.
class QuantumEncodingLayer(Model):
def __init__(self):
super().__init__()
def encode_pixels(self, pixel_values: torch.Tensor) -> Dict[str, QReg]:
# Split pixel values into groups of 4
pixel_groups = pixel_values.split(4)
# Initialize dictionary to store qubit outputs
qubit_outputs = {}
# Encode each group of 4 pixels into angles for RX, RY, RZ, RX gates
for i, pixel_group in enumerate(pixel_groups):
rx_angle = pixel_group[0] * (2 * torch.pi / 255)
ry_angle = pixel_group[1] * (2 * torch.pi / 255)
rz_angle = pixel_group[2] * (2 * torch.pi / 255)
rx2_angle = pixel_group[3] * (2 * torch.pi / 255)
# Apply gates to corresponding qubit
qubit_outputs[f"qubit_{i}"] = RX(rx_angle) & RY(ry_angle) & RZ(rz_angle) & RX(rx2_angle)
return qubit_outputs
# Quantum Layer for Entanglement
class QuantumEntanglementLayer(Model):
def __init__(self):
super().__init__()
def add_entanglement_layer(self) -> Dict[str, QReg]:
hwea_params = HardwareEfficientAnsatz(
num_qubits=_NUM_QUBITS,
connectivity_map=_CONNECTIVITY_MAP,
one_qubit_gates=[],
two_qubit_gates=["rzz, rxx, rzx, cz"],
)
return self.HardwareEfficientAnsatz(hwea_params)
# Hybrid Quantum Neural Network
class HybridQuantumNeuralNetwork(Model):
def __init__(self):
super().__init__()
# Instantiate Layer
self.classical_compression_layer = ClassicalCompressionLayer()
self.encoding_layer = QuantumEncodingLayer()
self.entanglement_layer = QuantumEntanglementLayer()
# Import Data
# Add Encoding Layer
encoding_out = self.encoding_layer.encode_pixels()
# Add Entanglement Layer
entanglement_out = self.entanglement_layer.add_rzz_layer()
# Add layers to the model
self.add(encoding_out, entanglement_out)
def forward(self, x):
# Classical Compression
compressed_data = self.classical_compression_layer(x)
# Quantum Encoding
encoding_result = self.encoding_layer(compressed_data)
# Quantum Entanglement
entanglement_result = self.entanglement_layer()
# Concatenate quantum and classical outputs
output = self.concatenate(
[encoding_result, entanglement_result, compressed_data]
)
return output
# hybrid_model = HybridQuantumNeuralNetwork()
# quantum_program = synthesize(hybrid_model.get_model())
Upon review with Tal, I learned that my class-based implementation was wrong and that we needed to have a single model for both the encoding and entanglement layers.
Week 9
In week 9, I traveled back home from my college (which took three days). We started doing peer programming on every alternate day. We tried to figure out the individual circuits for encoding, entanglement, and cz-block. We also discussed the post-processing. Here is the code we updated from the code I wrote in week 8:
import torch
import classiq
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import torch.nn.functional as F
from typing import Dict
from classiq import Model, synthesize, QReg, QFunc
from classiq.builtin_functions import HardwareEfficientAnsatz
from classiq.applications.qnn import QLayer
from classiq.execution import execute_qnn
from classiq.synthesis import SerializedQuantumProgram
from classiq.applications.qnn.types import (
MultipleArguments,
SavedResult,
ResultsCollection,
)
classiq.authenticate()
# constants
_NUM_QUBITS = 4
_REPS = 1
_FULLY_CONNECTED_MESH = [[0, 1], [1, 2], [2, 3], [3, 0]]
_LEARNING_RATE = 1.0
def add_entanglement(md: Model, prefix: str, in_wire=None) -> Dict[str, QReg]:
if in_wire is not None:
kwargs = { "in_wires": { "IN": in_wire["OUT"] } }
else:
kwargs = {}
hwea_params = HardwareEfficientAnsatz(
num_qubits=_NUM_QUBITS,
connectivity_map=_FULLY_CONNECTED_MESH,
reps=_REPS,
one_qubit_gates=[],
two_qubit_gates=["rzz", "rxx", "rzx"],
parameter_prefix=prefix,
)
return md.HardwareEfficientAnsatz(hwea_params, **kwargs)
model = Model()
out1 = add_entanglement(model, "input_")
out2 = add_entanglement(model, "weight_", out1)
quantum_program = synthesize(model.get_model())
def execute(quantum_program: SerializedQuantumProgram, arguments: MultipleArguments) -> ResultsCollection:
return execute_qnn(quantum_program, arguments)
# TODO: MODIFY THIS
# Post-process the result, returning a dict:
# Note: this function assumes that we only care about
# differentiating a single state (|0>)
# from all the rest of the states.
# In case of a different differentiation, this function should change.
def post_process(result: SavedResult) -> torch.Tensor:
"""
Take in a `SavedResult` with `ExecutionDetails` value type, and return the
probability of measuring |0> which equals the amount of `|0>` measurements
divided by the total amount of measurements.
"""
counts: dict = result.value.counts
# The probability of measuring |0>
p_zero: float = counts.get("0", 0.0) / sum(counts.values())
return torch.tensor(p_zero)
However, we still were confused about how to connect all these parts of the model.
Week 10
This week I successfully developed the full Model for our QNN after receiving help from Tal in the weekly meeting. Unfortunately, Asif and Roman missed this week's meeting. So I tried to ask doubts of all three from Tal. Here is the correct working implementation of the Model:
import classiq
import torch.nn as nn
import torchvision.transforms as transforms
from classiq import create_model, synthesize, show, QFunc, QArray, QBit, Output, allocate, RX, RY, RZ, RZZ, RXX, RYY, CZ
classiq.authenticate()
@QFunc
def encoding(q: QArray[QBit]) -> None:
"""
This function encodes the input data into the qubits. This input data is a 4x4 image pixel values
converted into angle for rotation gates (RX, RY, RZ, RX) in form of a 16x1 vector.
We encode 4 pixels per qubit.
Args:
q (QArray[QBit]): Array of four Qubits to encode the input data into.
"""
RX(theta="input_0", target=q[0]) # Pixel 0 on Qubit 0
RY(theta="input_1", target=q[0]) # Pixel 1 on Qubit 0
RZ(theta="input_2", target=q[0]) # Pixel 2 on Qubit 0
RX(theta="input_3", target=q[0]) # Pixel 3 on Qubit 0
RX(theta="input_4", target=q[1]) # Pixel 4 on Qubit 1
RY(theta="input_5", target=q[1]) # Pixel 5 on Qubit 1
RZ(theta="input_6", target=q[1]) # Pixel 6 on Qubit 1
RX(theta="input_7", target=q[1]) # Pixel 7 on Qubit 1
RX(theta="input_8", target=q[2]) # Pixel 8 on Qubit 2
RY(theta="input_9", target=q[2]) # Pixel 9 on Qubit 2
RZ(theta="input_10", target=q[2]) # Pixel 10 on Qubit 2
RX(theta="input_11", target=q[2]) # Pixel 11 on Qubit 2
RX(theta="input_12", target=q[3]) # Pixel 12 on Qubit 3
RY(theta="input_13", target=q[3]) # Pixel 13 on Qubit 3
RZ(theta="input_14", target=q[3]) # Pixel 14 on Qubit 3
RX(theta="input_15", target=q[3]) # Pixel 15 on Qubit 3
@QFunc
def mixing(q: QArray[QBit]) -> None:
"""
This function performs the mixing operation on the qubits.
This is done by applying a series of RZZ, RXX, RYY gates to form a
ring connection.
Args:
q (QArray[QBit]): Array of four Qubits to apply the mixing operation on.
"""
RZZ(theta="weight_0", target=q[0:2])
RZZ(theta="weight_1", target=q[1:3])
RZZ(theta="weight_2", target=q[2:4])
# RZZ(theta="weight_3", target=q[3:1])
RXX(theta="weight_4", target=q[0:2])
RXX(theta="weight_5", target=q[1:3])
RXX(theta="weight_6", target=q[2:4])
# RXX(theta="weight_7", target=q[3:1])
RYY(theta="weight_8", target=q[0:2])
RYY(theta="weight_9", target=q[1:3])
RYY(theta="weight_10", target=q[2:4])
# RYY(theta="weight_11", target=q[3:1])
@QFunc
def cz_block(q: QArray[QBit]) -> None:
"""
This function applies CZ gates between each qubit.
Args:
q (QArray[QBit]): Array of four Qubits to apply the entanglement operation on.
"""
CZ(control=q[0], target=q[1])
CZ(control=q[1], target=q[2])
CZ(control=q[2], target=q[3])
@QFunc
def main(res: Output[QArray[QBit]]) -> None:
"""
This is the main function from which model will be created.
It calls the other functions to perform the encoding, mixing and entanglement.
Args:
res (Output[QArray[QBit]]): Output QArray of QBits from which the model will be created.
"""
allocate(4, res)
encoding(q=res)
mixing(q=res)
cz_block(q=res)
# Create a model
model = create_model(main)
quantum_program = synthesize(model)
show(quantum_program)
Here is the resultant quantum program from this model:
Our next step will be to design the post-process function and then start the training part.
Week 11
This week I worked on getting the dataset, pre-processing it, creating the data loader for testing and training data, quantum neural network, and training and Testing functions. Let's discuss them one by one.
- Setting Device Agnostic Code
device = "cuda" if torch.cuda.is_available() else "cpu"
device
Now let's prepare the data to be passed into our quantum neural network.
- Defining functions for pre-processing data images and corresponding labels
Before getting our dataset, let's just quickly define how we want our input MNIST images should be pre-processed. As discussed earlier, the input MNIST images are all 28 × 28 px. We want to first center-crop them to 24 × 24 and then down-sample them to 4 × 4 for MNIST. Then we convert the image pixels into angles for passing them into Rotation gates later for encoding.
import torch
import torchvision.transforms as transforms
def input_transform(image):
"""
The input MNIST images are all 28 × 28 px. This function will firstly center-crop
them to 24 × 24 and then down-sample them to 4 × 4 for MNIST. Then we convert
the image pixels into angles for passing them into Rotation gates later for encoding.
"""
image = transforms.ToTensor()(image)
image = transforms.CenterCrop(24)(image)
image = transforms.Resize(size = (4,4))(image)
image = image.squeeze()
image_pixels = torch.flatten(image)
angles = torch.sqrt(image_pixels / 256)
return angles
I have also defined an empty function in which we can define how we want to transform labels of data before comparing it from the output of our Quantum Layer.
def target_transform(x):
return x
- Getting a dataset
We are using the MNIST dataset provided by torchvision.datasets. We are using the pre-processing functions input_transform and target_transform defined above to transform our data into a usable format.
from torchvision import datasets
# Setup training data
train_data = datasets.MNIST(
root="data",
train=True,
download=True,
transform=input_transform,
target_transform=target_transform
)
# Setup testing data
test_data = datasets.MNIST(
root="data",
train=False,
download=True,
transform=input_transform,
target_transform=target_transform
)
Let's see what we have got here:
len(train_data), len(test_data) ## Output: (60000, 10000)
Hmm.. We have got 60000 training data and 10000 testing data. Let's see what our data looks like:
# See the first training example
image, label = train_data[0]
image, label
## OUTPUT:
# (tensor([0.0000, 0.0000, 0.0317, 0.0378, 0.0000, 0.0336, 0.0000, 0.0000,
# 0.0000, 0.0000, 0.0477, 0.0000, 0.0295, 0.0620, 0.0000, 0.0000]), 5)
You can notice we have a tensor of 16 angles corresponding to an MNIST image of digit 5. Let's go further and prepare Dataloader for training and testing.
- Preparing Dataloader
Let's for now create a subset of training and testing data containing only 64 data for quick experimenting.
from torch.utils.data import Subset
# Define the size of the subset
subset_size = 64
# Create subsets of the datasets
train_subset = Subset(train_data, range(subset_size))
test_subset = Subset(test_data, range(subset_size))
Let's create 2 Batches of size 32 out of these:
# Setup the batch size hyperparameter
BATCH_SIZE = 32
# Turn datasets into iterables (batches)
train_dataloader = DataLoader(train_subset,
batch_size=BATCH_SIZE,
shuffle=True
)
test_dataloader = DataLoader(test_subset,
batch_size=BATCH_SIZE,
shuffle=False
)
Let's visualize this:
# Let's check out what we've created
print(f"Dataloaders: {train_dataloader, test_dataloader}")
print(f"Length of train dataloader: {len(train_dataloader)} batches of {BATCH_SIZE}")
print(f"Length of test dataloader: {len(test_dataloader)} batches of {BATCH_SIZE}")
## Output
# Dataloaders: (<torch.utils.data.dataloader.DataLoader object at 0x7f905f89b9d0>, <torch.utils.data.dataloader.DataLoader object at 0x7f906042cfd0>)
# Length of train dataloader: 2 batches of 32
# Length of test dataloader: 2 batches of 32
- Quantum Model
There is no change to the Quantum Model we discussed last week.
- Creating our Quantum Neural Network
We have passed the Quantum Layer we created last week into a Quantum Layer in a Quantum Neural Network.
import torch.nn as nn
from classiq.applications.qnn import QLayer
class Net(torch.nn.Module):
def __init__(self, *args, **kwargs) -> None:
super().__init__()
self.qlayer = QLayer(
quantum_program,
execute,
post_process,
*args,
**kwargs
)
def forward(self, x: torch.Tensor) -> torch.Tensor:
x = self.qlayer(x)
return x
qnn = Net()
The execute function passed into the QLayer is:
from classiq.execution import execute_qnn
from classiq.synthesis import SerializedQuantumProgram
from classiq.applications.qnn.types import (
MultipleArguments,
SavedResult,
ResultsCollection,
)
def execute(quantum_program: SerializedQuantumProgram, arguments: MultipleArguments) -> ResultsCollection:
return execute_qnn(quantum_program, arguments)
We haven't defined the Post Processing function passed into the QLayer yet (this will be the task of next week). For now, we are just using a template post-processing function from QNN Documentation provided by Classiq:
def post_process(result: SavedResult) -> torch.Tensor:
counts: dict = result.value.counts
# print(f"counts: {counts}")
# The probability of measuring |0>
p_zero: float = counts.get("0", 0.0) / sum(counts.values())
return torch.tensor(p_zero)
- Defining Loss Function and Optimizer
import torch.nn as nn
import torch.optim as optim
_LEARNING_RATE = 1.0
# choosing our loss function
loss_fn = nn.L1Loss()
# choosing our optimizer
optimizer = optim.SGD(qnn.parameters(), lr=_LEARNING_RATE)
- Defining Training and Testing Loop
from tqdm.auto import tqdm ## For showing a progress bar
def train(
model: nn.Module,
data_loader: DataLoader,
loss_fn: nn.modules.loss._Loss,
optimizer: optim.Optimizer,
epochs: int = 20,
) -> None:
train_loss = 0
model.to(device)
for epoch in tqdm(range(epochs)):
print(f"Epoch: {epoch}\n----------")
for batch, (data, label) in enumerate(data_loader):
# Send data to device (GPU or CPU)
data, label = data.to(device), label.to(device)
# 1. Forward pass
output = model(data)
# 2. Calculate loss
loss = loss_fn(output, label)
train_loss += loss
# 3. Optimizer zero grad
optimizer.zero_grad()
# 4. Loss backward
loss.backward()
# 5. Optimizer step
optimizer.step()
# Calculate loss per epoch and print out what's happening
train_loss /= len(data_loader)
print(f"Train loss: {train_loss:.5f}")
train(qnn, train_dataloader, loss_fn, optimizer, epochs=20)
def test(
model: nn.Module,
data_loader: DataLoader,
atol=1e-4
) -> float:
num_correct = 0
total = 0
# Put the model in eval mode
model.eval()
# Turn on inference mode context manager
with torch.inference_mode():
for data, labels in data_loader:
# Send data to GPU
data, labels = data.to(device), labels.to(device)
# 1. Forward pass: Let the model predict
predictions = model(data)
# Get a tensor of booleans, indicating if each label is close to the real label
is_prediction_correct = torch.isclose(predictions, labels.type(torch.float32), atol=atol)
print("Label: ", labels)
print("predictions: ", predictions)
print("is_prediction_correct: ", is_prediction_correct)
# Count the amount of `True` predictions
num_correct += is_prediction_correct.sum().item()
# Count the total evaluations
# the first dimension of `labels` is `batch_size`
total += labels.size(0)
# Calculate the accuracy
accuracy = float(num_correct) / float(total)
print(f"Test Accuracy of the model: {accuracy*100:.2f}")
return accuracy
test(qnn, test_dataloader)
Currently, I am getting an in-accurate Train Loss as we haven't implemented the post-processing yet. Our next step will be to implement it as discussed in the meeting with Tal.
Week 12
This week I focused on the Post-processing of measurement counts and the transformation of labels into one-hot encoded labels. I first converted the measurement counts dictionary into an array to get logits. I then trimmed this array to a length of 10 and then normalized it to get prediction probabilities. We optionally, converted these prediction probabilities to get prediction labels in a one-hot encoded format (but this is of no use during the training process). Here is the implementation of Post Processing Step:
def post_process(result: SavedResult) -> torch.Tensor:
counts: dict = result.value.counts
# Calculate logits from counts
logits: float = torch.zeros(16)
for key, value in counts.items():
logits[int(key, 2)] = value
# Trim the logits from length 16 to length 10 since we have only 10 labels
trimmed_logits = logits[:10]
# Calculate prediction probabilities from logits by normalizing it
pred_probs = torch.nn.functional.normalize(trimmed_logits, dim=0)
# Convert the prediction probabilities into prediction labels
pred_labels = torch.argmax(pred_probs)
### WRITE COUNTS, OUTPUT LOGITS, PRED PROBS, PRED LABELS to a file
output_file = open("post_process_output.txt", "a")
print("----------------------------------------------------------------------------------------------------------------------------------------------", file=output_file)
print(f"COUNTS:: \n {counts} \n", file=output_file)
print(f"LOGITS:: \n {logits} \n", file=output_file)
print(f"TRIMMED LOGITS:: \n {trimmed_logits} \n", file=output_file)
print(f"PREDICTION PROBABILITIES:: \n {pred_probs} \n", file=output_file)
print(f"PREDICTION LABELS:: \n {pred_labels} \n", file=output_file)
output_file.close()
return torch.tensor(pred_probs)
I also updated the target transform function to convert labels from integer digits to their one-hot encoding format:
def target_transform(label):
label_tensor = torch.LongTensor([label])
one_hot_label = torch.nn.functional.one_hot(label_tensor, 10)
return one_hot_label
Besides these, I have also added functionality to save prediction outputs during the training and testing process into output test files for visualization and inferencing purposes. We will remove these when we are done experimenting.
Now, here are the results of two trial runs:
Learning Rate: 1.0
Batch Size: 32
Classes: 0-9
| Exp. No. | No. of Batches | Epochs | Train Loss | Train Time (mins) | Test Accuracy |
| 1. | 2 | 2 | 0.28655 | 2 | 9.375 |
Next, I am preparing a mini-dataset of only "0" and "1" to run quick parallel experiments to increase the accuracy of our network.
Week 13
This week I focused on creating custom mnist datasets of different sizes and classes, making our code modular, and adding more visualization features for easier experimentation and inferencing.
I have created the following two datasets:
Mini dataset of size 128 images containing only labels 0 and 1
Mini dataset of size 128 images containing all labels from 0 to 1
I wrote multiple script files to achieve modularity:
- scripts/data_setup.py
This file contains functionalities to load different datasets (custom datasets of different sizes and classes or default MNIST datasets provided by PyTorch).
import os
from typing import Callable,Optional
from torchvision import datasets
from torch.utils.data import DataLoader, Subset
NUM_WORKERS = os.cpu_count()
def create_dataloaders_from_folders(
train_dir: str,
test_dir: str,
batch_size: int,
transform: Optional[Callable] = None,
target_transform: Optional[Callable] = None,
num_workers: int = NUM_WORKERS
):
"""Creates training and testing DataLoaders.
Takes in a training directory and testing directory path and turns
them into PyTorch Datasets and then into PyTorch DataLoaders.
Args:
train_dir: Path to training directory.
test_dir: Path to testing directory.
transform: function having torchvision transforms to perform on training and testing data.
target_transform: function having torchvision transforms to perform on training and testing data labels.
batch_size: Number of samples per batch in each of the DataLoaders.
num_workers: An integer for number of workers per DataLoader.
Returns:
A tuple of (train_dataloader, test_dataloader, class_names).
Where class_names is a list of the target classes.
Example usage:
train_dataloader, test_dataloader, class_names = \
= create_dataloaders_from_folders(train_dir=path/to/train_dir,
test_dir=path/to/test_dir,
transform=some_data_transform_function,
target_transform=some_label_transform_function,
batch_size=32,
num_workers=4)
"""
# Use ImageFolder to create dataset(s)
train_data = datasets.ImageFolder(train_dir, transform=transform, target_transform=target_transform)
test_data = datasets.ImageFolder(test_dir, transform=transform, target_transform=target_transform)
# Get class names
class_names = train_data.classes
# Turn images into data loaders
train_dataloader = DataLoader(
train_data,
batch_size=batch_size,
shuffle=True,
num_workers=num_workers,
pin_memory=True,
)
test_dataloader = DataLoader(
test_data,
batch_size=batch_size,
shuffle=False,
num_workers=num_workers,
pin_memory=True,
)
return train_dataloader, test_dataloader, class_names
def create_mnist_dataloaders(
batch_size: int,
root: str = "data",
transform: Optional[Callable] = None,
target_transform: Optional[Callable] = None,
num_workers: int = NUM_WORKERS,
create_subset: bool = False,
subset_size:int = 64
):
"""Creates training and testing DataLoaders.
Creates PyTorch Dataloaders from PyTorch MNIST Dataset.
Args:
root: folder name in which data will be downloaded.
transform: torchvision transforms to perform on training and testing data.
target_transform: function having torchvision transforms to perform on training and testing data labels.
batch_size: Number of samples per batch in each of the DataLoaders.
num_workers: An integer for number of workers per DataLoader.
create_subset: If True, it create a dataloaders from small subset of data.
subset_size: Size of the subset of data. Defaults to 64.
Returns:
A tuple of (train_dataloader, test_dataloader, class_names).
Where class_names is a list of the target classes.
Example usage:
train_dataloader, test_dataloader, class_names = \
= create_mnist_dataloaders(root="data"
transform=some_data_transform_function,
target_transform=some_label_transform_function,
batch_size=32,
num_workers=4)
"""
# Setup training data
train_data = datasets.MNIST(
root=root,
train=True,
download=True,
transform=transform,
target_transform=target_transform
)
# Setup testing data
test_data = datasets.MNIST(
root=root,
train=False,
download=True,
transform=transform,
target_transform=target_transform
)
# Get class names
class_names = train_data.classes
# Create subsets of the datasets
if create_subset:
train_data = Subset(train_data, range(subset_size))
test_data = Subset(test_data, range(subset_size))
# Turn datasets into iterables (batches)
train_dataloader = DataLoader(train_data,
batch_size=batch_size,
shuffle=True,
num_workers=num_workers,
)
test_dataloader = DataLoader(test_data,
batch_size=batch_size,
shuffle=False,
num_workers=num_workers,
)
return train_dataloader, test_dataloader, class_names
- scripts/data_transforms.py
This file contains functionalities for transforming data images and labels as discussed earlier.
import torch
import torchvision.transforms as transforms
def input_transform(image):
"""
The input MNIST images are all 28 × 28 px. This function will firstly center-crop
them to 24 × 24 and then down-sample them to 4 × 4 for MNIST. Then we convert
the image pixels into angles for passing them into Rotation gates later for encoding.
"""
image = transforms.Grayscale(num_output_channels=1)(image)
image = transforms.ToTensor()(image)
image = transforms.CenterCrop(24)(image)
image = transforms.Resize(size = (4,4), antialias=True)(image)
image = image.squeeze()
image_pixels = torch.flatten(image)
angles = torch.sqrt(image_pixels / 256)
return angles
def target_transform(label):
label_tensor = torch.LongTensor([label])
one_hot_label = torch.nn.functional.one_hot(label_tensor, 10)
return one_hot_label.squeeze()
def target_transform_bin(label):
label_tensor = torch.LongTensor([label])
one_hot_label = torch.nn.functional.one_hot(label_tensor, 2)
return one_hot_label.squeeze()
- scripts/train.py
This file contains the script for training our model. I have updated how results are saved to write them into a CSV file in further steps.
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
import torch.optim as optim
from typing import Dict, List
from tqdm.auto import tqdm
def train(
model: nn.Module,
data_loader: DataLoader,
loss_fn: nn.modules.loss._Loss,
optimizer: optim.Optimizer,
writer: torch.utils.tensorboard.writer.SummaryWriter,
epochs: int = 20,
device: str = 'cpu',
) -> Dict[str, List]:
model.to(device)
# Setup train loss value
train_loss = 0
# Create empty results dictionary
results = {
"train_loss": [],
}
# Loop through training steps for a number of epochs
for epoch in tqdm(range(epochs)):
print(f"Epoch: {epoch}\n----------")
for batch, (data, label) in enumerate(data_loader):
# Send data to device (GPU or CPU)
data, label = data.to(device), label.to(device)
# 1. Forward pass
output = model(data).to(device)
# 2. Calculate loss
loss = loss_fn(output, label)
train_loss += loss
# 3. Optimizer zero grad
optimizer.zero_grad()
# 4. Loss backward
loss.backward()
# 5. Optimizer step
optimizer.step()
# Calculate loss per epoch and print out what's happening
train_loss /= len(data_loader)
# Print out what's happening
print(
f"Epoch: {epoch+1} | "
f"Train loss: {train_loss:.5f}"
)
# Update results dictionary
results["train_loss"].append(train_loss.detach().item())
### Experiment Tracking ###
# See if there's a writer, if so, log to it
if writer:
# Add loss results to SummaryWriter
writer.add_scalars(
main_tag="Loss",
tag_scalar_dict={"train_loss": train_loss,},
global_step=epoch
)
# Close the writer
writer.close()
# Return the filled results at the end of the epochs
return results
- scripts/test.py
This file contains scripts for testing our model.
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
def test(
model: nn.Module,
data_loader: DataLoader,
atol=0,
device: str = 'cpu',
) -> float:
num_correct = 0
total = 0
# Put the model in eval mode
model.eval()
# Turn on inference mode context manager
with torch.inference_mode():
for data, labels in data_loader:
# Send data to GPU
data, labels = data.to(device), labels.to(device)
# 1. Forward pass: Let the model predict
predictions = model(data)
# Get a tensor of booleans, indicating if each label is close to the real label
is_prediction_correct = torch.isclose(predictions.argmax(dim=1), labels.argmax(dim=1), atol=atol)
### WRITE OUTPUT TO A FILE
# output_file = open("test_loop_output.txt", "a")
# print("----------------------------------------------------------------------------------------------------------------------------------------------", file=output_file)
# print(f"LABELS:: \n {labels} \n", file=output_file)
# print(f"PREDICTIONS:: \n {predictions} \n", file=output_file)
# print(f"IS PREDICTIONS CORRECT:: \n {is_prediction_correct} \n", file=output_file)
# output_file.close()
# Count the amount of `True` predictions
num_correct += is_prediction_correct.sum().item()
# Count the total evaluations
# the first dimension of `labels` is `batch_size`
total += labels.size(0)
# Calculate the accuracy
accuracy = float(num_correct) / float(total)
print(f"Test Accuracy of the model: {accuracy * 100:.2f}%")
return accuracy * 100
- scripts/helper.py
This file contains functionality for saving our results and and track experiments.
import torch
from torch.utils.tensorboard import SummaryWriter
from datetime import datetime
import os
import pandas as pd
def create_writer(experiment_name: str,
model_name: str,
extra: str=None) -> torch.utils.tensorboard.writer.SummaryWriter():
"""Creates a torch.utils.tensorboard.writer.SummaryWriter() instance saving to a specific log_dir.
log_dir is a combination of runs/timestamp/experiment_name/model_name/extra.
Where timestamp is the current date in YYYY-MM-DD format.
Args:
experiment_name (str): Name of experiment.
model_name (str): Name of model.
extra (str, optional): Anything extra to add to the directory. Defaults to None.
Returns:
torch.utils.tensorboard.writer.SummaryWriter(): Instance of a writer saving to log_dir.
Example usage:
# Create a writer saving to "runs/2022-06-04/data_10_percent/leqm3/5_epochs/"
writer = create_writer(experiment_name="data_10_percent",
model_name="leqm3",
extra="5_epochs")
# The above is the same as:
writer = SummaryWriter(log_dir="runs/2022-06-04/data_10_percent/leqm3/5_epochs/")
"""
# Get timestamp of current date (all experiments on certain day live in same folder)
timestamp = datetime.now().strftime("%Y-%m-%d") # returns current date in YYYY-MM-DD format
if extra:
# Create log directory path
log_dir = os.path.join("runs", timestamp, experiment_name, model_name, extra)
else:
log_dir = os.path.join("runs", timestamp, experiment_name, model_name)
print(f"[INFO] Created SummaryWriter, saving to: {log_dir}...")
return SummaryWriter(log_dir=log_dir)
def write_train_results(
experiment_name,
model_name,
epochs,
results,
):
output_dir = "outputs/train_results/"
# Check if the directory exists, and create it if not
if not os.path.exists(output_dir):
os.makedirs(output_dir)
output_file_name = f"{experiment_name}_{model_name}_epochs_{epochs}.csv"
file_path = f"outputs/train_results/{output_file_name}"
data = {
'Epoch': list(range(1, len(results['train_loss']) + 1)),
'Train Loss': results['train_loss']
}
df = pd.DataFrame(data)
if os.path.exists(file_path):
# If it exists, append data
df.to_csv(file_path, mode='a', index=False, header=False)
else:
df.to_csv(file_path, mode='w', index=False, header=True)
- scripts/save_model.py
This file contains the script for saving our trained models.
"""
Contains various utility functions for PyTorch model training and saving.
"""
import torch
from pathlib import Path
def save_model(model: torch.nn.Module, target_dir: str, model_name: str):
"""Saves a PyTorch model to a target directory.
Args:
model: A target PyTorch model to save.
target_dir: A directory for saving the model to.
model_name: A filename for the saved model. Should include
either ".pth" or ".pt" as the file extension.
Example usage:
save_model(model=model_0,
target_dir="models",
model_name="05_going_modular_tingvgg_model.pth")
"""
# Create target directory
target_dir_path = Path(target_dir)
target_dir_path.mkdir(parents=True, exist_ok=True)
# Create model save path
assert model_name.endswith(".pth") or model_name.endswith(
".pt"
), "model_name should end with '.pt' or '.pth'"
model_save_path = target_dir_path / model_name
# Save the model state_dict()
print(f"[INFO] Saving model to: {model_save_path}")
torch.save(obj=model.state_dict(), f=model_save_path)
- models/leqm3.py
This file contains the script for creating a Linear Entanglement Quantum Model for MNIST Data Classification with three linear entanglement layers of RXX, RYY, and RZZ.
"""_summary_
Linear Entanglement Quantum Model for MNIST Data Classification with three linear entanglement layers of RXX, RYY, and RZZ.
"""
from classiq import create_model, QFunc, QArray, QBit, Output, allocate, RX, RY, RZ, RZZ, RXX, RYY, CZ
@QFunc
def encoding(q: QArray[QBit]) -> None:
"""
This function encodes the input data into the qubits. This input data is a 4x4 image pixel values
converted into angle for rotation gates (RX, RY, RZ, RX) in form of a 16x1 vector.
We encode 4 pixels per qubit.
Args:
q (QArray[QBit]): Array of four Qubits to encode the input data into.
"""
RX(theta="input_0", target=q[0]) # Pixel 0 on Qubit 0
RY(theta="input_1", target=q[0]) # Pixel 1 on Qubit 0
RZ(theta="input_2", target=q[0]) # Pixel 2 on Qubit 0
RX(theta="input_3", target=q[0]) # Pixel 3 on Qubit 0
RX(theta="input_4", target=q[1]) # Pixel 4 on Qubit 1
RY(theta="input_5", target=q[1]) # Pixel 5 on Qubit 1
RZ(theta="input_6", target=q[1]) # Pixel 6 on Qubit 1
RX(theta="input_7", target=q[1]) # Pixel 7 on Qubit 1
RX(theta="input_8", target=q[2]) # Pixel 8 on Qubit 2
RY(theta="input_9", target=q[2]) # Pixel 9 on Qubit 2
RZ(theta="input_10", target=q[2]) # Pixel 10 on Qubit 2
RX(theta="input_11", target=q[2]) # Pixel 11 on Qubit 2
RX(theta="input_12", target=q[3]) # Pixel 12 on Qubit 3
RY(theta="input_13", target=q[3]) # Pixel 13 on Qubit 3
RZ(theta="input_14", target=q[3]) # Pixel 14 on Qubit 3
RX(theta="input_15", target=q[3]) # Pixel 15 on Qubit 3
@QFunc
def mixing(q: QArray[QBit]) -> None:
"""
This function performs the mixing operation on the qubits.
This is done by applying a series of RZZ, RXX, RYY gates to form a
ring connection.
Args:
q (QArray[QBit]): Array of four Qubits to apply the mixing operation on.
"""
RZZ(theta="weight_0", target=q[0:2])
RZZ(theta="weight_1", target=q[1:3])
RZZ(theta="weight_2", target=q[2:4])
RXX(theta="weight_4", target=q[0:2])
RXX(theta="weight_5", target=q[1:3])
RXX(theta="weight_6", target=q[2:4])
RYY(theta="weight_8", target=q[0:2])
RYY(theta="weight_9", target=q[1:3])
RYY(theta="weight_10", target=q[2:4])
@QFunc
def cz_block(q: QArray[QBit]) -> None:
"""
This function applies CZ gates between each qubit.
Args:
q (QArray[QBit]): Array of four Qubits to apply the entanglement operation on.
"""
CZ(control=q[0], target=q[1])
CZ(control=q[1], target=q[2])
CZ(control=q[2], target=q[3])
@QFunc
def main(res: Output[QArray[QBit]]) -> None:
"""
This is the main function from which model will be created.
It calls the other functions to perform the encoding, mixing and entanglement.
Args:
res (Output[QArray[QBit]]): Output QArray of QBits from which the model will be created.
"""
allocate(4, res)
encoding(q=res)
mixing(q=res)
cz_block(q=res)
def linear_entanglement_r3_quantum_model():
model = create_model(main)
return model
- models/qnn.py
This file contains scripts for creating a quantum neural network based on the model passed to it.
import torch
from classiq.execution import execute_qnn
from classiq.synthesis import SerializedQuantumProgram
from classiq.applications.qnn import QLayer
from classiq.applications.qnn.types import (
MultipleArguments,
SavedResult,
ResultsCollection,
)
def execute_fn(quantum_program: SerializedQuantumProgram, arguments: MultipleArguments) -> ResultsCollection:
return execute_qnn(quantum_program, arguments)
def post_process_fn(result: SavedResult) -> torch.Tensor:
counts: dict = result.value.counts
# Calculate logits from counts
logits: float = torch.zeros(16)
for key, value in counts.items():
logits[int(key, 2)] = value
# Trim the logits from length 16 to length 10 since we have only 10 labels
trimmed_logits = logits[:10]
# Calculate prediction probabilities from logits by normalizing it
pred_probs = torch.nn.functional.normalize(trimmed_logits, dim=0)
# Convert the prediction probabilities into prediction labels
# pred_labels = torch.argmax(pred_probs)
### WRITE COUNTS, OUTPUT LOGITS, PRED PROBS, PRED LABELS to a file
# output_file = open("post_process_output.txt", "a")
# print("----------------------------------------------------------------------------------------------------------------------------------------------", file=output_file)
# print(f"COUNTS:: \n {counts} \n", file=output_file)
# print(f"LOGITS:: \n {logits} \n", file=output_file)
# print(f"TRIMMED LOGITS:: \n {trimmed_logits} \n", file=output_file)
# print(f"PREDICTION PROBABILITIES:: \n {pred_probs} \n", file=output_file)
# print(f"PREDICTION LABELS:: \n {pred_labels} \n", file=output_file)
# output_file.close()
return pred_probs.clone().detach()
def post_process_bin_fn(result: SavedResult) -> torch.Tensor:
counts: dict = result.value.counts
# Calculate logits from counts
logits: float = torch.zeros(16)
for key, value in counts.items():
logits[int(key, 2)] = value
# Trim the logits from length 16 to length 10 since we have only 10 labels
trimmed_logits = logits[:2]
# Calculate prediction probabilities from logits by normalizing it
pred_probs = torch.nn.functional.normalize(trimmed_logits, dim=0)
# Convert the prediction probabilities into prediction labels
# pred_labels = torch.argmax(pred_probs)
### WRITE COUNTS, OUTPUT LOGITS, PRED PROBS, PRED LABELS to a file
# output_file = open("post_process_output.txt", "a")
# print("----------------------------------------------------------------------------------------------------------------------------------------------", file=output_file)
# print(f"COUNTS:: \n {counts} \n", file=output_file)
# print(f"LOGITS:: \n {logits} \n", file=output_file)
# print(f"TRIMMED LOGITS:: \n {trimmed_logits} \n", file=output_file)
# print(f"PREDICTION PROBABILITIES:: \n {pred_probs} \n", file=output_file)
# print(f"PREDICTION LABELS:: \n {pred_labels} \n", file=output_file)
# output_file.close()
return pred_probs.clone().detach()
class QNN(torch.nn.Module):
def __init__(self, quantum_program, execute, post_process, *args, **kwargs) -> None:
super().__init__()
self.qlayer = QLayer(
quantum_program,
execute=execute,
post_process=post_process,
*args,
**kwargs
)
def forward(self, x: torch.Tensor) -> torch.Tensor:
x = self.qlayer(x)
return x
Now using these files we can run many different numbers of experiments as we want based on different datasets of custom size and class labels.
Scenario 1: Experimentation using MNIST Dataset provided by Pytorch
Here is an example of how to run an experiment for a small (subset of size 64) MNIST dataset provided by PyTorch for 2 epochs.
import torch
import classiq
import torch.nn as nn
import torch.optim as optim
from torchinfo import summary
from models.leqm3 import linear_entanglement_r3_quantum_model
from models.qnn import execute_fn, post_process_fn, QNN
from scripts.helper import create_writer, write_train_results
from scripts.data_setup import create_mnist_dataloaders
from scripts.data_transforms import input_transform, target_transform
from scripts.train import train
from scripts.test import test
## Authenticate Classiq
classiq.authenticate()
## For setting up device agnostic code
device = "cuda" if torch.cuda.is_available() else "cpu"
device = 'cpu'
print(device)
## Clear Output Files
post_process_output_file = open("post_process_output.txt", "w")
print("-----------------------------------------------------------------------------------------------------------------", file=post_process_output_file)
print("--------------------------------------------POST PROCESS OUTPUT--------------------------------------------------", file=post_process_output_file)
print("-----------------------------------------------------------------------------------------------------------------", file=post_process_output_file)
post_process_output_file.close()
test_loop_output_file = open("test_loop_output.txt", "w")
print("-----------------------------------------------------------------------------------------------------------------", file=test_loop_output_file)
print("-----------------------------------------------TEST LOOP OUTPUT--------------------------------------------------", file=test_loop_output_file)
print("-----------------------------------------------------------------------------------------------------------------", file=test_loop_output_file)
test_loop_output_file.close()
# HYPER PARAMETERS
_LEARNING_RATE = 1.0
BATCH_SIZE = 64
EPOCHS = 2
# Create a Linear Entanglement Quantum Model for MNIST Data Classification with three linear entanglement layers of RXX, RYY, and RZZ.
quantum_model = linear_entanglement_r3_quantum_model()
quantum_program = classiq.synthesize(quantum_model)
# View Quantum Program on Classiq Platform
classiq.show(quantum_program)
qnn = QNN(
quantum_program=quantum_program,
execute=execute_fn,
post_process=post_process_fn,
)
summary(model=qnn, input_size=(32, 16), verbose=0, col_names=["input_size", "output_size", "num_params", "trainable"], col_width=20, row_settings=["var_names"])
# choosing our loss function
loss_fn = nn.L1Loss()
# choosing our optimizer
optimizer = optim.SGD(qnn.parameters(), lr=_LEARNING_RATE)
train_dataloader, test_dataloader, class_names = create_mnist_dataloaders(
root="data",
transform=input_transform,
target_transform=target_transform,
batch_size=BATCH_SIZE,
create_subset=True,
subset_size=64
)
# Let's check out what we've created
print(f"Dataloaders: {train_dataloader, test_dataloader}")
print(f"Length of train dataloader: {len(train_dataloader)} batches of {BATCH_SIZE}")
print(f"Length of test dataloader: {len(test_dataloader)} batches of {BATCH_SIZE}")
print(f"Our Dataset have following classes: {class_names}")
data, label = next(iter(train_dataloader))
print(f"Image shape: {data.shape} -> [batch_size, pixel_angle]")
print(f"Label shape: {label.shape} -> [batch_size, label_value]")
# Create a writer for tracking our experiment
writer = create_writer(experiment_name="data_0.1_percent", model_name="linear_entanglement_r3", extra=f"{EPOCHS}_epochs")
train_results = train(
model = qnn,
data_loader = train_dataloader,
loss_fn = loss_fn,
optimizer = optimizer,
writer = writer,
epochs = EPOCHS,
device = device
)
# Check out the model results
print(train_results)
write_train_results(experiment_name="data_0.1_percent", model_name="linear_entanglement_r3", epochs=EPOCHS, results=train_results)
# %load_ext tensorboard
# %tensorboard --logdir runs
test_results = test(
model = qnn,
data_loader = test_dataloader,
device = device
)
print(test_results)
# Save the trained model
save_model(
model=qnn,
target_dir='outputs/saved_models',
model_name='leqmr3_subset64.pt'
)
Here is the summary of our quantum neural network provided by torchinfo.summary:
Scenario 2: Experimentation using custom binary MNIST Dataset
Here is an example of how to run an experiment for 1280 MNIST images of 0 and 1 provided by PyTorch for 10 epochs.
import torch
import classiq
import torch.nn as nn
import torch.optim as optim
from torchinfo import summary
from pathlib import Path
from models.leqm3 import linear_entanglement_r3_quantum_model
from models.qnn import execute_fn, post_process_fn, QNN
from scripts.helper import create_writer, write_train_results
from scripts.data_setup import create_mnist_dataloaders
from scripts.data_transforms import input_transform, target_transform
from scripts.train import train
from scripts.test import test
## Authenticate Classiq
classiq.authenticate()
## For setting up device agnostic code
device = "cuda" if torch.cuda.is_available() else "cpu"
device = 'cpu'
print(device)
## HYPER PARAMETERS
_LEARNING_RATE = 1.0
BATCH_SIZE = 32
EPOCHS = 10
## Create a Linear Entanglement Quantum Model for MNIST Data Classification with three linear entanglement layers of RXX, RYY, and RZZ.
quantum_model = linear_entanglement_r3_quantum_model()
quantum_program = classiq.synthesize(quantum_model)
# View Quantum Program on Classiq Platform
classiq.show(quantum_program)
qnn = QNN(
quantum_program=quantum_program,
execute=execute_fn,
post_process=post_process_fn,
)
summary(model=qnn, input_size=(32, 16), verbose=0, col_names=["input_size", "output_size", "num_params", "trainable"], col_width=20, row_settings=["var_names"])
# choosing our loss function
loss_fn = nn.L1Loss()
# choosing our optimizer
optimizer = optim.SGD(qnn.parameters(), lr=_LEARNING_RATE)
train_dir = Path('mini_data_1280_bin/mini_data_1280_bin/train')
test_dir = Path('mini_data_1280_bin/mini_data_1280_bin/test')
train_dataloader, test_dataloader, class_names = create_dataloaders_from_folders(
train_dir=train_dir,
test_dir=test_dir,
transform=input_transform,
target_transform=target_transform_bin,
batch_size=BATCH_SIZE,
)
# Let's check out what we've created
print(f"Dataloaders: {train_dataloader, test_dataloader}")
print(f"Length of train dataloader: {len(train_dataloader)} batches of {BATCH_SIZE}")
print(f"Length of test dataloader: {len(test_dataloader)} batches of {BATCH_SIZE}")
print(f"Our Dataset have following classes: {class_names}")
data, label = next(iter(train_dataloader))
print(f"Image shape: {data.shape} -> [batch_size, pixel_angle]")
print(f"Label shape: {label.shape} -> [batch_size, label_value]")
# Create a writer for tracking our experiment
writer = create_writer(experiment_name="custom_data_1280", model_name="linear_entanglement_r3", extra=f"{EPOCHS}_epochs")
train_results = train(
model = qnn,
data_loader = train_dataloader,
loss_fn = loss_fn,
optimizer = optimizer,
writer = writer,
epochs = EPOCHS,
device = device
)
# Check out the model results
print(train_results)
write_train_results(experiment_name="data_1280_bin", model_name="linear_entanglement_r3", epochs=EPOCHS, results=train_results)
# %load_ext tensorboard
# %tensorboard --logdir runs
save_model(
model=qnn,
target_dir='outputs/saved_models',
model_name=f'exp_1_leqmr3_data_1280_bin_epoch{EPOCHS}.pt'
)
test_results = test(
model = qnn,
data_loader = test_dataloader,
device = device
)
print(test_results)
When using the custom binary dataset, here is the summary of our QNN.
In the upcoming weeks, I will be running different experiments, writing the final manuscript, and preparing a presentation for the final demo day.
Week 14
Experimentation
Learning Rate: 1.0
Batch Size: 32
| Exp. No. | No. of Batches | Epochs | Classes | Train Loss | Train Time (mins) | Test Accuracy |
| 1. | 2 | 2 | 0-9 | 0.28655 | 2 | 9.375% |
| 2. | 8 | 2 | 0-9 | 0.20318 | 3 | 7.422% |
| 3. | 8 | 10 | 0-9 | 0.19270 | 35 | 8.203% |
| 4. | 80 | 10 | 0-1 | 0.10146 | 165 | 50% |
