Pular para o conteúdo principal

Implante e execute um modelo de Qiskit Function para dinâmica hamiltoniana AQC + Trotter

Visão geral

Este é um modelo de Qiskit Function agnóstico em relação ao experimento para dinâmica hamiltoniana. Dado um Hamiltoniano de Pauli 1D com interação entre vizinhos mais próximos, um estado inicial preparado (opcional) e um conjunto de observáveis, ele executa a evolução temporal de Trotter, a compressão de circuito por compilação quântica aproximada (AQC) e a execução mitigada, retornando então a série temporal de cada observável. Troque a configuração (PRE) e a análise (POST) e o mesmo núcleo conduz um experimento diferente:

PRE (sua configuração)FUNCTION (implantada aqui)POST (sua análise)
Prepare um estado, como um circuito ou um estado produto, com um kick local opcionalSíntese de Trotter → compressão AQC → execução em statevector, fake, ou runtime, retornando O(t)\langle O \rangle(t)S(q,ω)S(q, \omega) para espalhamento de nêutrons, ou magnetização, transporte, dinâmica de quench, entre outros

O modelo é publicado no repositório de modelos de Qiskit Function, ao lado dos outros modelos de aplicação. Este notebook o implanta em sua própria conta do Qiskit Serverless. Execute-o uma vez, e qualquer notebook poderá então chamar a função com serverless.load("aqc-dynamics-function").

Para um exemplo científico completo, consulte Simule espalhamento de nêutrons com um workflow Serverless de dinâmica AQC + Trotter, que chama esta função para calcular o fator de estrutura dinâmica de KCuF3_3. Este notebook aborda, em vez disso, a implantação e o contrato de entrada.

Requisitos

Antes de começar, certifique-se de ter o seguinte no ambiente de kernel deste notebook:

  • Qiskit SDK v2.0 ou posterior (pip install qiskit).

  • O cliente Qiskit IBM Catalog (pip install qiskit-ibm-catalog), que implanta e executa cargas de trabalho no Qiskit Serverless.

As próprias dependências científicas da função (qiskit-addon-aqc-tensor, cotengrust, qiskit-aer) não precisam ser instaladas localmente.

Obtenha os arquivos de origem do modelo

A função é um pequeno pacote Python que o Qiskit Serverless executa na nuvem, então seu código-fonte precisa existir como arquivos locais que são enviados no momento da implantação. O pacote é publicado no repositório de modelos de Qiskit Function.

Baixe source_files

O download é um único zip, nomeado com o caminho completo do diretório no repositório:

qiskit-community qiskit-function-templates main physics aqc_trotter source_files.zip

  1. Descompacte-o no diretório que contém este notebook.

  2. Renomeie a pasta extraída desse nome longo para source_files.

Seu diretório de trabalho então ficará assim:

your-working-directory/
├── function-template-aqc-trotter.ipynb <- this notebook
└── source_files/ <- the renamed folder
├── __init__.py
├── program.py
└── source/
├── __init__.py
├── _serverless.py
├── app_function.py
├── aqc.py
├── build.py
├── execute.py
└── hamiltonian.py

O nome precisa ser exatamente source_files, pois é esse o working_dir que a Etapa 3 envia.

program.py é o ponto de entrada que o gateway invoca. Tudo em source/ é a implementação, dividida por etapa: síntese de Hamiltoniano e Trotter, compressão AQC e execução. Nada disso precisa ser editado para executar os exemplos a seguir. A Etapa 3 envia todo o diretório, então repita essa etapa sempre que você alterar um arquivo.

# Added by doQumentation — required packages for this notebook
!pip install -q numpy qiskit qiskit-ibm-catalog

1. Autenticação

Use qiskit-ibm-catalog para autenticar no QiskitServerless com sua chave de API (token) e CRN (instância), que você pode encontrar no painel do IBM Quantum® Platform. Com essas credenciais, você pode instanciar o cliente serverless localmente para enviar ou executar a função selecionada:

from qiskit_ibm_catalog import QiskitServerless
serverless = QiskitServerless(channel="ibm_quantum_platform", token="MY_TOKEN", instance="MY_CRN")

Você pode, opcionalmente, usar save_account() para salvar suas credenciais em seu ambiente local (consulte o guia Configure sua conta do IBM Cloud®). Observe que isso grava suas credenciais no mesmo arquivo que QiskitRuntimeService.save_account():

QiskitServerless.save_account(channel="ibm_quantum_platform", token="MY_TOKEN", instance="MY_CRN")

Se a conta estiver salva, não é necessário fornecer o token para autenticar:

from qiskit_ibm_catalog import QiskitServerless

