อินพุตและเอาต์พุตของ Executor

เวอร์ชันแพ็กเกจ

โค้ดในหน้านี้พัฒนาโดยใช้ requirements ต่อไปนี้ แนะนำให้ใช้เวอร์ชันเหล่านี้หรือใหม่กว่า

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

Executor primitive เป็นส่วนหนึ่งของ directed execution model ซึ่งให้ความยืดหยุ่นมากขึ้นในการกำหนดค่า error mitigation workflow

Inputs และ outputs ของ Executor primitive แตกต่างอย่างมากจาก Sampler และ Estimator primitives ตัวอย่างเช่น แทนที่จะรับ list ของ PUBs เป็น input Executor รับ QuantumProgram ซึ่งมี list ของ QuantumProgramItem objects container classes เหล่านี้ให้ความยืดหยุ่นมากกว่า PUB ซึ่งเป็นโครงสร้างข้อมูล tuple อย่างง่าย

Output ของ Executor คือ QuantumProgramResult ซึ่งเป็น iterable และมีหนึ่ง element สำหรับแต่ละ QuantumProgramItem input

อินพุต: โปรแกรมควอนตัม

ดังที่กล่าวไว้ก่อนหน้านี้ input ของ Executor primitive คือ QuantumProgram ซึ่งเป็น iterable ของ objects QuantumProgramItem objects เหล่านี้สามารถเป็นสองประเภท:

• CircuitItem ซึ่งโดยทั่วไปเก็บ circuit และค่า parameter (ถ้ามี)
• SamplexItem ซึ่งโดยทั่วไปเก็บสิ่งต่อไปนี้:
  • Template circuit
  • Samplex object ที่ใช้สร้าง parameter sets แบบสุ่มในขณะ runtime (เช่น เพื่อทำ twirling หรือ inject noise)
  • Arguments สำหรับ samplex ซึ่งอาจรวมถึงค่า parameter สำหรับ circuit ดั้งเดิม

item แต่ละตัวแทนงานคนละอย่างที่ Executor จะต้องทำ

ก่อนเริ่มต้น

ตัวอย่างโค้ดบางส่วนในหน้านี้ใช้ samplex ซึ่งเป็นส่วนหนึ่งของ Samplomatic package ดังนั้น ก่อนรัน code blocks เหล่านั้น คุณต้องติดตั้ง Samplomatic ตามที่แสดงใน code block ต่อไปนี้ สำหรับข้อมูลเพิ่มเติม ดู เอกสาร Samplomatic

pip install samplomatic

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

ตัวอย่าง: สร้าง QuantumProgram ด้วยสองงานที่แตกต่างกัน

ก่อนอื่นเริ่มต้น quantum program จากนั้น append program items โดยใช้ append_circuit_item หรือ append_samplex_item (ถ้ามี samplex) ตามที่แสดงในตัวอย่างต่อไปนี้

cell ต่อไปนี้เริ่มต้น QuantumProgram และระบุว่าควรรัน 1024 shots สำหรับทุก configuration ของแต่ละ item ในโปรแกรม

หมายเหตุ

ต่างจาก Sampler QuantumProgram รับค่า shot เพียงค่าเดียว ถ้าต้องการค่า shot ที่แตกต่างกัน คุณต้องมี QuantumProgram แยก ซึ่งจะเป็น job แยก

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

Append CircuitItem

จากนั้น append target circuit ที่ถูก transpile ตาม instruction set architecture (ISA) ของ backend เข้าไปใน QuantumProgram เนื่องจาก circuit นี้มีสอง parameters เราต้องให้ค่า parameter ด้วย (10 ชุดในตัวอย่างนี้) การรัน CircuitItem นี้คืองานแรกที่โปรแกรมจะดำเนินการ

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

Append SamplexItem

Circuit items รันโดยไม่มีการ randomize ใดๆ ในทางกลับกัน samplex items ให้คุณระบุวิธีการ randomize เนื้อหา cell ถัดไปใช้ฟังก์ชัน generate_boxing_pass_manager() เพื่อจัดกลุ่ม gates และ measurements ของ circuit เป็น boxes และเพิ่ม twirling annotation ให้แต่ละ box จากนั้นสร้างคู่ template circuit และ samplex โดยใช้ฟังก์ชัน build()

