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 opcional | Síntese de Trotter → compressão AQC → execução em statevector, fake, ou runtime, retornando | 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 KCuF. 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
-
Descompacte-o no diretório que contém este notebook.
-
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.
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.
| Input | Padrão | Descrição |
|---|---|---|
hamiltonian | obrigatório | Hamiltoniano 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_steps | obrigatório | Total de passos de Trotter. Evolui até T = t_steps * dt e reporta cada observável em cada t_k = k * dt. |
aqc_segments | obrigatório | Plano 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. |
dt | 0.2 | Tempo físico avançado por um passo de Trotter. |
initial_state | |0...0> | Um QuantumCircuit preparado para evoluir. Incorpore qualquer kick local neste circuito. |
observables | Z por site | Qualquer coisa que EstimatorV2 aceite como argumento observables. Um observável por coluna de saída. |
trotter_options | Suzuki de 2ª ordem | {"method": ..., "synthesis_settings": {...}}. reps e time são de responsabilidade da função. |
aqc_options | ver descrição | max_bond (32), cutoff (1e-8), autodiff_backend ("jax"), fidelity_target (None), optimizer_settings (L-BFGS-B, jac=True, maxiter=300). |
estimator_options | DD, twirling, TREX | EstimatorV2.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_name | menos ocupado | Nome do backend IBM® para runtime, ou um backend fake nomeado. |
batches | 1 | Divide os circuitos entre N jobs de runtime. Um batch envia um único job e não cria sessão. |
parallel_sim | False | Distribui os caminhos do simulador local por todos os núcleos disponíveis com o Ray. Sem efeito em runtime. |
return_circuits | False | Retorna 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.
backend | O que é | Credenciais | Observações |
|---|---|---|---|
"statevector" | StatevectorEstimator exato | Apenas conta Serverless | O caminho de referência exato. Sem tempo de QPU. |
"fake" | Simulação local com ruído em um backend fake do Qiskit | Apenas conta Serverless | Um 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 real | Conta Serverless e uma instância com acesso a QPU | backend_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 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_HARDWARE | preparação de estado, construção de Trotter, compressão AQC |
RUNNING: WAITING_FOR_QPU | na fila na QPU (apenas backend runtime) |
RUNNING: EXECUTING_QPU | circuitos em execução (simuladores locais marcam isso diretamente) |
RUNNING: POST_PROCESSING | montagem 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_optionsdefine 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. -
batchesdivide 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. Definirbatches=4envia 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
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:
-
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 identificadorjoba 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. -
Verificar status: execute novamente até que reporte
DONE. -
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
-
Percorra Simule espalhamento de nêutrons com um workflow Serverless de dinâmica AQC + Trotter, o exemplo complementar que chama esta função implantada para calcular o fator de estrutura dinâmica de KCuF.
-
Leia o modelo AQC Dynamics no GitHub para o contrato completo de entrada e saída, mais exemplos e detalhes de citação.
-
Navegue pelo repositório de modelos de Qiskit Function para outros modelos de aplicação construídos da mesma forma.
-
Leia o guia do Qiskit Serverless para gerenciar funções implantadas.
-
Aprofunde-se na etapa de compressão AQC com a documentação do Qiskit addon: AQC-Tensor.