How to Become a Senior AI Engineer in 2026 – 5 Years of Experience vs. Project Portfolio

import os
import sys
import logging
from typing import List, Dict, Optional
from dotenv import load_dotenv
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Pinecone as LangchainPinecone
from langchain_community.llms import VLLM
from langchain.chains import RetrievalQA
from pinecone import Pinecone, ServerlessSpec
import warnings
warnings.filterwarnings("ignore")

# Configure logging for error tracking
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Load environment variables from .env file
load_dotenv()
PINECONE_API_KEY = os.getenv("PINECONE_API_KEY")
PINECONE_ENV = os.getenv("PINECONE_ENV", "us-east-1")
VLLM_MODEL_PATH = os.getenv("VLLM_MODEL_PATH", "meta-llama/Llama-3.1-8B-Instruct")
DOC_DIR = os.getenv("DOC_DIR", "./docs")

class ProductionRAGPipeline:
    def __init__(self, index_name: str = "ai-eng-rag-2026"):
        self.index_name = index_name
        self.embeddings = None
        self.vectorstore = None
        self.llm = None
        self.qa_chain = None
        self._validate_env_vars()

    def _validate_env_vars(self) -> None:
        """Validate required environment variables are set"""
        required_vars = ["PINECONE_API_KEY"]
        missing = [var for var in required_vars if not os.getenv(var)]
        if missing:
            raise ValueError(f"Missing required environment variables: {missing}")

    def load_and_split_docs(self, chunk_size: int = 1024, chunk_overlap: int = 200) -> List:
        """Load documents from directory and split into chunks with error handling"""
        docs = []
        supported_extensions = [".pdf", ".txt", ".md"]
        try:
            for file in os.listdir(DOC_DIR):
                file_path = os.path.join(DOC_DIR, file)
                if not os.path.isfile(file_path):
                    continue
                ext = os.path.splitext(file)[1].lower()
                if ext not in supported_extensions:
                    logger.warning(f"Skipping unsupported file: {file}")
                    continue
                try:
                    if ext == ".pdf":
                        loader = PyPDFLoader(file_path)
                    else:
                        loader = TextLoader(file_path)
                    docs.extend(loader.load())
                    logger.info(f"Loaded {file}: {len(loader.load())} pages")
                except Exception as e:
                    logger.error(f"Failed to load {file}: {str(e)}")
                    continue
            if not docs:
                raise ValueError(f"No valid documents found in {DOC_DIR}")
            text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=chunk_size,
                chunk_overlap=chunk_overlap,
                length_function=len,
            )
            return text_splitter.split_documents(docs)
        except Exception as e:
            logger.error(f"Document loading failed: {str(e)}")
            raise

    def init_vectorstore(self, chunks: List) -> None:
        """Initialize Pinecone vectorstore with embeddings"""
        try:
            self.embeddings = HuggingFaceEmbeddings(
                model_name="sentence-transformers/all-MiniLM-L6-v2"
            )
            pc = Pinecone(api_key=PINECONE_API_KEY)
            # Create index if it doesn't exist
            if self.index_name not in pc.list_indexes().names():
                pc.create_index(
                    name=self.index_name,
                    dimension=384,  # Match embedding model dimension
                    spec=ServerlessSpec(cloud="aws", region=PINECONE_ENV)
                )
                logger.info(f"Created new Pinecone index: {self.index_name}")
            # Initialize LangChain Pinecone wrapper
            self.vectorstore = LangchainPinecone.from_documents(
                documents=chunks,
                embedding=self.embeddings,
                index_name=self.index_name
            )
            logger.info(f"Loaded {len(chunks)} chunks into Pinecone index")
        except Exception as e:
            logger.error(f"Vectorstore initialization failed: {str(e)}")
            raise

    def init_llm(self, max_new_tokens: int = 512) -> None:
        """Initialize vLLM instance for efficient inference"""
        try:
            self.llm = VLLM(
                model=VLLM_MODEL_PATH,
                trust_remote_code=True,
                max_new_tokens=max_new_tokens,
                temperature=0.1,
                gpu_memory_utilization=0.9,
            )
            logger.info(f"Initialized vLLM with model: {VLLM_MODEL_PATH}")
        except Exception as e:
            logger.error(f"LLM initialization failed: {str(e)}")
            raise

    def build_qa_chain(self) -> None:
        """Build retrieval QA chain with error handling"""
        try:
            if not self.vectorstore or not self.llm:
                raise ValueError("Vectorstore and LLM must be initialized first")
            self.qa_chain = RetrievalQA.from_chain_type(
                llm=self.llm,
                chain_type="stuff",
                retriever=self.vectorstore.as_retriever(search_kwargs={"k": 3}),
                return_source_documents=True
            )
            logger.info("QA chain built successfully")
        except Exception as e:
            logger.error(f"QA chain build failed: {str(e)}")
            raise

    def query(self, question: str) -> Dict:
        """Run query with error handling and logging"""
        try:
            if not self.qa_chain:
                raise ValueError("QA chain not initialized")
            result = self.qa_chain({"query": question})
            logger.info(f"Processed query: {question[:50]}...")
            return {
                "answer": result["result"],
                "sources": [doc.page_content for doc in result["source_documents"]]
            }
        except Exception as e:
            logger.error(f"Query failed: {str(e)}")
            return {"error": str(e)}

