Execute jobs quânticos a partir de um workflow

Versões dos pacotes

O código desta página foi desenvolvido usando os seguintes requisitos. Recomendamos usar estas versões ou mais recentes.

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

Um workflow quântico-clássico é executado como um único job de workflow gerenciado. Suas etapas clássicas são executadas no seu perfil de computação, e os jobs de primitiva que ele envia a uma QPU da IBM Quantum® são executados na sua instância e contam para o seu uso. A partir desse único job, você pode relatar cada etapa, encontrar os jobs de QPU e cancelá-los. Para iniciar um job e verificar seu status e logs, consulte Execute um workflow gerenciado.

Antes de começar, implante seu primeiro workflow gerenciado.

Crie o serviço Runtime

Crie o serviço Runtime no código do seu workflow com get_runtime_service() e use-o da mesma forma que você usa QiskitRuntimeService localmente. O workflow a seguir executa um circuito de Bell com a primitiva Sampler, relata cada etapa com update_status() e retorna as contagens:

# 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() retorna um QiskitRuntimeService configurado para o seu job de workflow gerenciado e oferece o seguinte:

• Credenciais. O serviço Runtime usa a chave de API, a instância e o canal que você usou para chamar run(), então o código do workflow não precisa de credenciais.

• Rastreamento. Um workflow registra todo job de primitiva executado por meio deste serviço Runtime sob o job de workflow gerenciado, incluindo jobs de Sampler, Estimator e Executor, e jobs em uma sessão ou em lote. Você pode listá-los com job.runtime_jobs().

• Cancelamento. Quando você chama job.cancel(), o workflow gerenciado também cancela as sessões, os lotes e os jobs de primitiva do workflow que ainda estão em execução.

Um workflow gerenciado não rastreia nem cancela os jobs de um serviço que você cria com QiskitRuntimeService(), embora os jobs ainda sejam executados. Se o workflow precisar usar outro serviço, cancele seus jobs em um handler de SIGTERM.

Execute primitivas em uma sessão ou em lote

Um workflow gerenciado também rastreia jobs de primitiva executados em uma sessão ou em lote. Crie a sessão ou o lote a partir de um backend retornado pelo serviço Runtime e crie as primitivas dentro do bloco with. Session funciona da mesma forma:

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]

Relate o progresso com update_status()

O workflow em Crie o serviço Runtime chama update_status() no início de cada etapa, para que o status do job mostre em que ponto o workflow está, como RUNNING: WAITING_FOR_QPU. Passe uma das seguintes constantes de Job:

Chamada no workflow	Status do 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() retorna True se o workflow alterou o status. Retorna False e grava um aviso nos logs se o valor não for uma dessas constantes, se o job não estiver mais ativo ou se o código for executado fora de um job. Ele não gera um erro, então um erro de digitação não interrompe o workflow.

Para o progresso dentro de uma etapa, como as iterações de um otimizador, grave linhas de log.

Encontre os jobs de QPU de uma execução

No seu notebook local, escolha uma QPU que sua instância possa acessar. client.backends() lista as disponíveis e client.least_busy() retorna a que tem menos jobs pendentes:

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']

Envie o workflow e passe o nome do backend ao executá-lo:

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

Depois que o job termina, obtenha os IDs dos jobs de QPU do workflow a partir do job de workflow gerenciado:

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

Cada ID é um ID de job de primitiva comum. Para ler o status e os resultados de um job, consulte Recupere os resultados de um job mais tarde. Se o workflow executou primitivas em uma sessão ou em lote, liste os IDs de sessão e de lote com runtime_sessions() e passe um deles a runtime_jobs() para obter apenas os jobs nele:

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']

O objeto job existe apenas no notebook que iniciou a execução. Em um novo notebook, obtenha o job novamente com client.job("<your-job-id>") e depois chame runtime_jobs() nele. Consulte Encontre um job a partir de uma nova sessão.

Para o uso de QPU e os tempos de execução de cada job de QPU, consulte Veja as métricas do job. Para o uso de toda a sua instância no período atual, chame client.usage(). Para ver quanto tempo cada etapa do job de workflow gerenciado levou, consulte Verifique quanto tempo cada etapa levou.

Cancele os jobs de QPU vinculados

Quando você cancela um job de workflow gerenciado com job.cancel(), o workflow também tenta cancelar o trabalho de QPU que o workflow iniciou:

• Sessões e lotes. O workflow cancela cada uma das sessões e cada um dos lotes do workflow, o que também cancela seus jobs na fila.

• Outros jobs de primitiva. O workflow cancela cada um dos outros jobs de primitiva do workflow que ainda estão na fila ou em execução.

job.cancel() retorna uma mensagem para cada job de QPU e sessão, para que você possa conferir o que foi cancelado:

run_workflow_remotely.ipynb

print(job.cancel())

Para um workflow que foi executado em um lote, a mensagem é parecida com a seguinte:

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

Se o cliente não consegue criar um QiskitRuntimeService a partir das suas credenciais locais, ele cancela apenas o job de workflow gerenciado. Cancele você mesmo os jobs de QPU usando os IDs obtidos de job.runtime_jobs().

Quando um workflow interrompe um job por conta própria, por exemplo porque atingiu o limite de 24 horas, ele não cancela seus jobs de QPU. Cancele-os usando os IDs obtidos de job.runtime_jobs().

Solucione problemas de jobs de QPU em um workflow

Problema	Causa
get_runtime_service() gera KeyError: 'QISKIT_IBM_CHANNEL'.	O código é executado no seu notebook local, e não em um job de workflow gerenciado. Use QiskitRuntimeService() localmente.
update_status() retorna False, e os logs mostram 'sub_status' not provided or is not valid.	O valor não é uma das constantes de Job, ou o job não está mais ativo.
job.runtime_jobs() retorna [].	O workflow criou seu serviço com QiskitRuntimeService(), ou ainda não executou uma primitiva.
Os jobs de uma sessão não são rastreados.	A sessão foi criada a partir de um backend de outro serviço. Obtenha o backend com get_runtime_service().
Os logs do workflow mostram Could not associate runtime job with serverless job.	O workflow não conseguiu registrar um job de primitiva. O job de primitiva ainda é executado, mas não é rastreado.
job.cancel() mostra No QiskitRuntimeService can be associated to the given token and instance.	O cliente não conseguiu criar um QiskitRuntimeService a partir das suas credenciais locais, então cancelou apenas o job de workflow gerenciado.
client.job() falha com Job [...] not found.	O ID do job está errado, ou o job pertence a outro usuário.

Próximos passos

Recomendações

• Execute um workflow gerenciado para verificar o status, o resultado e os logs de um job e encontrar a causa de um erro.