【代码贡献】优化量子核矩阵与量子距离计算：去除采样噪声、复用模拟器并修复多处缺陷

pyqpanda-algorithm/pyqpanda_alg/QKmeans/QuantumKmeans.py

@@ -13,13 +13,28 @@
 import numpy as np
 from copy import deepcopy
 from numpy import pi
-from pyqpanda3.core import CPUQVM, QCircuit, QProg, SWAP, U3, H, measure, draw_qprog
+from pyqpanda3.core import CPUQVM, QCircuit, QProg, SWAP, U3, H
 
 
 
 
-def _QuantumKmeansCircuit(theta0, phi0, theta, phi):
-    machine = CPUQVM()
+def _QuantumKmeansCircuit(theta0, phi0, theta, phi, machine=None):
+    """Swap-test circuit returning the ancilla probabilities ``{'0': .., '1': ..}``.
+
+    The probability of the ancilla being ``1`` is the normalised squared
+    distance between the two encoded states.
+
+    Two issues made this helper unnecessarily expensive, since ``fit`` calls it
+    once per (sample, centroid) pair on every iteration:
+
+    * a fresh ``CPUQVM`` was built on every call (now optional and shareable);
+    * ``draw_qprog(prog)`` was called unconditionally, which *builds a text
+      diagram of the circuit and throws it away* -- its return value was never
+      used. It has been removed from the hot path.
+
+    The program is also simulated without a measurement instruction, so the
+    ancilla probability is exact instead of a 1024-shot estimate.
+    """
     prog = QProg(3)
     qlist = prog.qubits()
     cir = QCircuit()
@@ -29,30 +44,29 @@ def _QuantumKmeansCircuit(theta0, phi0, theta, phi):
     cir << SWAP(qlist[0], qlist[1]).control(qlist[2])
     cir << H(qlist[2])
     prog << cir
-    prog << measure(qlist[2], qlist[2])
-
-    draw_qprog(prog)
 
-    machine.run(prog, 1024)
+    if machine is None:
+        machine = CPUQVM()
+    machine.run(prog, 1)
     result = machine.result().get_prob_dict([qlist[2]])
     return result
 
 
-def _point_centroid_distances(point, centroids, k):
+def _point_centroid_distances(point, centroids, k, machine=None):
     xval = [point[0]]
     for i in range(k):
-      xval.append(centroids[i][0])
+        xval.append(centroids[i][0])
     yval = [point[1]]
     for i in range(k):
-      yval.append(centroids[i][1])
+        yval.append(centroids[i][1])
 
     theta_t = [((x + 1) * pi / 2) for x in xval]
     theta_c = [((x + 1) * pi / 2) for x in yval]
 
     results_list = []
 
     for i in range(1, k + 1):
-        result = _QuantumKmeansCircuit(theta_c[0], theta_t[0], theta_c[i], theta_t[i])
+        result = _QuantumKmeansCircuit(theta_c[0], theta_t[0], theta_c[i], theta_t[i], machine)
         results_list.append(result['1'] if '1' in result else 0)
     return results_list
 
@@ -175,10 +189,13 @@ def fit(self, data):
 
         iter_counter = 1
         clusters = None
+        # One simulator reused across every distance evaluation in every
+        # iteration (was: a fresh CPUQVM per sample/centroid pair).
+        machine = CPUQVM()
         while abs(error - upper_error) > self.tol:
             centers = centers_new
 
-            distances = np.array(list(map(lambda x: _point_centroid_distances(x, centers, self.K), data)))
+            distances = np.array(list(map(lambda x: _point_centroid_distances(x, centers, self.K, machine), data)))
 
             clusters = np.argmin(distances, axis=1)
 

pyqpanda-algorithm/pyqpanda_alg/QSVM/quantum_kernel_svm.py

@@ -11,7 +11,7 @@
 # limitations under the License.
 
 import numpy as np
-from pyqpanda3.core import CPUQVM, QCircuit, QProg, CNOT, U1, U2, measure
+from pyqpanda3.core import CPUQVM, QCircuit, QProg, CNOT, U1, U2
 
 
 from ..plugin import *
@@ -21,6 +21,21 @@
 
 
 def _build_circuit(qlist, n_qbits, weights_x, weights_y):
+    """Build the kernel circuit for the ``n_qbits``-qubit feature map.
+
+    .. note::
+
+        The entangling blocks below are hard-coded to a single ``CNOT`` between
+        ``qlist[0]`` and ``qlist[1]``, so only ``n_qbits == 2`` is meaningful.
+        Previously any other value silently produced a wrong kernel matrix;
+        now it fails loudly.
+    """
+    if n_qbits != 2:
+        raise ValueError(
+            f"_build_circuit only supports n_qbits=2, got {n_qbits!r}. "
+            "The entangling block is hard-coded to CNOT(qlist[0], qlist[1])."
+        )
+
     circuit = QCircuit()
     for i in range(n_qbits):
         circuit << U2(qlist[i], 0, np.pi)
