Skip to content
~/sph.sh

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

Token-based pricing, production LLM uygulamaları için benzersiz maliyet zorlukları yaratır. Prompt caching, model routing ve token budget'ları ile kaliteden ödün vermeden maliyetleri %60-80 azaltmak için sistematik optimizasyon stratejilerini öğren.

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

python
# 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: 3input,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ı

python
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: 15input,15 input, 75 output per 1M token
  • Claude Opus 4.5: 5input,5 input, 25 output per 1M token (daha yeni, daha maliyet-etkin)
  • Claude Sonnet 4.6: 3input,3 input, 15 output per 1M token
  • Claude Haiku 3: 0.25input,0.25 input, 1.25 output per 1M token
  • Claude Haiku 4.5: 1input,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

python
import boto3import 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

python
# 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
python
# KÖTÜ: Dinamik içerik cache'i invalidate edersystem_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'dasystem_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

typescript
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ğiconst router = new IntelligentRouter();
// Basit sorgu -> gpt-4o-miniawait router.invoke(  "İade politikanız nedir?",  "Sen bir müşteri destek ajanısın");
// Karmaşık sorgu -> gpt-4-turboawait 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:

python
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 ederPROMPT_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

python
from dataclasses import dataclassfrom datetime import datetime, timedeltafrom typing import Dict, Optionalimport redis
@dataclassclass 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
@dataclassclass 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ımbudget_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

python
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

python
# Çözüm: max_tool_calls_per_turn ayarlaagent = 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

python
# KÖTÜ: Çok fazla chunkretriever = VectorStoreRetriever(    vector_store=vector_db,    search_kwargs={"k": 50}  # 25,000 token context)
# İYİ: Odaklanmış retrievalretriever = 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

python
import hashlibimport jsonfrom typing import Optionalimport redisfrom sentence_transformers import SentenceTransformerimport 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ımsemantic_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

python
# 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

python
import boto3from datetime import datetimefrom dataclasses import dataclass
@dataclassclass 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:

sql
-- Günlük model başına maliyetfields @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.

python
# Ö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.

python
# KÖTÜ: Dinamik timestamp cache'i her request'te invalidate edersystem_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'dasystem_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/ayHafta 2: 1,000 request/gün ile Beta = $500/ayHafta 3: 10,000 request/gün ile Production = $5,000/ayHafta 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

OptimizasyonMaliyet TasarrufuKalite EtkisiImplementation Çabası
Prompt caching%50-90YokDüşük (provider özelliği)
Model routing%30-50Düşük (%5-10 accuracy)Orta (routing logic)
Output limit'leri%20-40Düşük (özlülük)Düşük (parametre ayarı)
Semantic caching%40-60Orta (staleness)Orta (vector DB kurulumu)
Token budget'ları%10-30Yok (israfı önler)Orta (budget sistemi)
Batch inference%50Yok (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.

aws
finops
llm
cost-optimization
bedrock
openai
observability
performance

İlgili Yazılar