# Authenticate to the remote cluster
# In this case, loading a saved account
serverless = QiskitServerless()

# REPLACE WITH YOUR OWN CREDENTIALS or SAVED ACCOUNT
# serverless = QiskitServerless(channel="ibm_quantum_platform", token="MY_TOKEN", instance="MY_CRN")

2. Declare dependências

Pacotes que a função precisa além da imagem serverless base gerenciada.

nota

O gateway só instala nomes que estão na sua lista de permissões (requirements-dynamic-dependencies.txt), correspondidos pelo nome do pacote e fixados na versão permitida com ==. Qualquer outra coisa precisa chegar de forma transitiva (como dependência de um pacote na lista de permissões). A sintaxe [extras] é respeitada: qiskit-addon-aqc-tensor[quimb-jax] é o que instala quimb e jax. cotengrust é necessário para eficiência de memória durante a simulação de rede tensorial. qiskit-aer é listado separadamente para o backend fake (simulação local com ruído).

DEPENDENCIES = [
"qiskit-addon-aqc-tensor[quimb-jax]==0.3.1",
"qiskit-aer==0.17.2",
"cotengrust==0.2.0",
]

3. Defina e envie a função

from qiskit_ibm_catalog import QiskitFunction

fn = QiskitFunction(
title="aqc-dynamics-function",
entrypoint="program.py",
working_dir="source_files/",
dependencies=DEPENDENCIES,
)
serverless.upload(fn)
QiskitFunction(aqc-dynamics-function)

4. Verifique se foi registrada

next(p for p in serverless.list() if p.title == "aqc-dynamics-function")
QiskitFunction(aqc-dynamics-function)

Referência da função

Esta é uma breve introdução. Cada campo é documentado por completo no README do modelo AQC Dynamics: a tabela completa de entradas com suas regras de validação, os campos de saída, os backends de execução e mais exemplos elaborados. O que segue é a versão resumida, suficiente para acompanhar os exemplos a seguir.

Entradas

Cada execução é uma única chamada fn.run(...). Apenas as três primeiras entradas na tabela são obrigatórias: hamiltonian, t_steps e aqc_segments. Tudo depois delas é opcional e usa o padrão mostrado, então uma chamada mínima tem três argumentos, e o restante da tabela é a funcionalidade que você pode escolher ativar. O num_qubits do Hamiltoniano define o comprimento da cadeia, então não há uma entrada separada para o tamanho.

InputPadrãoDescrição
hamiltonianobrigatórioHamiltoniano de Pauli 1D com interação entre vizinhos mais próximos, como um SparsePauliOp. As strings são operadores de Pauli, então não há fator implícito de um meio.
t_stepsobrigatórioTotal de passos de Trotter. Evolui até T = t_steps * dt e reporta cada observável em cada t_k = k * dt.
aqc_segmentsobrigatórioPlano de compressão: uma lista de {"n_steps": k, "ansatz_steps": m}. sum(n_steps) passos são comprimidos; o restante é executado como Trotter simples.
dt0.2Tempo físico avançado por um passo de Trotter.
initial_state|0...0>Um QuantumCircuit preparado para evoluir. Incorpore qualquer kick local neste circuito.
observablesZ por siteQualquer coisa que EstimatorV2 aceite como argumento observables. Um observável por coluna de saída.
trotter_optionsSuzuki de 2ª ordem{"method": ..., "synthesis_settings": {...}}. reps e time são de responsabilidade da função.
aqc_optionsver descriçãomax_bond (32), cutoff (1e-8), autodiff_backend ("jax"), fidelity_target (None), optimizer_settings (L-BFGS-B, jac=True, maxiter=300).
estimator_optionsDD, twirling, TREXEstimatorV2.options, passado como está. Um dicionário fornecido substitui totalmente os padrões, em vez de mesclá-los.
transpiler_options{"optimization_level": 3}Argumentos nomeados de generate_preset_pass_manager. backend e target são rejeitados, já que o caminho de execução é responsável por eles.
backend"runtime""statevector", "fake", ou "runtime".
backend_namemenos ocupadoNome do backend IBM® para runtime, ou um backend fake nomeado.
batches1Divide os circuitos entre N jobs de runtime. Um batch envia um único job e não cria sessão.
parallel_simFalseDistribui os caminhos do simulador local por todos os núcleos disponíveis com o Ray. Sem efeito em runtime.
return_circuitsFalseRetorna os circuitos lógicos AQC + Trotter no resultado, junto com a série de observáveis.

Backends de execução

Os três caminhos compartilham o mesmo código e as mesmas configurações de mitigação. Eles diferem apenas em onde os circuitos são executados.

