Upload folder using huggingface_hub

.gitattributes

@@ -1,35 +1,3 @@
-*.7z filter=lfs diff=lfs merge=lfs -text
-*.arrow filter=lfs diff=lfs merge=lfs -text
-*.bin filter=lfs diff=lfs merge=lfs -text
-*.bz2 filter=lfs diff=lfs merge=lfs -text
-*.ckpt filter=lfs diff=lfs merge=lfs -text
-*.ftz filter=lfs diff=lfs merge=lfs -text
-*.gz filter=lfs diff=lfs merge=lfs -text
-*.h5 filter=lfs diff=lfs merge=lfs -text
-*.joblib filter=lfs diff=lfs merge=lfs -text
-*.lfs.* filter=lfs diff=lfs merge=lfs -text
-*.mlmodel filter=lfs diff=lfs merge=lfs -text
-*.model filter=lfs diff=lfs merge=lfs -text
-*.msgpack filter=lfs diff=lfs merge=lfs -text
-*.npy filter=lfs diff=lfs merge=lfs -text
-*.npz filter=lfs diff=lfs merge=lfs -text
-*.onnx filter=lfs diff=lfs merge=lfs -text
-*.ot filter=lfs diff=lfs merge=lfs -text
-*.parquet filter=lfs diff=lfs merge=lfs -text
-*.pb filter=lfs diff=lfs merge=lfs -text
-*.pickle filter=lfs diff=lfs merge=lfs -text
-*.pkl filter=lfs diff=lfs merge=lfs -text
-*.pt filter=lfs diff=lfs merge=lfs -text
-*.pth filter=lfs diff=lfs merge=lfs -text
-*.rar filter=lfs diff=lfs merge=lfs -text
 *.safetensors filter=lfs diff=lfs merge=lfs -text
-saved_model/**/* filter=lfs diff=lfs merge=lfs -text
-*.tar.* filter=lfs diff=lfs merge=lfs -text
-*.tar filter=lfs diff=lfs merge=lfs -text
-*.tflite filter=lfs diff=lfs merge=lfs -text
-*.tgz filter=lfs diff=lfs merge=lfs -text
-*.wasm filter=lfs diff=lfs merge=lfs -text
-*.xz filter=lfs diff=lfs merge=lfs -text
-*.zip filter=lfs diff=lfs merge=lfs -text
-*.zst filter=lfs diff=lfs merge=lfs -text
-*tfevents* filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Flame Corporation
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

MODEL_CARD.md

@@ -0,0 +1,58 @@
+---
+license: mit
+language:
+  - pt
+  - en
+pipeline_tag: text-generation
+tags:
+  - pytorch
+  - causal-lm
+  - decoder-only
+  - educational
+  - small-language-model
+---
+
+# Delta Ultra Mini
+
+Delta Ultra Mini is a compact decoder-only language model created by Flame Corporation. It is intended as an educational small LLM release for studying tokenizer training, causal language modeling, checkpoints, and local generation.
+
+## Model Details
+
+- Architecture: decoder-only causal Transformer
+- Parameters: about 50M
+- Context length: 512 tokens
+- Tokenizer: BPE
+- License: MIT
+
+## Intended Use
+
+- Learning how small language models are structured and trained
+- Running local inference experiments
+- Building and testing custom small datasets
+- Fine-tuning from a compact baseline
+
+## Not Intended For
+
+- Production assistant use without evaluation
+- High-stakes decision making
+- Security, medical, legal, or financial advice
+- Claims of strong general reasoning
+
+## Training Data
+
+The seed dataset is small and conversational. It is designed to validate the training pipeline, not to produce a high-quality assistant on its own. Users are encouraged to create larger, cleaner datasets and train new checkpoints.
+
+## Limitations
+
+Delta Ultra Mini can hallucinate, mix examples from the seed dataset, and answer incorrectly. The included checkpoint should be treated as an experimental baseline.
+
+## Quick Start
+
+```bash
+pip install -r requirements.txt
+python scripts/generate_delta.py --prompt "Quem e voce?" --checkpoint_path delta_checkpoint.pt --tokenizer_path tokenizer.json
+```
+
+## Citation
+
+If you publish work using this model, cite the repository or model page where you downloaded it.

README.md

@@ -1,3 +1,80 @@
----
-license: mit
----
+# Delta Ultra Mini
+
+Delta Ultra Mini is a compact decoder-only language model created by Flame Corporation. This model-only release contains the neural model code, tokenizer wrapper, training utilities, seed dataset, and simple local inference tools.
+
+This release intentionally does not include the REST API, API key server, browser SDK, or Python HTTP SDK.
+
+## Model
+
+- Architecture: decoder-only causal Transformer
+- Parameters: about 50M
+- Context length: 512 tokens
+- Tokenizer: BPE with chat tokens
+- License: MIT
+
+Delta Ultra Mini is an educational and experimental small LLM. It is useful for learning how a compact language model is structured, trained, checkpointed, and sampled. It is not a strong general assistant yet.
+
+## Install
+
+```bash
+pip install -r requirements.txt
+```
+
+## Files
+
+- `delta/model.py`: Transformer model
+- `delta/tokenizer.py`: tokenizer training/loading and chat formatting
+- `delta/generator.py`: local autoregressive generation
+- `delta/dataset.py`: text/jsonl dataset loader
+- `delta/trainer.py`: HuggingFace Trainer integration
+- `configs/ultra_mini.json`: model configuration
+- `tokenizer.json`: trained tokenizer
+- `data/`: small MIT-licensed seed dataset
+- `scripts/train_tokenizer.py`: tokenizer training entrypoint
+- `scripts/train_delta.py`: model training entrypoint
+- `scripts/generate_delta.py`: local inference entrypoint
+
+## Local Inference
+
+```bash
+python scripts/generate_delta.py --prompt "O que e PyTorch?" --checkpoint_path runs/delta-ultra-mini/delta_checkpoint.pt --tokenizer_path tokenizer.json
+```
+
+If your checkpoint is at the release root, use:
+
+```bash
+python scripts/generate_delta.py --prompt "Quem e voce?" --checkpoint_path delta_checkpoint.pt --tokenizer_path tokenizer.json
+```
+
+## Train Tokenizer
+
+```bash
+python scripts/train_tokenizer.py --corpus_files data/tokenizer_corpus.txt --output_path tokenizer.json
+```
+
+## Train Model
+
+```bash
+python scripts/train_delta.py --data_path data --output_dir runs/delta-ultra-mini --epochs 1 --batch_size 2 --tokenizer_path tokenizer.json
+```
+
+## Dataset
+
+The included dataset is a small seed dataset. It is meant to bootstrap experimentation and verify the pipeline. For better quality, create a larger dataset with varied examples, clean answers, validation splits, and careful review.
+
+Recommended format:
+
+```jsonl
+{"text":"[SYS] You are Delta. [SEP]\n[USR] Question [SEP]\n[ASS] Answer [SEP]"}
+```
+
+## Limitations
+
+- The seed checkpoint may memorize examples and generalize poorly.
+- The model is not safety-aligned like large production assistants.
+- It can produce incorrect or mixed answers.
+- It should be evaluated before any real use.
+
+## License
+
+MIT.

