Ai Cost Optimization

AI Cost Optimization

Purpose

Design and implement cost optimization strategies for LLM inference: token optimization, semantic caching, inference optimization, model routing, batching, distillation, and budget governance to minimize per-query cost while maintaining output quality. Provide production-ready patterns for cost monitoring, chargeback, and automated budget enforcement.

Agent Protocol

Trigger

User request includes: AI cost, token cost, LLM cost, prompt compression, caching for LLM, semantic cache, KV cache, quantization, model routing, cost optimization, batching, token counting, inference cost, context window, budget, chargeback, showback, cost allocation, FinOps, model distillation, speculative decoding.

Input Context

Required for a full optimization plan:

• Model(s) currently in use and pricing ($/1K tokens)
• Average tokens per query (input + output)
• Daily/monthly query volume
• Current cache setup (if any)
• Latency requirements (SLA in ms)
• Monthly budget and cost to date
• Number of users/teams for cost allocation

Output Artifact

A complete AI cost optimization plan covering: baseline measurement, token optimization, caching architecture, inference tuning, model routing, batching strategy, budget governance, monitoring setup, and projected savings.

Response Format

## AI Cost Optimization Plan
### Baseline
Model: {name} | Avg Tokens/Query: {N} | Daily Queries: {N}
Monthly Cost: ${N} | Cost/Query: ${N} | Budget: ${N}

### Token Optimization
- Prompt Compression: {method} | {savings: X%}
- System Prompt: {current tokens} → {optimized tokens}
- Context Window: {max tokens} → {reduced to N}
- Streaming: {enabled/disabled}

### Semantic Cache
Storage: {Redis / Momento / in-memory}
Embedding Model: {name} | Threshold: {similarity}
TTL: {duration} | Hit Rate Target: {X%}
Estimated Savings: {X%} | ${N}/month

### Inference Optimization
Quantization: {FP16 / INT8 / INT4 / none}
Flash Attention: {enabled/disabled}
KV Cache: {on/off} | Batch Size: {N}

### Model Routing
| Query Type | Model | Cost/Query | Allocation |
|---|---|---|---|
| Simple | {cheap model} | ${N} | {X%} |
| Complex | {expensive model} | ${N} | {Y%} |
| Reasoning | {reasoning model} | ${N} | {Z%} |

### Budget Governance
Daily Budget: ${N} | Monthly: ${N}
Soft Limit: {X%} | Hard Limit: {Y%}
Fallback Model: {name} on budget exceed
Chargeback: {per-team / per-user / none}

### Monitoring
Metrics Export: {Prometheus / Datadog / custom}
Alerts: {threshold / anomaly / model shift}
Dashboard: {Grafana / custom}

No preamble. No postamble. No explanations. No filler/hedging/transitions.

Completion Criteria

• [ ] Baseline cost measured: tokens per query, model pricing, daily volume.
• [ ] Token optimization applied: prompt compression target 40-60%, system prompt under 150 tokens.
• [ ] Semantic cache configured with embedding model, similarity threshold, TTL, hit rate target >30%.
• [ ] Inference optimization selected: quantization level, Flash Attention, KV cache tuning.
• [ ] Model routing rules defined with query classification and per-model cost.
• [ ] Batching strategy configured with max batch size, max wait, latency budget.
• [ ] Cost monitoring with Prometheus metrics, budget alerts, anomaly detection.
• [ ] Budget governance with soft/hard limits and automatic fallback model.
• [ ] Chargeback/showback allocation per team or user if multi-tenant.
• [ ] Optimization ROI tracked with before/after cost comparison.

Architecture Decision Framework

Decision Tree: Cost Optimization Strategy Selection

What is your primary constraint?
├── Cost reduction priority (reduce spend)
│   ├── Query volume  100K/day
│       ├── Continuous batching (vLLM/TensorRT-LLM)
│       ├── INT8 quantization with calibration
│       ├── Self-host for >100M tokens/day
│       ├── Prefix caching (shared prompts)
│       └── Model distillation pipeline
│
├── Latency constraint (response time SLA)
│   ├── SLA  2s (offline/batch)
│       ├── Large batch sizes (32-64)
│       ├── Continuous batching
│       ├── Prompt caching
│       └── Use cheapest acceptable model
│
├── Quality constraint (output quality must equal frontier)
│   ├── Use frontier model as primary
│   ├── Cascade: cheap model first, verify, upgrade on low confidence
│   ├── Semantic cache with high threshold (0.95+)
│   └── No quantization below FP16
│
└── Budget constraint (fixed monthly spend)
    ├── Set hard daily budget per model
    ├── Budget enforcement: fallback to cheaper on limit
    ├── Cost-aware router: downgrade model when budget tight
    ├── Chargeback per team to drive accountability
    └── Monthly cost review with optimization iteration

Decision Tree: Cache Architecture Selection

What is your query diversity?
├── Low diversity (10K unique queries/day)
    ├── Hybrid cache: exact + prefix + semantic
    ├── Embedding model: all-MiniLM-L6-v2 (100K)
    ├── LRU eviction with max capacity
    └── Cache warming for common queries

