Chapter 8: Production Deployment

8.1 From Lab to Production

Deploying RL systems in production requires careful consideration of performance, safety, monitoring, and maintenance. This chapter covers the engineering practices needed to successfully deploy RL agents in real-world systems.

Production Requirements

• Latency: Real-time inference (milliseconds to seconds)
• Reliability: High availability and fault tolerance
• Safety: Bounded behavior, fail-safes
• Monitoring: Track performance and detect drift
• Scalability: Handle production-scale load
• Explainability: Understand agent decisions

8.2 Model Serving Architecture

Inference Server

from fastapi import FastAPI
import torch
import uvicorn

app = FastAPI()

class RLModelServer:
    def __init__(self, model_path):
        // Load trained model
        self.model = torch.load(model_path)
        self.model.eval()

        // Move to GPU if available
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.model.to(self.device)

    def preprocess(self, observation):
        """Normalize and convert to tensor."""
        obs = torch.FloatTensor(observation).unsqueeze(0)
        return obs.to(self.device)

    def postprocess(self, action_tensor):
        """Convert tensor to action."""
        return action_tensor.cpu().numpy()[0]

    @torch.no_grad()
    def predict(self, observation):
        """Run inference."""
        obs_tensor = self.preprocess(observation)
        action_tensor = self.model(obs_tensor)
        action = self.postprocess(action_tensor)
        return action

// Global model server instance
model_server = RLModelServer(model_path='models/agent.pt')

@app.post("/predict")
async def predict(observation: list):
    """
    Inference endpoint.

    Request: {"observation": [0.1, 0.2, 0.3, ...]}
    Response: {"action": [1.5, -0.3, ...], "latency_ms": 12.5}
    """
    import time
    start = time.time()

    action = model_server.predict(observation)

    latency = (time.time() - start) * 1000

    return {
        "action": action.tolist(),
        "latency_ms": latency
    }