config.json

@@ -0,0 +1,14 @@
+{
+  "vocab_size": 32000,
+  "d_model": 512,
+  "n_heads": 8,
+  "n_layers": 8,
+  "d_ff": 2048,
+  "max_seq_len": 512,
+  "dropout": 0.1,
+  "tie_embeddings": true,
+  "pad_token_id": 0,
+  "bos_token_id": 2,
+  "eos_token_id": 3,
+  "use_cache": false
+}

configs/ultra_mini.json

@@ -0,0 +1,14 @@
+{
+  "vocab_size": 32000,
+  "d_model": 512,
+  "n_heads": 8,
+  "n_layers": 8,
+  "d_ff": 2048,
+  "max_seq_len": 512,
+  "dropout": 0.1,
+  "tie_embeddings": true,
+  "pad_token_id": 0,
+  "bos_token_id": 2,
+  "eos_token_id": 3,
+  "use_cache": true
+}

data/README.md

@@ -0,0 +1,31 @@
+# Delta Seed Dataset
+
+This directory contains the small seed dataset used to test and bootstrap Delta Ultra Mini.
+
+- `delta_seed.jsonl`: conversational JSONL records with a `text` field.
+- `tokenizer_corpus.txt`: plain text corpus for BPE tokenizer training.
+- `corpus.txt`: training text in the same chat-token format.
+
+This dataset is intentionally small. It is useful for validating the tokenizer, dataset loader, training loop, checkpoint saving, and local inference. It is not enough to create a strong general assistant.
+
+## License
+
+MIT, same as the project.
+
+## Format
+
+```jsonl
+{"text":"[SYS] You are Delta. [SEP]\n[USR] Question [SEP]\n[ASS] Answer [SEP]"}
+```
+
+## Train Tokenizer
+
+```powershell
+python scripts/train_tokenizer.py --corpus_files data/tokenizer_corpus.txt --output_path tokenizer.json
+```
+
+## Train Model
+
+```powershell
+python scripts/train_delta.py --data_path data --output_dir runs/delta-ultra-mini --epochs 1 --batch_size 2 --tokenizer_path tokenizer.json
+```

delta/__init__.py

@@ -0,0 +1,6 @@
+"""Delta Ultra Mini package."""
+
+from delta.identity import get_identity
+from delta.model import DeltaConfig, DeltaModel
+
+__all__ = ["DeltaConfig", "DeltaModel", "get_identity"]

delta/dataset.py

