mirror of
https://github.com/MLSysBook/TinyTorch.git
synced 2026-07-24 14:04:34 -05:00
- Removed 30 debugging and development artifact files - Kept core system, documentation, and demo files - tests/milestones: 9 clean files (system + docs) - milestones/05_2017_transformer: 5 clean files (demos) - Clear, focused directory structure - Ready for students and developers
79 KiB
79 KiB
In [ ]:
#| default_exp core.tensor
#| export
import numpy as np
# Constants for memory calculations
BYTES_PER_FLOAT32 = 4 # Standard float32 size in bytes
KB_TO_BYTES = 1024 # Kilobytes to bytes conversion
MB_TO_BYTES = 1024 * 1024 # Megabytes to bytes conversionIn [ ]:
#| export
class Tensor:
"""Educational tensor that grows with student knowledge.
This class starts simple but includes dormant features for future modules:
- requires_grad: Will be used for automatic differentiation (Module 05)
- grad: Will store computed gradients (Module 05)
- backward(): Will compute gradients (Module 05)
For now, focus on: data, shape, and basic operations.
"""
def __init__(self, data, requires_grad=False):
"""Create a new tensor from data."""
### BEGIN SOLUTION
self.data = np.array(data, dtype=np.float32)
self.shape = self.data.shape
self.size = self.data.size
self.dtype = self.data.dtype
self.requires_grad = requires_grad
self.grad = None
### END SOLUTION
def __repr__(self):
"""String representation of tensor for debugging."""
grad_info = f", requires_grad={self.requires_grad}" if self.requires_grad else ""
return f"Tensor(data={self.data}, shape={self.shape}{grad_info})"
def __str__(self):
"""Human-readable string representation."""
return f"Tensor({self.data})"
def numpy(self):
"""Return the underlying NumPy array."""
return self.data
def __add__(self, other):
"""Add two tensors element-wise with broadcasting support."""
### BEGIN SOLUTION
if isinstance(other, Tensor):
return Tensor(self.data + other.data)
else:
return Tensor(self.data + other)
### END SOLUTION
def __sub__(self, other):
"""Subtract two tensors element-wise."""
### BEGIN SOLUTION
if isinstance(other, Tensor):
return Tensor(self.data - other.data)
else:
return Tensor(self.data - other)
### END SOLUTION
def __mul__(self, other):
"""Multiply two tensors element-wise (NOT matrix multiplication)."""
### BEGIN SOLUTION
if isinstance(other, Tensor):
return Tensor(self.data * other.data)
else:
return Tensor(self.data * other)
### END SOLUTION
def __truediv__(self, other):
"""Divide two tensors element-wise."""
### BEGIN SOLUTION
if isinstance(other, Tensor):
return Tensor(self.data / other.data)
else:
return Tensor(self.data / other)
### END SOLUTION
def matmul(self, other):
"""Matrix multiplication of two tensors."""
### BEGIN SOLUTION
if not isinstance(other, Tensor):
raise TypeError(f"Expected Tensor for matrix multiplication, got {type(other)}")
if self.shape == () or other.shape == ():
return Tensor(self.data * other.data)
if len(self.shape) == 0 or len(other.shape) == 0:
return Tensor(self.data * other.data)
if len(self.shape) >= 2 and len(other.shape) >= 2:
if self.shape[-1] != other.shape[-2]:
raise ValueError(
f"Cannot perform matrix multiplication: {self.shape} @ {other.shape}. "
f"Inner dimensions must match: {self.shape[-1]} ≠ {other.shape[-2]}"
)
result_data = np.matmul(self.data, other.data)
return Tensor(result_data)
### END SOLUTION
def __getitem__(self, key):
"""Enable indexing and slicing operations on Tensors."""
### BEGIN SOLUTION
result_data = self.data[key]
if not isinstance(result_data, np.ndarray):
result_data = np.array(result_data)
result = Tensor(result_data, requires_grad=self.requires_grad)
return result
### END SOLUTION
def reshape(self, *shape):
"""Reshape tensor to new dimensions."""
### BEGIN SOLUTION
if len(shape) == 1 and isinstance(shape[0], (tuple, list)):
new_shape = tuple(shape[0])
else:
new_shape = shape
if -1 in new_shape:
if new_shape.count(-1) > 1:
raise ValueError("Can only specify one unknown dimension with -1")
known_size = 1
unknown_idx = new_shape.index(-1)
for i, dim in enumerate(new_shape):
if i != unknown_idx:
known_size *= dim
unknown_dim = self.size // known_size
new_shape = list(new_shape)
new_shape[unknown_idx] = unknown_dim
new_shape = tuple(new_shape)
if np.prod(new_shape) != self.size:
raise ValueError(
f"Cannot reshape tensor of size {self.size} to shape {new_shape}"
)
reshaped_data = np.reshape(self.data, new_shape)
result = Tensor(reshaped_data, requires_grad=self.requires_grad)
return result
### END SOLUTION
def transpose(self, dim0=None, dim1=None):
"""Transpose tensor dimensions."""
### BEGIN SOLUTION
if dim0 is None and dim1 is None:
if len(self.shape) < 2:
return Tensor(self.data.copy())
else:
axes = list(range(len(self.shape)))
axes[-2], axes[-1] = axes[-1], axes[-2]
transposed_data = np.transpose(self.data, axes)
else:
if dim0 is None or dim1 is None:
raise ValueError("Both dim0 and dim1 must be specified")
axes = list(range(len(self.shape)))
axes[dim0], axes[dim1] = axes[dim1], axes[dim0]
transposed_data = np.transpose(self.data, axes)
result = Tensor(transposed_data, requires_grad=self.requires_grad)
return result
### END SOLUTION
def sum(self, axis=None, keepdims=False):
"""Sum tensor along specified axis."""
### BEGIN SOLUTION
result = np.sum(self.data, axis=axis, keepdims=keepdims)
return Tensor(result)
### END SOLUTION
def mean(self, axis=None, keepdims=False):
"""Compute mean of tensor along specified axis."""
### BEGIN SOLUTION
result = np.mean(self.data, axis=axis, keepdims=keepdims)
return Tensor(result)
### END SOLUTION
def max(self, axis=None, keepdims=False):
"""Find maximum values along specified axis."""
### BEGIN SOLUTION
result = np.max(self.data, axis=axis, keepdims=keepdims)
return Tensor(result)
### END SOLUTION
def backward(self):
"""Compute gradients (implemented in Module 05: Autograd)."""
### BEGIN SOLUTION
pass
### END SOLUTIONIn [ ]:
def test_unit_tensor_creation():
"""🧪 Test Tensor creation with various data types."""
print("🧪 Unit Test: Tensor Creation...")
# Test scalar creation
scalar = Tensor(5.0)
assert scalar.data == 5.0
assert scalar.shape == ()
assert scalar.size == 1
assert scalar.requires_grad == False
assert scalar.grad is None
assert scalar.dtype == np.float32
# Test vector creation
vector = Tensor([1, 2, 3])
assert np.array_equal(vector.data, np.array([1, 2, 3], dtype=np.float32))
assert vector.shape == (3,)
assert vector.size == 3
# Test matrix creation
matrix = Tensor([[1, 2], [3, 4]])
assert np.array_equal(matrix.data, np.array([[1, 2], [3, 4]], dtype=np.float32))
assert matrix.shape == (2, 2)
assert matrix.size == 4
# Test gradient flag (dormant feature)
grad_tensor = Tensor([1, 2], requires_grad=True)
assert grad_tensor.requires_grad == True
assert grad_tensor.grad is None # Still None until Module 05
print("✅ Tensor creation works correctly!")
if __name__ == "__main__":
test_unit_tensor_creation()In [ ]:
def test_unit_arithmetic_operations():
"""🧪 Test arithmetic operations with broadcasting."""
print("🧪 Unit Test: Arithmetic Operations...")
# Test tensor + tensor
a = Tensor([1, 2, 3])
b = Tensor([4, 5, 6])
result = a + b
assert np.array_equal(result.data, np.array([5, 7, 9], dtype=np.float32))
# Test tensor + scalar (very common in ML)
result = a + 10
assert np.array_equal(result.data, np.array([11, 12, 13], dtype=np.float32))
# Test broadcasting with different shapes (matrix + vector)
matrix = Tensor([[1, 2], [3, 4]])
vector = Tensor([10, 20])
result = matrix + vector
expected = np.array([[11, 22], [13, 24]], dtype=np.float32)
assert np.array_equal(result.data, expected)
# Test subtraction (data centering)
result = b - a
assert np.array_equal(result.data, np.array([3, 3, 3], dtype=np.float32))
# Test multiplication (scaling)
result = a * 2
assert np.array_equal(result.data, np.array([2, 4, 6], dtype=np.float32))
# Test division (normalization)
result = b / 2
assert np.array_equal(result.data, np.array([2.0, 2.5, 3.0], dtype=np.float32))
# Test chaining operations (common in ML pipelines)
normalized = (a - 2) / 2 # Center and scale
expected = np.array([-0.5, 0.0, 0.5], dtype=np.float32)
assert np.allclose(normalized.data, expected)
print("✅ Arithmetic operations work correctly!")
if __name__ == "__main__":
test_unit_arithmetic_operations()In [ ]:
def test_unit_matrix_multiplication():
"""🧪 Test matrix multiplication operations."""
print("🧪 Unit Test: Matrix Multiplication...")
# Test 2×2 matrix multiplication (basic case)
a = Tensor([[1, 2], [3, 4]]) # 2×2
b = Tensor([[5, 6], [7, 8]]) # 2×2
result = a.matmul(b)
# Expected: [[1×5+2×7, 1×6+2×8], [3×5+4×7, 3×6+4×8]] = [[19, 22], [43, 50]]
expected = np.array([[19, 22], [43, 50]], dtype=np.float32)
assert np.array_equal(result.data, expected)
# Test rectangular matrices (common in neural networks)
c = Tensor([[1, 2, 3], [4, 5, 6]]) # 2×3 (like batch_size=2, features=3)
d = Tensor([[7, 8], [9, 10], [11, 12]]) # 3×2 (like features=3, outputs=2)
result = c.matmul(d)
# Expected: [[1×7+2×9+3×11, 1×8+2×10+3×12], [4×7+5×9+6×11, 4×8+5×10+6×12]]
expected = np.array([[58, 64], [139, 154]], dtype=np.float32)
assert np.array_equal(result.data, expected)
# Test matrix-vector multiplication (common in forward pass)
matrix = Tensor([[1, 2, 3], [4, 5, 6]]) # 2×3
vector = Tensor([1, 2, 3]) # 3×1 (conceptually)
result = matrix.matmul(vector)
# Expected: [1×1+2×2+3×3, 4×1+5×2+6×3] = [14, 32]
expected = np.array([14, 32], dtype=np.float32)
assert np.array_equal(result.data, expected)
# Test shape validation - should raise clear error
try:
incompatible_a = Tensor([[1, 2]]) # 1×2
incompatible_b = Tensor([[1], [2], [3]]) # 3×1
incompatible_a.matmul(incompatible_b) # 1×2 @ 3×1 should fail (2 ≠ 3)
assert False, "Should have raised ValueError for incompatible shapes"
except ValueError as e:
assert "Inner dimensions must match" in str(e)
assert "2 ≠ 3" in str(e) # Should show specific dimensions
print("✅ Matrix multiplication works correctly!")
if __name__ == "__main__":
test_unit_matrix_multiplication()In [ ]:
def test_unit_shape_manipulation():
"""🧪 Test reshape and transpose operations."""
print("🧪 Unit Test: Shape Manipulation...")
# Test basic reshape (flatten → matrix)
tensor = Tensor([1, 2, 3, 4, 5, 6]) # Shape: (6,)
reshaped = tensor.reshape(2, 3) # Shape: (2, 3)
assert reshaped.shape == (2, 3)
expected = np.array([[1, 2, 3], [4, 5, 6]], dtype=np.float32)
assert np.array_equal(reshaped.data, expected)
# Test reshape with tuple (alternative calling style)
reshaped2 = tensor.reshape((3, 2)) # Shape: (3, 2)
assert reshaped2.shape == (3, 2)
expected2 = np.array([[1, 2], [3, 4], [5, 6]], dtype=np.float32)
assert np.array_equal(reshaped2.data, expected2)
# Test reshape with -1 (automatic dimension inference)
auto_reshaped = tensor.reshape(2, -1) # Should infer -1 as 3
assert auto_reshaped.shape == (2, 3)
# Test reshape validation - should raise error for incompatible sizes
try:
tensor.reshape(2, 2) # 6 elements can't fit in 2×2=4
assert False, "Should have raised ValueError"
except ValueError as e:
assert "Total elements must match" in str(e)
assert "6 ≠ 4" in str(e)
# Test matrix transpose (most common case)
matrix = Tensor([[1, 2, 3], [4, 5, 6]]) # (2, 3)
transposed = matrix.transpose() # (3, 2)
assert transposed.shape == (3, 2)
expected = np.array([[1, 4], [2, 5], [3, 6]], dtype=np.float32)
assert np.array_equal(transposed.data, expected)
# Test 1D transpose (should be identity)
vector = Tensor([1, 2, 3])
vector_t = vector.transpose()
assert np.array_equal(vector.data, vector_t.data)
# Test specific dimension transpose
tensor_3d = Tensor([[[1, 2], [3, 4]], [[5, 6], [7, 8]]]) # (2, 2, 2)
swapped = tensor_3d.transpose(0, 2) # Swap first and last dimensions
assert swapped.shape == (2, 2, 2) # Same shape but data rearranged
# Test neural network reshape pattern (flatten for MLP)
batch_images = Tensor(np.random.rand(2, 3, 4)) # (batch=2, height=3, width=4)
flattened = batch_images.reshape(2, -1) # (batch=2, features=12)
assert flattened.shape == (2, 12)
print("✅ Shape manipulation works correctly!")
if __name__ == "__main__":
test_unit_shape_manipulation()In [ ]:
def test_unit_reduction_operations():
"""🧪 Test reduction operations."""
print("🧪 Unit Test: Reduction Operations...")
matrix = Tensor([[1, 2, 3], [4, 5, 6]]) # Shape: (2, 3)
# Test sum all elements (common for loss computation)
total = matrix.sum()
assert total.data == 21.0 # 1+2+3+4+5+6
assert total.shape == () # Scalar result
# Test sum along axis 0 (columns) - batch dimension reduction
col_sum = matrix.sum(axis=0)
expected_col = np.array([5, 7, 9], dtype=np.float32) # [1+4, 2+5, 3+6]
assert np.array_equal(col_sum.data, expected_col)
assert col_sum.shape == (3,)
# Test sum along axis 1 (rows) - feature dimension reduction
row_sum = matrix.sum(axis=1)
expected_row = np.array([6, 15], dtype=np.float32) # [1+2+3, 4+5+6]
assert np.array_equal(row_sum.data, expected_row)
assert row_sum.shape == (2,)
# Test mean (average loss computation)
avg = matrix.mean()
assert np.isclose(avg.data, 3.5) # 21/6
assert avg.shape == ()
# Test mean along axis (batch normalization pattern)
col_mean = matrix.mean(axis=0)
expected_mean = np.array([2.5, 3.5, 4.5], dtype=np.float32) # [5/2, 7/2, 9/2]
assert np.allclose(col_mean.data, expected_mean)
# Test max (finding best predictions)
maximum = matrix.max()
assert maximum.data == 6.0
assert maximum.shape == ()
# Test max along axis (argmax-like operation)
row_max = matrix.max(axis=1)
expected_max = np.array([3, 6], dtype=np.float32) # [max(1,2,3), max(4,5,6)]
assert np.array_equal(row_max.data, expected_max)
# Test keepdims (important for broadcasting)
sum_keepdims = matrix.sum(axis=1, keepdims=True)
assert sum_keepdims.shape == (2, 1) # Maintains 2D shape
expected_keepdims = np.array([[6], [15]], dtype=np.float32)
assert np.array_equal(sum_keepdims.data, expected_keepdims)
# Test 3D reduction (simulating global average pooling)
tensor_3d = Tensor([[[1, 2], [3, 4]], [[5, 6], [7, 8]]]) # (2, 2, 2)
spatial_mean = tensor_3d.mean(axis=(1, 2)) # Average across spatial dimensions
assert spatial_mean.shape == (2,) # One value per batch item
print("✅ Reduction operations work correctly!")
if __name__ == "__main__":
test_unit_reduction_operations()In [ ]:
def analyze_memory_layout():
"""📊 Demonstrate cache effects with row vs column access patterns."""
print("📊 Analyzing Memory Access Patterns...")
print("=" * 60)
# Create a moderately-sized matrix (large enough to show cache effects)
size = 2000
matrix = Tensor(np.random.rand(size, size))
import time
print(f"\nTesting with {size}×{size} matrix ({matrix.size * BYTES_PER_FLOAT32 / MB_TO_BYTES:.1f} MB)")
print("-" * 60)
# Test 1: Row-wise access (cache-friendly)
# Memory layout: [row0][row1][row2]... stored contiguously
print("\n🔬 Test 1: Row-wise Access (Cache-Friendly)")
start = time.time()
row_sums = []
for i in range(size):
row_sum = matrix.data[i, :].sum() # Access entire row sequentially
row_sums.append(row_sum)
row_time = time.time() - start
print(f" Time: {row_time*1000:.1f}ms")
print(f" Access pattern: Sequential (follows memory layout)")
# Test 2: Column-wise access (cache-unfriendly)
# Must jump between rows, poor spatial locality
print("\n🔬 Test 2: Column-wise Access (Cache-Unfriendly)")
start = time.time()
col_sums = []
for j in range(size):
col_sum = matrix.data[:, j].sum() # Access entire column with large strides
col_sums.append(col_sum)
col_time = time.time() - start
print(f" Time: {col_time*1000:.1f}ms")
print(f" Access pattern: Strided (jumps {size * BYTES_PER_FLOAT32} bytes per element)")
# Calculate slowdown
slowdown = col_time / row_time
print("\n" + "=" * 60)
print(f"📊 PERFORMANCE IMPACT:")
print(f" Slowdown factor: {slowdown:.2f}× ({col_time/row_time:.1f}× slower)")
print(f" Cache misses cause {(slowdown-1)*100:.0f}% performance loss")
# Educational insights
print("\n💡 KEY INSIGHTS:")
print(f" 1. Memory layout matters: Row-major (C-style) storage is sequential")
print(f" 2. Cache lines are ~64 bytes: Row access loads nearby elements \"for free\"")
print(f" 3. Column access misses cache: Must reload from DRAM every time")
print(f" 4. This is O(n) algorithm but {slowdown:.1f}× different wall-clock time!")
print("\n🚀 REAL-WORLD IMPLICATIONS:")
print(f" • CNNs use NCHW format (channels sequential) for cache efficiency")
print(f" • Matrix multiplication optimized with blocking (tile into cache-sized chunks)")
print(f" • Transpose is expensive ({slowdown:.1f}×) because it changes memory layout")
print(f" • This is why GPU frameworks obsess over memory coalescing")
print("\n" + "=" * 60)
# Run the systems analysis
if __name__ == "__main__":
analyze_memory_layout()In [ ]:
def test_module():
"""
Comprehensive test of entire module functionality.
This final test runs before module summary to ensure:
- All unit tests pass
- Functions work together correctly
- Module is ready for integration with TinyTorch
"""
print("🧪 RUNNING MODULE INTEGRATION TEST")
print("=" * 50)
# Run all unit tests
print("Running unit tests...")
test_unit_tensor_creation()
test_unit_arithmetic_operations()
test_unit_matrix_multiplication()
test_unit_shape_manipulation()
test_unit_reduction_operations()
print("\nRunning integration scenarios...")
# Test realistic neural network computation
print("🧪 Integration Test: Two-Layer Neural Network...")
# Create input data (2 samples, 3 features)
x = Tensor([[1, 2, 3], [4, 5, 6]])
# First layer: 3 inputs → 4 hidden units
W1 = Tensor([[0.1, 0.2, 0.3, 0.4],
[0.5, 0.6, 0.7, 0.8],
[0.9, 1.0, 1.1, 1.2]])
b1 = Tensor([0.1, 0.2, 0.3, 0.4])
# Forward pass: hidden = xW1 + b1
hidden = x.matmul(W1) + b1
assert hidden.shape == (2, 4), f"Expected (2, 4), got {hidden.shape}"
# Second layer: 4 hidden → 2 outputs
W2 = Tensor([[0.1, 0.2], [0.3, 0.4], [0.5, 0.6], [0.7, 0.8]])
b2 = Tensor([0.1, 0.2])
# Output layer: output = hiddenW2 + b2
output = hidden.matmul(W2) + b2
assert output.shape == (2, 2), f"Expected (2, 2), got {output.shape}"
# Verify data flows correctly (no NaN, reasonable values)
assert not np.isnan(output.data).any(), "Output contains NaN values"
assert np.isfinite(output.data).all(), "Output contains infinite values"
print("✅ Two-layer neural network computation works!")
# Test gradient attributes are preserved and functional
print("🧪 Integration Test: Gradient System Readiness...")
grad_tensor = Tensor([1, 2, 3], requires_grad=True)
result = grad_tensor + 5
assert grad_tensor.requires_grad == True, "requires_grad not preserved"
assert grad_tensor.grad is None, "grad should still be None"
# Test backward() doesn't crash (even though it does nothing)
grad_tensor.backward() # Should not raise any exception
print("✅ Gradient system ready for Module 05!")
# Test complex shape manipulations
print("🧪 Integration Test: Complex Shape Operations...")
data = Tensor([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12])
# Reshape to 3D tensor (simulating batch processing)
tensor_3d = data.reshape(2, 2, 3) # (batch=2, height=2, width=3)
assert tensor_3d.shape == (2, 2, 3)
# Global average pooling simulation
pooled = tensor_3d.mean(axis=(1, 2)) # Average across spatial dimensions
assert pooled.shape == (2,), f"Expected (2,), got {pooled.shape}"
# Flatten for MLP
flattened = tensor_3d.reshape(2, -1) # (batch, features)
assert flattened.shape == (2, 6)
# Transpose for different operations
transposed = tensor_3d.transpose() # Should transpose last two dims
assert transposed.shape == (2, 3, 2)
print("✅ Complex shape operations work!")
# Test broadcasting edge cases
print("🧪 Integration Test: Broadcasting Edge Cases...")
# Scalar broadcasting
scalar = Tensor(5.0)
vector = Tensor([1, 2, 3])
result = scalar + vector # Should broadcast scalar to vector shape
expected = np.array([6, 7, 8], dtype=np.float32)
assert np.array_equal(result.data, expected)
# Matrix + vector broadcasting
matrix = Tensor([[1, 2], [3, 4]])
vec = Tensor([10, 20])
result = matrix + vec
expected = np.array([[11, 22], [13, 24]], dtype=np.float32)
assert np.array_equal(result.data, expected)
print("✅ Broadcasting edge cases work!")
print("\n" + "=" * 50)
print("🎉 ALL TESTS PASSED! Module ready for export.")
print("Run: tito module complete 01_tensor")
# Run comprehensive module test
if __name__ == "__main__":
test_module()