การรัน SamplexItem นี้คืองานที่สองที่โปรแกรมจะดำเนินการ

ดูเอกสาร Samplomatic API สำหรับรายละเอียดทั้งหมดเกี่ยวกับ samplex และ arguments ของมัน ดู Samplomatic Transpiler guide สำหรับข้อมูลเกี่ยวกับการใช้ฟังก์ชัน 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()

เอาต์พุต

Output ของ Executor คือ QuantumProgramResult ซึ่งเป็น iterable มีหนึ่ง entry ต่อ QuantumProgramItem input ในลำดับเดียวกับ input items แต่ละ output item เหล่านี้เป็น dictionary ที่ keys เป็น strings สอดคล้องกับชื่อ classical registers ใน input circuits (รวมถึงอื่นๆ) ดังนั้นคุณไม่ต้องจำชื่อเหล่านี้เองอีกต่อไป ต่างจากตอนใช้ output ของ Sampler ที่ต้องจำ ค่า dictionary มีประเภทเป็น np.ndarray

ผลลัพธ์สำหรับตัวอย่างก่อนหน้ามี items เหล่านี้:

ผลลัพธ์ CircuitItem

Item แรกมีผลลัพธ์ของการรันงานแรก (CircuitItem) ในโปรแกรม มี key เดียวคือ meas ซึ่งเป็นชื่อ classical register ใน input circuit ค่าของ key นี้ map ไปยัง np.ndarray ที่มี shape (parameter sets, shots, register bits) ซึ่งคือ (10, 1024, 3) สำหรับตัวอย่างด้านบน

โค้ดต่อไปนี้แสดงวิธีเข้าถึงข้อมูลนี้:

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

แกนขวาสุดของ result_0 วิ่งไปตามบิตคลาสสิกในลำดับแบบ little-endian ดังนั้นดัชนี 0 เก็บผลลัพธ์ของบิตคลาสสิกตัวที่ 0 ซึ่งเป็นบิตที่มีนัยสำคัญน้อยที่สุด

ผลลัพธ์ SamplexItem

Item ที่สองมีผลลัพธ์ของการรันงานที่สอง (SamplexItem) ในโปรแกรม item นี้มีหลาย keys Key meas ซึ่งเป็นชื่อ classical register ของ input circuit map ไปยัง array ผลลัพธ์ของ register นั้น array นี้มี shape (randomizations, parameter sets, shots, classical bits) หรือ (28, 10, 1024, 3) ในตัวอย่างนี้

นอกจากนี้ output มี key measurement_flips.meas ซึ่งเป็น bit-flip corrections เพื่อยกเลิก measurement twirling สำหรับ register meas shape ของ output นี้จะเป็น (28, 10, 1, 3) สำหรับตัวอย่างนี้เนื่องจากต้องใช้เพียง shot เดียวเพื่อทำ bit-flip

โปรดทราบว่าแกนขวาสุดของ result_1 วิ่งไปตามบิตคลาสสิกในลำดับแบบ 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)

ดู execution span

การเรียก Executor.run() ครั้งเดียวส่ง Job เดียว อย่างไรก็ตาม ฝั่งเซิร์ฟเวอร์ บริการไม่จำเป็นต้องรันทุก configuration ของโปรแกรมควอนตัมพร้อมกัน แต่จะจัดกลุ่ม Circuit ที่ผูกค่าแล้วของ program item เป็น chunk และรันทีละ chunk บน QPU คุณไม่ต้องแบ่งโปรแกรมหรือส่ง Job เพิ่ม การทำ chunk เกิดขึ้นอัตโนมัติภายใน Job เดียว แอตทริบิวต์ metadata ของผลลัพธ์มีข้อมูลเวลาของ chunk เหล่านี้ใน result.metadata.chunk_timing timestamp เหล่านี้คือเวลาเริ่มและหยุดจริงของแต่ละ chunk บน QPU คุณใช้ข้อมูลนี้เพื่อดูว่าข้อมูลใดถูกเก็บบน QPU เมื่อใด เพื่อทำความเข้าใจว่า Executor กระจายงานอย่างไร หรือเพื่อวินิจฉัยปัญหาด้านประสิทธิภาพได้