@@ -0,0 +1,101 @@
+"""Dataset and collator for Delta causal language modeling."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from pathlib import Path
+from typing import Any
+
+import torch
+from torch.utils.data import Dataset
+
+from delta.tokenizer import DeltaTokenizer
+
+logging.basicConfig(level=os.getenv("DELTA_LOG_LEVEL", "INFO").upper())
+logger = logging.getLogger(__name__)
+
+
+def _read_texts(path: Path) -> list[str]:
+    """Read .txt files or .jsonl files containing a text field."""
+
+    files = [path] if path.is_file() else sorted(path.rglob("*"))
+    texts: list[str] = []
+    for file_path in files:
+        if file_path.suffix.lower() == ".txt":
+            texts.append(file_path.read_text(encoding="utf-8"))
+        elif file_path.suffix.lower() == ".jsonl":
+            with file_path.open("r", encoding="utf-8") as handle:
+                for line in handle:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    record = json.loads(line)
+                    if "text" in record:
+                        texts.append(str(record["text"]))
+    return texts
+
+
+class DeltaDataset(Dataset[dict[str, torch.Tensor]]):
+    """Sliding-window token dataset for language modeling."""
+
+    def __init__(
+        self,
+        data_path: str | Path,
+        tokenizer: DeltaTokenizer,
+        max_seq_len: int = 512,
+        stride: int = 256,
+    ) -> None:
+        self.data_path = Path(data_path)
+        self.tokenizer = tokenizer
+        self.max_seq_len = max_seq_len
+        self.stride = stride
+        texts = _read_texts(self.data_path)
+        if not texts:
+            raise ValueError(f"No .txt or .jsonl text records found in {self.data_path}")
+        self.windows: list[list[int]] = []
+        for text in texts:
+            ids = tokenizer.encode(text, add_special_tokens=True)
+            for start in range(0, max(1, len(ids) - 1), stride):
+                window = ids[start : start + max_seq_len]
+                if len(window) >= 2:
+                    self.windows.append(window)
+                if start + max_seq_len >= len(ids):
+                    break
+        logger.info("Loaded %s training windows from %s", len(self.windows), self.data_path)
+
+    def __len__(self) -> int:
+        """Return the number of windows."""
+
+        return len(self.windows)
+
+    def __getitem__(self, index: int) -> dict[str, torch.Tensor]:
+        """Return one token window."""
+
+        ids = torch.tensor(self.windows[index], dtype=torch.long)
+        return {"input_ids": ids, "labels": ids.clone()}
+
+
+class DeltaDataCollator:
+    """Dynamic padding collator for causal language modeling."""
+
+    def __init__(self, pad_token_id: int = 0) -> None:
+        self.pad_token_id = pad_token_id
+
+    def __call__(self, features: list[dict[str, torch.Tensor]]) -> dict[str, torch.Tensor]:
+        """Pad input ids and labels to the longest sample in the batch."""
+
+        max_len = max(feature["input_ids"].size(0) for feature in features)
+        input_ids = torch.full((len(features), max_len), self.pad_token_id, dtype=torch.long)
+        labels = torch.full((len(features), max_len), -100, dtype=torch.long)
+        attention_mask = torch.zeros((len(features), max_len), dtype=torch.long)
+        for row, feature in enumerate(features):
+            ids = feature["input_ids"]
+            length = ids.size(0)
+            input_ids[row, :length] = ids
+            labels[row, :length] = feature["labels"]
+            pad_positions = ids == self.pad_token_id
+            labels[row, :length][pad_positions] = -100
+            attention_mask[row, :length] = 1
+        return {"input_ids": input_ids, "labels": labels, "attention_mask": attention_mask}

delta/generator.py

@@ -0,0 +1,197 @@
+"""Text generation and chat helpers for Delta Ultra Mini."""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+from typing import Any, Generator
+
+import torch
+from torch.nn import functional as F
+
+from delta.identity import identity_response
+from delta.model import DeltaConfig, DeltaModel
+from delta.tokenizer import DEFAULT_SYSTEM_PROMPT, DeltaTokenizer
+
+logging.basicConfig(level=os.getenv("DELTA_LOG_LEVEL", "INFO").upper())
+logger = logging.getLogger(__name__)
+
+
+class DeltaGenerator:
+    """Autoregressive generator for Delta Ultra Mini."""
+
+    def __init__(self, model: DeltaModel, tokenizer: DeltaTokenizer, device: str | torch.device | None = None) -> None:
+        self.device = torch.device(device or ("cuda" if torch.cuda.is_available() else "cpu"))
+        self.model = model.to(self.device)
+        self.model.eval()
+        self.tokenizer = tokenizer
+
+    @classmethod
+    def from_files(
+        cls,
+        checkpoint_path: str | Path,
+        tokenizer_path: str | Path,
+        config_path: str | Path | None = None,
+        device: str | torch.device | None = None,
+    ) -> "DeltaGenerator":
+        """Create a generator from checkpoint, tokenizer, and optional config."""
+
+        checkpoint = torch.load(checkpoint_path, map_location="cpu")
+        config_data = checkpoint.get("config")
+        config = DeltaConfig.from_dict(config_data) if config_data else DeltaConfig.from_json(config_path or "configs/ultra_mini.json")
+        model = DeltaModel(config)
+        state = checkpoint.get("model_state_dict", checkpoint)
+        model.load_state_dict(state)
+        return cls(model=model, tokenizer=DeltaTokenizer(tokenizer_path), device=device)
+
+    def _filter_logits(
+        self,
+        logits: torch.Tensor,
+        temperature: float,
+        top_k: int,
+        top_p: float,
+    ) -> torch.Tensor:
+        """Apply temperature, top-k, and nucleus filtering."""
+
+        if temperature <= 0:
+            return logits
+        logits = logits / temperature
+        if top_k > 0:
+            values, _ = torch.topk(logits, min(top_k, logits.size(-1)))
+            logits = logits.masked_fill(logits < values[:, [-1]], torch.finfo(logits.dtype).min)
+        if 0.0 < top_p < 1.0:
+            sorted_logits, sorted_indices = torch.sort(logits, descending=True)
+            sorted_probs = F.softmax(sorted_logits, dim=-1)
+            cumulative = sorted_probs.cumsum(dim=-1)
+            remove = cumulative > top_p
+            remove[..., 1:] = remove[..., :-1].clone()
+            remove[..., 0] = False
+            indices_to_remove = remove.scatter(1, sorted_indices, remove)
+            logits = logits.masked_fill(indices_to_remove, torch.finfo(logits.dtype).min)
+        return logits
+
+    def _apply_repetition_penalty(
+        self,
+        logits: torch.Tensor,
+        generated: torch.Tensor,
+        repetition_penalty: float,
+    ) -> torch.Tensor:
+        """Penalize tokens that already appeared in the generated sequence."""
+
+        if repetition_penalty == 1.0:
+            return logits
+        for token_id in set(generated[0].tolist()):
+            score = logits[:, token_id]
+            logits[:, token_id] = torch.where(score < 0, score * repetition_penalty, score / repetition_penalty)
+        return logits
+
+    def _clean_completion_text(self, text: str) -> str:
+        """Trim leaked prompt/chat markers from decoded assistant text."""
+
+        cut_markers = ("[SYS]", "[USR]", "[ASS]", "[SEP]", DEFAULT_SYSTEM_PROMPT)
+        clean = text
+        for marker in cut_markers:
+            index = clean.find(marker)
+            if index >= 0:
+                clean = clean[:index]
+        return clean.strip()
+
+    @torch.inference_mode()
+    def generate(
+        self,
+        input_ids: list[int] | torch.Tensor,
+        max_new_tokens: int = 256,
+        temperature: float = 0.2,
+        top_k: int = 20,
+        top_p: float = 0.9,
+        repetition_penalty: float = 1.08,
+    ) -> list[int]:
+        """Generate token ids using KV cache and manual sampling."""
+
+        ids = torch.tensor([input_ids], dtype=torch.long, device=self.device) if isinstance(input_ids, list) else input_ids.to(self.device)
+        if ids.dim() == 1:
+            ids = ids.unsqueeze(0)
+        ids = ids[:, -(self.model.config.max_seq_len - 1) :]
+        max_new_tokens = min(max_new_tokens, self.model.config.max_seq_len - ids.size(1))
+        generated = ids.clone()
+        past_key_values = None
+        next_input = ids
+        stop_token_ids = self.tokenizer.chat_stop_token_ids
+        for _ in range(max_new_tokens):
+            outputs = self.model(next_input, past_key_values=past_key_values, use_cache=True)
+            logits = outputs["logits"][:, -1, :]
+            past_key_values = outputs["past_key_values"]
+            logits = self._apply_repetition_penalty(logits, generated, repetition_penalty)
+            if temperature <= 0:
+                next_token = torch.argmax(logits, dim=-1, keepdim=True)
+            else:
+                filtered = self._filter_logits(logits, temperature=temperature, top_k=top_k, top_p=top_p)
+                next_token = torch.multinomial(F.softmax(filtered, dim=-1), num_samples=1)
+            if int(next_token.item()) in stop_token_ids:
+                break
+            generated = torch.cat((generated, next_token), dim=1)
+            next_input = next_token
+        return generated[0].tolist()
+
+    def chat(self, messages: list[dict[str, Any]], persona: str | None = None, **gen_kwargs: Any) -> str:
+        """Generate an assistant response for chat messages."""
+
+        latest_user = next((str(m.get("content", "")) for m in reversed(messages) if m.get("role") == "user"), "")
+        intercepted = identity_response(latest_user)
+        if intercepted is not None:
+            return intercepted
+        prompt = self.tokenizer.format_chat(messages, persona=persona)
+        input_ids = self.tokenizer.encode(prompt, add_special_tokens=False)
+        input_ids = input_ids[-(self.model.config.max_seq_len - 1) :]
+        output_ids = self.generate(input_ids, **gen_kwargs)
+        new_ids = output_ids[len(input_ids) :]
+        text = self.tokenizer.decode(new_ids, skip_special_tokens=True)
+        return self._clean_completion_text(text)
+
+    def stream_chat(
+        self,
+        messages: list[dict[str, Any]],
+        persona: str | None = None,
+        **gen_kwargs: Any,
+    ) -> Generator[str, None, None]:
+        """Yield generated text token by token."""
+
+        latest_user = next((str(m.get("content", "")) for m in reversed(messages) if m.get("role") == "user"), "")
+        intercepted = identity_response(latest_user)
+        if intercepted is not None:
+            yield intercepted
+            return
+
+        prompt = self.tokenizer.format_chat(messages, persona=persona)
+        input_ids = self.tokenizer.encode(prompt, add_special_tokens=False)
+        input_ids = input_ids[-(self.model.config.max_seq_len - 1) :]
+        ids = torch.tensor([input_ids], dtype=torch.long, device=self.device)
+        generated = ids.clone()
+        past_key_values = None
+        next_input = ids
+        max_new_tokens = int(gen_kwargs.get("max_new_tokens", 256))
+        max_new_tokens = min(max_new_tokens, self.model.config.max_seq_len - ids.size(1))
+        temperature = float(gen_kwargs.get("temperature", 0.2))
+        top_k = int(gen_kwargs.get("top_k", 20))
+        top_p = float(gen_kwargs.get("top_p", 0.9))
+        repetition_penalty = float(gen_kwargs.get("repetition_penalty", 1.08))
+        stop_token_ids = self.tokenizer.chat_stop_token_ids
+        with torch.inference_mode():
+            for _ in range(max_new_tokens):
+                outputs = self.model(next_input, past_key_values=past_key_values, use_cache=True)
+                logits = outputs["logits"][:, -1, :]
+                past_key_values = outputs["past_key_values"]
+                logits = self._apply_repetition_penalty(logits, generated, repetition_penalty)
+                if temperature <= 0:
+                    next_token = torch.argmax(logits, dim=-1, keepdim=True)
+                else:
+                    filtered = self._filter_logits(logits, temperature=temperature, top_k=top_k, top_p=top_p)
+                    next_token = torch.multinomial(F.softmax(filtered, dim=-1), num_samples=1)
+                if int(next_token.item()) in stop_token_ids:
+                    return
+                generated = torch.cat((generated, next_token), dim=1)
+                next_input = next_token
+                text = self.tokenizer.decode([int(next_token.item())], skip_special_tokens=True)
+                if text:
+                    yield text

delta/identity.py

@@ -0,0 +1,93 @@
+"""Hardcoded identity helpers for Delta Ultra Mini.
+
+This module keeps the immutable public identity of the model in one place and
+provides a small interceptor for identity questions. The interceptor is used
+before neural generation so the model never has to infer facts about itself.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import re
+from typing import Any
+
+logging.basicConfig(level=os.getenv("DELTA_LOG_LEVEL", "INFO").upper())
+logger = logging.getLogger(__name__)
+
+_IDENTITY: dict[str, str] = {
+    "name": "Delta",
+    "version": "Ultra Mini",
+    "created_by": "Flame Corporation",
+    "parameters": "~50 milhões",
+    "purpose": "Assistente conversacional inteligente via API",
+}
+
+_IDENTITY_PATTERNS: tuple[re.Pattern[str], ...] = tuple(
+    re.compile(pattern, flags=re.IGNORECASE)
+    for pattern in (
+        r"qual.*seu nome",
+        r"who are you",
+        r"quem.*voc[eê]",
+        r"sua vers[aã]o",
+        r"your version",
+        r"quem.*criou",
+        r"who made you",
+        r"quantos par[aâ]metros",
+        r"how many parameters",
+    )
+)
+
+
+def get_identity() -> dict[str, str]:
+    """Return Delta's immutable identity.
+
+    Returns:
+        A copy of the identity dictionary.
+    """
+
+    return dict(_IDENTITY)
+
+
+def is_identity_question(text: str) -> bool:
+    """Check whether a user message asks about Delta's identity.
+
+    Args:
+        text: User-provided text.
+
+    Returns:
+        True when the text matches one of the hardcoded identity regexes.
+    """
+
+    return any(pattern.search(text) for pattern in _IDENTITY_PATTERNS)
+
+
+def identity_response(text: str) -> str | None:
+    """Return a deterministic identity answer when the input asks for it.
+
+    Args:
+        text: User-provided text.
+
+    Returns:
+        A precise identity answer, or None when the input should go to the
+        neural model.
+    """
+
+    if not is_identity_question(text):
+        return None
+
+    identity: dict[str, Any] = get_identity()
+    lowered = text.lower()
+    if re.search(r"quantos par[aâ]metros|how many parameters", lowered):
+        return f"Eu sou {identity['name']} {identity['version']} e tenho {identity['parameters']} de parâmetros."
+    if re.search(r"sua vers[aã]o|your version", lowered):
+        return f"Minha versão é {identity['version']}."
+    if re.search(r"quem.*criou|who made you", lowered):
+        return f"Fui criada pela {identity['created_by']}."
+    if re.search(r"qual.*seu nome", lowered):
+        return f"Meu nome é {identity['name']}."
+    return (
+        f"Eu sou {identity['name']} {identity['version']}, criada pela "
+        f"{identity['created_by']}. Meu propósito é ser um "
+        f"{identity['purpose'].lower()}."
+    )

