Gỡ lỗi cho Pipeline của Processor

Các pipeline của Processor có thể phức tạp, đặc biệt khi kết nối nhiều bước biến đổi. Khác với các lời gọi hàm đơn giản, pipeline thiếu khả năng quan sát tự nhiên, bạn không thể dễ dàng nhìn thấy điều gì xảy ra giữa các bước hoặc nơi nào gặp sự cố. Hướng dẫn này cung cấp các công cụ và kỹ thuật gỡ lỗi được thiết kế đặc biệt để giải quyết những thách thức này và giúp bạn hiểu luồng dữ liệu chạy qua các pipeline của mình.

Chúng ta sẽ khám phá ba phương pháp gỡ lỗi bổ trợ cho nhau: hooks để giám sát thời gian thực thi, step-through debugging để kiểm tra chi tiết, và feature validation để phát hiện các sai lệch về cấu trúc. Mỗi phương pháp phục vụ một mục đích khác nhau và khi kết hợp lại, chúng cung cấp khả năng quan sát toàn diện về hành vi của pipeline.

Hiểu về Hooks

Hooks là các hàm được gọi tại các điểm cụ thể trong quá trình thực thi pipeline. Chúng cung cấp cách để kiểm tra, giám sát hoặc sửa đổi dữ liệu mà không cần thay đổi mã pipeline của bạn. Hãy nghĩ về chúng như là “event listeners” cho pipeline của bạn.

Hook là gì?

Một hook là một hàm callback được gọi tự động tại các thời điểm cụ thể trong quá trình thực thi pipeline. Khái niệm này bắt nguồn từ lập trình hướng sự kiện, hãy tưởng tượng bạn có thể “gắn vào” luồng thực thi của pipeline để quan sát hoặc phản hồi với những gì đang diễn ra.

Hãy nghĩ về hooks như việc chèn các điểm kiểm tra (checkpoints) vào pipeline của bạn. Mỗi khi pipeline đạt đến một trong những điểm kiểm tra này, nó sẽ tạm dừng một chút để gọi hàm hook của bạn, cho bạn cơ hội kiểm tra trạng thái hiện tại, ghi log thông tin và xác thực dữ liệu.

Một hook đơn giản chỉ là một hàm nhận hai tham số:

• step_idx: int - Chỉ số của bước xử lý hiện tại (0, 1, 2, v.v.)
• transition: EnvTransition - Chuyển tiếp dữ liệu tại điểm đó trong pipeline

Điểm đẹp của hooks nằm ở tính chất không xâm lấn: bạn có thể thêm logic giám sát, xác thực hoặc gỡ lỗi mà không cần thay đổi dù chỉ một dòng trong mã pipeline của mình. Pipeline vẫn giữ được sự sạch sẽ và tập trung vào logic cốt lõi, trong khi hooks xử lý các mối quan tâm xuyên suốt như ghi log, giám sát và gỡ lỗi.

Hooks Trước và Sau

Pipeline hỗ trợ hai loại hook:

• Các hook trước (register_before_step_hook) - Được gọi trước khi mỗi bước được thực thi
• Sau các hook (register_after_step_hook) - Được gọi sau khi mỗi bước hoàn thành

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)

Thực hiện một Hook phát hiện NaN

Dưới đây là một ví dụ thực tế về một hook phát hiện các giá trị 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)

Cách Hooks Hoạt Động Bên Trong

Việc hiểu cơ chế bên trong giúp bạn sử dụng hooks hiệu quả hơn. Pipeline duy trì hai danh sách riêng biệt: một cho các hook trước bước và một cho các hook sau bước. Khi bạn đăng ký một hook, nó đơn giản được thêm vào danh sách tương ứng.

Trong quá trình thực thi, pipeline tuân theo một trình tự nghiêm ngặt: đối với mỗi bước xử lý, nó trước tiên gọi tất cả các before-hooks theo thứ tự đăng ký, sau đó thực hiện phép biến đổi của bước đó, và cuối cùng gọi tất cả các after-hooks theo thứ tự đăng ký. Điều này tạo ra một cấu trúc dự đoán được, giống như kẹp sandwich, bao quanh mỗi bước.

