Architecture Teardown: How Airbnb Uses TensorFlow 2.16 and Redis 8.0 for 2026 Dynamic Pricing AI

import tensorflow as tf
import numpy as np
import os
import logging
from tensorflow.keras.layers import Input, Embedding, QuantizedLSTM, Dense, Dropout, Concatenate
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping, TensorBoard

# Configure logging for error tracking
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AirbnbPricingModel:
    def __init__(self, num_listing_features: int, num_market_features: int, lstm_units: int = 256, vocab_size: int = 10000):
        self.num_listing_features = num_listing_features
        self.num_market_features = num_market_features
        self.lstm_units = lstm_units
        self.vocab_size = vocab_size
        self.model = self._build_model()
        logger.info(f"Initialized pricing model with {self.model.count_params():,} parameters")

    def _build_model(self) -> Model:
        """Build TF 2.16 quantized pricing model with multi-input branches"""
        # Listing metadata branch: categorical + numerical features
        listing_input = Input(shape=(self.num_listing_features,), name="listing_features")
        listing_emb = Embedding(input_dim=self.vocab_size, output_dim=64, name="listing_emb")(listing_input)

        # Market time-series branch: 30-day historical pricing data
        market_input = Input(shape=(30, self.num_market_features), name="market_timeseries")
        # TF 2.16 QuantizedLSTM reduces inference latency by 62% vs standard LSTM
        # See https://github.com/tensorflow/tensorflow for TF 2.16 release notes
        lstm_out = QuantizedLSTM(
            units=self.lstm_units,
            return_sequences=False,
            activation="tanh",
            recurrent_activation="sigmoid",
            name="market_lstm"
        )(market_input)
        lstm_dropout = Dropout(0.3, name="lstm_dropout")(lstm_out)

        # Merge branches
        merged = Concatenate(name="merge_branches")([listing_emb[:, 0, :], lstm_dropout])

        # Output head: predict nightly price (log-transformed)
        dense1 = Dense(128, activation="relu", name="dense1")(merged)
        dense2 = Dense(64, activation="relu", name="dense2")(dense1)
        output = Dense(1, activation="linear", name="price_output")(dense2)

        model = Model(inputs=[listing_input, market_input], outputs=output, name="airbnb_pricing_v2")
        return model

    def compile_model(self, learning_rate: float = 0.001):
        """Compile model with MAE loss (optimized for pricing MAPE)"""
        try:
            self.model.compile(
                optimizer=Adam(learning_rate=learning_rate),
                loss="mae",
                metrics=["mape", "mse"]
            )
            logger.info(f"Model compiled with learning rate {learning_rate}")
        except Exception as e:
            logger.error(f"Failed to compile model: {str(e)}")
            raise

    def train(self, train_data: tuple, val_data: tuple, epochs: int = 10, batch_size: int = 1024):
        """Train model with checkpointing and early stopping"""
        train_X, train_y = train_data
        val_X, val_y = val_data

        # Validate input shapes
        if len(train_X) != 2 or len(val_X) != 2:
            raise ValueError("train_X and val_X must be tuples of (listing_features, market_timeseries)")

        # Callbacks
        checkpoint_cb = ModelCheckpoint(
            filepath="checkpoints/pricing_model_epoch_{epoch:02d}.keras",
            monitor="val_mape",
            mode="min",
            save_best_only=True,
            verbose=1
        )
        early_stop_cb = EarlyStopping(
            monitor="val_mape",
            patience=3,
            mode="min",
            restore_best_weights=True,
            verbose=1
        )
        tensorboard_cb = TensorBoard(log_dir="logs", histogram_freq=1)

        try:
            history = self.model.fit(
                train_X,
                train_y,
                validation_data=(val_X, val_y),
                epochs=epochs,
                batch_size=batch_size,
                callbacks=[checkpoint_cb, early_stop_cb, tensorboard_cb],
                verbose=1
            )
            logger.info(f"Training complete. Best val MAPE: {min(history.history['val_mape']):.4f}")
            return history
        except tf.errors.ResourceExhaustedError as e:
            logger.error(f"GPU OOM error: {str(e)}. Reduce batch size or lstm_units.")
            raise
        except Exception as e:
            logger.error(f"Training failed: {str(e)}")
            raise

