{
  "instruction": "Summarize this customer review in one sentence",
  "input": "I bought this laptop last week and it's amazing...",
  "output": "Customer highly satisfied with laptop purchase, praising performance and battery life."
}

{
  "messages": [
    {"role": "user", "content": "What's your return policy?"},
    {"role": "assistant", "content": "We offer 30-day returns..."},
    {"role": "user", "content": "What about opened items?"},
    {"role": "assistant", "content": "Opened items can be returned..."}
  ]
}

{
  "text": "Technical documentation explaining API authentication..."
}

import pandas as pd
from collections import Counter

def assess_data_quality(dataset):
    """Comprehensive data quality assessment."""
    
    # Check for duplicates
    duplicates = dataset.duplicated().sum()
    print(f"Duplicates: {duplicates} ({duplicates/len(dataset)*100:.2f}%)")
    
    # Check for missing values
    missing = dataset.isnull().sum()
    print(f"Missing values:\n{missing}")
    
    # Check length distribution
    lengths = dataset['text'].str.len()
    print(f"Length stats:\n{lengths.describe()}")
    
    # Check for outliers (very short or very long examples)
    too_short = (lengths < 10).sum()
    too_long = (lengths > 10000).sum()
    print(f"Too short (<10 chars): {too_short}")
    print(f"Too long (>10k chars): {too_long}")
    
    # Check label distribution (for classification)
    if 'label' in dataset.columns:
        label_dist = Counter(dataset['label'])
        print(f"Label distribution: {label_dist}")
        
        # Warn about imbalance
        max_count = max(label_dist.values())
        min_count = min(label_dist.values())
        if max_count / min_count > 10:
            print("⚠️  Warning: Severe class imbalance detected")
    
    return {
        'duplicates': duplicates,
        'missing': missing.sum(),
        'length_stats': lengths.describe(),
        'outliers': too_short + too_long
    }

def clean_dataset(df):
    """Standard cleaning pipeline."""
    
    # Remove duplicates
    df = df.drop_duplicates(subset=['text'])
    
    # Remove null values
    df = df.dropna(subset=['text', 'label'])
    
    # Remove extremely short examples
    df = df[df['text'].str.len() >= 20]
    
    # Remove extremely long examples (or truncate)
    df = df[df['text'].str.len() <= 8000]
    
    # Normalize whitespace
    df['text'] = df['text'].str.strip()
    df['text'] = df['text'].str.replace(r'\s+', ' ', regex=True)
    
    # Remove special characters if needed (task-dependent)
    # df['text'] = df['text'].str.replace(r'[^\w\s]', '', regex=True)
    
    # Standardize labels
    df['label'] = df['label'].str.lower().str.strip()
    
    return df

{
  "prompt": "### Instruction:\nTranslate to French\n\n### Input:\nHello, how are you?\n\n### Response:\n",
  "completion": "Bonjour, comment allez-vous?"
}

{
  "messages": [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "What's the weather like?"},
    {"role": "assistant", "content": "I don't have access to real-time weather data."}
  ]
}

{
  "instruction": "Write a haiku about programming",
  "input": "",
  "output": "Code flows like water\nBugs emerge from the shadows\nDebug until dawn"
}

def convert_to_training_format(data, format_type="alpaca"):
    """Convert raw data to training format."""
    
    if format_type == "alpaca":
        formatted = []
        for item in data:
            formatted.append({
                "instruction": item['task'],
                "input": item.get('context', ''),
                "output": item['response']
            })
        return formatted
    
    elif format_type == "chat":
        formatted = []
        for item in data:
            formatted.append({
                "messages": [
                    {"role": "system", "content": "You are a helpful assistant."},
                    {"role": "user", "content": item['question']},
                    {"role": "assistant", "content": item['answer']}
                ]
            })
        return formatted
    
    return data

from sklearn.model_selection import train_test_split

# Standard split: 80% train, 10% validation, 10% test
train_data, temp_data = train_test_split(dataset, test_size=0.2, random_state=42)
val_data, test_data = train_test_split(temp_data, test_size=0.5, random_state=42)

