Intrările și ieșirile Executor

Versiuni de pachete

Codul de pe această pagină a fost dezvoltat folosind următoarele cerințe. Recomandăm să folosești aceste versiuni sau mai noi.

qiskit[all]~=2.6.1
qiskit-ibm-runtime~=0.50.0
samplomatic~=0.21.0

Primitiva Executor face parte din modelul de execuție direcționată, care oferă mai multă flexibilitate la personalizarea unui flux de lucru de atenuare a erorilor.

Intrările și ieșirile primitivei Executor sunt foarte diferite față de cele ale primitivelor Sampler și Estimator. De exemplu, în loc să primească o listă de PUB-uri ca intrare, Executor primește un QuantumProgram, care conține o listă de obiecte QuantumProgramItem. Aceste clase container îți oferă mai multă flexibilitate decât un PUB, care este o structură de date simplă de tip tuplu.

Ieșirea Executor este un QuantumProgramResult, care este iterabil și conține un element pentru fiecare QuantumProgramItem de intrare.

Intrări: Programe cuantice

Așa cum s-a menționat anterior, intrarea la o primitivă Executor este un QuantumProgram, care este un iterabil de obiecte QuantumProgramItem. Aceste obiecte pot fi de două tipuri:

• CircuitItem, care stochează de obicei un circuit și valorile sale de parametri (dacă există).
• SamplexItem, care stochează de obicei următoarele:
  • Un circuit șablon
  • Un obiect samplex, folosit pentru a genera seturi randomizate de parametri la runtime (de exemplu, pentru a efectua twirling sau a injecta zgomot)
  • Argumente pentru samplex, care pot include valori de parametri pentru circuitul original

Fiecare dintre aceste elemente reprezintă o sarcină diferită pe care Executor trebuie să o îndeplinească.

Înainte de a începe

Unele exemple de cod de pe această pagină folosesc samplex, care face parte din pachetul Samplomatic. Prin urmare, înainte de a rula acele blocuri de cod, trebuie să instalezi Samplomatic, așa cum se arată în blocul de cod următor. Pentru mai multe informații, consultă documentația Samplomatic.

pip install samplomatic

# For visualization support, include the visualization dependencies.
# pip install samplomatic[vis]

Exemplu: Creează un QuantumProgram cu două sarcini diferite

Mai întâi inițializează programul tău cuantic, apoi adaugă elemente de program la acesta folosind fie append_circuit_item, fie append_samplex_item (dacă există un samplex), după cum se arată în exemplele de mai jos.

Celula de mai jos inițializează un QuantumProgram și specifică că ar trebui să ruleze 1024 de shot-uri pentru fiecare configurație a fiecărui element din program.

notă

Spre deosebire de Sampler, un QuantumProgram acceptă o singură valoare de shot. Dacă dorești o valoare diferită de shot, ai nevoie de un QuantumProgram separat, care ar fi un job separat.

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

_needed = {"numpy": "numpy", "qiskit": "qiskit", "qiskit_ibm_runtime": "qiskit-ibm-runtime", "samplomatic": "samplomatic"}
_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")

from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime.quantum_program import QuantumProgram
from qiskit_ibm_runtime import Executor, QiskitRuntimeService
from qiskit.circuit import Parameter, QuantumCircuit
import numpy as np
from samplomatic import build
from samplomatic.transpiler import generate_boxing_pass_manager

# Initialize an empty program
program = QuantumProgram(shots=1024)

# Initialize and transpile a 3-qubit quantum circuit with 2 parameters.
circuit = QuantumCircuit(3)
circuit.h(0)
circuit.cx(0, 1)
circuit.cx(1, 2)
circuit.rz(Parameter("theta"), 0)
circuit.rz(Parameter("phi"), 1)

# `measure_all` adds a 3-bit classical register named "meas"
circuit.measure_all()

# Choose the least busy backend
service = QiskitRuntimeService()
backend = service.least_busy(operational=True, simulator=False)

# Generate a preset pass manager
# This will be used to convert the abstract circuit to an
# equivalent Instruction Set Architecture (ISA) circuit.
preset_pass_manager = generate_preset_pass_manager(
    backend=backend, optimization_level=0
)

# Transpile the circuit
isa_circuit = preset_pass_manager.run(circuit)

Adaugă un CircuitItem

Apoi, adaugă circuitul țintă, transpilat conform arhitecturii setului de instrucțiuni (ISA) a backend-ului, la QuantumProgram. Deoarece acest circuit are doi parametri, trebuie să furnizăm și valorile parametrilor (10 seturi în acest exemplu). Rularea acestui CircuitItem este prima sarcină pe care o va efectua programul.

# Append the transpiled circuit and an array
# containing 10 sets of parameter values to the program
program.append_circuit_item(
    isa_circuit,
    circuit_arguments=np.random.rand(
        10, 2
    ),  # 10 sets of parameter values and 2 parameters
)

Adaugă un SamplexItem