backendO que éCredenciaisObservações
"statevector"StatevectorEstimator exatoApenas conta ServerlessO caminho de referência exato. Sem tempo de QPU.
"fake"Simulação local com ruído em um backend fake do QiskitApenas conta ServerlessUm ensaio fiel do caminho runtime mitigado. Requer qiskit-aer. Usa por padrão o fake_sherbrooke de 127 qubits.
"runtime" (padrão)O EstimatorV2 mitigado em uma QPU realConta Serverless e uma instância com acesso a QPUbackend_name opcional; omiti-lo seleciona o dispositivo menos ocupado.

Ambos os caminhos de simulador ainda chamam a função implantada, então precisam de uma conta Serverless salva, mesmo não usando tempo de QPU. Os dois exemplos a seguir executam a mesma carga de trabalho primeiro em statevector, depois em runtime.

Saída

job.result() retorna um dicionário simples:

{
"times": [...], # length t_steps + 1, t_k = k * dt (t=0 is the prepared state)
"expectation_values": [[...]], # shape (n_times, n_observables)
"observable_labels": [...], # for example: ["Z_0", "ZZ_0_1"]
"metadata": {
"n", "t_steps", "dt", "tier",
"aqc_compressed_steps": 5, # total compressed steps (= sum of segment n_steps)
"aqc_segments": [ # per segment: the plan plus its own results
{"n_steps": 3, "ansatz_steps": 1, "steps": [1, 2, 3], "n_params": 133,
"fidelities": {"1": ..., "2": ..., "3": ...}},
{"n_steps": 2, "ansatz_steps": 2, "steps": [4, 5], "n_params": 245,
"fidelities": {"4": ..., "5": ...}},
],
"execution_backend",
"aqc_fidelities": {"1": ..., "2": ...}, # flat per-step fidelity, all compressed steps
"circuit_stats": { # per-step 2q depth and gate count, full Trotter vs AQC
"1": {"full_trotter": {"depth_2q": ..., "num_2q_gates": ...},
"aqc_trotter": {"depth_2q": ..., "num_2q_gates": ...}},
"2": {...},
},
"warnings": [...], # non-fatal notices; for example, a cotengrust fallback
"resource_usage": { # per stage; QPU_TIME is the charged QPU time
"RUNNING: OPTIMIZING_FOR_HARDWARE": {"CPU_TIME": ...},
"RUNNING: WAITING_FOR_QPU": {"CPU_TIME": ...},
"RUNNING: EXECUTING_QPU": {"QPU_TIME": ...},
},
},
# present only when return_circuits=True
"circuits": [QuantumCircuit, ...], # one per evolved step; circuits[i] is at times[i + 1]
}

aqc_fidelities e circuit_stats são os dois primeiros a ler: juntos, eles informam se a compressão se manteve fiel e se realmente economizou profundidade. Em runtime, resource_usage reporta o tempo de espera na fila separadamente do tempo de QPU pelo qual você é cobrado. Uma entrada rejeitada falha rapidamente como um ServerlessError estruturado (código 4615).

Exemplo de simulador

Execute a função primeiro no backend exato statevector. Isso não gasta tempo de QPU e valida a implantação de ponta a ponta. O modelo aqui é uma cadeia de Ising de campo transverso de oito qubits, e observables é omitido, então a função mede o ZZ por site padrão.

O plano de compressão é a entrada que vale a pena entender. Cada segmento {"n_steps": k, "ansatz_steps": m} comprime k passos consecutivos de Trotter em um ansatz construído a partir de um alvo de Trotter de m passos, e quaisquer passos além de sum(n_steps) são executados como Trotter simples. Os passos iniciais, com baixo emaranhamento, comprimem bem em um ansatz raso de camada única; os passos posteriores, com mais emaranhamento, precisam de um mais profundo.

from qiskit.quantum_info import SparsePauliOp

fn = serverless.load("aqc-dynamics-function")

n = 8
H = SparsePauliOp.from_sparse_list(
[("ZZ", [i, i + 1], 1.0) for i in range(n - 1)]
+ [("X", [i], 0.8) for i in range(n)],
num_qubits=n,
)

job = fn.run(
t_steps=8,
aqc_segments=[
{
"n_steps": 4,
"ansatz_steps": 1,
}, # early steps -> shallow 1-layer ansatz
{
"n_steps": 2,
"ansatz_steps": 2,
}, # later steps -> deeper 2-layer ansatz
],
hamiltonian=H,
aqc_options={"max_bond": 32},
backend="statevector",
)
print("job ID:", job.job_id)
job ID: ee1f3793-e995-427d-81d1-5924549beb38

