İşlemci Hattınızı Hata Ayıklama

İşleyici (Processor) hatları karmaşık olabilir, özellikle birden çok dönüştürme adımını zincirleme kullanırken. Basit fonksiyon çağrılarından farklı olarak, hatlarda doğal gözlemlenebilirlik yoktur; adımlar arasında ne olduğunu veya nerede bir sorun oluştuğunu kolayca göremezsiniz. Bu kılavuz, bu zorlukları ele almak ve hatlarınızdaki veri akışını anlamanıza yardımcı olmak için özel olarak tasarlanmış hata ayıklama araçlarını ve tekniklerini sunar.

Üç tamamlayıcı hata ayıklama yaklaşımını inceleyeceğiz: çalışma zamanı izleme için kancalar, ayrıntılı inceleme için adım adım hata ayıklama ve yapısal uyumsuzlukları yakalamak için özellik doğrulama. Her biri farklı bir amaca hizmet eder ve birlikte boru hattınızın davranışına ilişkin eksiksiz bir görünürlük sağlar.

Hook'ları Anlama

Hooks, pipeline çalıştırması sırasında belirli noktalarda çağrılan işlevlerdir. Pipeline kodunuzu değiştirmeden verileri denetlemenin, izlemenin veya değiştirmenin bir yolunu sunarlar. Bunları pipeline'ınız için “olay dinleyicileri” olarak düşünebilirsiniz.

Nedir bir Hook?

A hook, pipeline çalıştırması sırasında belirli anlarda otomatik olarak çağrılan bir geri çağırma (callback) işlevidir. Bu kavram olay tabanlı programlamadan gelir; pipeline'ın çalıştırma akışına “bağlanarak” (hook into) neler olduğunu gözlemleyebilir veya buna tepki verebilirsiniz.

Hook'ları, hattınıza (pipeline) kontrol noktaları eklemek gibi düşünün. Hat bu kontrol noktalarından birine her ulaştığında, kısa bir süre duraklayıp hook fonksiyonunuzu çağırır; bu da size mevcut durumu denetleme, bilgi kaydetme ve verileri doğrulama fırsatı verir.

Bir hook, yalnızca iki parametre kabul eden bir işlevdir:

• step_idx: int - İşleme adımının mevcut indeksi (0, 1, 2, vb.)
• transition: EnvTransition - Hattın o noktasındaki veri geçişi

Hook'ların güzelliği, müdahaleci olmayan doğalarından kaynaklanır: pipeline kodunuzda tek bir satır değiştirmeden izleme, doğrulama veya hata ayıklama mantığı ekleyebilirsiniz. Pipeline, çekirdek mantığına odaklanarak temiz kalırken; loglama, izleme ve hata ayıklama gibi enine kesen konuları hook'lar ele alır.

Önce ve Sonra Kancaları

Pipeline iki tür hook destekler:

• Önce kanca (register_before_step_hook) - Her adım çalıştırılmadan önce çağrılır
• Hook'lerden sonra (register_after_step_hook) - Her adım tamamlandıktan sonra çağrılır

def before_hook(step_idx: int, transition: EnvTransition):
    """Called before step processes the transition."""
    print(f"About to execute step {step_idx}")
    # Useful for: logging, validation, setup

def after_hook(step_idx: int, transition: EnvTransition):
    """Called after step has processed the transition."""
    print(f"Completed step {step_idx}")
    # Useful for: monitoring results, cleanup, debugging

processor.register_before_step_hook(before_hook)
processor.register_after_step_hook(after_hook)

NaN Algılama Kancası Uygulama

NaN değerlerini algılayan bir kancanın pratik bir örneği:

def check_nans(step_idx: int, transition: EnvTransition):
    """Check for NaN values in observations."""
    obs = transition.get(TransitionKey.OBSERVATION)
    if obs:
        for key, value in obs.items():
            if isinstance(value, torch.Tensor) and torch.isnan(value).any():
                print(f"NaN detected in {key} at step {step_idx}")

# Register the hook to run after each step
processor.register_after_step_hook(check_nans)

