Metin Vektörüne Dönüştürme: OpenAI Embeddings API Kullanımı

Yazar:komut.io ekibi
0 Yorum

Metin verilerini sayısal temsillere dönüştürmek, modern yapay zeka uygulamalarının temel taşlarından biri haline geldi. Özellikle semantik arama, doküman sınıflandırma veya öneri sistemleri üzerinde çalışıyorsan, embedding kavramını mutlaka duymuşsundur. OpenAI’nin Embeddings API’si bu işi hem hızlı hem de oldukça güvenilir bir şekilde yapıyor. Bugün bu API’yi gerçek dünya senaryolarıyla birlikte ele alacağız; sysadmin gözüyle bakarak log analizi, doküman arama ve otomatik ticket sınıflandırma gibi pratik kullanım alanlarına odaklanacağız.

Embedding Nedir, Neden İşimize Yarar?

Embedding, bir metin parçasını yüksek boyutlu bir sayı dizisine (vektöre) dönüştürme işlemidir. Bu vektörler, metinlerin anlamsal olarak birbirine ne kadar yakın olduğunu matematiksel olarak ifade eder. Yani “sunucu çöktü” ile “sistem yanıt vermiyor” cümleleri, kelime olarak farklı olsa da anlamsal olarak yakın olduklarından embedding uzayında birbirine yakın noktalarda konumlanır.

Bunu neden bir sysadmin olarak önemsememiz gerekiyor? Çünkü:

  • Log analizi: Binlerce log satırı arasında benzer hata kalıplarını bulmak için
  • Otomatik ticket yönlendirme: Helpdesk ticketlarını doğru ekibe yönlendirmek için
  • Doküman arama: İç wiki veya runbook aramalarını anlamsal hale getirmek için
  • Anomali tespiti: Normal davranıştan sapan log örüntülerini yakalamak için

OpenAI’nin text-embedding-3-small ve text-embedding-3-large modelleri şu an için en iyi performans/maliyet dengesini sunuyor.

API Anahtarını Ayarlamak ve Ortamı Hazırlamak

Önce çalışma ortamını hazırlayalım. Python kullanacağız çünkü OpenAI’nin resmi kütüphanesi Python için oldukça olgun.

# Sanal ortam oluştur
python3 -m venv embedding-env
source embedding-env/bin/activate

# Gerekli kütüphaneleri yükle
pip install openai numpy scikit-learn python-dotenv

# API anahtarını güvenli şekilde sakla
cat > .env << 'EOF'
OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
EOF

chmod 600 .env

API anahtarını asla kod içine gömmemek gerekiyor. .env dosyasını .gitignore‘a eklemeyi unutma:

echo ".env" >> .gitignore
echo "embedding-env/" >> .gitignore

İlk Embedding İsteği

Temel bir embedding isteği nasıl yapılır, görelim:

cat > basic_embedding.py << 'EOF'
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    """Verilen metin için embedding vektörü döndürür."""
    # Metni temizle - yeni satırlar embedding kalitesini düşürür
    text = text.replace("n", " ").strip()
    
    response = client.embeddings.create(
        input=text,
        model=model
    )
    
    return response.data[0].embedding

# Test edelim
test_text = "Nginx sunucusu 502 Bad Gateway hatası veriyor"
embedding = get_embedding(test_text)

print(f"Model: text-embedding-3-small")
print(f"Metin: {test_text}")
print(f"Vektör boyutu: {len(embedding)}")
print(f"İlk 5 değer: {embedding[:5]}")
EOF

python3 basic_embedding.py

text-embedding-3-small modeli 1536 boyutlu vektörler üretiyor. text-embedding-3-large ise 3072 boyutlu. Çoğu kullanım senaryosu için small model yeterli ve yaklaşık 5 kat daha ucuz.

Toplu Embedding İşlemi

Gerçek dünyada tek tek metin göndermek yerine toplu işlem yapmak hem daha verimli hem de daha ekonomik. Diyelim ki yüzlerce log satırını işlememiz gerekiyor:

cat > batch_embedding.py << 'EOF'
import os
import time
from openai import OpenAI
from dotenv import load_dotenv
from typing import List

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def get_embeddings_batch(
    texts: List[str], 
    model: str = "text-embedding-3-small",
    batch_size: int = 100
) -> List[List[float]]:
    """
    Büyük metin listelerini toplu olarak işler.
    API limiti: tek istekte max 2048 metin
    """
    all_embeddings = []
    
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        
        # Metinleri temizle
        cleaned_batch = [text.replace("n", " ").strip() for text in batch]
        
        try:
            response = client.embeddings.create(
                input=cleaned_batch,
                model=model
            )
            
            # Sıralamayı koru
            batch_embeddings = [item.embedding for item in sorted(
                response.data, key=lambda x: x.index
            )]
            all_embeddings.extend(batch_embeddings)
            
            print(f"İşlendi: {min(i + batch_size, len(texts))}/{len(texts)}")
            
            # Rate limiting için küçük bekleme
            if i + batch_size < len(texts):
                time.sleep(0.5)
                
        except Exception as e:
            print(f"Hata oluştu batch {i}: {e}")
            raise
    
    return all_embeddings

# Örnek log satırları
sample_logs = [
    "CRITICAL: Database connection pool exhausted, max 100 connections reached",
    "ERROR: Nginx upstream timeout after 30s for /api/v2/users",
    "WARNING: Disk usage on /var/log reached 85%",
    "INFO: Scheduled backup completed successfully",
    "ERROR: SSL certificate expires in 7 days for example.com",
    "CRITICAL: Memory usage at 95%, OOM killer activated",
    "WARNING: Failed login attempt from 192.168.1.105",
]

embeddings = get_embeddings_batch(sample_logs)
print(f"nToplam {len(embeddings)} embedding üretildi")
print(f"Her vektör boyutu: {len(embeddings[0])}")
EOF

python3 batch_embedding.py

Semantik Benzerlik Hesaplama

Embedding’lerin asıl gücü benzerlik hesaplamaktan geliyor. Cosine similarity kullanarak iki metnin ne kadar benzer olduğunu bulabiliriz:

cat > similarity_search.py << 'EOF'
import os
import numpy as np
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def cosine_similarity(vec1: list, vec2: list) -> float:
    """İki vektör arasındaki cosine similarity hesaplar. -1 ile 1 arası değer döner."""
    a = np.array(vec1)
    b = np.array(vec2)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

def semantic_search(query: str, documents: list, top_k: int = 3) -> list:
    """Sorgu ile en benzer dokümanları bulur."""
    
    # Tüm metinleri tek seferde embed et (daha verimli)
    all_texts = [query] + documents
    response = client.embeddings.create(
        input=[t.replace("n", " ") for t in all_texts],
        model="text-embedding-3-small"
    )
    
    embeddings = [item.embedding for item in sorted(response.data, key=lambda x: x.index)]
    query_embedding = embeddings[0]
    doc_embeddings = embeddings[1:]
    
    # Benzerlik skorlarını hesapla
    scores = [
        (doc, cosine_similarity(query_embedding, doc_emb))
        for doc, doc_emb in zip(documents, doc_embeddings)
    ]
    
    # Skora göre sırala
    scores.sort(key=lambda x: x[1], reverse=True)
    
    return scores[:top_k]

# Runbook veritabanı örneği
runbooks = [
    "Nginx 502 hatası için upstream timeout değerini artır ve backend sağlığını kontrol et",
    "MySQL bağlantı havuzu dolduğunda max_connections parametresini artır",
    "Disk dolduğunda /var/log altındaki eski logları temizle ve logrotate ayarla",
    "SSL sertifikası yenileme için certbot renew komutunu çalıştır",
    "Yüksek CPU kullanımında top ve ps aux ile prosesleri analiz et",
    "OOM durumunda kernel parametrelerini ve swap kullanımını kontrol et",
    "Başarısız SSH girişleri için fail2ban kurallarını gözden geçir",
    "Redis bağlantı hatalarında maxmemory-policy ayarını kontrol et",
]

