AI Projelerine Giriş: Heyecan ve Gizli Maliyetler
Son birkaç yıldır AI, hem ana projelerimde hem de yan ürünlerimde oldukça merkezi bir rol oynamaya başladı. İlk başta “vay be, ne kadar hızlı sonuç alıyorum” diye başladığım bu yolculukta, zamanla bazı “sessiz hataların” farkına vardım. Bu hatalar, doğrudan bir 500 Internal Server Error vermiyor, sistem crash etmiyor ama benim değerli zamanımı, bazen de cebimden çıkan parayı yavaş yavaş eritiyordu. Sanki arkadan bir musluk açık kalmış da ben farkında değilmişim gibiydi. Özellikle kendi finansal hesaplayıcılarımı geliştirdiğim yan ürünümde veya Android spam uygulamamda AI’ı entegre ederken bu türden birçok şeyle karşılaştım. Bu yazıda, bu gizli tuzakları ve onlardan nasıl ders çıkardığımı anlatmak istiyorum.
Bu durum, yıllardır sistem ve network tarafında edindiğim “her şeyin bir maliyeti var” prensibini AI dünyasında da farklı bir boyutta görmeme neden oldu. Bir VLAN segmentasyonu yaparken yanlış hesapladığım IP aralıkları nasıl sonra başımı ağrıttıysa, AI tarafında da yanlış bir varsayım saatlerimi yiyebiliyor. Bir yan ürünümde AI ile üretim planlama yaparken, ilk başta her şey yolunda görünüyordu. Ancak gerçek dünya verisiyle karşılaştığında, AI’ın “sağduyu” eksikliği veya ince detayları kaçırması, haftalar süren debug süreçlerine yol açtı. Özetle, AI size hızlı bir başlangıç verse de, detaylardaki şeytanları gözden kaçırmak, uzun vadede çok daha büyük bedellere mal olabiliyor.
API Maliyetlerini Anlamak ve Kontrol Altına Almak
AI modellerinin API’lerini kullanmaya başladığımda, başlangıçta token maliyetlerini hafife aldım. Küçük denemelerde maliyetler düşük gibi görünse de, kendi yan ürünümde finansal verileri analiz eden bir modül geliştirdiğimde, API çağrılarının faturaları hızla tavan yapmaya başladı. Özellikle prompt’ların ve model çıktılarının uzunluğu arttıkça, token kullanımı katlanarak artıyordu. Bir ayın sonunda gelen faturayı görünce şaşırmıştım. Bu, tıpkı bir zamanlar bir müşteri projesinde, yanlış konfigüre edilmiş bir Redis instance’ının sürekli OOM eviction policy’si yüzünden gereksiz yere CPU yakıp faturayı yükseltmesi gibiydi; görünürde her şey çalışıyor ama arka planda kaynaklar boşa gidiyor.
Bu durumu kontrol altına almak için ilk adımı, API çağrılarını ve token kullanımını detaylı bir şekilde loglamak oldu. Her çağrıda giden ve gelen token sayısını kaydederek, hangi senaryoların daha maliyetli olduğunu anlamaya çalıştım. Ardından, prompt’ları optimize etmeye başladım. Gereksiz kelimeleri atmak, daha kısa ve öz ifadeler kullanmak, modelden beklediğim çıktıyı mümkün olduğunca daraltmak gibi yöntemlerle token kullanımını düşürdüm. Örneğin, karmaşık bir finansal hesaplama senaryosunda, başlangıçta tüm finansal metinleri modele gönderirken, sonradan sadece gerekli veri noktalarını ve formülleri göndererek önemli ölçüde token tasarrufu sağladım.
import tiktoken
import openai
def count_tokens(text: str, model: str = "gpt-4"):
"""Metindeki token sayısını hesaplar."""
encoding = tiktoken.encoding_for_model(model)
return len(encoding.encode(text))
def analyze_cost(prompt: str, response: str, model: str = "gpt-4"):
"""Prompt ve response'un maliyetini (tahmini) hesaplar."""
input_tokens = count_tokens(prompt, model)
output_tokens = count_tokens(response, model)
# Örnek maliyetler (gerçek fiyatlar API sağlayıcısına göre değişir)
# Fiyatlar 2024 yılına ait, güncel fiyatları kontrol edin!
cost_per_input_k_tokens = 0.01 # $0.01 / 1K input tokens
cost_per_output_k_tokens = 0.03 # $0.03 / 1K output tokens
total_cost = (input_tokens / 1000 * cost_per_input_k_tokens) + \
(output_tokens / 1000 * cost_per_output_k_tokens)
print(f"Input Tokens: {input_tokens}, Output Tokens: {output_tokens}")
print(f"Tahmini Maliyet: ${total_cost:.4f}")
return total_cost
# Örnek kullanım
my_prompt = "Türkiye'de 2025 yılı için beklenen enflasyon oranı hakkında detaylı bir analiz yap ve bu analizi 500 kelimeyi geçmeyecek şekilde özetle."
my_response = "..." # Modelden gelen cevap
# analyze_cost(my_prompt, my_response, model="gpt-3.5-turbo")Bu tür bir maliyet analizi, hangi prompt’ların ve hangi iş akışlarının daha pahalı olduğunu net bir şekilde görmemi sağladı. Özellikle PostgreSQL performans tuning yazımda bahsettiğim gibi, veritabanı sorgularını optimize etmekle aynı mantık; gereksiz yükü ortadan kaldırmak.
Etkili Prompt Engineering: Deneme Sürecini Kısaltma Yolları
Prompt engineering, ilk başta basit gibi görünse de, istediğiniz çıktıyı tam olarak almak için harcadığınız zaman, AI projelerinin en sinsi maliyetlerinden biri olabiliyor. Bir üretim firmasının ERP’sinde AI ile üretim planlama modellerini test ederken, aynı çıktıyı farklı prompt’larla almak için saatlerimi harcadığımı bilirim. Bazen tek bir kelimenin, hatta bir noktalama işaretinin bile modelin davranışını tamamen değiştirebildiğini gördüm. Bu, tıpkı bir firewall politikasında ANY yerine spesifik portları tanımlarken harcadığım çabaya benziyordu; küçük bir detay, büyük bir fark yaratıyor. Kendi yan ürünümde bir prompt üzerinde 3-4 saat çalıştığım, sonra da “Acaba başka bir prompt daha mı iyi olurdu?” diye şüpheye düştüğüm çok oldu.
Bu deneme sürecini kısaltmak için kendime bazı prensipler edindim. İlk olarak, prompt’u olabildiğince açık ve net yazmaya çalışıyorum. Modelden ne beklediğimi, hangi formatta istediğimi ve hangi kısıtlamalara uyması gerektiğini baştan belirtiyorum. İkincisi, iteratif bir yaklaşım kullanıyorum. Küçük değişiklikler yapıp çıktıyı hemen kontrol ediyorum. Tüm prompt’u baştan yazmak yerine, adım adım iyileştiriyorum. Son olarak, prompt versiyonlaması yapıyorum. Kötü çalışan bir prompt’a geri dönebilmek veya farklı prompt’ları karşılaştırabilmek için değişiklikleri kaydediyorum. Bu, CI/CD reliability süreçlerindeki kod versiyonlamasına çok benziyor.
Özellikle RAG (Retrieval-Augmented Generation) tabanlı sistemlerde prompt’un kalitesi, çekilen bilginin doğru bir şekilde sentezlenmesi için hayati önem taşıyor. Yanlış veya eksik bir prompt, doğru bilgiyi çekseniz bile modelin “halüsinasyon” görmesine neden olabiliyor.
RAG Sistemlerinde Veri Bütünlüğü ve Güncelliği Sağlama
Retrieval-Augmented Generation (RAG) mimarileri, AI modellerinin güncel ve spesifik verilere erişimini sağlamak için harika bir yol. Ancak bu sistemleri kendi yan ürünümde kullanırken, veri tazeği ve bütünlüğü konusunda ciddi sorunlar yaşadım. Özellikle finansal hesaplayıcılarımda, veritabanından çekilen bilgilerin güncelliği hayati önem taşıyor. Bir gün, bir kullanıcının finansal hesaplamalarının yanlış çıktığını fark ettim. Sebebini araştırdığımda, RAG’ın kullandığı vektör veritabanındaki verilerin, ana veritabanındaki son değişikliklerden 48 saat geride olduğunu gördüm. Bu, tıpkı bir zamanlar bir bankanın iç platformunda, replikasyon gecikmeleri yüzünden raporların yanlış veriler göstermesi gibiydi; veri tutarsızlığı, güvenilirliği doğrudan etkiliyor.
Bu sorunu çözmek için, veri senkronizasyon stratejilerimi yeniden gözden geçirdim. İlk başta basit bir cron işiyle günlük senkronizasyon yaparken, bunu daha olay tabanlı bir yaklaşıma çevirdim. Ana veritabanında kritik bir değişiklik olduğunda, ilgili veriyi anında vektör veritabanına indeksleyen veya güncelleyen bir mekanizma kurdum. Bu, CDC (Change Data Capture) benzeri bir yaklaşımdı. Ayrıca, RAG’ın çektiği verinin “yaşını” kontrol eden bir mekanizma da ekledim. Eğer çekilen veri belirli bir eşiğin üzerinde eskiyse, modelden bu durumu belirtmesini veya daha güncel bir kaynak aramasını istedim.
# Örnek: Basit bir veri tazeği kontrolü
import datetime
def get_data_last_updated(record_id: str) -> datetime.datetime:
"""Veritabanından kaydın son güncellenme tarihini çeker."""
# Gerçek uygulamada buradan DB sorgusu yapılır.
# Örnek için rastgele bir tarih dönelim
if record_id == "finansal_bilgi_123":
return datetime.datetime.now() - datetime.timedelta(hours=2) # 2 saat önce
return datetime.datetime.now() - datetime.timedelta(days=3) # 3 gün önce
def check_data_freshness(record_id: str, max_stale_hours: int = 24) -> bool:
"""Verinin belirli bir taze eşiği içinde olup olmadığını kontrol eder."""
last_updated = get_data_last_updated(record_id)
if not last_updated:
return False # Veri bulunamadı
current_time = datetime.datetime.now()
stale_duration = current_time - last_updated
if stale_duration.total_seconds() / 3600 > max_stale_hours:
print(f"UYARI: {record_id} verisi {stale_duration.days} gün {stale_duration.seconds // 3600} saat önce güncellendi. Eşik: {max_stale_hours} saat.")
return False
print(f"{record_id} verisi taze.")
return True
# check_data_freshness("finansal_bilgi_123", max_stale_hours=12)
# check_data_freshness("eski_finansal_bilgi_456", max_stale_hours=12)Bu tür kontroller, RAG sistemlerinin güvenilirliğini artırmada kritik rol oynuyor. Aksi takdirde, model ne kadar iyi olursa olsun, yanlış veya eski bilgiyle “halüsinasyon” görmesi kaçınılmaz oluyor. Bu, tıpkı network tarafında DNS negative caching yüzünden eski IP adreslerine istek atılması gibi, altta yatan veri katmanındaki sorunların sistemin tamamını etkilemesi demek.
Agent Pattern’lerinde Stabiliteyi Güvence Altına Alma
AI agent pattern’leri, otonom görevleri yerine getirme potansiyeliyle beni her zaman heyecanlandırmıştır. Kendi görev yönetim uygulamamda, kullanıcıdan gelen doğal dil girdilerini anlayıp otomatik olarak alt görevler oluşturan ve bunları belirli bir sıraya göre tamamlamaya çalışan bir agent geliştirmeye çalıştım. İlk denemeler harikaydı, basit görevleri başarıyla yerine getiriyordu. Ancak daha karmaşık senaryolarda, agent’ın bir türlü sona ulaşamadığını, aynı adımları tekrar tekrar denediğini ve sonunda bir “sonsuz döngü”ye girdiğini fark ettim. Bu, CPU’yu %100’e çıkarıyor ve sistem kaynaklarını tüketiyordu. Bu durum, bir üretim ERP’sinde bir iş akışının, bir koşulun yanlış tanımlanması yüzünden sürekli aynı adımı tekrar etmesi ve kilitlenmesi gibiydi; akışın kontrolünü kaybetmek, ciddi performans sorunlarına yol açıyor.
Bu tür sonsuz döngüleri engellemek için agent tasarımına bazı kontrol mekanizmaları ekledim. İlk olarak, her agent adımını loglayarak hangi adımların tekrarlandığını görmeye çalıştım. İkincisi, bir “adım sayacı” ve “zaman aşımı” mekanizması uyguladım. Eğer agent belirli bir adım sayısını (örneğin 100 adım) aşarsa veya belirli bir süre içinde (örneğin 5 dakika) bir sonuca ulaşamazsa, görevi iptal edip kullanıcıya hata mesajı dönmesini sağladım. Üçüncüsü, agent’ın “hafızasını” optimize ettim. Gereksiz veya eski bilgileri hafızasından silerek, karar verme mekanizmasını daha odaklı hale getirdim.
import time
class AIAgent:
def __init__(self, max_steps: int = 50, timeout_seconds: int = 300):
self.step_count = 0
self.start_time = time.time()
self.max_steps = max_steps
self.timeout_seconds = timeout_seconds
self.history = [] # Agent'ın hafızası
def execute_step(self, task_description: str) -> str:
self.step_count += 1
elapsed_time = time.time() - self.start_time
if self.step_count > self.max_steps:
print(f"HATA: Agent {self.max_steps} adımı aştı, görevi iptal ediyor.")
return "ERROR: Max steps exceeded."
if elapsed_time > self.timeout_seconds:
print(f"HATA: Agent {self.timeout_seconds} saniye içinde tamamlanamadı, görevi iptal ediyor.")
return "ERROR: Timeout."
# Simülasyon: Agent'ın bir adımda yaptığı iş
print(f"Adım {self.step_count}: '{task_description}' işleniyor. Geçen süre: {elapsed_time:.2f}s")
self.history.append(f"Adım {self.step_count}: {task_description}")
# Gerçek uygulamada burada LLM çağrısı ve araç kullanımı olur
# Örnek olarak basit bir döngü simülasyonu
if self.step_count < 5:
return self.execute_step(f"Alt görev {self.step_count + 1} üzerinde çalış")
else:
return "GÖREV TAMAMLANDI."
# agent = AIAgent(max_steps=10)
# agent.execute_step("Kullanıcıdan gelen raporu özetle ve ekibe gönder.")Bu tarz limitler, agent’ların potansiyelini kullanırken aynı zamanda onları kontrol altında tutmamızı sağlıyor. Bir systemd unit’ine CPUQuota veya MemoryHigh limitleri koymakla aynı mantık; sistem kaynaklarının beklenmedik bir process tarafından tüketilmesini engellemek. Aksi halde, AI agent’lar da bir switch loop’u gibi network’ü boğmasa da, kendi sunucunuzu boğabilir.
Çoklu LLM Provider Yönetiminde Zorluklar ve Çözümler
AI uygulamalarında güvenilirliği artırmak için çoklu LLM (Large Language Model) provider stratejileri kullanıyorum. Kendi yan ürünlerimde, örneğin birincil sağlayıcı olarak Gemini Flash’ı kullanırken, olası kesintilere karşı Groq, Cerebras ve OpenRouter gibi alternatifleri fallback olarak konumlandırdım. Fikir harikaydı: bir sağlayıcı düşerse, diğerine geçiş yapacaktık ve kullanıcılar kesintisiz bir deneyim yaşayacaktı. Ancak bu mimari, beklenmedik bir debug kabusuna dönüştü. Bir hata oluştuğunda, hatanın hangi provider’dan geldiğini, neden fallback’e geçildiğini veya fallback’in neden başarısız olduğunu anlamak, tek bir provider kullanmaktan çok daha zordu. Bu, bir zamanlar şirket çıkışında 3 farklı ISP ile BGP routing decisions yönetirken yaşadığım karmaşaya benziyordu; her şey yolundayken harika, ama bir sorun çıktığında root cause’u bulmak çok zordu.
Bu karmaşayı yönetmek için, her bir provider çağrısını ayrı ayrı loglamaya başladım. Hangi provider’ın çağrıldığı, hangi parametrelerle çağrıldığı, ne kadar sürdüğü ve hangi hatayı döndürdüğü gibi bilgileri kaydettim. Ayrıca, fallback mekanizmasını daha sağlam hale getirdim. Bir provider’dan hata alındığında, sadece bir sonraki provider’a geçmek yerine, hatanın türüne göre farklı stratejiler belirledim. Örneğin, bir rate limit hatasında kısa bir bekleme süresi uygulayıp tekrar denemek, bir authentication error’da ise direkt diğer provider’a geçmek gibi. Bu, distributed tracing sistemlerinde bir isteğin farklı servisler arasında nasıl gezdiğini izlemeye benziyor.
import time
import random
def call_gemini_flash(prompt: str):
# Simülasyon: Gemini Flash çağrısı
if random.random() < 0.1: # %10 hata oranı
raise Exception("Gemini Flash API Error: Rate limit exceeded.")
time.sleep(0.5)
return f"Gemini Flash cevabı: {prompt[:20]}..."
def call_groq(prompt: str):
# Simülasyon: Groq çağrısı
if random.random() < 0.05: # %5 hata oranı
raise Exception("Groq API Error: Internal server error.")
time.sleep(0.3)
return f"Groq cevabı: {prompt[:20]}..."
def call_cerebras(prompt: str):
# Simülasyon: Cerebras çağrısı
if random.random() < 0.02: # %2 hata oranı
raise Exception("Cerebras API Error: Invalid request.")
time.sleep(0.7)
return f"Cerebras cevabı: {prompt[:20]}..."
def multi_provider_llm_call(prompt: str):
providers = [
("Gemini Flash", call_gemini_flash),
("Groq", call_groq),
("Cerebras", call_cerebras),
]
for provider_name, provider_func in providers:
try:
print(f"Deniyor: {provider_name}...")
response = provider_func(prompt)
print(f"Başarılı: {provider_name}")
return response
except Exception as e:
print(f"Hata: {provider_name} - {e}. Fallback yapılıyor...")
# Hata tipine göre farklı aksiyonlar alınabilir.
# Örneğin, rate limit için bekleyip tekrar deneme.
print("Tüm sağlayıcılar başarısız oldu.")
return "HATA: AI servislerine ulaşılamıyor."
# multi_provider_llm_call("Merhaba dünya, bana kısa bir hikaye anlat.")Bu yapı, bir provider’ın kesintiye uğraması durumunda bile sistemin çalışmaya devam etmesini sağlarken, aynı zamanda hangi provider’ın sorun çıkardığını ve nedenlerini daha net bir şekilde anlamama yardımcı oluyor. Bu, network segmentasyonu gibi kritik altyapı tasarımlarında yedekliliğin ve hata toleransının önemini bir kez daha gösteriyor.
AI Sistemlerinde Kapsamlı Gözlem (Observability) Kurulumu
AI projelerinin en sinsi hatalarından biri, operasyonel gözlem (observability) eksikliği yüzünden sorunların geç fark edilmesidir. Geleneksel yazılım sistemlerinde CPU, bellek, disk, ağ trafiği gibi metrikleri izlemek nispeten kolaydır. Ancak AI sistemlerinde, özellikle LLM’lerin iç davranışlarını veya agent’ların karar alma süreçlerini izlemek çok daha zordur. Kendi Android spam uygulamamın backend’inde AI destekli bir filtreleme modülü kullanıyordum. Kullanıcılar bir süre sonra spam mesajlarının hala geldiğini bildirmeye başladı, ancak sistem metriklerinde her şey yolundaydı. Sebebini anlamak 2 gün sürdü: AI modelinin çıktı kalitesi düşmüştü, ama bunu ölçen bir metriğim yoktu. Bu, bir zamanlar bir üretim firmasının ERP’sinde geç sevkiyat raporunun hep eksik gelmesi gibiydi; ana göstergeler iyi görünse de, altta yatan iş akışında sorun vardı.
Bu tür sorunları erkenden tespit etmek için AI sistemlerine özel gözlem stratejileri geliştirdim. İlk olarak, API çağrılarının başarı/başarısızlık oranları, ortalama yanıt süreleri ve token kullanım oranları gibi temel metrikleri toplamaya başladım. İkincisi, modelin çıktısını analiz eden “kalite metrikleri” tanımladım. Örneğin, bir metin özetleme görevinde, özetin orijinal metinle ne kadar alakalı olduğunu veya belirli anahtar kelimeleri içerip içermediğini ölçen basit heuristic’ler kullandım. Üçüncüsü, agent’ların karar ağacındaki her adımı ve aldığı kararları detaylı bir şekilde loglayarak, journald üzerinden izlenebilir hale getirdim.
import logging
import time
# Logger kurulumu
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def log_llm_call(model_name: str, prompt_length: int, response_length: int, duration_ms: int, success: bool, error_message: str = ""):
"""LLM çağrılarını loglar."""
status = "SUCCESS" if success else "FAILURE"
logging.info(f"LLM_CALL - Model: {model_name}, PromptLen: {prompt_length}, RespLen: {response_length}, Duration: {duration_ms}ms, Status: {status}, Error: '{error_message}'")
def log_agent_decision(agent_id: str, step: int, decision: str, context: str):
"""Agent'ın karar adımlarını loglar."""
logging.info(f"AGENT_DECISION - AgentID: {agent_id}, Step: {step}, Decision: '{decision}', Context: '{context[:50]}...'")
# Örnek kullanım
# try:
# start_time = time.time()
# # LLM çağrısı
# response_text = "Bu bir örnek cevap."
# duration = int((time.time() - start_time) * 1000)
# log_llm_call("gpt-3.5-turbo", 150, len(response_text), duration, True)
# except Exception as e:
# duration = int((time.time() - start_time) * 1000)
# log_llm_call("gpt-3.5-turbo", 150, 0, duration, False, str(e))
# log_agent_decision("task_agent_001", 3, "Veri tabanından bilgi çek", "Kullanıcı raporu özetlemek istiyor...")