print(f"Train: {len(train_data)} examples")
print(f"Validation: {len(val_data)} examples")
print(f"Test: {len(test_data)} examples")

import nlpaug.augmenter.word as naw
import nlpaug.augmenter.sentence as nas

# Synonym replacement
aug_synonym = naw.SynonymAug(aug_src='wordnet')
augmented_text = aug_synonym.augment(original_text)

# Back-translation (translate to another language and back)
aug_back_translation = naw.BackTranslationAug(
    from_model_name='facebook/wmt19-en-de',
    to_model_name='facebook/wmt19-de-en'
)
augmented_text = aug_back_translation.augment(original_text)

# Paraphrasing with LLMs
def paraphrase_with_llm(text, num_variations=3):
    prompt = f"Paraphrase the following text in {num_variations} different ways:\n\n{text}"
    # Call your LLM API
    return variations

from transformers import AutoModelForCausalLM, TrainingArguments, Trainer

model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b-hf")

training_args = TrainingArguments(
    output_dir="./results",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    learning_rate=2e-5,
    # All parameters are trainable
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=val_dataset
)

trainer.train()

from peft import LoraConfig, get_peft_model, TaskType

# Configure LoRA
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=8,  # Rank of adaptation matrices
    lora_alpha=32,  # Scaling factor
    lora_dropout=0.1,
    target_modules=["q_proj", "v_proj"]  # Which layers to adapt
)

# Wrap model with LoRA
model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b-hf")
model = get_peft_model(model, lora_config)

# Only ~1% of parameters are trainable
model.print_trainable_parameters()
# Output: trainable params: 4,194,304 || all params: 6,738,415,616 || trainable%: 0.06

from transformers import BitsAndBytesConfig
import torch

# Configure 4-bit quantization
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.float16,
    bnb_4bit_use_double_quant=True
)

# Load model in 4-bit
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-hf",
    quantization_config=bnb_config,
    device_map="auto"
)

# Apply LoRA on top
model = get_peft_model(model, lora_config)

training_args = TrainingArguments(
    learning_rate=2e-5,  # Conservative starting point
    # Or use a scheduler
    lr_scheduler_type="cosine",
    warmup_steps=100
)

from transformers import Trainer

class LRFinderTrainer(Trainer):
    def find_lr(self, start_lr=1e-7, end_lr=1, num_steps=100):
        lrs = []
        losses = []
        
        for lr in np.logspace(np.log10(start_lr), np.log10(end_lr), num_steps):
            self.optimizer.param_groups[0]['lr'] = lr
            loss = self.training_step(...)
            lrs.append(lr)
            losses.append(loss)
        
        # Plot and find the steepest descent
        import matplotlib.pyplot as plt
        plt.plot(lrs, losses)
        plt.xscale('log')
        plt.xlabel('Learning Rate')
        plt.ylabel('Loss')
        plt.show()

training_args = TrainingArguments(
    per_device_train_batch_size=4,  # Per GPU
    gradient_accumulation_steps=4,   # Effective batch size = 4 * 4 = 16
)

training_args = TrainingArguments(
    num_train_epochs=3,  # Common starting point
    # Or specify max steps
    max_steps=1000
)

training_args = TrainingArguments(
    # Regularization
    weight_decay=0.01,  # L2 regularization
    dropout=0.1,        # Dropout rate (if configurable)
    
    # Optimization
    adam_beta1=0.9,
    adam_beta2=0.999,
    adam_epsilon=1e-8,
    max_grad_norm=1.0,  # Gradient clipping
    
    # Evaluation
    evaluation_strategy="steps",
    eval_steps=100,
    save_strategy="steps",
    save_steps=100,
    load_best_model_at_end=True,
    metric_for_best_model="eval_loss"
)

# Mixed precision training (FP16/BF16)
training_args = TrainingArguments(
    fp16=True,  # For older GPUs
    # or
    bf16=True,  # For Ampere+ GPUs (better numerical stability)
)

# Gradient checkpointing (trade compute for memory)
model.gradient_checkpointing_enable()

# Flash Attention 2 (faster attention computation)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    use_flash_attention_2=True
)

# DeepSpeed for multi-GPU training
training_args = TrainingArguments(
    deepspeed="ds_config.json"
)