# Kullanıcı sorgusu
query = "Web sunucusu yavaş yanıt veriyor ve zaman aşımı hatası alıyorum"
results = semantic_search(query, runbooks, top_k=3)

print(f"Sorgu: {query}n")
print("En benzer runbook'lar:")
for doc, score in results:
    print(f"  Skor: {score:.4f} | {doc[:80]}...")
EOF

python3 similarity_search.py

Gerçek Dünya Senaryosu: Log Analizi ve Anomali Tespiti

Şimdi daha gerçekçi bir senaryoya bakalım. Üretim ortamındaki log dosyalarını analiz edip benzer hata örüntülerini gruplayalım:

cat > log_analyzer.py << 'EOF'
import os
import json
import numpy as np
from openai import OpenAI
from sklearn.cluster import KMeans
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# Simüle edilmiş üretim log verileri
production_logs = [
    "2024-01-15 10:23:45 ERROR app[web.1]: Connection refused to PostgreSQL at db-prod-01:5432",
    "2024-01-15 10:23:47 ERROR app[web.2]: Database connection failed: timeout after 5000ms",
    "2024-01-15 10:24:01 CRITICAL app[web.1]: Cannot connect to database, retrying...",
    "2024-01-15 10:31:22 ERROR app[worker.1]: Redis NOAUTH Authentication required",
    "2024-01-15 10:31:25 ERROR app[worker.2]: Redis connection error: AUTH failed",
    "2024-01-15 10:45:33 WARNING app[web.3]: Response time exceeded 2000ms for GET /api/orders",
    "2024-01-15 10:45:41 WARNING app[web.1]: Slow query detected: 3200ms SELECT * FROM orders",
    "2024-01-15 11:02:15 ERROR app[web.2]: 503 Service Unavailable from upstream payment-service",
    "2024-01-15 11:02:18 ERROR app[web.3]: Payment gateway timeout after 10s",
    "2024-01-15 11:15:00 INFO app[scheduler.1]: Cleanup job completed, removed 1423 temp files",
    "2024-01-15 11:45:22 ERROR app[web.1]: SSL handshake failed for api.external-provider.com",
    "2024-01-15 11:45:30 ERROR app[web.2]: Certificate verification failed: certificate expired",
]

def embed_logs(logs: list) -> np.ndarray:
    """Log satırlarını embed et ve numpy array döndür."""
    response = client.embeddings.create(
        input=[log.replace("n", " ") for log in logs],
        model="text-embedding-3-small"
    )
    embeddings = [item.embedding for item in sorted(response.data, key=lambda x: x.index)]
    return np.array(embeddings)

def cluster_logs(logs: list, n_clusters: int = 4) -> dict:
    """Logları semantik benzerliğe göre grupla."""
    embeddings = embed_logs(logs)
    
    kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
    cluster_labels = kmeans.fit_predict(embeddings)
    
    clusters = {}
    for idx, label in enumerate(cluster_labels):
        if label not in clusters:
            clusters[label] = []
        clusters[label].append(logs[idx])
    
    return clusters

# Logları grupla
clusters = cluster_logs(production_logs)

print("=== LOG CLUSTER ANALİZİ ===n")
for cluster_id, logs in sorted(clusters.items()):
    print(f"Cluster {cluster_id + 1} ({len(logs)} log):")
    for log in logs:
        # Timestamp kısmını at, sadece mesajı göster
        message = log[20:] if len(log) > 20 else log
        print(f"  - {message[:100]}")
    print()
EOF

python3 log_analyzer.py

Doküman Arama Sistemi Oluşturma ve Kaydetme