@app.get("/health")
async def health():
    return {"status": "healthy"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

TensorFlow Serving

// Export model for TF Serving
import tensorflow as tf

// Save model in SavedModel format
tf.saved_model.save(model, 'serving_model/1')

// Serve using TensorFlow Serving (Docker)
docker run -p 8501:8501 \
  --mount type=bind,source=$(pwd)/serving_model,target=/models/rl_agent \
  -e MODEL_NAME=rl_agent \
  -t tensorflow/serving

// Client code
import requests
import json

def predict(observation):
    url = 'http://localhost:8501/v1/models/rl_agent:predict'

    data = json.dumps({
        "signature_name": "serving_default",
        "instances": [observation.tolist()]
    })

    response = requests.post(url, data=data, headers={"content-type": "application/json"})
    prediction = response.json()['predictions'][0]

    return prediction

8.3 Distributed Training

Data Parallel Training

import torch
import torch.distributed as dist
import torch.multiprocessing as mp
from torch.nn.parallel import DistributedDataParallel as DDP

def setup(rank, world_size):
    os.environ['MASTER_ADDR'] = 'localhost'
    os.environ['MASTER_PORT'] = '12355'
    dist.init_process_group("nccl", rank=rank, world_size=world_size)

def cleanup():
    dist.destroy_process_group()

def train_worker(rank, world_size):
    setup(rank, world_size)

    // Create model and move to GPU
    model = PolicyNetwork().to(rank)
    ddp_model = DDP(model, device_ids=[rank])

    // Create environment
    env = gym.make('Humanoid-v2')

    // Training loop
    for episode in range(num_episodes):
        state = env.reset()
        done = False

        while not done:
            action = ddp_model(torch.FloatTensor(state).to(rank))
            next_state, reward, done, _ = env.step(action.cpu().numpy())

            // Compute loss
            loss = compute_loss(...)

            // Backward pass (gradients averaged across GPUs)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

            state = next_state

    cleanup()

def main():
    world_size = torch.cuda.device_count()
    mp.spawn(train_worker, args=(world_size,), nprocs=world_size, join=True)

if __name__ == "__main__":
    main()

Ray for Distributed RL

import ray
from ray.rllib.agents.ppo import PPOTrainer

ray.init()

// Configuration
config = {
    "env": "CartPole-v1",
    "num_workers": 16,           // Parallel environment workers
    "num_gpus": 4,               // GPUs for training
    "train_batch_size": 4000,
    "sgd_minibatch_size": 128,
    "num_sgd_iter": 10,
    "lr": 3e-4,
    "framework": "torch"
}

// Create trainer
trainer = PPOTrainer(config=config)

// Training
for iteration in range(1000):
    result = trainer.train()

    print(f"Iteration {iteration}")
    print(f"  Reward: {result['episode_reward_mean']:.2f}")
    print(f"  Length: {result['episode_len_mean']:.2f}")

    // Save checkpoint
    if iteration % 100 == 0:
        checkpoint = trainer.save()
        print(f"  Checkpoint saved: {checkpoint}")

// Export for deployment
trainer.export_policy_model("./exported_models")

8.4 Safety and Constraints

Safe Exploration

class SafetyLayer:
    """
    Safety wrapper that prevents unsafe actions.
    """
    def __init__(self, env, safety_constraints):
        self.env = env
        self.constraints = safety_constraints

    def is_safe(self, state, action):
        """Check if action satisfies all safety constraints."""
        for constraint in self.constraints:
            if not constraint.check(state, action):
                return False
        return True

    def get_safe_action(self, state, proposed_action):
        """
        Project unsafe action to nearest safe action.
        """
        if self.is_safe(state, proposed_action):
            return proposed_action

        // Find nearest safe action (optimization problem)
        safe_action = self.project_to_safe_set(state, proposed_action)
        return safe_action

// Usage
safe_env = SafetyLayer(env, safety_constraints=[
    MaxVelocityConstraint(max_speed=5.0),
    WorkspaceConstraint(bounds=workspace_bounds),
    CollisionConstraint(obstacles=obstacles)
])

action = policy(state)
safe_action = safe_env.get_safe_action(state, action)
next_state, reward, done = env.step(safe_action)

Constrained RL

class CPO:  // Constrained Policy Optimization
    """
    Optimize policy while satisfying safety constraints:

    maximize E[reward]
    subject to E[cost] ≤ threshold
    """
    def __init__(self, cost_limit=25):
        self.policy = PolicyNetwork()
        self.value = ValueNetwork()
        self.cost_value = ValueNetwork()  // Estimate cost
        self.cost_limit = cost_limit

    def update(self, trajectories):
        // Compute advantage and cost advantage
        advantages = self.compute_advantages(trajectories)
        cost_advantages = self.compute_cost_advantages(trajectories)

        // Current policy performance
        current_reward = self.evaluate_reward()
        current_cost = self.evaluate_cost()

        // Constrained optimization
        // Use Lagrangian: L = reward - λ * (cost - limit)
        policy_loss = -(advantages - λ * cost_advantages).mean()

        // Update policy
        optimizer.zero_grad()
        policy_loss.backward()

        // Project gradient to satisfy constraints
        self.project_gradients(current_cost, self.cost_limit)

        optimizer.step()

        // Update λ (Lagrange multiplier)
        if current_cost > self.cost_limit:
            λ *= 1.05  // Increase penalty
        else:
            λ *= 0.95  // Decrease penalty

8.5 Monitoring and Observability

Metrics Collection

from prometheus_client import Counter, Histogram, Gauge
import time

// Define metrics
inference_counter = Counter('rl_inference_total', 'Total inference requests')
inference_latency = Histogram('rl_inference_latency_seconds', 'Inference latency')
action_distribution = Histogram('rl_action_values', 'Distribution of action values')
reward_gauge = Gauge('rl_episode_reward', 'Episode reward')

class MonitoredAgent:
    def __init__(self, policy):
        self.policy = policy

    @inference_latency.time()
    def predict(self, observation):
        inference_counter.inc()

        action = self.policy(observation)

        // Track action distribution
        action_distribution.observe(action.mean())

        return action

    def on_episode_end(self, total_reward):
        reward_gauge.set(total_reward)

// Logging
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class LoggingAgent:
    def predict(self, observation):
        logger.info(f"Observation: {observation}")

        action = self.policy(observation)

        logger.info(f"Action: {action}")

        return action

Anomaly Detection

class AnomalyDetector:
    """
    Detect distribution shift and anomalous behavior.
    """
    def __init__(self):
        self.observation_stats = {
            'mean': None,
            'std': None,
            'min': None,
            'max': None
        }

    def fit(self, observations):
        """Learn normal observation distribution."""
        import numpy as np

        self.observation_stats['mean'] = np.mean(observations, axis=0)
        self.observation_stats['std'] = np.std(observations, axis=0)
        self.observation_stats['min'] = np.min(observations, axis=0)
        self.observation_stats['max'] = np.max(observations, axis=0)

    def detect_anomaly(self, observation):
        """Check if observation is anomalous."""
        import numpy as np

        // Z-score based detection
        z_score = np.abs(
            (observation - self.observation_stats['mean']) /
            (self.observation_stats['std'] + 1e-8)
        )

        is_anomaly = np.any(z_score > 3.0)  // 3-sigma rule

        if is_anomaly:
            logger.warning(f"Anomalous observation detected: {observation}")
            // Trigger alert, use fallback policy, etc.

        return is_anomaly

8.6 Online Learning and Adaptation

Continual Learning

class ContinualLearner:
    """
    Continue learning from production data.
    """
    def __init__(self, policy, update_frequency=1000):
        self.policy = policy
        self.buffer = ReplayBuffer(capacity=100000)
        self.update_frequency = update_frequency
        self.step_count = 0

    def act(self, observation):
        action = self.policy(observation)
        return action

    def observe(self, observation, action, reward, next_observation):
        // Store experience
        self.buffer.add(observation, action, reward, next_observation)

        self.step_count += 1

        // Periodic updates
        if self.step_count % self.update_frequency == 0:
            self.update_policy()

    def update_policy(self):
        """Update policy from recent experience."""
        if len(self.buffer) < batch_size:
            return

        batch = self.buffer.sample(batch_size)

        // Compute loss
        loss = compute_td_loss(self.policy, batch)

        // Update with small learning rate (stability)
        optimizer = torch.optim.Adam(self.policy.parameters(), lr=1e-5)
        optimizer.zero_grad()
        loss.backward()
        torch.nn.utils.clip_grad_norm_(self.policy.parameters(), 0.5)
        optimizer.step()

        logger.info(f"Policy updated. Loss: {loss.item():.4f}")

A/B Testing

class ABTestingFramework:
    """
    Compare multiple policy versions in production.
    """
    def __init__(self, policies, traffic_split=None):
        self.policies = policies  // {'baseline': policy_A, 'treatment': policy_B}

        if traffic_split is None:
            traffic_split = {k: 1.0/len(policies) for k in policies}

        self.traffic_split = traffic_split
        self.metrics = {k: [] for k in policies}

    def select_policy(self, user_id):
        """Route traffic to policy variants."""
        import random

        // Deterministic routing based on user_id
        hash_val = hash(user_id) % 100

        cumulative = 0
        for policy_name, percentage in self.traffic_split.items():
            cumulative += percentage * 100
            if hash_val < cumulative:
                return policy_name, self.policies[policy_name]

        return list(self.policies.items())[0]

    def record_metric(self, policy_name, reward):
        self.metrics[policy_name].append(reward)

    def analyze_results(self):
        """Statistical analysis of A/B test."""
        import numpy as np
        from scipy import stats

        baseline_rewards = self.metrics['baseline']
        treatment_rewards = self.metrics['treatment']

        // T-test
        t_stat, p_value = stats.ttest_ind(treatment_rewards, baseline_rewards)

        print(f"Baseline mean reward: {np.mean(baseline_rewards):.2f}")
        print(f"Treatment mean reward: {np.mean(treatment_rewards):.2f}")
        print(f"P-value: {p_value:.4f}")

        if p_value < 0.05:
            if np.mean(treatment_rewards) > np.mean(baseline_rewards):
                print("Treatment is significantly better!")
            else:
                print("Baseline is significantly better!")
        else:
            print("No significant difference")

8.7 Model Compression and Optimization

Quantization

import torch

// Post-training quantization
model = torch.load('policy.pt')
model.eval()

// Dynamic quantization (weights + activations)
quantized_model = torch.quantization.quantize_dynamic(
    model, {torch.nn.Linear}, dtype=torch.qint8
)

// Save quantized model
torch.save(quantized_model, 'policy_quantized.pt')

// Model size comparison
import os
original_size = os.path.getsize('policy.pt') / 1024  // KB
quantized_size = os.path.getsize('policy_quantized.pt') / 1024

print(f"Original: {original_size:.2f} KB")
print(f"Quantized: {quantized_size:.2f} KB")
print(f"Compression: {original_size/quantized_size:.2f}x")

Knowledge Distillation

class PolicyDistillation:
    """
    Train smaller student policy to mimic large teacher.
    """
    def __init__(self, teacher_policy, student_policy):
        self.teacher = teacher_policy
        self.student = student_policy

        self.teacher.eval()  // Freeze teacher

    def distill(self, states, temperature=2.0):
        """
        Train student to match teacher's action distribution.
        """
        with torch.no_grad():
            // Soft teacher predictions
            teacher_logits = self.teacher(states) / temperature
            teacher_probs = F.softmax(teacher_logits, dim=-1)

        // Student predictions
        student_logits = self.student(states) / temperature
        student_log_probs = F.log_softmax(student_logits, dim=-1)

        // KL divergence loss
        loss = F.kl_div(student_log_probs, teacher_probs, reduction='batchmean')

        return loss

// Training loop
teacher = LargePolicy()  // 10M parameters
student = SmallPolicy()  // 1M parameters

optimizer = torch.optim.Adam(student.parameters(), lr=1e-3)

for epoch in range(100):
    states = sample_states()

    loss = distillation.distill(states)

    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

print(f"Student trained! Parameters: {count_parameters(student)}")

8.8 Deployment Checklist

Deployment Stages

8.9 Case Studies

Google Data Center Cooling

DeepMind's RL system reduced cooling energy by 40%:

• Environment: Data center with hundreds of sensors and actuators
• State: Temperature, pressure, power consumption (time-series)
• Actions: Adjust cooling system settings
• Reward: -energy_consumption (minimize)
• Safety: Hard constraints on temperature ranges
• Deployment: Gradual rollout with human oversight

Recommendation Systems

class RecommenderRL:
    """
    RL for personalized recommendations.
    """
    def __init__(self):
        self.policy = SlateGenerationPolicy()

    def recommend(self, user_id, context):
        """
        Generate slate of recommendations.

        State: user history, context, time
        Action: slate of items to show
        Reward: clicks, watch time, satisfaction
        """
        state = self.get_state(user_id, context)

        // Generate candidate items
        candidates = self.candidate_generation(user_id)

        // Rank and select top-k
        slate = self.policy(state, candidates)

        return slate

    def update(self, user_id, slate, feedback):
        """Learn from user feedback."""
        reward = self.compute_reward(feedback)

        self.buffer.add(state, slate, reward)

        if len(self.buffer) > batch_size:
            self.train_step()

8.10 Future Directions

Emerging Trends

• Offline RL: Learn from logged data without environment interaction
• Meta-RL: Learn to learn, adapt quickly to new tasks
• Sim-to-Real: Better transfer from simulation to real world
• Explainable RL: Interpret agent decisions
• Foundation Models: Pre-trained RL models for transfer