Esegui job quantistici da un workflow

Versioni dei pacchetti

Il codice in questa pagina è stato sviluppato utilizzando i seguenti requisiti. Ti consigliamo di utilizzare queste versioni o successive.

qiskit[all]~=2.6.1
qiskit-ibm-runtime~=0.51.0
qiskit-serverless~=0.38.0

Un workflow quantistico-classico viene eseguito come un unico job di workflow gestito. Le sue fasi classiche vengono eseguite sul tuo profilo di calcolo, mentre i job primitive che invia a una QPU di IBM Quantum® vengono eseguiti sulla tua istanza e contano ai fini del tuo utilizzo. Da quel singolo job puoi segnalare ogni fase, trovare i job QPU e annullarli. Per avviare un job e controllarne stato e log, consulta Esegui un workflow gestito.

Prima di iniziare, distribuisci il tuo primo workflow gestito.

Crea il servizio Runtime

Crea il servizio Runtime nel codice del tuo workflow con get_runtime_service() e usalo allo stesso modo in cui usi QiskitRuntimeService in locale. Il workflow seguente esegue un circuito di Bell con la primitive Sampler, segnala ogni fase con update_status() e restituisce i conteggi:

# Added by doQumentation — installs the packages this notebook needs if they are missing
import importlib.util

_needed = {"qiskit": "qiskit", "qiskit_ibm_catalog": "qiskit-ibm-catalog", "qiskit_ibm_runtime": "qiskit-ibm-runtime", "qiskit_serverless": "qiskit-serverless"}
_missing = [pip for module, pip in _needed.items()
            if importlib.util.find_spec(module) is None]
# One at a time, so a package that fails to install does not block the others
for _pip in _missing:
    %pip install -q {_pip}
if not _missing:
    print("\u2713 All required packages are installed")

# This cell is hidden from users. It creates the working directory.
from pathlib import Path

Path("./source_files").mkdir(exist_ok=True)

%%writefile ./source_files/my_workflow.py
from qiskit import QuantumCircuit
from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime.executor_sampler import Sampler
from qiskit_serverless import (
    Job,
    get_arguments,
    get_runtime_service,
    save_result,
    update_status,
)

arguments = get_arguments()

service = get_runtime_service()
backend = service.backend(arguments["backend_name"])

circuit = QuantumCircuit(2)
circuit.h(0)
circuit.cx(0, 1)
circuit.measure_all()

update_status(Job.OPTIMIZING_HARDWARE)
pass_manager = generate_preset_pass_manager(optimization_level=1, backend=backend)
isa_circuit = pass_manager.run(circuit)

update_status(Job.WAITING_QPU)
sampler = Sampler(mode=backend)
primitive_job = sampler.run([isa_circuit])
counts = primitive_job.result()[0].data.meas.get_counts()

update_status(Job.POST_PROCESSING)
save_result({"backend": backend.name, "counts": counts})

Overwriting ./source_files/my_workflow.py

get_runtime_service() restituisce un QiskitRuntimeService configurato per il tuo job di workflow gestito e fornisce quanto segue:

• Credenziali. Il servizio Runtime usa la chiave API, l'istanza e il canale che hai usato per chiamare run(), quindi il codice del workflow non richiede credenziali.

• Tracciamento. Un workflow registra ogni job primitive eseguito tramite questo servizio Runtime sotto il job di workflow gestito, inclusi i job Sampler, Estimator ed Executor e i job in una sessione o in un batch. Puoi elencarli con job.runtime_jobs().

• Annullamento. Quando chiami job.cancel(), il workflow gestito annulla anche le sessioni, i batch e i job primitive del workflow ancora in esecuzione.

Un workflow gestito non traccia né annulla i job di un servizio creato con QiskitRuntimeService(), anche se i job vengono comunque eseguiti. Se il workflow deve usare un altro servizio, annulla i suoi job in un gestore SIGTERM.

Esegui le primitive in una sessione o in un batch

Un workflow gestito traccia anche i job primitive eseguiti in una sessione o in un batch. Crea la sessione o il batch da un backend restituito dal servizio Runtime e crea le primitive all'interno del blocco with. Session funziona allo stesso modo:

source_files/my_workflow.py

from qiskit_ibm_runtime import Batch

with Batch(backend=backend) as batch:
    sampler = Sampler(mode=batch)
    primitive_jobs = [sampler.run([isa_circuit]) for _ in range(2)]
    results = [primitive_job.result() for primitive_job in primitive_jobs]

Segnala l'avanzamento con update_status()

Il workflow in Crea il servizio Runtime chiama update_status() all'inizio di ogni fase, in modo che lo stato del job mostri a che punto si trova il workflow, ad esempio RUNNING: WAITING_FOR_QPU. Passa una delle seguenti costanti Job:

Chiamata nel workflow	Stato del job
update_status(Job.MAPPING)	RUNNING: MAPPING
update_status(Job.OPTIMIZING_HARDWARE)	RUNNING: OPTIMIZING_FOR_HARDWARE
update_status(Job.WAITING_QPU)	RUNNING: WAITING_FOR_QPU
update_status(Job.EXECUTING_QPU)	RUNNING: EXECUTING_QPU
update_status(Job.POST_PROCESSING)	RUNNING: POST_PROCESSING