Gerçek uygulamalarda embedding’leri her seferinde yeniden hesaplamak hem yavaş hem pahalı. Bunları disk üzerinde cache’leyelim:

cat > embedding_cache.py << 'EOF'
import os
import json
import hashlib
import numpy as np
from pathlib import Path
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

CACHE_DIR = Path("./embedding_cache")
CACHE_DIR.mkdir(exist_ok=True)

def get_text_hash(text: str) -> str:
    """Metin için deterministik hash üret."""
    return hashlib.sha256(text.encode()).hexdigest()[:16]

def load_cached_embedding(text: str, model: str) -> list | None:
    """Cache'den embedding yükle, yoksa None döndür."""
    cache_key = f"{model}_{get_text_hash(text)}"
    cache_file = CACHE_DIR / f"{cache_key}.json"
    
    if cache_file.exists():
        with open(cache_file, "r") as f:
            return json.load(f)["embedding"]
    return None

def save_embedding_cache(text: str, model: str, embedding: list):
    """Embedding'i cache'e kaydet."""
    cache_key = f"{model}_{get_text_hash(text)}"
    cache_file = CACHE_DIR / f"{cache_key}.json"
    
    with open(cache_file, "w") as f:
        json.dump({
            "text_preview": text[:100],
            "model": model,
            "embedding": embedding
        }, f)

def get_embedding_cached(text: str, model: str = "text-embedding-3-small") -> list:
    """Cache kontrolü yaparak embedding al."""
    cached = load_cached_embedding(text, model)
    
    if cached is not None:
        print(f"  Cache hit: {text[:50]}...")
        return cached
    
    print(f"  API çağrısı: {text[:50]}...")
    response = client.embeddings.create(
        input=text.replace("n", " "),
        model=model
    )
    embedding = response.data[0].embedding
    save_embedding_cache(text, model, embedding)
    return embedding

# Test: İlk çalıştırmada API, ikincisinde cache kullanacak
docs = [
    "Kubernetes pod'u CrashLoopBackOff durumunda kubectl logs ile kontrol et",
    "Docker container disk alanı için docker system prune komutunu kullan",
    "Ansible playbook hata ayıklaması için -vvv verbose flag ekle",
]

print("İlk çalıştırma (API çağrısı bekleniyor):")
embeddings_1 = [get_embedding_cached(doc) for doc in docs]

print("nİkinci çalıştırma (cache kullanılmalı):")
embeddings_2 = [get_embedding_cached(doc) for doc in docs]

print(f"nCache dosyaları: {list(CACHE_DIR.glob('*.json'))}")
EOF

python3 embedding_cache.py

Boyut Azaltma ile Maliyet Optimizasyonu

text-embedding-3-small ve text-embedding-3-large modelleri dimensions parametresini destekliyor. Bu sayede vektör boyutunu küçülterek storage ve hesaplama maliyetini düşürebiliriz:

cat > dimension_reduction.py << 'EOF'
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def get_embedding_with_dimensions(
    text: str, 
    dimensions: int,
    model: str = "text-embedding-3-small"
) -> list:
    """Belirli boyutta embedding üret."""
    response = client.embeddings.create(
        input=text.replace("n", " "),
        model=model,
        dimensions=dimensions
    )
    return response.data[0].embedding

import numpy as np

def cosine_similarity(v1, v2):
    a, b = np.array(v1), np.array(v2)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

query = "Database sunucusu bağlanamıyor"
doc1 = "PostgreSQL bağlantı hatası giderildi, max_connections artırıldı"
doc2 = "Nginx SSL sertifikası güncellendi"

# Farklı boyutlarda karşılaştır
for dims in [256, 512, 1024, 1536]:
    q_emb = get_embedding_with_dimensions(query, dims)
    d1_emb = get_embedding_with_dimensions(doc1, dims)
    d2_emb = get_embedding_with_dimensions(doc2, dims)
    
    sim1 = cosine_similarity(q_emb, d1_emb)
    sim2 = cosine_similarity(q_emb, d2_emb)
    
    print(f"Dims: {dims:4d} | DB hatası benzerliği: {sim1:.4f} | Nginx benzerliği: {sim2:.4f}")

