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

Production’da large language model çalıştırmak, çoğu cloud ekibinin güvendiği maliyet modelini bozuyor. Compute-hour faturalandırma öngörülebilir, token faturalandırma değil. Aynı özellik, prompt’un taşıdığı context’e, soruyu hangi modelin yanıtladığına ve agent’ın durmadan önce kaç tool çağırdığına bağlı olarak çağrı başına kuruşun altında da kalabiliyor, birkaç dolara da çıkabiliyor.

Başlangıç için doğru varsayılan, prompt caching ile katı bir output limiti. İkisi de provider özelliği ve ikisi de bir öğleden sonra sürüyor. Caching yanıtın içeriğine dokunmuyor; output limiti dokunuyor, çünkü üretim tavana gelince duruyor ve yapılandırılmış çıktı yarım dönebiliyor. Bu tavanı, özelliğin gerçekten ihtiyaç duyduğu yanıt uzunluklarına göre belirleyin. Model routing ile semantic caching bunlardan sonra geliyor; çünkü tasarrufu yanıt kalitesiyle takas ediyorlar ve production’a girmeden önce bir değerlendirme setinden geçmeleri gerekiyor.

Token-Based Faturalandırma Zorluğu#

Tek bir kötü tasarlanmış prompt, binlerce optimize edilmiş request’ten daha fazla token tüketebiliyor. Aşağıdaki örnekler karşılaştırmanın tek ölçekte kalması için her şeyi gpt-4-turbo fiyatlarıyla hesaplıyor (1M input token $10, 1M output token $30).

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

# Basit sorgu: "What's the weather?"
# Input: 50 token (user message + system prompt)
# Output: 30 token
# Maliyet: (50 * $10/1M) + (30 * $30/1M) = $0.0014

# Karmaşık RAG sorgusu: "Q4 satış trendlerini analiz et ve strateji öner"
# Input: 8,050 token (50 user + 6,000 system + 2,000 RAG context)
# Output: 500 token
# Maliyet: (8,050 * $10/1M) + (500 * $30/1M) = $0.0955 (68x daha pahalı)

# Tool-call storm: agent bir turn'de 20 tool çağırıyor
# Tool çağrıları boyunca input: 8,050 * 20 = 161,000 token
# Tool çağrıları boyunca output: 500 * 20 = 10,000 token
# Maliyet: (161,000 * $10/1M) + (10,000 * $30/1M) = $1.91 (1,364x daha pahalı)

Faturanın bir büyüklük mertebesi artması için mimaride hiçbir şeyin değişmesi gerekmiyor. Aynı özelliğin daha sık çağrılması ve daha uzun context taşıması yeterli. Makasın tam da uygulama proof-of-concept’ten production’a geçerken açılmasının sebebi bu.

Provider Pricing Modelleri#

AWS Bedrock, OpenAI ve Anthropic, aynı token’ı workload’a göre hangisinin daha ucuza geleceğini değiştirecek kadar farklı fiyatlıyor.

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)

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.

Converse API’de Cache Checkpoint’leri#

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-5-20250929-v1:0",
        messages=[
            {
                "role": "user",
                "content": [{"text": user_message}]
            }
        ],
        system=[
            {
                "text": SYSTEM_PROMPT,
                # Varsayılan TTL 5 dakika. Sonnet 4.5'te bir checkpoint'in
                # gerçekten cache'lemesi için en az 4,096 token gerekiyor.
                "cachePoint": {"type": "default"}
            }
        ]
    )

    # Maliyet analizi
    usage = response['usage']

    # İlk çağrı: 50 cache'siz input token, 10,000 token cache'e yazılıyor
    # (write'lar temel input ücretinin 1.25 katı üzerinden faturalanıyor)
    # TTL içindeki sonraki çağrılar: 50 cache'siz, 10,000 indirimli okuma

    # inputTokens sadece cache'siz token'ları sayıyor, toplam input:
    # inputTokens + cacheReadInputTokens + cacheWriteInputTokens
    print(f"Uncached input tokens: {usage.get('inputTokens', 0)}")
    print(f"Cache read tokens: {usage.get('cacheReadInputTokens', 0)}")
    print(f"Cache write tokens: {usage.get('cacheWriteInputTokens', 0)}")
    print(f"Output tokens: {usage.get('outputTokens', 0)}")

    return response

Maliyet Etki Analizi#

# Aynı 10,000 token'lık system prompt'a 100 request, her birinde 200
# token output, 1M input $3 ve 1M output $15 fiyatıyla

# Caching olmadan:
# (10,050 * 100 * $3/1M) + (200 * 100 * $15/1M) = $3.32