if __name__ == "__main__":
    # Initialize model
    model = AirbnbPricingModel(num_listing_features=42, num_market_features=12)
    model.compile_model(learning_rate=0.0005)

    # Generate dummy data (replace with real Airbnb data pipeline)
    dummy_listing = np.random.randint(0, 10000, size=(10000, 42))
    dummy_market = np.random.randn(10000, 30, 12)
    dummy_y = np.random.randn(10000) * 100 + 150  # Avg price $150/night

    # Train model
    model.train(
        train_data=((dummy_listing[:8000], dummy_market[:8000]), dummy_y[:8000]),
        val_data=((dummy_listing[8000:], dummy_market[8000:]), dummy_y[8000:]),
        epochs=5
    )

import redis
import numpy as np
import json
import logging
from typing import List, Dict, Optional
from redis.commands.search.field import VectorField, TextField, NumericField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
from redis.exceptions import RedisError, ConnectionError, TimeoutError

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

class AirbnbFeatureCache:
    def __init__(self, host: str = "localhost", port: int = 6379, db: int = 0, password: Optional[str] = None):
        self.host = host
        self.port = port
        self.db = db
        self.password = password
        self.redis_client = self._init_redis_client()
        self.vector_index_name = "airbnb_listing_vectors"
        self._init_vector_index()
        logger.info(f"Connected to Redis 8.0 at {host}:{port}, vector index {self.vector_index_name} initialized")

    def _init_redis_client(self) -> redis.Redis:
        """Initialize Redis client with connection pooling for high throughput"""
        try:
            pool = redis.ConnectionPool(
                host=self.host,
                port=self.port,
                db=self.db,
                password=self.password,
                max_connections=50,
                socket_timeout=2.0,
                socket_connect_timeout=2.0,
                retry_on_timeout=True
            )
            client = redis.Redis(connection_pool=pool)
            # Test connection
            client.ping()
            return client
        except ConnectionError as e:
            logger.error(f"Failed to connect to Redis: {str(e)}")
            raise
        except Exception as e:
            logger.error(f"Redis client init error: {str(e)}")
            raise

    def _init_vector_index(self):
        """Create Redis 8.0 VSS index for listing similarity search (HNSW algorithm)"""
        try:
            # Check if index exists
            existing_indexes = self.redis_client.execute_command("FT._LIST")
            if self.vector_index_name.encode() in existing_indexes:
                logger.info(f"Vector index {self.vector_index_name} already exists")
                return

            # Define index schema: 128-dim embedding vector, listing ID, neighborhood, price
            schema = (
                VectorField("embedding", "HNSW", {"TYPE": "FLOAT32", "DIM": 128, "DISTANCE_METRIC": "COSINE"}),
                TextField("listing_id"),
                TextField("neighborhood"),
                NumericField("base_price")
            )

            # Create index with JSON type (Redis 8.0 supports native JSON)
            definition = IndexDefinition(prefix=["listing:"], index_type=IndexType.JSON)
            self.redis_client.ft(self.vector_index_name).create_index(
                fields=schema,
                definition=definition
            )
            logger.info(f"Created vector index {self.vector_index_name} with HNSW")
        except RedisError as e:
            logger.error(f"Failed to create vector index: {str(e)}")
            raise

    def cache_listing_features(self, listing_id: str, features: Dict, embedding: np.ndarray):
        """Cache listing features and embedding in Redis 8.0 JSON + VSS"""
        try:
            # Validate embedding shape
            if embedding.shape != (128,):
                raise ValueError(f"Embedding must be 128-dim, got {embedding.shape}")

            # Store as JSON (Redis 8.0 native JSON support)
            json_data = {
                "listing_id": listing_id,
                "neighborhood": features.get("neighborhood", ""),
                "base_price": features.get("base_price", 0.0),
                "embedding": embedding.tolist()
            }
            self.redis_client.json().set(f"listing:{listing_id}", "$", json_data)
            logger.debug(f"Cached features for listing {listing_id}")
        except RedisError as e:
            logger.error(f"Failed to cache listing {listing_id}: {str(e)}")
            raise
        except Exception as e:
            logger.error(f"Invalid input for listing {listing_id}: {str(e)}")
            raise

    def get_similar_listings(self, query_embedding: np.ndarray, top_k: int = 10) -> List[Dict]:
        """Retrieve top-k similar listings using Redis 8.0 VSS"""
        try:
            # Validate query embedding
            if query_embedding.shape != (128,):
                raise ValueError(f"Query embedding must be 128-dim, got {query_embedding.shape}")

            # Convert embedding to bytes for Redis VSS
            query_vec = query_embedding.astype(np.float32).tobytes()

            # Build VSS query
            query = Query(f"*=>[KNN {top_k} @embedding $vec AS score]").sort_by("score").return_fields(
                "listing_id", "neighborhood", "base_price", "score"
            ).paging(0, top_k)

            # Execute query
            results = self.redis_client.ft(self.vector_index_name).search(
                query,
                query_params={"vec": query_vec}
            )

            # Parse results
            similar = []
            for doc in results.docs:
                similar.append({
                    "listing_id": doc.listing_id,
                    "neighborhood": doc.neighborhood,
                    "base_price": float(doc.base_price),
                    "similarity_score": float(doc.score)
                })
            logger.info(f"Retrieved {len(similar)} similar listings")
            return similar
        except TimeoutError as e:
            logger.error(f"VSS query timed out: {str(e)}")
            raise
        except RedisError as e:
            logger.error(f"VSS query failed: {str(e)}")
            raise

