Batch API ile OpenAI İsteklerini Toplu Gönderme

Yazar:komut.io ekibi
0 Yorum

Bir production ortamında günde on binlerce GPT isteği atmak zorunda kaldığınızda, ilk içgüdünüz muhtemelen paralel thread’ler açmak ya da rate limit hatalarıyla boğuşmak olur. Ben de aynı yolu denedim, aynı hataları aldım. Sonra OpenAI’nin Batch API’sini keşfettiğimde işler köklü biçimde değişti. Bu yazıda o deneyimi aktaracağım.

Batch API Nedir ve Neden Önemlidir?

OpenAI’nin standart API’si senkron çalışır: istek gönderirsin, cevap beklersin. Makul miktarda istek için bu yeterlidir. Ama işin içine veri zenginleştirme, toplu sınıflandırma, büyük ölçekli embedding üretimi girdiğinde senkron model hem pahalı hem de yönetimi zorlaşan bir altyapı gerektirir.

Batch API bu problemi farklı bir açıdan çözer. İsteklerinizi bir JSONL dosyasına paketlersiniz, bu dosyayı OpenAI’ye yüklersiniz ve sistem 24 saat içinde tüm yanıtları üretip size sunur. Karşılığında %50 maliyet indirimi alırsınız. Evet, yarı fiyat.

Bunun arkasındaki mantık şu: OpenAI, batch işleri düşük yük dönemlerinde çalıştırarak kendi altyapısını daha verimli kullanır. Siz de bant genişliği kısıtlamalarından azade, saatlik rate limit baskısı olmadan büyük işler çalıştırırsınız.

Hangi senaryolarda kullanıyorum ben bunu:

  • Ürün kataloglarında on binlerce açıklamayı yeniden yazmak
  • Log dosyalarından anomali tespiti için toplu analiz
  • Müşteri destek ticketlarını kategorilere ayırmak
  • Büyük doküman koleksiyonları için embedding üretmek
  • A/B test için aynı içeriğin farklı varyantlarını üretmek

Temel Gereksinimler ve Kurulum

Önce ortamı hazırlayalım. Python tarafında openai paketinin güncel versiyonu yeterli, ama ben üretim ortamlarında her zaman sanal ortam kullanıyorum:

python3 -m venv batch-env
source batch-env/bin/activate
pip install openai python-dotenv

# Versiyon kontrolü
pip show openai | grep Version

API anahtarınızı environment variable olarak ayarlayın, kod içine gömmeyin:

export OPENAI_API_KEY="sk-proj-..."

# Ya da .env dosyasına yazın
echo "OPENAI_API_KEY=sk-proj-..." > .env
chmod 600 .env

JSONL Dosyası Hazırlamak

Batch API’nin kalbi JSONL formatındaki istek dosyasıdır. Her satır bağımsız bir istek, her istek belirli bir şemaya uygun olmalı.

Şema şöyle:

  • custom_id: Her isteğe verdiğiniz benzersiz tanımlayıcı. Sonuçları eşleştirirken bunu kullanacaksınız.
  • method: HTTP metodu, her zaman POST
  • url: Hangi endpoint’i kullanacağınız, örneğin /v1/chat/completions
  • body: Normal API çağrısında gönderdiğiniz JSON gövdesi

Şimdi gerçek dünyaya yakın bir senaryo yazalım. Diyelim ki e-ticaret sitenizdeki ürün açıklamalarını SEO dostu hale getirmeniz gerekiyor:

import json

products = [
    {"id": "PRD-001", "name": "Deri Cüzdan", "desc": "Siyah renk, 6 kart bölmeli"},
    {"id": "PRD-002", "name": "Ahşap Saat", "desc": "Bambu gövde, minimal tasarım"},
    {"id": "PRD-003", "name": "Tuval Çanta", "desc": "Kanvas malzeme, fermuarlı"},
]

requests = []
for product in products:
    request = {
        "custom_id": f"seo-{product['id']}",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [
                {
                    "role": "system",
                    "content": "Sen bir e-ticaret SEO uzmanısın. Ürün açıklamalarını 100-150 kelimelik, anahtar kelime odaklı metinlere dönüştür."
                },
                {
                    "role": "user",
                    "content": f"Ürün: {product['name']}nMevcut açıklama: {product['desc']}nnSEO uyumlu açıklama yaz."
                }
            ],
            "max_tokens": 300,
            "temperature": 0.7
        }
    }
    requests.append(request)

