👁️Gözlemlenebilirlik ve İzleme

LLM sistemlerinin izlenmesi, geleneksel yazılım izlemesinden farklıdır. Latency, token kullanımı, maliyet, kalite metrikleri ve güvenlik olaylarının entegre olarak takip edilmesi gerekmektedir.

6.1 Langfuse Entegrasyonu

Langfuse, Komtaş'ın birincil LLM observability platformudur. GKE üzerinde self-hosted olarak çalışmaktadır.

from langfuse.decorators import observe, langfuse_context
from langfuse import Langfuse
import time

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host=os.environ.get("LANGFUSE_HOST", "http://langfuse:3000")
)

# Decorator tabanlı tracing
@observe(name="rag-answer-generation")
async def generate_rag_answer(question: str, user_id: str) -> str:
    # Metadata ekle
    langfuse_context.update_current_trace(
        user_id=user_id,
        tags=["production", "rag", "customer-support"],
        metadata={"product_version": "2.1.0"}
    )
    
    # Retrieval span
    with langfuse_context.observe(name="retrieval"):
        contexts = await retrieve_contexts(question)
        langfuse_context.update_current_observation(
            metadata={"retrieved_chunks": len(contexts), "query": question}
        )
    
    # Generation span
    with langfuse_context.observe(name="generation"):
        answer = await generate_answer(question, contexts)
        langfuse_context.update_current_observation(
            usage={"input": count_tokens(question + str(contexts)), "output": count_tokens(answer)},
            metadata={"model": "claude-opus-4-6"}
        )
    
    # RAGAS skor ekle
    langfuse_context.score_current_trace(
        name="faithfulness",
        value=await compute_faithfulness(answer, contexts),
        comment="RAGAS faithfulness skoru"
    )
    
    return answer

# Manuel trace API
def log_agent_execution(agent_name: str, input_data: dict, output: str, duration_ms: int):
    trace = langfuse.trace(
        name=f"agent-{agent_name}",
        input=input_data,
        output=output,
        metadata={"duration_ms": duration_ms}
    )
    
    # Span ekle
    trace.span(
        name="tool-execution",
        start_time=datetime.utcnow(),
        metadata={"tools_used": input_data.get("tools_called", [])}
    )
    
    return trace.id

Prompt Yönetimi

# Langfuse prompt versiyonlama
from langfuse import Langfuse

langfuse = Langfuse()

# Prompt kaydet
langfuse.create_prompt(
    name="contract-analysis",
    prompt="""Sen Komtaş sözleşme uzmanısın.
    Aşağıdaki sözleşmeyi analiz et: {{contract_text}}
    
    Şunları belirle:
    1. Taraflar ve yükümlülükler
    2. Sona erme tarihi
    3. Ceza maddeleri
    4. Risk değerlendirmesi""",
    config={"model": "claude-opus-4-6", "temperature": 0.1},
    labels=["production"]
)

# Prompt kullan (her zaman production sürümünü çek)
def get_contract_prompt(contract_text: str) -> str:
    prompt_obj = langfuse.get_prompt("contract-analysis")
    
    # Langfuse observability'ye bağla
    langfuse_context.update_current_observation(
        prompt=prompt_obj  # Hangi prompt versiyonunun kullanıldığını trace et
    )
    
    return prompt_obj.compile(contract_text=contract_text)

6.2 Maliyet İzleme ve Optimizasyon

Tüm model kullanımı OpenRouter üzerinden yapılmaktadır. Fiyatlar OpenRouter API'sinden alınmıştır (1M token başına USD, Nisan 2025).

# ─── Maliyet Optimizasyonu — OpenRouter Tabanlı Model Yönlendirme ───
# Tüm modeller tek endpoint üzerinden erişilir: openrouter.ai/api/v1
import os, redis
from openai import OpenAI   # OpenRouter, OpenAI-uyumlu API sunar
from datetime import date

