Unit 7

# Imperative: gradient descent step
def gradient_descent(X, y, lr=0.01, epochs=1000):
    m, n = X.shape
    theta = np.zeros(n)          # state

    for epoch in range(epochs):  # explicit loop
        predictions = X @ theta
        error = predictions - y
        gradient = X.T @ error / m
        theta = theta - lr * gradient   # state mutation

    return theta

# OOP: custom neural network layer (PyTorch)
import torch.nn as nn

class AttentionLayer(nn.Module):
    def __init__(self, embed_dim, heads):
        super().__init__()
        self.attn = nn.MultiheadAttention(
            embed_dim, heads
        )
        self.norm = nn.LayerNorm(embed_dim)

    def forward(self, x):
        attn_out, _ = self.attn(x, x, x)
        return self.norm(x + attn_out)  # residual

# Polymorphism – plug in any nn.Module
model = nn.Sequential(
    AttentionLayer(512, 8),
    nn.Linear(512, 256),
    nn.ReLU()
)

# Functional: data pipeline with map/filter
from functools import reduce

raw_data = [3.2, None, -1.0, 8.7, None, 5.5]

pipeline = (
    lambda data:
        list(map(
            lambda x: (x - 2.8) / 3.4,        # normalise
            filter(lambda x: x is not None,    # clean
                   filter(lambda x: x >= 0,   # remove neg
                          filter(None.__ne__, data)))
        ))
)

clean = pipeline(raw_data)  # [0.118, 1.735, 0.794]

# JAX: functional auto-diff
import jax
grad_loss = jax.grad(lambda params: loss_fn(params, X, y))

% Prolog — Medical Expert System
symptom(patient1, fever).
symptom(patient1, cough).
symptom(patient1, fatigue).

disease(flu) :- symptom(P, fever),
               symptom(P, cough),
               symptom(P, fatigue).

treatment(flu, rest).
treatment(flu, antiviral).

% Query: what disease does patient1 have?
% ?- disease(D), treatment(D, T).
% D = flu, T = rest ;
% D = flu, T = antiviral.

# Event-driven: real-time fraud detection
import asyncio

class FraudDetector:
    async def on_transaction(self, event):
        score = await self.model.predict(event)
        if score > 0.85:
            await self.alert(event['id'], score)

    async def alert(self, tx_id, score):
        print(f"🚨 FRAUD: tx={tx_id} score={score:.2f}")

# Event loop — handles thousands concurrently
async def main():
    detector = FraudDetector()
    async for event in kafka_stream("transactions"):
        asyncio.create_task(
            detector.on_transaction(event)
        )

asyncio.run(main())

# Python (OOP) calls Prolog (Logic)
from pyswip import Prolog

class Patient:
    def __init__(self, name, symptoms):
        self.name = name
        self.symptoms = symptoms   # list of strings

class DiagnosisEngine:
    def __init__(self):
        self.prolog = Prolog()
        self.prolog.consult("diagnosis.pl")   # load rules

    def diagnose(self, patient: Patient):
        # Assert patient facts into Prolog
        for s in patient.symptoms:
            self.prolog.assertz(f"symptom({patient.name},{s})")

        # Query Prolog logic engine
        results = list(self.prolog.query(
            f"disease({patient.name}, D), treatment(D,T)"
        ))
        return results

# Usage
p = Patient("alice", ["fever","cough","fatigue"])
engine = DiagnosisEngine()
print(engine.diagnose(p))
# → [{'D': 'flu', 'T': 'rest'}, ...]

# 1. SQL (Declarative) — fetch data
query = """
    SELECT user_id, product_id, rating, timestamp
    FROM purchases
    WHERE timestamp > '2024-01-01'
"""
df = pd.read_sql(query, engine)

# 2. Pandas (Functional transforms) — feature engineering
user_features = (df
    .groupby('user_id')
    .agg(avg_rating=('rating','mean'),
         purchase_count=('product_id','count'))
    .reset_index()
)

# 3. TensorFlow (Dataflow) — collaborative filtering
model = tf.keras.Sequential([
    tf.keras.layers.Embedding(n_users, 32),
    tf.keras.layers.Flatten(),
    tf.keras.layers.Dense(64, activation='relu'),
    tf.keras.layers.Dense(n_products, activation='sigmoid')
])
model.compile(optimizer='adam', loss='binary_crossentropy')
model.fit(X_train, y_train, epochs=10)

# 4. FastAPI (Event-Driven) — serve predictions
@app.get("/recommend/{user_id}")
async def recommend(user_id: int):
    return model.predict([[user_id]]).tolist()

# ROS2 Python node — object detection subscriber
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from vision_msgs.msg import Detection2DArray
import torch

class DetectionNode(Node):             # OOP: extends Node
    def __init__(self):
        super().__init__('detector')
        # Event-Driven: subscribe to camera topic
        self.sub = self.create_subscription(
            Image, '/camera/image_raw',
            self.on_image, 10           # callback
        )
        self.pub = self.create_publisher(
            Detection2DArray, '/detections', 10
        )
        self.model = torch.hub.load(   # Dataflow: DL model
            'ultralytics/yolov5', 'yolov5s'
        )

    def on_image(self, msg):           # event handler
        results = self.model(msg.data)
        detections = self.parse(results)
        self.pub.publish(detections)   # publish to ROS topic