# JSONL dosyasına yaz
with open("batch_requests.jsonl", "w", encoding="utf-8") as f:
    for req in requests:
        f.write(json.dumps(req, ensure_ascii=False) + "n")

print(f"{len(requests)} istek dosyaya yazıldı.")

Dosyayı Yüklemek ve Batch İşi Başlatmak

Dosya hazır olduğunda iki adım var: önce dosyayı OpenAI’ye yükle, sonra batch işini oluştur.

from openai import OpenAI
from dotenv import load_dotenv
import os

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

# Dosyayı yükle
with open("batch_requests.jsonl", "rb") as f:
    batch_file = client.files.create(
        file=f,
        purpose="batch"
    )

print(f"Dosya yüklendi. File ID: {batch_file.id}")
print(f"Dosya boyutu: {batch_file.bytes} byte")

# Batch işini başlat
batch_job = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
    metadata={
        "project": "seo-enrichment",
        "version": "1.0",
        "created_by": "data-team"
    }
)

print(f"Batch iş başlatıldı.")
print(f"Batch ID: {batch_job.id}")
print(f"Durum: {batch_job.status}")

# Bu ID'yi kaydedin!
with open("batch_job_id.txt", "w") as f:
    f.write(batch_job.id)

metadata alanını ciddiye alın. Üretimde aynı anda birden fazla batch işi çalışıyor olabilir. Hangi işin ne için başlatıldığını metadata’ya yazmazsanız, iki hafta sonra dosya listesine bakıp ne olduğunu anlamaya çalışırsınız.

Batch Durumunu İzlemek

Batch işi asenkron çalıştığından, tamamlanıp tamamlanmadığını periyodik olarak kontrol etmeniz gerekir. Üretimde bunu bir cron job ya da uzun süre çalışan bir daemon script’le yapıyorum:

import time
import sys
from openai import OpenAI

client = OpenAI()

def check_batch_status(batch_id: str) -> dict:
    batch = client.batches.retrieve(batch_id)
    
    status_info = {
        "id": batch.id,
        "status": batch.status,
        "created_at": batch.created_at,
        "request_counts": batch.request_counts,
        "output_file_id": batch.output_file_id,
        "error_file_id": batch.error_file_id,
    }
    
    return status_info, batch

def wait_for_completion(batch_id: str, check_interval: int = 60):
    """
    Batch tamamlanana kadar bekle.
    check_interval: Kaç saniyede bir kontrol edilsin (default: 60)
    """
    terminal_states = {"completed", "failed", "expired", "cancelled"}
    
    while True:
        info, batch = check_batch_status(batch_id)
        
        counts = batch.request_counts
        print(f"[{time.strftime('%H:%M:%S')}] Durum: {info['status']}")
        
        if counts:
            print(f"  Toplam: {counts.total} | "
                  f"Tamamlanan: {counts.completed} | "
                  f"Başarısız: {counts.failed}")
        
        if info["status"] in terminal_states:
            return info, batch
        
        time.sleep(check_interval)

# Kullanım
batch_id = open("batch_job_id.txt").read().strip()
info, batch = wait_for_completion(batch_id, check_interval=30)
print(f"nSonuç durumu: {info['status']}")

status değerleri şunlar olabilir:

  • validating: Yüklenen dosya doğrulanıyor
  • in_progress: İstekler işleniyor
  • finalizing: Sonuçlar hazırlanıyor
  • completed: Her şey tamamlandı
  • failed: Bir sorun oluştu
  • expired: 24 saat içinde tamamlanamadı
  • cancelled: Manuel olarak iptal edildi

Sonuçları İndirmek ve İşlemek

İş tamamlandığında çıktı dosyasını indirip parse etmeniz gerekiyor:

import json
from openai import OpenAI

client = OpenAI()

