File size: 8,823 Bytes
ae1e90a | 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 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 | #!/usr/bin/env python3
"""
ISL Router: Complexity-Based Task Routing
Routes tasks between local OLMoE and external tools based on ISL principles
"""
import re
from typing import Dict, List, Tuple
from enum import Enum
class TaskComplexity(Enum):
"""Task complexity levels for routing decisions"""
TRIVIAL = 1 # Direct lookup, simple calc
SIMPLE = 2 # Basic reasoning, OLMoE handles
MODERATE = 3 # Needs OLMoE + maybe tools
COMPLEX = 4 # Requires external specialist
UNCERTAIN = 5 # Needs search/verification
class RoutingDecision(Enum):
"""Routing strategies"""
LOCAL_ONLY = "local" # Pure OLMoE
LOCAL_WITH_TOOLS = "hybrid" # OLMoE + tools
DELEGATE = "external" # External specialist
SEARCH_FIRST = "search" # Search then OLMoE
class ISLRouter:
"""
ISL-Style Router for Hybrid MoE System
Philosophy:
- Measure complexity BEFORE calling expensive models
- Delegate only when needed
- Avoid over-thinking simple tasks
- Log when OLMoE was enough
"""
def __init__(self):
# Complexity scoring patterns
self.trivial_patterns = [
r"what is \d+\s*[\+\-\*\/]\s*\d+", # Simple math
r"define \w+", # Definitions
r"what time", # Time queries
]
self.complex_patterns = [
r"prove|theorem|lemma", # Math proofs
r"implement.*algorithm", # Complex code
r"analyze.*security", # Security analysis
r"compare.*benchmark", # Deep comparisons
]
self.search_patterns = [
r"latest|recent|current|today", # Time-sensitive
r"who is|who was", # Entity queries
r"news|breaking", # News
]
# Tool availability
self.available_tools = {
'calculator': True,
'search': True,
'code_executor': True,
'specialist_llm': False # Expensive, use sparingly
}
def score_complexity(self, query: str) -> Tuple[TaskComplexity, float]:
"""
Score task complexity using ISL heuristics
Returns:
(complexity_level, confidence_score)
"""
query_lower = query.lower()
# Check trivial patterns
for pattern in self.trivial_patterns:
if re.search(pattern, query_lower):
return TaskComplexity.TRIVIAL, 0.9
# Check search patterns
for pattern in self.search_patterns:
if re.search(pattern, query_lower):
return TaskComplexity.UNCERTAIN, 0.8
# Check complex patterns
for pattern in self.complex_patterns:
if re.search(pattern, query_lower):
return TaskComplexity.COMPLEX, 0.85
# Heuristics
word_count = len(query.split())
has_code = '```' in query or 'def ' in query or 'function' in query or 'write a' in query_lower
has_math = any(op in query for op in ['+', '-', '*', '/', '=', '^']) or 'calculate' in query_lower
# Scoring logic
if word_count < 10 and not has_code:
# Check if it's a math problem that needs calculator
if has_math and 'calculate' in query_lower:
return TaskComplexity.MODERATE, 0.8 # Trigger tools
return TaskComplexity.SIMPLE, 0.7
elif has_code or word_count > 50:
return TaskComplexity.MODERATE, 0.6
else:
return TaskComplexity.SIMPLE, 0.65
def route(self, query: str) -> Dict:
"""
Make routing decision based on complexity
Returns:
{
'strategy': RoutingDecision,
'complexity': TaskComplexity,
'confidence': float,
'tools': List[str],
'reasoning': str
}
"""
complexity, confidence = self.score_complexity(query)
# Routing logic
if complexity == TaskComplexity.TRIVIAL:
return {
'strategy': RoutingDecision.LOCAL_ONLY,
'complexity': complexity,
'confidence': confidence,
'tools': [],
'reasoning': 'Trivial task, OLMoE handles directly'
}
elif complexity == TaskComplexity.SIMPLE:
return {
'strategy': RoutingDecision.LOCAL_ONLY,
'complexity': complexity,
'confidence': confidence,
'tools': [],
'reasoning': 'Simple reasoning, OLMoE sufficient'
}
elif complexity == TaskComplexity.MODERATE:
# Check if tools would help
tools_needed = []
if 'calculate' in query.lower() or any(op in query for op in ['+', '-', '*', '/']):
tools_needed.append('calculator')
if 'search' in query.lower() or 'find' in query.lower():
tools_needed.append('search')
return {
'strategy': RoutingDecision.LOCAL_WITH_TOOLS if tools_needed else RoutingDecision.LOCAL_ONLY,
'complexity': complexity,
'confidence': confidence,
'tools': tools_needed,
'reasoning': f'Moderate task, OLMoE + {len(tools_needed)} tools'
}
elif complexity == TaskComplexity.UNCERTAIN:
return {
'strategy': RoutingDecision.SEARCH_FIRST,
'complexity': complexity,
'confidence': confidence,
'tools': ['search'],
'reasoning': 'Time-sensitive or factual, search first'
}
else: # COMPLEX
return {
'strategy': RoutingDecision.DELEGATE,
'complexity': complexity,
'confidence': confidence,
'tools': ['specialist_llm'],
'reasoning': 'Complex task, needs specialist model'
}
def should_use_olmoe(self, routing: Dict) -> bool:
"""Check if OLMoE should be used"""
return routing['strategy'] in [
RoutingDecision.LOCAL_ONLY,
RoutingDecision.LOCAL_WITH_TOOLS,
RoutingDecision.SEARCH_FIRST
]
def estimate_tokens(self, query: str, routing: Dict) -> int:
"""Estimate token usage for budgeting"""
base_tokens = len(query.split()) * 1.3 # Rough estimate
if routing['strategy'] == RoutingDecision.LOCAL_ONLY:
return int(base_tokens * 50) # Response multiplier
elif routing['strategy'] == RoutingDecision.LOCAL_WITH_TOOLS:
return int(base_tokens * 80) # Tools add overhead
elif routing['strategy'] == RoutingDecision.SEARCH_FIRST:
return int(base_tokens * 100) # Search + synthesis
else:
return int(base_tokens * 200) # Specialist is expensive
def log_decision(self, query: str, routing: Dict, was_successful: bool):
"""Log routing decisions for analysis"""
log_entry = {
'query': query[:100],
'strategy': routing['strategy'].value,
'complexity': routing['complexity'].name,
'confidence': routing['confidence'],
'tools_used': routing['tools'],
'success': was_successful
}
# In production, write to file/DB
print(f"📊 Routing Log: {log_entry}")
def test_router():
"""Test ISL router with sample queries"""
router = ISLRouter()
test_queries = [
"What is 2 + 2?",
"Explain quantum computing",
"Who won the latest election?",
"Prove the Riemann hypothesis",
"Write a Python function to reverse a string",
"What's the weather today?",
"Fix this grammar: Me and him goes to store",
]
print("="*70)
print("🧪 ISL Router Test")
print("="*70)
for query in test_queries:
routing = router.route(query)
tokens = router.estimate_tokens(query, routing)
print(f"\n📝 Query: {query}")
print(f" Strategy: {routing['strategy'].value}")
print(f" Complexity: {routing['complexity'].name}")
print(f" Confidence: {routing['confidence']:.0%}")
print(f" Tools: {routing['tools'] or 'None'}")
print(f" Est. Tokens: ~{tokens}")
print(f" Reasoning: {routing['reasoning']}")
if __name__ == "__main__":
test_router()
|