Execute um workflow gerenciado

Versões dos pacotes

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

qiskit-serverless~=0.38.0

Um job de workflow gerenciado funciona como um job em lote que você envia a um agendador de HPC. run() envia o job e retorna um objeto Job imediatamente. Use o objeto Job para as tarefas que squeue, o arquivo de saída do job e scancel cobrem em um cluster. O job continua em execução depois que você fecha seu notebook.

Os exemplos desta página usam um pequeno workflow que lê um ângulo, theta, e falha se o ângulo estiver fora do intervalo. Faça o upload dele conforme descrito em Implante seu primeiro workflow gerenciado.

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

_needed = {"qiskit_ibm_catalog": "qiskit-ibm-catalog", "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
import math

from qiskit_serverless import get_arguments, save_result

arguments = get_arguments()
theta = arguments["theta"]
if not 0 <= theta <= math.pi:
    raise ValueError(f"theta must be between 0 and pi, got {theta}")

print(f"Computing the expectation value for theta={theta}", flush=True)
save_result({"theta": theta, "expectation_value": math.cos(theta)})

Writing ./source_files/my_workflow.py

Inicie um job

Obtenha o workflow pelo título e chame run() com os argumentos que o workflow lê:

# This cell is hidden from users. It uploads the workflow under a title that
# other test notebooks do not use, and makes load("my-workflow") return it.
# It also silences the account warning that job.cancel() logs.
import logging

from qiskit_ibm_catalog import ManagedWorkflow, ManagedWorkflowClient

logging.getLogger("qiskit_ibm_runtime").setLevel(logging.ERROR)

_test_title = "docs-run-managed-workflow"
ManagedWorkflowClient().upload(
    ManagedWorkflow(
        title=_test_title,
        entrypoint="my_workflow.py",
        working_dir="./source_files/",
    )
)
_load = ManagedWorkflowClient.load
ManagedWorkflowClient.load = lambda self, title, *a, **k: _load(
    self, _test_title if title == "my-workflow" else title, *a, **k
)

run_workflow_remotely.ipynb

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
my_workflow = client.load("my-workflow")

job = my_workflow.run(theta=0.5)
job_id = job.job_id

run() retorna depois que o workflow coloca o job na fila. Salve o ID do job para poder encontrar o job a partir de uma nova sessão.

• Para escolher quais argumentos um workflow aceita e verificá-los antes de o job começar, consulte Defina entradas e saídas.

• Para executar o job com um tamanho de computação diferente, passe function_size. Consulte Escolha um tamanho para cada execução.

Acompanhe um job

Verifique o status

Logo depois de chamar run(), o status geralmente é QUEUED: run_workflow_remotely.ipynb

job.status()

'QUEUED'

Status	Significado
QUEUED	O job aguarda um slot de computação.
INITIALIZING	O workflow está iniciando um ambiente com o perfil de computação do job.
RUNNING	O ponto de entrada do workflow está em execução.
RUNNING: <stage>	O ponto de entrada reportou uma etapa com update_status(), como RUNNING: WAITING_FOR_QPU.
CANCELING	Você cancelou o job e o workflow está interrompendo-o.
DONE	O ponto de entrada terminou com sucesso.
ERROR	O ponto de entrada gerou um erro ou terminou com código diferente de zero. Consulte Encontre a causa de um erro.
CANCELED	Você cancelou o job, ou ele atingiu o limite de 24 horas.

DONE, ERROR e CANCELED são finais, e job.in_terminal_state() retorna True para eles. Um loop de status que espera outro valor, como SUCCEEDED, nunca termina.

Espere o job terminar

Para esperar até o job terminar, chame wait_for_final_state(). Ele retorna o status final e, com verbose=True, imprime cada mudança de status. Se você definir timeout em segundos, wait_for_final_state() levanta um TimeoutError quando o tempo se esgota antes de o job terminar. O job continua em execução:

run_workflow_remotely.ipynb

job.wait_for_final_state(timeout=600, verbose=True)

Waiting for job 3f2a1c0e-6b1d-4c0e-9a51-2f0c4d1e7a90
0:00:00  QUEUED
0:02:10  RUNNING
0:03:05  RUNNING: WAITING_FOR_QPU
0:41:44  DONE
'DONE'

wait_for_final_state() verifica o status após 1, 2, 4, 8, 16 e 32 segundos, e depois a cada 60 segundos, de modo que percebe o fim de um job curto em poucos segundos e faz poucas requisições para um job longo. Para verificar em um intervalo fixo, passe poll_interval em segundos.

Leia os logs

Tudo o que o workflow escreve na saída padrão ou no erro padrão vai para os logs do job. Você pode lê-los com job.logs() enquanto o job executa, antes de ele ter um resultado:

# This cell is hidden from users. It waits for the job to end, so that the
# logs are complete.
job.wait_for_final_state()

run_workflow_remotely.ipynb

print(job.logs())

Computing the expectation value for theta=0.5

# This cell is hidden from users. It checks the logs.
assert "Computing the expectation value for theta=0.5" in job.logs()

• O workflow envia novas linhas de log aproximadamente a cada 15 segundos, e uma última vez quando o job termina ou é cancelado. Logo depois que um job começa, job.logs() retorna No logs yet.

• O workflow mantém os 50 MB mais recentes de logs de cada job e adiciona uma linha no topo informando que os logs foram truncados.

Para decidir o que seu workflow registra nos logs, consulte Escreva logs. Para mostrar apenas as linhas que correspondem a uma expressão regular, passe include para job.filtered_logs(). Para ocultar linhas, passe exclude.

Obtenha o resultado

job.result() espera até o job terminar e então retorna o valor que o workflow passou para save_result(): run_workflow_remotely.ipynb

job.result()

{'theta': 0.5, 'expectation_value': 0.8775825618903728}

# This cell is hidden from users. It checks the result.
assert _ == {"theta": 0.5, "expectation_value": 0.8775825618903728}, _  # noqa: F821

Por padrão, job.result() verifica o status a cada 30 segundos e espera sem limite de tempo. Passe maxwait para limitar o número de verificações. Por exemplo, job.result(maxwait=40) espera cerca de 20 minutos. Para verificar uma única vez sem esperar, passe wait=False. Se job.result() parar de esperar antes de o job terminar, ele retorna {}.

Veja quanto tempo cada etapa levou

job.events() lista toda mudança de status e de etapa de um job com seu horário, para que você veja quanto tempo o job esperou por computação e quanto tempo cada etapa levou: run_workflow_remotely.ipynb

for event in reversed(job.events()):
    print(
        event.created,
        event.data.get("status") or event.data.get("sub_status"),
    )

2026-11-02T14:43:07.134377Z QUEUED
2026-11-02T14:43:24.474199Z PENDING
2026-11-02T14:46:07.309519Z RUNNING
2026-11-02T14:46:21.115038Z SUCCEEDED

# This cell is hidden from users. It checks the events.
_statuses = [e.data.get("status") for e in job.events()]
assert "QUEUED" in _statuses and "SUCCEEDED" in _statuses, _statuses

Os eventos usam os nomes de status armazenados. PENDING é INITIALIZING, e SUCCEEDED é DONE. Para o uso de QPU de uma execução, consulte Encontre os jobs de QPU de uma execução.

Encontre a causa de um erro

Quando o workflow levanta uma exceção, o job termina com o status ERROR. Por exemplo, execute o workflow com um ângulo que ele rejeita: run_workflow_remotely.ipynb

failed_job = my_workflow.run(theta=-1.0)

# This cell is hidden from users. It waits for the job to end.
failed_job.wait_for_final_state()

run_workflow_remotely.ipynb

failed_job.status()

'ERROR'

Os logs mostram o traceback: run_workflow_remotely.ipynb

print(failed_job.logs())

Traceback (most recent call last):
  File "/function_user_data/my_workflow.py", line 8, in <module>
    raise ValueError(f"theta must be between 0 and pi, got {theta}")
ValueError: theta must be between 0 and pi, got -1.0

job.result() raises a QiskitServerlessException: run_workflow_remotely.ipynb

from qiskit_serverless.exception import QiskitServerlessException

try:
    failed_job.result()
except QiskitServerlessException as error:
    print(error)

raise ValueError(f"theta must be between 0 and pi, got {theta}")
ValueError: theta must be between 0 and pi, got -1.0

# This cell is hidden from users. It checks the error.
assert failed_job.status() == "ERROR", failed_job.status()
assert (
    "ValueError: theta must be between 0 and pi, got -1.0"
    in failed_job.logs()
)
assert failed_job.events(type="ERROR") == []

Se o workflow reporta erros com send_error(), leia-os com job.events(type="ERROR") e leia suas mensagens com job.error_message(). Para um job sem erros reportados, job.events(type="ERROR") retorna uma lista vazia, e job.error_message() retorna {}.

run_workflow_remotely.ipynb

for event in job.events(type="ERROR"):
    print(event.created, event.data.get("code"), event.data.get("message"))

Gerencie jobs

Cancele um job

Cancele um job com job.cancel(). O status muda para CANCELING enquanto o workflow interrompe o job e depois para CANCELED: run_workflow_remotely.ipynb

job_to_cancel = my_workflow.run(theta=1.0)
job_to_cancel.cancel()

'Job has been stopped. No active runtime job ID associated with this serverless job ID.'

# This cell is hidden from users. It checks the cancellation.
assert (
    job_to_cancel.wait_for_final_state() == "CANCELED"
), job_to_cancel.status()

job.cancel() retorna uma mensagem sobre o job e sobre seus jobs de QPU vinculados, que ele também tenta cancelar. Para um job em execução, o workflow envia SIGTERM ao seu ponto de entrada. Para salvar resultados parciais, consulte Faça a limpeza quando um job for cancelado. Se o job já terminou, job.cancel() retorna Job already in terminal state. e não altera nada.

Encontre um job a partir de uma nova sessão

O objeto Job e o workflow existem apenas na sua sessão atual do Python. Depois de reiniciar o kernel de um notebook, passe o ID do job salvo para client.job(), ou liste seus jobs recentes com client.jobs(): run_workflow_remotely.ipynb

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
job = client.job(job_id)
my_workflow = client.load("my-workflow")

client.jobs(function=my_workflow, status="ERROR", limit=5)

[<Job | e6b1ffa1-cd26-437d-bd53-b265dd4ba952>,
 <Job | 11308f2a-8d7a-40a0-a970-7fdeca2b171b>,
 <Job | af597785-3470-469a-9051-607e0185892e>,
 <Job | a9051931-3bc2-4b80-9ce6-c39d28ce260c>,
 <Job | e329e8e3-e955-4188-9bfd-8f3050ee2d39>]

# This cell is hidden from users. It checks the job lookups.
assert job.result()["theta"] == 0.5
assert failed_job.job_id in [j.job_id for j in _]  # noqa: F821

client.jobs() retorna seus 10 jobs mais recentes, do mais novo para o mais antigo. Filtre com function, status (da tabela de status) ou created_after (um timestamp ISO 8601) e navegue pelas páginas com limit e offset. Para listar os workflows que você enviou, chame client.list().

A tabela Workloads na IBM Quantum Platform também lista seus jobs, com status, perfil de computação e tempos de execução.

Limite de tempo

Um workflow gerenciado cancela um job que executa por mais de 24 horas, e o job termina com o status CANCELED. Durante execuções longas, salve checkpoints no armazenamento de arquivos para não perder trabalho.

Solucione problemas de jobs

Problema	Causa
run() falha com Code: 429 e Active job limit reached.	Você atingiu o limite de jobs ativos. Espere alguns jobs terminarem ou cancele jobs de que você não precisa mais.
O job permanece QUEUED e run() avisou your account, plan or subscription is blocked.	O job não começa até que o bloqueio na sua conta, plano ou assinatura seja removido. Entre em contato com seu administrador.
job.logs() retorna No logs yet.	O job está na fila ou acabou de começar. Espere cerca de 15 segundos.
job.result() retorna {}	O job não terminou, foi cancelado ou não salvou resultado. Consulte Salve resultados.
job.error_message() retorna {}	O workflow levantou uma exceção sem reportar um erro. Leia o traceback com job.logs().
O job está CANCELED, mas você não o cancelou	O job atingiu o limite de 24 horas.

Próximos passos

Recomendações

• Escreva logs e trate erros para configurar os logs e erros que um job reporta.
• Execute uma varredura de parâmetros para executar o workflow uma vez para cada conjunto de entradas.