AI Workload'ları için FinOps: Production'da LLM Maliyet Yönetimi

Abstract

Production’da Large Language Model çalıştırmak, geleneksel cloud infrastructure’dan temelden farklı bir maliyet modeli sunuyor. Token-based pricing, maliyetlerin kullanım pattern’larına, prompt tasarımına ve model seçimine göre 100 kat değişebileceği anlamına geliyor. Öngörülebilir compute-hour faturalandırmanın aksine, LLM harcamaları kötü optimize edilmiş prompt’lardan veya sınırsız tool kullanımından beklenmedik şekilde artabiliyor.

Bu rehber, prompt caching (%90 tasarruf), intelligent model routing (%30-50 azalma), token budget enforcement ve semantic caching gibi sistematik LLM maliyet optimizasyon yaklaşımlarını inceliyor. Bu pattern’ları uygulayan ekipler tipik olarak kaliteyi koruyarak %60-80 maliyet azalması sağlıyor.

Token-Based Faturalandırma Zorluğu

Geleneksel cloud FinOps prensipleri doğrudan LLM workload’larına çevrilemiyor. Tek bir kötü tasarlanmış prompt, binlerce optimize edilmiş request’ten daha fazla token tüketebiliyor.

Maliyet Değişkenliği Örneği

# Basit sorgu: "What's the weather?"
# Input: 50 token (user message + system prompt)
# Output: 30 token
# Maliyet (GPT-4): $0.0008

# Karmaşık RAG sorgusu: "Q4 satış trendlerini analiz et ve strateji öner"
# Input: 8,050 token (50 user + 10,000 system + 2,000 RAG context)
# Output: 500 token
# Maliyet (GPT-4): $0.095 (119x daha pahalı)

# Tool-call storm: Agent bir turn'de 20 tool çağırıyor
# Tool başına input: 8,050 token × 20 = 161,000 token
# Tool başına output: 500 token × 20 = 10,000 token
# Maliyet (GPT-4): $1.91 (2,387x daha pahalı)

Problem, uygulamalar proof-of-concept’ten production’a scale olurken daha da büyüyor. Çalıştığım ekiplerde aylık faturaların lansmanın ardından birkaç hafta içinde 500 dolardan 15,000 dolara çıktığını gördüm.

Provider Pricing Modellerini Anlamak

Farklı provider’lar, toplam sahip olma maliyetini önemli ölçüde etkileyen farklı pricing yapıları sunuyor.

AWS Bedrock Pricing Katmanları

Standard (On-Demand): Taahhütsüz token-based faturalandırma

• Claude Sonnet 4.6: $3 input,$15 output per 1M token
• En esnek seçenek, en yüksek token başına maliyet

Batch Inference: Asenkron workload’lar için %50 indirim

• Gece raporları, bulk doküman analizi için ideal
• Real-time olmayan processing kabul edilebilir

Provisioned Throughput: Yüksek hacimli senaryolar için zamana dayalı pricing

• Reserve edilmiş kapasite, öngörülebilir maliyetler
• Örnek: Claude Haiku 4.5 Provisioned Throughput ile (6 aylık taahhüt)

OpenAI Pricing Yapısı

PRICING = {
    'gpt-4-turbo': {
        'input': 10.00 / 1_000_000,
        'output': 30.00 / 1_000_000
    },
    'gpt-4o': {
        'input': 2.50 / 1_000_000,
        'output': 10.00 / 1_000_000
    },
    'gpt-4o-mini': {
        'input': 0.15 / 1_000_000,
        'output': 0.60 / 1_000_000
    }
}

# Önemli insight: Output token'ları input'tan 2-5x daha pahalı
# GPT-4: Output 3x daha pahalı ($30 vs $10 per 1M)
# GPT-4o: Output 4x daha pahalı ($10 vs $2.50 per 1M)

Anthropic Direct Pricing

• Claude Opus 4.1: $15 input,$75 output per 1M token
• Claude Opus 4.5: $5 input,$25 output per 1M token (daha yeni, daha maliyet-etkin)
• Claude Sonnet 4.6: $3 input,$15 output per 1M token
• Claude Haiku 3: $0.25 input,$1.25 output per 1M token
• Claude Haiku 4.5: $1 input,$5 output per 1M token (daha yeni nesil)
• Prompt Caching: Cache’lenen token’larda %90 indirim, %85 latency azalması
• Cache Write Premium: Cache write’larında %25 premium (içeriği cache’leme için tek seferlik maliyet)

Optimizasyon Stratejisi 1: Prompt Caching

Prompt caching, minimal implementation çabasıyla en yüksek maliyet azalmasını sağlıyor. Statik prompt bileşenlerini cacheable olarak işaretleyerek, cache TTL süresi içindeki sonraki request’ler bu token’larda %90 indirim alıyor.

