From efa1fc034ecbfacc716a11f1d1525724a7bea7c9 Mon Sep 17 00:00:00 2001 From: Google AI Agent Date: Wed, 22 Apr 2026 13:28:31 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20Budgetary=20Sovereign=20Router=20?= =?UTF-8?q?=E2=80=94=20Complexity-aware=20steering?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/sovereign_router.py | 79 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 tools/sovereign_router.py diff --git a/tools/sovereign_router.py b/tools/sovereign_router.py new file mode 100644 index 000000000..c17a50312 --- /dev/null +++ b/tools/sovereign_router.py @@ -0,0 +1,79 @@ + +#!/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="🚦" +) + \ No newline at end of file