chunk_timing คือลิสต์ที่มี ChunkSpan หนึ่งอันต่อ chunk ที่รัน แต่ละ ChunkSpan มีแอตทริบิวต์ต่อไปนี้:

• start และ stop: timestamp แบบ UTC ที่ chunk เริ่มและหยุดรันบน QPU

• parts: ลิสต์ของอ็อบเจกต์ ChunkPart ที่อธิบายว่า program item ใดมีส่วนร่วมใน chunk แต่ละ ChunkPart มีสิ่งต่อไปนี้:

  • idx_item: index ของ program item ตามลำดับที่ item ถูกเพิ่มเข้าไปใน QuantumProgram

  • size: จำนวน configuration (องค์ประกอบของ shape ของ item) จาก item นั้นที่รันใน chunk นี้ ตัวอย่างเช่น SamplexItem ในตัวอย่างก่อนหน้ามี shape (28, 10) จึงมี 280 configuration ทั้งหมด

โค้ดต่อไปนี้พิมพ์ระยะเวลาของแต่ละ chunk และเนื้อหาของแต่ละ chunk สำหรับ Job ที่ส่งไปก่อนหน้า:

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

พร็อพเพอร์ตี QuantumProgramResult.timing ห่อลิสต์เดียวกันไว้ในอ็อบเจกต์ ChunkTiming ซึ่งรายงานข้อมูลโดยรวมของชุด chunk ด้วย:

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

Visualize execution span

ใช้ฟังก์ชัน draw_chunk_timings เพื่อ visualize เวลาของ chunk บนกราฟแท่งแบบโต้ตอบ แกน x คือเวลา และแต่ละ chunk ถูกวาดเป็นส่วนแนวนอนที่ครอบคลุมเวลาเริ่มและหยุด แกน y แสดงจำนวน configuration สะสมที่เสร็จแล้ว (หรือเปอร์เซ็นต์ของเวิร์กโหลดทั้งหมด หากคุณตั้ง normalize_y=True) เลื่อนเมาส์ไปเหนือส่วนหนึ่งเพื่อดู timestamp ระยะเวลา และ part ของ chunk นั้น

ฟังก์ชันนี้ส่งคืนรูป plotly ซึ่งคุณแสดงผลโดยตรง บันทึกลงไฟล์ หรือทำทั้งสองอย่างได้ รับได้ทั้งลิสต์ chunk_timing หรืออ็อบเจกต์ ChunkTiming คุณส่งได้หลายอัน เช่น จาก Job ต่างๆ เพื่อวาดบนพล็อตเดียวกัน ตั้ง common_start=True เพื่อจัดแนวให้เริ่มที่เวลาเดียวกันเพื่อเปรียบเทียบ

ต้องใช้ Plotly

draw_chunk_timings ต้องใช้ plotly ติดตั้งได้โดยรัน pip install plotly

ช่องว่างระหว่าง chunk

การ visualize อาจแสดงช่องว่างระหว่างการรัน chunk ทุก Job ที่ส่งไปยัง QPU จริงต้องผ่านการประมวลผลก่อนแบบคลาสสิก เช่น การแปลงเป็นพัลส์ที่ส่งไปยังฮาร์ดแวร์ได้ เนื่องจากการประมวลผลก่อนนี้ทำแบบขนานได้ บริการจึงนำ Job ออกจากคิวและเริ่มหลาย Job พร้อมกัน อย่างไรก็ตาม QPU ประมวลผลได้ครั้งละหนึ่งหน่วยงานเท่านั้น ดังนั้นหาก Job สร้างหลาย chunk chunk ของมันอาจสลับกับ chunk ของ Job ผู้ใช้อื่น และแต่ละ chunk อาจต้องรอคิวบน QPU นี่คือเหตุผลที่อาจเกิดช่องว่างระหว่าง chunk

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

ขั้นตอนถัดไป

คำแนะนำ

• สำรวจ ตัวอย่าง ที่ใช้ Executor
• เรียนรู้เกี่ยวกับ directed execution model
• ทำความเข้าใจ Executor broadcasting