Spaces:
Configuration error
Configuration error
File size: 7,070 Bytes
50cdd10 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 | # -*- coding: utf-8 -*-
"""
Backend Selector Module
Intelligent backend selection based on edit characteristics
"""
import numpy as np
from typing import Dict, List, Optional
class PerformancePredictor:
"""Predict backend performance for given edit"""
def __init__(self):
"""Initialize performance predictor"""
# Performance profiles for each backend
self.profiles = {
'russian': {
'cyrillic_boost': 0.05,
'base_success': 0.873,
'base_latency': 85,
'domain_strengths': {'scientific': 0.03, 'math': 0.02}
},
'ibm': {
'cyrillic_boost': 0.0,
'base_success': 0.891,
'base_latency': 92,
'domain_strengths': {'code': 0.03, 'text': 0.02}
}
}
def predict_performance(
self,
backend: str,
edit_features: Dict
) -> Dict[str, float]:
"""
Predict performance for backend on edit
Args:
backend: Backend name
edit_features: Edit characteristics
Returns:
Predicted performance metrics
"""
profile = self.profiles.get(backend, self.profiles['ibm'])
# Base success rate
success_rate = profile['base_success']
# Language boost
if edit_features.get('lang') == 'ru' and backend == 'russian':
success_rate += profile['cyrillic_boost']
# Domain boost
domain = edit_features.get('domain', 'text')
if domain in profile['domain_strengths']:
success_rate += profile['domain_strengths'][domain]
# Latency
latency = profile['base_latency']
# Add noise
success_rate += np.random.normal(0, 0.01)
latency += np.random.normal(0, 5)
return {
'predicted_success_rate': min(1.0, max(0.0, success_rate)),
'predicted_latency_ms': max(10, latency),
'confidence': 0.85
}
class BackendSelector:
"""Select optimal backend for edit"""
def __init__(
self,
backends: List[str] = None,
selection_strategy: str = 'performance'
):
"""
Initialize backend selector
Args:
backends: Available backends
selection_strategy: Strategy ('performance', 'latency', 'balanced')
"""
self.backends = backends or ['russian', 'ibm']
self.selection_strategy = selection_strategy
self.predictor = PerformancePredictor()
self.selection_history = []
def select_backend(
self,
edit: Dict,
constraints: Optional[Dict] = None
) -> str:
"""
Select optimal backend for edit
Args:
edit: Edit to process
constraints: Optional constraints (max_latency, min_success_rate)
Returns:
Selected backend name
"""
constraints = constraints or {}
# Extract edit features
features = {
'lang': edit.get('lang', 'en'),
'domain': edit.get('domain', 'text'),
'complexity': edit.get('complexity', 'medium')
}
# Predict performance for each backend
predictions = {}
for backend in self.backends:
pred = self.predictor.predict_performance(backend, features)
predictions[backend] = pred
# Select based on strategy
if self.selection_strategy == 'performance':
selected = max(
predictions.items(),
key=lambda x: x[1]['predicted_success_rate']
)[0]
elif self.selection_strategy == 'latency':
selected = min(
predictions.items(),
key=lambda x: x[1]['predicted_latency_ms']
)[0]
else: # balanced
selected = max(
predictions.items(),
key=lambda x: x[1]['predicted_success_rate'] / (x[1]['predicted_latency_ms'] / 100)
)[0]
# Apply constraints
if constraints:
max_latency = constraints.get('max_latency_ms')
min_success = constraints.get('min_success_rate')
if max_latency and predictions[selected]['predicted_latency_ms'] > max_latency:
# Find alternative
for backend, pred in predictions.items():
if pred['predicted_latency_ms'] <= max_latency:
selected = backend
break
if min_success and predictions[selected]['predicted_success_rate'] < min_success:
# Find alternative
for backend, pred in predictions.items():
if pred['predicted_success_rate'] >= min_success:
selected = backend
break
# Record selection
self.selection_history.append({
'edit_id': edit.get('id', 'unknown'),
'selected_backend': selected,
'predictions': predictions
})
return selected
def batch_select(
self,
edits: List[Dict],
constraints: Optional[Dict] = None
) -> Dict[str, List[Dict]]:
"""
Select backends for batch of edits
Args:
edits: List of edits
constraints: Optional constraints
Returns:
Dict mapping backend names to edit lists
"""
backend_assignments = {backend: [] for backend in self.backends}
for edit in edits:
selected = self.select_backend(edit, constraints)
backend_assignments[selected].append(edit)
return backend_assignments
def get_selection_stats(self) -> Dict:
"""Get selection statistics"""
if not self.selection_history:
return {}
backend_counts = {backend: 0 for backend in self.backends}
for selection in self.selection_history:
backend_counts[selection['selected_backend']] += 1
total = len(self.selection_history)
return {
'total_selections': total,
'backend_distribution': {
backend: count / total
for backend, count in backend_counts.items()
}
}
# Convenience function
def select_best_backend(edit: Dict, backends: List[str] = None) -> str:
"""Quick backend selection"""
selector = BackendSelector(backends=backends)
return selector.select_backend(edit)
|