# Sonuç: 256 boyut bile çoğu senaryo için yeterli ayrışmayı sağlar
EOF

python3 dimension_reduction.py

Helpdesk Ticket Sınıflandırma Sistemi

Şimdi her şeyi bir araya getirelim. Gelen IT ticketlarını otomatik olarak kategorize eden bir sistem kuralım:

cat > ticket_classifier.py << 'EOF'
import os
import numpy as np
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# Kategori örnekleri (few-shot için referans noktalar)
categories = {
    "network": [
        "VPN bağlantısı kurulamıyor",
        "Ağ geçidi yanıt vermiyor, internet erişimi yok",
        "Firewall kuralı eklenmesi gerekiyor",
    ],
    "server": [
        "Sunucu yanıt vermiyor, SSH bağlantısı yok",
        "CPU kullanımı %100'de, sistem yavaş",
        "Disk alanı doldu, servisler çöktü",
    ],
    "database": [
        "Veritabanı sorguları çok yavaş çalışıyor",
        "MySQL replikasyon geride kaldı",
        "Backup görevleri başarısız oluyor",
    ],
    "security": [
        "Şüpheli giriş denemeleri tespit edildi",
        "Kullanıcı hesabı kilitlendi, açılması gerekiyor",
        "SSL sertifikası süresi dolmuş",
    ],
}

def embed_all_categories(cats: dict) -> dict:
    """Tüm kategori örneklerini embed et."""
    all_texts = []
    category_map = []
    
    for cat_name, examples in cats.items():
        for example in examples:
            all_texts.append(example)
            category_map.append(cat_name)
    
    response = client.embeddings.create(
        input=all_texts,
        model="text-embedding-3-small"
    )
    
    embeddings = [item.embedding for item in sorted(response.data, key=lambda x: x.index)]
    
    result = {}
    for text, cat, emb in zip(all_texts, category_map, embeddings):
        if cat not in result:
            result[cat] = []
        result[cat].append(emb)
    
    return result

def classify_ticket(ticket_text: str, category_embeddings: dict) -> tuple:
    """Ticket metnini en uygun kategoriye ata."""
    response = client.embeddings.create(
        input=ticket_text.replace("n", " "),
        model="text-embedding-3-small"
    )
    ticket_emb = np.array(response.data[0].embedding)
    
    best_category = None
    best_score = -1
    
    for cat_name, cat_embeddings in category_embeddings.items():
        # Kategori için ortalama benzerlik al
        scores = []
        for cat_emb in cat_embeddings:
            cat_vec = np.array(cat_emb)
            score = float(np.dot(ticket_emb, cat_vec) / 
                         (np.linalg.norm(ticket_emb) * np.linalg.norm(cat_vec)))
            scores.append(score)
        
        avg_score = np.mean(scores)
        if avg_score > best_score:
            best_score = avg_score
            best_category = cat_name
    
    return best_category, best_score

# Kategori embedding'lerini hazırla
print("Kategori veritabanı hazırlanıyor...")
cat_embeddings = embed_all_categories(categories)

# Test ticketları
test_tickets = [
    "Sabahtan beri uzak masaüstüne bağlanamıyorum, port 3389 timeout alıyor",
    "Prod veritabanında deadlock oluşuyor, transaction'lar takılı kalıyor",
    "Bir kullanıcı brute force ile sisteme girmeye çalışıyor gibi",
    "Web sunucumuzun RAM kullanımı gece 3'te aniden artmış",
]

print("n=== OTOMATİK TICKET SINIFLANDIRMA ===n")
for ticket in test_tickets:
    category, confidence = classify_ticket(ticket, cat_embeddings)
    print(f"Ticket : {ticket}")
    print(f"Kategori: {category.upper()} (güven: {confidence:.4f})")
    print()
