Beyond Frontier Models: A Technical Deep Dive & Practitioner's Guide to LLM Tuning, Optimization, Deployment, and Architecture

# Training data format for instruction tuning
training_examples = [
    {
        "instruction": "Summarize the following article in 3 bullet points.",
        "input": "<article text>",
        "output": "• Key point 1\n• Key point 2\n• Key point 3"
    },
    # Diverse phrasing matters!
    {
        "instruction": "Give me a 3-point summary of this text.",
        "input": "<article text>",
        "output": "..."
    }
]

# SFT training loop
def sft_training_step(model, batch, optimizer):
    inputs = tokenize(batch["instruction"] + batch["input"])
    targets = tokenize(batch["output"])

    logits = model(inputs)
    loss = cross_entropy_loss(logits, targets)

    loss.backward()
    optimizer.step()
    return loss

# Direct Preference Optimization (DPO) - simpler than RLHF
def dpo_loss(model, ref_model, preferred, rejected, beta=0.1):
    """
    DPO loss without explicit reward modeling.
    beta controls deviation from reference model.
    """
    # Log probabilities under current and reference models
    pi_preferred = model.log_prob(preferred)
    pi_rejected = model.log_prob(rejected)
    ref_preferred = ref_model.log_prob(preferred)
    ref_rejected = ref_model.log_prob(rejected)

    # DPO objective
    preferred_ratio = pi_preferred - ref_preferred
    rejected_ratio = pi_rejected - ref_rejected

    loss = -torch.log(torch.sigmoid(beta * (preferred_ratio - rejected_ratio)))
    return loss.mean()

# Domain adaptation strategy
domain_training_config = {
    "learning_rate": 1e-5,  # Lower than initial pre-training
    "warmup_ratio": 0.1,
    "epochs": 2,  # Don't overtrain - causes forgetting
    "data_mix": {
        "domain_specific": 0.7,
        "general": 0.3  # Prevents catastrophic forgetting
    }
}

# Cross-Entropy: Standard next-token prediction
def cross_entropy_loss(logits, targets):
    return F.cross_entropy(logits.view(-1, vocab_size), targets.view(-1))

# Triplet Loss: For embedding training
def triplet_loss(anchor, positive, negative, margin=0.2):
    pos_dist = F.pairwise_distance(anchor, positive)
    neg_dist = F.pairwise_distance(anchor, negative)
    return F.relu(pos_dist - neg_dist + margin).mean()

# KL Divergence: For distillation or regularization
def kl_divergence_loss(student_logits, teacher_logits, temperature=2.0):
    student_probs = F.log_softmax(student_logits / temperature, dim=-1)
    teacher_probs = F.softmax(teacher_logits / temperature, dim=-1)
    return F.kl_div(student_probs, teacher_probs, reduction='batchmean')

# Contrastive Loss (InfoNCE): For embedding models
def contrastive_loss(embeddings, temperature=0.07):
    similarity = embeddings @ embeddings.T / temperature
    labels = torch.arange(len(embeddings))
    return F.cross_entropy(similarity, labels)

class LoRALayer(nn.Module):
    """
    LoRA: Learns low-rank updates to frozen weights.
    W_new = W_frozen + (A @ B) * scaling
    """
    def __init__(self, in_features, out_features, rank=8, alpha=16):
        super().__init__()
        self.rank = rank
        self.scaling = alpha / rank

        # Frozen original weights
        self.weight = nn.Parameter(torch.randn(out_features, in_features),
                                   requires_grad=False)

        # Trainable low-rank matrices
        self.lora_A = nn.Parameter(torch.randn(rank, in_features) * 0.01)
        self.lora_B = nn.Parameter(torch.zeros(out_features, rank))

    def forward(self, x):
        # Original forward + low-rank update
        original = F.linear(x, self.weight)
        lora_update = F.linear(F.linear(x, self.lora_A), self.lora_B)
        return original + lora_update * self.scaling

# QLoRA configuration for memory-efficient fine-tuning
qlora_config = {
    "load_in_4bit": True,
    "bnb_4bit_quant_type": "nf4",  # NormalFloat4 - optimal for weights
    "bnb_4bit_compute_dtype": torch.bfloat16,
    "bnb_4bit_use_double_quant": True,  # Quantize the quantization constants
    "lora_r": 64,
    "lora_alpha": 16,
    "lora_dropout": 0.1,
    "target_modules": ["q_proj", "k_proj", "v_proj", "o_proj"]
}