delta/model.py

@@ -0,0 +1,331 @@
+"""PyTorch implementation of the Delta Ultra Mini decoder-only Transformer."""
+
+from __future__ import annotations
+
+import json
+import logging
+import math
+import os
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any
+
+import torch
+from torch import nn
+from torch.nn import functional as F
+
+logging.basicConfig(level=os.getenv("DELTA_LOG_LEVEL", "INFO").upper())
+logger = logging.getLogger(__name__)
+
+
+@dataclass(slots=True)
+class DeltaConfig:
+    """Configuration for Delta Ultra Mini.
+
+    Attributes:
+        vocab_size: Token vocabulary size.
+        d_model: Embedding and hidden size.
+        n_heads: Number of attention heads.
+        n_layers: Number of decoder blocks.
+        d_ff: Feed-forward hidden size.
+        max_seq_len: Maximum context length.
+        dropout: Dropout probability.
+        tie_embeddings: Whether output projection shares token embedding weight.
+        pad_token_id: Padding token id.
+        bos_token_id: Beginning-of-sequence token id.
+        eos_token_id: End-of-sequence token id.
+    """
+
+    vocab_size: int = 32000
+    d_model: int = 512
+    n_heads: int = 8
+    n_layers: int = 8
+    d_ff: int = 2048
+    max_seq_len: int = 512
+    dropout: float = 0.1
+    tie_embeddings: bool = True
+    pad_token_id: int = 0
+    bos_token_id: int = 2
+    eos_token_id: int = 3
+    use_cache: bool = True
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "DeltaConfig":
+        """Build a config from a dictionary."""
+
+        valid = {field for field in cls.__dataclass_fields__}
+        return cls(**{key: value for key, value in data.items() if key in valid})
+
+    @classmethod
+    def from_json(cls, path: str | Path) -> "DeltaConfig":
+        """Load a config from a JSON file."""
+
+        with Path(path).open("r", encoding="utf-8") as handle:
+            return cls.from_dict(json.load(handle))
+
+    def to_dict(self) -> dict[str, Any]:
+        """Serialize config to a dictionary."""
+
+        return asdict(self)
+
+
+class RMSNorm(nn.Module):
+    """Root Mean Square normalization without mean-centering."""
+
+    def __init__(self, dim: int, eps: float = 1e-6) -> None:
+        super().__init__()
+        self.eps = eps
+        self.weight = nn.Parameter(torch.ones(dim))
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        """Normalize the last dimension of x."""
+
+        normed = x * torch.rsqrt(x.pow(2).mean(dim=-1, keepdim=True) + self.eps)
+        return normed * self.weight
+
+
+class RotaryEmbedding(nn.Module):
+    """Rotary positional embedding cache for attention heads."""
+
+    def __init__(self, dim: int, max_seq_len: int = 512, base: float = 10000.0) -> None:
+        super().__init__()
+        inv_freq = 1.0 / (base ** (torch.arange(0, dim, 2).float() / dim))
+        positions = torch.arange(max_seq_len, dtype=torch.float)
+        freqs = torch.outer(positions, inv_freq)
+        emb = torch.cat((freqs, freqs), dim=-1)
+        self.register_buffer("cos_cached", emb.cos()[None, None, :, :], persistent=False)
+        self.register_buffer("sin_cached", emb.sin()[None, None, :, :], persistent=False)
+
+    def forward(self, seq_len: int, offset: int = 0) -> tuple[torch.Tensor, torch.Tensor]:
+        """Return cosine and sine caches for a sequence span."""
+
+        end = offset + seq_len
+        return self.cos_cached[:, :, offset:end, :], self.sin_cached[:, :, offset:end, :]
+
+
+def _rotate_half(x: torch.Tensor) -> torch.Tensor:
+    """Rotate pairs of hidden dimensions for RoPE."""
+
+    x1, x2 = x.chunk(2, dim=-1)
+    return torch.cat((-x2, x1), dim=-1)
+
+
+def apply_rotary(x: torch.Tensor, cos: torch.Tensor, sin: torch.Tensor) -> torch.Tensor:
+    """Apply rotary embedding to q or k tensors."""
+
+    return (x * cos) + (_rotate_half(x) * sin)
+
+
+class CausalSelfAttention(nn.Module):
+    """Multi-head causal self-attention with optional KV cache."""
+
+    def __init__(self, config: DeltaConfig) -> None:
+        super().__init__()
+        if config.d_model % config.n_heads != 0:
+            raise ValueError("d_model must be divisible by n_heads")
+        self.n_heads = config.n_heads
+        self.head_dim = config.d_model // config.n_heads
+        self.qkv_proj = nn.Linear(config.d_model, 3 * config.d_model, bias=False)
+        self.out_proj = nn.Linear(config.d_model, config.d_model, bias=False)
+        self.dropout = nn.Dropout(config.dropout)
+        self.rope = RotaryEmbedding(self.head_dim, config.max_seq_len)
+        mask = torch.tril(torch.ones(config.max_seq_len, config.max_seq_len, dtype=torch.bool))
+        self.register_buffer("causal_mask", mask, persistent=False)
+
+    def forward(
+        self,
+        x: torch.Tensor,
+        past_key_value: tuple[torch.Tensor, torch.Tensor] | None = None,
+        use_cache: bool = False,
+    ) -> tuple[torch.Tensor, tuple[torch.Tensor, torch.Tensor] | None]:
+        """Run attention.
+
+        Args:
+            x: Input tensor of shape (batch, seq, hidden).
+            past_key_value: Optional cached key and value tensors.
+            use_cache: Whether to return a new cache.
+
+        Returns:
+            Attention output and optional key/value cache.
+        """
+
+        batch_size, seq_len, hidden_size = x.shape
+        qkv = self.qkv_proj(x)
+        q, k, v = qkv.split(hidden_size, dim=-1)
+        q = q.view(batch_size, seq_len, self.n_heads, self.head_dim).transpose(1, 2)
+        k = k.view(batch_size, seq_len, self.n_heads, self.head_dim).transpose(1, 2)
+        v = v.view(batch_size, seq_len, self.n_heads, self.head_dim).transpose(1, 2)
+
+        past_len = 0 if past_key_value is None else past_key_value[0].size(2)
+        cos, sin = self.rope(seq_len, offset=past_len)
+        q = apply_rotary(q, cos.to(q.device, q.dtype), sin.to(q.device, q.dtype))
+        k = apply_rotary(k, cos.to(k.device, k.dtype), sin.to(k.device, k.dtype))
+
+        if past_key_value is not None:
+            past_k, past_v = past_key_value
+            k = torch.cat((past_k, k), dim=2)
+            v = torch.cat((past_v, v), dim=2)
+
+        present = (k, v) if use_cache else None
+        attn_scores = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.head_dim)
+        total_len = k.size(2)
+        if past_len == 0:
+            mask = self.causal_mask[:seq_len, :total_len]
+            attn_scores = attn_scores.masked_fill(~mask[None, None, :, :], torch.finfo(attn_scores.dtype).min)
+        attn_weights = F.softmax(attn_scores, dim=-1)
+        attn_weights = self.dropout(attn_weights)
+        y = torch.matmul(attn_weights, v)
+        y = y.transpose(1, 2).contiguous().view(batch_size, seq_len, hidden_size)
+        return self.out_proj(y), present
+
+
+class SwiGLUFeedForward(nn.Module):
+    """SwiGLU feed-forward network."""
+
+    def __init__(self, config: DeltaConfig) -> None:
+        super().__init__()
+        self.gate_proj = nn.Linear(config.d_model, config.d_ff, bias=False)
+        self.up_proj = nn.Linear(config.d_model, config.d_ff, bias=False)
+        self.down_proj = nn.Linear(config.d_ff, config.d_model, bias=False)
+        self.dropout = nn.Dropout(config.dropout)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        """Apply SwiGLU transformation."""
+
+        return self.down_proj(self.dropout(F.silu(self.gate_proj(x)) * self.up_proj(x)))
+
+
+class DeltaDecoderBlock(nn.Module):
+    """One Delta decoder block: RMSNorm, attention, RMSNorm, SwiGLU FFN."""
+
+    def __init__(self, config: DeltaConfig) -> None:
+        super().__init__()
+        self.attn_norm = RMSNorm(config.d_model)
+        self.attn = CausalSelfAttention(config)
+        self.ffn_norm = RMSNorm(config.d_model)
+        self.ffn = SwiGLUFeedForward(config)
+
+    def forward(
+        self,
+        x: torch.Tensor,
+        past_key_value: tuple[torch.Tensor, torch.Tensor] | None = None,
+        use_cache: bool = False,
+    ) -> tuple[torch.Tensor, tuple[torch.Tensor, torch.Tensor] | None]:
+        """Run one decoder block."""
+
+        attn_out, present = self.attn(self.attn_norm(x), past_key_value=past_key_value, use_cache=use_cache)
+        x = x + attn_out
+        x = x + self.ffn(self.ffn_norm(x))
+        return x, present
+
+
+class DeltaModel(nn.Module):
+    """Delta Ultra Mini causal language model."""
+
+    def __init__(self, config: DeltaConfig | dict[str, Any] | None = None) -> None:
+        super().__init__()
+        self.config = DeltaConfig.from_dict(config) if isinstance(config, dict) else (config or DeltaConfig())
+        self.embed_tokens = nn.Embedding(self.config.vocab_size, self.config.d_model)
+        self.drop = nn.Dropout(self.config.dropout)
+        self.layers = nn.ModuleList(DeltaDecoderBlock(self.config) for _ in range(self.config.n_layers))
+        self.norm = RMSNorm(self.config.d_model)
+        self.lm_head = nn.Linear(self.config.d_model, self.config.vocab_size, bias=False)
+        if self.config.tie_embeddings:
+            self.lm_head.weight = self.embed_tokens.weight
+        self.apply(self._init_weights)
+        total_params = self.num_parameters()
+        logger.info("DeltaModel initialized with %s parameters", f"{total_params:,}")
+        print(f"DeltaModel parameters: {total_params:,}")
+        if not 48_000_000 <= total_params <= 55_000_000:
+            raise ValueError(f"Delta Ultra Mini must have 48M-55M parameters, got {total_params:,}")
+
+    def _init_weights(self, module: nn.Module) -> None:
+        """Initialize weights with GPT-style normal initialization."""
+
+        if isinstance(module, nn.Linear):
+            nn.init.normal_(module.weight, mean=0.0, std=0.02)
+            if module.bias is not None:
+                nn.init.zeros_(module.bias)
+        elif isinstance(module, nn.Embedding):
+            nn.init.normal_(module.weight, mean=0.0, std=0.02)
+
+    def num_parameters(self, only_trainable: bool = True, exclude_embeddings: bool = False) -> int:
+        """Return the number of model parameters.
+
+        Args:
+            only_trainable: Count only parameters with requires_grad.
+            exclude_embeddings: Exclude embedding parameters for Trainer FLOPs estimates.
+        """
+
+        total = 0
+        for name, parameter in self.named_parameters():
+            if only_trainable and not parameter.requires_grad:
+                continue
+            if exclude_embeddings and "embed_tokens" in name:
+                continue
+            total += parameter.numel()
+        return total
+
+    def forward(
+        self,
+        input_ids: torch.Tensor,
+        labels: torch.Tensor | None = None,
+        past_key_values: list[tuple[torch.Tensor, torch.Tensor]] | None = None,
+        use_cache: bool = False,
+        **_: Any,
+    ) -> dict[str, torch.Tensor | list[tuple[torch.Tensor, torch.Tensor]] | None]:
+        """Run causal language modeling forward pass."""
+
+        if input_ids.size(1) > self.config.max_seq_len:
+            input_ids = input_ids[:, -self.config.max_seq_len :]
+            if labels is not None:
+                labels = labels[:, -self.config.max_seq_len :]
+        x = self.drop(self.embed_tokens(input_ids))
+        next_cache: list[tuple[torch.Tensor, torch.Tensor]] = []
+        for index, layer in enumerate(self.layers):
+            past = None if past_key_values is None else past_key_values[index]
+            x, present = layer(x, past_key_value=past, use_cache=use_cache)
+            if present is not None:
+                next_cache.append(present)
+        logits = self.lm_head(self.norm(x))
+        loss = None
+        if labels is not None:
+            shift_logits = logits[:, :-1, :].contiguous()
+            shift_labels = labels[:, 1:].contiguous()
+            loss = F.cross_entropy(
+                shift_logits.view(-1, shift_logits.size(-1)),
+                shift_labels.view(-1),
+                ignore_index=-100,
+            )
+        return {"loss": loss, "logits": logits, "past_key_values": next_cache if use_cache else None}
+
+    def save_checkpoint(
+        self,
+        path: str | Path,
+        optimizer: torch.optim.Optimizer | None = None,
+        scheduler: Any | None = None,
+        step: int = 0,
+    ) -> None:
+        """Save a full training checkpoint."""
+
+        checkpoint: dict[str, Any] = {
+            "model_state_dict": self.state_dict(),
+            "step": step,
+            "config": self.config.to_dict(),
+        }
+        if optimizer is not None:
+            checkpoint["optimizer_state_dict"] = optimizer.state_dict()
+        if scheduler is not None:
+            checkpoint["scheduler_state_dict"] = scheduler.state_dict()
+        path = Path(path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        torch.save(checkpoint, path)
+
+    @classmethod
+    def load_checkpoint(cls, path: str | Path, map_location: str | torch.device = "cpu") -> "DeltaModel":
+        """Load a model from a checkpoint file."""
+
+        checkpoint = torch.load(path, map_location=map_location)
+        model = cls(DeltaConfig.from_dict(checkpoint["config"]))
+        model.load_state_dict(checkpoint["model_state_dict"])
+        return model

delta/tokenizer.py

@@ -0,0 +1,137 @@
+"""BPE tokenizer training and chat formatting for Delta Ultra Mini."""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+from typing import Any
+
+from tokenizers import Tokenizer
+from tokenizers.decoders import ByteLevel as ByteLevelDecoder
+from tokenizers.models import BPE
+from tokenizers.pre_tokenizers import ByteLevel
+from tokenizers.processors import TemplateProcessing
+from tokenizers.trainers import BpeTrainer
+
+logging.basicConfig(level=os.getenv("DELTA_LOG_LEVEL", "INFO").upper())
+logger = logging.getLogger(__name__)
+
+SPECIAL_TOKENS: list[str] = ["[PAD]", "[UNK]", "[BOS]", "[EOS]", "[SYS]", "[USR]", "[ASS]", "[SEP]"]
+DEFAULT_SYSTEM_PROMPT = (
+    "Você é Delta, assistente criada pela Flame Corporation. "
+    "Responda de forma clara, útil e amigável."
+)
+
+
+def train_tokenizer(corpus_files: list[str] | list[Path], output_path: str | Path) -> None:
+    """Train a BPE tokenizer from raw text files.
+
+    Args:
+        corpus_files: Paths to corpus files.
+        output_path: Destination tokenizer JSON path.
+    """
+
+    tokenizer = Tokenizer(BPE(unk_token="[UNK]"))
+    tokenizer.pre_tokenizer = ByteLevel(add_prefix_space=False)
+    tokenizer.decoder = ByteLevelDecoder()
+    trainer = BpeTrainer(vocab_size=32000, special_tokens=SPECIAL_TOKENS, show_progress=True)
+    tokenizer.train([str(path) for path in corpus_files], trainer)
+    tokenizer.post_processor = TemplateProcessing(
+        single="[BOS] $A [EOS]",
+        pair="[BOS] $A [SEP] $B [EOS]",
+        special_tokens=[
+            ("[BOS]", tokenizer.token_to_id("[BOS]")),
+            ("[EOS]", tokenizer.token_to_id("[EOS]")),
+            ("[SEP]", tokenizer.token_to_id("[SEP]")),
+        ],
+    )
+    output = Path(output_path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    tokenizer.save(str(output))
+    logger.info("Tokenizer saved to %s", output)
+
+
+def load_tokenizer(path: str | Path) -> "DeltaTokenizer":
+    """Load a DeltaTokenizer from disk."""
+
+    return DeltaTokenizer(path)
+
+
+class DeltaTokenizer:
+    """Thin wrapper around HuggingFace tokenizers.Tokenizer."""
+
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path)
+        self.tokenizer = Tokenizer.from_file(str(self.path))
+        self.pad_token_id = self.tokenizer.token_to_id("[PAD]")
+        self.unk_token_id = self.tokenizer.token_to_id("[UNK]")
+        self.bos_token_id = self.tokenizer.token_to_id("[BOS]")
+        self.eos_token_id = self.tokenizer.token_to_id("[EOS]")
+        self.sys_token_id = self.tokenizer.token_to_id("[SYS]")
+        self.usr_token_id = self.tokenizer.token_to_id("[USR]")
+        self.ass_token_id = self.tokenizer.token_to_id("[ASS]")
+        self.sep_token_id = self.tokenizer.token_to_id("[SEP]")
+        self.special_tokens = set(SPECIAL_TOKENS)
+
+    @property
+    def chat_stop_token_ids(self) -> set[int]:
+        """Token ids that should end an assistant completion."""
+
+        return {
+            token_id
+            for token_id in (
+                self.eos_token_id,
+                self.sep_token_id,
+                self.sys_token_id,
+                self.usr_token_id,
+                self.ass_token_id,
+                self.bos_token_id,
+                self.pad_token_id,
+            )
+            if token_id is not None
+        }
+
+    def encode(self, text: str, add_special_tokens: bool = True) -> list[int]:
+        """Encode a string into token ids."""
+
+        return self.tokenizer.encode(text, add_special_tokens=add_special_tokens).ids
+
+    def decode(self, ids: list[int], skip_special_tokens: bool = True) -> str:
+        """Decode token ids into text."""
+
+        return self.tokenizer.decode(ids, skip_special_tokens=skip_special_tokens)
+
+    def batch_encode(self, texts: list[str], add_special_tokens: bool = True) -> list[list[int]]:
+        """Encode a batch of strings."""
+
+        return [encoding.ids for encoding in self.tokenizer.encode_batch(texts, add_special_tokens=add_special_tokens)]
+
+    def format_chat(self, messages: list[dict[str, Any]], persona: str | None = None) -> str:
+        """Format a multi-turn conversation for Delta.
+
+        Args:
+            messages: Conversation turns with role and content.
+            persona: Optional system prompt.
+
+        Returns:
+            Prompt text ending with an assistant tag for continuation.
+        """
+
+        system = persona or DEFAULT_SYSTEM_PROMPT
+        parts = [f"[SYS] {system} [SEP]"]
+        for message in messages:
+            role = str(message.get("role", "")).lower()
+            content = str(message.get("content", "")).strip()
+            if role == "user":
+                parts.append(f"[USR] {content} [SEP]")
+            elif role == "assistant":
+                parts.append(f"[ASS] {content} [SEP]")
+            elif role == "system":
+                parts[0] = f"[SYS] {content} [SEP]"
+        if not parts[-1].startswith("[ASS]"):
+            parts.append("[ASS]")
+        else:
+            parts.append("[USR]  [SEP]")
+            parts.append("[ASS]")
+        return "\n".join(parts)

delta/trainer.py

@@ -0,0 +1,106 @@
+"""Training utilities for Delta Ultra Mini using HuggingFace Trainer."""
+
+from __future__ import annotations
+
+import logging
+import os
+import json
+from pathlib import Path
+from typing import Any
+
+import torch
+from transformers import Trainer, TrainingArguments
+from transformers.utils import SAFE_WEIGHTS_NAME
+
+from delta.dataset import DeltaDataCollator, DeltaDataset
+from delta.model import DeltaConfig, DeltaModel
+from delta.tokenizer import load_tokenizer
+
+logging.basicConfig(level=os.getenv("DELTA_LOG_LEVEL", "INFO").upper())
+logger = logging.getLogger(__name__)
+
+
+class DeltaTrainer(Trainer):
+    """Trainer override that saves tied weights safely with safetensors."""
+
+    def _save(self, output_dir: str | None = None, state_dict: dict[str, torch.Tensor] | None = None) -> None:
+        output_dir = output_dir or self.args.output_dir
+        os.makedirs(output_dir, exist_ok=True)
+
+        state_dict = state_dict or self.model.state_dict()
+        weights_path = os.path.join(output_dir, SAFE_WEIGHTS_NAME)
+
+        try:
+            import safetensors.torch
+
+            safetensors.torch.save_file(state_dict, weights_path, metadata={"format": "pt"})
+        except RuntimeError as exc:
+            # When embeddings are tied, tensors share the same underlying storage.
+            # `save_file(state_dict, ...)` refuses that; `save_model(model, ...)` handles it.
+            if "share memory" not in str(exc):
+                raise
+            import safetensors.torch
+
+            safetensors.torch.save_model(self.model, weights_path, metadata={"format": "pt"})
+
+
+def train(config_dict: dict[str, Any]) -> Trainer:
+    """Train Delta Ultra Mini.
+
+    Args:
+        config_dict: Training and model configuration.
+
+    Returns:
+        The configured Trainer after training.
+    """
+
+    data_path = Path(config_dict["data_path"])
+    output_dir = Path(config_dict["output_dir"])
+    tokenizer_path = Path(config_dict.get("tokenizer_path", output_dir / "tokenizer.json"))
+    epochs = float(config_dict.get("epochs", 1))
+    batch_size = int(config_dict.get("batch_size", 2))
+    resume_from_checkpoint = config_dict.get("resume_from_checkpoint")
+    model_config = DeltaConfig.from_dict(config_dict.get("model", config_dict))
+    tokenizer = load_tokenizer(tokenizer_path)
+    dataset = DeltaDataset(
+        data_path=data_path,
+        tokenizer=tokenizer,
+        max_seq_len=model_config.max_seq_len,
+        stride=int(config_dict.get("stride", 256)),
+    )
+    model = DeltaModel(model_config)
+    training_args = TrainingArguments(
+        output_dir=str(output_dir),
+        num_train_epochs=epochs,
+        per_device_train_batch_size=batch_size,
+        learning_rate=float(config_dict.get("learning_rate", 3e-4)),
+        weight_decay=float(config_dict.get("weight_decay", 0.01)),
+        warmup_steps=int(config_dict.get("warmup_steps", 500)),
+        lr_scheduler_type="cosine",
+        max_grad_norm=float(config_dict.get("max_grad_norm", 1.0)),
+        fp16=torch.cuda.is_available(),
+        save_steps=int(config_dict.get("save_steps", 1000)),
+        logging_steps=int(config_dict.get("logging_steps", 50)),
+        report_to=[],
+        remove_unused_columns=False,
+        dataloader_pin_memory=torch.cuda.is_available(),
+    )
+    trainer = DeltaTrainer(
+        model=model,
+        args=training_args,
+        train_dataset=dataset,
+        data_collator=DeltaDataCollator(tokenizer.pad_token_id),
+    )
+    trainer.train(resume_from_checkpoint=resume_from_checkpoint)
+    trainer.save_model(str(output_dir))
+    output_dir.mkdir(parents=True, exist_ok=True)
+    model.save_checkpoint(
+        output_dir / "delta_checkpoint.pt",
+        optimizer=trainer.optimizer,
+        scheduler=trainer.lr_scheduler,
+        step=int(trainer.state.global_step),
+    )
+    with (output_dir / "config.json").open("w", encoding="utf-8") as handle:
+        json.dump(model_config.to_dict(), handle, ensure_ascii=False, indent=2)
+    logger.info("Saved Delta checkpoint to %s", output_dir / "delta_checkpoint.pt")
+    return trainer

delta_checkpoint.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6ff56db700e129603ddb8d75ed058a2aa6d707c59f176e643238913229e2ee10
+size 599439053

model.safetensors

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:37ae14167b33ca2943e5a665a2e04acc6d7718694cff13900544e1b1fd42ac4f
+size 199794416

requirements.txt

@@ -0,0 +1,4 @@
+torch
+tokenizers
+transformers
+accelerate>=1.1.0

scripts/generate_delta.py

@@ -0,0 +1,58 @@
+"""Command line local inference for Delta Ultra Mini without the REST API."""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import os
+import sys
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from delta.generator import DeltaGenerator
+
+logging.basicConfig(level=os.getenv("DELTA_LOG_LEVEL", "INFO").upper())
+logger = logging.getLogger(__name__)
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse command line arguments."""
+
+    parser = argparse.ArgumentParser(description="Generate text with Delta Ultra Mini.")
+    parser.add_argument("--prompt", required=True, help="User prompt.")
+    parser.add_argument("--checkpoint_path", default="runs/delta-ultra-mini/delta_checkpoint.pt")
+    parser.add_argument("--tokenizer_path", default="tokenizer.json")
+    parser.add_argument("--config_path", default="configs/ultra_mini.json")
+    parser.add_argument("--max_new_tokens", type=int, default=128)
+    parser.add_argument("--temperature", type=float, default=0.2)
+    parser.add_argument("--top_k", type=int, default=20)
+    parser.add_argument("--top_p", type=float, default=0.9)
+    parser.add_argument("--repetition_penalty", type=float, default=1.08)
+    return parser.parse_args()
+
+
+def main() -> None:
+    """Load the checkpoint and print one assistant response."""
+
+    args = parse_args()
+    generator = DeltaGenerator.from_files(
+        checkpoint_path=args.checkpoint_path,
+        tokenizer_path=args.tokenizer_path,
+        config_path=args.config_path,
+    )
+    response = generator.chat(
+        [{"role": "user", "content": args.prompt}],
+        max_new_tokens=args.max_new_tokens,
+        temperature=args.temperature,
+        top_k=args.top_k,
+        top_p=args.top_p,
+        repetition_penalty=args.repetition_penalty,
+    )
+    print(response)
+
+
+if __name__ == "__main__":
+    main()

scripts/train_delta.py

@@ -0,0 +1,58 @@
+"""Command line training entrypoint for Delta Ultra Mini."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import os
+import sys
+from pathlib import Path
+from typing import Any
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from delta.trainer import train
+
+logging.basicConfig(level=os.getenv("DELTA_LOG_LEVEL", "INFO").upper())
+logger = logging.getLogger(__name__)
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse training arguments."""
+
+    parser = argparse.ArgumentParser(description="Train Delta Ultra Mini.")
+    parser.add_argument("--data_path", required=True, help="Directory or file containing .txt/.jsonl training data.")
+    parser.add_argument("--output_dir", required=True, help="Output directory for checkpoints.")
+    parser.add_argument("--epochs", type=float, default=1.0, help="Number of training epochs.")
+    parser.add_argument("--batch_size", type=int, default=2, help="Per-device train batch size.")
+    parser.add_argument("--resume_from_checkpoint", default=None, help="Checkpoint path or true to resume latest.")
+    parser.add_argument("--tokenizer_path", default=None, help="Path to tokenizer.json.")
+    parser.add_argument("--config_path", default="configs/ultra_mini.json", help="Model config JSON.")
+    return parser.parse_args()
+
+
+def main() -> None:
+    """Run Trainer-based model training."""
+
+    args = parse_args()
+    with Path(args.config_path).open("r", encoding="utf-8") as handle:
+        model_config: dict[str, Any] = json.load(handle)
+    output_dir = Path(args.output_dir)
+    config = {
+        "data_path": args.data_path,
+        "output_dir": str(output_dir),
+        "epochs": args.epochs,
+        "batch_size": args.batch_size,
+        "resume_from_checkpoint": args.resume_from_checkpoint,
+        "tokenizer_path": args.tokenizer_path or str(output_dir / "tokenizer.json"),
+        "model": model_config,
+    }
+    train(config)
+    logger.info("Training complete.")
+
+
+if __name__ == "__main__":
+    main()

scripts/train_tokenizer.py

@@ -0,0 +1,39 @@
+"""Command line tokenizer trainer for Delta Ultra Mini."""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import os
+import sys
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from delta.tokenizer import train_tokenizer
+
+logging.basicConfig(level=os.getenv("DELTA_LOG_LEVEL", "INFO").upper())
+logger = logging.getLogger(__name__)
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse command line arguments."""
+
+    parser = argparse.ArgumentParser(description="Train Delta BPE tokenizer.")
+    parser.add_argument("--corpus_files", nargs="+", required=True, help="Input .txt corpus files.")
+    parser.add_argument("--output_path", required=True, help="Output tokenizer JSON path.")
+    return parser.parse_args()
+
+
+def main() -> None:
+    """Train and save a tokenizer."""
+
+    args = parse_args()
+    train_tokenizer([Path(path) for path in args.corpus_files], Path(args.output_path))
+    logger.info("Tokenizer training complete.")
+
+
+if __name__ == "__main__":
+    main()