def download_results(batch_id: str, output_path: str = "batch_results.jsonl"):
    batch = client.batches.retrieve(batch_id)
    
    if batch.status != "completed":
        print(f"Batch henüz tamamlanmadı. Durum: {batch.status}")
        return None
    
    if not batch.output_file_id:
        print("Çıktı dosyası bulunamadı.")
        return None
    
    # Sonuç dosyasını indir
    content = client.files.content(batch.output_file_id)
    
    with open(output_path, "wb") as f:
        f.write(content.content)
    
    print(f"Sonuçlar {output_path} dosyasına kaydedildi.")
    
    # Hata dosyası varsa onu da indir
    if batch.error_file_id:
        error_content = client.files.content(batch.error_file_id)
        with open("batch_errors.jsonl", "wb") as f:
            f.write(error_content.content)
        print("Hatalar batch_errors.jsonl dosyasına kaydedildi.")
    
    return output_path

def parse_results(results_file: str) -> dict:
    """
    Sonuçları custom_id'ye göre indexle.
    """
    results = {}
    
    with open(results_file, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            
            result = json.loads(line)
            custom_id = result["custom_id"]
            
            if result["error"]:
                results[custom_id] = {
                    "success": False,
                    "error": result["error"]
                }
            else:
                # Başarılı yanıtı çıkar
                content = result["response"]["body"]["choices"][0]["message"]["content"]
                results[custom_id] = {
                    "success": True,
                    "content": content,
                    "usage": result["response"]["body"]["usage"]
                }
    
    return results

# Kullanım
batch_id = open("batch_job_id.txt").read().strip()
results_file = download_results(batch_id)

if results_file:
    results = parse_results(results_file)
    
    for custom_id, result in results.items():
        if result["success"]:
            print(f"n--- {custom_id} ---")
            print(result["content"][:200])
        else:
            print(f"n[HATA] {custom_id}: {result['error']}")

Üretim Ortamı için Komple Script

Tüm bunları bir araya getiren, log tutabilen ve hata yönetimi yapabilen bir script:

#!/usr/bin/env python3
"""
Batch API İş Yöneticisi
Kullanım: python batch_manager.py --input data.jsonl --project my-project
"""

import argparse
import json
import logging
import os
import sys
import time
from pathlib import Path
from openai import OpenAI

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("batch_manager.log"),
        logging.StreamHandler(sys.stdout)
    ]
)
log = logging.getLogger(__name__)

client = OpenAI()

def submit_batch(input_file: str, project: str) -> str:
    log.info(f"Dosya yükleniyor: {input_file}")
    
    with open(input_file, "rb") as f:
        uploaded = client.files.create(file=f, purpose="batch")
    
    log.info(f"Dosya yüklendi: {uploaded.id}")
    
    batch = client.batches.create(
        input_file_id=uploaded.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
        metadata={"project": project, "input_file": input_file}
    )
    
    log.info(f"Batch başlatıldı: {batch.id}")
    return batch.id

def monitor_and_retrieve(batch_id: str, output_dir: str = ".") -> bool:
    output_path = Path(output_dir)
    output_path.mkdir(exist_ok=True)
    
    while True:
        batch = client.batches.retrieve(batch_id)
        counts = batch.request_counts
        
        log.info(
            f"Durum: {batch.status} | "
            f"Toplam: {getattr(counts, 'total', '?')} | "
            f"Tamamlanan: {getattr(counts, 'completed', '?')} | "
            f"Başarısız: {getattr(counts, 'failed', '?')}"
        )
        
        if batch.status == "completed":
            result_file = output_path / f"{batch_id}_results.jsonl"
            content = client.files.content(batch.output_file_id)
            result_file.write_bytes(content.content)
            log.info(f"Sonuçlar kaydedildi: {result_file}")
            
            if batch.error_file_id:
                error_file = output_path / f"{batch_id}_errors.jsonl"
                error_content = client.files.content(batch.error_file_id)
                error_file.write_bytes(error_content.content)
                log.warning(f"Hatalar kaydedildi: {error_file}")
            
            return True
        
        elif batch.status in {"failed", "expired", "cancelled"}:
            log.error(f"Batch başarısız: {batch.status}")
            return False
        
        time.sleep(60)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True, help="JSONL input dosyası")
    parser.add_argument("--project", required=True, help="Proje adı")
    parser.add_argument("--output-dir", default="results", help="Sonuç dizini")
    args = parser.parse_args()
    
    batch_id = submit_batch(args.input, args.project)
    success = monitor_and_retrieve(batch_id, args.output_dir)
    sys.exit(0 if success else 1)