generation_config = {
    # Temperature: Controls randomness
    # 0.0-0.3: Factual, deterministic tasks
    # 0.7-1.0: Creative tasks
    "temperature": 0.7,

    # Top-p (nucleus sampling): Dynamic vocabulary filtering
    # Samples from smallest set with cumulative probability >= p
    "top_p": 0.9,

    # Top-k: Fixed vocabulary limit
    "top_k": 50,

    # Repetition control
    "presence_penalty": 0.6,   # Penalize tokens that appeared
    "frequency_penalty": 0.3,  # Penalize based on frequency
}

def generate_with_config(model, prompt, config):
    logits = model(prompt)

    # Apply temperature
    logits = logits / config["temperature"]

    # Apply top-k filtering
    if config["top_k"] > 0:
        indices_to_remove = logits < torch.topk(logits, config["top_k"])[0][..., -1, None]
        logits[indices_to_remove] = float('-inf')

    # Apply top-p filtering
    sorted_logits, sorted_indices = torch.sort(logits, descending=True)
    cumulative_probs = torch.cumsum(F.softmax(sorted_logits, dim=-1), dim=-1)
    sorted_indices_to_remove = cumulative_probs > config["top_p"]
    sorted_indices_to_remove[..., 1:] = sorted_indices_to_remove[..., :-1].clone()
    sorted_indices_to_remove[..., 0] = 0
    indices_to_remove = sorted_indices_to_remove.scatter(1, sorted_indices, sorted_indices_to_remove)
    logits[indices_to_remove] = float('-inf')

    # Sample
    probs = F.softmax(logits, dim=-1)
    return torch.multinomial(probs, num_samples=1)

def calculate_memory_requirements(
    num_params_billions,
    precision="fp16",
    batch_size=1,
    sequence_length=2048
):
    """Estimate GPU memory for inference."""

    bytes_per_param = {
        "fp32": 4,
        "fp16": 2,
        "bf16": 2,
        "int8": 1,
        "int4": 0.5
    }

    # Model weights
    model_memory_gb = (num_params_billions * 1e9 *
                       bytes_per_param[precision]) / 1e9

    # KV cache (for each layer)
    # Assuming ~100 layers, hidden_size ≈ num_params^0.5 * 1000
    hidden_size = int((num_params_billions ** 0.5) * 1000)
    num_layers = min(int(num_params_billions * 10), 100)
    kv_cache_per_token = 2 * num_layers * hidden_size * bytes_per_param[precision]
    kv_cache_gb = (batch_size * sequence_length * kv_cache_per_token) / 1e9

    # Activation memory (rough estimate)
    activation_gb = model_memory_gb * 0.1

    return {
        "model_weights_gb": model_memory_gb,
        "kv_cache_gb": kv_cache_gb,
        "activations_gb": activation_gb,
        "total_gb": model_memory_gb + kv_cache_gb + activation_gb
    }

# Example: 70B model in fp16
# calculate_memory_requirements(70, "fp16") → ~150GB total
# calculate_memory_requirements(70, "int4") → ~45GB total

def speculative_decode(target_model, draft_model, prompt, num_speculative=4):
    """
    Use small draft model to propose tokens, verify with target model.
    Can achieve 2-3x speedup.
    """
    generated = prompt

    while not is_complete(generated):
        # Draft model generates multiple tokens quickly
        draft_tokens = []
        draft_probs = []
        for _ in range(num_speculative):
            logits = draft_model(generated + draft_tokens)
            token = sample(logits)
            draft_tokens.append(token)
            draft_probs.append(F.softmax(logits, dim=-1))

        # Target model verifies all at once (single forward pass)
        target_logits = target_model(generated + draft_tokens)
        target_probs = F.softmax(target_logits, dim=-1)

        # Accept tokens where target agrees with draft
        accepted = 0
        for i, (draft_p, target_p) in enumerate(zip(draft_probs, target_probs)):
            # Acceptance probability
            r = random.random()
            if r < min(1, target_p[draft_tokens[i]] / draft_p[draft_tokens[i]]):
                accepted += 1
            else:
                break

        generated += draft_tokens[:accepted]

        # Sample one more from target if we rejected early
        if accepted < num_speculative:
            generated += sample(target_logits[accepted])

    return generated

