05: Persona Routing Optimization
FLUX Adaptive LLM Routing for 5-10x Cost Reduction
Pillar: #5 of 12 - Core Priority
Timeline: 8 Weeks
Dependencies: None
Expected ROI: 5-10x LLM cost reduction, 3x average latency improvement
Executive Summary
Maestro's 11 personas make hundreds of LLM calls per workflow, all routed to full-power models (Opus/GPT-4) regardless of query complexity. Analysis reveals 70% of calls could use cheaper models with no quality loss. This pillar implements FLUX (Flexible LLM Unified eXecution) - a confidence-based adaptive routing system that selects the optimal model tier per query.
Key Insight: A simple "What's the status?" query shouldn't cost the same as "Design a microservices architecture."
Problem Statement
Current State (with Code Evidence)
# File: llm-router/src/llm_router/router.py
# Current: Static persona → provider mapping
class PersonaRouter:
def select_provider(self, persona: str) -> Provider:
return self.policy.get(persona, self.default_provider)
# Always returns same model regardless of query complexity
# File: adapters/context_manager.py (line 57)
# Token estimation uses rough 4-char heuristic
total_chars = len(context.system_prompt)
for message in context.messages:
total_chars += len(message.content)
return total_chars // 4 # 15-25% estimation error
Cost Analysis
Blended waste: ~70% of LLM spend is unnecessary.
Solution Architecture
FLUX Routing Flow
┌─────────────────┐
│ HAIKU │ ← Confidence > 0.90
│ (fast/cheap) │ 70% of queries
Request → Context → Confidence ────┼─────────────────┤
Builder Estimator │ SONNET │ ← Confidence 0.70-0.90
↓ │ (balanced) │ 20% of queries
[Features] ├─────────────────┤
↓ │ OPUS │ ← Confidence < 0.70
[Classifier] │ (full power) │ 10% of queries
└─────────────────┘
↓
[Quality Monitor]
↓
[Escalation if needed]
Core Components
1. Confidence Estimator
class ConfidenceEstimator:
"""
Trained on historical Maestro traces.
<1ms inference on CPU via ONNX.
"""
def __init__(self):
self.classifier = load_model("flux_classifier.onnx")
self.thresholds = {
"haiku": 0.90,
"sonnet": 0.70,
"opus": 0.0
}
def estimate(self, persona: str, context: Context, query: str) -> ModelRecommendation:
features = self._extract_features(persona, context, query)
confidence = self.classifier.predict(features)
if confidence > self.thresholds["haiku"]:
return ModelRecommendation("haiku", confidence)
elif confidence > self.thresholds["sonnet"]:
return ModelRecommendation("sonnet", confidence)
else:
return ModelRecommendation("opus", confidence)
def _extract_features(self, persona: str, context: Context, query: str) -> Dict:
return {
"persona_type": self.persona_encoder.encode(persona),
"context_tokens": len(context),
"query_complexity": self._analyze_complexity(query),
"has_code_request": "```" in query or "code" in query.lower(),
"has_reasoning_markers": any(m in query.lower() for m in
["explain", "why", "analyze", "design", "architect"]),
"historical_success": self.persona_history.get_success_rate(persona),
}
2. Early Exit Adapter
class FluxAdapter(ClaudeAdapter):
"""
Progressive model escalation with early exit.
"""
async def invoke(self, context: Context) -> Response:
recommendation = self.confidence_estimator.estimate(
context.persona, context, context.query
)
response = await self._invoke_model(recommendation.model, context)
if self._needs_escalation(response, recommendation):
response = await self._escalate(response, context)
self._record_outcome(recommendation, response)
return response
def _needs_escalation(self, response: Response, rec: ModelRecommendation) -> bool:
return (
response.contains_uncertainty_markers or
response.finish_reason != "stop" or
response.quality_score < 0.8
)
3. Pattern Cache
class PatternCache:
"""
Materialize responses for frequently repeated queries.
Cache if: >5% frequency AND consistent responses.
"""
def check(self, persona: str, query: str) -> Optional[Response]:
key = self._make_key(persona, query)
if key in self.cache and self.cache[key].is_valid():
return self.cache[key].response
return None
Integration Points
Files to Modify
LLM Router Modification
# BEFORE (router.py)
def select_provider(self, persona: str) -> Provider:
return self.policy.get(persona, self.default_provider)
# AFTER
def select_provider(self, persona: str, context: Context, query: str) -> Provider:
cached = self.pattern_cache.check(persona, query)
if cached:
return CachedProvider(cached)
recommendation = self.confidence_estimator.estimate(persona, context, query)
self.trace_recorder.record(persona, context, recommendation)
return self.get_provider_for_model(recommendation.model)
Implementation Plan
Week 1-2: Data Collection & Analysis
- [ ] Extract training data from TouchpointCollector (10K+ samples)
- [ ] Label historical calls for model downgrade potential
- [ ] Analyze query complexity patterns per persona
- [ ] Build query structure feature extractor
Week 3-4: Confidence Classifier
- [ ] Train confidence classifier (target: 85% accuracy)
- [ ] Optimize for <1ms inference latency
- [ ] Export to ONNX for production
- [ ] Build inference wrapper
Week 5-6: Router Integration
- [ ] Modify llm-router for FLUX routing
- [ ] Implement early exit adapter
- [ ] Add fallback escalation logic
- [ ] Wire up FLUX cost tracking
Week 7-8: Production Rollout
- [ ] Implement pattern cache
- [ ] A/B test against baseline
- [ ] Tune thresholds based on production metrics
- [ ] Full deployment with monitoring
Expected Results
Cost Reduction
Latency Reduction
Risks & Mitigations
Success Criteria
Alternatives Considered
FLUX chosen: Adapts to actual query complexity with automatic escalation safety net.
- #04 RAG Next Gen: Semantic compression + FLUX = compound savings
- #08 Observability: FLUX metrics in AgentOps dashboard
- #03 Agent Orchestration: FLUX informs persona selection
Status: Ready for Implementation
Owner: LLM Optimization Team
