Comprehensive comparison for AI technology in Fine-tuning applications

import baseten
import os
import time
from typing import Dict, List, Optional
import json

# Initialize Baseten client with API key
baseten.login(os.environ.get("BASETEN_API_KEY"))

class CustomerSupportFineTuner:
    """
    Production-ready fine-tuning pipeline for customer support chatbot.
    Trains a model on company-specific support tickets and responses.
    """
    
    def __init__(self, base_model: str = "mistralai/Mistral-7B-v0.1"):
        self.base_model = base_model
        self.fine_tuned_model_id = None
        
    def prepare_training_data(self, tickets: List[Dict]) -> str:
        """
        Format customer support tickets into JSONL training format.
        """
        training_data = []
        
        for ticket in tickets:
            if not ticket.get("question") or not ticket.get("answer"):
                continue
                
            formatted_entry = {
                "messages": [
                    {"role": "system", "content": "You are a helpful customer support agent for TechCorp."},
                    {"role": "user", "content": ticket["question"]},
                    {"role": "assistant", "content": ticket["answer"]}
                ]
            }
            training_data.append(json.dumps(formatted_entry))
        
        # Write to temporary JSONL file
        training_file = "/tmp/support_training.jsonl"
        with open(training_file, "w") as f:
            f.write("\n".join(training_data))
            
        return training_file
    
    def start_fine_tuning(self, training_file: str, validation_file: Optional[str] = None) -> str:
        """
        Initiate fine-tuning job on Baseten with error handling.
        """
        try:
            # Upload training dataset
            print("Uploading training data...")
            train_dataset = baseten.upload_dataset(training_file)
            
            val_dataset = None
            if validation_file:
                val_dataset = baseten.upload_dataset(validation_file)
            
            # Configure fine-tuning parameters
            config = {
                "base_model": self.base_model,
                "training_dataset": train_dataset.id,
                "validation_dataset": val_dataset.id if val_dataset else None,
                "hyperparameters": {
                    "learning_rate": 2e-5,
                    "num_epochs": 3,
                    "batch_size": 4,
                    "warmup_steps": 100
                },
                "model_name": "customer-support-v1"
            }
            
            # Start fine-tuning job
            print(f"Starting fine-tuning job for {self.base_model}...")
            job = baseten.fine_tune(**config)
            
            return job.id
            
        except baseten.errors.InvalidDatasetError as e:
            print(f"Dataset validation failed: {e}")
            raise
        except baseten.errors.QuotaExceededError as e:
            print(f"Quota exceeded: {e}")
            raise
        except Exception as e:
            print(f"Unexpected error during fine-tuning: {e}")
            raise
    
    def monitor_training(self, job_id: str, poll_interval: int = 30) -> Dict:
        """
        Monitor fine-tuning progress with status updates.
        """
        print(f"Monitoring job {job_id}...")
        
        while True:
            try:
                job = baseten.get_fine_tune_job(job_id)
                status = job.status
                
                print(f"Status: {status} | Progress: {job.progress}%")
                
                if status == "completed":
                    self.fine_tuned_model_id = job.model_id
                    print(f"Fine-tuning completed! Model ID: {self.fine_tuned_model_id}")
                    return {
                        "status": "success",
                        "model_id": self.fine_tuned_model_id,
                        "metrics": job.metrics
                    }
                    
                elif status == "failed":
                    print(f"Fine-tuning failed: {job.error_message}")
                    return {"status": "failed", "error": job.error_message}
                    
                elif status == "cancelled":
                    print("Fine-tuning was cancelled")
                    return {"status": "cancelled"}
                    
                time.sleep(poll_interval)
                
            except Exception as e:
                print(f"Error monitoring job: {e}")
                time.sleep(poll_interval)
    
    def deploy_model(self) -> str:
        """
        Deploy fine-tuned model to production endpoint.
        """
        if not self.fine_tuned_model_id:
            raise ValueError("No fine-tuned model available for deployment")
            
        try:
            deployment = baseten.deploy(
                model_id=self.fine_tuned_model_id,
                name="customer-support-production",
                autoscaling_config={
                    "min_replicas": 1,
                    "max_replicas": 10,
                    "target_concurrency": 5
                }
            )
            
            print(f"Model deployed successfully! Endpoint: {deployment.endpoint}")
            return deployment.endpoint
            
        except Exception as e:
            print(f"Deployment failed: {e}")
            raise

# Example usage
if __name__ == "__main__":
    # Sample customer support data
    support_tickets = [
        {"question": "How do I reset my password?", "answer": "Visit account.techcorp.com/reset and enter your email."},
        {"question": "What's your refund policy?", "answer": "We offer 30-day money-back guarantee on all products."},
        {"question": "How do I cancel my subscription?", "answer": "Go to Settings > Billing > Cancel Subscription."}
    ]
    
    tuner = CustomerSupportFineTuner()
    training_file = tuner.prepare_training_data(support_tickets)
    job_id = tuner.start_fine_tuning(training_file)
    result = tuner.monitor_training(job_id)
    
    if result["status"] == "success":
        endpoint = tuner.deploy_model()
        print(f"Production endpoint ready: {endpoint}")