class PagedKVCache:
    """
    vLLM-style paged attention for efficient memory usage.
    Allocates memory in pages, enables sharing across requests.
    """
    def __init__(self, page_size=16, max_pages=1000):
        self.page_size = page_size
        self.pages = {}  # page_id -> tensor
        self.page_table = {}  # request_id -> [page_ids]

    def allocate_page(self):
        page_id = len(self.pages)
        self.pages[page_id] = torch.zeros(self.page_size, hidden_size)
        return page_id

    def get_kv_cache(self, request_id, position):
        page_idx = position // self.page_size
        offset = position % self.page_size

        if request_id not in self.page_table:
            self.page_table[request_id] = []

        while len(self.page_table[request_id]) <= page_idx:
            self.page_table[request_id].append(self.allocate_page())

        page_id = self.page_table[request_id][page_idx]
        return self.pages[page_id][offset]

class HybridRetriever:
    """
    Combines dense (semantic) and sparse (keyword) retrieval.
    """
    def __init__(self, dense_model, sparse_index, dense_index):
        self.dense_model = dense_model
        self.sparse_index = sparse_index  # BM25, TF-IDF
        self.dense_index = dense_index    # FAISS, Pinecone

    def retrieve(self, query, k=10, dense_weight=0.7):
        # Dense retrieval
        query_embedding = self.dense_model.encode(query)
        dense_results = self.dense_index.search(query_embedding, k=k*2)

        # Sparse retrieval
        sparse_results = self.sparse_index.search(query, k=k*2)

        # Reciprocal Rank Fusion
        scores = {}
        for rank, (doc_id, _) in enumerate(dense_results):
            scores[doc_id] = scores.get(doc_id, 0) + dense_weight / (rank + 60)

        for rank, (doc_id, _) in enumerate(sparse_results):
            scores[doc_id] = scores.get(doc_id, 0) + (1 - dense_weight) / (rank + 60)

        # Return top-k by fused score
        ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return ranked[:k]

# Multi-stage retrieval with reranking
class MultiStageRAG:
    def __init__(self, retriever, reranker, generator):
        self.retriever = retriever
        self.reranker = reranker  # Cross-encoder for precise ranking
        self.generator = generator

    def query(self, question, k_retrieve=50, k_rerank=5):
        # Stage 1: Fast retrieval (bi-encoder)
        candidates = self.retriever.retrieve(question, k=k_retrieve)

        # Stage 2: Precise reranking (cross-encoder)
        reranked = self.reranker.rerank(question, candidates, k=k_rerank)

        # Stage 3: Generate with retrieved context
        context = "\n\n".join([doc.text for doc in reranked])
        prompt = f"""Context:\n{context}\n\nQuestion: {question}\n\nAnswer:"""

        return self.generator.generate(prompt)

# Query decomposition for complex questions
def decompose_query(question, llm):
    """Break complex questions into sub-questions."""
    prompt = f"""Break this question into simpler sub-questions:
    Question: {question}

    Sub-questions:"""

    sub_questions = llm.generate(prompt).split('\n')
    return [q.strip() for q in sub_questions if q.strip()]

class LLMEvaluator:
    def __init__(self, model):
        self.model = model
        self.metrics = {}

    def evaluate_generation_quality(self, test_set):
        """Quantitative metrics for text generation."""
        results = {
            "perplexity": [],
            "bleu": [],
            "rouge_l": [],
            "bertscore": []
        }

        for example in test_set:
            output = self.model.generate(example["input"])

            # Perplexity
            results["perplexity"].append(
                self.calculate_perplexity(example["input"], output)
            )

            # BLEU for translation-like tasks
            results["bleu"].append(
                sentence_bleu([example["reference"]], output)
            )

            # ROUGE-L for summarization
            results["rouge_l"].append(
                rouge_scorer.score(example["reference"], output)["rougeL"].fmeasure
            )

        return {k: np.mean(v) for k, v in results.items()}

    def evaluate_with_llm_judge(self, test_set, judge_model):
        """LLM-as-a-judge evaluation."""
        judge_prompt = """Rate the following response on a scale of 1-5:
        Question: {question}
        Response: {response}

        Criteria:
        - Accuracy: Is the information correct?
        - Completeness: Does it fully answer the question?
        - Clarity: Is it well-written and clear?

        Provide scores as JSON: {{"accuracy": X, "completeness": X, "clarity": X}}
        """

        scores = []
        for example in test_set:
            response = self.model.generate(example["input"])
            judgment = judge_model.generate(
                judge_prompt.format(question=example["input"], response=response)
            )
            scores.append(json.loads(judgment))

        return aggregate_scores(scores)