EOF

python3 ticket_classifier.py

Maliyet ve Rate Limit Yönetimi

OpenAI Embeddings API’si kullanırken dikkat edilmesi gereken bazı pratik noktalar var:

  • text-embedding-3-small: 1 milyon token başına yaklaşık 0.02 USD. Çoğu kullanım için idealdir.
  • text-embedding-3-large: 1 milyon token başına yaklaşık 0.13 USD. Yüksek hassasiyet gereken durumlarda kullanılır.
  • Rate limits: Varsayılan olarak dakikada 1 milyon token. Production’da bu limite takılabilirsin.
  • Token sayımı: Ortalama olarak 1 token yaklaşık 4 karakter veya 0.75 kelimedir.
  • Batch boyutu: Tek istekte maksimum 2048 metin, ancak toplam token başına 8191 token limiti var.

Rate limit yönetimi için exponential backoff implementasyonu:

cat > rate_limit_handler.py << 'EOF'
import os
import time
import random
from openai import OpenAI, RateLimitError, APIError
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def get_embedding_with_retry(
    text: str,
    model: str = "text-embedding-3-small",
    max_retries: int = 5
) -> list:
    """Rate limit ve geçici hatalara karşı retry mekanizmalı embedding."""
    
    for attempt in range(max_retries):
        try:
            response = client.embeddings.create(
                input=text.replace("n", " "),
                model=model
            )
            return response.data[0].embedding
            
        except RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            
            # Exponential backoff + jitter
            wait_time = (2 ** attempt) + random.uniform(0, 1)
            print(f"Rate limit aşıldı. {wait_time:.1f}s bekleniyor... (deneme {attempt + 1})")
            time.sleep(wait_time)
            
        except APIError as e:
            if attempt == max_retries - 1:
                raise
            
            wait_time = 2 ** attempt
            print(f"API hatası: {e}. {wait_time}s bekleniyor...")
            time.sleep(wait_time)
    
    raise Exception("Maximum retry sayısına ulaşıldı")

# Test
test = "Kubernetes deployment rolling update sırasında pod'lar crash oluyor"
emb = get_embedding_with_retry(test)
print(f"Başarıyla embed edildi: {len(emb)} boyutlu vektör")
EOF

python3 rate_limit_handler.py

Sonuç

OpenAI Embeddings API’si, sysadmin araç setine ciddi bir güç katıyor. Log analizi, ticket sınıflandırma ve doküman arama gibi senaryolarda kelime bazlı aramadan anlamsal aramaya geçmek, operasyonel verimliliği gözle görülür şekilde artırıyor. Özellikle binlerce log satırı arasında “bu hataya benzer bir şeyi daha önce gördük mü?” sorusunu saniyeler içinde yanıtlayabilmek, incident response süreçlerini hızlandırıyor.

Pratik öneriler olarak şunları söyleyebilirim: Başlangıç için her zaman text-embedding-3-small kullan ve yeterli olup olmadığını test et. Embedding’leri mutlaka cache’le, aynı metni birden fazla kez işlemek hem yavaş hem pahalı. Production’a geçmeden önce rate limit ve retry mekanizmalarını doğru implement et. Büyük veri setleri için embedding’leri önceden hesaplayıp vector database’e (Pinecone, Weaviate veya pgvector) kaydet; böylece query zamanında sadece sorgu embedding’ini hesaplamak yeterli olur.

Embedding teknolojisi, LLM dünyasının görünmez kahramanı. Chatbot’lar ve kod üretme araçları kadar konuşulmasa da bir RAG sistemi kurmadan, bir arama motoru geliştirmeden veya anomali tespiti yapmadan herhangi ciddi bir yapay zeka entegrasyonu inşa etmek neredeyse imkansız. Bu temeli sağlam atmak, ilerleyen adımlarda kuracağın her şeyi daha güvenilir ve ölçeklenebilir hale getiriyor.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir