Status: Needs Review

This page has not been reviewed for accuracy and completeness. Content may be outdated or contain errors.

---

Execution Stages

Control when nodes execute during pipeline operation via stage-aware graph execution.

Execution stages enable conditional node execution based on context (training, validation, testing, or inference). Essential for loss nodes (training only), metrics (validation/test only), and optimization-specific operations.

Key capabilities:

• Five execution stages: ALWAYS, TRAIN, VAL, TEST, INFERENCE
• Stage-aware filtering in pipeline executor
• Context object carries stage, epoch, batch_idx, global_step
• Default to ALWAYS for core processing nodes
• Restrict stages only when needed

---

Stage System Overview

Every node declares the stages it runs in on the class body, next to _category and _tags:

from cuvis_ai_core.node import Node
from cuvis_ai_schemas.enums import ExecutionStage

class MyNode(Node):
    # A set literal (or stage names such as "train") is normalized to a
    # frozenset when the class is created; an unknown name fails at import.
    EXECUTION_STAGES = {ExecutionStage.TRAIN, ExecutionStage.VAL}

Instances read the declaration through node.execution_stages, which is read-only: every instance of a class runs in the same stages, and a pipeline yaml picks stages by picking the class (see Pattern 5 for running the evaluation nodes of a trained pipeline).

Key Concepts:

Concept	Description
ExecutionStage Enum	ALWAYS, TRAIN, VAL, TEST, INFERENCE
EXECUTION_STAGES	Class-level declaration, a frozenset[ExecutionStage]
node.execution_stages	Read-only view of the class declaration; not a constructor argument
Context Object	Runtime info (stage, epoch, batch_idx) passed to nodes
Stage Filtering	Pipeline skips nodes not matching current stage
Default	Node.EXECUTION_STAGES is {ExecutionStage.ALWAYS}

Execution Flow

graph LR
    A[Pipeline.forward<br/>context.stage=TRAIN] --> B{For Each Node}
    B --> C{Node.execution_stages<br/>contains stage?}
    C -->|Yes| D[Execute Node]
    C -->|No| E[Skip Node]
    D --> F[Continue]
    E --> F
    classDef success fill:#d4edda,stroke:#28a745,color:#155724
    classDef error fill:#f8d7da,stroke:#dc3545,color:#721c24
    class D success
    class E error

---

The Five Execution Stages

1. ALWAYS (Default)

Node executes unconditionally in all stages.

Use Cases:

• Core data transformation
• Feature extraction
• Essential preprocessing

class FeatureExtractor(Node):
    # Inherits Node.EXECUTION_STAGES == {ExecutionStage.ALWAYS}

    def forward(self, data, **_):
        # Runs during TRAIN, VAL, TEST, INFERENCE
        features = self.extract_features(data)
        return {"features": features}

# Default behavior
node1 = FeatureExtractor()  # runs in TRAIN, VAL, TEST and INFERENCE

---

2. TRAIN

Node only executes during training. Use for: data augmentation, dropout, training loss.

class TrainingAugmentation(Node):
    EXECUTION_STAGES = {ExecutionStage.TRAIN}

3. VAL (Validation)

Node only executes during validation. Use for: validation metrics, model selection criteria.

class ValidationMetrics(Node):
    EXECUTION_STAGES = {ExecutionStage.VAL}

4. TEST

Node only executes during testing. Use for: final metrics, performance benchmarking.

class TestEvaluator(Node):
    EXECUTION_STAGES = {ExecutionStage.TEST}

5. INFERENCE

Node only executes during inference/prediction. Use for: production post-processing, output formatting.

class InferencePostProcessor(Node):
    EXECUTION_STAGES = {ExecutionStage.INFERENCE}

---

Stage-Aware Node Patterns

Pattern 1: Single Stage

class TrainOnlyAugmentation(Node):
    EXECUTION_STAGES = {ExecutionStage.TRAIN}

Pattern 2: Multiple Stages

class MetricsNode(Node):
    # Execute during validation and test only
    EXECUTION_STAGES = {ExecutionStage.VAL, ExecutionStage.TEST}

Pattern 3: Training-Aware (Common for Loss)

from cuvis_ai.node.losses import LossNode

class MyLoss(LossNode):
    # LossNode declares EXECUTION_STAGES = {TRAIN, VAL, TEST}; subclasses inherit it
    def forward(self, predictions, targets, **_):
        return {"loss": self.compute_loss(predictions, targets)}

Pattern 4: Stage-Conditional Behavior

