Architectural Blueprints for Reusable, Scalable & Maintainable AI Systems
Reusable solutions to recurring software engineering challenges in data processing, model training, inference, and deployment pipelines.
Sequential processing of data using an arbitrary chain of transformations. Essential for data preprocessing and inference frameworks.
preprocess ➔ inference ➔ postprocessN must match input type of stage N+1.Function Sequence Pipeline:
from typing import Union, List
def preprocess(input: Union[str, Image]) -> Tensor:
return load_and_transform(input)
def inference(input: Tensor) -> Tensor:
return model(input)
def postprocess(input: Tensor) -> str:
return decode_logits(input)
# Pipeline Execution Loop
pipeline = [preprocess, inference, postprocess]
data = "sample_input.jpg"
for step in pipeline:
data = step(data)
return data
Keras-style Sequential Layer Pipeline:
class KerasModel:
def __init__(self):
self.layers = []
def add_layer(self, layer):
self.layers.append(layer)
def forward(self, input):
for layer in self.layers:
input = layer(input)
return input
# Usage
model = KerasModel()
model.add_layer(Linear(784, 128))
model.add_layer(ReLU())
output = model.forward(raw_tensor)
A generalization of pipelines into a Directed Acyclic Graph (DAG). Handles non-linear dependencies, branching, and parallel execution.
1. Graph Definition (Adjacency List):
# Representing DAG with Python dictionary
graph = {
'input': ['a'],
'a': ['b', 'e'],
'b': ['c', 'd'],
'd': ['e']
}
2. Workflow Execution Engine Concept:
class Step:
def __init__(self, inputs, outputs):
self.inputs = inputs
self.outputs = outputs
self.dependencies_met = False
self.resources = {"cpu": 2, "gpu": 1}
def execute(self):
if self.dependencies_met:
# Run model / transformation step
pass
class WorkflowEngine:
def __init__(self, dag):
self.dag = dag
def execute(self):
# Resolve topological order & run
pass
Originating from LISP (homoiconicity), code and data share the same representation. Functions can be treated as inspectable, serializable data objects.
(+ 1 2) ;; Executable Function call
'(+ 1 2) ;; Quoted String / Data structure3).
A model is a Callable Function (forward pass), but its parameters are Data Arrays (tensors).
model = myModel()
# 1. Used as a Function:
output = model(torch.randn(100))
# 2. Inspected as Data (Weights / Pickle):
weights = model.state_dict()
serialized_bytes = pickle.dumps(model)
Provides a clean, uniform interface to sequentially access elements or mini-batches from a data source without exposing its underlying storage logic.
for batch in dataset:
output = model(batch)
Implemented via Python's standard __iter__() and __next__() magic methods.
from typing import List
class CustomDataset:
def __init__(self, data: List[str]):
self.data = data
self.index = 0
def __iter__(self):
self.index = 0
return self
def __next__(self, batch_size: int = 1):
if self.index >= len(self.data):
raise StopIteration
batch = self.data[self.index : self.index + batch_size]
self.index += batch_size
return batch
Decouples client request reception from model inference execution using an asynchronous queue system. Essential for multi-model serving.
from dataclasses import dataclass
from typing import Union, Tuple, List
@dataclass
class Job:
model_name: str
input_data: Union[str, bytes]
endpoint: Tuple[str, int] # (host, port)
class JobProcessor:
def __init__(self):
self.jobs: List[Job] = []
def add_job(self, job: Job):
self.jobs.append(job)
def process_next(self):
if not self.jobs:
return None
job = self.jobs.pop(0) # FIFO Queue
return self.execute(job)
def execute(self, job: Job):
# Run inference using model registry & publish result to endpoint
result = run_model_inference(job.model_name, job.input_data)
return self.expose(result, job.endpoint)
Defines a one-to-many dependency between objects. When the subject changes state (e.g., end of epoch), all registered observers are automatically notified.
Extends training framework functionality (TensorBoard logging, LR schedulers, early stopping) without mutating core framework code.
from abc import ABC, abstractmethod
class Observer(ABC):
@abstractmethod
def update(self, state: dict):
pass
class ModelSubject:
def __init__(self):
self.observers: List[Observer] = []
self.state = {}
def attach(self, observer: Observer):
self.observers.append(observer)
def notify(self):
for observer in self.observers:
observer.update(self.state)
class ChangeLRObserver(Observer):
def update(self, state: dict):
if state.get("loss_increased"):
state["lr"] *= 0.1
print("Reduced learning rate")
class LogObserver(Observer):
def update(self, state: dict):
with open("log.txt", "a") as f:
f.write(f"Epoch {state['epoch']}: loss={state['loss']}\n")
Standardized high-level abstraction interface (model.fit(data)) popularized by Scikit-Learn.
class LearnerModel:
def __init__(self, model, loss_fn, optimizer):
self.model = model
self.loss_fn = loss_fn
self.optimizer = optimizer
def fit(self, dataset, epochs=5):
for epoch in range(epochs):
for X_batch, y_batch in dataset:
# 1. Forward Pass
preds = self.model(X_batch)
loss = self.loss_fn(preds, y_batch)
# 2. Gradient Update
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
⚠️ Avoid Python for loops in numerical code!
Sequential model invocation incurs severe GPU kernel launch overhead and low hardware utilization.
torch.stack() and invoke the model ONCE.
# ❌ BAD: Slow Python Loop (O(N) GPU calls)
for item in inputs:
output = model.forward(item)
# ✅ GOOD: Vectorized Batch (1 GPU call)
tensor_batch = torch.stack(inputs)
batch_output = model.forward(tensor_batch)
Dynamically adds responsibilities or behaviors to functions/classes without altering their internal code.
Common standard decorators: @memoize, @lru_cache, @profile, @app.get().
from line_profiler import LineProfiler
# Custom Profiling Decorator
def profile(func):
def inner(*args, **kwargs):
profiler = LineProfiler()
profiler.add_function(func)
profiler.enable_by_count()
return func(*args, **kwargs)
return inner
# Application Usage
@profile
def train_epoch_slow():
# Model compute steps...
pass
Defines a family of algorithms, encapsulates each one, and makes them interchangeable at runtime.
A Trainer class accepts an abstract Optimizer strategy interface. Users can swap between SGDOptimizer and AdamOptimizer seamlessly.
from abc import ABC, abstractmethod
class Optimizer(ABC):
@abstractmethod
def step(self, weights, grads) -> Tensor:
pass
class SGDOptimizer(Optimizer):
def step(self, weights, grads):
return weights - 0.01 * grads
class AdamOptimizer(Optimizer):
def step(self, weights, grads):
# Adam momentum calculation step
return updated_weights
class Trainer:
def __init__(self, optimizer: Optimizer):
self.optimizer = optimizer # Strategy injected
Dynamic construction of Directed Acyclic Computation Graphs (Tape-based Autograd) during forward execution for exact backpropagation.
Cache-oblivious matrix algorithms and tiled block matrix multiplications to maximize CPU L1/L2/L3 cache hit ratios.
Parallelism strategies for LLMs & massive models:
• Data Parallel (DDP)
• Pipeline Parallel
• Tensor Parallel
| Design Pattern | Core Concept | Primary ML Use Case |
|---|---|---|
| Pipeline | Sequential step execution | Preprocess ➔ Inference ➔ Postprocess |
| Workflow (DAG) | Directed Acyclic Graph | Airflow ETL, TorchServe Ensembles |
| Function as Data | Code/Data duality | Model Pickling & ONNX Serialization |
| Iterator | Streaming batch accessor | PyTorch DataLoader streaming |
| Job Queues | Async request worker queue | Multi-model web inference serving |
| Callbacks/Observer | Event-driven state hooks | TensorBoard logging, LR schedulers |
| Learner | Unified model.fit() encapsulation | Scikit-learn / Fast.ai interfaces |
| Batch/Vectorization | SIMD tensor stacking | Eliminating loops, GPU acceleration |
| Strategy | Interchangeable algorithms | Pluggable Optimizers & Losses |
Questions & Architectural Discussion