AWS Bedrock ile Implementation

import boto3
import json

bedrock_runtime = boto3.client('bedrock-runtime')

# Şirket politikaları ile büyük system prompt (10,000 token)
SYSTEM_PROMPT = """Sen Acme Corp için bir müşteri destek ajanısın.

Şirket Politikaları:
[... 8,000 token politika, prosedür, FAQ ...]

İletişim Stili:
- Profesyonel ama samimi
- Kısa yanıtlar (max 200 kelime)
- Her zaman ilgili politika referansları ekle

Tool Kullanım Kılavuzu:
[... 2,000 token tool dokümantasyonu ...]
"""

def invoke_with_caching(user_message: str):
    response = bedrock_runtime.converse(
        modelId="anthropic.claude-sonnet-4-20250929-v1:0",  # -v1:0 suffix ekle
        messages=[
            {
                "role": "user",
                "content": [{"text": user_message}]
            }
        ],
        system=[
            {
                "text": SYSTEM_PROMPT,
                "cachePoint": {"type": "default"}  # 5 dakika cache
            }
        ]
    )

    # Maliyet analizi
    usage = response['usage']

    # İlk çağrı: Tam input token maliyeti + cache write premium (%25)
    # input_tokens: 10,000 (system) + 50 (user message) = 10,050
    # cacheReadInputTokensCount: 0
    # cacheCreationInputTokensCount: 10,000 (ilk write'da %25 premium ile)

    # 5 dakika içindeki sonraki çağrılar:
    # input_tokens: 50 (sadece user message)
    # cacheReadInputTokensCount: 10,000 (%90 indirim)

    print(f"Input tokens: {usage.get('inputTokens', 0)}")
    print(f"Cached input tokens: {usage.get('cacheReadInputTokensCount', 0)}")
    print(f"Cache creation tokens: {usage.get('cacheCreationInputTokensCount', 0)}")
    print(f"Output tokens: {usage.get('outputTokens', 0)}")

    return response

Maliyet Etki Analizi

# Caching olmadan (100 request/gün):
# Günlük maliyet: (10,050 * 100 * $3/1M) + (200 * 100 * $15/1M) = $3.32
# Aylık maliyet: $3.32 * 30 = $99.60

# Caching ile (%80 cache hit rate varsayımı):
# İlk request: (10,050 * $3/1M) + (200 * $15/1M) = $0.033
# Cache'li request'ler: (50 * $3/1M) + (10,000 * $0.30/1M) + (200 * $15/1M) = $0.0066
# Günlük maliyet: $0.033 + (99 * $0.0066) = $0.686
# Aylık maliyet: $0.686 * 30 = $20.58
# Tasarruf: %79 azalma

Implementation Best Practice’leri

Prompt’ları Caching için Yapılandır:

• Statik içeriği (politikalar, talimatlar) önce yerleştir
• Dinamik context (user data, timestamp’ler) user message’larda olsun
• Cache’li bölümleri gereksiz yere değiştirme

Yaygın Hatalar:

• Dinamik Timestamp’ler: System prompt’a current_time eklemek her request’te cache’i invalidate eder
• Kesintili Traffic: 5 dakikalık TTL, trafikteki boşlukların cache’i invalidate ettiği anlamına gelir
• Prompt Versiyonlama: Prompt değişikliklerini düşük trafikli dönemlerde deploy et

# KÖTÜ: Dinamik içerik cache'i invalidate eder
system_prompt = f"""
Sen bir destek ajanısın.
Şu anki zaman: {datetime.now().isoformat()}  # Her request'te değişir!
[... prompt'un geri kalanı ...]
"""

# İYİ: Statik prompt, dinamik context user message'da
system_prompt = """
Sen bir destek ajanısın.
[... statik politikalar ve talimatlar ...]
"""

user_message = f"""
Şu anki zaman: {datetime.now().isoformat()}
Kullanıcı sorusu: {question}
"""

Optimizasyon Stratejisi 2: Intelligent Model Routing

Tüm sorular en güçlü (ve pahalı) modeli gerektirmiyor. Sorguları complexity’e göre route etmek, minimal kalite etkisiyle maliyetleri %30-50 azaltabiliyor.

Custom Routing Implementation

import OpenAI from 'openai';

interface ModelRoutingConfig {
  simpleThreshold: number;  // < 0.3 = basit sorgu
  complexThreshold: number;  // > 0.7 = karmaşık sorgu
  models: {
    simple: string;
    medium: string;
    complex: string;
  };
}

