Solving a QUBO problem flexibly¶

In addition to the Object API (Solver), qubo-solver also exposes a functional API. It is slightly more verbose, but gives better insight and control: some things are only possible with the functional API.

Classical approach¶

We'll start with the classical approach, since the quantum one has some specificities. Rather than going through Solver, we call the solving algorithm directly — here, tabu search — on a batch of starting bitstrings. See the documentation for other classical solvers.

In [ ]:

from __future__ import annotations

from qubosolver import (
    Instance,
    analysis,
    bitstrings,
    matrix,
    solving,
    torch_rng,
)

Q = matrix.tensor(
    [
        [-0.2, 0.0, 1.0],
        [0.0, 0.0, 1.5],
        [1.0, 1.5, 0.0],
    ]
)
instance = Instance(Q)

# Run tabu search in parallel from 5 starting bitstrings
starts = bitstrings.rand(5, instance.size, rng=torch_rng(15))
solution = solving.tabu_search.solve(instance, starts=starts, time_limit=10.0)

print(analysis.to_dataframe([solution]))

from __future__ import annotations from qubosolver import ( Instance, analysis, bitstrings, matrix, solving, torch_rng, ) Q = matrix.tensor( [ [-0.2, 0.0, 1.0], [0.0, 0.0, 1.5], [1.0, 1.5, 0.0], ] ) instance = Instance(Q) # Run tabu search in parallel from 5 starting bitstrings starts = bitstrings.rand(5, instance.size, rng=torch_rng(15)) solution = solving.tabu_search.solve(instance, starts=starts, time_limit=10.0) print(analysis.to_dataframe([solution]))

  labels bitstrings  costs  counts  probs
0      0        100   -0.2       1    0.2
1      0        110   -0.2       4    0.8

Quantum approach¶

In [ ]:

import qoolqit

from qubosolver import (
    Instance,
    LocalEmulator,
    Solution,
    analysis,
    drive_shaping,
    embedding,
    matrix,
    solving,
)

Q = matrix.tensor(
    [
        [-0.2, 0.0, 1.0],
        [0.0, 0.0, 1.5],
        [1.0, 1.5, 0.0],
    ]
)
instance = Instance(Q)

device = qoolqit.AnalogDevice()
backend = LocalEmulator()

register = embedding.blade.embed_for_device(instance, device)
drive = drive_shaping.proportional_diagonal.build_drive(instance, register, device=device)
program = solving.analog_quantum_sampling.compile(register, drive, device)
job = backend.run(program)
solution = Solution.from_results(job.results(), instance)

print(analysis.to_dataframe([solution]))

import qoolqit from qubosolver import ( Instance, LocalEmulator, Solution, analysis, drive_shaping, embedding, matrix, solving, ) Q = matrix.tensor( [ [-0.2, 0.0, 1.0], [0.0, 0.0, 1.5], [1.0, 1.5, 0.0], ] ) instance = Instance(Q) device = qoolqit.AnalogDevice() backend = LocalEmulator() register = embedding.blade.embed_for_device(instance, device) drive = drive_shaping.proportional_diagonal.build_drive(instance, register, device=device) program = solving.analog_quantum_sampling.compile(register, drive, device) job = backend.run(program) solution = Solution.from_results(job.results(), instance) print(analysis.to_dataframe([solution]))

  labels bitstrings  costs  counts  probs
0      0        110   -0.2     442  0.442
1      0        100   -0.2      69  0.069
2      0        001    0.0     278  0.278
3      0        010    0.0      60  0.060
4      0        000    0.0     151  0.151

In [ ]:

import qoolqit
from pasqal_cloud import PasqalCloudConnection
from qoolqit.execution import QPU

from qubosolver import (
    Instance,
    RemoteEmulator,
    Solution,
    analysis,
    drive_shaping,
    embedding,
    matrix,
    solving,
)

# Replace with your username, project id and password on the Pasqal Cloud.
USERNAME = "#TO_PROVIDE"
PROJECT_ID = "#TO_PROVIDE"
PASSWORD = None

Q = matrix.tensor(
    [
        [-0.2, 0.0, 1.0],
        [0.0, 0.0, 1.5],
        [1.0, 1.5, 0.0],
    ]
)
instance = Instance(Q)

if PASSWORD is not None:
    # Setup connection
    connection = PasqalCloudConnection(
        username=USERNAME,
        password=PASSWORD,
        project_id=PROJECT_ID,
    )
else:
    # Use a mock local connection for tutorial
    from qubosolver.utils._local_connection import LocalConnection

    connection = LocalConnection()

emulate = True
if emulate:
    device = qoolqit.AnalogDevice()
    backend = RemoteEmulator(connection=connection)
else:
    print(f"Available devices: {connection.fetch_available_devices()}")
    device = qoolqit.Device.from_connection(connection, "FRESNEL_CAN1")
    backend = QPU(connection=connection, num_shots=1000)

