Comprehensive comparison for Observability technology in AI applications

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model, TaskType, PeftModel
from datasets import load_dataset
from trl import SFTTrainer
import logging
import os
from typing import Optional

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

class ProductReviewFineTuner:
    """Fine-tune LLM for product review generation using PEFT LoRA."""
    
    def __init__(self, model_name: str = "meta-llama/Llama-2-7b-hf", output_dir: str = "./lora_model"):
        self.model_name = model_name
        self.output_dir = output_dir
        self.model = None
        self.tokenizer = None
        
    def setup_model(self) -> None:
        """Initialize base model and tokenizer with error handling."""
        try:
            logger.info(f"Loading model: {self.model_name}")
            self.tokenizer = AutoTokenizer.from_pretrained(self.model_name, trust_remote_code=True)
            self.tokenizer.pad_token = self.tokenizer.eos_token
            
            self.model = AutoModelForCausalLM.from_pretrained(
                self.model_name,
                torch_dtype=torch.float16,
                device_map="auto",
                trust_remote_code=True
            )
            
            lora_config = LoraConfig(
                r=16,
                lora_alpha=32,
                target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
                lora_dropout=0.05,
                bias="none",
                task_type=TaskType.CAUSAL_LM
            )
            
            self.model = get_peft_model(self.model, lora_config)
            self.model.print_trainable_parameters()
            logger.info("Model setup complete")
            
        except Exception as e:
            logger.error(f"Model setup failed: {str(e)}")
            raise
    
    def train(self, dataset_name: str = "amazon_polarity", max_steps: int = 500) -> None:
        """Train model with PEFT using product review dataset."""
        try:
            dataset = load_dataset(dataset_name, split="train[:5000]")
            
            def format_prompt(example):
                return f"Review: {example['content']}\nRating: {example['label']}"
            
            training_args = TrainingArguments(
                output_dir=self.output_dir,
                per_device_train_batch_size=4,
                gradient_accumulation_steps=4,
                learning_rate=2e-4,
                logging_steps=10,
                max_steps=max_steps,
                save_steps=100,
                fp16=True,
                optim="paged_adamw_8bit"
            )
            
            trainer = SFTTrainer(
                model=self.model,
                train_dataset=dataset,
                formatting_func=format_prompt,
                args=training_args,
                max_seq_length=512
            )
            
            logger.info("Starting training...")
            trainer.train()
            trainer.save_model()
            logger.info(f"Training complete. Model saved to {self.output_dir}")
            
        except Exception as e:
            logger.error(f"Training failed: {str(e)}")
            raise
    
    def load_and_inference(self, prompt: str, adapter_path: Optional[str] = None) -> str:
        """Load trained adapter and generate review."""
        try:
            adapter_path = adapter_path or self.output_dir
            
            if not os.path.exists(adapter_path):
                raise FileNotFoundError(f"Adapter not found at {adapter_path}")
            
            base_model = AutoModelForCausalLM.from_pretrained(
                self.model_name,
                torch_dtype=torch.float16,
                device_map="auto"
            )
            
            model = PeftModel.from_pretrained(base_model, adapter_path)
            model.eval()
            
            inputs = self.tokenizer(prompt, return_tensors="pt").to(model.device)
            
            with torch.no_grad():
                outputs = model.generate(
                    **inputs,
                    max_new_tokens=100,
                    temperature=0.7,
                    do_sample=True
                )
            
            response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
            return response
            
        except Exception as e:
            logger.error(f"Inference failed: {str(e)}")
            raise

if __name__ == "__main__":
    finetuner = ProductReviewFineTuner()
    finetuner.setup_model()
    finetuner.train()
    result = finetuner.load_and_inference("Generate a positive review for a laptop:")
    print(result)