Skip to main content
IBM Quantum Platform

MultiStagePassManager

class qiskit.passmanager.MultiStagePassManager(**stages)

GitHub

Bases: Generic[IR, IR_OUT]

Um gerenciador de passagens em etapas compatível com vários IRs.

Este gerenciador de passagens executa etapas sequenciais e nomeadas no programa de entrada. Um estágio pode ser definido como Task, um iterável dele ou como um BasePassManager. Se um for BasePassManager definido como estágio, apenas as tarefas nele contidas serão executadas; as conversões de entrada e saída definidas por seus _passmanager_backend métodos _passmanager_frontend e _não_ serão aplicadas.

As etapas podem:

  • manter o IR, por exemplo, se estiver definido como BasePassManager[IR] ou Task[IR, IR], ou
  • reduzir o IR, por exemplo, para Task[IR1, IR2].

É de responsabilidade do usuário configurar os estágios de forma compatível, de modo que a resposta de saída (IR) do estágio atual corresponda à resposta de entrada (IR) do próximo estágio. Os nomes das etapas e a ordem em que são executadas são definidos no momento da criação do gerenciador de passagens. A implementação de uma etapa pode ser modificada atribuindo-se uma nova implementação ao atributo do objeto correspondente.

Se for fornecida uma função de retorno de chamada ao run() método, ela deve ser capaz de lidar com o fato de que sua entrada IR pode ser de qualquer tipo gerado pelas tarefas no gerenciador de passagens.

A função de retorno é chamada com a seguinte assinatura:

def callback(
    task: Task,  # the executed task
    passmanager_ir: Any,  # the IR after the task execution
    property_set: PropertySet,  # the property set after execution
    running_time: float,  # the time the task ran
    count: int  # the number of executed tasks so far
):
    ...

Todos os argumentos são passados como argumentos de palavra-chave.

Nota

Embora o objeto Task defina a interface da tarefa, as passagens personalizadas devem derivar apenas da classe base GenericPass. A base Task é uma interface interna, e versões futuras do Qiskit podem impor mais restrições aos tipos disponíveis de Task.

Um exemplo de fluxo de trabalho é:

from qiskit.circuit import QuantumCircuit
from qiskit.dagcircuit import DAGCircuit
from qiskit.passmanager import GenericPass, MultiStagePassManager
from qiskit.transpiler import generate_preset_pass_manager, Target, CouplingMap

class CustomPauliIR:
    # A custom IR of global Pauli strings
    def __init__(self, num_qubits):
        self.num_qubits = num_qubits
        self.paulis = []

    def apply(self, pauli: str):
        assert len(pauli) == self.num_qubits
        self.paulis.append(pauli)

class CustomPauliOptimization(GenericPass[CustomPauliIR, CustomPauliIR]):
    # A pass run on the custom Pauli IR
    def run(self, passmanager_ir: CustomPauliIR) -> CustomPauliIR:
        to_remove = []
        for i, pauli in enumerate(passmanager_ir.paulis):
            if all(p == "I" for p in pauli):
                to_remove.append(i)

        for i in reversed(to_remove):
            del passmanager_ir.paulis[i]

        return passmanager_ir

class PauliToDAG(GenericPass[CustomPauliIR, DAGCircuit]):
    # A pass converting CustomPauliIR to DAGCircuit
    def run(self, passmanager_ir: CustomPauliIR) -> DAGCircuit:
        circuit = QuantumCircuit(passmanager_ir.num_qubits)
        for pauli in passmanager_ir.paulis:
            circuit.pauli(pauli, circuit.qubits)
        return circuit.to_dag()

def callback(task, passmanager_ir, property_set, running_time, count):
    if isinstance(passmanager_ir, CustomPauliIR):
        print("PauliIR:", task.__class__.__name__, passmanager_ir.paulis)
    else:
        print("DAGCircuit:", task.__class__.__name__, passmanager_ir.count_ops())

target = Target.from_configuration(
    basis_gates=["u", "cx"], coupling_map=CouplingMap.from_line(3)
)
multi_pm = MultiStagePassManager(
    pauli_opt=CustomPauliOptimization(),
    pauli_to_dag=PauliToDAG(),
    dag_opt=generate_preset_pass_manager(target=target),
)

