Debug Pipeline Prosesor Anda

Pipeline prosesor dapat menjadi kompleks, terutama saat merangkai beberapa langkah transformasi. Berbeda dengan panggilan fungsi sederhana, pipeline tidak memiliki observabilitas alami; Anda tidak dapat dengan mudah melihat apa yang terjadi di antara setiap langkah atau di mana masalah terjadi. Panduan ini menyediakan alat dan teknik debugging yang dirancang khusus untuk mengatasi tantangan ini dan membantu Anda memahami aliran data melalui pipeline Anda.

Kita akan menjelajahi tiga pendekatan debugging yang saling melengkapi: hooks untuk pemantauan runtime, step-through debugging untuk inspeksi terperinci, dan feature validation untuk menangkap ketidakcocokan struktural. Masing-masing memiliki tujuan berbeda dan secara bersama-sama memberikan visibilitas lengkap terhadap perilaku pipeline Anda.

Memahami Hooks

Hooks adalah fungsi yang dipanggil pada titik-titik tertentu selama eksekusi pipeline. Mereka menyediakan cara untuk memeriksa, memantau, atau memodifikasi data tanpa mengubah kode pipeline Anda. Anggaplah mereka sebagai “event listeners” untuk pipeline Anda.

Apa itu Hook?

Hook adalah fungsi callback yang dipanggil secara otomatis pada momen-momen tertentu selama eksekusi pipeline. Konsep ini berasal dari pemrograman berbasis event, bayangkan Anda dapat “menyematkan diri” ke dalam alur eksekusi pipeline untuk mengamati atau bereaksi terhadap apa yang sedang terjadi.

Anggaplah hooks sebagai checkpoint yang disisipkan ke dalam pipeline Anda. Setiap kali pipeline mencapai salah satu checkpoint ini, ia akan berhenti sejenak untuk memanggil fungsi hook Anda, memberi Anda kesempatan untuk memeriksa status saat ini, mencatat informasi, dan memvalidasi data.

Hook hanyalah fungsi yang menerima dua parameter:

• step_idx: int - Indeks dari langkah pemrosesan saat ini (0, 1, 2, dst.)
• transition: EnvTransition - Transisi data pada titik tersebut dalam pipeline

Keindahan dari hooks terletak pada sifatnya yang non-invasif: Anda dapat menambahkan logika pemantauan, validasi, atau debugging tanpa mengubah satu baris pun kode pipeline Anda. Pipeline tetap bersih dan berfokus pada logika intinya, sementara hooks menangani kekhawatiran lintas-cutting seperti logging, monitoring, dan debugging.

Hook Sebelum vs Sesudah

Pipeline mendukung dua jenis hook:

• Hook sebelum (register_before_step_hook) - Dipanggil sebelum setiap langkah dieksekusi
• Setelah hook (register_after_step_hook) - Dipanggil setelah setiap langkah selesai

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)

Implementasi Hook Deteksi NaN

Berikut adalah contoh praktis hook yang mendeteksi nilai NaN:

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)

Cara Kerja Hooks Secara Internal

Memahami mekanisme internal membantu Anda menggunakan hook secara lebih efektif. Pipeline mempertahankan dua daftar terpisah: satu untuk hook before-step dan yang lain untuk hook after-step. Saat Anda mendaftarkan hook, hook tersebut hanya ditambahkan ke daftar yang sesuai.

Selama eksekusi, pipeline mengikuti urutan yang ketat: untuk setiap langkah pemrosesan, pipeline pertama-tama memanggil semua before-hook sesuai urutan registrasi, kemudian menjalankan transformasi langkah yang sebenarnya, dan akhirnya memanggil semua after-hook sesuai urutan registrasi. Hal ini menciptakan struktur yang dapat diprediksi, menyerupai sandwich, di sekitar setiap langkah.

Insight utamanya adalah bahwa hook tidak mengubah logika inti pipeline—mereka murni bersifat aditif. Metode _forward dari pipeline mengorkestrasi interaksi antara hook dan langkah pemrosesan, memastikan bahwa kode debugging atau monitoring Anda berjalan pada saat yang tepat tanpa mengganggu aliran data utama.

Berikut adalah tampilan sederhana tentang cara pipeline mengeksekusi hook:

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)

Alur Eksekusi

Alur eksekusinya terlihat seperti ini:

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

Sebagai contoh, dengan 3 langkah dan kedua jenis hook:

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

Beberapa Hook

Anda dapat mendaftarkan beberapa hook dari tipe yang sama - mereka dieksekusi sesuai urutan pendaftaran:

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)

Meskipun hook sangat berguna untuk memantau masalah tertentu (seperti deteksi NaN) atau mengumpulkan metrik selama eksekusi pipeline normal, terkadang Anda perlu menyelidiki lebih dalam. Ketika Anda ingin memahami persis apa yang terjadi di setiap langkah atau men-debug logika transformasi yang kompleks, step-through debugging memberikan inspeksi terperinci yang Anda butuhkan.

Debugging Langkah-Demi-Langkah