Elementele de circuit sunt executate fără nicio randomizare. În schimb, elementele samplex îți permit să specifici cum să le randomizezi conținutul. Celula următoare folosește funcția generate_boxing_pass_manager() pentru a grupa porțile și măsurătorile circuitului în cutii și a adăuga o adnotare de twirling fiecărei cutii. Apoi generează o pereche circuit șablon și samplex folosind funcția build().

Rularea acestui SamplexItem este a doua sarcină pe care o va efectua programul.

Consultă documentația API Samplomatic pentru detalii complete despre samplex și argumentele sale. Consultă Ghidul Transpiler Samplomatic pentru informații despre utilizarea funcției generate_boxing_pass_manager().

# Transpile the circuit, additionally grouping gates and measurements into annotated boxes
preset_pass_manager = generate_preset_pass_manager(
    backend=backend, optimization_level=0
)

# Use the boxing pass manager to group gates
# and measurements into boxes and add
# a`Twirl` annotation.
preset_pass_manager.post_scheduling = generate_boxing_pass_manager(
    # Add gate twirling
    enable_gates=True,
    # Add measurement twirling
    enable_measures=True,
)
boxed_circuit = preset_pass_manager.run(circuit)

# Build the template circuit and the samplex.  The template circuit has parametric gates
# without fixed values and the samplex randomly generates the parameter
# values on the server side at runtime to perform twirling.
template_circuit, samplex = build(boxed_circuit)

# Determine what arguments are required by the samplex.
# Input the arguments in samplex_arguments.
print(samplex.inputs())

TensorInterface(<
  - 'parameter_values' <float64[2]>: Input parameter values to use during sampling.
>)

# Append the template circuit and samplex as a samplex item
program.append_samplex_item(
    template_circuit,
    samplex=samplex,
    samplex_arguments={
        # the arguments required by the samplex.sample method
        "parameter_values": np.random.rand(10, 2),
    },
    shape=(28, 10),  # 28 randomizations and 10 sets of parameter values
)

# Initialize an Executor with the default options
executor = Executor(mode=backend)

# Submit the job
job = executor.run(program)

# Retrieve the result
result = job.result()

Ieșiri

Ieșirea Executor este un QuantumProgramResult, care este iterabil. Conține câte un element pentru fiecare QuantumProgramItem de intrare, în aceeași ordine ca elementele de intrare. Fiecare dintre aceste elemente de ieșire este un dicționar ale cărui chei sunt șiruri de caractere care corespund numelor registrelor clasice din circuitele de intrare (printre altele), astfel că nu mai trebuie să memorezi aceste nume ca la ieșirea Sampler. Valorile dicționarului sunt de tip np.ndarray.

Rezultatul din exemplul anterior conține aceste elemente:

Rezultatul CircuitItem

Primul element conține rezultatele rulării primei sarcini (un CircuitItem) din program. Conține o singură cheie, meas, care este numele registrului clasic din circuitul de intrare. Valoarea acestei chei corespunde unui np.ndarray de formă (parameter sets, shots, register bits), adică (10, 1024, 3) pentru exemplul de mai sus.

Codul următor ilustrează cum să accesezi aceste informații:

# Access the results of the classical register of task #0, a CircuitItem
result_0 = result[0]["meas"]
print(f"Result shape: {result_0.shape}")

Result shape: (10, 1024, 3)

Axa cea mai din dreapta a lui result_0 parcurge biții clasici în ordine little-endian. Astfel, indexul 0 conține rezultatul bitului clasic 0, bitul cel mai puțin semnificativ.

Rezultatul SamplexItem

Al doilea element conține rezultatele rulării celei de-a doua sarcini (un SamplexItem) din program. Acest element conține mai multe chei. Cheia meas, care este numele registrului clasic al circuitului de intrare, corespunde tabloului de rezultate al acelui registru. Acest tablou are forma (randomizations, parameter sets, shots, classical bits), adică (28, 10, 1024, 3) în acest exemplu.

În plus, ieșirea conține o cheie measurement_flips.meas, care reprezintă corecțiile de inversare a biților pentru a anula twirling-ul măsurătorilor pentru registrul meas. Forma acestei ieșiri va fi (28, 10, 1, 3) pentru exemplul nostru, deoarece este necesar un singur shot pentru a efectua inversarea biților.

Reține că axa cea mai din dreapta a lui result_1 parcurge biții clasici în ordine little-endian.

# Access the results of the classical register of task #1
result_1 = result[1]["meas"]
print(f"Result shape: {result_1.shape}")

# Access the bit-flip corrections
flips_1 = result[1]["measurement_flips.meas"]
print(f"Bit-flip corrections shape: {flips_1.shape}")

# Undo the bit flips via classical XOR
unflipped_result_1 = result_1 ^ flips_1

Result shape: (28, 10, 1024, 3)
Bit-flip corrections shape: (28, 10, 1, 3)

Vizualizează execution spans

