Enterprise LLM Fine-Tuning: Complete Guide to Custom AI Models for Business

This pipeline processes insurance claims data, ensuring quality and consistency before fine-tuning. Notice the validation steps and quality metrics. The key insight: we're not just processing text—we're building a system that understands business context, regulatory requirements, and decision-making patterns.

import pandas as pd
import json
from typing import List, Dict
from sklearn.model_selection import train_test_split

class EnterpriseDataProcessor:
    def __init__(self, min_quality_score: float = 0.8):
        self.min_quality_score = min_quality_score
        self.quality_metrics = {}
    
    def validate_data_quality(self, dataset: List[Dict]) -> Dict:
        """Validate data quality using multiple metrics"""
        quality_report = {
            'total_samples': len(dataset),
            'avg_length': 0,
            'consistency_score': 0,
            'coverage_score': 0
        }
        
        # Calculate average text length
        lengths = [len(item['text']) for item in dataset]
        quality_report['avg_length'] = sum(lengths) / len(lengths)
        
        # Check annotation consistency
        consistency_scores = self._calculate_consistency(dataset)
        quality_report['consistency_score'] = sum(consistency_scores) / len(consistency_scores)
        
        # Check domain coverage
        quality_report['coverage_score'] = self._calculate_coverage(dataset)
        
        return quality_report
    
    def create_instruction_dataset(self, raw_data: List[Dict]) -> List[Dict]:
        """Convert raw data to instruction-following format"""
        formatted_data = []
        
        for item in raw_data:
            # Create instruction-following format
            instruction = f"Analyze this insurance claim and determine: {item['task_description']}"
            
            formatted_item = {
                "instruction": instruction,
                "input": item['claim_text'],
                "output": item['analysis_result'],
                "metadata": {
                    "claim_type": item['claim_type'],
                    "complexity": item['complexity_score'],
                    "department": item['department']
                }
            }
            
            formatted_data.append(formatted_item)
        
        return formatted_data
    
    def split_dataset(self, data: List[Dict], test_size: float = 0.2):
        """Split data with stratification for balanced representation"""
        # Stratify by claim type to ensure balanced splits
        claim_types = [item['metadata']['claim_type'] for item in data]
        
        train_data, test_data = train_test_split(
            data, 
            test_size=test_size, 
            stratify=claim_types,
            random_state=42
        )
        
        return train_data, test_data

This framework helps you choose the right fine-tuning method based on your specific requirements. Most enterprise projects should start with LoRA.

def select_fine_tuning_method(
    task_complexity: str,
    data_size: int,
    latency_requirement: float,
    budget: float,
    infrastructure: str
) -> str:
    """
    Select the optimal fine-tuning method based on enterprise requirements
    """
    
    # Decision tree for method selection
    if data_size < 1000:
        return "PEFT"  # Small datasets benefit from parameter efficiency
    
    if latency_requirement < 0.5:  # Sub-second response needed
        if infrastructure == "edge":
            return "PEFT"
        else:
            return "LoRA"
    
    if task_complexity == "high" and budget > 1000:
        return "Full Fine-Tuning"  # Maximum performance needed
    
    if budget < 200:
        return "PEFT"  # Cost-constrained projects
    
    # Default recommendation for most enterprise use cases
    return "LoRA"

# Example usage for different enterprise scenarios
scenarios = [
    {
        "name": "Customer Support Chatbot",
        "task_complexity": "medium",
        "data_size": 5000,
        "latency_requirement": 0.3,
        "budget": 500,
        "infrastructure": "cloud",
        "recommended_method": "LoRA"
    },
    {
        "name": "Medical Document Analysis",
        "task_complexity": "high",
        "data_size": 15000,
        "latency_requirement": 2.0,
        "budget": 2000,
        "infrastructure": "on-premise",
        "recommended_method": "Full Fine-Tuning"
    },
    {
        "name": "Mobile App Assistant",
        "task_complexity": "low",
        "data_size": 2000,
        "latency_requirement": 0.2,
        "budget": 150,
        "infrastructure": "edge",
        "recommended_method": "PEFT"
    }
]

