Qiskit Sampler 与 Estimator
Sampler 和 Estimator 都接收 Qiskit 线路：Sampler 返回测量分布，Estimator 返回观测量的期望值及估计标准误。本页从一个 Sampler 实验开始，再介绍 Estimator、参数数组和多观测量。
| 目标 | 选择 | 结果入口 |
|---|---|---|
| 拿到测量结果，每个参数点都有自己的 counts | QuantumSourceSampler / SamplerV2 | result[pub_index].data.<经典寄存器名>.get_counts(...) |
| 估计 Pauli 观测量的期望值 | QuantumSourceEstimator / EstimatorV2 | result[pub_index].data.evs 和 data.stds |
| 只提交一条普通测量线路，直接取 Qiskit Result | QuantumSourceBackend | job.result().get_counts()；见 Qiskit 入门 |
每个实验按 prepare → 查看报价 → confirm → result 的顺序运行。prepare 准备报价，confirm 确认执行，result 读取结果；run 将准备和确认合并为一步。确认真机实验会使用设备资源，并按报价计费。
1. 创建 Service 和所选设备
先配置平台 API 地址、平台 SDK 凭据和设备目录中的 short_code。以下实验共用同一个 service 与 backend。
import os
from qiskit import QuantumCircuit
from quantum_source_qiskit import QuantumSourceService
service = QuantumSourceService(
base_url=os.environ["QUANTUM_SOURCE_API_URL"],
currency="OPEN",
)
backend = service.return_backend(os.environ["QUANTUM_SOURCE_BACKEND"])
print(backend.name)
print(backend.limits)
Service 负责认证、钱包和客户端连接，backend 代表当前设备。需要使用 FULL 时，在创建 Service 时选择 currency="FULL"。通过 limits 查看 shots 等设备限制。
2. Sampler：准备一个 Bell 采样实验
这里创建名为 c 的经典寄存器；最终结果也从 data.c 读取。Sampler 的输入必须包含经典测量寄存器，并应将需要采样的量子位测量到其中。
bell = QuantumCircuit(2, 2)
bell.h(0)
bell.cx(0, 1)
bell.measure([0, 1], [0, 1])
sampler = service.create_sampler(
backend=backend,
options={"default_shots": 128},
)
sample_prepared = sampler.prepare([bell])
sample_prepared.quote
sampler.prepare([bell]) 返回 PreparedQuantumSourcePrimitive，保存本次实验的完整报价。quote.task_count、quote.total_shots 和金额分别显示任务数量、总采样次数和费用。options.default_shots 设置采样器的默认采样次数。
print(sample_prepared.quote.to_dict())
审阅上面的设备、任务数、采样总量、金额和币种后，只有决定执行这一次实验时才运行下一单元。
sample_job = sample_prepared.confirm()
确认后返回 QuantumSourcePrimitiveJob，在本机后台等待平台结果并整理 Qiskit 数据。若提交仍待平台确认，返回的 PreparedSubmissionReceipt 可按本页恢复步骤继续查询。
3. 从原生结果中读取 counts、寄存器和参数点
sample_result = sample_job.result(timeout=300)
pub_result = sample_result[0]
counts = pub_result.data.c.get_counts()
print(counts)
print("实际采样数：", pub_result.data.c.num_shots)
print("平台任务号：", sample_result.metadata.get("platform_job_ids", []))
sample_result 是 Qiskit PrimitiveResult。[0] 选择第一个 PUB，pub_result.data 是 DataBin，data.c 选择经典寄存器 c 对应的 BitArray。get_counts() 返回比特串到测量次数的映射。
| 线路的写法 | 对应结果字段 | 原因 |
|---|---|---|
| QuantumCircuit(2, 2)，然后 measure(...) | pub_result.data.c | 自动创建的经典寄存器名为 c |
| 没有经典寄存器的 QuantumCircuit(2)，然后 measure_all() | pub_result.data.meas | measure_all 创建名为 meas 的经典寄存器 |
| 自己添加 ClassicalRegister(..., "readout") | pub_result.data.readout | 数据字段跟随实际寄存器名，不固定为 c 或 meas |
| 一个 PUB 含两个参数点 | data.c.get_counts(0) 与 get_counts(1) | 选择该寄存器的不同参数点 |
# 继续读取刚取得的 PUB 结果。
print("经典寄存器：", list(pub_result.data))
print("PUB 数组形状：", pub_result.data.shape)
print("样本顺序说明：", pub_result.metadata.get("sample_order"))
# 有多个经典寄存器时，可将它们按 Qiskit 规则合并后查看。
combined_counts = pub_result.join_data().get_counts()
print(combined_counts)
BitArray 由平台返回的整数 counts 展开得到。sample_order="counts_expansion" 表示样本按计数展开顺序排列，适合读取各位串的次数和统计分布。
4. PUB 与 shots：默认值和覆盖顺序
PUB 可以理解为一次计算的输入单元。Sampler 接受单个已测量线路，或 (线路, 参数值)，或 (线路, 参数值, shots)。最外层列表是多个 PUB；一个 PUB 的参数数组还可能展开成多个实际任务。
| 优先级 | 设置方式 | 影响范围 |
|---|---|---|
| 最高 | (circuit, parameter_values, 128) | 只影响这个 PUB；其中每个参数点各自使用 128 shots |
| 其次 | sampler.prepare(pubs, shots=256)，run 同理 | 给本次调用中未明确 shots 的 PUB 提供默认值 |
| 其次 | create_sampler(..., options={"default_shots": 512}) | 给未提供以上两项的 PUB 提供采样器默认值 |
| 最终默认 | 未做任何上述设置 | 当前 Sampler 默认 100 shots；仍须符合目标设备限制 |
# 准备一个包含两个 PUB 的实验，并查看总报价。
two_pub_prepared = sampler.prepare(
[(bell, None, 128), bell.copy()],
shots=256,
)
print(two_pub_prepared.quote.to_dict())
第一项采样 128 次，第二项使用本次默认值 256，共 384 shots。两项分别准备任务，接受整批报价后即可确认执行。
5. 参数数组：为每个参数点采样
数值数组最后一轴对应线路参数顺序，前面的轴构成参数点形状。先看 circuit.parameters 的实际顺序，再组织数据。本例有一个参数 theta 和两个参数点，每点 128 shots。
import numpy as np
from qiskit.circuit import Parameter
theta = Parameter("theta")
parameter_circuit = QuantumCircuit(1, 1)
parameter_circuit.ry(theta, 0)
parameter_circuit.measure(0, 0)
print(list(parameter_circuit.parameters))
values = np.array([[0.0], [0.5]]) # 形状 (2, 1)：2 个参数点、每点 1 个参数
parameter_prepared = sampler.prepare([(parameter_circuit, values, 128)])
parameter_prepared.quote
客户端先绑定两组参数，再准备两条线路，总计 256 shots。若决定运行此独立实验，按相同的 prepare → 审价 → confirm 流程继续；下面两个单元是它自己的执行和结果步骤。
parameter_job = parameter_prepared.confirm()
parameter_result = parameter_job.result(timeout=300)
bits = parameter_result[0].data.c
print("参数形状：", bits.shape) # (2,)
print("theta=0.0：", bits.get_counts(0))
print("theta=0.5：", bits.get_counts(1))
如果 values 形状为 (2, 3, 1)，结果参数形状就是 (2, 3)。用 get_counts((row, column)) 读取指定参数点，或用 get_counts() 汇总全部参数点。
6. Estimator：另一个独立的期望值实验
Estimator 接收状态制备线路和观测量。插件自动生成测量基变换及测量线路，平台执行这些线路后，由客户端计算期望值和标准误。
from qiskit.quantum_info import SparsePauliOp
state = QuantumCircuit(1)
state.ry(0.5, 0)
observable = SparsePauliOp.from_list([("Z", 1.0)])
estimator = service.create_estimator(
backend=backend,
options={"default_precision": 0.1},
)
estimate_prepared = estimator.prepare([(state, observable)])
estimate_prepared.quote
这个 PUB 是 (线路, 观测量)。Z 的系数为 1，所求是这个状态下的 ⟨Z⟩。prepare 返回同样的 primitive 准备对象；可以先查看完整报价，再决定执行。仅当选择运行这个独立实验时，继续以下单元。
estimate_job = estimate_prepared.confirm()
estimate_result = estimate_job.result(timeout=300)
print("期望值：", estimate_result[0].data.evs)
print("估计标准误：", estimate_result[0].data.stds)
print("平台任务号：", estimate_result.metadata.get("platform_job_ids", []))
evs 与 stds 是 Qiskit 原生数组；单个观测量和单个参数点对应标量形状数组。evs 保存期望值，stds 保存采样估计的统计标准误。可将实际 evs 与 Ry(0.5) 后测 Z 的理论值 cos(0.5) 比较。
7. precision 怎样变成真实采样预算
| 优先级或选项 | 当前行为 | 为什么要看报价 |
|---|---|---|
| PUB 的第 4 项 precision | (circuit, observable, parameter_values, precision)；优先于方法默认 | 同一调用中的不同 PUB 可以有不同 shots |
| prepare/run(..., precision=...) | 给本次未指定精度的 PUB 提供默认值 | 每个实际测量线路分别计算采样次数 |
| options.default_precision | 工厂 options 设置默认；Qiskit 2.5.2 默认 0.015625 | 该默认对应每个实际测量线路 4096 shots |
| options.abelian_grouping | 所核对的 Qiskit 默认 True；合并逐比特可交换的 Pauli 测量 | 是否能共享测量基会改变实际任务数 |
# 查看估计器选项及采样次数公式。
import math
print(estimator.options)
for precision in (0.1, 0.05, 0.015625):
print(precision, "->", math.ceil(1 / precision**2), "shots / 测量线路")
采样次数按 ceil(1 / precision**2) 计算：precision 越小，采样数越大。选择正的有限数，并使对应 shots 位于设备允许范围内。estimator.options 可查看当前安装的 Qiskit BackendEstimatorV2 所提供的默认选项。
8. 多观测量与参数广播
下面的观测量数组形状为 (2, 1)，参数点形状为 (2,)，广播后返回 (2, 2)：两行分别对应 Z 和 X，两列分别对应两个 theta。先准备这个实验并查看总报价。
import numpy as np
from qiskit.circuit import Parameter
sweep_theta = Parameter("sweep_theta")
sweep_state = QuantumCircuit(1)
sweep_state.ry(sweep_theta, 0)
observables = [["Z"], ["X"]]
sweep_values = np.array([[0.0], [0.5]])
broadcast_prepared = estimator.prepare(
[(sweep_state, observables, sweep_values)],
precision=0.1,
)
print(broadcast_prepared.quote.to_dict())
客户端按参数点和可共享测量基展开任务。这里的 Z 和 X 分别测量，两个参数点也分别执行。quote.task_count、quote.total_shots 和 amount 显示展开后的任务数、总采样量和费用。接受报价后，继续确认并读取结果。
broadcast_job = broadcast_prepared.confirm()
broadcast_result = broadcast_job.result(timeout=300)
evs = broadcast_result[0].data.evs
stds = broadcast_result[0].data.stds
print(evs.shape) # (2, 2)
print("Z 在两个参数点的估计：", evs[0])
print("X 在两个参数点的估计：", evs[1])
print(stds)
["Z", "X"] 的形状为 (2,)，与两个参数点逐项对应；[["Z"], ["X"]] 增加观测量轴，使每个观测量分别作用于每个参数点。根据实验所需的对应关系选择形状。
9. 直接运行、设备选项和固定布局
程序已经决定执行时，可以用 run 合并 prepare 与 confirm。下面展示 Sampler 和 Estimator 的直接执行写法，按需要选择相应入口。
# Sampler 直接执行：
# sample_job = sampler.run([bell], shots=128)
# Estimator 直接执行：
# estimate_job = estimator.run([(state, observable)], precision=0.1)
| 设置 | 放在哪里 | 边界 |
|---|---|---|
| 默认采样数或估计设置 | create_sampler/create_estimator 的 options | Sampler 仅 default_shots；Estimator 使用 Qiskit BackendEstimatorV2 选项 |
| 设备专属参数 | configuration_data 或当前 schema 声明的单次关键字 | 通过 backend.configuration_schema 查看当前设备的选项 |
| 固定物理位置 | prepare/run 的 physical_qubits 或 initial_layout | 选择其中一个入口，或直接提交已有布局的线路 |
| 审价策略 | auto_approve_quote=False 或显式 prepare | run 不自动确认时抛 QuoteApprovalRequired，准备对象位于异常 prepared 属性 |
| 钱包 | 创建 QuantumSourceService 时的 currency | 选择新报价使用的 OPEN 或 FULL 钱包 |
固定两个物理位置时，先从设备能力中选择可用、直接连接且支持所需门的编号。例如 sampler.prepare([bell], shots=128, physical_qubits=[1, 11]) 将两个逻辑比特分别固定在物理 1、11。固定位置采用 mapping=False。Estimator 也接受布局参数，其测量基变换门同样按所选物理位置编译。
Primitive 的等待时间通过结果句柄的 result(timeout=...) 设置，具体用法如下。
10. 等待、超时与恢复原提交
| 操作 | 作用 | 后续步骤 |
|---|---|---|
| primitive_prepared.confirm() | 确认实验并启动本机后台结果处理 | 保存返回的任务句柄 |
| primitive_job.result(timeout=300) | 本次调用等待后台结果最多 300 秒 | 读取返回的 PrimitiveResult |
| 当前 result 调用超时 | 后台任务继续处理 | 稍后用原 job 再调用 result |
| primitive_prepared.recover() | 查询原已尝试提交，必要时恢复本机结果处理 | 取得任务句柄后继续读取 result |
PrimitiveJob 在本机后台等待平台结果并计算 BitArray、evs/stds。读取期间保持 Notebook 内核和 service 开启。关闭内核后，云端已接受任务继续处理；跨会话读取时，保留实验定义和任务标识，通过 Core 查询平台记录并下载原始结果。完整 PUB 结果整理依赖本次会话中的实验定义和本地处理。
# 只查询上面 Sampler 实验的状态；正常排队时 job_id() 可能尚无值。
print(sample_job.status())
print(sample_job.submission_ids)
print(sample_job.platform_job_ids)
# 如果仅是上一次 result 等待超时，可直接再次等待原 job：
# sample_result = sample_job.result(timeout=300)
# 若后台归约因原提交待核实等异常已经退出，可只读恢复：
# recovered = sample_prepared.recover()
# 检查 recovered 是任务句柄还是仍待核实的 PreparedSubmissionReceipt，
# 只有取得任务句柄后再调用 recovered.result(timeout=300)。
primitive_job.cancel() 向平台申请取消尚未领取的待发送任务。多任务批次可以分别成功或失败，通过取消报告查看各项结果；仍在处理的任务继续沿原句柄查询。
11. 读取设备提供的原生概率
标准 Sampler 通过 counts 构造 BitArray。对提供原生概率通道的设备，可使用 sampler.run_probabilities(...)，从 data.probabilities 中读取 QuasiDistribution。
# 使用设备的原生概率通道执行实验：
# probability_job = sampler.run_probabilities([bell], shots=128)
# probability_result = probability_job.result(timeout=300)
# distribution = probability_result[0].data.probabilities
# print(distribution)
单个参数点的 probabilities 是一个分布；参数网格对应相同形状的一组分布。选择 counts 结果时使用标准 prepare/run，选择设备原生概率结果时使用 run_probabilities。
12. 完成后关闭连接
# 仅在所选实验的结果等待与处理完成后运行。
service.close()
完成结果等待与读取后，调用 close 释放客户端连接。平台上的任务记录和结果继续保留。
API 参考与延伸阅读
Sampler、Estimator、准备对象和结果对象的签名、参数与返回值集中收录在 API 参考中。
下一步
下载本页 Jupyter Notebook
收录本页 Python 示例与说明，便于在 Jupyter Notebook 中学习和运行；安装命令保留为说明。