Cost Optimization

Strategies for optimizing agent system costs: compute, tokens, storage, and operational efficiency.

Cost Analysis Framework

Cost Components

Total Cost = Compute + Tokens + Storage + Network + Operations

Token Cost Optimization

Model Selection Strategy

class ModelSelector:
    def __init__(self):
        self.models = {
            "gpt-4o": ModelProfile(cost_per_1k_in=0.005, cost_per_1k_out=0.015, quality=1.0, speed=1.0),
            "gpt-4o-mini": ModelProfile(cost_per_1k_in=0.00015, cost_per_1k_out=0.0006, quality=0.85, speed=2.0),
            "gpt-3.5-turbo": ModelProfile(cost_per_1k_in=0.0005, cost_per_1k_out=0.0015, quality=0.7, speed=3.0),
            "claude-3.5-sonnet": ModelProfile(cost_per_1k_in=0.003, cost_per_1k_out=0.015, quality=0.95, speed=1.2),
            "claude-3-haiku": ModelProfile(cost_per_1k_in=0.00025, cost_per_1k_out=0.00125, quality=0.75, speed=4.0),
            "gemini-1.5-flash": ModelProfile(cost_per_1k_in=0.000075, cost_per_1k_out=0.0003, quality=0.8, speed=3.0),
        }

    def select_model(self, task: Task, budget_per_request: float = None, 
                     quality_threshold: float = 0.7) -> str:
        """Select cheapest model meeting quality threshold"""
        candidates = [
            m for m in self.models.values() 
            if m.quality >= quality_threshold
        ]

        if budget_per_request:
            candidates = [m for m in candidates 
                         if m.estimated_cost_per_request() <= budget_per_request]

        return min(candidates, key=lambda m: m.estimated_cost_per_request()).name

    def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        m = self.models[model]
        return (input_tokens * m.cost_per_1k_in + output_tokens * m.cost_per_1k_out) / 1000

Prompt Compression

class PromptCompressor:
    def __init__(self, tokenizer, target_ratio: float = 0.5):
        self.tokenizer = tokenizer
        self.target_ratio = target_ratio

    def compress(self, text: str, preserve_keywords: list[str] = None) -> str:
        """Compress prompt while preserving key information"""
        tokens = self.tokenizer.encode(text)
        target_tokens = int(len(tokens) * self.target_ratio)

        if len(tokens) <= target_tokens:
            return text

        # Strategy 1: Remove redundant whitespace
        compressed = self._remove_redundancy(text)

        # Strategy 2: Summarize with LLM (if still too long)
        if len(self.tokenizer.encode(compressed)) > target_tokens:
            compressed = self._llm_summarize(text, target_tokens)

        # Strategy 3: Preserve keywords, truncate rest
        if len(self.tokenizer.encode(compressed)) > target_tokens:
            compressed = self._preserve_keywords(compressed, preserve_keywords, target_tokens)

        return compressed

    def _preserve_keywords(self, text: str, keywords: list[str], max_tokens: int) -> str:
        # Keep sentences containing keywords
        sentences = sent_tokenize(text)
        scored = []
        for sent in sentences:
            score = sum(1 for kw in keywords if kw.lower() in sent.lower())
            scored.append((score, sent))

        # Sort by score, take top until token limit
        scored.sort(key=lambda x: x[0], reverse=True)
        result = []
        tokens = 0
        for score, sent in scored:
            sent_tokens = len(self.tokenizer.encode(sent))
            if tokens + sent_tokens <= max_tokens:
                result.append(sent)
                tokens += sent_tokens
            else:
                break
        return " ".join(result)

Prompt Caching

class PromptCache:
    def __init__(self, redis: Redis, ttl: int = 3600):
        self.redis = redis
        self.ttl = ttl

    def cache_key(self, prompt: str, model: str, params: dict) -> str:
        content = f"{model}:{json.dumps(prompt, sort_keys=True)}:{json.dumps(params, sort_keys=True)}"
        return f"prompt_cache:{hashlib.sha256(prompt.encode()).hexdigest()[:16]}"

    async def get(self, prompt: str, model: str, params: dict) -> Optional[str]:
        key = self.cache_key(prompt, model, params)
        return await self.redis.get(key)

    async def set(self, prompt: str, model: str, params: dict, response: str):
        key = self.cache_key(prompt, model, params)
        await self.redis.setex(f"prompt_cache:{key}", self.ttl, response)

    async def get_or_compute(self, prompt: str, model: str, params: dict, 
                             compute_fn: Callable) -> str:
        cached = await self.get(prompt, model, {})
        if cached:
            return cached

        response = await compute_fn(prompt, model)
        await self.set(prompt, model, {}, response)
        return response