This is the complete fine-tuning pipeline used for enterprise clients. It includes data validation, model training, evaluation, and deployment preparation.

import torch
import transformers
from transformers import (
    AutoTokenizer, 
    AutoModelForCausalLM, 
    TrainingArguments, 
    Trainer,
    DataCollatorForLanguageModeling
)
from peft import LoraConfig, get_peft_model, TaskType
from datasets import Dataset
import wandb
from typing import Dict, List
import json

class EnterpriseFineTuner:
    def __init__(self, model_name: str, output_dir: str):
        self.model_name = model_name
        self.output_dir = output_dir
        self.tokenizer = None
        self.model = None
        self.training_args = None
        
    def setup_model_and_tokenizer(self):
        """Initialize model and tokenizer with enterprise-grade settings"""
        print(f"Loading {self.model_name}...")
        
        # Load tokenizer with proper settings
        self.tokenizer = AutoTokenizer.from_pretrained(
            self.model_name,
            trust_remote_code=True,
            padding_side="right"
        )
        
        # Add padding token if it doesn't exist
        if self.tokenizer.pad_token is None:
            self.tokenizer.pad_token = self.tokenizer.eos_token
        
        # Load model with memory optimization
        self.model = AutoModelForCausalLM.from_pretrained(
            self.model_name,
            torch_dtype=torch.float16,  # Use half precision for memory efficiency
            device_map="auto",
            trust_remote_code=True
        )
        
        # Resize token embeddings if needed
        self.model.resize_token_embeddings(len(self.tokenizer))
        
        print(f"Model loaded successfully. Parameters: {self.model.num_parameters():,}")
    
    def setup_lora_config(self, r: int = 16, alpha: int = 32):
        """Configure LoRA for parameter-efficient fine-tuning"""
        lora_config = LoraConfig(
            task_type=TaskType.CAUSAL_LM,
            r=r,  # Rank of adaptation
            lora_alpha=alpha,  # Scaling parameter
            lora_dropout=0.1,
            target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],  # Target attention layers
            bias="none"
        )
        
        self.model = get_peft_model(self.model, lora_config)
        self.model.print_trainable_parameters()
        
        return lora_config
    
    def prepare_dataset(self, data: List[Dict]) -> Dataset:
        """Prepare dataset for training with proper formatting"""
        def format_instruction(example):
            """Format data as instruction-following examples"""
            instruction = example['instruction']
            input_text = example.get('input', '')
            output = example['output']
            
            # Create instruction-following format
            if input_text:
                prompt = f"### Instruction:\n{instruction}\n\n### Input:\n{input_text}\n\n### Response:\n{output}"
            else:
                prompt = f"### Instruction:\n{instruction}\n\n### Response:\n{output}"
            
            return {"text": prompt}
        
        # Format all examples
        formatted_data = [format_instruction(example) for example in data]
        
        # Tokenize the dataset
        def tokenize_function(examples):
            return self.tokenizer(
                examples["text"],
                truncation=True,
                padding=True,
                max_length=512,  # Adjust based on your data
                return_tensors="pt"
            )
        
        dataset = Dataset.from_list(formatted_data)
        tokenized_dataset = dataset.map(tokenize_function, batched=True)
        
        return tokenized_dataset
    
    def setup_training_args(self, 
                          num_epochs: int = 3,
                          batch_size: int = 4,
                          learning_rate: float = 2e-4,
                          warmup_steps: int = 100):
        """Configure training arguments for enterprise deployment"""
        self.training_args = TrainingArguments(
            output_dir=self.output_dir,
            num_train_epochs=num_epochs,
            per_device_train_batch_size=batch_size,
            per_device_eval_batch_size=batch_size,
            warmup_steps=warmup_steps,
            learning_rate=learning_rate,
            logging_steps=10,
            logging_dir=f"{self.output_dir}/logs",
            save_steps=500,
            save_total_limit=3,
            evaluation_strategy="steps",
            eval_steps=500,
            load_best_model_at_end=True,
            metric_for_best_model="eval_loss",
            greater_is_better=False,
            report_to="wandb",  # Logging to Weights & Biases
            run_name=f"enterprise_finetuning_{self.model_name}",
            fp16=True,  # Use mixed precision training
            dataloader_pin_memory=True,
            dataloader_num_workers=4,
            remove_unused_columns=False
        )
        
        return self.training_args
    
    def train(self, train_dataset, eval_dataset=None):
        """Execute the fine-tuning process"""
        # Data collator for language modeling
        data_collator = DataCollatorForLanguageModeling(
            tokenizer=self.tokenizer,
            mlm=False  # We're doing causal LM, not masked LM
        )
        
        # Initialize trainer
        trainer = Trainer(
            model=self.model,
            args=self.training_args,
            train_dataset=train_dataset,
            eval_dataset=eval_dataset,
            data_collator=data_collator,
            tokenizer=self.tokenizer
        )
        
        # Start training
        print("Starting fine-tuning...")
        trainer.train()
        
        # Save the final model
        trainer.save_model()
        self.tokenizer.save_pretrained(self.output_dir)
        
        print(f"Fine-tuning completed! Model saved to {self.output_dir}")
        
        return trainer

# Example usage
def main():
    # Initialize fine-tuner
    fine_tuner = EnterpriseFineTuner(
        model_name="meta-llama/Llama-2-7b-hf",
        output_dir="./enterprise_model"
    )
    
    # Setup model and tokenizer
    fine_tuner.setup_model_and_tokenizer()
    
    # Configure LoRA
    fine_tuner.setup_lora_config(r=16, alpha=32)
    
    # Prepare training data
    training_data = load_enterprise_data()  # Your data loading function
    train_dataset = fine_tuner.prepare_dataset(training_data)
    
    # Setup training arguments
    fine_tuner.setup_training_args(
        num_epochs=3,
        batch_size=4,
        learning_rate=2e-4
    )
    
    # Train the model
    trainer = fine_tuner.train(train_dataset)
    
    print("Fine-tuning pipeline completed successfully!")

if __name__ == "__main__":
    main()

This FastAPI server provides enterprise-grade model serving with proper error handling, monitoring, and security features.

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
from peft import PeftModel
import time
import logging
from typing import Optional, List
import redis
import json
from prometheus_client import Counter, Histogram, generate_latest
import uvicorn

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

# Prometheus metrics
REQUEST_COUNT = Counter('model_requests_total', 'Total model requests', ['model', 'status'])
REQUEST_DURATION = Histogram('model_request_duration_seconds', 'Request duration', ['model'])

class EnterpriseModelServer:
    def __init__(self, model_path: str, base_model_name: str):
        self.model_path = model_path
        self.base_model_name = base_model_name
        self.model = None
        self.tokenizer = None
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        self.security = HTTPBearer()
        
    def load_model(self):
        """Load the fine-tuned model with enterprise optimizations"""
        logger.info(f"Loading model from {self.model_path}")
        
        # Load tokenizer
        self.tokenizer = AutoTokenizer.from_pretrained(self.model_path)
        
        # Load base model
        base_model = AutoModelForCausalLM.from_pretrained(
            self.base_model_name,
            torch_dtype=torch.float16,
            device_map="auto"
        )
        
        # Load fine-tuned weights
        self.model = PeftModel.from_pretrained(base_model, self.model_path)
        self.model.eval()
        
        logger.info("Model loaded successfully")
    
    def generate_response(self, 
                         prompt: str, 
                         max_length: int = 512,
                         temperature: float = 0.7,
                         top_p: float = 0.9) -> str:
        """Generate response with caching and error handling"""
        
        # Check cache first
        cache_key = f"response:{hash(prompt)}"
        cached_response = self.redis_client.get(cache_key)
        
        if cached_response:
            logger.info("Cache hit")
            return json.loads(cached_response)['response']
        
        try:
            # Tokenize input
            inputs = self.tokenizer(
                prompt, 
                return_tensors="pt", 
                truncation=True, 
                max_length=256
            ).to(self.model.device)
            
            # Generate response
            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_length=max_length,
                    temperature=temperature,
                    top_p=top_p,
                    do_sample=True,
                    pad_token_id=self.tokenizer.eos_token_id
                )
            
            # Decode response
            response = self.tokenizer.decode(
                outputs[0][inputs['input_ids'].shape[1]:], 
                skip_special_tokens=True
            )
            
            # Cache response
            self.redis_client.setex(
                cache_key, 
                3600,  # 1 hour cache
                json.dumps({'response': response})
            )
            
            return response
            
        except Exception as e:
            logger.error(f"Generation error: {str(e)}")
            raise HTTPException(status_code=500, detail="Model generation failed")

