79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
|
|
#!/usr/bin/env python3
|
|
"""
|
|
Budgetary Sovereign Router — Complexity-Aware Inference Steering.
|
|
|
|
Uses a deterministic GOFAI scoring model to determine if a prompt
|
|
requires high-reasoning (Cloud LLM) or commodity-reasoning (Local LLM).
|
|
"""
|
|
|
|
import re
|
|
import logging
|
|
from typing import List, Dict, Any
|
|
from tools.registry import registry, tool_error, tool_result
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ROUTER_SCHEMA = {
|
|
"name": "sovereign_router",
|
|
"description": "Analyzes a prompt and recommends the most cost-effective inference path. It uses a GOFAI model to detect complexity markers, potentially saving 90% in cloud costs for 'Small Fry' operations.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"prompt": {"type": "string", "description": "The prompt or task description to analyze."}
|
|
},
|
|
"required": ["prompt"]
|
|
}
|
|
}
|
|
|
|
class ComplexityScore:
|
|
HIGH_REASONING_MARKERS = [
|
|
r"refactor", r"architect", r"design pattern", r"security audit",
|
|
r"complex", r"debug a crash", r"optimize performance"
|
|
]
|
|
COMMODITY_MARKERS = [
|
|
r"summarize", r"extract json", r"clean up typos", r"format",
|
|
r"write a test for", r"todo", r"explain"
|
|
]
|
|
|
|
@classmethod
|
|
def score(cls, text: str) -> int:
|
|
score = 0
|
|
text = text.lower()
|
|
for marker in cls.HIGH_REASONING_MARKERS:
|
|
if re.search(marker, text):
|
|
score += 5
|
|
for marker in cls.COMMODITY_MARKERS:
|
|
if re.search(marker, text):
|
|
score -= 3
|
|
# Length penalty
|
|
if len(text) > 2000:
|
|
score += 2
|
|
return score
|
|
|
|
def route_prompt(prompt: str):
|
|
"""Determine routing path."""
|
|
score = ComplexityScore.score(prompt)
|
|
|
|
threshold = 2
|
|
recommendation = "CLOUD" if score >= threshold else "LOCAL"
|
|
|
|
return tool_result(
|
|
status="Routing Determined",
|
|
score=score,
|
|
recommendation=recommendation,
|
|
reason=f"Prompt complexity score is {score}. Threshold for Cloud is {threshold}.",
|
|
action=f"Routing this request to ${recommendation} inference engine."
|
|
)
|
|
|
|
def _handle_router(args, **kwargs):
|
|
return route_prompt(args.get("prompt"))
|
|
|
|
registry.register(
|
|
name="sovereign_router",
|
|
toolset="dispatch",
|
|
schema=ROUTER_SCHEMA,
|
|
handler=_handle_router,
|
|
emoji="🚦"
|
|
)
|
|
|