ml_model.py

# ml_model.py
import torch
import torch.nn as nn

class LayerNorm(nn.Module):
    def __init__(self, emb_dim):
        super().__init__()
        self.eps = 1e-5
        self.scale = nn.Parameter(torch.ones(emb_dim))
        self.shift = nn.Parameter(torch.zeros(emb_dim))

    def forward(self, x):
        mean = x.mean(dim=-1, keepdim=True)
        var = x.var(dim=-1, keepdim=True)
        norm_x = (x - mean) / torch.sqrt(var + self.eps)
        return self.scale * norm_x + self.shift


class GELU(nn.Module):
    def __init__(self):
        super().__init__()

    def forward(self, x):
        return 0.5 * x * (1 + torch.tanh(
            torch.sqrt(torch.tensor(2.0 / torch.pi)) * 
            (x + 0.044715 * torch.pow(x, 3))
        ))


class MultiHeadAttentionWrapper_V2(nn.Module):
    def __init__(self, d_in, d_out, context_length, dropout, num_heads, qkv_bias):
        super().__init__()
        assert (d_out % num_heads == 0), "d_out must be divisible by num_heads"
        self.d_out = d_out
        self.num_heads = num_heads
        self.head_dim = d_out // num_heads

        self.W_query = nn.Linear(d_in, d_out, bias=qkv_bias)
        self.W_key = nn.Linear(d_in, d_out, bias=qkv_bias)
        self.W_value = nn.Linear(d_in, d_out, bias=qkv_bias)
        self.out_proj = nn.Linear(d_out, d_out)
        self.dropout = nn.Dropout(dropout)
        self.register_buffer(
            "mask",
            torch.triu(torch.ones(context_length, context_length), diagonal=1)
        )

    def forward(self, x):
        b, num_tokens, d_in = x.shape
        keys = self.W_key(x)
        queries = self.W_query(x)
        values = self.W_value(x)

        keys = keys.view(b, num_tokens, self.num_heads, self.head_dim)
        queries = queries.view(b, num_tokens, self.num_heads, self.head_dim)
        values = values.view(b, num_tokens, self.num_heads, self.head_dim)

        keys = keys.transpose(1, 2)
        queries = queries.transpose(1, 2)
        values = values.transpose(1, 2)

        attn_scores = queries @ keys.transpose(2, 3)

        mask_bool = self.mask.bool()[:num_tokens, :num_tokens]
        attn_scores.masked_fill_(mask_bool, -torch.inf)

        attn_weights = torch.softmax(attn_scores / keys.shape[-1]**0.5, dim=-1)
        attn_weights = self.dropout(attn_weights)

        context_vec = (attn_weights @ values).transpose(1, 2)
        context_vec = context_vec.reshape(b, num_tokens, self.d_out)
        context_vec = self.out_proj(context_vec)

        return context_vec


class FeedForward(nn.Module):
    def __init__(self, cfg):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Linear(cfg["emb_dim"], 4 * cfg["emb_dim"]),
            GELU(),
            nn.Linear(4 * cfg["emb_dim"], cfg["emb_dim"])
        )
    
    def forward(self, x):
        return self.layers(x)


class TransformerBlock(nn.Module):
    def __init__(self, cfg):
        super().__init__()
        self.att = MultiHeadAttentionWrapper_V2(
            d_in=cfg['emb_dim'],
            d_out=cfg['emb_dim'],
            context_length=cfg["context_length"],
            num_heads=cfg["n_heads"],
            dropout=cfg["drop_rate"],
            qkv_bias=cfg["qkv_bias"]
        )
        self.ff = FeedForward(cfg)
        self.norm1 = LayerNorm(cfg["emb_dim"])
        self.norm2 = LayerNorm(cfg["emb_dim"])
        self.drop_shortcut = nn.Dropout(cfg["drop_rate"])

    def forward(self, x):
        x = x + self.drop_shortcut(self.att(self.norm1(x)))
        x = x + self.drop_shortcut(self.ff(self.norm2(x)))
        return x


class GPTModel(nn.Module):
    def __init__(self, cfg):
        super().__init__()
        self.tok_emb = nn.Embedding(cfg["vocab_size"], cfg["emb_dim"])
        self.pos_emb = nn.Embedding(cfg["context_length"], cfg["emb_dim"])
        self.drop_emb = nn.Dropout(cfg["drop_rate"])
        self.trf_blocks = nn.Sequential(
            *[TransformerBlock(cfg) for _ in range(cfg["n_layers"])]
        )
        self.final_norm = LayerNorm(cfg["emb_dim"])
        self.out_head = nn.Linear(cfg["emb_dim"], cfg["vocab_size"], bias=False)
    
    def forward(self, in_idx):
        batch_size, seq_len = in_idx.shape
        tok_embeds = self.tok_emb(in_idx)
        pos_embeds = self.pos_emb(torch.arange(seq_len, device=in_idx.device))
        x = tok_embeds + pos_embeds
        x = self.drop_emb(x)
        x = self.trf_blocks(x)
        x = self.final_norm(x)
        logits = self.out_head(x)
        return logits