interface QueryComplexity {
  score: number;
  factors: {
    wordCount: number;
    questionType: string;
    contextRequired: boolean;
    multiStepReasoning: boolean;
  };
}

class IntelligentRouter {
  private openai: OpenAI;
  private config: ModelRoutingConfig;

  constructor() {
    this.openai = new OpenAI();
    this.config = {
      simpleThreshold: 0.3,
      complexThreshold: 0.7,
      models: {
        simple: 'gpt-4o-mini',  // $0.15 input, $0.60 output per 1M
        medium: 'gpt-4o',  // $2.50 input, $10.00 output per 1M
        complex: 'gpt-4-turbo'  // $10.00 input, $30.00 output per 1M
      }
    };
  }

  /**
   * Heuristik kullanarak sorgu complexity'sini analiz et
   * Production sistemleri hafif bir classifier model kullanabilir
   */
  analyzeComplexity(query: string): QueryComplexity {
    const words = query.split(/\s+/);
    const wordCount = words.length;

    // Soru türünü tespit et
    const questionType = this.detectQuestionType(query);

    // Multi-step reasoning göstergelerini kontrol et
    const multiStepKeywords = ['karşılaştır', 'analiz', 'tasarla', 'uygula',
                                'değerlendir', 'öner', 'strateji'];
    const multiStepReasoning = multiStepKeywords.some(kw =>
      query.toLowerCase().includes(kw)
    );

    // Context gerekli mi (önceki konuşma, doküman referansları)
    const contextRequired = query.toLowerCase().includes('önceki') ||
                           query.toLowerCase().includes('daha önce') ||
                           query.toLowerCase().includes('bahsettiğ');

    // Complexity score hesapla (0.0 - 1.0)
    let score = 0.0;

    // Kelime sayısı faktörü (daha uzun = potansiyel olarak daha karmaşık)
    if (wordCount < 10) score += 0.1;
    else if (wordCount < 30) score += 0.3;
    else score += 0.5;

    // Soru tipi faktörü
    if (questionType === 'factual') score += 0.1;
    else if (questionType === 'analytical') score += 0.5;
    else score += 0.3;

    // Multi-step reasoning önemli complexity ekler
    if (multiStepReasoning) score += 0.3;

    // Context gereksinimi complexity ekler
    if (contextRequired) score += 0.2;

    // 0.0 - 1.0 aralığına normalize et
    score = Math.min(1.0, score);

    return {
      score,
      factors: {
        wordCount,
        questionType,
        contextRequired,
        multiStepReasoning
      }
    };
  }

  private detectQuestionType(query: string): string {
    const lower = query.toLowerCase();

    // Faktörel sorular
    if (lower.match(/^(ne|nedir|kim|nerede) /)) return 'factual';

    // Analitik sorular
    if (lower.match(/(nasıl|neden|açıkla|karşılaştır|analiz)/)) return 'analytical';

    // Prosedürel sorular
    if (lower.match(/(nasıl yapılır|adımlar|süreç|uygula)/)) return 'procedural';

    return 'general';
  }

  selectModel(complexity: QueryComplexity): string {
    if (complexity.score < this.config.simpleThreshold) {
      return this.config.models.simple;
    } else if (complexity.score < this.config.complexThreshold) {
      return this.config.models.medium;
    } else {
      return this.config.models.complex;
    }
  }

  async invoke(query: string, systemPrompt: string) {
    const complexity = this.analyzeComplexity(query);
    const model = this.selectModel(complexity);

    console.log(`Query complexity: ${complexity.score.toFixed(2)} -> ${model}`);

    const response = await this.openai.chat.completions.create({
      model,
      messages: [
        { role: 'system', content: systemPrompt },
        { role: 'user', content: query }
      ],
      temperature: 0.7
    });

    return {
      response: response.choices[0].message.content,
      model,
      complexity: complexity.score,
      usage: response.usage
    };
  }
}

// Kullanım örneği
const router = new IntelligentRouter();

// Basit sorgu -> gpt-4o-mini
await router.invoke(
  "İade politikanız nedir?",
  "Sen bir müşteri destek ajanısın"
);

// Karmaşık sorgu -> gpt-4-turbo
await router.invoke(
  "Enterprise ve business planlarımızı karşılaştır, 500 çalışanlı orta ölçekli bir şirket için hangisi daha iyi olur, 3 yıllık ölçeklenebilirlik ve maliyeti göz önünde bulundurarak analiz et",
  "Sen bir müşteri destek ajanısın"
);

AWS Bedrock Intelligent Prompt Routing

AWS Bedrock kullanan ekipler için intelligent routing, prompt router üzerinden mevcut:

import boto3

bedrock = boto3.client('bedrock-runtime')

# Prompt router ARN kullan ("family ARN" değil)
# Prompt router sorgu complexity'sini analiz eder ve uygun model'e route eder
PROMPT_ROUTER_ARN = "arn:aws:bedrock:us-east-1::prompt-router/anthropic.claude"

response = bedrock.converse(
    modelId=PROMPT_ROUTER_ARN,  # Bedrock prompt router otomatik Haiku veya Sonnet'e route eder
    messages=[
        {
            "role": "user",
            "content": [{"text": "Seattle'da hava nasıl?"}]
        }
    ]
)

# Bedrock otomatik route eder:
# - Basit sorgular -> Claude Haiku 4.5 ($1.00/1M input)
# - Karmaşık sorgular -> Claude 3.5 Sonnet ($3/1M input)
# Ortalama maliyet azalması: Kalite kaybı olmadan %30
# Gözlemlenen routing: RAG dataset'lerinde %87 Haiku, %13 Sonnet

Beklenen Sonuçlar

Optimizasyon Stratejisi 3: Token Budget Enforcement

Sınırsız token tüketimi maliyet fırtınalarına yol açıyor. Hard limit’ler uygulamak, sistem fonksiyonelliğini korurken kontrolsüz harcamaları önlüyor.

Budget Tracking Implementation

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, Optional
import redis

@dataclass
class TokenBudget:
    max_input_tokens_per_request: int
    max_output_tokens_per_request: int
    max_tokens_per_user_daily: int
    max_tokens_per_team_monthly: int

@dataclass
class BudgetUsage:
    user_id: str
    team_id: str
    tokens_used_today: int
    tokens_used_this_month: int
    last_reset: datetime

class TokenBudgetEnforcer:
    def __init__(self, budget: TokenBudget):
        self.budget = budget
        self.redis_client = redis.Redis(host='localhost', decode_responses=True)

    def check_and_reserve(
        self,
        user_id: str,
        team_id: str,
        estimated_input_tokens: int,
        estimated_output_tokens: int
    ) -> tuple[bool, Optional[str]]:
        """
        Request'in budget dahilinde olup olmadığını kontrol et ve token'ları reserve et.
        (allowed, error_message) döndürür
        """

        # Request başına limit'leri kontrol et
        if estimated_input_tokens > self.budget.max_input_tokens_per_request:
            return False, f"Input token ({estimated_input_tokens}) request limiti aşıyor ({self.budget.max_input_tokens_per_request})"

        if estimated_output_tokens > self.budget.max_output_tokens_per_request:
            return False, f"Output token ({estimated_output_tokens}) request limiti aşıyor ({self.budget.max_output_tokens_per_request})"

        # Günlük user limit'ini kontrol et
        user_daily_key = f"budget:user:{user_id}:daily"
        user_tokens_today = int(self.redis_client.get(user_daily_key) or 0)

        total_estimated = estimated_input_tokens + estimated_output_tokens

        if user_tokens_today + total_estimated > self.budget.max_tokens_per_user_daily:
            return False, f"Kullanıcı günlük limiti aşıldı ({user_tokens_today}/{self.budget.max_tokens_per_user_daily})"

        # Aylık team limit'ini kontrol et
        team_monthly_key = f"budget:team:{team_id}:monthly"
        team_tokens_this_month = int(self.redis_client.get(team_monthly_key) or 0)

        if team_tokens_this_month + total_estimated > self.budget.max_tokens_per_team_monthly:
            return False, f"Takım aylık limiti aşıldı ({team_tokens_this_month}/{self.budget.max_tokens_per_team_monthly})"

        # Token'ları reserve et (optimistic locking)
        pipe = self.redis_client.pipeline()

        # User günlük counter'ını artır (gece yarısında expire oluyor)
        tomorrow = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1)
        seconds_until_midnight = int((tomorrow - datetime.now()).total_seconds())
        pipe.incrby(user_daily_key, total_estimated)
        pipe.expire(user_daily_key, seconds_until_midnight)

        # Team aylık counter'ını artır (ay sonunda expire oluyor)
        next_month = (datetime.now().replace(day=1) + timedelta(days=32)).replace(day=1)
        seconds_until_month_end = int((next_month - datetime.now()).total_seconds())
        pipe.incrby(team_monthly_key, total_estimated)
        pipe.expire(team_monthly_key, seconds_until_month_end)

        pipe.execute()

        return True, None

    def record_actual_usage(
        self,
        user_id: str,
        team_id: str,
        actual_input_tokens: int,
        actual_output_tokens: int,
        estimated_input_tokens: int,
        estimated_output_tokens: int
    ):
        """
        Gerçek vs tahmin edilen kullanıma göre budget'ı ayarla.
        """
        actual_total = actual_input_tokens + actual_output_tokens
        estimated_total = estimated_input_tokens + estimated_output_tokens
        difference = actual_total - estimated_total

        if difference != 0:
            pipe = self.redis_client.pipeline()
            pipe.incrby(f"budget:user:{user_id}:daily", difference)
            pipe.incrby(f"budget:team:{team_id}:monthly", difference)
            pipe.execute()

# LLM uygulamasında kullanım
budget_enforcer = TokenBudgetEnforcer(
    budget=TokenBudget(
        max_input_tokens_per_request=8000,  # Büyük context'leri önle
        max_output_tokens_per_request=2000,  # Response uzunluğunu sınırla
        max_tokens_per_user_daily=100_000,  # ~$0.50/gün per user (GPT-4)
        max_tokens_per_team_monthly=10_000_000  # ~$100/ay per team
    )
)

def invoke_llm_with_budget(user_id: str, team_id: str, prompt: str):
    # Token'ları tahmin et (yaklaşık)
    estimated_input = len(prompt.split()) * 1.3  # Tokenization için hesapla
    estimated_output = 500  # Muhafazakar tahmin

    # Budget'ı kontrol et
    allowed, error = budget_enforcer.check_and_reserve(
        user_id, team_id, int(estimated_input), estimated_output
    )

    if not allowed:
        raise BudgetExceededError(error)

    # LLM'i çağır
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=budget_enforcer.budget.max_output_tokens_per_request
    )

    # Gerçek kullanımı kaydet
    budget_enforcer.record_actual_usage(
        user_id,
        team_id,
        response.usage.prompt_tokens,
        response.usage.completion_tokens,
        int(estimated_input),
        estimated_output
    )

    return response.choices[0].message.content

Alert Konfigürasyonu

def check_budget_alerts(user_id: str, team_id: str, redis_client, budget):
    """
    %70, %90, %100 budget threshold'larında alert tetikle
    """
    user_daily_key = f"budget:user:{user_id}:daily"
    user_tokens_today = int(redis_client.get(user_daily_key) or 0)

    daily_limit = budget.max_tokens_per_user_daily
    usage_percentage = (user_tokens_today / daily_limit) * 100

    if usage_percentage >= 100:
        send_alert(
            level="CRITICAL",
            message=f"User {user_id} günlük budget'ı aştı",
            action="BLOCK"
        )
    elif usage_percentage >= 90:
        send_alert(
            level="WARNING",
            message=f"User {user_id} günlük budget'ın %90'ında",
            action="NOTIFY"
        )
    elif usage_percentage >= 70:
        send_alert(
            level="INFO",
            message=f"User {user_id} günlük budget'ın %70'inde",
            action="MONITOR"
        )

Yaygın Budget Hataları

Tool-Call Storm’ları: Agent’lar limit olmadan 50+ tool çağırıyor, milyonlarca token tüketiyor

# Çözüm: max_tool_calls_per_turn ayarla
agent = Agent(
    tools=[get_product_details, get_reviews, get_pricing],
    max_tool_calls_per_turn=5,  # Hard limit
    instructions="Mümkün olduğunda batch query'ler kullan."
)

RAG Over-Retrieval: 5 chunk yeterken 50 chunk retrieve etmek

# KÖTÜ: Çok fazla chunk
retriever = VectorStoreRetriever(
    vector_store=vector_db,
    search_kwargs={"k": 50}  # 25,000 token context
)

# İYİ: Odaklanmış retrieval
retriever = VectorStoreRetriever(
    vector_store=vector_db,
    search_kwargs={"k": 5}  # 2,500 token (%90 azalma)
)

Optimizasyon Stratejisi 4: Semantic Caching

Geleneksel caching sadece tam eşleşen sorguları eşleştirir. Semantic caching, semantik olarak benzer sorular için response’ları cache’lemek üzere vector similarity kullanarak cache hit rate’ini dramatik şekilde artırıyor.

Vector Similarity ile Implementation

import hashlib
import json
from typing import Optional
import redis
from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticCache:
    def __init__(
        self,
        redis_client: redis.Redis,
        similarity_threshold: float = 0.95,
        ttl_seconds: int = 3600
    ):
        self.redis = redis_client
        self.similarity_threshold = similarity_threshold
        self.ttl_seconds = ttl_seconds

        # Semantic matching için hafif embedding model
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')

    def _get_embedding(self, text: str) -> np.ndarray:
        """Query için embedding vector oluştur"""
        return self.embedding_model.encode(text, normalize_embeddings=True)

    def _cosine_similarity(self, vec1: np.ndarray, vec2: np.ndarray) -> float:
        """İki vector arasında cosine similarity hesapla"""
        return np.dot(vec1, vec2)  # Vector'lar zaten normalize

    def get(self, query: str, system_prompt: str = "") -> Optional[dict]:
        """
        Semantik olarak benzer sorgu varsa cache'lenmiş response'u al
        """
        cache_key_prefix = f"semantic_cache:{hashlib.md5(system_prompt.encode()).hexdigest()}"

        # Bu system prompt için tüm cache'lenmiş sorguları al
        cached_keys = self.redis.keys(f"{cache_key_prefix}:*")

        if not cached_keys:
            return None

        query_embedding = self._get_embedding(query)

        best_match = None
        best_similarity = 0.0

        # En benzer cache'lenmiş sorguyu bul
        for key in cached_keys:
            cached_data = self.redis.get(key)
            if not cached_data:
                continue

            cached = json.loads(cached_data)
            cached_embedding = np.array(cached['embedding'])

            similarity = self._cosine_similarity(query_embedding, cached_embedding)

            if similarity > best_similarity:
                best_similarity = similarity
                best_match = cached

        # Similarity threshold'u aşarsa cache'lenmiş response'u döndür
        if best_similarity >= self.similarity_threshold:
            return {
                'response': best_match['response'],
                'similarity': best_similarity,
                'cached': True,
                'original_query': best_match['query']
            }

        return None

    def set(self, query: str, response: str, system_prompt: str = ""):
        """
        Query-response çiftini semantic embedding ile cache'le
        """
        cache_key_prefix = f"semantic_cache:{hashlib.md5(system_prompt.encode()).hexdigest()}"
        query_hash = hashlib.md5(query.encode()).hexdigest()
        cache_key = f"{cache_key_prefix}:{query_hash}"

        embedding = self._get_embedding(query)

        cache_data = {
            'query': query,
            'response': response,
            'embedding': embedding.tolist(),
            'timestamp': datetime.utcnow().isoformat()
        }

        self.redis.setex(
            cache_key,
            self.ttl_seconds,
            json.dumps(cache_data)
        )

# Production'da kullanım
semantic_cache = SemanticCache(
    redis_client=redis.Redis(host='localhost', decode_responses=False),
    similarity_threshold=0.95,  # %95 similarity gerekli
    ttl_seconds=3600  # 1 saat cache
)

def invoke_with_semantic_cache(query: str, system_prompt: str):
    # Önce semantic cache'i kontrol et
    cached = semantic_cache.get(query, system_prompt)

    if cached:
        print(f"Cache hit! Similarity: {cached['similarity']:.2%}")
        print(f"Original query: {cached['original_query']}")
        return cached['response']

    # Cache miss - LLM'i çağır
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": query}
        ]
    )

    result = response.choices[0].message.content

    # Gelecekteki semantik olarak benzer sorgular için cache'le
    semantic_cache.set(query, result, system_prompt)

    return result

# Örnek: Semantik olarak benzer sorgular
# Sorgu 1: "İade politikanız nedir?"
# Sorgu 2: "Paramı nasıl geri alırım?"
# Sorgu 3: "Ürünleri iade edip para iadesi alabilir miyim?"
# Üçü de > %95 similarity ile eşleşir ve cache'lenmiş response döner

Performans Etkisi

# Tam eşleşme caching: %10-20 hit rate (sadece özdeş sorgular)
# Semantic caching: %40-60 hit rate (semantik olarak benzer sorgular)
# Maliyet azalması: Müşteri destek/FAQ use case'leri için %40-60

# Trade-off'lar:
# - Embedding hesaplama: MiniLM için ~1ms (minimal overhead)
# - Redis memory: Cache'lenmiş sorgu başına ~384 byte
# - Similarity tuning: Çok düşük = yanlış cevaplar, çok yüksek = daha az hit

Maliyet İzleme ve Observability

Token tüketimine real-time görünürlük olmadan, maliyet problemleri fatura gelene kadar gizli kalıyor.

CloudWatch Metrics Implementation

import boto3
from datetime import datetime
from dataclasses import dataclass

@dataclass
class CostMetrics:
    timestamp: datetime
    model: str
    input_tokens: int
    output_tokens: int
    cached_tokens: int
    total_cost: float
    user_id: str
    team_id: str
    request_type: str  # 'simple', 'medium', 'complex'

