Tracker#
What is Tracker?#
Tracker is a tool package used for tracking and analyzing the execution process of quantum programs. It helps you record the execution status of quantum computing tasks, including execution time, parameter changes, result data, and other information, enabling you to better understand and optimize quantum programs.
Getting Started with Tracker#
Import Required Modules#
In your Python code, import the following modules:
from qpanda3_runtime import RuntimeService, Tracker
First Tracker Example#
Let’s start with a simple example to understand how to use Tracker to track quantum operations:
# Initialize RuntimeService
service = RuntimeService(auto_track=True, trace_record_file='./trace_record.jsonl')
service.login('your_valid_api_key')
# Use context manager for tracking
with Tracker(service, name="quantum_operation") as t:
# Record data
t.record({"operation": "quantum_circuit", "qubit_count": 2})
# Here you can execute your quantum operations
# ...
# Record result
t.record({"result": "success"})
This example demonstrates the basic usage of Tracker:
Enable auto-tracking when initializing RuntimeService
Use
withstatement to create a tracking contextRecord data within the context
Three Usage Modes of Tracker#
1. Context Manager Mode#
This is the most commonly used mode, using with statement to automatically manage the start and end of tracking:
with Tracker(service, name="my_operation") as t:
t.record({"step": "start", "params": [0.1, 0.2]})
# Execute operation
t.record({"step": "completed", "result": 0.5})
2. Decorator Mode#
Use @tracker decorator to automatically track function execution:
from qpanda3_runtime import tracker
@tracker(service, data={"function_name": "my_function"})
def my_function():
# Function logic
return "result"
# Function will be tracked automatically when called
result = my_function()
3. Manual Tracking Mode#
If you need more fine-grained control, you can manually manage the tracking process:
t = Tracker(service)
t.start("manual_operation")
t.record({"step": "start"})
# Execute operation
t.record({"step": "end", "result": "completed"})
t.end({"status": "success"})
Tips for Recording Data#
Adding Labels#
Add labels to records for easier subsequent analysis:
with Tracker(service, name="quantum_operation") as t:
t.record({"operation": "circuit_execution"}, labels=["quantum", "circuit"])
# Execute operation
t.record({"result": "success"}, labels=["quantum", "result"])
Recording Execution Time#
Record the execution time of operations:
import time
start_time = time.time()
with Tracker(service, name="time_tracking") as t:
# Execute operation
time.sleep(1)
end_time = time.time()
t.record({"execution_time": end_time - start_time})
Nested Tracking#
Track complex execution flows:
with Tracker(service, name="outer_operation") as outer:
outer.record({"step": "start"})
with Tracker(service, name="inner_operation") as inner:
inner.record({"step": "processing"})
# Execute inner operation
inner.record({"step": "completed"})
outer.record({"step": "completed"})
Practical Application: Error Debugging and Task Recovery#
2. View Tracking Records with Label “ERROR”#
import os
import json
from pyqpanda3.core import QProg, X, CNOT, measure, CZ, H
from qpanda3_runtime import RuntimeService, Tracker
def retrieve_error_message():
# Load tracking records
trace_file = r'./trace_record260330.jsonl'
# Open file, find lines with 'ERROR' in labels
with open(trace_file, 'r', encoding='utf-8') as file:
for line in file:
# Retrieve corresponding record lines through labels
if 'ERROR' in line:
# Convert to json dictionary
record = json.loads(line)
# Check again if 'ERROR' is in labels
if 'ERROR' not in record['labels']:
continue
print(record)
if __name__ == '__main__':
retrieve_error_message()
Output result:
{
"row_id": 12,
"row_type": "CONTEXT_END",
"row_parent_id": 10,
"data": {
"error": "{'position': 'runtime_server[qcloud]', 'error_code': 400, 'error_info': 'resource is null', 'error_type': 'InvalidRequest', 'trace_record': ['qcloud_code: 10002', 'obj_str: {\"code\":10002,\"message\":\"resource is null\",\"obj\":null,\"success\":false}'], 'help_info': ''}",
"task_metadata": {
"type": "SampleQTaskInfo",
"qtask_type": "sample",
"circuits": {
"type": "List[QProgUnionType]",
"data": [
"QINIT 2\nCREG 2\nX q[0]\nCNOT q[0],q[1]\nMEASURE q[0],c[0]\nMEASURE q[1],c[1]\n",
"QINIT 3\nCREG 3\nCZ q[1],q[2]\nMEASURE q[2],c[2]\n",
"QINIT 3\nCREG 3\nH q[1]\nCNOT q[1],q[2]\nMEASURE q[2],c[2]\n"
]
},
"device": {
"type": "QDevice",
"basic_info": {
"channel": "qcloud",
"chip_id": "20",
"chipId": 20,
"chipName": "硅基量子计算机",
"chipNameEn": "Silicon Quantum Computer",
"machineType": 5,
"nameDetail": "\"Silicon Quantum Computer\"",
"onlineFlag": true
},
"config_info": {}
},
"measure_qubits": null,
"specified_block": null,
"shots": 10,
"is_amend": true,
"is_mapping": true,
"is_optimization": true,
"point_label": 0,
"task_describe": ""
}
},
"labels": ["ERROR"],
"timestamp": "2026-03-30T14:04:41.274114"
}
Note: This is the content of line 12 in file trace_record260330.jsonl, which is a json string.
3. View Error Messages During Program Execution#
import os
import json
from pyqpanda3.core import QProg, X, CNOT, measure, CZ, H
from qpanda3_runtime import RuntimeService, Tracker
def retrieve_error_notification():
# Load tracking records
trace_file = r'./trace_record260330.jsonl'
# Open file, find lines with 'ERROR' in labels
with open(trace_file, 'r', encoding='utf-8') as file:
for line in file:
# Retrieve corresponding record lines through labels
if 'ERROR' in line:
# Convert to json dictionary
record = json.loads(line)
# Check again if 'ERROR' is in labels
if 'ERROR' not in record['labels']:
continue
error_info = record['data']['error']
# 'error_info': 'resource is null', indicates the corresponding chip is unavailable
print('error_info:',error_info)
if __name__ == '__main__':
retrieve_error_notification()
Execution result:
error_info: {'position': 'runtime_server[qcloud]', 'error_code': 400, 'error_info': 'resource is null', 'error_type': 'InvalidRequest', 'trace_record': ['qcloud_code: 10002', 'obj_str: {"code":10002,"message":"resource is null","obj":null,"success":false}'], 'help_info': ''}
Note:
This is the error field content from line 12 of file
trace_record260330.jsonlThis error message was originally generated by the ‘qcloud’ access channel, with error code 400 and error message “resource is null”, meaning the corresponding chip is unavailable
4. Recover the Actual Interface Parameters Passed When Submitting the Task#
# Auto-work feature demonstration
import os
import json
from pyqpanda3.core import QProg, X, CNOT, measure, CZ, H
from qpanda3_runtime import RuntimeService, Tracker
from qpanda3_runtime.task.qtask_info import SampleQTaskInfo,EstimateQTaskInfo
def reproduce_function_input_parameters_when_task_was_submitted():
# Load tracking records
service = RuntimeService()
service.login('your_valid_api_key')
# Define retrieval function
trace_file = r'./trace_record260330.jsonl'
# Open file, find lines with 'ERROR' in labels
with open(trace_file, 'r', encoding='utf-8') as file:
for line in file:
# Retrieve corresponding record lines through labels
if 'ERROR' in line:
# Convert to json dictionary
record = json.loads(line)
# Check again if 'ERROR' is in labels
if 'ERROR' not in record['labels']:
continue
task_metadata = record['data']['task_metadata']
if task_metadata['qtask_type'] == 'sample':
task_info:SampleQTaskInfo = SampleQTaskInfo.from_json_dict(service,task_metadata)
circuits, device, measure_qubits, specified_block, shots, is_amend, is_mapping, is_optimization, point_label, task_describe = task_info.input_for_service_sample()
print('circuits:',circuits)
print('device:',device)
print('measure_qubits:',measure_qubits)
print('specified_block:',specified_block)
print('shots:',shots)
print('is_amend:',is_amend)
print('is_mapping:',is_mapping)
print('is_optimization:',is_optimization)
print('point_label:',point_label)
print('task_describe:',task_describe)
else:
task_info:EstimateQTaskInfo = EstimateQTaskInfo.from_json_dict(service,task_metadata)
circuit_with_observable, device, specified_block, shots, is_amend, is_mapping, is_optimization, task_describe = task_info.input_for_service_estimate()
print('circuit_with_observable:',circuit_with_observable)
print('device:',device)
print('specified_block:',specified_block)
print('shots:',shots)
print('is_amend:',is_amend)
print('is_mapping:',is_mapping)
print('is_optimization:',is_optimization)
print('task_describe:',task_describe)
if __name__ == '__main__':
reproduce_function_input_parameters_when_task_was_submitted()
Execution result:
circuits: [<pyqpanda3.core.core.QProg object at 0x00000263B5265030>, <pyqpanda3.core.core.QProg object at 0x00000263B53133F0>, <pyqpanda3.core.core.QProg object at 0x00000263B5313430>]
device: {'channel': 'qcloud', 'chip_id': '20', 'chipId': 20, 'chipName': '硅基量子计算机', 'chipNameEn': 'Silicon Quantum Computer', 'machineType': 5, 'nameDetail': '"Silicon Quantum Computer"', 'onlineFlag': True}
measure_qubits: None
specified_block: None
shots: 10
is_amend: True
is_mapping: True
is_optimization: True
point_label: 0
task_describe:
5. Change to Available Computing Device and Resubmit Task#
import os
import json
from pyqpanda3.core import QProg, X, CNOT, measure, CZ, H
from qpanda3_runtime import RuntimeService, Tracker
from qpanda3_runtime.task.qtask_info import SampleQTaskInfo,EstimateQTaskInfo
def change_device_and_resubmit_task():
# Load tracking records
service = RuntimeService(auto_track=True, trace_record_file='./trace_record_4.jsonl')
service.login('your_valid_api_key')
# Define retrieval function
trace_file = r'./trace_record260330.jsonl'
# Open file, find lines with 'ERROR' in labels
with open(trace_file, 'r', encoding='utf-8') as file:
for line in file:
# Retrieve corresponding record lines through labels
if 'ERROR' in line:
# Convert to json dictionary
record = json.loads(line)
# Check again if 'ERROR' is in labels
if 'ERROR' not in record['labels']:
continue
task_metadata = record['data']['task_metadata']
if task_metadata['qtask_type'] == 'sample':
task_info:SampleQTaskInfo = SampleQTaskInfo.from_json_dict(service,task_metadata)
circuits, device, measure_qubits, specified_block, shots, is_amend, is_mapping, is_optimization, point_label, task_describe = task_info.input_for_service_sample()
task = service.sample(circuits, service.device('72'), measure_qubits, specified_block, shots, is_amend, is_mapping, is_optimization, point_label)
else:
task_info:EstimateQTaskInfo = EstimateQTaskInfo.from_json_dict(service,task_metadata)
circuit_with_observable, device, specified_block, shots, is_amend, is_mapping, is_optimization, task_describe = task_info.input_for_service_estimate()
task = service.estimate(circuit_with_observable, service.device('72'), specified_block, shots, is_amend, is_mapping, is_optimization)
res = task.get_result_sync()
print('res:',res)
if __name__ == '__main__':
change_device_and_resubmit_task()
Execution result:
res: [[{'key': ['0x3'], 'value': [1]}, {'key': ['0x0', '0x1'], 'value': [0.9, 0.1]}, {'key': ['0x0', '0x1'], 'value': [0.5, 0.5]}]]
Note: After changing to an available device, the computing task was submitted successfully and executed to obtain computing results.
Practical Application: Using Tracker in Variational Computing Tasks#
Variational Quantum Computing is an important quantum computing paradigm, usually involving iterative optimization of parameterized quantum circuits. Tracker can well track the execution process of such complex tasks.
Using Function Decorator to Track Functions#
1. Simple Decorator Usage: Track Function Name and Return Value#
service = RuntimeService(auto_track=False, trace_record_file='./trace_record_vqtask.jsonl')
service.login('your_valid_api_key')
@tracker(service)
def update_gate_params(old_params):
# Used to simulate parameter update, can be replaced with actual parameter update logic
return old_params * 2
2. Decorator Usage with TrackerRef: Record Additional Data Inside Function#
service = RuntimeService(auto_track=False, trace_record_file='./trace_record_vqtask.jsonl')
service.login('your_valid_api_key')
ref = TrackerRef()
@tracker(service, ref)
def check_for_stop_iter(params, count):
# Simulate checking whether to stop iteration
if count >= 2:
ref.record({"count": count}, labels=["ITERATION"])
return True
return False
Using with Syntax to Track Arbitrary Code Blocks#
In variational quantum computing tasks, usually need to track the entire iteration process:
service = RuntimeService(auto_track=False, trace_record_file='./trace_record_vqtask.jsonl')
service.login('your_valid_api_key')
with Tracker(service, name="vqsession") as t:
with service.vqsession(
vqcircuit=vqc,
device=service.device('180'),
shots=10,
life_time=300
) as vqs:
while True: # Simulate iteration process
vqtask = vqs.run_vqtask(params, measure_qubits)
task_id = vqtask.id()
print(f'task_id:{task_id}')
res = vqtask.get_result_sync()
print(f'res:{res}')
t.record({"task_id": task_id, "result": res}, labels=["VQTASK", "RESULT"])
count += 1
if check_for_stop_iter(params, count):
print('Iteration Completed')
break
else:
params = update_gate_params(params)
Complete Example: Variational Quantum Computing Task Tracking#
Below is a complete variational quantum computing task example, demonstrating how to comprehensively use the above technologies:
import os
from pyqpanda3.vqcircuit import VQCircuit
from pyqpanda3.core import *
import numpy as np
from qpanda3_runtime import RuntimeService, Tracker, tracker, TrackerRef
# Create quantum circuit
vqc = VQCircuit()
vqc.set_Param([2])
vqc << RX(0, vqc.Param([0])) << RY(1, vqc.Param([1]))
# Initialize parameters
init_params = np.array([0.14, 1.13])
service = RuntimeService(auto_track=False, trace_record_file='./trace_record_vqtask.jsonl')
service.login('your_valid_api_key')
@tracker(service)
def update_gate_params(old_params):
# Used to simulate parameter update, can be replaced with actual parameter update logic
return old_params * 2
ref = TrackerRef()
@tracker(service, ref)
def check_for_stop_iter(params, count):
# Simulate checking whether to stop iteration
if count >= 2:
ref.record({"count": count}, labels=["ITERATION"])
return True
return False
measure_qubits = [0, 1]
params = init_params
count = 0
with Tracker(service, name="vqsession") as t:
with service.vqsession(
vqcircuit=vqc,
device=service.device('72'),
shots=10,
life_time=300
) as vqs:
while True: # Simulate iteration process
vqtask = vqs.run_vqtask(params, measure_qubits)
task_id = vqtask.id()
print(f'task_id:{task_id}')
res = vqtask.get_result_sync()
print(f'res:{res}')
t.record({"task_id": task_id, "result": res}, labels=["VQTASK", "RESULT"])
count += 1
if check_for_stop_iter(params, count):
print('Iteration Completed')
break
else:
params = update_gate_params(params)
Execution Result and Record Analysis#
After executing the above code, the tracking records are written to file ./trace_record_vqtask.jsonl. The following is the analysis of key records:
1. Call Record of check_for_stop_iter Function#
First call (count=1, returns False):
{"row_id":4,"row_type":"CONTEXT_START","row_parent_id":1,"data":{"node_name":"check_for_stop_iter"},"labels":[],"timestamp":"2026-03-30T16:31:17.958228"}
{"row_id":5,"row_type":"DATA","row_parent_id":4,"data":null,"labels":[],"timestamp":"2026-03-30T16:31:17.958228"}
{"row_id":6,"row_type":"CONTEXT_END","row_parent_id":4,"data":{"result":"False"},"labels":[],"timestamp":"2026-03-30T16:31:17.958228"}
Second call (count=2, returns True):
{"row_id":11,"row_type":"CONTEXT_START","row_parent_id":1,"data":{"node_name":"check_for_stop_iter"},"labels":[],"timestamp":"2026-03-30T16:31:17.958228"}
{"row_id":12,"row_type":"DATA","row_parent_id":11,"data":null,"labels":[],"timestamp":"2026-03-30T16:31:17.958228"}
{"row_id":13,"row_type":"DATA","row_parent_id":11,"data":{"count":2},"labels":["ITERATION"],"timestamp":"2026-03-30T16:31:17.958228"}
{"row_id":14,"row_type":"CONTEXT_END","row_parent_id":11,"data":{"result":"True"},"labels":[],"timestamp":"2026-03-30T16:31:17.958228"}
Note: During the second call, {"count": 2} data was recorded through ref.record(), and ["ITERATION"] label was added.
2. Call Record of update_gate_params Function#
{"row_id":7,"row_type":"CONTEXT_START","row_parent_id":1,"data":{"node_name":"update_gate_params"},"labels":[],"timestamp":"2026-03-30T16:31:17.958228"}
{"row_id":8,"row_type":"DATA","row_parent_id":7,"data":null,"labels":[],"timestamp":"2026-03-30T16:31:17.958228"}
{"row_id":9,"row_type":"CONTEXT_END","row_parent_id":7,"data":{"result":"[0.28 2.26]"},"labels":[],"timestamp":"2026-03-30T16:31:17.958228"}
The updated parameters are [0.28 2.26].
3. Record of t.record()#
{"row_id":10,"row_type":"DATA","row_parent_id":1,"data":{"task_id":"E4190A1832FC564D94C7E225CE82C6EB","result":[[{"key":["0x0","0x2","0x3"],"value":[0.1,0.8,0.1]}]]},"labels":["VQTASK","RESULT"],"timestamp":"2026-03-30T16:31:17.958228"}
This record contains the task ID and computing result, with ["VQTASK", "RESULT"] labels added.
Summary#
Through this example, we can see the powerful application of Tracker in variational quantum computing tasks:
Decorator mode can automatically track function calls and return values
TrackerRef allows recording additional data inside decorated functions
with syntax can track complex code blocks and iteration processes
Label system facilitates classifying and filtering records
These features make Tracker a powerful tool for debugging and optimizing variational quantum computing tasks.