Meta AI Engineer Interview Preparation Guide - Junior Level

import torch
import torch.nn as nn
import torch.optim as optim
from contextlib import ExitStack
import logging
import gc
import time

logger = logging.getLogger(__name__)

def train_one_epoch(model, dataloader, optimizer, scaler, device, epoch, checkpoint_path=None):
    model.train()
    try:
        for batch in dataloader:
            inputs, targets = [t.to(device, non_blocking=True) for t in batch]

            # Mixed precision context manager (optional)
            with torch.cuda.amp.autocast(enabled=scaler is not None):
                outputs = model(inputs)
                loss = nn.functional.cross_entropy(outputs, targets)

            optimizer.zero_grad()
            if scaler is not None:
                scaler.scale(loss).backward()
                scaler.step(optimizer)
                scaler.update()
            else:
                loss.backward()
                optimizer.step()

    except KeyboardInterrupt:
        logger.warning("Training interrupted by user. Saving checkpoint...")
        if checkpoint_path:
            _save_checkpoint(model, optimizer, scaler, epoch, checkpoint_path)
        raise  # re-raise so caller can decide to stop

    except RuntimeError as e:
        # Common GPU OOM or other runtime errors
        logger.exception("RuntimeError during training: %s", e)
        if "out of memory" in str(e).lower():
            logger.error("CUDA OOM detected. Attempting deterministic cleanup and saving partial checkpoint.")
        if checkpoint_path:
            _save_checkpoint(model, optimizer, scaler, epoch, checkpoint_path + ".error")
        # Clean up GPU state deterministically
        _cleanup_gpu()
        raise

    except Exception as e:
        # Catch remaining unexpected errors, log full traceback, save state
        logger.exception("Unexpected error during training: %s", e)
        if checkpoint_path:
            _save_checkpoint(model, optimizer, scaler, epoch, checkpoint_path + ".error")
        _cleanup_gpu()
        raise

    finally:
        # Always ensure any temporary buffers are cleared
        _cleanup_step_tensors()
        torch.cuda.synchronize() if device.type == "cuda" else None
        torch.cuda.empty_cache() if device.type == "cuda" else None

def _save_checkpoint(model, optimizer, scaler, epoch, path):
    # Move optimizer state to CPU to minimize GPU memory before saving
    opt_state = {k: v.cpu() if isinstance(v, torch.Tensor) and v.is_cuda else v
                 for k, v in optimizer.state_dict().items()}
    ckpt = {
        "epoch": epoch,
        "model": model.state_dict(),
        "optimizer": optimizer.state_dict(),  # optimizer.state_dict() is now safe
        "scaler": scaler.state_dict() if scaler is not None else None,
        "timestamp": time.time(),
    }
    torch.save(ckpt, path)
    logger.info("Checkpoint saved to %s", path)

def _cleanup_gpu():
    # Delete references, run GC, synchronize CUDA and empty cache
    try:
        gc.collect()
        torch.cuda.synchronize()
        torch.cuda.empty_cache()
    except Exception:
        # Avoid raising in cleanup
        logger.exception("Error during GPU cleanup")

def _cleanup_step_tensors():
    # If you keep per-step globals, delete them here. Example placeholder.
    pass

class TrieNode:
    __slots__ = ("children", "is_word", "prefix_count")
    def __init__(self, use_array=False):
        # choose dict (default) or fixed array of length 26 for children
        self.children = {} if not use_array else [None]*26
        self.is_word = False
        self.prefix_count = 0

class Trie:
    def __init__(self, use_array=False):
        self.root = TrieNode(use_array=use_array)
        self.use_array = use_array

    def _idx(self, ch):
        return ord(ch) - 97

    def insert(self, word: str):
        node = self.root
        node.prefix_count += 1
        for ch in word:
            if self.use_array:
                i = self._idx(ch)
                if node.children[i] is None:
                    node.children[i] = TrieNode(use_array=True)
                node = node.children[i]
            else:
                if ch not in node.children:
                    node.children[ch] = TrieNode(use_array=False)
                node = node.children[ch]
            node.prefix_count += 1
        node.is_word = True

    def search(self, word: str) -> bool:
        node = self.root
        for ch in word:
            node = node.children[self._idx(ch)] if self.use_array else node.children.get(ch)
            if not node:
                return False
        return node.is_word

    def starts_with(self, prefix: str) -> bool:
        node = self.root
        for ch in prefix:
            node = node.children[self._idx(ch)] if self.use_array else node.children.get(ch)
            if not node:
                return False
        return True

    def count_prefix(self, prefix: str) -> int:
        node = self.root
        for ch in prefix:
            node = node.children[self._idx(ch)] if self.use_array else node.children.get(ch)
            if not node:
                return 0
        return node.prefix_count