Un singur apel Executor.run() trimite un singur job. Pe partea serverului, însă, serviciul nu rulează neapărat toate configurațiile programului cuantic în același timp. În schimb, grupează circuitele legate ale elementelor programului în bucăți (chunks) și rulează câte o bucată pe rând pe QPU. Nu trebuie să împarți programul sau să trimiți joburi suplimentare; împărțirea în bucăți se face automat în cadrul unui singur job. Atributul metadata al rezultatului conține informații de temporizare despre aceste bucăți în result.metadata.chunk_timing. Aceste marcaje de timp sunt momentele reale de start și de oprire ale fiecărei bucăți pe QPU. Poți folosi aceste informații pentru a vedea când au fost colectate anumite date pe QPU, pentru a înțelege cum a distribuit Executor munca sau pentru a diagnostica probleme de performanță.

chunk_timing este o listă care conține câte un ChunkSpan pentru fiecare bucată executată. Fiecare ChunkSpan are următoarele atribute:

• start și stop: Marcajele de timp UTC la care bucata a început și s-a oprit din execuție pe QPU.

• parts: O listă de obiecte ChunkPart care descriu ce elemente de program au contribuit la bucată. Fiecare ChunkPart conține următoarele:

  • idx_item: Indexul elementului de program, în ordinea în care elementele au fost adăugate la QuantumProgram.

  • size: Numărul de configurații (elemente din forma elementului) din acel element care au fost executate în această bucată. De exemplu, SamplexItem din exemplul anterior are forma (28, 10), deci are 280 de configurații în total.

Următorul cod afișează durata fiecărei bucăți și conținutul fiecărei bucăți pentru jobul trimis anterior:

for idx, span in enumerate(result.metadata.chunk_timing):
    duration = (span.stop - span.start).total_seconds()
    print(f"Chunk {idx}: {span.start} to {span.stop} ({duration:.3f} s)")
    for part in span.parts:
        print(f"  Item {part.idx_item}: {part.size} configurations")

Chunk 0: 2026-11-06 01:47:46.009718+00:00 to 2026-11-06 01:49:05.577720+00:00 (79.568 s)
  Item 0: 10 configurations
  Item 1: 280 configurations

Proprietatea QuantumProgramResult.timing încadrează aceeași listă într-un obiect ChunkTiming, care raportează și informații globale despre colecția de bucăți:

timing = result.timing

print("Number of chunks:", len(timing))
print("   Start of the first chunk:", timing.start)
print("      End of the last chunk:", timing.stop)
print("        Total duration (s):", timing.duration)

Number of chunks: 1
   Start of the first chunk: 2026-11-06 01:47:46.009718+00:00
      End of the last chunk: 2026-11-06 01:49:05.577720+00:00
        Total duration (s): 79.568002

Vizualizează execution spans

Folosește funcția draw_chunk_timings pentru a vizualiza temporizările bucăților pe un grafic cu bare interactiv. Axa x este timpul, iar fiecare bucată este desenată ca un segment orizontal care se întinde între momentele ei de start și de oprire. Axa y arată numărul cumulat de configurații finalizate (sau procentul din sarcina totală, dacă setezi normalize_y=True). Treci cu cursorul peste un segment pentru a vedea marcajele de timp, durata și părțile acelei bucăți.

Funcția returnează o figură plotly, pe care o poți afișa direct, salva într-un fișier sau ambele. Acceptă fie lista chunk_timing, fie un obiect ChunkTiming. Poți transmite mai multe, de exemplu din joburi diferite, pentru a le desena pe același grafic. Setează common_start=True pentru a le alinia la un moment comun de start, în vederea comparării.

Plotly necesar

draw_chunk_timings necesită plotly. Îl poți instala rulând pip install plotly.

Goluri între bucăți

Vizualizarea poate arăta goluri între execuțiile bucăților. Toate joburile trimise unui QPU real trec printr-o preprocesare clasică, precum traducerea în pulsuri care pot fi trimise hardware-ului. Deoarece această preprocesare se poate face în paralel, serviciul scoate din coadă și pornește simultan mai multe joburi. QPU-ul, însă, poate procesa o singură unitate de lucru pe rând. Prin urmare, dacă un job generează mai multe bucăți, bucățile lui pot fi intercalate cu cele ale joburilor altor utilizatori, iar fiecare bucată poate trebui să își aștepte rândul pe QPU. De aceea pot apărea goluri între bucăți.

from qiskit_ibm_runtime.visualization import draw_chunk_timings

fig = draw_chunk_timings(result.timing, names="Executor job")

# Alternatively, call the `draw()` method of the ChunkTiming object
# fig = result.timing.draw()

# Uncomment the following line to display the figure
# fig.show(renderer="notebook")

# Save to a file
# fig.write_html("chunk_timings.html")

Pașii următori

Recomandări

• Explorează exemple care folosesc Executor.
• Află despre modelul de execuție direcționată.
• Înțelege broadcasting-ul Executor.