class LLMCostTracker:
    def __init__(self):
        self.cloudwatch = boto3.client('cloudwatch')

        # Provider pricing (2025 güncel)
        self.pricing = {
            'gpt-4-turbo': {
                'input': 10.00 / 1_000_000,
                'output': 30.00 / 1_000_000
            },
            'gpt-4o': {
                'input': 2.50 / 1_000_000,
                'output': 10.00 / 1_000_000
            },
            'gpt-4o-mini': {
                'input': 0.15 / 1_000_000,
                'output': 0.60 / 1_000_000
            },
            'claude-sonnet-3.5': {
                'input': 3.00 / 1_000_000,
                'output': 15.00 / 1_000_000,
                'cached_input': 0.30 / 1_000_000  # %90 indirim
            }
        }

    def calculate_cost(self, metrics: CostMetrics) -> float:
        """Token kullanımı ve model pricing'e göre maliyeti hesapla"""
        pricing = self.pricing.get(metrics.model)
        if not pricing:
            raise ValueError(f"Bilinmeyen model: {metrics.model}")

        input_cost = metrics.input_tokens * pricing['input']
        output_cost = metrics.output_tokens * pricing['output']

        # Caching indirimini uygula (varsa)
        if metrics.cached_tokens > 0 and 'cached_input' in pricing:
            cached_cost = metrics.cached_tokens * pricing['cached_input']
            # Cached token'lar zaten input_tokens'da sayılmış, ayarla
            uncached_tokens = metrics.input_tokens - metrics.cached_tokens
            input_cost = (uncached_tokens * pricing['input']) + cached_cost

        return input_cost + output_cost

    def publish_metrics(self, metrics: CostMetrics):
        """Dashboard görselleştirmesi için CloudWatch'a metric'leri yayınla"""

        cost = self.calculate_cost(metrics)

        metric_data = [
            {
                'MetricName': 'TokenUsage',
                'Dimensions': [
                    {'Name': 'Model', 'Value': metrics.model},
                    {'Name': 'TokenType', 'Value': 'Input'}
                ],
                'Value': metrics.input_tokens,
                'Unit': 'Count',
                'Timestamp': metrics.timestamp
            },
            {
                'MetricName': 'TokenUsage',
                'Dimensions': [
                    {'Name': 'Model', 'Value': metrics.model},
                    {'Name': 'TokenType', 'Value': 'Output'}
                ],
                'Value': metrics.output_tokens,
                'Unit': 'Count',
                'Timestamp': metrics.timestamp
            },
            {
                'MetricName': 'LLMCost',
                'Dimensions': [
                    {'Name': 'Model', 'Value': metrics.model},
                    {'Name': 'Team', 'Value': metrics.team_id},
                    {'Name': 'RequestType', 'Value': metrics.request_type}
                ],
                'Value': cost,
                'Unit': 'None',  # Dolar
                'Timestamp': metrics.timestamp
            }
        ]

        # Caching kullanılıyorsa cache hit rate metric ekle
        if metrics.cached_tokens > 0:
            cache_hit_rate = (metrics.cached_tokens / metrics.input_tokens) * 100
            metric_data.append({
                'MetricName': 'CacheHitRate',
                'Dimensions': [{'Name': 'Model', 'Value': metrics.model}],
                'Value': cache_hit_rate,
                'Unit': 'Percent',
                'Timestamp': metrics.timestamp
            })

        self.cloudwatch.put_metric_data(
            Namespace='LLM/Costs',
            MetricData=metric_data
        )

    def create_cost_anomaly_alarm(self, threshold_dollars: float):
        """Maliyet anomalileri için CloudWatch alarm oluştur"""
        self.cloudwatch.put_metric_alarm(
            AlarmName='LLM-Daily-Cost-Anomaly',
            ComparisonOperator='GreaterThanThreshold',
            EvaluationPeriods=1,
            MetricName='LLMCost',
            Namespace='LLM/Costs',
            Period=86400,  # 24 saat
            Statistic='Sum',
            Threshold=threshold_dollars,
            ActionsEnabled=True,
            AlarmActions=[
                'arn:aws:sns:us-east-1:123456789:llm-cost-alerts'
            ],
            AlarmDescription=f'Günlük LLM maliyeti ${threshold_dollars} aştığında uyar'
        )

Önemli Metrikler Dashboard

CloudWatch Insights Query’leri:

-- Günlük model başına maliyet
fields @timestamp, model, sum(cost) as daily_cost
| filter namespace = "LLM/Costs"
| stats sum(daily_cost) by model, bin(@timestamp, 1d)

-- En pahalı 10 kullanıcı
fields user_id, sum(cost) as user_cost
| filter namespace = "LLM/Costs"
| stats sum(user_cost) by user_id
| sort user_cost desc
| limit 10

-- Cache etkinliği (maliyet tasarrufu)
fields @timestamp,
       sum(cached_tokens) / sum(input_tokens) * 100 as cache_hit_rate,
       sum(cached_tokens) * (standard_price - cached_price) as savings