# Process your data - the hook will be called automatically
output = processor(input_data)

# Remove the hook when done debugging
processor.unregister_after_step_hook(check_nans)

Hook'ların İç Mekanizması

Dahili mekanizmayı anlamak, hook'ları daha etkili kullanmanıza yardımcı olur. Pipeline, iki ayrı liste sürdürür: biri before-step hook'ları için, diğeri after-step hook'ları için. Bir hook kaydettiğinizde, ilgili listeye basitçe eklenir.

Çalışma sırasında, pipeline katı bir sırayı izler: her işleme adımı için önce kayıt sırasına göre tüm before-hook'ları çağırır, ardından gerçek adım dönüşümünü çalıştırır ve son olarak kayıt sırasına göre tüm after-hook'ları çağırır. Bu, her adımın etrafında öngörülebilir, sandviç benzeri bir yapı oluşturur.

Temel içgörü, hook'ların çekirdek pipeline mantığını değiştirmemesidir; bunlar tamamen ekleyicidir. Pipeline'ın _forward metodu, hata ayıklama veya izleme kodunuzun ana veri akışını engellemeden tam olarak doğru anlarda çalışmasını sağlamak için hook'lar ve işleme adımları arasındaki bu dansı koordine eder.

İşte pipeline'ın hook'ları nasıl çalıştırdığının basitleştirilmiş bir görünümü:

class DataProcessorPipeline:
    def __init__(self):
        self.steps = [...]
        self.before_step_hooks = []  # List of before hooks
        self.after_step_hooks = []   # List of after hooks

    def _forward(self, transition):
        """Internal method that processes the transition through all steps."""
        for step_idx, processor_step in enumerate(self.steps):
            # 1. Call all BEFORE hooks
            for hook in self.before_step_hooks:
                hook(step_idx, transition)

            # 2. Execute the actual processing step
            transition = processor_step(transition)

            # 3. Call all AFTER hooks
            for hook in self.after_step_hooks:
                hook(step_idx, transition)

        return transition

    def register_before_step_hook(self, hook_fn):
        self.before_step_hooks.append(hook_fn)

    def register_after_step_hook(self, hook_fn):
        self.after_step_hooks.append(hook_fn)

Çalışma Akışı

Çalışma akışı şu şekilde görünür:

Input → Before Hook → Step 0 → After Hook → Before Hook → Step 1 → After Hook → ... → Output

Örneğin, 3 adım ve her iki kancası türü ile:

def timing_before(step_idx, transition):
    print(f"⏱️  Starting step {step_idx}")

def validation_after(step_idx, transition):
    print(f"✅ Completed step {step_idx}")

processor.register_before_step_hook(timing_before)
processor.register_after_step_hook(validation_after)

# This will output:
# ⏱️  Starting step 0
# ✅ Completed step 0
# ⏱️  Starting step 1
# ✅ Completed step 1
# ⏱️  Starting step 2
# ✅ Completed step 2

Çoklu Kancalar

Aynı türden birden çok hook kaydedebilirsiniz - kaydedildikleri sırada çalışırlar:

def log_shapes(step_idx: int, transition: EnvTransition):
    obs = transition.get(TransitionKey.OBSERVATION)
    if obs:
        print(f"Step {step_idx} observation shapes:")
        for key, value in obs.items():
            if isinstance(value, torch.Tensor):
                print(f"  {key}: {value.shape}")

processor.register_after_step_hook(check_nans)      # Executes first
processor.register_after_step_hook(log_shapes)     # Executes second

# Both hooks will be called after each step in registration order
output = processor(input_data)

Hook'lar, belirli sorunları izlemek (NaN algılama gibi) veya normal pipeline çalışması sırasında metrikleri toplamak için mükemmel olsa da, bazen daha derine inmeniz gerekir. Her adımda tam olarak ne olduğunu anlamak veya karmaşık dönüştürme mantığını hata ayıklamak istediğinizde, adım adım hata ayıklama (step-through debugging) size gereken ayrıntılı denetimi sağlar.

Adım Adım Hata Ayıklama