Bu script’i şöyle çalıştırırsınız:

python batch_manager.py 
  --input batch_requests.jsonl 
  --project "seo-enrichment-v2" 
  --output-dir /data/results

# Systemd service olarak çalıştırmak için
# /etc/systemd/system/batch-job.service dosyası oluşturun

Mevcut Batch İşlerini Listelemek

Aktif ve geçmiş batch işlerinizi görmek için:

from openai import OpenAI

client = OpenAI()

# Son 10 batch işini listele
batches = client.batches.list(limit=10)

for batch in batches.data:
    counts = batch.request_counts
    print(f"ID: {batch.id}")
    print(f"  Durum: {batch.status}")
    print(f"  Metadata: {batch.metadata}")
    
    if counts:
        print(f"  İstekler: {counts.total} toplam, "
              f"{counts.completed} tamamlandı, "
              f"{counts.failed} başarısız")
    print()

Sınırlamalar ve Dikkat Edilmesi Gerekenler

Batch API her duruma uygun değil, bunu açıkça söylemek gerekiyor.

  • 24 saat penceresi: Batch işiniz 24 saat içinde tamamlanmazsa expired durumuna düşer. Bu olursa işlenemeyen istekler için tekrar denemeniz gerekir.
  • Dosya boyutu sınırı: Tek bir batch dosyası maksimum 100 MB olabilir ve en fazla 50.000 istek içerebilir. Daha büyük işleri parçalara bölün.
  • Gerçek zamanlı değil: Batch API anlık yanıt gerektiren uygulamalar için uygun değildir. Kullanıcıya anlık geri bildirim verecekseniz standart API’yi kullanın.
  • Model desteği: Tüm modeller Batch API’yi desteklemez. GPT-4o, GPT-4o-mini ve embedding modelleri destekleniyor. Kullanmadan önce dokümanı kontrol edin.
  • Sonuç sırası garantisi yok: Batch işi tamamlandığında sonuçlar istek sıralamanızla aynı olmayabilir. custom_id alanını kullanarak eşleştirme yapın.

Büyük işleri yönetirken bir ipucu: batch dosyalarınızı 10.000’er istek gibi mantıklı parçalara bölün. Böylece bir parça başarısız olduğunda tümünü yeniden çalıştırmanıza gerek kalmaz.

Gerçek Dünya: Maliyet Hesabı

Pratik bir örnek verelim. Günde 100.000 gpt-4o-mini isteği attığınızı varsayın, her istek ortalama 500 input token ve 200 output token kullansın.

Standart API ile:

  • Input: 100.000 x 500 token = 50M token
  • Output: 100.000 x 200 token = 20M token
  • Toplam maliyet: Güncel fiyatlarla hesaplayın, Batch API %50 daha ucuz

Aylık bazda bu fark ciddi bir bütçe kalemi oluşturuyor. Batch API’nin “24 saat bekleyebiliyorum” dediğiniz her iş için standart tercihiniz olması gerektiğini düşünüyorum.

Sonuç

Batch API, büyük ölçekli LLM işlemleri için hem maliyet hem de yönetim açısından ciddi avantajlar sunuyor. Özellikle veri zenginleştirme, toplu analiz ve içerik üretimi gibi gerçek zamanlı olmayan iş yüklerinde standart API yerine Batch API kullanmak, infrastructure maliyetlerinizi yarı yarıya düşürebilir.

Uygulamada dikkat etmeniz gereken iki kritik nokta var: custom_id alanını anlamlı değerlerle doldurun çünkü sonuçları geri eşleştirmeniz bu ID’ye bağlı, ve metadata’yı ihmal etmeyin çünkü üretimde işlerin takibi metadata olmadan kaosa döner.

Kendi deneyimimden söylüyorum: Batch API’yi kullanmaya başladıktan sonra rate limit hataları tamamen hayatımdan çıktı, maliyet %40-50 arasında düştü ve büyük veri işleme süreçlerim çok daha öngörülebilir hale geldi. Eğer hala her isteği senkron atıyorsanız ve büyük hacimli işler yapıyorsanız, Batch API’ye geçmek için beklemeniz gereken hiçbir neden yok.

Bir yanıt yazın

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