{
  "train_batch_size": 32,
  "gradient_accumulation_steps": 4,
  "optimizer": {
    "type": "AdamW",
    "params": {
      "lr": 2e-5
    }
  },
  "fp16": {
    "enabled": true
  },
  "zero_optimization": {
    "stage": 2,
    "offload_optimizer": {
      "device": "cpu"
    }
  }
}

from transformers import TrainerCallback

class DetailedLoggingCallback(TrainerCallback):
    def on_log(self, args, state, control, logs=None, **kwargs):
        if logs:
            print(f"Step {state.global_step}:")
            print(f"  Training Loss: {logs.get('loss', 'N/A'):.4f}")
            print(f"  Learning Rate: {logs.get('learning_rate', 'N/A'):.2e}")
            
            if 'eval_loss' in logs:
                print(f"  Validation Loss: {logs['eval_loss']:.4f}")
                
                # Check for overfitting
                train_loss = logs.get('loss', float('inf'))
                val_loss = logs['eval_loss']
                
                if val_loss > train_loss * 1.2:
                    print("  ⚠️  Warning: Possible overfitting detected")

trainer = Trainer(
    model=model,
    args=training_args,
    callbacks=[DetailedLoggingCallback()]
)

# Early stopping
from transformers import EarlyStoppingCallback

early_stopping = EarlyStoppingCallback(
    early_stopping_patience=3,  # Stop if no improvement for 3 evaluations
    early_stopping_threshold=0.01  # Minimum improvement threshold
)

training_args = TrainingArguments(
    # Regularization techniques
    weight_decay=0.01,
    dropout=0.1,
    
    # Data augmentation (if applicable)
    # More training data
    
    # Reduce model capacity (for LoRA)
    # lora_r=4 instead of 8
    
    # Fewer epochs
    num_train_epochs=3,
    
    # Evaluation and checkpointing
    evaluation_strategy="steps",
    eval_steps=50,
    save_total_limit=3,  # Keep only best 3 checkpoints
    load_best_model_at_end=True
)

import wandb

wandb.init(project="my-fine-tuning", name="llama-7b-lora")

training_args = TrainingArguments(
    report_to="wandb",
    logging_steps=10
)

# Automatically logs:
# - Training/validation loss
# - Learning rate schedule
# - Gradient norms
# - System metrics (GPU usage, memory)

from datasets import load_metric

def compute_metrics(eval_pred):
    """Compute metrics during training."""
    predictions, labels = eval_pred
    
    # For generation tasks
    decoded_preds = tokenizer.batch_decode(predictions, skip_special_tokens=True)
    decoded_labels = tokenizer.batch_decode(labels, skip_special_tokens=True)
    
    # ROUGE for summarization
    rouge = load_metric("rouge")
    rouge_scores = rouge.compute(
        predictions=decoded_preds,
        references=decoded_labels
    )
    
    # BLEU for translation
    bleu = load_metric("bleu")
    bleu_score = bleu.compute(
        predictions=decoded_preds,
        references=decoded_labels
    )
    
    return {
        "rouge1": rouge_scores["rouge1"].mid.fmeasure,
        "rouge2": rouge_scores["rouge2"].mid.fmeasure,
        "rougeL": rouge_scores["rougeL"].mid.fmeasure,
        "bleu": bleu_score["bleu"]
    }

trainer = Trainer(
    model=model,
    args=training_args,
    compute_metrics=compute_metrics
)

# For classification
from sklearn.metrics import accuracy_score, f1_score, precision_recall_fscore_support

def compute_classification_metrics(eval_pred):
    predictions, labels = eval_pred
    predictions = np.argmax(predictions, axis=1)
    
    accuracy = accuracy_score(labels, predictions)
    precision, recall, f1, _ = precision_recall_fscore_support(
        labels, predictions, average='weighted'
    )
    
    return {
        "accuracy": accuracy,
        "f1": f1,
        "precision": precision,
        "recall": recall
    }