# Caching ile, TTL penceresi içinde 1 write ve 99 read:
# Write request: (50 * $3/1M) + (10,000 * $3.75/1M) + (200 * $15/1M) = $0.0407
# Read request'ler: (50 * $3/1M) + (10,000 * $0.30/1M) + (200 * $15/1M) = $0.0062
# Toplam: $0.0407 + (99 * $0.0062) = $0.65
# Azalma: %80

# Write premium'u, seyrek trafikte caching'in maliyeti artırabilmesinin
# sebebi: okunmadan süresi dolan bir cache saf 1.25 kat ek ücret demek.

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: TTL her hit’te sıfırlanıyor; düzenli trafik cache’i sıcak tutuyor, boşluklar ise süresinin dolmasına yol açıyor
• 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}
"""

Intelligent Model Routing#

Tüm sorular en güçlü (ve pahalı) modeli gerektirmiyor. Complexity’e göre route etmek, trafiğin basit çoğunluğunu daha ucuz bir modele taşıyor; tasarruf da iki model arasındaki fiyat farkı ile güvenle taşıyabildiğiniz trafik oranının çarpımı. Bu oranı, production’da güvenmeden önce router’ın arkasındaki bir değerlendirme setiyle ölçün.

Complexity Skoruna Göre Routing#

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 aynı fikri yönetilen bir endpoint olarak sunuyor. Prompt router her request için yanıt kalitesini tahmin ediyor ve aynı aileden tam olarak iki model arasında seçim yapıyor:

import boto3

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

# Router ARN'leri ListPromptRouters'dan veya konsoldaki Prompt Routers
# sayfasından geliyor. Bir router aynı aileden iki model arasında seçiyor.
router_arn = bedrock.list_prompt_routers()['promptRouterSummaries'][0]['promptRouterArn']

response = runtime.converse(
    modelId=router_arn,
    messages=[
        {
            "role": "user",
            "content": [{"text": "Seattle'da hava nasıl?"}]
        }
    ]
)

# Yanıt, request'i hangi modelin karşıladığını bildiriyor.
# Production'da güvenmeden önce iki kısıt:
# - Routing sadece İngilizce prompt'lar için optimize edilmiş durumda.
# - Tasarruf, iki model arasındaki fiyat farkı ile ucuz modele düşen
#   trafik oranının çarpımı; oranı varsaymak yerine ölçün.

Beklenen Sonuçlar#

Aşağıdaki diyagram 60/30/10’luk bir trafik dağılımı varsayıyor. Dağılım değişince routing’in bütün gerekçesi de baştan değişiyor.

Token Budget Enforcement#

Sınırsız token tüketimi maliyet fırtınalarına yol açıyor; hard limit’ler normal request’leri engellemeden bu fırtınaları durduruyor.

Token Kontrolü ve Rezervasyonu#

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,  # 1M input $10 ile ~$1/gün per user
        max_tokens_per_team_monthly=10_000_000  # aynı ücretle ~$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)
)

Semantic Caching#

Geleneksel caching sadece tam eşleşen sorguları eşleştiriyor. Semantic caching ise vector similarity kullanarak aynı anlama gelen soruların tek bir yanıtı paylaşmasını sağlıyor; tekrar eden trafikte hit rate’i böyle yükseltiyor. Beraberinde iki sorunun benzer görünüp aslında benzer olmaması riskini de getiriyor, bu yüzden similarity threshold’u teknik bir ayar olduğu kadar ürün kararı.

Vector Similarity ile Eşleştirme#

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#

# Hit rate trafiğin bir özelliği. Destek kuyrukları ve FAQ
# endpoint'leri birkaç intent etrafında sıkı kümeleniyor; açık uçlu
# asistanlar neredeyse hiç kümelenmiyor. Tasarruf bütçelemeden önce bir
# günlük gerçek sorguyu matcher üzerinden tekrar oynatın.

# Trade-off'lar:
# - Embedding hesaplama: küçük bir lokal modelde tek bir ek forward
#   pass, engellediği LLM çağrısının yanında ihmal edilebilir
# - Redis memory: MiniLM-L6 için entry başına 384 float boyut,
#   4 byte'lık float'larla yaklaşık 1.5 KB
# - Similarity tuning: çok düşük = kendinden emin yanlış cevaplar,
#   çok yüksek = cache hiç tutmuyor

Maliyet İzleme ve Observability#

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

CloudWatch’a Maliyet Metriği Yayınlama#

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:123456789012:llm-cost-alerts'
            ],
            AlarmDescription=f'Günlük LLM maliyeti ${threshold_dollars} aştığında uyar'
        )

Metrik Panosu#

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)

Token maliyeti request hacmini neredeyse lineer takip ediyor; bu da büyümeyi hem kolay öngörülebilir hem de kolay gözden kaçar hale getiriyor:

100 request/gün    = $50/ay
1,000 request/gün  = $500/ay
10,000 request/gün = $5,000/ay
50,000 request/gün = $25,000/ay

İlk günden instrument edin, metric’leri aynı değişiklikte CloudWatch’a yayınlayın ve budget alert’lerini ilk sürprizden sonra değil launch’tan önce ayarlayın.

Optimizasyon Etki Matrisi#

Tasarruf trafik dağılımınıza bağlı olduğu için asıl karşılaştırma, her tekniğin hangi kaldıraca bastığı ve bunun kalite ile emek olarak neye mal olduğu. İlk üç satır hemen her workload’da güvenle açılabilir; son üç satır muhakeme gerektiriyor.

Optimizasyon	Maliyet kaldıracı	Kalite etkisi	Implementation çabası
Prompt caching	Tekrarlanan prefix cache-read ücretiyle faturalanıyor	Yok	Düşük (provider özelliği)
Batch inference	Provider batch indirimi (Bedrock’ta %50)	Yok, sadece async	Düşük (provider özelliği)
Output limit’leri	Faturanın pahalı tarafı output	Düşük, cevaplar kısalıyor	Düşük (parametre ayarı)
Model routing	Trafiğin basit kısmını ucuz model karşılıyor	Değerlendirme seti gerekiyor	Orta (routing logic)
Semantic caching	Neredeyse aynı sorularda çağrıyı tamamen atlıyor	Orta, bayatlama ve yanlış eşleşme	Orta (vector store)
Token budget’ları	Kontrolsüz döngülerin etki alanını sınırlıyor	Yok, israfı önlüyor	Orta (budget sistemi)

Bu Yaklaşım Nerede Geçerli#

Varsayılan sıralama; trafik hacmi, gereken yanıt uzunluğu ya da görev şekli bu varsayımlardan birini bozana kadar geçerliliğini koruyor.

Trafik, TTL dolmadan cache’e ulaşamayacak kadar seyrekse 1.25 katlık write premium’u caching’i net zarara çeviriyor; burada batch inference ya da daha küçük bir model daha iyi kaldıraç. Cevapların uzun ve ayrıntılı kalması gerekiyorsa output limit’leri ürüne karşı çalışıyor, emek routing’e gitmeli. Workload genel bir asistan değil de tek bir dar görevse routing’i tamamen atlayın: değerlendirme setinizi geçen en ucuz modeli seçin ve tasarruf ettiğiniz karmaşıklığı başka yere harcayın. Bir router ancak trafik gerçekten karışıksa işe yarar.

Kaynaklar#

• FinOps Çerçevesi Genel Bakış (yeni sekmede açılır) - Bulut finansal yönetimi için Bildir, Optimize Et ve Çalıştır yaşam döngüsünü kapsayan FinOps Foundation çalışma modeli
• FinOps İlkeleri (yeni sekmede açılır) - 2025’te yapay zeka ve bulut dışı maliyet kapsamlarını içerecek şekilde güncellenen FinOps uygulamasının altı temel ilkesi
• AWS Faturalandırma ve Maliyet Yönetimi Nedir? (yeni sekmede açılır) - Cost Explorer, Budgets ve Cost Optimization Hub’ın AWS dokümantasyonu genel bakışı
• AWS Cost Explorer (yeni sekmede açılır) - Servise, etikete ve zaman dilimine göre AWS harcamalarını analiz etmek ve görselleştirmek için servis dokümantasyonu
• AWS Budgets (yeni sekmede açılır) - Yapay zeka ve makine öğrenmesi iş yükü maliyet kontrolü için otomatik uyarılarla maliyet ve kullanım bütçesi eşiklerini belirleme
• AWS Cost Optimization Hub (yeni sekmede açılır) - Reserved Instance’lar ve Savings Plan’larını hesaba katan doğru boyutlandırma ve tasarruf önerilerinin birleşik görünümü
• Daha hızlı model çıkarımı için prompt caching (yeni sekmede açılır) - Amazon Bedrock cache checkpoint’leri, model başına token alt sınırları, TTL davranışı ve Converse API’nin döndürdüğü usage alanları
• Amazon Bedrock’ta intelligent prompt routing (yeni sekmede açılır) - Varsayılan ve yapılandırılmış router’ların aynı aileden iki model arasında nasıl seçim yaptığı, sadece İngilizce için optimize edilmiş olma kısıtı dahil
• Amazon Bedrock fiyatlandırması (yeni sekmede açılır) - Model başına token ücretleri, batch inference indirimi ve yukarıdaki maliyet hesaplarında kullanılan cache okuma/yazma oranları
• Anthropic fiyatlandırması (yeni sekmede açılır) - Claude model ailesi için yayımlanan input ve output token ücretleri
• OpenAI API fiyatlandırması (yeni sekmede açılır) - Routing örneğinde kullanılan GPT modelleri için yayımlanan token ücretleri