class AdaptiveNormalizer(Node):
    EXECUTION_STAGES = {ExecutionStage.TRAIN, ExecutionStage.INFERENCE}

    def forward(self, data, context: Context, **_):
        if context.stage == ExecutionStage.TRAIN:
            self.update_running_stats(data)
        normalized = self.normalize(data)
        return {"normalized": normalized}

Pattern 5: Evaluation Nodes on a Trained Pipeline

A pipeline yaml cannot move a node between stages: hparams are constructor arguments and execution_stages is not one. A yaml written for an earlier release may still carry the inert execution_stages: null, which is accepted and dropped; any other value fails at load with a TypeError naming the node. The TensorBoardMonitorNode and the artifact visualizers run in train/val/test, so restore-pipeline and Predictor.predict() (the inference stage) write no TensorBoard output. To log a trained pipeline's artifacts and metrics over a labeled split, run it at the test stage:

from cuvis_ai_core.training.predictor import Predictor
from cuvis_ai_schemas.enums import ExecutionStage

Predictor(pipeline, datamodule).predict(stage=ExecutionStage.TEST)

Plugin manifests and the gRPC NodeInfo expose the class declaration (category and lifecycle tags), which is also exactly what the pipeline runs.

Trained pipelines at inference

A pipeline trained in CuvisNEXT is loaded for prediction at ExecutionStage.INFERENCE, which prunes its losses, metrics, visualizers and TensorBoard sink; nothing else is needed to run the trained graph. Runs saved before cuvis-ai 0.14.0 carry a QuantileBinaryDecider, which flags a fixed fraction of every frame and cannot be gated from the picker; retrain them to get the two-stage decider with its optional gate.

---

Pipeline Execution with Context

Context Object

Runtime information passed to pipeline.

from cuvis_ai_schemas.enums import ExecutionStage
from cuvis_ai_schemas.execution import Context

context = Context(
    stage=ExecutionStage.TRAIN,  # Current stage
    epoch=5,                     # Training epoch
    batch_idx=42,                # Batch index
    global_step=1337             # Global training step
)

Attributes:

Attribute	Type	Description
stage	ExecutionStage	Current stage
epoch	int	Training epoch (0-indexed)
batch_idx	int	Batch index in epoch (0-indexed)
global_step	int	Global step counter

Passing Context to Pipeline

from cuvis_ai_core.pipeline.pipeline import CuvisPipeline
from cuvis_ai_schemas.enums import ExecutionStage
from cuvis_ai_schemas.execution import Context

pipeline = CuvisPipeline()

# Training
train_context = Context(stage=ExecutionStage.TRAIN, epoch=0, batch_idx=0)
train_outputs = pipeline.forward(batch=train_batch, context=train_context)

# Validation
val_context = Context(stage=ExecutionStage.VAL, epoch=0, batch_idx=0)
val_outputs = pipeline.forward(batch=val_batch, context=val_context)

# Inference
inference_context = Context(stage=ExecutionStage.INFERENCE)
inference_outputs = pipeline.forward(batch=inference_batch, context=inference_context)

---

Data Flow Patterns

Training vs Inference Paths

graph TD
    A[Data Input<br/>ALWAYS] --> B[Augmentation<br/>TRAIN]
    A --> C[Feature Extraction<br/>ALWAYS]
    B --> C
    C --> D[Model<br/>ALWAYS]
    D --> E[Loss<br/>TRAIN/VAL/TEST]
    D --> F[Metrics<br/>VAL/TEST]
    D --> G[Inference Optimizer<br/>INFERENCE]
    classDef success fill:#d4edda,stroke:#28a745,color:#155724
    classDef error fill:#f8d7da,stroke:#dc3545,color:#721c24
    class E,F success
    class G error

---

Lifecycle Tags

Execution stages are invisible outside the process: they are not in the pipeline yaml, not in plugin manifests and not on gRPC. What the manifests and NodeInfo do expose is the node _category and its _tags, and three tags describe the lifecycle: TRAINING, EVALUATION, INFERENCE. For the declaration and the exposed metadata to tell the same story, tests/test_node_lifecycle_consistency.py enforces four rules on every builtin class:

Rule	Statement
R0	Category LOSS or REGULARIZER never runs at inference.
R1	A node that never runs at inference carries TRAINING or EVALUATION.
R2	A node tagged TRAINING or EVALUATION but not INFERENCE never runs at inference.
R3	A node tagged INFERENCE runs at inference.

Category decides only for losses: METRIC, SINK and VISUALIZER are too coarse (video writers are sinks that exist for inference; DistinctLabelCount is a metric that runs at inference, so it carries INFERENCE). Nodes tagged AUGMENTATION are exempt from R2: the occlusion family sits inline (cube -> occlusion -> model), so pruning it by stage would leave the model's input unsatisfied; it stays ALWAYS until it gains a stage-aware pass-through. The rules read the class declarations, which are exactly what every instance runs with.

