5 Python Decorators to Stabilize Your Machine Learning Pipeline

AI projects quickly accumulate boilerplate code for API calls, retries, logging, caching, and validation, which distracts from core model logic. Experienced Python engineers address this by encapsulating cross‑cutting concerns in reusable decorators.

1. Concurrency Limiter

When many inference requests run in parallel, GPUs or external APIs can be overwhelmed. A concurrency limiter caps the number of simultaneous tasks using a semaphore, queuing excess work.

Avoids GPU memory exhaustion under heavy load.

Reduces API rate‑limit failures caused by uncontrolled parallelism.

Improves stability of chatbots or recommendation systems during traffic spikes.

import threading
import time

semaphore = threading.Semaphore(3)

def concurrency_limit(func):
    def wrapper(*args, **kwargs):
        with semaphore:
            print(f"Running {func.__name__}")
            return func(*args, **kwargs)
    return wrapper

@concurrency_limit
def process_ai_task(task):
    print(f"Task {task} started")
    time.sleep(2)
    print(f"Task {task} completed")

2. Structured Machine Learning Logger

Large‑scale ML projects generate massive unstructured logs, making debugging difficult. A structured logger emits JSON records containing the function name, execution time, and status, which are machine‑readable and easier to monitor.

Facilitates debugging of distributed training and inference pipelines.

Exposes performance bottlenecks and improves reliability of production AI systems.

Integrates with enterprise‑grade observability pipelines.

import time
import json
from functools import wraps

def ml_logger(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        log = {
            "function": func.__name__,
            "execution_time": round(time.time() - start, 2),
            "status": "success"
        }
        print(json.dumps(log, indent=4))
        return result
    return wrapper

@ml_logger
def train_model():
    time.sleep(2)
    return "Training Complete"

3. Feature Injector

Raw input data often lacks engineered features required by models. The feature injector adds derived fields before the model runs, separating feature engineering from prediction logic for better maintainability.

Extracts feature engineering from core prediction code.

Reduces duplicated preprocessing across multiple AI pipelines.

Accelerates extensions in recommendation, fraud detection, and predictive analytics systems.

from functools import wraps

def feature_injector(func):
    @wraps(func)
    def wrapper(data):
        data["salary_per_age"] = data["salary"] / data["age"]
        data["is_high_income"] = data["salary"] > 100000
        return func(data)
    return wrapper

@feature_injector
def predict(data):
    print(data)

predict({
    "age": 25,
    "salary": 50000
})

4. Deterministic Seed Setter

Randomness can cause identical training scripts to produce different results. A deterministic seed setter synchronizes the random state across libraries (e.g., random and numpy) so experiments are reproducible.

Ensures consistent outcomes across multiple runs and environments.

Makes hyper‑parameter search and benchmark comparisons fair.

Simplifies debugging of nondeterministic neural‑network failures.

import random
import numpy as np
from functools import wraps

def deterministic_seed(seed=42):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            random.seed(seed)
            np.random.seed(seed)
            return func(*args, **kwargs)
        return wrapper
    return decorator

@deterministic_seed(seed=42)
def train_model():
    print(random.randint(1, 100))
    print(np.random.rand())

5. Dev‑Mode Fallback

During development, external AI APIs or cloud services may be unavailable. The dev‑mode fallback decorator catches exceptions, returns a safe mock response, and prevents the application from crashing.

Keeps the development workflow running when external services are unstable.

Allows front‑end teams to test against stable mocks even if the back‑end AI system is not ready.

Improves resilience for offline development, temporary outages, and experimental deployments.

from functools import wraps

DEV_MODE = True

def dev_fallback(mock_response):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if DEV_MODE:
                    print(f"Fallback activated: {e}")
                    return mock_response
                raise e
        return wrapper
    return decorator

@dev_fallback(mock_response="Mock AI response")
def call_llm():
    raise Exception("API unavailable")

These five decorators—Concurrency Limiter, Structured Logger, Feature Injector, Deterministic Seed Setter, and Dev‑Mode Fallback—are practical patterns that decouple infrastructure concerns from core ML logic, making LLM apps, recommendation systems, AI agents, and predictive pipelines more maintainable and robust.