TensorFlow и Keras

🎯 Цель

После прочтения этой главы:

• Знаете разницу между TensorFlow/Keras и PyTorch
• Строите модель через Keras Sequential и Functional API
• Знаете экосистему TensorFlow (TF Serving, TF Lite, TF.js)
• Осознанно выбираете свой основной framework

**Внимание:**Ваш основной framework — PyTorch(industry default). TF/Keras изучаем просто для ознакомления, потому что:

• Всё ещё есть в старых проектах
• Интеграция с Google Cloud (Vertex AI)
• TF Lite — mobile/edge deployment

Что нужно изучить

• TensorFlow 2.x — eager execution, tf.function
• Keras Sequential API — последовательное добавление слоёв
• Keras Functional API — сложные архитектуры
• Model.compile, fit, predict — high-level API
• Callbacks — EarlyStopping, ModelCheckpoint, TensorBoard
• tf.data.Dataset — efficient data pipeline
• TF Serving — production deployment
• TF Lite — mobile/edge inference
• TF.js — inference в браузере

Библиотеки

pip install tensorflow
# или
pip install tensorflow-macos tensorflow-metal  # Для Mac M-chip

Версия: TensorFlow 2.15+(только 2.x).

PyTorch vs TensorFlow/Keras

**Совет:**Изучайте PyTorch как основу, а TF/Keras на уровне «знакомства».

Примеры кода

Sequential API — самый простой

import tensorflow as tf
from tensorflow.keras import layers, models

model = models.Sequential([
    layers.Input(shape=(784,)),
    layers.Dense(256, activation="relu"),
    layers.Dropout(0.3),
    layers.Dense(128, activation="relu"),
    layers.Dropout(0.3),
    layers.Dense(10, activation="softmax"),
])

model.compile(
    optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3),
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)

model.summary()

Training (fit API)

from tensorflow.keras.datasets import mnist

(X_train, y_train), (X_test, y_test) = mnist.load_data()
X_train = X_train.reshape(-1, 784) / 255.0
X_test = X_test.reshape(-1, 784) / 255.0

history = model.fit(
    X_train, y_train,
    epochs=10,
    batch_size=128,
    validation_split=0.1,
    verbose=1,
)

# Evaluation
test_loss, test_acc = model.evaluate(X_test, y_test, verbose=0)
print(f"Test accuracy: {test_acc:.4f}")

Functional API — посложнее

inputs = layers.Input(shape=(784,))
x = layers.Dense(256, activation="relu")(inputs)
x = layers.Dropout(0.3)(x)

# Branching (преимущество Functional API)
branch_a = layers.Dense(64, activation="relu")(x)
branch_b = layers.Dense(64, activation="relu")(x)
merged = layers.Concatenate()([branch_a, branch_b])

outputs = layers.Dense(10, activation="softmax")(merged)
model = models.Model(inputs=inputs, outputs=outputs)

Callbacks

from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint, TensorBoard

callbacks = [
    EarlyStopping(monitor="val_loss", patience=5, restore_best_weights=True),
    ModelCheckpoint("best_model.keras", monitor="val_accuracy", save_best_only=True),
    TensorBoard(log_dir="./logs"),
]

model.fit(X_train, y_train, epochs=50, callbacks=callbacks, validation_split=0.1)
# TensorBoard: $ tensorboard --logdir=./logs

Custom training loop (PyTorch-like)

optimizer = tf.keras.optimizers.Adam()
loss_fn = tf.keras.losses.SparseCategoricalCrossentropy()
acc_metric = tf.keras.metrics.SparseCategoricalAccuracy()

@tf.function  # graph mode (быстрее)
def train_step(X, y):
    with tf.GradientTape() as tape:
        logits = model(X, training=True)
        loss = loss_fn(y, logits)
    grads = tape.gradient(loss, model.trainable_weights)
    optimizer.apply_gradients(zip(grads, model.trainable_weights))
    acc_metric.update_state(y, logits)
    return loss

for epoch in range(10):
    for X_batch, y_batch in train_dataset:
        loss = train_step(X_batch, y_batch)
    print(f"Epoch {epoch+1}: loss={loss:.4f}, acc={acc_metric.result():.4f}")
    acc_metric.reset_state()