# Pydantic models for API
class GenerationRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=1000)
    max_length: Optional[int] = Field(512, ge=50, le=1024)
    temperature: Optional[float] = Field(0.7, ge=0.1, le=2.0)
    top_p: Optional[float] = Field(0.9, ge=0.1, le=1.0)

class GenerationResponse(BaseModel):
    response: str
    model: str
    timestamp: float
    cached: bool = False

# Initialize FastAPI app
app = FastAPI(
    title="Enterprise LLM API",
    description="Production-ready fine-tuned language model API",
    version="1.0.0"
)

# Global model server instance
model_server = EnterpriseModelServer(
    model_path="./enterprise_model",
    base_model_name="meta-llama/Llama-2-7b-hf"
)

@app.on_event("startup")
async def startup_event():
    """Load model on startup"""
    model_server.load_model()

@app.post("/generate", response_model=GenerationResponse)
async def generate_text(
    request: GenerationRequest,
    credentials: HTTPAuthorizationCredentials = Depends(model_server.security)
):
    """Generate text using the fine-tuned model"""
    
    # Validate API key (implement your auth logic)
    if not validate_api_key(credentials.credentials):
        raise HTTPException(status_code=401, detail="Invalid API key")
    
    # Record metrics
    start_time = time.time()
    
    try:
        # Generate response
        response = model_server.generate_response(
            prompt=request.prompt,
            max_length=request.max_length,
            temperature=request.temperature,
            top_p=request.top_p
        )
        
        # Record successful request
        REQUEST_COUNT.labels(model="enterprise_llm", status="success").inc()
        
        return GenerationResponse(
            response=response,
            model="enterprise_llm",
            timestamp=time.time(),
            cached=False
        )
        
    except Exception as e:
        # Record failed request
        REQUEST_COUNT.labels(model="enterprise_llm", status="error").inc()
        raise HTTPException(status_code=500, detail=str(e))
    
    finally:
        # Record request duration
        REQUEST_DURATION.labels(model="enterprise_llm").observe(time.time() - start_time)

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

@app.get("/metrics")
async def metrics():
    """Prometheus metrics endpoint"""
    return generate_latest()

def validate_api_key(api_key: str) -> bool:
    """Validate API key (implement your authentication logic)"""
    # This is a placeholder - implement proper API key validation
    return api_key == "your-secure-api-key"

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Multi-stage Docker build for production deployment
FROM nvidia/cuda:11.8-devel-ubuntu20.04 as base

# Install system dependencies
RUN apt-get update && apt-get install -y \
    python3.9 \
    python3.9-dev \
    python3-pip \
    git \
    curl \
    && rm -rf /var/lib/apt/lists/*

# Set Python 3.9 as default
RUN update-alternatives --install /usr/bin/python python /usr/bin/python3.9 1
RUN update-alternatives --install /usr/bin/pip pip /usr/bin/pip3 1

# Install Python dependencies
COPY requirements.txt /app/requirements.txt
WORKDIR /app
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . /app

# Create non-root user for security
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# Expose port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# Start the application
CMD ["python", "model_server.py"]