if __name__ == "__main__":
    try:
        pipeline = ProductionRAGPipeline()
        chunks = pipeline.load_and_split_docs()
        pipeline.init_vectorstore(chunks)
        pipeline.init_llm()
        pipeline.build_qa_chain()
        # Example query
        response = pipeline.query("What are the requirements for senior AI engineer roles in 2026?")
        print(f"Answer: {response['answer']}")
    except Exception as e:
        logger.critical(f"Pipeline failed to start: {str(e)}")
        sys.exit(1)

import os
import sys
import tensorflow as tf
import mlflow
import mlflow.tensorflow
import numpy as np
from tensorflow.keras import layers, models
from tensorflow.keras.datasets import cifar10
from tensorflow.keras.utils import to_categorical
import logging
from typing import Tuple, Dict

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

class LegacyImageClassifier:
    def __init__(self, run_name: str = "legacy-cifar10-training"):
        self.run_name = run_name
        self.model = None
        self.x_train = None
        self.y_train = None
        self.x_test = None
        self.y_test = None
        self.history = None

    def load_and_preprocess_data(self) -> None:
        """Load CIFAR-10 data and preprocess with error handling"""
        try:
            logger.info("Loading CIFAR-10 dataset")
            (x_train, y_train), (x_test, y_test) = cifar10.load_data()
            # Normalize pixel values
            x_train = x_train.astype("float32") / 255.0
            x_test = x_test.astype("float32") / 255.0
            # One-hot encode labels
            y_train = to_categorical(y_train, 10)
            y_test = to_categorical(y_test, 10)
            self.x_train, self.y_train = x_train, y_train
            self.x_test, self.y_test = x_test, y_test
            logger.info(f"Data loaded: Train {x_train.shape}, Test {x_test.shape}")
        except Exception as e:
            logger.error(f"Data loading failed: {str(e)}")
            raise

    def build_model(self, learning_rate: float = 0.001) -> None:
        """Build legacy CNN model with TensorFlow 2.16.1"""
        try:
            logger.info("Building CNN model")
            self.model = models.Sequential([
                layers.Conv2D(32, (3, 3), activation="relu", input_shape=(32, 32, 3)),
                layers.MaxPooling2D((2, 2)),
                layers.Conv2D(64, (3, 3), activation="relu"),
                layers.MaxPooling2D((2, 2)),
                layers.Conv2D(64, (3, 3), activation="relu"),
                layers.Flatten(),
                layers.Dense(64, activation="relu"),
                layers.Dense(10, activation="softmax")
            ])
            self.model.compile(
                optimizer=tf.keras.optimizers.Adam(learning_rate=learning_rate),
                loss="categorical_crossentropy",
                metrics=["accuracy"]
            )
            logger.info(f"Model built: {self.model.summary()}")
        except Exception as e:
            logger.error(f"Model build failed: {str(e)}")
            raise

    def train_model(self, epochs: int = 10, batch_size: int = 64) -> None:
        """Train model with MLflow tracking"""
        try:
            mlflow.set_experiment("legacy-image-classification")
            with mlflow.start_run(run_name=self.run_name):
                mlflow.tensorflow.autolog()
                logger.info(f"Training model for {epochs} epochs")
                self.history = self.model.fit(
                    self.x_train,
                    self.y_train,
                    epochs=epochs,
                    batch_size=batch_size,
                    validation_split=0.2,
                    verbose=1
                )
                # Evaluate model
                test_loss, test_acc = self.model.evaluate(self.x_test, self.y_test, verbose=0)
                mlflow.log_metric("test_accuracy", test_acc)
                mlflow.log_metric("test_loss", test_loss)
                logger.info(f"Test accuracy: {test_acc:.4f}, Test loss: {test_loss:.4f}")
        except Exception as e:
            logger.error(f"Training failed: {str(e)}")
            raise

    def save_model(self, path: str = "./legacy-model") -> None:
        """Save model to disk with error handling"""
        try:
            if not self.model:
                raise ValueError("Model not trained yet")
            os.makedirs(path, exist_ok=True)
            self.model.save(path)
            mlflow.tensorflow.log_model(self.model, "model")
            logger.info(f"Model saved to {path}")
        except Exception as e:
            logger.error(f"Model save failed: {str(e)}")
            raise

    def evaluate_model(self) -> Dict:
        """Evaluate model and return metrics"""
        try:
            if not self.model:
                raise ValueError("Model not trained")
            test_loss, test_acc = self.model.evaluate(self.x_test, self.y_test, verbose=0)
            predictions = self.model.predict(self.x_test)
            return {
                "test_accuracy": float(test_acc),
                "test_loss": float(test_loss),
                "predictions_shape": predictions.shape
            }
        except Exception as e:
            logger.error(f"Evaluation failed: {str(e)}")
            raise