Acompanhe a execução e leia o resultado

status() reporta tanto o ciclo de vida geral do job quanto o subestado por etapa que a função publica conforme é executada. As mesmas etapas se aplicam à execução em hardware mais adiante neste guia:

QUEUED -> INITIALIZING -> RUNNING: OPTIMIZING_FOR_HARDWARE -> RUNNING: WAITING_FOR_QPU -> RUNNING: EXECUTING_QPU -> RUNNING: POST_PROCESSING -> DONE

Valor de status()Etapa
RUNNING: OPTIMIZING_FOR_HARDWAREpreparação de estado, construção de Trotter, compressão AQC
RUNNING: WAITING_FOR_QPUna fila na QPU (apenas backend runtime)
RUNNING: EXECUTING_QPUcircuitos em execução (simuladores locais marcam isso diretamente)
RUNNING: POST_PROCESSINGmontagem do dicionário de resultado

Os estados finais são DONE, ERROR e CANCELED. Esta execução em statevector não tem fila de QPU, então ela pula RUNNING: WAITING_FOR_QPU. Use job.logs() a qualquer momento para ver os logs por etapa, incluindo a fidelidade AQC alcançada em cada passo.

print(job.status()) # re-run until this reports DONE
DONE
import numpy as np

result = job.result()
ev = np.array(result["expectation_values"])

print("observables:", result["observable_labels"])
print("shape:", ev.shape, "-> (n_times, n_observables)")
print("first row (t = 0, the prepared state):", np.round(ev[0], 4))
print("last row (t = t_steps * dt):", np.round(ev[-1], 4))
print(
"AQC fidelities:",
{k: round(v, 4) for k, v in result["metadata"]["aqc_fidelities"].items()},
)

# What the compression bought: 2-qubit depth at the final time step.
stats = result["metadata"]["circuit_stats"][
str(result["metadata"]["t_steps"])
]
print(
"2q depth at the final step:",
stats["full_trotter"]["depth_2q"],
"(full Trotter) ->",
stats["aqc_trotter"]["depth_2q"],
"(AQC + Trotter)",
)
observables: ['Z_0', 'Z_1', 'Z_2', 'Z_3', 'Z_4', 'Z_5', 'Z_6', 'Z_7']
shape: (9, 8) -> (n_times, n_observables)
first row (t = 0, the prepared state): [1. 1. 1. 1. 1. 1. 1. 1.]
last row (t = t_steps * dt): [0.1442 0.2956 0.4686 0.4877 0.4869 0.4686 0.2963 0.1441]
AQC fidelities: {'1': 1.0, '2': 1.0, '3': 1.0, '4': 1.0, '5': 1.0, '6': 0.9999}
2q depth at the final step: 210 (full Trotter) -> 79 (AQC + Trotter)

Exemplo de hardware

Uma chamada de função com backend="runtime" faz o transpile e executa em um processador IBM Quantum real, com a mitigação de erros integrada da função: dynamical decoupling (XY4), gate twirling e twirled readout error extinction (TREX). backend_name seleciona o dispositivo; omita-o e a função escolherá o menos ocupado.

Nada no código científico muda. O que difere do exemplo de simulador é o comprimento da cadeia, o número de passos de Trotter, o plano de compressão, o backend e as configurações explícitas de mitigação abordadas na seção a seguir.

Dimensionando o job para o hardware de controle

estimator_options é a entrada que vale a pena configurar deliberadamente. O gate twirling constrói num_randomizations circuitos randomizados separados para cada PUB, e o job inteiro, cada PUB com todas as suas randomizações, precisa caber na memória de instruções do sistema de controle clássico da QPU. A função usa por padrão 1000 randomizações, então uma evolução de 10 passos envia 11 PUBs de 1000 circuitos cada: aproximadamente 11.000 instâncias de circuito em um único job.

Se exceder o que o sistema de controle comporta, o job falha com o erro 6073. Job limits fornece os limites e como contá-los, sendo o principal 26,8 milhões de instruções do sistema de controle por qubit, aplicado por job em vez de por PUB. O dynamical decoupling adiciona gates que contam para esse limite.

Duas entradas controlam o tamanho:

  • estimator_options define o orçamento de shots. O total de shots é num_randomizations * shots_per_randomization, então você pode trocar randomizações por shots por randomização, mantendo as estatísticas e ainda reduzindo o programa. A célula a seguir usa 100 randomizações com 200 shots cada, o que equivale a 20.000 shots por observável e cerca de um décimo das instâncias de circuito que os padrões enviariam. Consulte TwirlingOptions e Opções do Estimator para o conjunto completo de campos.

  • batches divide os PUBs entre esse número de jobs de runtime separados, o que é o remédio que o próprio erro 6073 sugere e por que o enquadramento por job importa. Definir batches=4 envia aproximadamente três PUBs por job, em vez de onze de uma vez, e os jobs saem juntos em um batch, de modo que o grupo entra na fila uma única vez, em vez de cada job entrar na fila separadamente.