update_status() restituisce True se il workflow ha cambiato lo stato. Restituisce False e scrive un avviso nei log se il valore non è una di queste costanti, se il job non è più attivo o se il codice viene eseguito fuori da un job. Non solleva un errore, quindi un refuso non interrompe il workflow.

Per l'avanzamento all'interno di una fase, come le iterazioni di un ottimizzatore, scrivi righe di log.

Trova i job QPU di un'esecuzione

Nel tuo notebook locale, scegli una QPU a cui la tua istanza può accedere. client.backends() le elenca e client.least_busy() restituisce quella con il minor numero di job in sospeso:

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
print([backend.name for backend in client.backends(simulator=False)])

backend_name = client.least_busy(operational=True, simulator=False).name

['ibm_boston', 'ibm_fez', 'ibm_kingston', 'ibm_marrakesh', 'ibm_phoenix', 'ibm_pittsburgh']

Carica il workflow e passa il nome del backend quando lo esegui:

from qiskit_ibm_catalog import ManagedWorkflow

my_workflow = client.upload(
    ManagedWorkflow(
        title="my-workflow",
        entrypoint="my_workflow.py",
        working_dir="./source_files/",
    )
)

job = my_workflow.run(backend_name=backend_name)

# This cell is hidden from users. It waits for the job to finish.
assert job.wait_for_final_state() == "DONE", job.logs()

Dopo la conclusione del job, ottieni dal job di workflow gestito gli ID dei job QPU del workflow:

runtime_job_ids = job.runtime_jobs()
runtime_job_ids

['dbl6lst8ft8s738secf0']

# This cell is hidden from users. It checks the tracked job.
assert len(runtime_job_ids) == 1, runtime_job_ids

Ogni ID è un normale ID di job primitive. Per leggere lo stato e i risultati di un job, consulta Recupera i risultati di un job in un secondo momento. Se il workflow ha eseguito primitive in una sessione o in un batch, elenca gli ID di sessione e batch con runtime_sessions() e passane uno a runtime_jobs() per ottenere solo i job al suo interno:

run_workflow_remotely.ipynb

for session_id in job.runtime_sessions():
    print(session_id, job.runtime_jobs(session_id))

dc101077-d085-40fb-8fbe-78852e355994 ['dbjqp48nceqs73aki1k0', 'dbjqp4l6p51s73clb91g']

L'oggetto job esiste solo nel notebook che ha avviato l'esecuzione. In un nuovo notebook, ottieni di nuovo il job con client.job("<your-job-id>"), quindi chiama runtime_jobs() su di esso. Consulta Trova un job da una nuova sessione.

Per l'utilizzo della QPU e i tempi di esecuzione di ogni job QPU, consulta Visualizza le metriche del job. Per l'utilizzo dell'intera istanza nel periodo corrente, chiama client.usage(). Per vedere quanto è durata ogni fase del job di workflow gestito, consulta Controlla quanto è durata ogni fase.

Annulla i job QPU collegati

Quando annulli un job di workflow gestito con job.cancel(), il workflow prova anche ad annullare il lavoro QPU che ha avviato:

• Sessioni e batch. Il workflow annulla ogni sessione e ogni batch del workflow, il che annulla anche i loro job in coda.

• Altri job primitive. Il workflow annulla ogni altro job primitive del workflow ancora in coda o in esecuzione.

job.cancel() restituisce un messaggio per ogni job QPU e sessione, così puoi verificare cosa è stato annullato:

run_workflow_remotely.ipynb

print(job.cancel())

Per un workflow eseguito in un batch, il messaggio è simile al seguente:

Job is stopping. Canceled runtime session: eb8874c9-a15e-45f8-9a90-a9fb35267f9d and associated runtime jobs.

Se il client non riesce a creare un QiskitRuntimeService dalle tue credenziali locali, annulla solo il job di workflow gestito. Annulla tu stesso i job QPU con i loro ID da job.runtime_jobs().

Quando un workflow interrompe un job da sé, ad esempio perché ha raggiunto il limite di 24 ore, non annulla i suoi job QPU. Annullali con i loro ID da job.runtime_jobs().

Risoluzione dei problemi dei job QPU in un workflow

Problema	Causa
get_runtime_service() solleva KeyError: 'QISKIT_IBM_CHANNEL'.	Il codice viene eseguito nel tuo notebook locale, non in un job di workflow gestito. Usa QiskitRuntimeService() in locale.
update_status() restituisce False e i log mostrano 'sub_status' not provided or is not valid.	Il valore non è una delle costanti Job, oppure il job non è più attivo.
job.runtime_jobs() restituisce [].	Il workflow ha creato il suo servizio con QiskitRuntimeService(), oppure non ha ancora eseguito una primitive.
I job di una sessione non vengono tracciati.	La sessione è stata creata da un backend di un servizio diverso. Ottieni il backend da get_runtime_service().
I log del workflow mostrano Could not associate runtime job with serverless job.	Il workflow non è riuscito a registrare un job primitive. Il job primitive viene comunque eseguito, ma non viene tracciato.
job.cancel() mostra No QiskitRuntimeService can be associated to the given token and instance.	Il client non è riuscito a creare un QiskitRuntimeService dalle tue credenziali locali, quindi ha annullato solo il job di workflow gestito.
client.job() fallisce con Job [...] not found.	L'ID del job è errato, oppure il job appartiene a un altro utente.

Passi successivi

Raccomandazioni

• Esegui un workflow gestito per controllare stato, risultato e log di un job e trovare la causa di un errore.