Semantic Caching (Embedding-based)

class SemanticPromptCache:
    def __init__(self, redis: Redis, embedder: EmbeddingModel, 
                 similarity_threshold: float = 0.95, ttl: int = 3600):
        self.redis = redis
        self.embedder = embedder
        self.threshold = similarity_threshold
        self.ttl = ttl

    async def get(self, prompt: str) -> Optional[str]:
        embedding = await self.embedder.embed(prompt)

        # Search for similar prompts
        similar = await self.search_similar(prompt, top_k=5)

        for cached_prompt, cached_response, similarity in similar:
            if similarity >= self.threshold:
                return cached_response
        return None

    async def set(self, prompt: str, response: str):
        embedding = await self.embedder.embed(prompt)
        key = f"semantic_cache:{hashlib.sha256(prompt.encode()).hexdigest()[:16]}"

        await self.redis.hset(f"semantic_cache:{prompt[:50]}", mapping={
            "prompt": prompt,
            "response": json.dumps(response),
            "embedding": json.dumps(embedding.tolist())
        })
        await self.redis.expire(f"semantic_cache:{prompt[:50]}", self.ttl)

Compute Cost Optimization

Right-Sizing Instances

class InstanceRightSizer:
    def __init__(self, metrics_client: MetricsClient):
        self.metrics = metrics_client

    async def analyze(self, deployment: str, window: str = "7d") -> RightSizingRecommendation:
        metrics = await self.metrics.get_container_metrics(deployment, window)

        cpu_util = statistics.mean(m.cpu_utilization for m in metrics)
        memory_util = statistics.mean(m.memory_utilization for m in metrics)

        # Current instance
        current = self.get_current_instance_type()

        # Find optimal
        optimal = self.find_optimal_instance(cpu_util, memory_util)

        savings = self.calculate_savings(current, optimal)

        return RightSizingRecommendation(
            current_instance=current,
            recommended_instance=optimal,
            cpu_utilization=cpu_util,
            memory_utilization=memory_util,
            estimated_monthly_savings=savings
        )

    def find_optimal_instance(self, cpu_util: float, mem_util: float) -> InstanceType:
        # Target 70% utilization
        target_cpu = 0.7
        target_mem = 0.7

        cpu_factor = cpu_util / target_cpu if cpu_util > 0 else 1
        mem_factor = memory_util / target_mem if memory_util > 0 else 1

        scale_factor = max(cpu_factor, mem_factor)

        # Find instance with appropriate capacity
        for instance in INSTANCE_TYPES:
            if (instance.cpu >= current.cpu / scale_factor and 
                instance.memory >= current.memory / scale_factor):
                return instance

        return current  # Already optimal

Spot Instance Strategy

class SpotInstanceManager:
    def __init__(self, cloud_provider: CloudProvider):
        self.provider = cloud_provider

    async def create_spot_fleet(self, spec: SpotFleetSpec) -> SpotFleet:
        # Diversify across instance types and AZs
        instance_types = self.get_diverse_instance_types(spec.requirements)

        fleet = await self.provider.create_spot_fleet(
            instance_types=instance_types,
            target_capacity=spec.target_capacity,
            allocation_strategy="diversified",
            max_price=spec.max_price,
            allocation_strategy="lowest-price",
            instance_interruption_behavior="terminate"
        )

        # Add fallback to on-demand
        await self.add_on_demand_fallback(fleet, spec.fallback_percentage)

        return fleet

    async def handle_interruption(self, instance_id: str):
        # Graceful draining
        await self.drain_node(instance_id)

        # Replace with new spot or on-demand
        await self.replace_instance(instance_id)

Serverless Cost Optimization

Right-Sizing Lambda

class LambdaOptimizer:
    def __init__(self, cloudwatch: CloudWatchClient):
        self.cw = cloudwatch

    async def optimize_function(self, function_name: str) -> OptimizationResult:
        # Get metrics
        metrics = await self.get_metrics(function_name, days=14)

        # Analyze
        avg_duration = statistics.mean(m.duration for m in metrics.durations)
        max_memory_used = max(m.max_memory_used for m in metrics.memory)
        avg_memory_used = statistics.mean(m.avg_memory_used for m in metrics.memory)

        # Current config
        current_memory = self.get_function_memory(function_name)

        # Optimal memory (power of 2, min 128MB)
        optimal_memory = self.calculate_optimal_memory(
            avg_memory_used, max_memory_used
        )

        # Estimate savings
        current_cost = self.estimate_monthly_cost(current_memory, metrics)
        optimal_cost = self.estimate_monthly_cost(optimal_memory, metrics)

        return OptimizationResult(
            current_memory=current_memory,
            recommended_memory=optimal_memory,
            current_monthly_cost=current_cost,
            projected_monthly_cost=optimal_cost,
            savings_pct=(current_cost - optimal_cost) / current_cost * 100
        )