Điểm mấu chốt là các hook không thay đổi logic của pipeline cốt lõi — chúng hoàn toàn mang tính bổ sung. Phương thức _forward của pipeline điều phối sự phối hợp giữa các hook và các bước xử lý, đảm bảo rằng mã gỡ lỗi hoặc giám sát của bạn được thực hiện vào đúng thời điểm mà không gây cản trở cho luồng dữ liệu chính.

Dưới đây là cách đơn giản hóa để xem pipeline thực thi các 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)

Luồng thực thi

Luồng thực thi trông như sau:

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

Ví dụ, với 3 bước và cả hai loại 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

Nhiều Hooks

Bạn có thể đăng ký nhiều hook cùng một loại - chúng sẽ được thực thi theo thứ tự đăng ký:

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)

Mặc dù hooks rất hữu ích để theo dõi các vấn đề cụ thể (như phát hiện NaN) hoặc thu thập các chỉ số trong quá trình thực thi pipeline bình thường, đôi khi bạn cần đi sâu hơn. Khi bạn muốn hiểu chính xác điều gì xảy ra ở từng bước hoặc gỡ lỗi cho logic biến đổi phức tạp, việc gỡ lỗi từng bước (step-through debugging) cung cấp cho bạn khả năng kiểm tra chi tiết cần thiết.

Gỡ lỗi từng bước

Gỡ lỗi từng bước giống như có một bản phát lại chậm cho pipeline của bạn. Thay vì chỉ nhìn thấy dữ liệu của bạn bị biến đổi trong một khoảnh khắc mờ nhòe từ đầu vào đến đầu ra, bạn có thể tạm dừng và kiểm tra những gì xảy ra sau từng bước riêng lẻ.

Cách tiếp cận này đặc biệt có giá trị khi bạn đang cố gắng hiểu một pipeline phức tạp, gỡ lỗi cho hành vi bất ngờ, hoặc xác minh rằng mỗi phép biến đổi đang hoạt động như mong đợi. Khác với hooks, vốn rất hữu ích cho việc giám sát tự động, gỡ lỗi từng bước (step-through debugging) cho bạn quyền kiểm soát thủ công và tương tác đối với quá trình kiểm tra.

Phương thức step_through() là một generator, trả về trạng thái chuyển tiếp sau mỗi bước xử lý, cho phép bạn kiểm tra các kết quả trung gian. Hãy nghĩ về nó như việc tạo ra một loạt các bản chụp (snapshot) của dữ liệu khi nó chảy qua pipeline—mỗi bản chụp cho bạn thấy chính xác dữ liệu của bạn trông như thế nào sau khi một phép biến đổi khác được áp dụng.

Cách hoạt động của Step-Through

Phương thức step_through() thay đổi cơ bản cách pipeline thực thi. Thay vì chạy tất cả các bước theo trình tự và chỉ trả về kết quả cuối cùng, nó biến pipeline thành một iterator (trình lặp) để trả về các kết quả trung gian.

Dưới đây là những gì diễn ra bên trong: phương thức bắt đầu bằng cách chuyển đổi dữ liệu đầu vào của bạn sang định dạng chuyển tiếp nội bộ của pipeline, sau đó trả về trạng thái ban đầu này. Tiếp theo, nó áp dụng bước xử lý đầu tiên và trả về kết quả. Sau đó, nó áp dụng bước thứ hai lên kết quả đó và lại trả về, cứ thế tiếp tục. Mỗi yield cung cấp cho bạn một bản chụp hoàn chỉnh của quá trình chuyển tiếp tại thời điểm đó.