if __name__ == "__main__":
    try:
        classifier = LegacyImageClassifier()
        classifier.load_and_preprocess_data()
        classifier.build_model()
        classifier.train_model(epochs=10, batch_size=64)
        classifier.save_model()
        metrics = classifier.evaluate_model()
        print(f"Final metrics: {metrics}")
    except Exception as e:
        logger.critical(f"Pipeline failed: {str(e)}")
        sys.exit(1)

import time
import numpy as np
import logging
from typing import List, Dict
from vllm import LLM, SamplingParams
import tensorflow as tf
import tfx
from tfx.components import InfraValidator
import os
import sys

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

class InferenceBenchmark:
    def __init__(self, prompt: str = "Explain how to become a senior AI engineer in 2026", num_runs: int = 100):
        self.prompt = prompt
        self.num_runs = num_runs
        self.vllm_latencies = []
        self.tfx_latencies = []
        self.vllm_model = None
        self.tfx_model = None

    def init_vllm(self, model_path: str = "meta-llama/Llama-3.1-8B-Instruct") -> None:
        """Initialize vLLM instance for benchmarking"""
        try:
            logger.info(f"Initializing vLLM with model: {model_path}")
            self.vllm_model = LLM(
                model=model_path,
                gpu_memory_utilization=0.9,
                trust_remote_code=True
            )
            self.sampling_params = SamplingParams(
                temperature=0.1,
                max_tokens=512,
                top_p=0.95
            )
            logger.info("vLLM initialized successfully")
        except Exception as e:
            logger.error(f"vLLM initialization failed: {str(e)}")
            raise

    def init_tfx(self, model_path: str = "./legacy-tfx-model") -> None:
        """Initialize TFX-served model for benchmarking (TensorFlow 2.16.1)"""
        try:
            logger.info(f"Loading TFX model from {model_path}")
            self.tfx_model = tf.saved_model.load(model_path)
            logger.info("TFX model loaded successfully")
        except Exception as e:
            logger.error(f"TFX initialization failed: {str(e)}")
            raise

    def run_vllm_benchmark(self) -> List[float]:
        """Run vLLM inference benchmark with error handling"""
        try:
            if not self.vllm_model:
                raise ValueError("vLLM model not initialized")
            logger.info(f"Running vLLM benchmark: {self.num_runs} runs")
            for i in range(self.num_runs):
                start = time.perf_counter()
                outputs = self.vllm_model.generate([self.prompt], self.sampling_params)
                end = time.perf_counter()
                latency = (end - start) * 1000  # ms
                self.vllm_latencies.append(latency)
                if i % 10 == 0:
                    logger.info(f"vLLM run {i}: {latency:.2f}ms")
            return self.vllm_latencies
        except Exception as e:
            logger.error(f"vLLM benchmark failed: {str(e)}")
            raise

    def run_tfx_benchmark(self) -> List[float]:
        """Run TFX inference benchmark with error handling"""
        try:
            if not self.tfx_model:
                raise ValueError("TFX model not initialized")
            logger.info(f"Running TFX benchmark: {self.num_runs} runs")
            # Preprocess input for TFX model (simplified for example)
            input_tensor = tf.constant([self.prompt])
            for i in range(self.num_runs):
                start = time.perf_counter()
                _ = self.tfx_model(input_tensor)
                end = time.perf_counter()
                latency = (end - start) * 1000  # ms
                self.tfx_latencies.append(latency)
                if i % 10 == 0:
                    logger.info(f"TFX run {i}: {latency:.2f}ms")
            return self.tfx_latencies
        except Exception as e:
            logger.error(f"TFX benchmark failed: {str(e)}")
            raise

    def calculate_stats(self, latencies: List[float]) -> Dict:
        """Calculate benchmark statistics"""
        try:
            return {
                "p50_latency_ms": float(np.percentile(latencies, 50)),
                "p99_latency_ms": float(np.percentile(latencies, 99)),
                "avg_latency_ms": float(np.mean(latencies)),
                "std_latency_ms": float(np.std(latencies)),
                "min_latency_ms": float(np.min(latencies)),
                "max_latency_ms": float(np.max(latencies))
            }
        except Exception as e:
            logger.error(f"Stats calculation failed: {str(e)}")
            raise

    def print_results(self) -> None:
        """Print benchmark results"""
        try:
            vllm_stats = self.calculate_stats(self.vllm_latencies)
            tfx_stats = self.calculate_stats(self.tfx_latencies)
            print("\n=== Benchmark Results ===")
            print(f"Prompt: {self.prompt}")
            print(f"Number of runs: {self.num_runs}")
            print(f"Hardware: NVIDIA A100 80GB, Driver 535.154.05, CUDA 12.1")
            print(f"Software: vLLM 0.4.0, TensorFlow 2.16.1, TFX 1.12.0")
            print("\n--- vLLM Stats ---")
            for k, v in vllm_stats.items():
                print(f"{k}: {v:.2f}")
            print("\n--- TFX Stats ---")
            for k, v in tfx_stats.items():
                print(f"{k}: {v:.2f}")
            print(f"\nvLLM is {tfx_stats['p99_latency_ms'] / vllm_stats['p99_latency_ms']:.2f}x faster at p99 latency")
        except Exception as e:
            logger.error(f"Result printing failed: {str(e)}")
            raise