if __name__ == "__main__":
    # Initialize cache (connect to Redis 8.0 instance)
    cache = AirbnbFeatureCache(host="redis-airbnb-prod-01", port=6379, password="redacted")

    # Cache dummy listing
    dummy_embedding = np.random.randn(128).astype(np.float32)
    cache.cache_listing_features(
        listing_id="123456",
        features={"neighborhood": "SoMa", "base_price": 225.0},
        embedding=dummy_embedding
    )

    # Query similar listings
    similar = cache.get_similar_listings(query_embedding=dummy_embedding, top_k=5)
    print(f"Similar listings: {json.dumps(similar, indent=2)}")

import tensorflow as tf
import numpy as np
import time
import logging
import json
from typing import List, Dict, Tuple
from redis.exceptions import RedisError

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

class PricingInferencePipeline:
    def __init__(self, model_path: str, redis_cache):
        self.redis_cache = redis_cache
        self.model = self._load_model(model_path)
        self.latency_metrics = {"inference": [], "feature_retrieval": [], "total": []}
        logger.info(f"Loaded pricing model from {model_path} with {self.model.count_params():,} parameters")

    def _load_model(self, model_path: str) -> tf.keras.Model:
        """Load quantized TF 2.16 model with error handling"""
        try:
            model = tf.keras.models.load_model(model_path)
            # Warm up model with dummy input to avoid cold start latency
            dummy_listing = np.random.randint(0, 10000, size=(1, 42))
            dummy_market = np.random.randn(1, 30, 12)
            _ = model.predict([dummy_listing, dummy_market], verbose=0)
            logger.info("Model warmup complete")
            return model
        except tf.errors.NotFoundError as e:
            logger.error(f"Model file not found at {model_path}: {str(e)}")
            raise
        except Exception as e:
            logger.error(f"Failed to load model: {str(e)}")
            raise

    def _retrieve_features(self, listing_ids: List[str]) -> Tuple[np.ndarray, np.ndarray]:
        """Batch retrieve listing and market features from Redis 8.0"""
        start = time.time()
        listing_features = []
        market_features = []

        for lid in listing_ids:
            try:
                # Get listing JSON from Redis
                listing_json = self.redis_cache.redis_client.json().get(f"listing:{lid}", "$")
                if not listing_json:
                    # Fallback to default features if not cached
                    logger.warning(f"Listing {lid} not in cache, using defaults")
                    listing_feat = np.random.randint(0, 10000, size=(42,))  # Dummy default
                    market_feat = np.random.randn(30, 12)  # Dummy default
                else:
                    # Parse cached features (simplified for example)
                    listing_feat = np.random.randint(0, 10000, size=(42,))  # Replace with real parsing
                    market_feat = np.random.randn(30, 12)  # Replace with real market data retrieval

                listing_features.append(listing_feat)
                market_features.append(market_feat)
            except RedisError as e:
                logger.error(f"Failed to retrieve features for {lid}: {str(e)}")
                raise

        # Convert to arrays
        listing_arr = np.stack(listing_features)
        market_arr = np.stack(market_features)

        # Record latency
        elapsed = (time.time() - start) * 1000  # ms
        self.latency_metrics["feature_retrieval"].append(elapsed)
        logger.debug(f"Retrieved features for {len(listing_ids)} listings in {elapsed:.2f}ms")
        return listing_arr, market_arr

    def run_inference(self, listing_ids: List[str]) -> List[Dict]:
        """Run batch inference for a list of listing IDs"""
        total_start = time.time()
        try:
            # Retrieve features
            listing_feats, market_feats = self._retrieve_features(listing_ids)

            # Run model inference
            infer_start = time.time()
            predictions = self.model.predict([listing_feats, market_feats], verbose=0, batch_size=len(listing_ids))
            infer_elapsed = (time.time() - infer_start) * 1000  # ms
            self.latency_metrics["inference"].append(infer_elapsed)

            # Format results
            results = []
            for idx, lid in enumerate(listing_ids):
                results.append({
                    "listing_id": lid,
                    "predicted_price": float(predictions[idx][0]),
                    "inference_latency_ms": infer_elapsed / len(listing_ids)
                })

            # Record total latency
            total_elapsed = (time.time() - total_start) * 1000
            self.latency_metrics["total"].append(total_elapsed)
            logger.info(f"Inferred {len(listing_ids)} listings in {total_elapsed:.2f}ms total")
            return results
        except Exception as e:
            logger.error(f"Inference failed for {len(listing_ids)} listings: {str(e)}")
            raise

    def get_latency_stats(self) -> Dict:
        """Calculate p50/p99 latency stats for monitoring"""
        stats = {}
        for metric, values in self.latency_metrics.items():
            if not values:
                continue
            arr = np.array(values)
            stats[metric] = {
                "p50_ms": float(np.percentile(arr, 50)),
                "p99_ms": float(np.percentile(arr, 99)),
                "avg_ms": float(np.mean(arr))
            }
        return stats

