Escribir registros y gestionar errores

Configura el código de tu workflow para escribir registros, informar de errores a tu notebook local y guardar el trabajo parcial cuando se cancela un trabajo. Para leer estas salidas desde tu notebook local, consulta Ejecutar un managed workflow.

En el workflow	En tu notebook local
Escribe con un logger del módulo logging, o llama a print().	job.logs() devuelve las líneas.
Lanza una excepción, o termina con un código distinto de cero.	job.status() devuelve ERROR, y job.logs() muestra el traceback.
Llama a send_error().	job.error_message() y job.events(type="ERROR") devuelven el código, el mensaje y los detalles.
Llama a save_result() en un manejador de SIGTERM.	job.result() devuelve el resultado parcial de un trabajo cancelado.

Escribir registros

Todo lo que el workflow escribe en la salida estándar o en el error estándar va a los registros del trabajo. Lee los registros con job.logs() mientras el trabajo se ejecuta o después de que termine.

Configurar un logger

Usa el módulo logging de la biblioteca estándar de Python. logging.basicConfig() escribe en el error estándar, que llega a los registros sin retraso. El formato del siguiente ejemplo añade una marca de tiempo a cada línea:

source_files/my_workflow.py

import logging

logging.basicConfig(
    level=logging.WARNING,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger("my_workflow")
logger.setLevel(logging.INFO)

logger.info("Transpiled %d circuits: depth %d -> %d", len(circuits), 1250, 730)

2026-11-01 21:57:35,195 INFO my_workflow: Transpiled 12 circuits: depth 1250 -> 730

level=logging.WARNING oculta los mensajes INFO de otros paquetes, y logger.setLevel(logging.INFO) conserva los mensajes INFO de tu propio logger. qiskit_ibm_runtime define su propio nivel con la variable de entorno QISKIT_IBM_RUNTIME_LOG_LEVEL.

print() también llega a los registros, pero la salida estándar usa búfer, por lo que las líneas impresas podrían no aparecer hasta que termine el trabajo. Si usas print(), pasa flush=True.

Definir el nivel de registro en cada ejecución

Para depurar una ejecución sin cambiar el código, pasa el nivel de registro como argumento:

run_workflow_remotely.ipynb

job = my_workflow.run(circuits=circuits, log_level="DEBUG")

Para habilitar esto, define el nivel de tu logger a partir del argumento:

source_files/my_workflow.py

from qiskit_serverless import get_arguments

arguments = get_arguments()
logger.setLevel(arguments.get("log_level", "INFO"))

Si el workflow tiene un esquema de argumentos, declara log_level en él, por ejemplo como {"enum": ["DEBUG", "INFO", "WARNING"]}.

Práctica recomendada: decide qué registrar

Registra lo suficiente para reconstruir una ejecución a partir de sus registros:

• Al inicio, registra lo que recibió la ejecución. Registra el nombre del backend, el número de circuitos y los demás ajustes que cambian entre ejecuciones.

• Registra el inicio y el final de cada etapa. Usa mensajes fijos que puedas buscar.

• Registra las iteraciones a intervalos fijos. Por ejemplo, registra una línea como iteration 10/200, energy=-1.1000 cada 10 iteraciones, en lugar de en cada iteración.

• Registra las excepciones con logger.exception() dentro de un bloque except. La línea incluye el traceback completo, incluso si el workflow se recupera del error.

• Desactiva las barras de progreso. Los registros almacenan cada actualización de una barra de progreso, como la de tqdm, como una línea separada. Pasa disable=True a tqdm, o registra a intervalos en su lugar.

• Mantén los registros pequeños. No registres datos grandes, como cada shot. Un managed workflow conserva solo los 50 MB más recientes de registros de cada trabajo.

• Usa las palabras "error" y "exception" solo en mensajes sobre fallos. Si un trabajo fallido no informó de su error, la excepción que lanza job.result() contiene solo las líneas de registro que incluyen estas palabras.

Práctica recomendada: informa de los errores con send_error()

Cuando el workflow lanza una excepción, el trabajo termina con el estado ERROR y el traceback va a los registros. Un managed workflow no almacena el error, ni siquiera para un ServerlessError. job.error_message() devuelve {}, y la excepción que lanza job.result() contiene solo las líneas de registro que incluyen error o exception. Las comprobaciones de tipos y rangos de los argumentos corresponden a un esquema de argumentos, que rechaza una ejecución incorrecta antes de que comience el trabajo.

Informa con send_error() de cada error que detenga el workflow, para que tu notebook local reciba un error breve con un código y detalles:

run_workflow_remotely.ipynb

print(job.error_message())

| Message: The circuit has more qubits than the backend.
| Code: 4500
| Exception: InputError
| Details:
|   - circuit_qubits: 10
|   - backend_qubits: 5

job.result() lanza una QiskitServerlessException con el mismo texto.

Para habilitar esto, lanza ServerlessError donde ocurre el error y envuelve el código de tu punto de entrada en un bloque try que envíe cualquier error con send_error(). Un workflow añade este manejador a las Catalog Functions, pero no a los managed workflows:

source_files/my_workflow.py

from qiskit_serverless import ServerlessError, send_error

class InputError(ServerlessError):
    """The arguments are valid, but the workflow cannot use them."""

def main():
    ...
    if num_qubits > backend.num_qubits:
        raise InputError(
            code="4500",
            message="The circuit has more qubits than the backend.",
            details={"circuit_qubits": num_qubits, "backend_qubits": backend.num_qubits},
        )
    ...

try:
    main()
except Exception as error:
    send_error(
        code=getattr(error, "code", "1"),
        message=getattr(error, "message", repr(error)),
        exception=type(error).__name__,
        args=getattr(error, "details", None),
    )
    raise

• Vuelve a lanzar el error. El raise final conserva el traceback en los registros y termina el trabajo con el estado ERROR.

• Crea una subclase de ServerlessError para nombrar un grupo de errores. El manejador envía el nombre de la clase, como InputError, en el campo exception.

• Otras excepciones también llegan al manejador. Reciben el código 1, y su repr() se convierte en el mensaje, como ValueError('bad shots').

Informa de un problema y continúa

send_error() no detiene el flujo de trabajo ni cambia su estado, por lo que también puedes informar de problemas de los que el flujo de trabajo se recupera, como cada circuito que omite:

source_files/my_workflow.py

valid_circuits = []
for index, circuit in enumerate(circuits):
    if circuit.num_qubits > backend.num_qubits:
        send_error(
            code="4501",
            message="Skipped a circuit that is wider than the backend.",
            exception="InputError",
            args={"circuit_index": index},
        )
        continue
    valid_circuits.append(circuit)

job.events(type="ERROR") devuelve todos los errores que informó el flujo de trabajo, incluidos los errores de un job que terminó con el estado DONE. Consulta Find the cause of an error.

• Pasa números y cadenas en args. error_message() muestra solo el primer elemento de una lista.

• Comprueba el valor devuelto. Si un flujo de trabajo no puede almacenar el error, por ejemplo porque el código se ejecuta fuera de un job, send_error() escribe el error en los registros y devuelve False.

Limpia los recursos cuando se cancela un job

Cuando cancelas un job, el flujo de trabajo envía SIGTERM a tu punto de entrada, sube los registros finales y apaga el entorno. Sin un controlador, Python se detiene de inmediato y los bloques finally no se ejecutan. El flujo de trabajo también intenta cancelar los jobs de QPU y las sesiones que el flujo de trabajo inició con get_runtime_service(). Consulta Cancel linked QPU jobs.

Para conservar el trabajo parcial, registra un controlador para SIGTERM. Así, tu notebook local obtiene el resultado parcial de un job cancelado:

run_workflow_remotely.ipynb

job.cancel()
print(job.result())  # {'status': 'canceled', 'partial_results': [...]}

Para habilitarlo, agrega el controlador al inicio de tu punto de entrada:

source_files/my_workflow.py

import signal
import sys

from qiskit_serverless import save_result

partial_results = []

def handle_sigterm(signum, frame):
    save_result({"status": "canceled", "partial_results": partial_results})
    sys.exit(0)

signal.signal(signal.SIGTERM, handle_sigterm)

sys.exit() en el controlador genera SystemExit en el código que se está ejecutando, por lo que los bloques finally se ejecutan antes de que termine el script. El job sigue terminando con el estado CANCELED.

Mantén el controlador breve, porque el flujo de trabajo apaga el entorno poco después de SIGTERM. Para ejecuciones largas, guarda también puntos de control, de modo que una ejecución posterior pueda reanudarse después de que un job falle o alcance el límite de 24 horas. Consulta Best practice: checkpoint long jobs.

Próximos pasos

Recomendaciones

• Ejecuta un flujo de trabajo gestionado para iniciar un job y leer su estado, sus registros y su resultado.
• Gestiona archivos entre ejecuciones para guardar puntos de control desde los que una ejecución posterior pueda reanudarse.
• Elige un perfil de cómputo para ejecutar tu flujo de trabajo con más recursos de CPU, memoria o GPU.