def longest_palindromic_subsequence(s):
    n = len(s)
    if n == 0:
        return "", 0, 0  # lps str, length, min insertions

    dp = [[0]*n for _ in range(n)]
    # base: single chars
    for i in range(n):
        dp[i][i] = 1

    # fill for substrings of increasing length
    for length in range(2, n+1):
        for i in range(0, n-length+1):
            j = i + length - 1
            if s[i] == s[j]:
                dp[i][j] = 2 + (dp[i+1][j-1] if i+1 <= j-1 else 0)
            else:
                dp[i][j] = max(dp[i+1][j], dp[i][j-1])

    # reconstruct one LPS
    i, j = 0, n-1
    left = []
    right = []
    while i <= j:
        if i == j:
            # center character for odd-length LPS
            left.append(s[i])
            break
        if s[i] == s[j]:
            left.append(s[i])
            right.append(s[j])
            i += 1
            j -= 1
        else:
            # move toward the direction of larger dp
            if dp[i+1][j] >= dp[i][j-1]:
                i += 1
            else:
                j -= 1

    lps = "".join(left) + "".join(reversed(right))
    lps_len = dp[0][n-1]
    min_insertions = n - lps_len
    return lps, lps_len, min_insertions

# Example
s = "abcda"
print(longest_palindromic_subsequence(s))
# Output: ('aba' or 'aca' etc., 3, 2)

from abc import ABC, abstractmethod
import json
from typing import Dict, Any

class MetricLogger(ABC):
    @abstractmethod
    def log(self, step: int, metrics: Dict[str, float]) -> None:
        """Record metrics for a given step."""
        pass

    @abstractmethod
    def flush(self) -> None:
        """Ensure all buffered metrics are persisted."""
        pass

# No-op implementation (useful in unit tests)
class NoOpLogger(MetricLogger):
    def log(self, step, metrics):
        return None
    def flush(self):
        return None

# Simple file-writer (easy to inspect in CI)
class FileLogger(MetricLogger):
    def __init__(self, path):
        self.path = path
        self._buf = []
    def log(self, step, metrics):
        entry = {"step": step, "metrics": metrics}
        self._buf.append(entry)
        # simple immediate write for determinism
        with open(self.path, "a") as f:
            f.write(json.dumps(entry) + "\n")
    def flush(self):
        self._buf.clear()

# Example commercial service adapter (stub)
class CommercialLogger(MetricLogger):
    def __init__(self, client):
        self.client = client
    def log(self, step, metrics):
        # translate to service API
        self.client.send({"timestamp_step": step, "payload": metrics})
    def flush(self):
        self.client.flush()

# Atomic<T> supports load(acquire), store(release), compare_and_swap(expected, new) (CAS)

def make_node(val=None):
    return Node(val, Atomic(None))

# Initialization: single dummy node
head = Atomic(make_node())   # points to Node
tail = Atomic(head.load())   # initially same as head

def enqueue(val):
    node = make_node(val)
    while True:
        t = tail.load()              # acquire
        n = t.next.load()            # acquire
        if t is tail.load():         # still consistent
            if n is None:
                # try to link new node at tail.next
                if t.next.compare_and_swap(None, node):
                    # advance tail (helping)
                    tail.compare_and_swap(t, node)
                    return
            else:
                # tail is lagging; help advance
                tail.compare_and_swap(t, n)

def dequeue():
    while True:
        h = head.load()
        t = tail.load()
        n = h.next.load()
        if h is head.load():
            if h is t:
                if n is None:
                    return EMPTY
                # tail lagging; help advance
                tail.compare_and_swap(t, n)
            else:
                val = n.value
                # try to swing head to next (linearization point)
                if head.compare_and_swap(h, n):
                    # safe to reclaim h (see reclamation techniques)
                    return val

import pandas as pd

def impute_purchase_amount(df: pd.DataFrame, user_col='user_id', value_col='purchase_amount', min_obs=3):
    """
    Impute missing purchase_amount per user median, fallback to global median if user has <min_obs.
    Operates non-destructively (returns a copy).
    """
    df = df.copy()
    # global median (ignores NaNs)
    global_median = df[value_col].median()

    # per-user: count non-null and median
    stats = (df.groupby(user_col)[value_col]
               .agg(count_nonnull=lambda x: x.notna().sum(),
                    median_value='median')
               .reset_index())

    # Users eligible for per-user median
    eligible = stats.loc[stats['count_nonnull'] >= min_obs, [user_col, 'median_value']]
    user_median_map = pd.Series(eligible['median_value'].values, index=eligible[user_col]).to_dict()

    # Map per-user median to rows; missing mapping -> fallback to global_median
    df['_user_median'] = df[user_col].map(user_median_map)
    df[value_col] = df[value_col].fillna(df['_user_median']).fillna(global_median)
    df = df.drop(columns=['_user_median'])
    return df