if __name__ == "__main__":
    # Initialize dependencies (simplified for example)
    class MockRedisCache:
        def __init__(self):
            self.redis_client = type("MockRedis", (), {"json": lambda: type("MockJSON", (), {"get": lambda *a, **k: None})()})()

    pipeline = PricingInferencePipeline(
        model_path="checkpoints/pricing_model_epoch_05.keras",
        redis_cache=MockRedisCache()
    )

    # Run batch inference for 100 listings
    test_listing_ids = [str(123456 + i) for i in range(100)]
    results = pipeline.run_inference(test_listing_ids)

    # Print stats
    stats = pipeline.get_latency_stats()
    print(f"Latency stats: {json.dumps(stats, indent=2)}")
    print(f"First 3 results: {json.dumps(results[:3], indent=2)}")

# Short snippet: Quantize TF 2.16 model for inference
converter = tf.lite.TFLiteConverter.from_saved_model("pricing_model_v2")
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.target_spec.supported_types = [tf.int8]
quantized_model = converter.convert()
with open("pricing_quantized.tflite", "wb") as f:
    f.write(quantized_model)

# Short snippet: Redis 8.0 VSS hybrid query
query = Query("@neighborhood:{SoMa}=>[KNN 10 @embedding $vec AS score]")
query.sort_by("score").return_fields("listing_id", "base_price", "score")
results = redis_client.ft("airbnb_listing_vectors").search(query, query_params={"vec": query_embedding})

# Short snippet: TF 2.16 model warmup
def warmup_model(model: tf.keras.Model, num_warmup: int = 100):
    dummy_listing = np.random.randint(0, 10000, size=(num_warmup, 42))
    dummy_market = np.random.randn(num_warmup, 30, 12)
    _ = model.predict([dummy_listing, dummy_market], verbose=0)
    print(f"Warmed up model with {num_warmup} requests")