def generate_text_better(model, idx, max_new_tokens, context_size, temperature=1.0, top_k=None):
    """Generate text using the model with temperature and top-k sampling"""
    for _ in range(max_new_tokens):
        idx_cond = idx[:, -context_size:]
        with torch.no_grad():
            logits = model(idx_cond)
        logits = logits[:, -1, :] / temperature
        
        if top_k is not None:
            v, _ = torch.topk(logits, min(top_k, logits.size(-1)))
            logits[logits < v[:, [-1]]] = -float('Inf')
        
        probas = torch.softmax(logits, dim=-1)
        idx_next = torch.multinomial(probas, num_samples=1)
        idx = torch.cat((idx, idx_next), dim=1)
    return idx


def text_token_ids(text, tokenizer):
    """Convert text to token IDs"""
    encoded = tokenizer.encode(text, allowed_special={"<|endoftext|>"})
    encoded_tensor = torch.tensor(encoded).unsqueeze(0)
    return encoded_tensor


def token_text_ids(tokens, tokenizer):
    """Convert token IDs back to text"""
    flat = tokens.squeeze(0)
    return tokenizer.decode(flat.tolist())


# ============================================================================
# SUMMARIZATION UTILITIES
# ============================================================================

def format_court_doc_prompt(document, instruction="Ringkaskan dokumen pengadilan berikut:"):
    """
    Format Indonesian court document for summarization task.

    Args:
        document: The full court document text
        instruction: The instruction prompt in Indonesian (default: "Summarize the following court document:")

    Returns:
        Formatted prompt string
    """
    prompt = f"""{instruction}

Dokumen:
{document}

Ringkasan:"""
    return prompt


def format_training_example(document, summary, instruction="Ringkaskan dokumen pengadilan berikut:"):
    """
    Format a training example with document and its summary.

    Args:
        document: The full court document text
        summary: The target summary
        instruction: The instruction prompt in Indonesian

    Returns:
        Complete training text with document and summary
    """
    return f"""{instruction}

Dokumen:
{document}

Ringkasan:
{summary}<|endoftext|>"""


def preprocess_court_documents(documents, summaries, tokenizer, max_length=2048):
    """
    Preprocess court documents and summaries for training.

    Args:
        documents: List of court document texts
        summaries: List of corresponding summaries
        tokenizer: The tokenizer to use
        max_length: Maximum sequence length

    Returns:
        List of tokenized training examples
    """
    training_data = []

    for doc, summ in zip(documents, summaries):
        formatted = format_training_example(doc, summ)

        # Tokenize
        encoded = tokenizer.encode(formatted, allowed_special={"<|endoftext|>"})

        # Truncate if needed
        if len(encoded) > max_length:
            encoded = encoded[:max_length]

        training_data.append(torch.tensor(encoded))

    return training_data


def generate_summary(model, document, tokenizer, cfg, max_summary_tokens=256,
                     temperature=0.7, top_k=50, instruction="Ringkaskan dokumen pengadilan berikut:"):
    """
    Generate a summary for an Indonesian court document.

    Args:
        model: The trained GPT model
        document: The court document text to summarize
        tokenizer: The tokenizer
        cfg: Model configuration dict
        max_summary_tokens: Maximum length of generated summary
        temperature: Sampling temperature (lower = more focused)
        top_k: Top-k sampling parameter
        instruction: Instruction prompt in Indonesian

    Returns:
        Generated summary text
    """
    model.eval()

    # Format the prompt
    prompt = format_court_doc_prompt(document, instruction)

    # Tokenize
    encoded = tokenizer.encode(prompt, allowed_special={"<|endoftext|>"})
    encoded_tensor = torch.tensor(encoded).unsqueeze(0)

    # Move to same device as model
    device = next(model.parameters()).device
    encoded_tensor = encoded_tensor.to(device)

    # Generate
    with torch.no_grad():
        output = generate_text_better(
            model=model,
            idx=encoded_tensor,
            max_new_tokens=max_summary_tokens,
            context_size=cfg["context_length"],
            temperature=temperature,
            top_k=top_k
        )

    # Decode
    generated_text = tokenizer.decode(output.squeeze(0).tolist())

    # Extract just the summary part (after "Ringkasan:")
    if "Ringkasan:" in generated_text:
        summary = generated_text.split("Ringkasan:")[-1].strip()
        # Remove endoftext token if present
        summary = summary.replace("<|endoftext|>", "").strip()
        return summary

    return generated_text


def calc_loss_batch(input_batch, target_batch, model, device):
    """Calculate loss for a batch of data"""
    input_batch = input_batch.to(device)
    target_batch = target_batch.to(device)
    logits = model(input_batch)
    loss = torch.nn.functional.cross_entropy(
        logits.flatten(0, 1), target_batch.flatten()
    )
    return loss