---

Best Practices

1. Use Semantic Stage Selection

# GOOD: Loss computes during training phases
class TrainingLoss(LossNode):
    pass  # Inherits {TRAIN, VAL, TEST} from LossNode

# BAD: Loss executing during inference
class BadLoss(Node):
    _category = NodeCategory.LOSS
    EXECUTION_STAGES = {ExecutionStage.ALWAYS}  # the consistency test rejects this

1. Default to ALWAYS for Core Nodes

# GOOD: Applies everywhere
class FeatureExtractor(Node):
    pass  # Inherits Node.EXECUTION_STAGES == {ALWAYS}

# BAD: Unnecessary restriction
class OverRestricted(Node):
    EXECUTION_STAGES = {ExecutionStage.TRAIN, ExecutionStage.INFERENCE}

1. Separate Concerns with Multiple Nodes

# GOOD: Separate nodes for different behaviors
class TrainingPostProcessor(Node):
    EXECUTION_STAGES = {ExecutionStage.TRAIN}

class InferencePostProcessor(Node):
    EXECUTION_STAGES = {ExecutionStage.INFERENCE}

1. Document Stage Decisions — Add docstrings explaining why a node is restricted to specific stages.

2. Test All Stages — Verify pipeline behavior in TRAIN, VAL, TEST, and INFERENCE stages.

---

Troubleshooting

Node Not Executing

Diagnosis:

logger.info(f"Node {node.name} stages: {node.execution_stages}")
logger.info(f"Current stage: {context.stage}")

Solution: Verify stage match

# Node class restricted to TRAIN
class MyNode(Node):
    EXECUTION_STAGES = {ExecutionStage.TRAIN}

# But running VAL - won't execute!
context = Context(stage=ExecutionStage.VAL)

# Fix: add VAL to the class declaration
class MyNode(Node):
    EXECUTION_STAGES = {ExecutionStage.TRAIN, ExecutionStage.VAL}

Unexpected Node Execution

Solution: Check the parent class declaration

class MyMetric(MetricNode):  # Parent declares EXECUTION_STAGES = {VAL, TEST}
    # A subclass inherits the parent's stages unless it redeclares EXECUTION_STAGES
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.threshold = 0.5

Loss Not Computing in Validation

Solution: Use LossNode base class

from cuvis_ai.node.losses import LossNode

class MyLoss(LossNode):
    # Auto-configured {TRAIN, VAL, TEST}
    def forward(self, predictions, targets, **_):
        return {"loss": self.compute_loss(predictions, targets)}

Context Not Available in Node

Fix: Add context to INPUT_SPECS

class MyNode(Node):
    INPUT_SPECS = {
        "data": PortSpec(dtype=torch.float32, shape=(-1, -1)),
        "context": PortSpec(dtype=Context, shape=()),  # Add this
    }

    def forward(self, data, context: Context, **_):
        if context.stage == ExecutionStage.TRAIN:
            return {"output": self.train_process(data)}
        return {"output": self.eval_process(data)}

Pipeline automatically injects context for nodes declaring it in INPUT_SPECS.

Stage Enum Comparison Failing

Solution: Compare enum to enum, not string

# BAD
if context.stage == "train":  # Never matches!
    pass

# GOOD
if context.stage == ExecutionStage.TRAIN:
    pass

# ALSO GOOD
if context.stage.value == "train":
    pass

Optimization Tips

1. Minimize Overhead in High-Frequency Stages

   # BAD: Expensive visualization in training
   class BadVisualizer(Node):
       EXECUTION_STAGES = {ExecutionStage.ALWAYS}
   
       def forward(self, data, **_):
           expensive_viz = self.render_3d_plot(data)  # Runs every batch!
           return {"visualization": expensive_viz}
   
   # GOOD: Visualization only during validation and test
   class GoodVisualizer(Node):
       EXECUTION_STAGES = {ExecutionStage.VAL, ExecutionStage.TEST}
   

2. Reduce Complexity in Inference

   class AdaptiveProcessor(Node):
       def forward(self, data, context: Context, **_):
           if context.stage == ExecutionStage.INFERENCE:
               return {"result": self.fast_forward(data)}
           else:
               return {"result": self.full_forward(data)}
   

3. Skip Gradients in Non-Training

   # Automatically handled by trainer
   with torch.no_grad():
       outputs = pipeline.forward(batch=val_batch, context=val_context)