program = CustomPauliIR(3)
program.apply("XYZ")
program.apply("III")
program.apply("ZZI")

out = multi_pm.run(program, callback=callback)
print(out.count_ops())

Essa classe está relacionada a StagedPassManager na medida em que ambas possuem um modelo de execução em etapas. O StagedPassManager, no entanto, permite DAGCircuit apenas como seu IR, possui conversões implícitas de e para um QuantumCircuit nos níveis de entrada e saída, e possui ganchos implícitos pre_* de e post_* nos estágios nos quais é possível gravar.

A lógica de execução de um StagedPassManager é, grosso modo, equivalente a:

from qiskit.circuit import QuantumCircuit
from qiskit.dagcircuit import DAGCircuit
from qiskit.passmanager import GenericPass, MultiStagePassManager
from qiskit.transpiler import TranspileLayout

class CircuitToDAG(GenericPass[QuantumCircuit, DAGCircuit]):
    def run(self, passmanager_ir: QuantumCircuit) -> DAGCircuit:
        self.property_set["original_qubit_indices"] = {
            bit: i for i, bit in enumerate(passmanager_ir.qubits)
        }
        self.property_set["num_input_qubits"] = passmanager_ir.num_qubits
        return passmanager_ir.to_dag()

class DAGToCircuit(GenericPass[DAGCircuit, QuantumCircuit]):
    def run(self, passmanager_ir: DAGCircuit) -> QuantumCircuit:
        qc = passmanager_ir.to_circuit(copy_operations=False)
        qc._layout = TranspileLayout.from_property_set(passmanager_ir, self.property_set)
        return qc

multi_pm = MultiStagePassManager(
    input=CircuitToDAG(),
    # ... stages of StagedPassManager ...
    output=DAGToCircuit()
)

input_circuit = QuantumCircuit(1)
output_circuit = multi_pm.run(input_circuit)
Aviso

O modelo de execução atual lineariza a passagem em um FlowControllerLinear para executar as tarefas. Esse modelo subjacente está sujeito a alterações, e não é seguro basear-se nessa suposição. As interfaces públicas dessa classe, no entanto, são estáveis.

Parâmetros

stages (BasePassManager[Any] | Task[Any, Any] | Iterable[Task[Any, Any]]) –

As etapas como gerentes de passagem. Elas serão executadas na ordem indicada e devem ter IRs compatíveis.

Os nomes das etapas são definidos pelo construtor; não é possível adicionar novas etapas posteriormente, mas é possível substituir a implementação de cada etapa reatribuindo-a ao seu atributo.


Atributos

stages

Os nomes artísticos. Esses são imutáveis.

Os próprios estágios podem ser modificados por meio de uma gravação no atributo que possui o mesmo nome do estágio.


Métodos

run

run(in_programs, callback=None, *, property_set=None)

GitHub

Execute o gerenciador de passagens em um conjunto de programas de entrada.

Parâmetros

  • in_programs (IR | list[IR] | tuple[IR]) – Os programas nos quais o gerenciador de senhas deve ser executado.
  • callback (Callable[[Task, Any, PropertySet, float, int], None] | None) – Uma função de retorno de chamada passada para cada tarefa individual.
  • property_set (PropertySet | None) – Uma propriedade opcional definida para ser passada ao gerenciador de passagens. Isso será alterado diretamente, caso seja fornecido. Isso não pode ser usado com o operador “multiple” em programas.

Retorna

Os programas de saída.

Tipo de retorno

IR_OUT | Iterable [ IR_OUT ]

to_flow_controller

to_flow_controller()

GitHub

Converta esse gerenciador de passagem em múltiplas etapas em um controlador de fluxo linear.

Essa conversão normaliza esse gerenciador de passagem em um Task[IR, IR_OUT] e permite que ele seja aninhado dentro de um MultiStagePassManager ou de outros fluxos de execução.

Tipo de retorno

Flow[* ControllerLinearIR*, IR_OUT]

Esta página foi útil?
Relate um bug, erro de digitação ou solicite conteúdo no GitHub.