Provisioned Concurrency Optimization

class ProvisionedConcurrencyOptimizer:
    def __init__(self, lambda_client: LambdaClient):
        self.client = lambda_client

    async def optimize(self, function_name: str) -> ConcurrencyConfig:
        # Get utilization metrics
        metrics = await self.get_concurrency_metrics(function_name, days=7)

        # Calculate optimal provisioned concurrency
        p90_concurrent = np.percentile(metrics.concurrent_executions, 90)
        p99_concurrent = np.percentile(metrics.concurrent_executions, 99)

        # Target: cover P99 with 20% buffer
        recommended = int(p99_concurrent * 1.2)

        # Cost analysis
        provisioned_cost = self.calculate_provisioned_cost(recommended)
        on_demand_cost = self.estimate_on_demand_cost(metrics)

        return ConcurrencyConfig(
            recommended_provisioned=recommended,
            current_provisioned=self.get_current_provisioned(function_name),
            estimated_savings=on_demand_cost - provisioned_cost
        )

Storage Cost Optimization

Tiered Storage

class StorageTierManager:
    def __init__(self, s3_client: S3Client):
        self.s3 = s3_client

    async def apply_lifecycle_policy(self, bucket: str, policy: LifecyclePolicy):
        """Apply intelligent tiering"""
        rules = []

        # Hot -> Standard (0-30 days)
        rules.append(LifecycleRule(
            id="hot-to-standard",
            status="Enabled",
            filter={"Prefix": ""},
            transitions=[
                Transition(StorageClass="STANDARD_IA", Days=30),
                Transition(StorageClass="GLACIER_IR", Days=90),
                Transition(StorageClass="GLACIER", Days=180),
                Transition(StorageClass="DEEP_ARCHIVE", Days=365)
            ]
        ))

        # Delete incomplete multipart uploads
        rules.append(LifecycleRule(
            id="cleanup-multipart",
            status="Enabled",
            abort_incomplete_multipart_upload=AbortIncompleteMultipartUpload(
                DaysAfterInitiation=7
            )
        ))

        # Expire old versions
        rules.append(LifecycleRule(
            id="expire-old-versions",
            status="Enabled",
            noncurrent_version_transitions=[
                NoncurrentVersionTransition(StorageClass="STANDARD_IA", NoncurrentDays=30),
                NoncurrentVersionTransition(StorageClass="GLACIER", NoncurrentDays=90)
            ],
            noncurrent_version_expiration=NoncurrentVersionExpiration(NoncurrentDays=365)
        ))

        await self.s3.put_bucket_lifecycle_configuration(
            Bucket=bucket,
            LifecycleConfiguration={"Rules": rules}
        )

Intelligent Tiering (S3)

# Enable S3 Intelligent-Tiering
await s3.put_bucket_intelligent_tiering_configuration(
    Bucket=bucket,
    Id="auto-tiering",
    IntelligentTieringConfiguration={
        "Id": "auto-tiering",
        "Status": "Enabled",
        "Tierings": [
            {"Days": 1, "AccessTier": "ARCHIVE_INSTANT_ACCESS"},
            {"Days": 30, "AccessTier": "ARCHIVE"},
            {"Days": 90, "AccessTier": "DEEP_ARCHIVE"}
        ]
    )
)

Database Storage Optimization

class DatabaseStorageOptimizer:
    def __init__(self, pg_pool: AsyncConnectionPool):
        self.pool = pg_pool

    async def analyze_table_bloat(self) -> list[BloatReport]:
        query = """
        SELECT 
            schemaname, tablename,
            pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as total_size,
            pg_size_pretty(pg_relation_size(schemaname||'.'||tablename)) as table_size,
            pg_size_pretty(pg_total_relation_size - pg_relation_size) as bloat_size,
            round(100 * (pg_total_relation_size - pg_relation_size) / pg_total_relation_size) as bloat_pct
        FROM pg_stat_user_tables
        WHERE pg_total_relation_size > 100 * 1024 * 1024  -- > 100MB
        ORDER BY bloat_pct DESC;
        """
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(query)

        return [BloatReport(**row) for row in rows]

    async def reclaim_space(self, table: str) -> ReclaimResult:
        # pg_repack for online vacuum
        result = await self.execute_pg_repack(table)
        return ReclaimResult(table=table, reclaimed_bytes=result.reclaimed)