Decision Tree: Model Routing Strategy

What is your query complexity distribution?
├── >70% simple queries
│   ├── Route simple → GPT-4o-mini / Claude Haiku ($0.15/M tokens)
│   ├── Route complex → GPT-4o / Claude Sonnet ($2.50/M tokens)
│   └── Estimated savings: 50-70%
├── 40-70% simple queries
│   ├── Three-tier routing: simple → cheap, medium → mid, hard → frontier
│   ├── Use classifier (ML or heuristic) for routing
│   └── Estimated savings: 30-50%
└──  str:
        return hashlib.sha256(f"{prompt}:{model}".encode()).hexdigest()

    def _estimate_tokens(self, text: str) -> int:
        return len(text) // 4

    def _compute_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        cfg = self.models[model]
        return (input_tokens * cfg.input_price_per_1k + output_tokens * cfg.output_price_per_1k) / 1000

    async def _classify_query(self, prompt: str) -> str:
        word_count = len(prompt.split())
        if word_count  str:
        routing = {
            "simple": "gpt-4o-mini",
            "complex": "gpt-4o",
        }
        return routing.get(classification, "gpt-4o-mini")

    async def semantic_search(self, query_embed: np.ndarray) -> Optional[str]:
        now = time.time()
        best_score = self.threshold
        best_response = None
        expired_keys = []
        for key, (response, embed, ts) in self.cache.items():
            if now - ts > self.ttl:
                expired_keys.append(key)
                continue
            score = float(np.dot(query_embed, embed))
            if score > best_score:
                best_score = score
                best_response = response
        for k in expired_keys:
            del self.cache[k]
        return best_response

    async def infer(self, prompt: str, user_id: Optional[str] = None) -> tuple[str, float]:
        start = time.time()
        exact_key = self._make_exact_key(prompt, "")  # check across models
        if exact_key in self.cache:
            self.cache_hits += 1
            response, _, _ = self.cache[exact_key]
            latency = (time.time() - start) * 1000
            self.cost_records.append(CostRecord(
                timestamp=start, model="cache", input_tokens=0,
                output_tokens=0, total_cost=0, cache_hit=True,
                latency_ms=latency, user_id=user_id, route="cache"
            ))
            return response, 0.0

        if self.embedder:
            query_embed = np.array(self.embedder(prompt))
            cached = await self.semantic_search(query_embed)
            if cached:
                self.cache_hits += 1
                latency = (time.time() - start) * 1000
                self.cost_records.append(CostRecord(
                    timestamp=start, model="cache", input_tokens=0,
                    output_tokens=0, total_cost=0, cache_hit=True,
                    latency_ms=latency, user_id=user_id, route="semantic_cache"
                ))
                return cached, 0.0

        self.cache_misses += 1
        classification = await self._classify_query(prompt)
        model_name = self._select_model(classification)
        response = await self.llm_call(model=model_name, prompt=prompt)
        latency = (time.time() - start) * 1000
        output_tokens = self._estimate_tokens(response)
        input_tokens = self._estimate_tokens(prompt)
        cost = self._compute_cost(model_name, input_tokens, output_tokens)
        self.cost_records.append(CostRecord(
            timestamp=start, model=model_name, input_tokens=input_tokens,
            output_tokens=output_tokens, total_cost=cost, cache_hit=False,
            latency_ms=latency, user_id=user_id, route=classification
        ))
        if self.embedder:
            query_embed = np.array(self.embedder(prompt))
            self.cache[exact_key] = (response, query_embed, time.time())
        return response, cost

    def cost_summary(self, days: int = 30) -> dict:
        cutoff = time.time() - days * 86400
        recent = [r for r in self.cost_records if r.timestamp >= cutoff]
        total = sum(r.total_cost for r in recent)
        by_model = defaultdict(float)
        by_user = defaultdict(float)
        for r in recent:
            by_model[r.model] += r.total_cost
            if r.user_id:
                by_user[r.user_id] += r.total_cost
        total_calls = len(recent)
        return {
            "period_days": days,
            "total_cost": round(total, 2),
            "total_calls": total_calls,
            "avg_cost_per_call": round(total / max(total_calls, 1), 6),
            "cache_hit_rate": round(self.cache_hits / max(self.cache_hits + self.cache_misses, 1), 3),
            "cost_by_model": dict(by_model),
            "cost_by_user": dict(by_user),
        }

Pattern 2: Semantic Cache with Embedding Selection