if __name__ == "__main__":
    try:
        benchmark = InferenceBenchmark(num_runs=100)
        benchmark.init_vllm()
        benchmark.init_tfx()
        benchmark.run_vllm_benchmark()
        benchmark.run_tfx_benchmark()
        benchmark.print_results()
    except Exception as e:
        logger.critical(f"Benchmark failed: {str(e)}")
        sys.exit(1)

import streamlit as st
from transformers import pipeline

@st.cache_resource
def load_model():
    return pipeline("sentiment-analysis", model="distilbert-base-uncased-finetuned-sst-2-english")

model = load_model()
st.title("Production Sentiment Analyzer")
user_input = st.text_area("Enter text to analyze:")
if st.button("Analyze"):
    result = model(user_input)[0]
    st.write(f"Sentiment: {result['label']}, Confidence: {result['score']:.2f}")

import pytest
from production_rag import ProductionRAGPipeline
import time

@pytest.fixture
def pipeline():
    return ProductionRAGPipeline()

def test_rag_latency(pipeline):
    start = time.perf_counter()
    response = pipeline.query("What is RAG?")
    latency = (time.perf_counter() - start) * 1000
    assert latency < 200  # p99 < 200ms
    assert "error" not in response

## Description
Fixes #1234 where vLLM crashes when using trust_remote_code=True with quantized models.

## Changes
- Added validation for quantized model configs
- Added unit test for quantized model loading
- Updated documentation to note quantized model requirements

## Benchmark Results
- No regression in latency for Llama-3.1-8B-Instruct
- Quantized model load time reduced by 12%