@@ -50,17 +65,31 @@ def _build_circuit(qlist, n_qbits, weights_x, weights_y):
     return circuit
 
 
-def _run_circuit(n_qbits, weights_x, weights_y):
-    machine = CPUQVM()
+def _run_circuit(n_qbits, weights_x, weights_y, machine=None):
+    """Evaluate the kernel circuit and return the *exact* output probabilities.
+
+    Performance notes
+    -----------------
+    Two things used to make this helper the hot spot when building an
+    ``N x N`` kernel matrix:
+
+    1. A brand new ``CPUQVM`` was constructed on every call (``N*N`` times).
+    2. The program contained a ``measure`` instruction, so the probabilities
+       were **shot-sampled** (1024 shots -> ~3% relative noise) rather than
+       computed exactly. The kernel matrix therefore changed between runs.
+
+    The program is now simulated without any measurement instruction, so the
+    state-vector simulator returns the exact amplitudes/probabilities, and a
+    caller-supplied ``machine`` can be reused across the whole matrix.
+    """
     prog = QProg(n_qbits)
     qubits = prog.qubits()
-    circuit = QCircuit()
-    circuit << _build_circuit(qubits, n_qbits, weights_x, weights_y)
-    prog << circuit
-    prog << measure(qubits, qubits)
-    machine.run(prog, 1024)
-    result = machine.result().get_counts()
-    return result
+    prog << _build_circuit(qubits, n_qbits, weights_x, weights_y)
+
+    if machine is None:
+        machine = CPUQVM()
+    machine.run(prog, 1)
+    return machine.result().get_prob_dict(qubits)
 
 
 class QuantumKernel_vqnet:
@@ -306,6 +335,12 @@ def qsvm_classification():
                     qsvm_classification()
 
         """
+        if self._n_qbits != 2:
+            raise ValueError(
+                f"QuantumKernel_vqnet only supports n_qbits=2, got {self._n_qbits!r}. "
+                "Pass n_qbits explicitly, e.g. QuantumKernel_vqnet(n_qbits=2)."
+            )
+
         if not isinstance(x_vec, np.ndarray):
             x_vec = np.asarray(x_vec)
         if y_vec is not None and not isinstance(y_vec, np.ndarray):
@@ -350,10 +385,11 @@ def qsvm_classification():
             mus = np.asarray(mus.flat)
             nus = np.asarray(nus.flat)
 
-        is_statevector_sim = False
-        measurement = not is_statevector_sim
         measurement_basis = "0" * self._n_qbits
 
+        # One simulator reused for the whole matrix (was: one per matrix entry).
+        machine = CPUQVM()
+
         for idx in range(0, len(mus), self._batch_size):
             to_be_computed_data_pair = []
             to_be_computed_index = []
@@ -362,28 +398,39 @@ def qsvm_classification():
                 j = nus[sub_idx]
                 x_i = x_vec[i]
                 y_j = y_vec[j]
-                if not np.all(x_i == y_j):
+                if np.all(x_i == y_j):
+                    # Identical feature vectors encode to the same state, so the
+                    # fidelity is exactly 1. The symmetric branch pre-fills the
+                    # diagonal, but the asymmetric branch used to leave these
+                    # entries at 0 -- record them explicitly for both cases.
+                    kernel[i, j] = 1.0
+                    if is_symmetric:
+                        kernel[j, i] = 1.0
+                else:
                     to_be_computed_data_pair.append((x_i, y_j))
                     to_be_computed_index.append((i, j))
 
             matrix_elements = []
             for x, y in to_be_computed_data_pair:
-                result = _run_circuit(self._n_qbits, x, y)
-                try:
-                    counts = result[measurement_basis]
-                    states = np.sum(list(result.values()))
-                    probability = counts / states
-                except:
-                    probability = 0.0001
-                matrix_elements.append(probability)
+                probabilities = _run_circuit(self._n_qbits, x, y, machine=machine)
+                matrix_elements.append(float(probabilities.get(measurement_basis, 0.0)))
 
             for (i, j), value in zip(to_be_computed_index, matrix_elements):
                 kernel[i, j] = value
                 if is_symmetric:
                     kernel[j, i] = kernel[i, j]
 
         if is_symmetric:
-            D, U = np.linalg.eig(kernel)
-            kernel = U @ np.diag(np.maximum(0, D)) @ U.transpose()
+            # ``kernel`` is real symmetric: use ``eigh`` (real spectrum, ascending)
+            # instead of ``eig``, which could return a complex dtype.
+            D, U = np.linalg.eigh(kernel)
+            kernel = U @ np.diag(np.maximum(0.0, D)) @ U.T
+            kernel = np.real_if_close(kernel)
+
+        # Exact fidelities live in [0, 1]; clip away floating-point overshoot and
+        # restore the unit diagonal that a quantum kernel must satisfy.
+        kernel = np.clip(kernel, 0.0, 1.0)
+        if is_symmetric:
+            np.fill_diagonal(kernel, 1.0)
 
         return kernel

pyqpanda-algorithm/pyqpanda_alg/QSVR/QSVR.py

@@ -12,10 +12,8 @@
 
 from pyqpanda3.core import CPUQVM, QCircuit, QProg, RX, RY, CZ, H
 import numpy as np
-from sklearn.model_selection import train_test_split
 from sklearn.preprocessing import StandardScaler
 from sklearn.svm import SVR
-from sklearn.metrics import mean_squared_error, r2_score
 import matplotlib.pyplot as plt
 from sklearn.decomposition import PCA
 
@@ -108,34 +106,72 @@ def cir_real(qb, cs):
             Qcir << RX(qb[i], cs[n - 1 - i])
         return Qcir
 
-    def dist(self, x, y):
-        machine = CPUQVM()
+    def dist(self, x, y, machine=None):
+        """Fidelity ``|<psi(x)|psi(y)>|^2`` between the two encoded states.
+
+        The circuit carries **no measurement instruction**, so the state-vector
+        simulator returns the exact probability instead of a shot-sampled
+        frequency. A ``machine`` may be supplied so that an entire kernel
+        matrix is built with a single simulator instead of one per entry.
+        """
+        if machine is None:
+            machine = CPUQVM()
         prog = QProg(2)
         qv = prog.qubits()
         prog << self.cir_real(qv, x) << self.cir_real(qv, y).dagger()
-        machine.run(prog, 1000)
+        machine.run(prog, 1)
         re = machine.result().get_prob_dict(qv)
         re = parse_quantum_result_dict(re, qv, select_max=-1)['0' * 2]
         return re
 
     def k_kernel(self, X, Y):
-        matrix = np.zeros((len(X), len(Y)))
-        for i in range(len(X)):
-            for j in range(len(Y)):
-                matrix[i][j] = self.dist(X[i], Y[j])
+        """Build the quantum kernel matrix between ``X`` and ``Y``.
+
+        When ``X`` and ``Y`` hold the same samples the matrix is symmetric, so
+        only the upper triangle is simulated and mirrored. This is the case
+        scikit-learn uses during ``fit``, so roughly half of the circuit
+        simulations are avoided there.
+        """
+        X = np.asarray(X)
+        Y = np.asarray(Y)
+        n, m = len(X), len(Y)
+        matrix = np.zeros((n, m))
+
+        # Single simulator shared by every entry of the matrix.
+        machine = CPUQVM()
+
+        if n == m and np.array_equal(X, Y):
+            for i in range(n):
+                matrix[i, i] = 1.0
+                for j in range(i + 1, m):
+                    value = self.dist(X[i], Y[j], machine)
+                    matrix[i, j] = value
+                    matrix[j, i] = value
+        else:
+            for i in range(n):
+                for j in range(m):
+                    matrix[i, j] = self.dist(X[i], Y[j], machine)
+
         return matrix
 
-    def get_res(self):
+    def _fit(self):
+        """Fit the SVR on the training data (shared by ``get_res``/``show_res``)."""
         svr = SVR(kernel=self.k_kernel, gamma=0.1)
         svr.fit(self.x, self.y)
+        return svr
+
+    def get_res(self):
+        svr = self._fit()
         y_ppp = svr.predict(self.x)
         return y_ppp, self.y
 
     def show_res(self):
-        svr = SVR(kernel=self.k_kernel, gamma=0.1)
-        svr.fit(self.x, self.y)
-        x0_test = np.linspace(min(self.x[:, 0]), max(self.x[:, 1]), 30)
-        x1_test = np.linspace(min(self.x[:, 0]), max(self.x[:, 1]), 30)
+        svr = self._fit()
+        # NOTE: each axis must be sampled over its own range. The original code
+        # used ``max(self.x[:, 1])`` for *both* axes, so the plotted surface was
+        # computed on the wrong grid.
+        x0_test = np.linspace(min(self.x[:, 0]), max(self.x[:, 0]), 30)
+        x1_test = np.linspace(min(self.x[:, 1]), max(self.x[:, 1]), 30)
         X0_test, X1_test = np.meshgrid(x0_test, x1_test)
         X_test = np.c_[X0_test.ravel(), X1_test.ravel()]
         y_pred = svr.predict(X_test).reshape(X0_test.shape)