def calc_loss_loader(data_loader, model, device, num_batches=None):
    """Calculate average loss over data loader"""
    total_loss = 0.
    if num_batches is None:
        num_batches = len(data_loader)
    else:
        num_batches = min(num_batches, len(data_loader))

    for i, (input_batch, target_batch) in enumerate(data_loader):
        if i >= num_batches:
            break
        loss = calc_loss_batch(input_batch, target_batch, model, device)
        total_loss += loss.item()

    return total_loss / num_batches


def train_model_summarization(model, train_loader, val_loader, optimizer, device,
                              num_epochs, eval_freq, eval_iter, start_context,
                              tokenizer, cfg):
    """
    Train the model for Indonesian court document summarization.

    Args:
        model: GPTModel instance
        train_loader: Training data loader
        val_loader: Validation data loader
        optimizer: Optimizer (e.g., AdamW)
        device: Device to train on (cuda/cpu)
        num_epochs: Number of training epochs
        eval_freq: Evaluate every N steps
        eval_iter: Number of batches for evaluation
        start_context: Sample document for testing during training
        tokenizer: Tokenizer for decoding
        cfg: Model configuration

    Returns:
        Lists of training losses, validation losses, and tracked tokens
    """
    train_losses, val_losses, track_tokens_seen = [], [], []
    tokens_seen = 0
    global_step = -1

    for epoch in range(num_epochs):
        model.train()

        for input_batch, target_batch in train_loader:
            optimizer.zero_grad()
            loss = calc_loss_batch(input_batch, target_batch, model, device)
            loss.backward()
            optimizer.step()
            tokens_seen += input_batch.numel()
            global_step += 1

            # Evaluate periodically
            if global_step % eval_freq == 0:
                train_loss = calc_loss_loader(train_loader, model, device, num_batches=eval_iter)
                val_loss = calc_loss_loader(val_loader, model, device, num_batches=eval_iter)
                train_losses.append(train_loss)
                val_losses.append(val_loss)
                track_tokens_seen.append(tokens_seen)
                print(f"Ep {epoch+1} (Step {global_step:06d}): "
                      f"Train loss {train_loss:.3f}, Val loss {val_loss:.3f}")

        # Generate sample summary at end of each epoch
        print(f"\n--- Sample Summary after Epoch {epoch+1} ---")
        sample_summary = generate_summary(
            model=model,
            document=start_context,
            tokenizer=tokenizer,
            cfg=cfg,
            max_summary_tokens=150,
            temperature=0.7,
            top_k=50
        )
        print(sample_summary)
        print("-" * 50 + "\n")

    return train_losses, val_losses, track_tokens_seen


# ============================================================================
# CONFIGURATION FOR INDONESIAN COURT DOCUMENT SUMMARIZATION
# ============================================================================

SUMMARIZATION_CONFIG = {
    "vocab_size": 50257,      # GPT-2 vocab size (works with tiktoken)
    "context_length": 2048,   # Longer context for court documents
    "emb_dim": 768,           # Embedding dimension
    "n_heads": 12,            # Number of attention heads
    "n_layers": 12,           # Number of transformer blocks
    "drop_rate": 0.1,         # Dropout rate
    "qkv_bias": False         # Use bias in attention projections
}


# Example usage function
def example_summarization_pipeline():
    """
    Example of how to use the model for Indonesian court document summarization.
    This is a template - adjust paths and data as needed.
    """
    import tiktoken

    # Initialize tokenizer
    tokenizer = tiktoken.get_encoding("gpt2")

    # Initialize model
    model = GPTModel(SUMMARIZATION_CONFIG)
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model.to(device)

    # Example: Load your court documents and summaries
    # documents = load_court_documents()  # Your data loading function
    # summaries = load_summaries()         # Your data loading function

    # Example: Preprocess data
    # training_data = preprocess_court_documents(documents, summaries, tokenizer)

    # Example: Create data loaders
    # from torch.utils.data import DataLoader, Dataset
    # train_loader = DataLoader(your_dataset, batch_size=4, shuffle=True)
    # val_loader = DataLoader(your_val_dataset, batch_size=4, shuffle=False)

    # Example: Train
    # optimizer = torch.optim.AdamW(model.parameters(), lr=5e-5, weight_decay=0.1)
    # train_model_summarization(
    #     model=model,
    #     train_loader=train_loader,
    #     val_loader=val_loader,
    #     optimizer=optimizer,
    #     device=device,
    #     num_epochs=5,
    #     eval_freq=100,
    #     eval_iter=10,
    #     start_context="Sample court document...",
    #     tokenizer=tokenizer,
    #     cfg=SUMMARIZATION_CONFIG
    # )

    # Example: Generate summary
    court_doc = "Putusan Pengadilan Negeri Jakarta Pusat..."
    summary = generate_summary(
        model=model,
        document=court_doc,
        tokenizer=tokenizer,
        cfg=SUMMARIZATION_CONFIG
    )
    print(f"Summary: {summary}")

    return model, tokenizer