# For question answering
def compute_qa_metrics(eval_pred):
    predictions, labels = eval_pred
    
    # Exact match
    exact_matches = sum(p == l for p, l in zip(predictions, labels))
    exact_match = exact_matches / len(predictions)
    
    # F1 score (token overlap)
    f1_scores = [compute_f1(p, l) for p, l in zip(predictions, labels)]
    f1 = sum(f1_scores) / len(f1_scores)
    
    return {
        "exact_match": exact_match,
        "f1": f1
    }

def evaluate_samples(model, tokenizer, test_samples, num_samples=10):
    """Generate and review sample outputs."""
    
    for i, sample in enumerate(test_samples[:num_samples]):
        print(f"\n{'='*80}")
        print(f"Sample {i+1}")
        print(f"{'='*80}")
        print(f"Input: {sample['input']}")
        print(f"\nExpected: {sample['output']}")
        
        # Generate prediction
        inputs = tokenizer(sample['input'], return_tensors="pt").to(model.device)
        outputs = model.generate(**inputs, max_new_tokens=256)
        prediction = tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        print(f"\nPredicted: {prediction}")
        print(f"{'='*80}")

# Run after each checkpoint
evaluate_samples(model, tokenizer, test_dataset)

# Estimate training cost
def estimate_training_cost(
    num_examples,
    batch_size,
    num_epochs,
    gpu_cost_per_hour,
    examples_per_second
):
    """Estimate total training cost."""
    
    total_examples = num_examples * num_epochs
    total_seconds = total_examples / examples_per_second
    total_hours = total_seconds / 3600
    
    total_cost = total_hours * gpu_cost_per_hour
    
    print(f"Estimated training time: {total_hours:.2f} hours")
    print(f"Estimated cost: ${total_cost:.2f}")
    
    return total_cost

# Example: 10k examples, 3 epochs, A100 GPU
estimate_training_cost(
    num_examples=10000,
    batch_size=4,
    num_epochs=3,
    gpu_cost_per_hour=2.50,  # A100 spot instance
    examples_per_second=2
)

# Estimate data generation cost
def estimate_data_generation_cost(
    num_examples,
    tokens_per_example,
    cost_per_1k_tokens
):
    """Estimate cost of generating training data with GPT-4."""
    
    total_tokens = num_examples * tokens_per_example
    total_cost = (total_tokens / 1000) * cost_per_1k_tokens
    
    print(f"Total tokens: {total_tokens:,}")
    print(f"Estimated cost: ${total_cost:.2f}")
    
    return total_cost

# Example: Generate 5k examples with GPT-4
estimate_data_generation_cost(
    num_examples=5000,
    tokens_per_example=500,  # Input + output
    cost_per_1k_tokens=0.03  # GPT-4 pricing
)
# Output: Estimated cost: $75.00

from transformers import AutoModelForCausalLM, BitsAndBytesConfig

# 8-bit quantization (2-3x smaller, minimal quality loss)
model = AutoModelForCausalLM.from_pretrained(
    "your-fine-tuned-model",
    load_in_8bit=True,
    device_map="auto"
)

# 4-bit quantization (4x smaller, slight quality loss)
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.float16
)

model = AutoModelForCausalLM.from_pretrained(
    "your-fine-tuned-model",
    quantization_config=bnb_config,
    device_map="auto"
)

# Convert to GGUF for llama.cpp (CPU/Metal inference)
python convert-hf-to-gguf.py \
    --model-dir ./your-fine-tuned-model \
    --outfile model.gguf \
    --outtype q4_k_m  # 4-bit quantization

# Run inference with llama.cpp
./llama-cli -m model.gguf -p "Your prompt here"

from optimum.onnxruntime import ORTModelForCausalLM

# Convert to ONNX
model = ORTModelForCausalLM.from_pretrained(
    "your-fine-tuned-model",
    export=True
)

# Save optimized model
model.save_pretrained("./onnx-model")

# Inference is faster and more portable

import torch.nn.utils.prune as prune

# Remove least important weights
for name, module in model.named_modules():
    if isinstance(module, torch.nn.Linear):
        prune.l1_unstructured(module, name='weight', amount=0.2)  # Remove 20%

# Make pruning permanent
for name, module in model.named_modules():
    if isinstance(module, torch.nn.Linear):
        prune.remove(module, 'weight')

