Podstawy AI i ML dla inżynierów

Uczenie maszynowe dzieli się na trzy paradygmaty według tego, jaki sygnał uczący ma dostęp do modelu: etykiety, strukturę danych bez etykiet, albo nagrodę ze środowiska. Wybór paradygmatu wynika z natury problemu i dostępności danych, nie z mody na konkretny algorytm.

from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)
clf = RandomForestClassifier(n_estimators=300, max_depth=8, random_state=42)
clf.fit(X_train, y_train)
accuracy = clf.score(X_test, y_test)

from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler

X_scaled = StandardScaler().fit_transform(X)
kmeans = KMeans(n_clusters=4, n_init="auto", random_state=42)
labels = kmeans.fit_predict(X_scaled)

import numpy as np
import gymnasium as gym

env = gym.make("FrozenLake-v1", is_slippery=False)
Q = np.zeros((env.observation_space.n, env.action_space.n))
alpha, gamma, epsilon = 0.1, 0.95, 0.1

for episode in range(5000):
    state, _ = env.reset()
    terminated = truncated = False
    while not (terminated or truncated):
        action = env.action_space.sample() if np.random.rand() < epsilon else np.argmax(Q[state])
        next_state, reward, terminated, truncated, _ = env.step(action)
        Q[state, action] += alpha * (reward + gamma * np.max(Q[next_state]) - Q[state, action])
        state = next_state

Każdy model popełnia błąd generalizacji z dwóch niezależnych źródeł. Bias to systematyczne uproszczenie zbyt sztywnego modelu. Wariancja to nadwrażliwość zbyt elastycznego modelu na szum w danych treningowych. Zrozumienie kompromisu bias-wariancja to podstawa diagnozy, dlaczego model działa źle, zanim zmienisz architekturę.

import numpy as np
from sklearn.linear_model import LinearRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import PolynomialFeatures
from sklearn.metrics import mean_squared_error

rng = np.random.default_rng(42)
X = rng.uniform(-3, 3, size=(40, 1))
y = 0.5 * X[:, 0] ** 2 - X[:, 0] + rng.normal(0, 1, size=40)

X_train, X_test = X[:30], X[30:]
y_train, y_test = y[:30], y[30:]

for degree in (1, 4, 15):
    model = make_pipeline(PolynomialFeatures(degree), LinearRegression())
    model.fit(X_train, y_train)
    train_mse = mean_squared_error(y_train, model.predict(X_train))
    test_mse = mean_squared_error(y_test, model.predict(X_test))
    print(f"degree={degree}: train_mse={train_mse:.2f}, test_mse={test_mse:.2f}")

1. Wykreśl krzywe uczenia (learning curves) Błąd treningowy i walidacyjny w funkcji liczby przykładów treningowych.
2. Zdiagnozuj wzorzec Obie krzywe wysoko i blisko siebie → underfitting. Duża luka między nimi → overfitting.
3. Zastosuj odpowiednią interwencję Underfitting: zwiększ pojemność modelu lub liczbę cech. Overfitting: regularyzacja, więcej danych, uproszczenie modelu.
4. Zweryfikuj na zbiorze testowym Dopiero po ustabilizowaniu walidacji, jednorazowo, żeby nie zanieczyścić decyzji o architekturze.

• Regularyzacja L1 (Lasso): wymusza rzadkość wag, efektywna selekcja cech
• Regularyzacja L2 (Ridge/weight decay): tłumi duże wagi bez zerowania ich
• Dropout: losowe wyłączanie neuronów podczas treningu sieci głębokich
• Early stopping: zatrzymanie treningu, gdy błąd walidacyjny przestaje spadać
• Augmentacja danych: sztuczne powiększenie zbioru treningowego bez zbierania nowych danych

Poprawny podział danych to fundament wiarygodnej ewaluacji modelu. Błąd metodologiczny na tym etapie daje fałszywie optymistyczne wyniki w laboratorium i katastrofę w produkcji. Model był bowiem oceniany na danych, które w praktyce nigdy nie powinien był 'widzieć' przed predykcją.