class HallucinationMitigation:
    def __init__(self, model, retriever, fact_checker):
        self.model = model
        self.retriever = retriever
        self.fact_checker = fact_checker

    def generate_with_grounding(self, query):
        # Retrieve relevant facts first
        facts = self.retriever.retrieve(query, k=5)
        fact_context = "\n".join([f"- {fact}" for fact in facts])

        prompt = f"""Based ONLY on these facts:
{fact_context}

Answer the question. If the facts don't contain the answer, say "I don't have enough information."

Question: {query}
Answer:"""

        response = self.model.generate(prompt)
        return response

    def verify_response(self, query, response):
        """Post-generation fact checking."""
        # Extract claims from response
        claims = self.extract_claims(response)

        verified_claims = []
        for claim in claims:
            # Check against retrieval
            evidence = self.retriever.retrieve(claim, k=3)
            is_supported = self.fact_checker.verify(claim, evidence)
            verified_claims.append({
                "claim": claim,
                "supported": is_supported,
                "evidence": evidence
            })

        return verified_claims

    def self_consistency_check(self, query, n_samples=5):
        """Generate multiple responses and check consistency."""
        responses = [self.model.generate(query, temperature=0.7)
                     for _ in range(n_samples)]

        # Extract key facts from each response
        fact_sets = [self.extract_facts(r) for r in responses]

        # Find consistent facts (appear in majority)
        all_facts = {}
        for facts in fact_sets:
            for fact in facts:
                all_facts[fact] = all_facts.get(fact, 0) + 1

        consistent = [f for f, count in all_facts.items()
                      if count >= n_samples // 2 + 1]

        return consistent

def attention_based_hallucination_detection(model, prompt, response):
    """
    Lookback ratio: Do generated tokens attend to input or just previous generations?
    Low lookback ratio suggests hallucination.
    """
    full_sequence = prompt + response
    _, attention_weights = model(full_sequence, output_attentions=True)

    prompt_length = len(tokenize(prompt))
    response_length = len(tokenize(response))

    # For each response token, calculate attention to prompt vs. response
    lookback_ratios = []
    for i in range(response_length):
        response_position = prompt_length + i
        attn = attention_weights[-1][0, :, response_position, :]  # Last layer

        prompt_attention = attn[:, :prompt_length].sum()
        total_attention = attn[:, :response_position].sum()

        lookback_ratios.append(prompt_attention / total_attention)

    return np.mean(lookback_ratios)  # Low = likely hallucination

def calculate_kv_cache_size(
    num_layers,
    hidden_size,
    num_heads,
    sequence_length,
    batch_size,
    precision_bytes=2  # fp16
):
    """
    KV cache grows linearly with sequence length.
    This is often the memory bottleneck for long contexts.
    """
    head_dim = hidden_size // num_heads

    # K and V for each layer, each head
    kv_per_token = 2 * num_layers * num_heads * head_dim * precision_bytes

    total_bytes = batch_size * sequence_length * kv_per_token
    return total_bytes / (1024 ** 3)  # GB

# Example: LLaMA 70B with 32K context
# calculate_kv_cache_size(80, 8192, 64, 32768, 1) → ~40GB just for KV cache!

class ChainOfThoughtGenerator:
    """Implement reasoning patterns."""

    def __init__(self, model):
        self.model = model

    def generate_with_cot(self, question):
        prompt = f"""{question}

Let's think through this step by step:
1."""
        return self.model.generate(prompt)

    def self_consistency(self, question, n_paths=5):
        """Generate multiple reasoning paths, vote on answer."""
        answers = []
        for _ in range(n_paths):
            response = self.generate_with_cot(question)
            answer = self.extract_final_answer(response)
            answers.append(answer)

        # Majority vote
        from collections import Counter
        return Counter(answers).most_common(1)[0][0]

    def tree_of_thought(self, question, breadth=3, depth=3):
        """Explore multiple reasoning branches."""
        def expand_node(state, depth_remaining):
            if depth_remaining == 0:
                return self.evaluate_state(state)

            # Generate multiple next steps
            next_steps = self.generate_next_steps(state, n=breadth)

            # Recursively explore each branch
            best_score = float('-inf')
            best_path = None

            for step in next_steps:
                new_state = state + "\n" + step
                score, path = expand_node(new_state, depth_remaining - 1)

                if score > best_score:
                    best_score = score
                    best_path = path

            return best_score, best_path

        initial_state = f"Question: {question}\nReasoning:"
        _, best_reasoning = expand_node(initial_state, depth)
        return best_reasoning