class EmbeddingCache:
    def __init__(
        self,
        model_name: str = "all-MiniLM-L6-v2",
        threshold: float = 0.92,
        ttl_seconds: int = 3600,
        max_entries: int = 50000,
        storage_backend: str = "memory",
        redis_client=None,
    ):
        if storage_backend == "memory":
            self.store = {}
        elif storage_backend == "redis":
            self.store = redis_client
        else:
            raise ValueError(f"Unknown storage: {storage_backend}")
        self.model_name = model_name
        self.threshold = threshold
        self.ttl = ttl_seconds
        self.max_entries = max_entries
        self.hits = 0
        self.misses = 0
        self._init_embedder()

    def _init_embedder(self):
        from sentence_transformers import SentenceTransformer
        self.encoder = SentenceTransformer(self.model_name)

    def _embed(self, text: str) -> np.ndarray:
        return self.encoder.encode(text, normalize_embeddings=True)

    def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
        return float(np.dot(a, b))

    def get(self, query: str) -> Optional[str]:
        query_embed = self._embed(query)
        now = time.time()
        expired_keys = []
        best_match = None
        best_score = self.threshold
        for key, entry in self.store.items():
            if now - entry["ts"] > self.ttl:
                expired_keys.append(key)
                continue
            score = self._cosine_similarity(query_embed, entry["embed"])
            if score > best_score:
                best_score = score
                best_match = entry["response"]
        for k in expired_keys:
            del self.store[k]
        if best_match:
            self.hits += 1
            return best_match
        self.misses += 1
        return None

    def set(self, query: str, response: str):
        if len(self.store) >= self.max_entries:
            oldest = min(self.store.keys(), key=lambda k: self.store[k]["ts"])
            del self.store[oldest]
        embed = self._embed(query)
        key = hashlib.md5(query.encode()).hexdigest()
        self.store[key] = {
            "embed": embed,
            "response": response,
            "ts": time.time(),
        }

    def tune_threshold(self, eval_pairs: list[tuple[str, str, bool]]) -> float:
        best_f1 = 0
        best_thresh = self.threshold
        for thresh_pct in range(80, 99):
            t = thresh_pct / 100.0
            tp = fp = fn = tn = 0
            for q1, q2, should_match in eval_pairs:
                e1 = self._embed(q1)
                e2 = self._embed(q2)
                sim = self._cosine_similarity(e1, e2)
                predicted = sim >= t
                if predicted and should_match:
                    tp += 1
                elif predicted and not should_match:
                    fp += 1
                elif not predicted and should_match:
                    fn += 1
                else:
                    tn += 1
            precision = tp / max(tp + fp, 1)
            recall = tp / max(tp + fn, 1)
            f1 = 2 * precision * recall / max(precision + recall, 1e-6)
            if f1 > best_f1:
                best_f1 = f1
                best_thresh = t
        self.threshold = best_thresh
        return best_thresh

    def stats(self) -> dict:
        total = self.hits + self.misses
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": round(self.hits / max(total, 1), 3),
            "size": len(self.store),
            "threshold": self.threshold,
            "ttl": self.ttl,
            "model": self.model_name,
        }

Pattern 3: Model Cascade with Budget Enforcement

@dataclass
class CascadeTier:
    model: str
    cost_per_call: float
    max_retries: int = 0
    fallback_on_error: bool = True

class ModelCascade:
    def __init__(
        self,
        llm_call: Callable,
        tiers: list[CascadeTier],
        daily_budget: float = 100.0,
    ):
        self.llm_call = llm_call
        self.tiers = tiers
        self.daily_spend = 0.0
        self.daily_budget = daily_budget
        self.monthly_budget = daily_budget * 30
        self.monthly_spend = 0.0
        self.cascade_stats: dict[str, int] = defaultdict(int)

    def _check_budget(self, estimated_cost: float) -> bool:
        if self.daily_spend + estimated_cost > self.daily_budget:
            return False
        if self.monthly_spend + estimated_cost > self.monthly_budget:
            return False
        return True

    def _get_cheapest_tier(self) -> CascadeTier:
        return min(self.tiers, key=lambda t: t.cost_per_call)

    async def execute(self, prompt: str, min_quality: str = "simple") -> tuple[str, str, float]:
        for tier in self.tiers:
            est_cost = tier.cost_per_call
            if not self._check_budget(est_cost):
                fallback = self._get_cheapest_tier()
                result = await self.llm_call(model=fallback.model, prompt=prompt)
                cost = fallback.cost_per_call
                self.daily_spend += cost
                self.monthly_spend += cost
                self.cascade_stats["budget_fallback"] += 1
                return result, fallback.model, cost

            try:
                result = await self.llm_call(model=tier.model, prompt=prompt)
                cost = tier.cost_per_call
                self.daily_spend += cost
                self.monthly_spend += cost
                self.cascade_stats[tier.model] += 1
                return result, tier.model, cost
            except Exception as e:
                if tier.fallback_on_error and tier != self.tiers[-1]:
                    continue
                raise

        fallback = self._get_cheapest_tier()
        result = await self.llm_call(model=fallback.model, prompt=prompt)
        cost = fallback.cost_per_call
        self.daily_spend += cost
        self.monthly_spend += cost
        return result, fallback.model, cost

    def budget_status(self) -> dict:
        return

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [j4flmao](https://github.com/j4flmao)
- **Source:** [j4flmao/agent-skills](https://github.com/j4flmao/agent-skills)
- **License:** MIT

Install and usage instructions live in the source repository linked above.