from sagemaker.huggingface import HuggingFaceModel

# Deploy to SageMaker
huggingface_model = HuggingFaceModel(
    model_data="s3://your-bucket/model.tar.gz",
    role=role,
    transformers_version="4.26",
    pytorch_version="1.13",
    py_version="py39"
)

predictor = huggingface_model.deploy(
    initial_instance_count=1,
    instance_type="ml.g5.xlarge"  # GPU instance
)

# Inference
result = predictor.predict({
    "inputs": "Your prompt here"
})

from google.cloud import aiplatform

# Deploy model
model = aiplatform.Model.upload(
    display_name="fine-tuned-llm",
    artifact_uri="gs://your-bucket/model",
    serving_container_image_uri="us-docker.pkg.dev/vertex-ai/prediction/pytorch-gpu:latest"
)

endpoint = model.deploy(
    machine_type="n1-standard-4",
    accelerator_type="NVIDIA_TESLA_T4",
    accelerator_count=1
)

from vllm import LLM, SamplingParams

# Initialize vLLM
llm = LLM(
    model="your-fine-tuned-model",
    tensor_parallel_size=1,  # Number of GPUs
    dtype="float16",
    max_model_len=4096
)

# Sampling parameters
sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=256
)

# Batch inference (very efficient)
prompts = ["Prompt 1", "Prompt 2", "Prompt 3"]
outputs = llm.generate(prompts, sampling_params)

for output in outputs:
    print(output.outputs[0].text)

from fastapi import FastAPI
from pydantic import BaseModel
from vllm import LLM, SamplingParams

app = FastAPI()
llm = LLM(model="your-fine-tuned-model")

class GenerateRequest(BaseModel):
    prompt: str
    max_tokens: int = 256
    temperature: float = 0.7

@app.post("/generate")
async def generate(request: GenerateRequest):
    sampling_params = SamplingParams(
        temperature=request.temperature,
        max_tokens=request.max_tokens
    )
    
    outputs = llm.generate([request.prompt], sampling_params)
    
    return {
        "generated_text": outputs[0].outputs[0].text,
        "tokens_generated": len(outputs[0].outputs[0].token_ids)
    }

# Run with: uvicorn app:app --host 0.0.0.0 --port 8000

import modal

stub = modal.Stub("fine-tuned-llm")

@stub.function(
    gpu="A10G",
    image=modal.Image.debian_slim().pip_install(
        "transformers", "torch", "accelerate"
    ),
    timeout=300
)
def generate(prompt: str):
    from transformers import AutoModelForCausalLM, AutoTokenizer
    
    model = AutoModelForCausalLM.from_pretrained("your-model")
    tokenizer = AutoTokenizer.from_pretrained("your-model")
    
    inputs = tokenizer(prompt, return_tensors="pt")
    outputs = model.generate(**inputs, max_new_tokens=256)
    
    return tokenizer.decode(outputs[0], skip_special_tokens=True)

@stub.local_entrypoint()
def main():
    result = generate.remote("Your prompt here")
    print(result)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
import time

app = FastAPI(title="Fine-Tuned LLM API", version="1.0.0")

class GenerateRequest(BaseModel):
    prompt: str = Field(..., description="Input prompt")
    max_tokens: int = Field(256, ge=1, le=2048)
    temperature: float = Field(0.7, ge=0.0, le=2.0)
    top_p: float = Field(0.9, ge=0.0, le=1.0)
    stop_sequences: Optional[List[str]] = None

class GenerateResponse(BaseModel):
    generated_text: str
    tokens_generated: int
    latency_ms: float
    model_version: str

@app.post("/v1/generate", response_model=GenerateResponse)
async def generate(request: GenerateRequest):
    """Generate text from prompt."""
    
    start_time = time.time()
    
    try:
        # Your generation logic
        output = model.generate(
            request.prompt,
            max_tokens=request.max_tokens,
            temperature=request.temperature,
            top_p=request.top_p,
            stop=request.stop_sequences
        )
        
        latency = (time.time() - start_time) * 1000
        
        return GenerateResponse(
            generated_text=output.text,
            tokens_generated=output.num_tokens,
            latency_ms=latency,
            model_version="v1.0.0"
        )
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {"status": "healthy", "model_loaded": model is not None}

