Cost Aware Llm Pipeline

affaan-m/ECC/skills/cost-aware-llm-pipeline

by affaan-mef648e01899ba3e8dc6371642deaaf64b4477775No license275K starsListed Oct 9, 2026Updated Oct 9, 2026Repository updated 4 days ago

Cost optimization patterns for LLM API usage — model routing by task complexity, budget tracking, retry logic, and prompt caching. Use when LLM spend needs to come down, or when routing tasks across model tiers and budgets.

Instructions onlyAI & Agents

AI-generated overview

Patterns for cutting LLM API costs via model routing, budget tracking, retry logic and prompt caching.

What it does: Provides reference patterns and code snippets for building cost-aware LLM API pipelines: routing tasks to cheaper or more expensive models by complexity, tracking cumulative spend with immutable records, retrying only transient errors, and caching long system prompts. It also includes a model pricing reference table and lists of best practices and anti-patterns. It produces guidance and example code rather than a runnable tool.
When to use it: Use when LLM API spend needs to be reduced, when routing tasks across model tiers, or when batch processing needs budget guardrails. It suits applications calling Claude, OpenAI or similar APIs, and multi-model architectures that need intelligent routing.
Requirements: No scripts ship with the skill; it is instructions and code examples only. The examples reference Python, the anthropic client library and its error types, and assume access to an LLM API.

Cost-Aware LLM Pipeline

Patterns for controlling LLM API costs while maintaining quality. Combines model routing, budget tracking, retry logic, and prompt caching into a composable pipeline.

When to Activate

Building applications that call LLM APIs (Claude, GPT, etc.)
Processing batches of items with varying complexity
Need to stay within a budget for API spend
Optimizing cost without sacrificing quality on complex tasks

Core Concepts

1. Model Routing by Task Complexity

Automatically select cheaper models for simple tasks, reserving expensive models for complex ones.

python

MODEL_SONNET = "claude-sonnet-5"MODEL_HAIKU = "claude-haiku-4-5-20251001"
_SONNET_TEXT_THRESHOLD = 10_000  # chars_SONNET_ITEM_THRESHOLD = 30     # items
def select_model(    text_length: int,    item_count: int,    force_model: str | None = None,) -> str:    """Select model based on task complexity."""    if force_model is not None:        return force_model    if text_length >= _SONNET_TEXT_THRESHOLD or item_count >= _SONNET_ITEM_THRESHOLD:        return MODEL_SONNET  # Complex task    return MODEL_HAIKU  # Simple task (3-4x cheaper)

MODEL_SONNET = "claude-sonnet-5"MODEL_HAIKU = "claude-haiku-4-5-20251001"
_SONNET_TEXT_THRESHOLD = 10_000  # chars_SONNET_ITEM_THRESHOLD = 30     # items
def select_model(    text_length: int,    item_count: int,    force_model: str | None = None,) -> str:    """Select model based on task complexity."""    if force_model is not None:        return force_model    if text_length >= _SONNET_TEXT_THRESHOLD or item_count >= _SONNET_ITEM_THRESHOLD:        return MODEL_SONNET  # Complex task    return MODEL_HAIKU  # Simple task (3-4x cheaper)

2. Immutable Cost Tracking

Track cumulative spend with frozen dataclasses. Each API call returns a new tracker — never mutates state.

python

from dataclasses import dataclass
@dataclass(frozen=True, slots=True)class CostRecord:    model: str    input_tokens: int    output_tokens: int    cost_usd: float
@dataclass(frozen=True, slots=True)class CostTracker:    budget_limit: float = 1.00    records: tuple[CostRecord, ...] = ()
    def add(self, record: CostRecord) -> "CostTracker":        """Return new tracker with added record (never mutates self)."""        return CostTracker(            budget_limit=self.budget_limit,            records=(*self.records, record),        )
    @property    def total_cost(self) -> float:        return sum(r.cost_usd for r in self.records)
    @property    def over_budget(self) -> bool:        return self.total_cost > self.budget_limit

from dataclasses import dataclass
@dataclass(frozen=True, slots=True)class CostRecord:    model: str    input_tokens: int    output_tokens: int    cost_usd: float
@dataclass(frozen=True, slots=True)class CostTracker:    budget_limit: float = 1.00    records: tuple[CostRecord, ...] = ()
    def add(self, record: CostRecord) -> "CostTracker":        """Return new tracker with added record (never mutates self)."""        return CostTracker(            budget_limit=self.budget_limit,            records=(*self.records, record),        )
    @property    def total_cost(self) -> float:        return sum(r.cost_usd for r in self.records)
    @property    def over_budget(self) -> bool:        return self.total_cost > self.budget_limit

3. Narrow Retry Logic

Retry only on transient errors. Fail fast on authentication or bad request errors.

python

from anthropic import (    APIConnectionError,    InternalServerError,    RateLimitError,)
_RETRYABLE_ERRORS = (APIConnectionError, RateLimitError, InternalServerError)_MAX_RETRIES = 3
def call_with_retry(func, *, max_retries: int = _MAX_RETRIES):    """Retry only on transient errors, fail fast on others."""    for attempt in range(max_retries):        try:            return func()        except _RETRYABLE_ERRORS:            if attempt == max_retries - 1:                raise            time.sleep(2 ** attempt)  # Exponential backoff    # AuthenticationError, BadRequestError etc. → raise immediately

