LLM Cost Optimization

Overview

Reducing LLM API costs in production through caching and batching. This guide covers practical implementation for production ML systems.

Why This Matters in MLOps

Setup

bash
Install required tools
pip install python mlflow pandas numpy scikit-learn
Or with Docker
docker pull python:3.11-slim

Core Implementation

python
import os
import json
import logging
from datetime import datetime
from pathlib import Path
logger = logging.getLogger(__name__)
class LLMCostOptimization:
    """
    LLM Cost Optimization implementation.
    
    Handles: cost management
    Tool: python
    """
    
    def __init__(self, config: dict = None):
        self.config = config or self._default_config()
        self._setup()
    
    def _default_config(self) -> dict:
        return {
            "tool": "python",
            "environment": os.getenv("ENVIRONMENT", "development"),
            "log_level": "INFO",
        }
    
    def _setup(self):
        """Initialize python connection and resources."""
        logging.basicConfig(level=self.config.get("log_level", "INFO"))
        logger.info(f"Initialized LLM Cost Optimization with config: {self.config}")
    
    def run(self, **kwargs) -> dict:
        """Execute cost management."""
        start = datetime.utcnow()
        
        try:
            result = self._execute(**kwargs)
            
            elapsed = (datetime.utcnow() - start).total_seconds()
            logger.info(f"LLM Cost Optimization completed in {elapsed:.2f}s")
            
            return {
                "status": "success",
                "result": result,
                "elapsed_seconds": elapsed
            }
        
        except Exception as e:
            logger.error(f"LLM Cost Optimization failed: {e}")
            return {
                "status": "failed", 
                "error": str(e)
            }
    
    def _execute(self, **kwargs) -> dict:
        """Core cost management logic. Override to customize."""
        return {"completed": True, "tool": "python"}
Configuration
config = {
    "tool": "python",
    "tracking_uri": os.getenv("MLFLOW_TRACKING_URI", "http://localhost:5000"),
    "artifact_root": "./artifacts",
}
Initialize
processor = LLMCostOptimization(config)
result = processor.run()
print(json.dumps(result, indent=2))

PYTHON Integration

python
Specific python integration for cost management
import subprocess
def setup_python():
    """Configure python for cost management."""
    
    # Initialize project
    print(f"Setting up python for cost management...")
    
    # Example configuration
    config = {
        "project": "my-ml-project",
        "tool": "python",
        "specialty": "cost management",
        "version": "1.0.0"
    }
    
    # Save configuration
    Path(".python").mkdir(exist_ok=True)
    with open(f".python/config.json", "w") as f:
        json.dump(config, f, indent=2)
    
    print(f"python configured for cost management")
    return config
config = setup_python()

Monitoring and Alerting

python
from dataclasses import dataclass
import time
class MLOpsMonitor:
    """Monitor cost management metrics."""
    
    def __init__(self):
        self.metrics: list[MetricSnapshot] = []
        self.thresholds = {
            "error_rate": 0.05,
            "latency_p99_ms": 1000,
            "data_drift_score": 0.3
        }
    
    def record(self, metric: str, value: float, labels: dict = None):
        snapshot = MetricSnapshot(
            timestamp=time.time(),
            metric_name=metric,
            value=value,
            labels=labels or {}
        )
        self.metrics.append(snapshot)
        self._check_threshold(metric, value)
    
    def _check_threshold(self, metric: str, value: float):
        threshold = self.thresholds.get(metric)
        if threshold and value > threshold:
            logger.warning(f"ALERT: {metric}={value:.3f} exceeds threshold {threshold}")
monitor = MLOpsMonitor()

CI/CD Integration

yaml
.github/workflows/ml-pipeline.yml
name: ML Pipeline
on:
  push:
    paths: ['src/', 'data/']
jobs:
  train-and-evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run cost management
        run: python -m src.llm_cost_optimization
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_URI }}
      
      - name: Check model quality
        run: python -m src.validate_model

Best Practices

Resources