Upload qads/quantum/core.py

qads/quantum/core.py

@@ -0,0 +1,237 @@
+"""Quantum Decision Core - Main integration module."""
+import numpy as np
+from typing import Dict, Any, Optional, List, Tuple
+from dataclasses import dataclass
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class QuantumState:
+    """Represents a quantum-encoded state."""
+    amplitudes: np.ndarray
+    probabilities: np.ndarray
+    entropy: float
+    n_qubits: int
+    
+    @classmethod
+    def from_classical(cls, state_vector: np.ndarray, n_qubits: int) -> 'QuantumState':
+        """Encode classical state into quantum amplitudes."""
+        # Normalize
+        amplitudes = state_vector / np.linalg.norm(state_vector)
+        probabilities = np.abs(amplitudes) ** 2
+        entropy = -np.sum(probabilities * np.log2(probabilities + 1e-10))
+        return cls(amplitudes, probabilities, entropy, n_qubits)
+
+
+class QuantumDecisionCore:
+    """
+    Main quantum decision core integrating QAOA, VQC, and uncertainty analysis.
+    """
+    
+    def __init__(self, config: Any):
+        self.config = config
+        self.n_qubits = config.n_qubits
+        self.n_layers = config.n_layers
+        self.shots = config.shots
+        self.entropy_threshold = config.entropy_threshold
+        
+        # Sub-modules
+        self.qaoa = None
+        self.vqc = None
+        self.uncertainty_analyzer = None
+        self.kernel_attention = None
+        
+        # Metrics
+        self.metrics = {
+            'quantum_calls': 0,
+            'avg_optimization_time': 0.0,
+            'avg_entropy': 0.0,
+            'activation_count': 0
+        }
+        
+        self._initialize_modules()
+        
+    def _initialize_modules(self):
+        """Initialize quantum sub-modules."""
+        try:
+            from .qaoa import QAOAOptimizer
+            from .vqc import VariationalQuantumCircuit
+            from .uncertainty import QuantumUncertaintyAnalyzer
+            from .kernels import QuantumKernelAttention
+            
+            self.qaoa = QAOAOptimizer(self.config)
+            self.vqc = VariationalQuantumCircuit(self.config)
+            self.uncertainty_analyzer = QuantumUncertaintyAnalyzer(self.config)
+            self.kernel_attention = QuantumKernelAttention(self.config)
+            logger.info("Quantum modules initialized successfully")
+        except ImportError as e:
+            logger.warning(f"Quantum libraries not available: {e}")
+            
+    def encode_state(self, classical_state: np.ndarray) -> QuantumState:
+        """Encode classical state into quantum representation."""
+        # Pad or truncate to match qubit count
+        target_dim = 2 ** self.n_qubits
+        if len(classical_state) < target_dim:
+            padded = np.zeros(target_dim)
+            padded[:len(classical_state)] = classical_state
+            classical_state = padded
+        else:
+            classical_state = classical_state[:target_dim]
+            
+        return QuantumState.from_classical(classical_state, self.n_qubits)
+    
+    def optimize_path(self, 
+                     cost_matrix: np.ndarray,
+                     constraints: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        """Use QAOA to optimize path/cost function."""
+        if self.qaoa is None:
+            return self._classical_fallback(cost_matrix)
+            
+        self.metrics['quantum_calls'] += 1
+        result = self.qaoa.optimize(cost_matrix, constraints)
+        self._update_metrics(result)
+        return result
+    
+    def analyze_uncertainty(self, state: np.ndarray) -> Dict[str, float]:
+        """Analyze uncertainty using VQC."""
+        if self.uncertainty_analyzer is None:
+            return self._classical_uncertainty(state)
+            
+        result = self.uncertainty_analyzer.analyze(state)
+        self.metrics['avg_entropy'] = (
+            (self.metrics['avg_entropy'] * (self.metrics['quantum_calls'] - 1) + result['entropy']) /
+            self.metrics['quantum_calls']
+        )
+        return result
+    
+    def compute_attention(self, 
+                         query: np.ndarray, 
+                         keys: np.ndarray) -> np.ndarray:
+        """Compute quantum kernel attention."""
+        if self.kernel_attention is None:
+            return self._classical_attention(query, keys)
+            
+        return self.kernel_attention.compute(query, keys)
+    
+    def should_activate_quantum(self, world_state: Dict[str, Any]) -> bool:
+        """Determine if quantum computation should be activated."""
+        entropy = world_state.get('entropy', 0.0)
+        uncertainty = world_state.get('uncertainty', 0.0)
+        obstacle_density = world_state.get('obstacle_density', 0.0)
+        
+        # Activation logic based on complexity metrics
+        should_activate = (
+            entropy > self.config.activation_entropy or
+            uncertainty > 0.5 or
+            obstacle_density > 0.4
+        )
+        
+        if should_activate:
+            self.metrics['activation_count'] += 1
+            
+        return should_activate
+    
+    def evaluate_trajectories(self, 
+                            trajectories: List[np.ndarray],
+                            world_state: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """Evaluate multiple trajectories with quantum scoring."""
+        results = []
+        
+        for traj in trajectories:
+            # Encode trajectory
+            q_state = self.encode_state(traj.flatten())
+            
+            # Analyze uncertainty
+            uncertainty = self.analyze_uncertainty(traj.flatten())
+            
+            # Compute quantum score
+            quantum_score = self._compute_quantum_score(q_state, uncertainty, world_state)
+            
+            results.append({
+                'trajectory': traj,
+                'quantum_score': quantum_score,
+                'entropy': q_state.entropy,
+                'uncertainty': uncertainty,
+                'confidence': 1.0 - uncertainty.get('entropy', 0.0)
+            })
+            
+        return sorted(results, key=lambda x: x['quantum_score'], reverse=True)
+    
+    def _compute_quantum_score(self, 
+                              q_state: QuantumState,
+                              uncertainty: Dict[str, float],
+                              world_state: Dict[str, Any]) -> float:
+        """Compute composite quantum score for decision making."""
+        # Components
+        coherence = 1.0 - q_state.entropy / self.n_qubits  # Higher is better
+        confidence = 1.0 - uncertainty.get('entropy', 0.0)
+        
+        # Weighted combination
+        score = (
+            0.4 * coherence +
+            0.3 * confidence +
+            0.2 * (1.0 - world_state.get('risk_score', 0.0)) +
+            0.1 * (1.0 - world_state.get('obstacle_density', 0.0))
+        )
+        
+        return float(np.clip(score, 0.0, 1.0))
+    
+    def _classical_fallback(self, cost_matrix: np.ndarray) -> Dict[str, Any]:
+        """Classical fallback when quantum is unavailable."""
+        # Simple greedy optimization
+        n = cost_matrix.shape[0]
+        path = [0]
+        visited = {0}
+        
+        while len(path) < n:
+            current = path[-1]
+            next_node = min(
+                (i for i in range(n) if i not in visited),
+                key=lambda i: cost_matrix[current, i]
+            )
+            path.append(next_node)
+            visited.add(next_node)
+            
+        cost = sum(cost_matrix[path[i], path[i+1]] for i in range(n-1))
+        
+        return {
+            'path': path,
+            'cost': float(cost),
+            'quantum_used': False,
+            'iterations': 1
+        }
+    
+    def _classical_uncertainty(self, state: np.ndarray) -> Dict[str, float]:
+        """Classical uncertainty estimation fallback."""
+        probs = np.abs(state) ** 2
+        probs = probs / (probs.sum() + 1e-10)
+        entropy = -np.sum(probs * np.log2(probs + 1e-10))
+        
+        return {
+            'entropy': float(entropy),
+            'confidence': float(1.0 - entropy / np.log2(len(state))),
+            'risk_score': float(np.std(state)),
+            'quantum_used': False
+        }
+    
+    def _classical_attention(self, query: np.ndarray, keys: np.ndarray) -> np.ndarray:
+        """Classical dot-product attention fallback."""
+        scores = np.dot(keys, query)
+        scores = scores / np.sqrt(len(query))
+        exp_scores = np.exp(scores - np.max(scores))
+        attention = exp_scores / exp_scores.sum()
+        return attention
+    
+    def _update_metrics(self, result: Dict[str, Any]):
+        """Update internal metrics."""
+        if 'optimization_time' in result:
+            n = self.metrics['quantum_calls']
+            self.metrics['avg_optimization_time'] = (
+                (self.metrics['avg_optimization_time'] * (n - 1) + result['optimization_time']) / n
+            )
+    
+    def get_metrics(self) -> Dict[str, float]:
+        """Return quantum computation metrics."""
+        return self.metrics.copy()