tf.data.Dataset — efficient pipeline

import tensorflow as tf

dataset = tf.data.Dataset.from_tensor_slices((X_train, y_train))
dataset = (dataset
    .shuffle(buffer_size=10000)
    .batch(128)
    .prefetch(tf.data.AUTOTUNE))

for X_batch, y_batch in dataset.take(1):
    print(X_batch.shape, y_batch.shape)

Сохранение и загрузка

# Сохранение (Keras format — новый стандарт)
model.save("model.keras")

# Загрузка
loaded = tf.keras.models.load_model("model.keras")

# Только weights
model.save_weights("weights.h5")
new_model.load_weights("weights.h5")

# SavedModel format (для TF Serving)
model.export("saved_model_dir/")

TFLite — mobile/edge

converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()

with open("model.tflite", "wb") as f:
    f.write(tflite_model)

# Loading (mobile)
interpreter = tf.lite.Interpreter(model_path="model.tflite")
interpreter.allocate_tensors()

Интеграция с backend

TF Serving (RECOMMENDED for TF models)

# Через Docker
docker run -p 8501:8501 \
  --mount type=bind,source=$(pwd)/saved_model_dir,target=/models/my_model \
  -e MODEL_NAME=my_model \
  tensorflow/serving

# REST API
curl -X POST http://localhost:8501/v1/models/my_model:predict \
  -d '{"instances": [[1.0, 2.0, 3.0, ...]]}'

FastAPI proxy to TF Serving

import httpx
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
TF_SERVING_URL = "http://localhost:8501/v1/models/my_model:predict"

class Input(BaseModel):
    features: list[float]

@app.post("/predict")
async def predict(data: Input):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            TF_SERVING_URL,
            json={"instances": [data.features]},
        )
    return response.json()

Keras модель напрямую в FastAPI

import tensorflow as tf
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    app.state.model = tf.keras.models.load_model("model.keras")
    yield

app = FastAPI(lifespan=lifespan)

@app.post("/predict")
def predict(data: Input):
    X = tf.constant([data.features])
    logits = app.state.model(X, training=False).numpy()
    return {"prediction": int(logits.argmax()), "confidence": float(logits.max())}

Ресурсы

• TensorFlow tutorials — tensorflow.org/tutorials
• Keras docs — keras.io
• “Deep Learning with Python” — François Chollet (создатель Keras, 2-е издание)
• Andrew Ng — TensorFlow Professional Certificate(Coursera)
• TensorFlow YouTube channel — official tutorials

🏋️ Упражнения

🟢 Easy

1. Создайте 3-layer MLP через Sequential API и обучите на MNIST.
2. Интерпретируйте число параметров, показанное model.summary().
3. Используйте callback EarlyStopping для автоматической остановки training.

🟡 Medium

1. Custom callback: напишите custom callback, выводящий loss/acc каждые 5 epoch.
2. Functional API: создайте модель с 2 input (numeric + categorical).
3. Same model, two frameworks: напишите одинаковую MLP в PyTorch и Keras, сравните accuracy и training time.

🔴 Hard

1. TF Serving deployment: deploy MNIST модель в TF Serving через Docker, делайте predict через REST API.
2. TF Lite mobile: экспортируйте модель в .tflite, делайте inference в Python или Android emulator.

Capstone

notebooks/month-03/03_keras_mnist.ipynb:

• Создайте модели через Keras Sequential и Functional API на MNIST
• Логирование в TensorBoard
• EarlyStopping + ModelCheckpoint через Callbacks
• Export в TF Lite
• Сравнение с PyTorch capstone

✅ Чек-лист

• Знаю разницу между Sequential и Functional API
• Знаю workflow model.compile / fit / evaluate
• Использую Callbacks (EarlyStopping, ModelCheckpoint)
• Умею создавать tf.data.Dataset pipeline
• Умею сохранять модель в форматах .keras, SavedModel, TFLite
• Понимаю концепцию TF Serving
• Обосновываю выбор между PyTorch и TF/Keras

Переходим к Техники обучения — фокус снова на PyTorch.