# OpenRouter istemcisi — base_url ile Anthropic/OpenAI/Google farkı ortadan kalkar
OPENROUTER_CLIENT = OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key=os.getenv("OPENROUTER_API_KEY"),
)

# Görev × karmaşıklık → model  (maliyet sırasına göre optimize edilmiş)
ROUTING_MATRIX = {
    # Yüksek hacim / basit → en düşük maliyet
    ("classification",  "low"):    "meta-llama/llama-3.3-70b-instruct",       # $0.10/$0.32
    ("extraction",      "low"):    "google/gemini-2.5-flash-lite",             # $0.10/$0.40
    ("summarization",   "low"):    "mistralai/mistral-small-3.2-24b-instruct", # $0.075/$0.20
    # Orta karmaşıklık → denge noktası
    ("classification",  "medium"): "google/gemini-2.5-flash",                  # $0.30/$2.50
    ("summarization",   "medium"): "anthropic/claude-sonnet-4.6",              # $3/$15
    ("code_generation", "medium"): "deepseek/deepseek-chat",                   # $0.32/$0.89
    ("data_analysis",   "medium"): "openai/gpt-4.1",                           # $2/$8
    ("translation",     "medium"): "qwen/qwen3-235b-a22b",                     # $0.46/$1.82
    # Kritik / yüksek kalite → premium modeller
    ("analysis",        "high"):   "anthropic/claude-sonnet-4.6",              # $3/$15
    ("legal_review",    "high"):   "anthropic/claude-opus-4.6",                # $5/$25
    ("reasoning",       "high"):   "openai/o3",                                # $2/$8
    ("code_review",     "high"):   "deepseek/deepseek-r1",                     # $0.70/$2.50
    ("strategic",       "high"):   "x-ai/grok-3",                              # $3/$15
    ("long_document",   "high"):   "google/gemini-2.5-pro",                    # $1.25/$10
}

# 1M token başına USD maliyet (OpenRouter, Nisan 2025)
MODEL_COSTS = {
    "meta-llama/llama-3.3-70b-instruct":         {"in": 0.100,  "out": 0.320},
    "google/gemini-2.5-flash-lite":              {"in": 0.100,  "out": 0.400},
    "mistralai/mistral-small-3.2-24b-instruct":  {"in": 0.075,  "out": 0.200},
    "google/gemini-2.5-flash":                   {"in": 0.300,  "out": 2.500},
    "anthropic/claude-sonnet-4.6":               {"in": 3.000,  "out": 15.00},
    "deepseek/deepseek-chat":                    {"in": 0.320,  "out": 0.890},
    "openai/gpt-4.1":                            {"in": 2.000,  "out": 8.000},
    "qwen/qwen3-235b-a22b":                      {"in": 0.455,  "out": 1.820},
    "anthropic/claude-opus-4.6":                 {"in": 5.000,  "out": 25.00},
    "openai/o3":                                 {"in": 2.000,  "out": 8.000},
    "deepseek/deepseek-r1":                      {"in": 0.700,  "out": 2.500},
    "x-ai/grok-3":                               {"in": 3.000,  "out": 15.00},
    "google/gemini-2.5-pro":                     {"in": 1.250,  "out": 10.00},
}

def route_to_model(task_type: str, complexity: str) -> str:
    """Görev türü ve karmaşıklığa göre en uygun OpenRouter modelini döndürür."""
    return ROUTING_MATRIX.get((task_type, complexity), "anthropic/claude-sonnet-4.6")

def estimate_cost_usd(model: str, input_tokens: int, output_tokens: int) -> float:
    """Tahmini maliyeti USD cinsinden hesaplar. (1M token başına fiyat × kullanım)"""
    c = MODEL_COSTS.get(model, {"in": 3.0, "out": 15.0})
    return (input_tokens * c["in"] + output_tokens * c["out"]) / 1_000_000

