Core API¶

The core package contains the tensor object and operation base class used by every differentiable computation.

Variable¶

Variable wraps a NumPy array, stores an optional accumulated gradient, and points to the Function that produced it.

import numpy as np
from MiniTorch import Variable

x = Variable(np.array([1.0, 2.0], dtype=np.float32), name="x")
loss = (x ** 2).sum()
loss.backward()

Note

MiniTorch gradients are themselves Variable objects. Read the raw array through x.grad.data.

Variable ¶

Variable(data: Optional[ndarray], name: Optional[str] = None)

Source code in MiniTorch/core/variable.py

def __init__(self, data: Optional[np.ndarray], name: Optional[str] = None) -> None:
    if data is not None:
        if not isinstance(data, np.ndarray):
            raise TypeError(f"{type(data)} is not supported")

    self.data: Optional[np.ndarray] = data
    self.name: Optional[str] = name
    self.grad: Optional[Variable] = None
    self.creator: Optional[Function] = None
    self.generation: int = 0

T `property` ¶

T: Variable

backward ¶

backward(retain_grad: bool = False, create_graph: bool = False) -> None

Source code in MiniTorch/core/variable.py

def backward(self, retain_grad: bool = False, create_graph: bool = False) -> None:
    if not Config.enable_backprob:
        raise RuntimeError("backward() is not allowed in no_grad mode")

    if self.grad is None:
        self.grad = Variable(np.ones_like(self.data))

    if self.creator is None:
        return

    # Build a compact topological tape in O(functions + edges). Forward
    # execution already defines a DAG, so backward can simply replay this
    # tape in reverse instead of repeatedly sorting a ready list.
    tape: list[Function] = []
    discovered: set[Function] = set()
    stack: list[tuple[Function, bool]] = [(self.creator, False)]

    while stack:
        func, expanded = stack.pop()
        if expanded:
            tape.append(func)
            continue
        if func in discovered:
            continue
        discovered.add(func)
        stack.append((func, True))
        if func.inputs is not None:
            for input_var in func.inputs:
                if input_var.creator is not None and input_var.creator not in discovered:
                    stack.append((input_var.creator, False))

    for func in reversed(tape):
        if func.outputs is None or func.inputs is None:
            continue

        outputs = [output_ref() for output_ref in func.outputs]
        if create_graph:
            output_grads = [
                output.grad if output is not None else None
                for output in outputs
            ]
            with using_config("enable_backprob", True):
                input_grads = func.backward(*output_grads)
            if not isinstance(input_grads, tuple):
                input_grads = (input_grads,)

            for input_var, grad in zip(func.inputs, input_grads):
                if grad is None:
                    continue
                input_var.grad = grad if input_var.grad is None else input_var.grad + grad
        else:
            output_grad_arrays = [
                output.grad.data
                if output is not None and output.grad is not None
                else None
                for output in outputs
            ]
            array_backward = getattr(func, "backward_array", None)
            try:
                if array_backward is None:
                    raise NotImplementedError
                input_grad_arrays = array_backward(*output_grad_arrays)
            except NotImplementedError:
                # Compatibility path for third-party operations that have
                # not implemented the raw-array training API yet.
                output_grads = [
                    output.grad if output is not None else None
                    for output in outputs
                ]
                with using_config("enable_backprob", False):
                    legacy_grads = func.backward(*output_grads)
                if not isinstance(legacy_grads, tuple):
                    legacy_grads = (legacy_grads,)
                input_grad_arrays = tuple(
                    grad.data if isinstance(grad, Variable) else grad
                    for grad in legacy_grads
                )

            if not isinstance(input_grad_arrays, tuple):
                input_grad_arrays = (input_grad_arrays,)

            for input_var, grad_array in zip(func.inputs, input_grad_arrays):
                if grad_array is None:
                    continue
                grad_array = np.asarray(grad_array)
                if input_var.grad is None:
                    input_var.grad = Variable(grad_array)
                else:
                    # Allocate the accumulation result instead of mutating
                    # in place: some derivative rules intentionally return
                    # aliased upstream gradients.
                    input_var.grad.data = np.asarray(
                        input_var.grad.data + grad_array
                    )

        if not retain_grad and not create_graph:
            for output in outputs:
                if output is not None and output is not self:
                    output.grad = None

clear_grad ¶

clear_grad() -> None

Source code in MiniTorch/core/variable.py

def clear_grad(self) -> None:
    self.grad = None

reshape ¶

reshape(*shape: Any) -> Variable

Source code in MiniTorch/core/variable.py

def reshape(self, *shape: Any) -> Variable:
    resolved: Any = shape[0] if (len(shape) == 1 and isinstance(shape[0], (tuple, list))) else shape
    from MiniTorch.ops.reshape import reshape
    return reshape(self, resolved)

Function¶

Subclass Function to implement a differentiable operation. The ordinary first-order path can provide backward_array to propagate raw NumPy arrays without building another autograd graph.