1. Preprocessing dopasowany na całym zbiorze przed podziałem (skalowanie, imputacja, selekcja cech).
2. Wyciek czasowy: losowe tasowanie danych chronologicznych, model 'widzi' przyszłość w treningu.
3. Wyciek cechy docelowej (target leakage): cecha wejściowa jest pochodną etykiety i nie będzie dostępna w momencie predykcji w produkcji.

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.svm import SVC
from sklearn.model_selection import cross_val_score, StratifiedKFold

pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("clf", SVC(kernel="rbf", C=1.0)),
])

cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
scores = cross_val_score(pipeline, X, y, cv=cv, scoring="f1_macro")
print(f"F1 macro: {scores.mean():.3f} +/- {scores.std():.3f}")

• Czy preprocessing był dopasowany wyłącznie na zbiorze treningowym?
• Czy jakakolwiek cecha jest dostępna dopiero po zaistnieniu etykiety (target leakage)?
• Czy dane mają porządek czasowy i czy podział go respektuje?
• Czy zbiór testowy był użyty więcej niż raz do podejmowania decyzji o modelu?

Embedding to gęsty wektor liczbowy o stałym wymiarze. Reprezentuje znaczenie obiektu (słowa, zdania, obrazu, użytkownika) tak, że obiekty semantycznie podobne mają wektory blisko siebie w przestrzeni wektorowej. To fundament wyszukiwania semantycznego, systemów rekomendacji i RAG (retrieval-augmented generation).

import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")
embeddings = model.encode([
    "Panel fotowoltaiczny z pęknięciem ogniwa",
    "Uszkodzona cela w module PV",
])

def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

similarity = cosine_similarity(embeddings[0], embeddings[1])

• Wyszukiwanie semantyczne: dopasowanie po znaczeniu, nie po dokładnych słowach kluczowych
• RAG: pobranie kontekstu najbardziej relewantnego semantycznie do zapytania przed wygenerowaniem odpowiedzi przez LLM
• Systemy rekomendacji: podobieństwo produktów lub użytkowników w przestrzeni wektorowej
• Deduplikacja: wykrywanie niemal identycznych treści mimo różnic w sformułowaniu

Accuracy jest najbardziej intuicyjną metryką i najczęściej złym wyborem w realnych problemach klasyfikacji, gdzie klasy są niezbalansowane. Precision, recall, F1 i AUC-ROC rozbijają jakość modelu na komponenty, które faktycznie odpowiadają na pytania biznesowe: ile fałszywych alarmów, ile przeoczonych przypadków.

from sklearn.metrics import classification_report, roc_auc_score
from sklearn.linear_model import LogisticRegression

clf = LogisticRegression(class_weight="balanced", max_iter=1000)
clf.fit(X_train, y_train)

y_pred = clf.predict(X_test)
y_proba = clf.predict_proba(X_test)[:, 1]

print(classification_report(y_test, y_pred, target_names=["ok", "defekt"]))
print(f"AUC-ROC: {roc_auc_score(y_test, y_proba):.3f}")

1. Zidentyfikuj koszt każdego typu błędu Ile kosztuje fałszywy alarm, ile kosztuje przeoczenie.
2. Sprawdź balans klas Jeśli klasa mniejszościowa poniżej ~10-20%, accuracy jest niewiarygodne.
3. Wybierz metrykę zgodną z kosztem Recall gdy przeoczenie jest drogie, precision gdy fałszywy alarm jest drogi, F1 gdy oba porównywalne.
4. Dostrój próg decyzyjny Na podstawie krzywej precision-recall, nie zostawiaj domyślnego 0.5.

Transformer, wprowadzony w pracy 'Attention Is All You Need' (2017), zastąpił rekurencyjne sieci (RNN, LSTM) jako dominującą architekturę dla sekwencji dzięki jednej zmianie koncepcyjnej. Każdy token 'patrzy' na wszystkie inne tokeny naraz (mechanizm uwagi, attention), a nie krok po kroku.

import numpy as np

def scaled_dot_product_attention(Q, K, V):
    d_k = Q.shape[-1]
    scores = Q @ K.T / np.sqrt(d_k)
    weights = np.exp(scores - scores.max(axis=-1, keepdims=True))
    weights /= weights.sum(axis=-1, keepdims=True)
    return weights @ V