def call_via_openrouter(task_type: str, complexity: str,
                        prompt: str, max_tokens: int = 2048) -> dict:
    """Model seçimi, API çağrısı ve maliyet takibini tek fonksiyonda birleştirir."""
    model = route_to_model(task_type, complexity)
    response = OPENROUTER_CLIENT.chat.completions.create(
        model=model,
        max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompt}],
        extra_headers={              # OpenRouter'ın zorunlu header'ları
            "HTTP-Referer": "https://komtas.com",
            "X-Title": "Komtas CoPilot",
        }
    )
    cost = estimate_cost_usd(model,
                             response.usage.prompt_tokens,
                             response.usage.completion_tokens)
    return {
        "content":   response.choices[0].message.content,
        "model":     model,
        "tokens":    response.usage.total_tokens,
        "cost_usd":  round(cost, 6),
    }

# ─── Token / Maliyet Bütçe Takibi (Redis tabanlı, USD bazlı) ───
class TokenBudgetTracker:
    def __init__(self, daily_limit_usd: float = 50.0):
        self.daily_limit_usd = daily_limit_usd
        self.redis = redis.Redis(host="redis", port=6379, decode_responses=True)

    def record_usage(self, user_id: str, model: str,
                     input_tokens: int, output_tokens: int) -> dict:
        """Kullanımı kaydeder; günlük USD limitini aşarsa exception fırlatır."""
        cost = estimate_cost_usd(model, input_tokens, output_tokens)
        key   = "cost_budget:%s:%s" % (user_id, date.today())
        pipe  = self.redis.pipeline()    # Atomik işlem — race condition önler
        pipe.incrbyfloat(key, cost)
        pipe.expire(key, 86400)          # 24 saat TTL — günlük sıfırlama
        cumulative = float(pipe.execute()[0])
        if cumulative > self.daily_limit_usd:
            raise Exception("Günlük $%.0f limiti aşıldı: $%.4f" % (self.daily_limit_usd, cumulative))
        return {"cost_usd": cost, "daily_total_usd": round(cumulative, 4)}

# ─── Kullanım Örneği ───
result = call_via_openrouter(
    task_type="legal_review", complexity="high",
    prompt="Aşağıdaki sözleşmeyi KVKK uyumluluğu açısından değerlendir..."
)
print("Model: %s  |  Token: %d  |  Maliyet: $%.4f" % (result['model'], result['tokens'], result['cost_usd']))

6.3 SLO/SLA Tanımları ve Denetim Kaydı

# KVKK/GDPR uyumlu denetim kaydı
import hashlib
from datetime import datetime, timezone

class ImmutableAuditLogger:
    """Değiştirilemez denetim kaydı — KVKK, GDPR, SOC2 uyumlu"""
    
    def __init__(self, storage_client):
        self.storage = storage_client  # GCS bucket
        self.previous_hash = None
    
    async def log(self, event: dict):
        """Her log kaydı önceki kaydın hash'ini içerir (blockchain benzeri)"""
        
        record = {
            "event_id": str(uuid.uuid4()),
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "event_type": event.get("type"),
            "user_id": event.get("user_id"),
            "session_id": event.get("session_id"),
            "action": event.get("action"),
            # PII içermez — sadece anonimleştirilmiş referanslar
            "data_categories": event.get("data_categories", []),
            "legal_basis": event.get("legal_basis", "legitimate_interest"),
            "previous_hash": self.previous_hash,
            "retention_until": (datetime.now() + timedelta(days=730)).isoformat(),  # KVKK: 2 yıl
        }
        
        # Hash zinciri
        record_json = json.dumps(record, sort_keys=True, ensure_ascii=False)
        current_hash = hashlib.sha256(record_json.encode()).hexdigest()
        record["hash"] = current_hash
        self.previous_hash = current_hash
        
        # GCS'e yaz (immutable bucket policy ile)
        blob_name = f"audit/{datetime.now().strftime('%Y/%m/%d')}/{record['event_id']}.json"
        await self.storage.upload(blob_name, record_json)
        
        return record["event_id"]