Network Cost Optimization

CDN & Caching

class CDNOptimizer:
    def __init__(self, cdn_client: CDNClient):
        self.cdn = cdn_client

    async def optimize_cache_policy(self, distribution_id: str) -> CachePolicy:
        # Analyze cache hit ratios
        stats = await self.get_cache_stats(distribution_id)

        # Optimal TTL by content type
        policies = {
            "static_assets": CachePolicy(ttl=31536000, vary_headers=["Accept-Encoding"]),
            "api_responses": CachePolicy(ttl=300, vary_headers=["Authorization", "Accept"]),
            "user_content": CachePolicy(ttl=0, no_cache=True),
            "api_static": CachePolicy(ttl=3600, vary_headers=["Accept"])
        }

        return CachePolicyConfig(
            behaviors=self.map_paths_to_policies(stats),
            default_ttl=3600
        )

Data Transfer Optimization

class DataTransferOptimizer:
    def __init__(self, cloud_provider: CloudProvider):
        self.provider = cloud_provider

    async def optimize_cross_region_transfer(self) -> TransferOptimization:
        # Use VPC endpoints for S3/DynamoDB
        # Enable VPC peering instead of public internet
        # Use S3 Transfer Acceleration for large uploads
        # Enable compression (gzip, brotli)
        # Use multipart upload for large files

        recommendations = [
            "Enable VPC endpoints for S3/DynamoDB",
            "Use VPC peering for cross-region",
            "Enable S3 Transfer Acceleration for >100MB",
            "Enable compression (gzip/brotli) for all responses",
            "Use multipart upload for files >100MB",
            "Implement request coalescing for duplicate requests"
        ]

        return TransferOptimization(recommendations=recommendations)

Monitoring & Alerting for Costs

Cost Anomaly Detection

class CostAnomalyDetector:
    def __init__(self, cost_explorer: CostExplorerClient):
        self.ce = cost_explorer

    async def detect_anomalies(self, account_id: str, days: int = 30) -> list[CostAnomaly]:
        # Get historical costs
        costs = await self.get_daily_costs(account_id, days)

        # Statistical anomaly detection
        daily_costs = [c.amount for c in costs]
        mean = statistics.mean(daily_costs)
        stdev = statistics.stdev(daily_costs) if len(daily_costs) > 1 else 0

        anomalies = []
        for cost in costs[-7:]:  # Last 7 days
            z_score = abs(cost.amount - mean) / stdev if stdev > 0 else 0
            if z_score > 3:  # 3 sigma
                anomalies.append(CostAnomaly(
                    date=cost.date,
                    amount=cost.amount,
                    expected=mean,
                    z_score=z_score,
                    severity="high" if z_score > 4 else "medium"
                ))

        return anomalies

Budget Alerts

class BudgetAlertManager:
    def __init__(self, budgets_client: BudgetsClient):
        self.client = budgets_client

    async def create_budget_alert(self, config: BudgetConfig):
        await self.client.create_budget(Budget(
            budget_name=config.name,
            budget_limit=config.limit,
            time_unit=config.time_unit,
            cost_filters=config.filters,
            notifications=[
                BudgetNotification(
                    notification_type="ACTUAL",
                    comparison_operator="GREATER_THAN",
                    threshold=80,
                    notification_type="EMAIL",
                    subscribers=[config.email]
                ),
                BudgetNotification(
                    notification_type="FORECASTED",
                    comparison_operator="GREATER_THAN",
                    threshold=100,
                    notification_type="EMAIL",
                    subscribers=[config.email]
                )
            ]
        ))

Cost Allocation & Chargeback

Tag-Based Cost Allocation

class CostAllocator:
    def __init__(self, cost_explorer: CostExplorerClient):
        self.ce = cost_explorer

    async def allocate_costs(self, period: TimeRange, 
                             allocation_tags: list[str]) -> CostAllocation:
        # Get costs with tags
        costs = await self.ce.get_cost_and_usage(
            TimePeriod={"Start": period.start, "End": period.end},
            Granularity="DAILY",
            Metrics=["UnblendedCost"],
            GroupBy=[{"Type": "TAG", "Key": tag} for tag in allocation_tags]
        )

        # Aggregate by tag
        allocations = defaultdict(float)
        for result in costs:
            for group in result["Groups"]:
                key = group["Keys"][0] if group["Keys"] else "untagged"
                cost = float(group["Metrics"]["UnblendedCost"]["Amount"])
                allocations[key] += cost

        return CostAllocation(
            period=period,
            allocations=dict(allocations),
            total=sum(allocations.values())
        )