@app.get("/metrics")
async def metrics():
    """Prometheus-compatible metrics."""
    return {
        "requests_total": request_counter,
        "average_latency_ms": avg_latency,
        "errors_total": error_counter
    }

from fastapi.responses import StreamingResponse
import asyncio

@app.post("/v1/generate/stream")
async def generate_stream(request: GenerateRequest):
    """Stream generated tokens as they're produced."""
    
    async def token_generator():
        for token in model.generate_stream(request.prompt):
            yield f"data: {token}\n\n"
            await asyncio.sleep(0)  # Allow other tasks to run
    
    return StreamingResponse(
        token_generator(),
        media_type="text/event-stream"
    )

import requests

class LLMClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_key}"}
    
    def generate(self, prompt: str, **kwargs):
        """Generate text from prompt."""
        response = requests.post(
            f"{self.base_url}/v1/generate",
            json={"prompt": prompt, **kwargs},
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()
    
    def generate_stream(self, prompt: str, **kwargs):
        """Stream generated tokens."""
        response = requests.post(
            f"{self.base_url}/v1/generate/stream",
            json={"prompt": prompt, **kwargs},
            headers=self.headers,
            stream=True
        )
        
        for line in response.iter_lines():
            if line.startswith(b"data: "):
                yield line[6:].decode()

# Usage
client = LLMClient("https://api.example.com", "your-api-key")
result = client.generate("Write a poem about AI")
print(result["generated_text"])

from prometheus_client import Counter, Histogram, Gauge
import time

# Request metrics
request_counter = Counter(
    'llm_requests_total',
    'Total number of requests',
    ['endpoint', 'status']
)

request_latency = Histogram(
    'llm_request_latency_seconds',
    'Request latency in seconds',
    ['endpoint']
)

# Model metrics
tokens_generated = Counter(
    'llm_tokens_generated_total',
    'Total tokens generated'
)

active_requests = Gauge(
    'llm_active_requests',
    'Number of active requests'
)

# Error tracking
error_counter = Counter(
    'llm_errors_total',
    'Total number of errors',
    ['error_type']
)

# Usage in endpoint
@app.post("/v1/generate")
async def generate(request: GenerateRequest):
    active_requests.inc()
    start_time = time.time()
    
    try:
        output = model.generate(request.prompt)
        
        # Record metrics
        request_counter.labels(endpoint='generate', status='success').inc()
        tokens_generated.inc(output.num_tokens)
        
        return output
    
    except Exception as e:
        error_counter.labels(error_type=type(e).__name__).inc()
        request_counter.labels(endpoint='generate', status='error').inc()
        raise
    
    finally:
        latency = time.time() - start_time
        request_latency.labels(endpoint='generate').observe(latency)
        active_requests.dec()

from typing import List
import numpy as np

class QualityMonitor:
    def __init__(self):
        self.outputs = []
        self.scores = []
    
    def log_output(self, prompt: str, output: str, metadata: dict):
        """Log output for quality analysis."""
        self.outputs.append({
            'prompt': prompt,
            'output': output,
            'timestamp': time.time(),
            **metadata
        })
    
    def detect_anomalies(self):
        """Detect unusual outputs."""
        recent_outputs = self.outputs[-100:]
        
        # Check for repetition
        for output in recent_outputs:
            text = output['output']
            if self._has_excessive_repetition(text):
                self._alert('excessive_repetition', output)
        
        # Check for length anomalies
        lengths = [len(o['output']) for o in recent_outputs]
        mean_length = np.mean(lengths)
        std_length = np.std(lengths)
        
        for output in recent_outputs:
            length = len(output['output'])
            if abs(length - mean_length) > 3 * std_length:
                self._alert('length_anomaly', output)
    
    def _has_excessive_repetition(self, text: str, threshold: float = 0.3):
        """Check if text has excessive repetition."""
        words = text.split()
        if len(words) < 10:
            return False
        
        unique_ratio = len(set(words)) / len(words)
        return unique_ratio < threshold
    
    def _alert(self, alert_type: str, output: dict):
        """Send alert for quality issue."""
        print(f"⚠️  Quality Alert: {alert_type}")
        print(f"Prompt: {output['prompt'][:100]}...")
        print(f"Output: {output['output'][:100]}...")
        # Send to monitoring system (Slack, PagerDuty, etc.)

monitor = QualityMonitor()

@app.post("/v1/generate")
async def generate(request: GenerateRequest):
    output = model.generate(request.prompt)
    
    # Log for quality monitoring
    monitor.log_output(
        prompt=request.prompt,
        output=output.text,
        metadata={
            'temperature': request.temperature,
            'tokens': output.num_tokens
        }
    )
    
    return output

import random
from enum import Enum

class ModelVariant(Enum):
    CONTROL = "v1.0.0"
    TREATMENT = "v1.1.0"

class ABTestManager:
    def __init__(self, treatment_percentage: float = 0.1):
        self.treatment_percentage = treatment_percentage
        self.results = {'control': [], 'treatment': []}
    
    def get_variant(self, user_id: str) -> ModelVariant:
        """Consistently assign user to variant."""
        # Hash user_id for consistent assignment
        hash_val = hash(user_id) % 100
        
        if hash_val < self.treatment_percentage * 100:
            return ModelVariant.TREATMENT
        return ModelVariant.CONTROL
    
    def log_result(self, variant: ModelVariant, latency: float, quality_score: float):
        """Log result for analysis."""
        variant_key = 'treatment' if variant == ModelVariant.TREATMENT else 'control'
        self.results[variant_key].append({
            'latency': latency,
            'quality_score': quality_score,
            'timestamp': time.time()
        })
    
    def analyze_results(self):
        """Compare variants statistically."""
        control_latencies = [r['latency'] for r in self.results['control']]
        treatment_latencies = [r['latency'] for r in self.results['treatment']]
        
        print(f"Control - Mean latency: {np.mean(control_latencies):.3f}s")
        print(f"Treatment - Mean latency: {np.mean(treatment_latencies):.3f}s")
        
        # Statistical significance test
        from scipy import stats
        t_stat, p_value = stats.ttest_ind(control_latencies, treatment_latencies)
        print(f"P-value: {p_value:.4f}")

ab_test = ABTestManager(treatment_percentage=0.1)

@app.post("/v1/generate")
async def generate(request: GenerateRequest, user_id: str):
    variant = ab_test.get_variant(user_id)
    
    # Load appropriate model
    model = models[variant.value]
    
    start_time = time.time()
    output = model.generate(request.prompt)
    latency = time.time() - start_time
    
    # Log for A/B test
    ab_test.log_result(variant, latency, quality_score=0.9)  # Calculate actual score
    
    return output

import logging
from pythonjsonlogger import jsonlogger

# Structured logging
logger = logging.getLogger()
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

@app.post("/v1/generate")
async def generate(request: GenerateRequest):
    request_id = str(uuid.uuid4())
    
    logger.info("Request received", extra={
        'request_id': request_id,
        'prompt_length': len(request.prompt),
        'temperature': request.temperature
    })
    
    try:
        output = model.generate(request.prompt)
        
        logger.info("Request completed", extra={
            'request_id': request_id,
            'tokens_generated': output.num_tokens,
            'latency_ms': latency
        })
        
        return output
    
    except Exception as e:
        logger.error("Request failed", extra={
            'request_id': request_id,
            'error': str(e),
            'error_type': type(e).__name__
        })
        raise

import mlflow
import mlflow.pytorch

# During training
with mlflow.start_run():
    # Log parameters
    mlflow.log_params({
        'learning_rate': 2e-5,
        'batch_size': 4,
        'num_epochs': 3,
        'lora_r': 8
    })
    
    # Log metrics
    mlflow.log_metrics({
        'train_loss': 0.45,
        'eval_loss': 0.52,
        'eval_accuracy': 0.89
    })
    
    # Log model
    mlflow.pytorch.log_model(
        model,
        "model",
        registered_model_name="fine-tuned-llm"
    )
    
    # Log artifacts
    mlflow.log_artifact("training_config.json")
    mlflow.log_artifact("dataset_stats.json")

# Load specific version for deployment
model_uri = "models:/fine-tuned-llm/production"
model = mlflow.pytorch.load_model(model_uri)

import wandb

# Initialize run
run = wandb.init(
    project="llm-fine-tuning",
    name="llama-7b-lora-v3",
    config={
        'learning_rate': 2e-5,
        'batch_size': 4,
        'lora_r': 8
    }
)

# Log during training
wandb.log({
    'train_loss': loss,
    'eval_loss': eval_loss,
    'learning_rate': lr
})

# Save model
wandb.save('model.pt')

# Mark as production
run.tags = ['production', 'v1.2.0']

# Tag model versions
git tag -a v1.0.0 -m "Initial production model"
git push origin v1.0.0

# Store model metadata
cat > model_card.md << EOF
# Model: fine-tuned-llm-v1.0.0

## Training Details
- Base model: meta-llama/Llama-2-7b-hf
- Training data: 10,000 examples
- Training date: 2025-12-15
- Training duration: 4 hours
- Hardware: 1x A100 GPU

## Performance
- Validation loss: 0.52
- Accuracy: 89%
- ROUGE-L: 0.76

## Deployment
- Quantization: 8-bit
- Inference latency: 150ms (p95)
- Memory usage: 8GB
EOF

# Kubernetes deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-api
spec:
  replicas: 3  # Scale to 3 instances
  selector:
    matchLabels:
      app: llm-api
  template:
    metadata:
      labels:
        app: llm-api
    spec:
      containers:
      - name: llm-api
        image: your-registry/llm-api:v1.0.0
        resources:
          requests:
            memory: "16Gi"
            nvidia.com/gpu: 1
          limits:
            memory: "16Gi"
            nvidia.com/gpu: 1
        ports:
        - containerPort: 8000
---
apiVersion: v1
kind: Service
metadata:
  name: llm-api-service
spec:
  selector:
    app: llm-api
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: llm-api-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: llm-api
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Pods
    pods:
      metric:
        name: llm_active_requests
      target:
        type: AverageValue
        averageValue: "10"

from functools import lru_cache
import hashlib
import redis

# In-memory cache
@lru_cache(maxsize=1000)
def generate_cached(prompt: str, temperature: float):
    return model.generate(prompt, temperature=temperature)

# Redis cache for distributed systems
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def generate_with_redis_cache(prompt: str, temperature: float):
    # Create cache key
    cache_key = hashlib.md5(
        f"{prompt}:{temperature}".encode()
    ).hexdigest()
    
    # Check cache
    cached = redis_client.get(cache_key)
    if cached:
        return cached.decode()
    
    # Generate and cache
    output = model.generate(prompt, temperature=temperature)
    redis_client.setex(
        cache_key,
        3600,  # 1 hour TTL
        output
    )
    
    return output

import asyncio
from collections import deque

class BatchProcessor:
    def __init__(self, max_batch_size: int = 8, max_wait_ms: int = 100):
        self.max_batch_size = max_batch_size
        self.max_wait_ms = max_wait_ms
        self.queue = deque()
        self.processing = False
    
    async def add_request(self, prompt: str):
        """Add request to batch queue."""
        future = asyncio.Future()
        self.queue.append((prompt, future))
        
        if not self.processing:
            asyncio.create_task(self._process_batch())
        
        return await future
    
    async def _process_batch(self):
        """Process accumulated requests as batch."""
        self.processing = True
        await asyncio.sleep(self.max_wait_ms / 1000)
        
        # Collect batch
        batch = []
        futures = []
        
        while self.queue and len(batch) < self.max_batch_size:
            prompt, future = self.queue.popleft()
            batch.append(prompt)
            futures.append(future)
        
        if batch:
            # Process batch
            outputs = model.generate_batch(batch)
            
            # Return results
            for future, output in zip(futures, outputs):
                future.set_result(output)
        
        self.processing = False

batch_processor = BatchProcessor()

@app.post("/v1/generate")
async def generate(request: GenerateRequest):
    output = await batch_processor.add_request(request.prompt)
    return output