Adım adım hata ayıklama, hattınız için yavaş çekim bir tekrar izleme gibidir. Verilerinizin girdiden çıktıya tek bir hızlı belirsizlik içinde dönüştürülmesini izlemek yerine, her bir bireysel adımın ardından ne olduğunu durdurup inceleyebilirsiniz.

Bu yaklaşım, karmaşık bir hattı anlamaya çalışırken, beklenmeyen davranışları hata ayıklarken veya her dönüştürmenin beklenildiği gibi çalıştığını doğrularken özellikle değerlidir. Otomatik izleme için harika olan kanca (hook) mekanizmalarının aksine, adım adım hata ayıklama, inceleme süreci üzerinde manuel ve etkileşimli kontrol sağlar.

step_through() yöntemi, her işleme adımından sonra geçiş durumunu üreten bir üreteçtir ve bu sayede ara sonuçları incelemenizi sağlar. Bunu, verileriniz boru hattından akarken oluşturulan bir dizi anlık görüntü olarak düşünebilirsiniz; her anlık görüntü, bir sonraki dönüştürme uygulandıktan sonra verilerinizin tam olarak nasıl göründüğünü gösterir.

Step-Through Nasıl Çalışır

step_through() yöntemi, hattın (pipeline) çalışma biçimini temelden değiştirir. Tüm adımları sırayla çalıştırıp yalnızca nihai sonucu döndürmek yerine, hattı ara sonuçları üreten bir yineleyiciye (iterator) dönüştürür.

İçeride neler oluyor: yöntem, girdi verinizi hattın dahili geçiş biçimine dönüştürerek başlar ve ardından bu başlangıç durumunu verir. Ardından ilk işleme adımını uygular ve sonucu verir. Sonra ikinci adımı bu sonuca uygular ve yine verir, böylece devam eder. Her yield, o andaki geçişin tam bir anlık görüntüsünü size sunar.

Bu üreteç deseni tembel (lazy) olduğu için güçlüdür—pipeline yalnızca siz istediğinizde bir sonraki adımı hesaplar. Bu, herhangi bir noktada durmanıza, mevcut durumu ayrıntılı olarak incelemenize ve devam edip etmeyeceğinize karar vermenize olanak tanır. Sorunlu tek bir adımı hata ayıklamak için tüm pipeline'ı çalıştırmak zorunda değilsiniz.

Tüm hattı çalıştırıp yalnızca nihai sonucu görmek yerine, step_through() her adımın ardından durur ve size ara geçişi verir:

# This creates a generator that yields intermediate states.
# The first yield is the input itself; yield i (i >= 1) is the output of step i - 1.
for i, intermediate_result in enumerate(processor.step_through(input_data)):
    print("=== Input ===" if i == 0 else f"=== After step {i - 1} ===")

    # Inspect the observation at this stage
    obs = intermediate_result.get(TransitionKey.OBSERVATION)
    if obs:
        for key, value in obs.items():
            if isinstance(value, torch.Tensor):
                print(f"{key}: shape={value.shape}, dtype={value.dtype}")

Kesme Noktaları ile Etkileşimli Hata Ayıklama

Adım adım döngüde kesme noktaları ekleyerek etkileşimli olarak hata ayıklama yapabilirsiniz:

# Step through the pipeline with debugging
for i, intermediate in enumerate(processor.step_through(data)):
    print("Input" if i == 0 else f"After step {i - 1}: {processor.steps[i - 1].__class__.__name__}")

    # Set a breakpoint to inspect the current state
    breakpoint()  # Debugger will pause here

    # You can now inspect 'intermediate' in the debugger:
    # - Check tensor shapes and values
    # - Verify expected transformations
    # - Look for unexpected changes

Hata ayıklayıcı oturumu sırasında şunları yapabilirsiniz:

• Gözlem verilerini görmek için intermediate[TransitionKey.OBSERVATION] değerini inceleyin
• Eylem dönüşümleri için intermediate[TransitionKey.ACTION] alanını kontrol edin
• Geçişin herhangi bir bölümünü inceleyerek her adımın ne yaptığını anlayın