rclpy.init()
rclpy.spin(DetectionNode())

Benefits

• Each component uses the most suitable tool/paradigm
• Encourages modular, loosely-coupled architecture
• Independent scaling of subsystems
• Easier to replace/upgrade individual components
• Access to best-of-breed open-source libraries
• Parallel development by specialised teams

Challenges

• Increased operational complexity (DevOps overhead)
• Data serialisation overhead at language boundaries
• Debugging across languages and paradigms is harder
• Inter-service latency in distributed pipelines
• Consistency: different type systems and null handling
• Team skill requirements span multiple languages

• Handling missing values: mean/median imputation, interpolation, or removal
• Outlier detection: IQR, Z-score, isolation forests
• Normalisation / scaling: Min-Max, Z-score, RobustScaler
• Encoding categoricals: One-hot, label, target encoding
• Text cleaning: tokenisation, stop-word removal, stemming, lemmatisation
• Image preprocessing: resize, crop, histogram equalisation

import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.impute import SimpleImputer

# Load raw data
df = pd.read_csv('data.csv')

# 1. Missing value imputation
imputer = SimpleImputer(strategy='median')
df[['age','income']] = imputer.fit_transform(
    df[['age','income']]
)

# 2. Remove outliers (IQR method)
Q1, Q3 = df['income'].quantile([0.25, 0.75])
IQR = Q3 - Q1
df = df[df['income'].between(Q1 - 1.5*IQR,
                              Q3 + 1.5*IQR)]

# 3. Encode categoricals
df = pd.get_dummies(df, columns=['city','gender'])

# 4. Scale numeric features
scaler = StandardScaler()
numeric = ['age','income','score']
df[numeric] = scaler.fit_transform(df[numeric])

print(f"Clean dataset: {df.shape} rows/cols")

import pandas as pd
from sklearn.decomposition import PCA

# Example: feature engineering for ride-share data
df['hour']        = pd.to_datetime(df['pickup']).dt.hour
df['is_rush']     = df['hour'].between(7, 9) | df['hour'].between(17, 19)
df['is_weekend']  = pd.to_datetime(df['pickup']).dt.dayofweek >= 5
df['distance_km'] = haversine(df['lat1'],df['lon1'],
                               df['lat2'],df['lon2'])
df['price_per_km'] = df['fare'] / (df['distance_km'] + 1e-9)

# Embeddings for categorical 'zone'
from sklearn.preprocessing import OrdinalEncoder
enc = OrdinalEncoder()
df['zone_enc'] = enc.fit_transform(df[['zone']])

# Dimensionality reduction on sensor data
pca = PCA(n_components=10)
sensor_cols = [c for c in df if c.startswith('sensor')]
df_pca = pca.fit_transform(df[sensor_cols])
print(f"Explained variance: {pca.explained_variance_ratio_.sum():.2%}")

import torch, torch.nn as nn, optuna

def objective(trial):
    lr    = trial.suggest_float('lr', 1e-5, 1e-2, log=True)
    batch = trial.suggest_categorical('batch', [16,32,64,128])
    drop  = trial.suggest_float('dropout', 0.1, 0.5)

    model = MyModel(dropout=drop)
    opt   = torch.optim.Adam(model.parameters(), lr=lr)
    loader = DataLoader(train_ds, batch_size=batch)

    for epoch in range(20):
        model.train()
        for X, y in loader:
            opt.zero_grad()
            loss = nn.CrossEntropyLoss()(model(X), y)
            loss.backward()
            opt.step()

    # Return validation metric
    return evaluate(model, val_loader)

study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=50)
print("Best:", study.best_params)

• Foundation Models & Fine-tuning: Pre-trained LLMs adapted to domain-specific tasks with minimal data (LoRA, QLoRA)
• Multimodal AI: Systems that reason across text, images, audio, video simultaneously (GPT-4V, Gemini)
• Agentic AI / AutoGPT: AI agents that autonomously plan, use tools, and execute multi-step tasks
• Neuromorphic & Quantum AI: Brain-inspired hardware and quantum computing for exponential speedups
• Edge AI proliferation: Efficient models on smartphones, IoT sensors, wearables — no cloud needed
• Federated Learning: Training across distributed data without centralising sensitive records

• AI-assisted programming: Tools like GitHub Copilot, Cursor — AI writing the AI systems of the future
• Automated ML (AutoML): NAS, AutoFeaturization, AutoDeploy reduce human intervention in SDLC
• Causal AI: Models that reason about cause-and-effect, not just correlations — more robust decisions
• Regulation & Standards: EU AI Act, ISO/IEC 42001 — formal standards for AI system development
• Hybrid Neurosymbolic AI: Combining neural networks with symbolic reasoning for interpretable, data-efficient systems
• AI Safety research: Alignment, interpretability, robustness — making AI reliably beneficial