from anthropic import (    APIConnectionError,    InternalServerError,    RateLimitError,)
_RETRYABLE_ERRORS = (APIConnectionError, RateLimitError, InternalServerError)_MAX_RETRIES = 3
def call_with_retry(func, *, max_retries: int = _MAX_RETRIES):    """Retry only on transient errors, fail fast on others."""    for attempt in range(max_retries):        try:            return func()        except _RETRYABLE_ERRORS:            if attempt == max_retries - 1:                raise            time.sleep(2 ** attempt)  # Exponential backoff    # AuthenticationError, BadRequestError etc. → raise immediately

4. Prompt Caching

Cache long system prompts to avoid resending them on every request.

python

messages = [    {        "role": "user",        "content": [            {                "type": "text",                "text": system_prompt,                "cache_control": {"type": "ephemeral"},  # Cache this            },            {                "type": "text",                "text": user_input,  # Variable part            },        ],    }]

messages = [    {        "role": "user",        "content": [            {                "type": "text",                "text": system_prompt,                "cache_control": {"type": "ephemeral"},  # Cache this            },            {                "type": "text",                "text": user_input,  # Variable part            },        ],    }]

Composition

Combine all four techniques in a single pipeline function:

python

def process(text: str, config: Config, tracker: CostTracker) -> tuple[Result, CostTracker]:    # 1. Route model    model = select_model(len(text), estimated_items, config.force_model)
    # 2. Check budget    if tracker.over_budget:        raise BudgetExceededError(tracker.total_cost, tracker.budget_limit)
    # 3. Call with retry + caching    response = call_with_retry(lambda: client.messages.create(        model=model,        messages=build_cached_messages(system_prompt, text),    ))
    # 4. Track cost (immutable)    record = CostRecord(model=model, input_tokens=..., output_tokens=..., cost_usd=...)    tracker = tracker.add(record)
    return parse_result(response), tracker

def process(text: str, config: Config, tracker: CostTracker) -> tuple[Result, CostTracker]:    # 1. Route model    model = select_model(len(text), estimated_items, config.force_model)
    # 2. Check budget    if tracker.over_budget:        raise BudgetExceededError(tracker.total_cost, tracker.budget_limit)
    # 3. Call with retry + caching    response = call_with_retry(lambda: client.messages.create(        model=model,        messages=build_cached_messages(system_prompt, text),    ))
    # 4. Track cost (immutable)    record = CostRecord(model=model, input_tokens=..., output_tokens=..., cost_usd=...)    tracker = tracker.add(record)
    return parse_result(response), tracker

Pricing Reference (2026)

Model	Input ($/1M tokens)	Output ($/1M tokens)	Relative Cost
Haiku 3.5 (legacy)	$0.80	$4.00	0.8x
Haiku 4.5	$1.00	$5.00	1x
Sonnet 5	$2.00	$10.00	2x
Sonnet 4.6	$3.00	$15.00	3x
Opus 4.8	$5.00	$25.00	5x
Fable 5 / Mythos 5	$10.00	$50.00	10x
Opus 4.0 / 4.1 (legacy)	$15.00	$75.00	15x

Best Practices

Start with the cheapest model and only route to expensive models when complexity thresholds are met
Set explicit budget limits before processing batches — fail early rather than overspend
Log model selection decisions so you can tune thresholds based on real data
Use prompt caching for system prompts over 1024 tokens — saves both cost and latency
Never retry on authentication or validation errors — only transient failures (network, rate limit, server error)

Anti-Patterns to Avoid

Using the most expensive model for all requests regardless of complexity
Retrying on all errors (wastes budget on permanent failures)
Mutating cost tracking state (makes debugging and auditing difficult)
Hardcoding model names throughout the codebase (use constants or config)
Ignoring prompt caching for repetitive system prompts

When to Use

Any application calling Claude, OpenAI, or similar LLM APIs
Batch processing pipelines where cost adds up quickly
Multi-model architectures that need intelligent routing
Production systems that need budget guardrails

Source and attribution

Source:affaan-m/ECCinskills/cost-aware-llm-pipelineat commitef648e0

License: No license

Content belongs to its original authors. SourceWeft indexes it from a public repository.

Report or request removal

Cost Aware Llm Pipeline

Cost-Aware LLM Pipeline

When to Activate

Core Concepts

1. Model Routing by Task Complexity

2. Immutable Cost Tracking

3. Narrow Retry Logic

4. Prompt Caching

Composition

Pricing Reference (2026)

Best Practices

Anti-Patterns to Avoid

When to Use

Source and attribution

More from affaan-m/ECC

Unified Notifications Ops

Verification Loop

Ui To Vue

Unified Memory

Visa Doc Translate

Team Builder

More in AI & Agents

Ppt Template Creator

Skill Development

Plugin Structure

Command Development

Claude Md Improver

Find Skills