Debugging langkah-demi-langkah ibarat memiliki replay lambat untuk pipeline Anda. Alih-alih melihat data Anda ditransformasi dalam satu kilatan cepat dari input ke output, Anda dapat berhenti dan memeriksa apa yang terjadi setelah setiap langkah individual.

Pendekatan ini sangat berharga ketika Anda berusaha memahami pipeline yang kompleks, men-debug perilaku yang tidak terduga, atau memverifikasi bahwa setiap transformasi bekerja sesuai harapan. Berbeda dengan hook, yang sangat baik untuk pemantauan otomatis, step-through debugging memberi Anda kontrol manual dan interaktif atas proses inspeksi.

Metode step_through() adalah generator yang menghasilkan keadaan transisi setelah setiap langkah pemrosesan, sehingga Anda dapat memeriksa hasil antara. Anggap saja ini sebagai pembuatan serangkaian snapshot data Anda saat mengalir melalui pipeline—setiap snapshot menunjukkan persis bagaimana data Anda terlihat setelah satu transformasi lagi diterapkan.

Cara Kerja Step-Through

Metode step_through() secara fundamental mengubah cara pipeline dieksekusi. Alih-alih menjalankan semua langkah secara berurutan dan hanya mengembalikan hasil akhir, metode ini mengubah pipeline menjadi iterator yang menghasilkan hasil perantara.

Berikut yang terjadi secara internal: metode ini dimulai dengan mengonversi data input Anda ke format transisi internal pipeline, lalu menghasilkan (yield) state awal ini. Selanjutnya, metode ini menerapkan langkah pemrosesan pertama dan menghasilkan hasilnya. Kemudian, metode ini menerapkan langkah kedua pada hasil tersebut dan menghasilkan lagi, dan seterusnya. Setiap yield memberikan Anda snapshot lengkap dari transisi pada titik tersebut.

Pola generator ini sangat kuat karena bersifat *lazy*—pipeline hanya menghitung langkah berikutnya saat Anda memintanya. Artinya, Anda dapat berhenti di titik mana pun, memeriksa keadaan saat ini secara menyeluruh, dan memutuskan apakah akan melanjutkan. Anda tidak dipaksa untuk menjalankan seluruh pipeline hanya untuk men-debug satu langkah yang bermasalah.

Alih-alih menjalankan seluruh pipeline dan hanya melihat hasil akhirnya, step_through() berhenti setelah setiap langkah dan memberikan transisi perantara:

# 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}")

Debugging Interaktif dengan Breakpoints

Anda dapat menambahkan breakpoint di loop step-through untuk melakukan debug secara interaktif:

# 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

Selama sesi debugger, Anda dapat:

• Periksa intermediate[TransitionKey.OBSERVATION] untuk melihat data observasi
• Periksa intermediate[TransitionKey.ACTION] untuk transformasi aksi
• Periksa bagian mana pun dari transisi untuk memahami apa yang dilakukan oleh setiap langkah

Debugging step-through sangat cocok untuk memahami transformasi data, tetapi bagaimana dengan struktur data tersebut? Meskipun hook dan step-through membantu Anda men-debug perilaku runtime, Anda juga perlu memastikan bahwa pipeline Anda menghasilkan data dalam format yang diharapkan oleh komponen downstream. Di sinilah validasi kontrak fitur berperan.

Memvalidasi Kontrak Fitur

Kontrak fitur mendefinisikan struktur data yang pipeline Anda harapkan sebagai input dan hasilkan sebagai output. Validasi kontrak ini membantu mendeteksi ketidakcocokan sejak dini.

Memahami Kontrak Fitur

Setiap langkah pemroses memiliki metode transform_features() yang menjelaskan bagaimana struktur data diubah:

# 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}")

Memverifikasi Fitur yang Diharapkan

Periksa bahwa pipeline Anda menghasilkan fitur yang Anda harapkan:

# 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}")

Validasi ini membantu memastikan bahwa pipeline Anda akan berfungsi dengan benar bersama komponen hilir yang mengharapkan struktur data tertentu.

Ringkasan

Sekarang setelah Anda memahami tiga pendekatan debugging, Anda dapat menangani masalah pipeline apa pun secara sistematis:

1. Hooks - Untuk pemantauan dan validasi saat runtime tanpa memodifikasi kode pipeline
2. Step-through - Untuk memeriksa keadaan perantara dan memahami transformasi
3. Validasi fitur - Untuk memastikan kontrak struktur data terpenuhi

Kapan menggunakan masing-masing pendekatan:

• Mulai dengan step-through debugging ketika Anda perlu memahami apa yang dilakukan pipeline Anda atau ketika terjadi sesuatu yang tidak terduga
• Tambahkan hooks untuk pemantauan berkelanjutan selama pengembangan dan produksi guna mendeteksi masalah secara otomatis
• Gunakan feature validation sebelum deployment untuk memastikan pipeline Anda berfungsi dengan komponen downstream

Ketiga alat ini bekerja sama untuk memberikan observabilitas lengkap yang secara alami tidak dimiliki oleh pipeline yang kompleks. Dengan hook yang memantau masalah, step-through yang membantu Anda memahami perilaku, dan validasi fitur yang memastikan kompatibilitas, Anda akan dapat men-debug pipeline apa pun dengan percaya diri dan efisien.