Lembre-se de que um estimator_options fornecido substitui totalmente os padrões da função, em vez de mesclar-se a eles, então o dynamical decoupling e o TREX são reafirmados na célula a seguir para mantê-los ativados.

from qiskit.quantum_info import SparsePauliOp

fn = serverless.load("aqc-dynamics-function")

n = 10
H = SparsePauliOp.from_sparse_list(
[("ZZ", [i, i + 1], 1.0) for i in range(n - 1)]
+ [("X", [i], 0.8) for i in range(n)],
num_qubits=n,
)

job = fn.run(
t_steps=10,
aqc_segments=[
{
"n_steps": 3,
"ansatz_steps": 1,
}, # early steps -> shallow 1-layer ansatz
{
"n_steps": 3,
"ansatz_steps": 2,
}, # later steps -> deeper 2-layer ansatz
],
hamiltonian=H,
aqc_options={"max_bond": 32},
backend="runtime",
backend_name="ibm_marrakesh",
# The function defaults to 1000 twirling randomizations, which was too large
# for this device. Total shots is num_randomizations *
# shots_per_randomization, so this is 20,000 shots per observable.
estimator_options={
"dynamical_decoupling": {"enable": True, "sequence_type": "XY4"},
"twirling": {
"enable_gates": True,
"num_randomizations": 100,
"shots_per_randomization": 200,
},
"resilience": {"measure_mitigation": True},
},
)
print("job ID (save this to reconnect later):", job.job_id)
job ID (save this to reconnect later): 7229a8bf-9f83-4785-8dd4-489844abc2d9
Reconectando a um job de longa duração

Uma execução em hardware não é rápida, e a maior parte do tempo é clássica, e não na QPU. A compressão AQC é executada dentro da função antes que qualquer coisa chegue à QPU, e a fila da QPU se soma a isso. Você não precisa manter este notebook ou kernel aberto enquanto ele é executado.

Copie o ID do job impresso pela célula anterior e salve-o. As próximas três células permitem que você retome a execução mais tarde:

  1. Reconectar, necessário apenas em uma nova sessão de kernel: execute novamente a célula de Autenticação para recriar serverless, depois reconstrua o identificador job a partir do ID que você salvou. Pule esta célula se ainda estiver na sessão em que você enviou o job, pois o identificador já está ativo.

  2. Verificar status: execute novamente até que reporte DONE.

  3. Buscar o resultado: execute apenas quando o status for DONE.

Cole o ID salvo no lugar do placeholder na célula de reconexão a seguir.

# Reconnect to a previously submitted job by its ID. Only needed in a NEW kernel
# session; if you are still in the session where you submitted, the `job` handle
# from the preceding cell is already live, so skip this cell. Replace the ID that follows with your own.
job = serverless.get_job_by_id("<your job ID>")
# Re-run this until it reports DONE, then fetch the result in the following cell.
print(job.status())
DONE
import numpy as np

# Run this only once the preceding status cell reports DONE. result() blocks until
# the job finishes, so calling it earlier just waits.
result = job.result()
ev = np.array(result["expectation_values"])

print("backend:", result["metadata"]["execution_backend"])
print("shape:", ev.shape, "-> (n_times, n_observables)")
print("last row (t = t_steps * dt):", np.round(ev[-1], 4))
print(
"AQC fidelities:",
{k: round(v, 4) for k, v in result["metadata"]["aqc_fidelities"].items()},
)

# What the compression bought: 2-qubit depth at the final time step.
stats = result["metadata"]["circuit_stats"][
str(result["metadata"]["t_steps"])
]
print(
"2q depth at the final step:",
stats["full_trotter"]["depth_2q"],
"(full Trotter) ->",
stats["aqc_trotter"]["depth_2q"],
"(AQC + Trotter)",
)
backend: runtime
shape: (11, 10) -> (n_times, n_observables)
last row (t = t_steps * dt): [0.1504 0.1361 0.218 0.2144 0.2275 0.1783 0.1749 0.1599 0.0915 0.0922]
AQC fidelities: {'1': 1.0, '2': 1.0, '3': 1.0, '4': 1.0, '5': 0.9999, '6': 0.9999}
2q depth at the final step: 342 (full Trotter) -> 171 (AQC + Trotter)

Próximos passos

Recomendações