Q = np.random.randn(4, 8)  # 4 tokeny, wymiar reprezentacji 8
K = np.random.randn(4, 8)
V = np.random.randn(4, 8)
output = scaled_dot_product_attention(Q, K, V)

• Warstwa self-attention (multi-head)
• Warstwa feed-forward (dwuwarstwowa sieć gęsta stosowana niezależnie do każdego tokenu)
• Połączenia rezydualne (residual connections) wokół obu podwarstw
• Layer normalization stabilizująca trening głębokich stosów bloków

Dojrzałość inżynierska w AI/ML objawia się nie tylko umiejętnością zbudowania modelu, ale umiejętnością rozpoznania, kiedy model jest złym pomysłem. Presja żeby 'użyć AI' bo jest modne, jest realnym źródłem kosztownych, kruchych systemów produkcyjnych.

• Problem rozwiązywalny jedną jasną regułą biznesową lub zapytaniem SQL
• Zbiór danych zbyt mały, by model nauczył się generalizowalnego wzorca (rząd dziesiątek przykładów)
• Wymóg regulacyjny pełnej wyjaśnialności decyzji, którego czarnoskrzynkowy model nie spełnia
• Koszt błędu ekstremalnie wysoki (bezpieczeństwo, zdrowie) bez możliwości nadzoru człowieka nad każdą decyzją
• Środowisko zmienia się szybciej niż cykl retrainingu modelu

def classify_with_guardrail(model, x, confidence_threshold: float = 0.85):
    proba = model.predict_proba([x])[0]
    predicted_class = proba.argmax()
    confidence = proba[predicted_class]

    if confidence < confidence_threshold:
        return {"decision": "eskalacja_do_czlowieka", "confidence": confidence}
    return {"decision": predicted_class, "confidence": confidence}

1. Zdefiniuj problem bez wspominania o ML Jeśli da się go rozwiązać regułą lub zapytaniem, zrób to.
2. Oceń dostępność i jakość danych Brak wystarczających, reprezentatywnych danych dyskwalifikuje ML niezależnie od architektury.
3. Policz koszt błędu i wymóg wyjaśnialności Wysoki koszt bez nadzoru człowieka i wymóg pełnej audytowalności to sygnał ostrzegawczy.
4. Zaprojektuj guardrail zanim wdrożysz Próg pewności, eskalacja do człowieka, plan rollbacku, zawsze zanim model dotknie realnych decyzji.

W praktyce inżynierskiej jakość i reprezentatywność danych treningowych decydują o wyniku projektu ML częściej niż wybór architektury czy hiperparametrów. Nurt data-centric AI formalizuje to: poprawa etykiet i usunięcie szumu zwykle poprawia model bardziej niż zmiana algorytmu.

import numpy as np
from scipy.stats import ks_2samp

def detect_feature_drift(reference: np.ndarray, production: np.ndarray, alpha: float = 0.01) -> bool:
    statistic, p_value = ks_2samp(reference, production)
    return p_value < alpha  # True = statystycznie istotny dryf

drift_detected = detect_feature_drift(
    reference_batch["panel_temp_delta"],
    production_batch["panel_temp_delta"],
)

• Latencja: budżet czasowy na pojedynczą inferencję zgodny z wymogiem UX/SLA
• Koszt: koszt inferencji na żądanie przy skali produkcyjnej, nie tylko koszt treningu
• Monitoring: dashboard metryk jakości predykcji i rozkładu danych wejściowych w czasie
• Alerting: automatyczne powiadomienie, gdy metryka lub dryf przekroczy próg
• Plan rollbacku: możliwość szybkiego powrotu do poprzedniej wersji modelu

1. Walidacja danych przed treningiem Automatyczne sprawdzenie schematu, zakresów wartości i braków danych.
2. Shadow deployment Nowy model liczy predykcje równolegle z produkcyjnym, bez wpływu na decyzje, do porównania metryk.
3. Canary release Stopniowe przekierowanie małego procenta ruchu na nowy model przed pełnym wdrożeniem.
4. Monitoring ciągły Śledzenie dryfu danych, latencji, kosztu i metryk jakości predykcji w czasie.
5. Plan rollbacku Automatyczny lub ręczny powrót do poprzedniej wersji przy przekroczeniu progów alertowych.