Adım adım hata ayıklama, veri dönüşümlerini anlamak için idealdir, ancak bu verinin yapısı ne durumda? Kancalar ve adım adım hata ayıklama çalışma zamanı davranışını hata ayıklamanıza yardımcı olsa da, hattınızın aşağı akış bileşenlerinin beklediği biçimde veri ürettiğinden de emin olmanız gerekir. İşte özellik sözleşmesi doğrulaması burada devreye girer.

Özellik Sözleşmelerini Doğrulama

Özellik sözleşmeleri, hattınızın girdi olarak beklediği ve çıktı olarak ürettiği veri yapısını tanımlar. Bu sözleşmeleri doğrulamak, uyumsuzlukları erken aşamada yakalamaya yardımcı olur.

Özellik Sözleşmelerini Anlama

Her işlemci adımının, veri yapısını nasıl değiştirdiğini açıklayan bir transform_features() yöntemi vardır:

# Get the expected output features from your pipeline
initial_features = {
    PipelineFeatureType.OBSERVATION: {
        "observation.state": PolicyFeature(type=FeatureType.STATE, shape=(7,)),
        "observation.image": PolicyFeature(type=FeatureType.VISUAL, shape=(3, 224, 224))
    },
    PipelineFeatureType.ACTION: {
        "action": PolicyFeature(type=FeatureType.ACTION, shape=(4,))
    }
}

# Check what your pipeline will output
output_features = processor.transform_features(initial_features)

print("Input features:")
for feature_type, features in initial_features.items():
    print(f"  {feature_type}:")
    for key, feature in features.items():
        print(f"    {key}: {feature.type.value}, shape={feature.shape}")

print("\nOutput features:")
for feature_type, features in output_features.items():
    print(f"  {feature_type}:")
    for key, feature in features.items():
        print(f"    {key}: {feature.type.value}, shape={feature.shape}")

Beklenen Özelliklerin Doğrulanması

Hattınızın beklediğiniz özellikleri ürettiğinden emin olun:

# Define what features you expect the pipeline to produce
expected_keys = ["observation.state", "observation.image", "action"]

print("Validating feature contract...")
for expected_key in expected_keys:
    found = False
    for feature_type, features in output_features.items():
        if expected_key in features:
            feature = features[expected_key]
            print(f"✅ {expected_key}: {feature.type.value}, shape={feature.shape}")
            found = True
            break

    if not found:
        print(f"❌ Missing expected feature: {expected_key}")

Bu doğrulama, belirli veri yapılarını bekleyen aşağı akış bileşenleriyle hattınızın doğru çalışmasını sağlamak için yardımcı olur.

Özet

Üç hata ayıklama yaklaşımını anladığınıza göre, herhangi bir boru hattı sorununu sistematik olarak ele alabilirsiniz:

1. Hook'lar - Pipeline kodunu değiştirmeden çalışma zamanı izleme ve doğrulama için
2. Adım adım inceleme - Ara durumları denetlemek ve dönüşümleri anlamak için
3. Özellik doğrulama - Veri yapısı sözleşmelerinin karşılandığından emin olmak için

Her yaklaşımın ne zaman kullanılacağı:

• Ne zaman pipeline'ınızın ne yaptığını anlamaya ya da beklenmedik bir durumla karşılaştığınızda adım adım hata ayıklamayı kullanmaya başlayın
• Geliştirme ve üretim sırasında sorunları otomatik olarak yakalamak için sürekli izleme amacıyla kanca ekleyin
• Dağıtım öncesinde hattınızın aşağı akış bileşenleriyle çalıştığını doğrulamak için özellik doğrulaması kullanın

Bu üç araç birlikte çalışarak karmaşık boru hatlarının doğal olarak eksik bıraktığı tam gözlemlenebilirliği sağlar. Sorunları izleyen kancalar, davranışı anlamanıza yardımcı olan adım adım inceleme ve uyumluluğu garanti eden özellik doğrulaması sayesinde herhangi bir boru hatını güvenle ve verimli bir şekilde hata ayıklayabilirsiniz.