Mô hình generator này rất mạnh mẽ vì nó mang tính lười (lazy)—pipeline chỉ tính toán bước tiếp theo khi bạn yêu cầu. Điều này có nghĩa là bạn có thể dừng lại ở bất kỳ thời điểm nào, kiểm tra kỹ lưỡng trạng thái hiện tại và quyết định xem có nên tiếp tục hay không. Bạn không bị buộc phải chạy toàn bộ pipeline chỉ để gỡ lỗi một bước có vấn đề.

Thay vì chạy toàn bộ pipeline và chỉ xem kết quả cuối cùng, step_through() sẽ tạm dừng sau mỗi bước và cung cấp cho bạn trạng thái chuyển tiếp trung gian:

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

Gỡ lỗi tương tác bằng điểm dừng

Bạn có thể thêm các điểm dừng (breakpoints) trong vòng lặp chạy từng bước (step-through) để gỡ lỗi tương tác:

# 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

Trong phiên gỡ lỗi, bạn có thể:

• Kiểm tra intermediate[TransitionKey.OBSERVATION] để xem dữ liệu quan sát
• Kiểm tra intermediate[TransitionKey.ACTION] để xem các phép biến đổi hành động
• Kiểm tra bất kỳ phần nào của quá trình chuyển đổi để hiểu từng bước thực hiện điều gì

Gỡ lỗi theo từng bước (step-through debugging) rất hữu ích để hiểu các phép biến đổi dữ liệu, nhưng còn cấu trúc của dữ liệu đó thì sao? Trong khi hooks và step-through giúp bạn gỡ lỗi hành vi lúc chạy, bạn cũng cần đảm bảo pipeline của mình tạo ra dữ liệu đúng định dạng mà các thành phần phía sau kỳ vọng. Đây chính là lúc việc xác thực hợp đồng đặc trưng (feature contract validation) phát huy tác dụng.

Xác thực Hợp đồng Tính năng

Hợp đồng tính năng (Feature contracts) xác định cấu trúc dữ liệu mà pipeline của bạn kỳ vọng nhận vào và tạo ra. Kiểm tra tính hợp lệ của các hợp đồng này giúp phát hiện sớm các sự không khớp.

Hiểu về Hợp đồng Tính năng

Mỗi bước xử lý có phương thức transform_features() mô tả cách nó thay đổi cấu trúc dữ liệu:

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

Xác minh các tính năng mong đợi

Kiểm tra xem pipeline của bạn có tạo ra các đặc trưng như mong đợi không:

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

Việc xác thực này giúp đảm bảo pipeline của bạn sẽ hoạt động chính xác với các thành phần phía sau kỳ vọng các cấu trúc dữ liệu cụ thể.

Tóm tắt

Giờ đây, khi bạn đã hiểu ba cách tiếp cận gỡ lỗi, bạn có thể xử lý mọi vấn đề trong pipeline một cách có hệ thống:

1. Hooks - Để giám sát và xác thực thời gian chạy mà không cần sửa đổi mã pipeline
2. Điều tra từng bước - Để kiểm tra các trạng thái trung gian và hiểu các phép biến đổi
3. Đánh giá tính năng - Để đảm bảo các thỏa thuận về cấu trúc dữ liệu được đáp ứng

Khi nào nên sử dụng từng phương pháp:

• Bắt đầu với điều khiển từng bước khi bạn cần hiểu pipeline của mình hoạt động như thế nào hoặc khi có điều gì đó bất ngờ xảy ra
• Thêm hooks để giám sát liên tục trong quá trình phát triển và sản xuất nhằm tự động phát hiện các vấn đề
• Sử dụng feature validation trước khi triển khai để đảm bảo pipeline của bạn hoạt động với các thành phần phía sau

Ba công cụ này phối hợp với nhau để mang lại khả năng quan sát đầy đủ mà các pipeline phức tạp thường thiếu. Với hooks giám sát các vấn đề, step-through giúp bạn hiểu rõ hành vi, và feature validation đảm bảo tính tương thích, bạn sẽ có thể gỡ lỗi bất kỳ pipeline nào một cách tự tin và hiệu quả.