扩展量子后端

通过 runtime.register_quantum_backend()，第三方库或用户代码可以提供量子执行能力。CPU 模拟器与已注册的 Provider 使用相同的电路处理、结果校验和生命周期管理。

注册与创建

from pivotq.providers import BackendCapabilities

runtime.register_quantum_backend(
    "my-device", MyProvider,
    capabilities=BackendCapabilities(max_qubits=device_qubit_limit, is_simulated=False),
)
backend = runtime.quantum_backend("my-device", endpoint="device-address")

MyProvider 是部署方提供或用户实现的类，device_qubit_limit 来自实际设备能力。endpoint 仅示范该类可接收的配置项；实际配置参数由 Provider 定义。SDK 不会只凭名称或地址自动适配任意 QPU。

执行时通过 MyProvider(**config) 构造实例。注册与创建后端时均不会创建 MyProvider 实例，也不会连接设备；执行进程首次调用工厂函数时才创建实例。配置须可序列化，并在创建后端时生成独立快照。

注册属于当前 Runtime，名称不能重复或覆盖已有后端。backend.describe() 返回注册时提供的静态能力，不探测设备。

Provider 协议

接口	约定
`run(request: QuantumRequest)`	同步返回 `ProviderResult`
`close()`	幂等释放客户端资源；无资源时可为空方法
`BackendCapabilities`	max_qubits、min_qubits、is_simulated、supports_seed
`QuantumRequest`	circuit、shots、seed、request_id
`ProviderResult`	shots、source、可选 counts/probabilities、metadata

输入电路是已绑定参数、去掉末尾测量的独立副本。Provider 负责把它编译到设备门集并读取所有逻辑量子位。结果位串必须为 q[n-1]…q0，不含寄存器分隔空格；设备路由后的物理布局和厂商位序由 Provider 还原。

SDK 保存用户原始测量映射，统一处理部分测量、位重排和未写入的经典位。Provider 不需要重复实现这些逻辑。首版不支持中途测量、动态控制、reset 或 initialize。

结果校验与故障

counts 与 probabilities 至少提供一种，具体类型取决于后端实际提供的数据；计数总和必须等于设备确认的 shots，概率必须有效且总和为 1。两者同时返回时必须一致。只提供概率时，用户得到的 QuantumResult.counts 为 None。SDK 不补造原始计数，也不静默归一化错误分布。

source 明确填写结果来源。metadata 可含 backend_job_id、device_id，不能覆盖 source、request_id、backend、is_simulated、bit_order，也不能包含不可序列化对象或凭据。

确认请求尚未提交时，输入问题使用 pivotq.errors.ValidationError；请求可能已经执行但响应丢失时，使用 ResultUnknownError 并保留设备作业标识。真实设备 Provider 的未分类异常或无效结果也按“执行状态未知”处理；供核实的信息会随依赖关系传递给后续 CPU 任务。SDK 不自动重试或回退模拟器。

Task 与 Actor

模拟 Provider 默认使用 Task，每次调用都会创建并关闭实例，申请 1 CPU；可设置 execution="actor" 保留状态，默认串行。num_cpus 可调整模拟 Provider 的 CPU 请求。

真实设备 Provider 使用固定的串行 Actor、0 CPU 资源请求。串行约束作用于同一个后端对象；多个客户端对同一设备的全局排队仍由设备服务负责。

可运行的离线扩展教程

python packages/framework/examples/custom_backend.py
python packages/framework/examples/custom_backend.py --executor ray

教程中的程序注册用户实现的 CPU 量子模拟器，验证部分测量与结果依赖，全程不访问真实设备。完整教程代码如下：

"""Register an offline quantum provider: python custom_backend.py --executor ray.

The same provider can be implemented in a separate installed Python package.
No private PivotQ interfaces are needed, and no physical device is contacted.
"""

import argparse
import json

from pivotq import Runtime
from pivotq.providers import BackendCapabilities, ProviderResult, QuantumRequest


class CustomStatevectorProvider:
    def __init__(self, *, source="example_statevector_samples"):
        self.source = source

    def run(self, request: QuantumRequest) -> ProviderResult:
        from qiskit.quantum_info import Statevector

        # PivotQ supplies a bound, measurement-free copy. Return all logical
        # qubits in q[n-1]...q0 order; PivotQ restores the user's measurements.
        state = Statevector.from_instruction(request.circuit)
        state.seed(request.seed)
        counts = state.sample_counts(shots=request.shots)
        return ProviderResult(
            shots=request.shots,
            source=self.source,
            counts={str(key): int(count) for key, count in counts.items()},
        )

    def close(self):
        # Close an HTTP client or vendor SDK session here if one is used.
        pass


def build_circuit():
    from pivotq import QuantumCircuit

    circuit = QuantumCircuit(3, 2)
    circuit.x(0)
    circuit.h(2)
    circuit.measure(0, 1)
    circuit.measure(2, 0)
    return circuit


def summarize(result):
    return {
        "backend": result.backend,
        "is_simulated": result.is_simulated,
        "counts": result.counts,
        "probabilities": result.probabilities,
        "shots": result.shots,
        "bit_order": result.bit_order,
        "source": result.metadata["source"],
    }


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--executor", choices=("local", "ray"), default="local")
    parser.add_argument("--shots", type=int, default=1024)
    parser.add_argument("--seed", type=int, default=7)
    args = parser.parse_args()
    with Runtime(executor=args.executor) as runtime:
        runtime.register_quantum_backend(
            "custom-simulator", CustomStatevectorProvider,
            capabilities=BackendCapabilities(max_qubits=16, is_simulated=True, supports_seed=True),
        )
        backend = runtime.quantum_backend("custom-simulator", source="example_statevector_samples")
        circuit = runtime.submit(build_circuit)
        measurement = backend.submit(circuit, shots=args.shots, seed=args.seed)
        summary = runtime.submit(summarize, measurement)
        print(json.dumps(runtime.get(summary), ensure_ascii=False, sort_keys=True))
        runtime.release(summary, measurement, circuit)


if __name__ == "__main__":
    main()