| filter namespace = "LLM/Costs" and model = "claude-sonnet-3.5"
| stats avg(cache_hit_rate), sum(savings) by bin(@timestamp, 1h)

Yaygın Hatalar ve Öğrenilenler

Output Token Maliyetlerini Göz Ardı Etmek

Output token’ları input token’larından 2-5x daha pahalı, ancak optimizasyon genellikle sadece input’a odaklanıyor.

# Örnek: RAG uygulaması
# Input: 8,000 token (2,000 user query + 6,000 retrieved context)
# Output: 2,000 token (detaylı cevap)

# GPT-4 maliyet:
# - Input: 8,000 × $10/1M = $0.08
# - Output: 2,000 × $30/1M = $0.06
# Output token'ların %20'si olmasına rağmen toplam maliyetin %43'ü

# Çözüm: Agresif max_tokens limit'leri + özlülük prompt'ları
system_prompt = """
150 kelimenin altında öz cevaplar ver.
Kapsamlı detay yerine netliğe öncelik ver.
"""

response = openai.chat.completions.create(
    model="gpt-4",
    messages=[...],
    max_tokens=200  # Hard limit
)

Küçük Prompt Değişikliklerinden Cache Invalidation

Küçük prompt varyasyonları tüm cache’i invalidate ediyor, etkinliği yok ediyor.

# KÖTÜ: Dinamik timestamp cache'i her request'te invalidate eder
system_prompt = f"""
Sen bir destek ajanısın.
Şu anki zaman: {datetime.now().isoformat()}  # Her seferinde farklı!
[... prompt'un geri kalanı ...]
"""

# İYİ: Sadece statik içerik, dinamik context user message'da
system_prompt = """
Sen bir destek ajanısın.
[... statik politikalar ...]
"""

user_message = f"""
Şu anki zaman: {datetime.now().isoformat()}
Kullanıcı sorusu: {question}
"""

Fatura Şoku Gelinceye Kadar İzleme Yok

Observability olmadan production’a deploy etmek, problemleri hasar verdikten sonra keşfetmek demek.

Çalıştığım bir projeden örnek timeline:

Hafta 1: 100 request/gün ile POC = $50/ay
Hafta 2: 1,000 request/gün ile Beta = $500/ay
Hafta 3: 10,000 request/gün ile Production = $5,000/ay
Hafta 4: Özellik viral oldu, 50,000 request/gün = $25,000/ay

Çözüm: İlk günden instrument et, CloudWatch’a hemen metric’leri yayınla, launch’tan önce budget alert’leri ayarla.

Optimizasyon Etki Matrisi

Optimizasyon	Maliyet Tasarrufu	Kalite Etkisi	Implementation Çabası
Prompt caching	%50-90	Yok	Düşük (provider özelliği)
Model routing	%30-50	Düşük (%5-10 accuracy)	Orta (routing logic)
Output limit’leri	%20-40	Düşük (özlülük)	Düşük (parametre ayarı)
Semantic caching	%40-60	Orta (staleness)	Orta (vector DB kurulumu)
Token budget’ları	%10-30	Yok (israfı önler)	Orta (budget sistemi)
Batch inference	%50	Yok (sadece async)	Düşük (provider özelliği)

Önemli Çıkarımlar

Token-Based Faturalandırma Yeni Bir Düşünce Biçimi Gerektiriyor: Geleneksel cloud maliyetleri öngörülebilir ve lineer. LLM maliyetleri kullanım pattern’larına göre 100 kat değişiyor. Optimizasyon zorunlu.

Output Token’ları Daha Pahalı: max_tokens limit’leri ve özlülük için prompt engineering ile kısa response’lara odaklan.

Prompt Caching Kolay Kazanç: Cache’lenmiş token’larda %90 indirim (Anthropic), tipik uygulamalar için %50-70 maliyet azalması, sıfır kod değişikliği gerekli.

Model Routing Maliyet ve Kaliteyi Dengeler: Sorguların %70’i daha ucuz modeller kullanabilir. Intelligent routing %30-50 tasarruf sağlar. AWS Bedrock sıfır konfigürasyon routing sunuyor.

Observability Sürprizleri Önler: İlk günden tüm LLM çağrılarını instrument et. %70, %90, %100’de budget alert’leri ayarla. Haftalık metric’leri gözden geçir.

Optimizasyon Birleşir: Birden fazla tekniği birleştirmek, kaliteyi koruyarak toplam %60-80 maliyet azalması sağlayabiliyor.

LLM maliyet yönetimi geleneksel cloud FinOps’tan temelden farklı, ama bu pattern’ların sistematik uygulaması maliyetleri öngörülebilir ve kontrol edilebilir hale getiriyor.