register = embedding.blade.embed_for_device(instance, device)
drive = drive_shaping.proportional_diagonal.build_drive(instance, register, device=device)
program = solving.analog_quantum_sampling.compile(register, drive, device)
job = backend.run(program)
solution = Solution.from_results(job.results(), instance)

print(analysis.to_dataframe([solution]))

import qoolqit from pasqal_cloud import PasqalCloudConnection from qoolqit.execution import QPU from qubosolver import ( Instance, RemoteEmulator, Solution, analysis, drive_shaping, embedding, matrix, solving, ) # Replace with your username, project id and password on the Pasqal Cloud. USERNAME = "#TO_PROVIDE" PROJECT_ID = "#TO_PROVIDE" PASSWORD = None Q = matrix.tensor( [ [-0.2, 0.0, 1.0], [0.0, 0.0, 1.5], [1.0, 1.5, 0.0], ] ) instance = Instance(Q) if PASSWORD is not None: # Setup connection connection = PasqalCloudConnection( username=USERNAME, password=PASSWORD, project_id=PROJECT_ID, ) else: # Use a mock local connection for tutorial from qubosolver.utils._local_connection import LocalConnection connection = LocalConnection() emulate = True if emulate: device = qoolqit.AnalogDevice() backend = RemoteEmulator(connection=connection) else: print(f"Available devices: {connection.fetch_available_devices()}") device = qoolqit.Device.from_connection(connection, "FRESNEL_CAN1") backend = QPU(connection=connection, num_shots=1000) register = embedding.blade.embed_for_device(instance, device) drive = drive_shaping.proportional_diagonal.build_drive(instance, register, device=device) program = solving.analog_quantum_sampling.compile(register, drive, device) job = backend.run(program) solution = Solution.from_results(job.results(), instance) print(analysis.to_dataframe([solution]))

  labels bitstrings  costs  counts  probs
0      0        110   -0.2     454  0.454
1      0        100   -0.2      65  0.065
2      0        000    0.0     150  0.150
3      0        001    0.0     270  0.270
4      0        010    0.0      61  0.061

Saving and retrieving a remote job¶

Remote runs, whether on a QPU or a remote emulator, are submitted asynchronously: backend.run(program) returns as soon as the job is queued, without waiting for results. This lets you save the job's identifiers and the instance, disconnect, and retrieve the results later — from the same session or a different one — rather than blocking until the run completes.

In [ ]:

import json
import pathlib

from qoolqit.execution.job import JobStatus, get_batch_id, retrieve_remote_job

metadata = {
    "job_id": job.job_id(),
    "batch_id": get_batch_id(job),
}

output_directory = pathlib.Path.cwd() / "tmp" / "qubosolver-in-full"
output_directory.mkdir(parents=True, exist_ok=True)
metadata_file = output_directory / "metadata.json"
data_file = output_directory / "data.bin"

with metadata_file.open("w") as f:
    json.dump(metadata, f)
with data_file.open("wb") as f:
    instance.save(f)

import json import pathlib from qoolqit.execution.job import JobStatus, get_batch_id, retrieve_remote_job metadata = { "job_id": job.job_id(), "batch_id": get_batch_id(job), } output_directory = pathlib.Path.cwd() / "tmp" / "qubosolver-in-full" output_directory.mkdir(parents=True, exist_ok=True) metadata_file = output_directory / "metadata.json" data_file = output_directory / "data.bin" with metadata_file.open("w") as f: json.dump(metadata, f) with data_file.open("wb") as f: instance.save(f)

In [ ]:

with metadata_file.open("r") as f:
    metadata = json.load(f)
with data_file.open("rb") as f:
    reloaded_instance = Instance.load(f)

reloaded_job = retrieve_remote_job(connection, metadata["job_id"], batch_id=metadata["batch_id"])

status = reloaded_job.get_status()
print(f"Job status: {status}")

if status == JobStatus.DONE:
    solution = Solution.from_results(reloaded_job.results(), reloaded_instance)
    print(analysis.to_dataframe([solution]))

with metadata_file.open("r") as f: metadata = json.load(f) with data_file.open("rb") as f: reloaded_instance = Instance.load(f) reloaded_job = retrieve_remote_job(connection, metadata["job_id"], batch_id=metadata["batch_id"]) status = reloaded_job.get_status() print(f"Job status: {status}") if status == JobStatus.DONE: solution = Solution.from_results(reloaded_job.results(), reloaded_instance) print(analysis.to_dataframe([solution]))

Job status: JobStatus.DONE
  labels bitstrings  costs  counts  probs
0      0        110   -0.2     454  0.454
1      0        100   -0.2      65  0.065
2      0        000    0.0     150  0.150
3      0        001    0.0     270  0.270
4      0        010    0.0      61  0.061

The functional API mirrors each step the Solver performs internally — embedding, drive shaping, compiling, running, and interpreting results — as a separate call you can inspect, swap out, or run independently. See the other tutorials for a closer look at each of these steps.