CheckPoint#
什么是 CheckPoint?#
CheckPoint 是 qpanda3_runtime 中的一个实用工具,用于在量子计算任务中保存和恢复状态。想象一下你在玩一个游戏时可以随时保存进度,CheckPoint 就是量子计算中的”保存点”功能。
为什么需要 CheckPoint?#
在量子计算中,许多任务需要长时间运行,比如:
变分量子算法(VQE、QAOA)的迭代优化
大规模量子电路的采样任务
复杂量子实验的多次尝试
使用 CheckPoint 可以:
防止数据丢失:任务意外中断时可以从检查点恢复
方便调试:保存中间状态便于分析问题
优化资源:可以暂停任务,稍后继续执行
记录进度:保存任务执行的历史记录
快速开始#
让我们通过一个最简单的例子快速了解 CheckPoint 的基本用法:
# 导入必要的模块
from qpanda3_runtime import CheckPoint
# 创建 CheckPoint 实例
checkpoint = CheckPoint()
# 保存一个简单的检查点
task_state = {"task_id": "my_task", "status": "running", "progress": 0.5}
user_data = {"algorithm": "test", "iteration": 1}
saved_file = checkpoint.save(
filepath="my_first_checkpoint.json",
task_state=task_state,
user_data=user_data
)
print(f"检查点已保存到: {saved_file}")
# 加载检查点
loaded_data = checkpoint.load(saved_file)
print(f"加载的任务状态: {loaded_data['task_state']}")
print(f"加载的用户数据: {loaded_data['user_data']}")
这个简单的例子展示了 CheckPoint 的核心功能:保存和恢复数据。接下来我们将详细学习如何使用这些功能。
基本使用#
1. 保存检查点#
保存检查点就像给任务拍一张快照。你需要提供两种信息:
from qpanda3_runtime import CheckPoint
checkpoint = CheckPoint()
# 任务状态:描述任务当前状态的数据
task_state = {
"task_id": "quantum_simulation_001",
"status": "completed",
"result": {"energy": -1.234, "fidelity": 0.95},
"circuit_info": {"qubits": 5, "depth": 10}
}
# 用户数据:你自定义的附加信息
user_data = {
"experiment_name": "H2_molecule_VQE",
"parameters": [0.1, 0.2, 0.3, 0.4],
"iteration": 50,
"timestamp": "2026-03-30T10:30:00",
"notes": "第50次迭代,接近收敛"
}
# 保存检查点
filepath = checkpoint.save(
filepath="vqe_iteration_50.json", # 可选,不指定会自动生成
task_state=task_state,
user_data=user_data
)
print(f"检查点保存成功: {filepath}")
重要提示:
task_state可以是字典,也可以是具有get_task_state()方法的对象user_data可以是任何 JSON 可序列化的 Python 对象如果不指定
filepath,CheckPoint 会自动生成一个基于时间戳的文件名
2. 加载检查点#
加载检查点可以恢复之前保存的状态:
from qpanda3_runtime import CheckPoint
checkpoint = CheckPoint()
try:
# 加载检查点
data = checkpoint.load("vqe_iteration_50.json")
print("=== 检查点加载成功 ===")
# 访问不同类型的数据
meta = data['meta'] # 元数据(版本、创建时间)
task_state = data['task_state'] # 任务状态
user_data = data['user_data'] # 用户数据
print(f"版本: {meta.get('version')}")
print(f"创建时间: {meta.get('created_at')}")
print(f"任务ID: {task_state.get('task_id')}")
print(f"实验名称: {user_data.get('experiment_name')}")
print(f"迭代次数: {user_data.get('iteration')}")
# 使用恢复的数据
recovered_params = user_data.get('parameters', [])
print(f"恢复的参数: {recovered_params}")
except FileNotFoundError:
print("错误: 检查点文件不存在")
except Exception as e:
print(f"加载失败: {e}")
3. 处理版本兼容性#
CheckPoint 会自动检查版本兼容性。如果遇到版本不匹配,可以使用 export_data() 方法:
from qpanda3_runtime import CheckPoint
checkpoint = CheckPoint()
try:
# 尝试加载(可能因为版本不匹配而失败)
data = checkpoint.load("old_checkpoint.json")
except Exception as e:
print(f"加载失败: {e}")
print("尝试导出数据(不检查版本)...")
# 导出数据,不进行版本检查
exported = checkpoint.export_data("old_checkpoint.json")
print(f"导出的元数据: {exported['meta']}")
print(f"导出的用户数据: {exported['user_data']}")
实际应用场景#
场景1:量子迭代优化中的检查点#
在变分量子算法(如 VQE)中,通常需要多次迭代优化参数。使用 CheckPoint 可以保存每次迭代的状态:
import numpy as np
from qpanda3_runtime import RuntimeService, CheckPoint
from pyqpanda3.vqcircuit import VQCircuit
from pyqpanda3.core import RX,RY,RZ
def run_vqe_with_checkpoints():
"""在 VQE 算法中使用检查点"""
# 初始化服务
service = RuntimeService()
service.login('your_valid_api_key')
# 创建量子电路
vqc = VQCircuit()
vqc.set_Param([2]) # 2个参数
vqc << RX(0, vqc.Param([0])) << RY(1, vqc.Param([1]))
# 初始参数
params = np.array([0.14, 1.13])
# 初始化 CheckPoint
checkpoint = CheckPoint()
print("=== 开始 VQE 优化 ===")
with service.vqsession(
vqcircuit=vqc,
device=service.device('WK_C180'),
shots=100,
life_time=300
) as vqs:
for iteration in range(10): # 迭代10次
# 运行量子任务
vqtask = vqs.run_vqtask(params, [0, 1])
result = vqtask.get_result_sync()
# 保存检查点
checkpoint.save(
filepath=f"vqe_iteration_{iteration:03d}.json",
task_state=vqtask.get_task_state(),
user_data={
'iteration': iteration,
'parameters': params.tolist(),
'energy': calculate_energy(result),
'timestamp': time.time()
}
)
print(f"迭代 {iteration}: 参数={params}, 能量={calculate_energy(result)}")
# 更新参数(简化示例)
params = params * 0.9 # 简单的参数更新
print("=== 优化完成 ===")
# 从检查点恢复特定迭代
print("\n=== 从检查点恢复 ===")
iteration_to_recover = 5
recovered_data = checkpoint.load(f"vqe_iteration_{iteration_to_recover:03d}.json")
recovered_params = np.array(recovered_data['user_data']['parameters'])
print(f"恢复第{iteration_to_recover}次迭代的参数: {recovered_params}")
def calculate_energy(result):
"""简化能量计算(实际中会更复杂)"""
return -1 # 示例计算
if __name__ == "__main__":
import time
run_vqe_with_checkpoints()
场景2:采样任务中的检查点#
对于长时间运行的采样任务,可以使用 CheckPoint 保存中间结果:
from qpanda3_runtime import RuntimeService, CheckPoint
from pyqpanda3.core import QProg, X, CNOT, measure
def sample_task_with_checkpoint():
"""采样任务中使用检查点"""
service = RuntimeService()
service.login('your_valid_api_key')
# 创建量子程序
prog = QProg()
prog << X(0) << CNOT(0, 1) << measure(0, 0) << measure(1, 1)
# 提交采样任务
task = service.sample(
circuits=[prog.originir()],
device=service.device('WK_C180'),
shots=1000
)
checkpoint = CheckPoint()
# 保存任务状态
checkpoint.save(
filepath="sampling_task.json",
task_state=task.get_task_state(),
user_data={
'circuit': 'Bell state',
'shots': 1000,
'device': 'WK_C180',
'description': '贝尔态采样任务'
}
)
print("采样任务检查点已保存")
# 稍后可以恢复任务
recovered_task = service.recover_qtask_manager("sampling_task.json")
print(f"恢复的任务ID: {recovered_task.id()}")
if __name__ == "__main__":
sample_task_with_checkpoint()
场景3:保存复杂的实验配置#
你可以使用 CheckPoint 保存完整的实验配置和执行历史:
from qpanda3_runtime import CheckPoint
from datetime import datetime
import numpy as np
def save_complete_experiment():
"""保存完整的实验配置和结果"""
checkpoint = CheckPoint()
# 复杂的实验配置
experiment_config = {
'algorithm': 'QAOA',
'problem': {
'type': 'maxcut',
'graph': [[0, 1], [1, 2], [2, 0], [0, 3], [1, 3]],
'weights': [1.0, 2.0, 1.5, 0.8, 1.2]
},
'parameters': {
'p': 3, # QAOA 层数
'beta': [0.1, 0.2, 0.3],
'gamma': [0.5, 0.6, 0.7],
'optimizer': 'COBYLA',
'max_iterations': 100
},
'hardware': {
'device': '72',
'shots': 5000,
'calibration_date': '2026-03-01'
}
}
# 执行历史
execution_history = []
for i in range(5):
execution_history.append({
'iteration': i,
'energy': -2.5 + i * 0.1, # 模拟能量改进
'parameters': {'beta': 0.1 * i, 'gamma': 0.5 * i},
'timestamp': datetime.now().isoformat(),
'duration_seconds': 30.5 + i * 5
})
# 性能指标
metrics = {
'total_execution_time': 180.5,
'average_iteration_time': 36.1,
'convergence_rate': 0.92,
'memory_usage_mb': 128.7,
'success_rate': 0.98
}
# 保存检查点
checkpoint.save(
filepath=f"qaoa_experiment_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json",
task_state={
'experiment_id': 'qaoa_maxcut_001',
'status': 'completed',
'final_energy': execution_history[-1]['energy'],
'iterations_completed': len(execution_history)
},
user_data={
'config': experiment_config,
'history': execution_history,
'metrics': metrics,
'notes': '5次迭代后的QAOA实验结果'
}
)
print("完整实验检查点已保存")
if __name__ == "__main__":
save_complete_experiment()
最佳实践#
1. 文件命名和组织#
为检查点文件使用有意义的名称,便于管理:
# 好的命名方式
checkpoint.save(
filepath=f"vqe_h2_iteration_{iteration:03d}_{timestamp}.json",
...
)
# 按项目组织文件
project_name = "quantum_chemistry"
experiment_id = "h2_bond_length_scan"
checkpoint.save(
filepath=f"checkpoints/{project_name}/{experiment_id}/iteration_{iteration}.json",
...
)
2. 定期保存策略#
根据任务特点制定合适的保存频率:
def iterative_algorithm(checkpoint_interval=10):
"""定期保存检查点的迭代算法"""
checkpoint = CheckPoint()
for iteration in range(100):
# 执行迭代...
# 定期保存检查点
if iteration % checkpoint_interval == 0:
checkpoint.save(
filepath=f"iteration_{iteration}.json",
task_state=current_state,
user_data={'iteration': iteration}
)
print(f"已保存第{iteration}次迭代的检查点")
# 重要里程碑时额外保存
if is_important_milestone(iteration):
checkpoint.save(
filepath=f"milestone_{iteration}.json",
task_state=current_state,
user_data={'milestone': get_milestone_name(iteration)}
)
3. 版本管理#
处理软件版本更新时的兼容性问题:
def load_with_version_check(filepath):
"""带版本检查的加载函数"""
checkpoint = CheckPoint()
try:
# 尝试正常加载
return checkpoint.load(filepath)
except Exception as e:
print(f"加载失败: {e}")
# 导出数据进行分析
exported = checkpoint.export_data(filepath)
old_version = exported['meta']['version']
current_version = checkpoint.version
print(f"版本不匹配: {old_version} -> {current_version}")
# 根据版本差异执行迁移
if old_version == "1.0.0" and current_version == "1.1.0":
return migrate_from_1_0_to_1_1(exported)
else:
raise Exception(f"不支持的版本迁移: {old_version} -> {current_version}")