Showback/Chargeback Reports

class ChargebackReportGenerator:
    def __init__(self, cost_allocator: CostAllocator):
        self.allocator = cost_allocator

    async def generate_monthly_report(self, month: str) -> ChargebackReport:
        allocation = await self.allocator.allocate_costs(
            period=TimeRange.from_month(month),
            allocation_tags=["team", "project", "environment"]
        )

        # Generate per-team breakdown
        team_costs = defaultdict(float)
        for tag, cost in allocation.allocations.items():
            if tag.startswith("team:"):
                team_costs[tag[5:]] += cost

        return ChargebackReport(
            month=month,
            total_cost=allocation.total,
            by_team=dict(team_costs),
            by_project=self.aggregate_by_tag(allocation, "project"),
            by_environment=self.aggregate_by_tag(allocation, "environment"),
            trends=self.calculate_trends(month),
            recommendations=self.generate_recommendations(allocation)
        )

Kubernetes Cost Optimization

Resource Quotas & Limits

# Namespace resource quota
apiVersion: v1
kind: ResourceQuota
metadata:
  name: agent-namespace-quota
  namespace: agents
spec:
  hard:
    requests.cpu: "100"
    requests.memory: "200Gi"
    limits.cpu: "200"
    limits.memory: "400Gi"
    requests.nvidia.com/gpu: "4"
    persistentvolumeclaims: "20"
    services.loadbalancers: "5"
---
# LimitRange for default container limits
apiVersion: v1
kind: LimitRange
metadata:
  name: agent-limits
  namespace: agents
spec:
  limits:
  - default:
      cpu: "2000m"
      memory: "2Gi"
    defaultRequest:
      cpu: "250m"
      memory: "512Mi"
    max:
      cpu: "4000m"
      memory: "8Gi"
    type: Container

KubeCost Integration

# kubecost-values.yaml
kubecost:
  prometheus:
    server:
      global:
        scrape_interval: 30s
      retention: 15d
  costs:
    defaultClusterId: production-cluster
  networkCosts:
    enabled: true
  kubecostToken: ${KUBECOST_TOKEN}

FinOps Practices

FinOps Framework

class FinOpsManager:
    def __init__(self):
        self.principles = [
            "Team owns cost of their services",
            "Cost is a first-class metric",
            "Decisions driven by business value",
            "Forecasting and planning are essential",
            "Continuous optimization"
        ]

    async def generate_finops_report(self) -> FinOpsReport:
        return FinOpsReport(
            maturity=self.assess_maturity(),
            cost_allocation=self.get_cost_allocation(),
            optimization_opportunities=self.get_optimizations(),
            forecasting=self.generate_forecast(),
            governance=self.assess_governance()
        )

    def assess_maturity(self) -> MaturityLevel:
        # Crawl, Walk, Run
        checks = {
            "cost_visibility": self.has_cost_visibility(),
            "allocation": self.has_cost_allocation(),
            "optimization": self.has_automated_optimization(),
            "forecasting": self.has_forecasting(),
            "governance": self.has_governance(),
            "culture": self.has_finops_culture()
        }
        score = sum(checks.values()) / len(checks)
        if score >= 0.8: return MaturityLevel.RUN
        if score >= 0.5: return MaturityLevel.WALK
        return MaturityLevel.CRAWL

Related Notes

• Deployment Architecture — Deployment patterns
• Scaling Strategies — Scaling strategies
• Observability — Cost monitoring
• Reliability — Reliability vs cost
• Deployment Architecture — Deployment
• CI CD Best Practices — CI/CD costs
• Infrastructure as Code — IaC costs
• Observability — Cost observability
• CI CD Best Practices — CI/CD costs
• Supply Chain Security — Supply chain costs
• Research Directions — Cost optimization research

Sources: AWS Well-Architected Cost Optimization, Google Cloud Cost Optimization, Azure Cost Management, FinOps Foundation, AWS Compute Optimizer, AWS Cost Explorer, AWS Budgets, Kubecost, CloudHealth, CloudCheckr, Cloudability, Cloudability, FinOps Foundation, AWS Well-Architected Framework, Google Cloud FinOps, Azure Cost Management, CNCF FinOps, CNCF FinOps Working Group.