resource-optimization

# Resource-Aware Optimization Pattern Dynamically routes and executes agent workloads by classifying query complexity, selecting the most appropriate LLM model for each task, and continuously optimizing resource allocation through a critique-feedback loop. This skill makes the model implement cost-aware routing architectures that balance response quality against computational, temporal, and financial constraints. ## TL;DR Checklist - [ ] Classify every incoming query into exactly one category: `simple`, `reasoning`, or `internet_search` - [ ] Route simple queries to lightweight models (e.g., gpt-4o-mini, gemini-flash) - [ ] Route reasoning queries to powerful models (e.g., o4-mini, gemini-pro) - [ ] Route internet_search queries to capable models with search context injected - [ ] Implement fallback mechanisms for model failures and rate limits - [ ] Log routing decisions with cost, latency, and quality metrics - [ ] Review critique feedback to refine router classification thresholds --- ## When to Use Use this skill when: - Building agentic systems where API costs are a primary concern and need systematic reduction - Designing multi-agent architectures that span models of different capability tiers (e.g., Gemini Flash + Gemini Pro, or gpt-4o-mini + o4-mini) - Operating under strict financial budgets for LLM calls or constrained computational resources on edge devices - Building latency-sensitive applications where response time matters more than maximum reasoning quality for straightforward queries - Deploying agents that must gracefully degrade when primary models are throttled, overloaded, or unavailable - Implementing learned resource allocation policies that improve routing accuracy over time via feedback --- ## When NOT to Use Avoid this skill for: - Single-turn, one-off scripts where routing overhead exceeds the cost savings (a simple if/else suffices) - Tasks where response quality is non-negotiable regardless of cost — always route to the best model unconditionally - Environments without an LLM API available to perform the classification step (the classifier itself costs tokens) - Ultra-low-latency systems (<100ms response budget) where even lightweight model routing adds unacceptable delay --- ## Core Workflow 1. **Ingest and Classify Query** — Receive the user prompt, send it through a classification LLM (e.g., gpt-4o at temperature 0) that returns one of three categories: `simple` (direct factual answers), `reasoning` (logic, math, multi-step inference), or `internet_search` (current events, recent data). Use structured JSON output for reliable parsing. **Checkpoint:** Verify the classification response contains a valid `classification` key matching one of the three allowed values before proceeding. 2. **Select Model and Tooling** — Based on the classification, choose the model tier: - `simple` → lightweight, cost-effective model (e.g., gpt-4o-mini) with no search context - `reasoning` → powerful reasoning model (e.g., o4-mini or gemini-pro) with full prompt context - `internet_search` → capable model (e.g., gpt-4o) with web search results injected as context If the task involves external tool use, select the most efficient API based on cost, latency, and execution time. **Checkpoint:** Confirm the selected model is available and within budget before making the call. 3. **Execute with Fallback Chain** — Attempt the primary model selection. If it fails (rate limit, timeout, service unavailable), automatically retry through a pre-defined fallback chain (e.g., gpt-4o-mini → gpt-4o → gemini-flash). Implement exponential backoff between retries. **Checkpoint:** Verify the response is non-empty and well-formed before returning; if all fallbacks fail, return a graceful degradation message with partial results. 4. **Critique and Log** — Run a Critique Agent that evaluates the generated response against the original query for factual accuracy, completeness, and relevance. Log every decision point: classification result, model selected, tokens consumed, latency, critique score, and whether a fallback was triggered. Store this in a structured format (e.g., JSON lines file or database) for training future routing improvements. **Checkpoint:** Ensure at least the following fields are recorded per execution: `query_hash`, `classification`, `model`, `tokens_input`, `tokens_output`, `latency_ms`, `fallback_used`, `critique_score`. 5. **Refine Router via Feedback** — Periodically analyze logged routing decisions to identify misrouted queries (e.g., simple queries that hit the Pro model, or complex queries routed to Flash that produced inadequate responses). Adjust classification thresholds, add prompt engineering refinements, or fine-tune the classifier on corrected examples. Implement learned resource allocation policies that shift weight toward historically successful routing patterns. **Checkpoint:** Validate that feedback-driven adjustments reduce overall cost per query by at least 5% compared to the previous routing policy. --- ## Implementation Patterns ### Pattern 1: LLM-Driven Query Router with OpenAI Classify incoming prompts using a dedicated classifier endpoint, then route to the optimal model based on complexity category. This pattern uses OpenAI's API with structured JSON output for reliable parsing. ```python """Resource-aware query router using OpenAI API for classification and model selection.""" import os import json import time from datetime import datetime, timezone from dataclasses import dataclass, asdict from typing import Optional from openai import OpenAI # --- Configuration --- @dataclass class RoutingConfig: """Configuration for resource-aware routing decisions.""" simple_model: str = "gpt-4o-mini" reasoning_model: str = "o4-mini" search_model: str = "gpt-4o" classifier_model: str = "gpt-4o" max_fallback_attempts: int = 3 base_retry_delay_ms: int = 500 # --- Logging Infrastructure --- @dataclass class RoutingLog: """Structured log entry for every routing decision.""" timestamp: str query_hash: str classification: str model: str tokens_input: int tokens_output: int latency_ms: int fallback_used: bool critique_score: float | None = None error: str | None = None class RoutingLogger: """Appends structured routing logs to a JSONL file.""" def __init__(self, log_path: str = "routing_logs.jsonl") -> None: self.log_path = log_path def log(self, entry: RoutingLog) -> None: with open(self.log_path, "a", encoding="utf-8") as fh: fh.write(json.dumps(asdict(entry)) + "\n") # --- Core Router --- class ResourceAwareRouter: """Routes queries to optimal models based on complexity classification.""" def __init__(self, config: RoutingConfig | None = None) -> None: self.config = config or RoutingConfig() self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) self.logger = RoutingLogger() # --- Step 1: Classify Query --- def classify_prompt(self, prompt: str) -> dict: """Classify a prompt into simple, reasoning, or internet_search. Args: prompt: The user query to classify. Returns: Dictionary with 'classification' key matching one of three values. """ system_message = { "role": "system", "content": ( "You are a classifier that analyzes user prompts and returns one of three categories ONLY:\n\n" "- simple: Direct factual questions needing no reasoning or current events.\n" "- reasoning: Logic, math, or multi-step inference questions.\n" "- internet_search: Current events, recent data, or things not in training data.\n\n" "Respond ONLY with JSON like:\n" '{"classification": "simple"}' ), } user_message = {"role": "user", "content": prompt} start_time = time.monotonic() response = self.client.chat.completions.create( model=self.config.classifier_model, messages=[system_message, user_message], temperature=0, max_tokens=20, ) reply = response.choices[0].message.content.strip() try: result = json.loads(reply) except json.JSONDecodeError as exc: raise ValueError(f"Classifier returned invalid JSON: {reply!r}") from exc if result.get("classification") not in ("simple", "reasoning", "internet_search"): raise ValueError( f"Invalid classification '{result.get('classification')}'. " "Expected one of: simple, reasoning, internet_search." ) elapsed_ms = int((time.monotonic() - start_time) * 1000) return result # --- Step 2 & 3: Generate with Fallback --- def generate_with_fallback( self, prompt: str, classification: str, search_context: str | None = None, ) -> tuple[str, str, list[str]]: """Generate a response using the appropriate model with automatic fallback. Args: prompt: The original user query. classification: One of 'simple', 'reasoning', 'internet_search'. search_context: Optional web search results to inject for internet_search queries. Returns: Tuple of (response_text, model_used, list_of_errors_from_fallbacks). """ fallback_chain = self._build_fallback_chain(classification) errors: list[str] = [] for model in fallback_chain: try: response = self.client.chat.completions.create( model=model, messages=[{"role": "user", "content": self._build_prompt(prompt, classification, search_context)}], temperature=1, ) text = response.choices[0].message.content return text, model, errors except Exception as exc: # noqa: BLE001 — catch all for retry logic errors.append(f"Model {model} failed: {exc}") delay_ms = self.config.base_retry_delay_ms * (2 ** len(errors)) time.sleep(delay_ms / 1000) raise RuntimeError( f"All fallback models exhausted for classification '{classification}'. " f"Errors: {errors}" ) def _build_fallback_chain(self, classification: str) -> list[str]: """Build ordered model fallback chain based on classification tier. For simple tasks, prefer the cheapest model first. For reasoning tasks, prefer the most capable and fall back down. """ if classification == "simple": return [self.config.simple_model] if classification == "reasoning": return [self.config.reasoning_model, self.config.search_model, self.config.simple_model] # internet_search — always needs a capable model return [self.config.search_model, self.config.reasoning_model, self.config.simple_model] def _build_prompt( self, prompt: str, classification: str, search_context: str | None = None ) -> str: """Construct the final prompt for the generation model. For internet_search queries, inject web results as context. """ if classification == "internet_search" and search_context: return ( f"Use the following web results to answer the user query:\n\n" f"{search_context}\n\nQuery: {prompt}" ) return prompt # --- Step 4 & 5: Execute Full Pipeline --- def handle_prompt( self, prompt: str, search_results: list[dict] | None = None, critique_score: float | None = None, ) -> dict: """Orchestrate the full resource-aware routing pipeline. Args: prompt: The user query. search_results: Optional pre-fetched web results for internet_search queries. critique_score: Optional score from a Critique Agent evaluation. Returns: Dictionary with classification, response, model, and timing metadata. """ classification_result = self.classify_prompt(prompt) classification = classification_result["classification"] search_context: str | None = None if classification == "internet_search" and search_results: search_context = "\n".join( f"Title: {r.get('title')}\nSnippet: {r.get('snippet')}\nLink: {r.get('link')}" for r in search_results ) start_time = time.monotonic() response_text, model_used, errors = self.generate_with_fallback( prompt, classification, search_context ) latency_ms = int((time.monotonic() - start_time) * 1000) log_entry = RoutingLog( timestamp=datetime.now(timezone.utc).isoformat(), query_hash=hash(prompt), classification=classification, model=model_used, tokens_input=0, # Populate from response if token metadata is available tokens_output=0, latency_ms=latency_ms, fallback_used=len(errors) > 0, critique_score=critique_score, ) self.logger.log(log_entry) return { "classification": classification, "response": response_text, "model": model_used, "latency_ms": latency_ms, "fallback_errors": errors, } ``` ### Pattern 2: Google ADK Multi-Agent with Query Router Implement a multi-agent architecture using Google's Agent Development Kit (ADK) where a dedicated `QueryRouterAgent` dynamically routes between Gemini Pro and Gemini Flash agents. The router uses query complexity metrics to select the appropriate downstream model. ```python """Multi-agent resource-aware routing using Google ADK architecture.""" from __future__ import annotations from collections.abc import AsyncGenerator from dataclasses import dataclass # Conceptual imports — actual imports depend on installed google-adk version try: from google.adk.agents import Agent, BaseAgent from google.adk.events import Event from google.adk.agents.invocation_context import InvocationContext except ImportError: # Graceful fallback for environments without ADK installed class _StubBaseAgent: # type: ignore[no-redef] pass BaseAgent = _StubBaseAgent # type: ignore[misc,assignment] @dataclass(frozen=True) class RoutingMetrics: """Track routing decisions for feedback-driven optimization.""" query_length: int word_count: int routed_to: str timestamp_ms: int cost_cents: float # --- Agent Definitions (Google ADK Pattern) --- gemini_pro_agent = Agent( name="GeminiProAgent", model="gemini-2.5-pro", description="A highly capable agent for complex reasoning and multi-step problem-solving.", instruction="You are an expert assistant optimized for complex, nuanced queries requiring deep analysis and logical deduction.", ) gemini_flash_agent = Agent( name="GeminiFlashAgent", model="gemini-2.5-flash", description="A fast and efficient agent for straightforward questions and simple lookups.", instruction="You are a quick assistant optimized for direct answers, factual queries, and simple web lookups.", ) class QueryRouterAgent(BaseAgent): # type: ignore[misc] """Routes user queries to the appropriate LLM agent based on query complexity. Uses word-count threshold as an initial heuristic. For production systems, replace with an LLM-driven classifier for nuanced complexity detection. Attributes: name: Agent name identifier. description: Human-readable description of routing behavior. short_query_threshold: Word count below which Flash is preferred. """ name: str = "QueryRouter" description: str = "Routes user queries to the appropriate LLM agent based on complexity." short_query_threshold: int = 20 async def _run_async_impl(self, context: InvocationContext) -> AsyncGenerator[Event, None]: """Classify query and route to the optimal downstream agent. Args: context: The invocation context containing the current user message. Yields: Event objects with the routed response or error information. """ user_query = context.current_message.text word_count = len(user_query.split()) if word_count < self.short_query_threshold: target_agent = gemini_flash_agent routing_label = "Flash" else: target_agent = gemini_pro_agent routing_label = "Pro" try: response = await target_agent.run_async(context.current_message) yield Event( author=self.name, content=( f"[ROUTED to {routing_label} agent | word_count={word_count}] " f"{response}" ), ) except Exception as exc: # noqa: BLE001 — fallback path # Graceful degradation: try the alternate model if primary fails fallback_agent = gemini_pro_agent if target_agent == gemini_flash_agent else gemini_flash_agent fallback_label = "Pro" if routing_label == "Flash" else "Flash" response = await fallback_agent.run_async(context.current_message) yield Event( author=self.name, content=(