inference.py

#!/usr/bin/env python3
"""Standalone pretrain inference for kerzgrr/Tercet-base.

Downloads model assets from the Hub (cached after first run), loads TinyGDN,
and streams raw text continuations. This is the pretrained base model — not
instruction-tuned. The chat / SFT checkpoint is available at kerzgrr/Tercet.

Examples:
  python inference.py --prompt "The history of computing begins"
  python inference.py
  python inference.py --prompt "Once upon a time" --temperature 0.9 --top-p 0.95
"""

from __future__ import annotations

import argparse
import json
import os
import platform
import shutil
import subprocess
import sys
import time
import warnings
from pathlib import Path

import torch
from safetensors.torch import load_file
from tokenizers import Tokenizer


def _silence_runtime_warnings() -> None:
    """Hide known-noisy Triton / SDPA warnings that don't affect results."""
    patterns = (
        r"tl\.make_block_ptr is deprecated",
        r"Memory efficient kernel not used because",
        r"Memory Efficient attention has been runtime disabled",
        r"Flash attention kernel not used because",
        r"Torch was not compiled with flash attention",
        r"cuDNN attention kernel not used because",
        r"cuDNN attention has been runtime disabled",
    )
    for pattern in patterns:
        warnings.filterwarnings("ignore", message=pattern)

REPO_ID = "kerzgrr/Tercet-base"
FLA_COMMIT = "cbb0a72efb55c18ca0ef4f298298317573ad2cb3"
FLA_REPO = "https://github.com/fla-org/flash-linear-attention.git"
PATCH_FILES = (
    "fla/__init__.py",
    "fla/ops/__init__.py",
    "fla/layers/__init__.py",
    "fla/ops/simple_gla/__init__.py",
)


def _download(filename: str, local_dir: Path | None) -> Path:
    from huggingface_hub import hf_hub_download

    return Path(
        hf_hub_download(
            repo_id=REPO_ID,
            filename=filename,
            local_dir=str(local_dir) if local_dir else None,
        )
    )


def _run(cmd: list[str], *, cwd: Path | None = None, env: dict | None = None) -> None:
    print("+", " ".join(cmd), flush=True)
    merged = os.environ.copy()
    if env:
        merged.update(env)
    # Avoid Windows cp1252 crashes while reading FLA's README during setup.
    merged.setdefault("PYTHONUTF8", "1")
    merged.setdefault("PYTHONIOENCODING", "utf-8")
    subprocess.check_call(cmd, cwd=str(cwd) if cwd else None, env=merged)


def _pip_install(*args: str) -> None:
    _run([sys.executable, "-m", "pip", "install", *args])


def _fla_importable() -> tuple[bool, str]:
    try:
        from fla.layers.gdn2 import GatedDeltaNet2  # noqa: F401
    except Exception as error:  # noqa: BLE001
        return False, str(error)
    return True, ""


def _cache_root() -> Path:
    override = os.environ.get("MONOSTICH_CACHE")
    if override:
        path = Path(override).expanduser().resolve()
    else:
        path = Path.home() / ".cache" / "monostich-2-base"
    path.mkdir(parents=True, exist_ok=True)
    return path


def _ensure_git() -> None:
    if shutil.which("git") is None:
        raise RuntimeError(
            "git is required to auto-install flash-linear-attention. "
            "Install Git and ensure it is on PATH."
        )


def _apply_windows_fla_patches(fla_root: Path, local_dir: Path | None) -> None:
    print("Applying Windows FLA import patches from the Hub …", flush=True)
    for relative in PATCH_FILES:
        source = _download(f"windows_fla_patches/{relative}", local_dir)
        target = fla_root / relative
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(source, target)
        print(f"  patched {relative}", flush=True)


def _install_fla(local_dir: Path | None) -> None:
    print("flash-linear-attention missing/broken — installing automatically …", flush=True)
    _pip_install("einops", "numpy")
    is_windows = platform.system() == "Windows"

    if not is_windows:
        _pip_install(
            "--no-deps",
            f"git+{FLA_REPO}@{FLA_COMMIT}",
        )
        return

    # Windows: editable clone + Hub patches (stock FLA + Triton 3.7 breaks on import).
    _ensure_git()
    fla_root = _cache_root() / "flash-linear-attention"
    if (fla_root / ".git").is_dir():
        _run(["git", "fetch", "--depth", "1", "origin", FLA_COMMIT], cwd=fla_root)
        _run(["git", "checkout", "--force", FLA_COMMIT], cwd=fla_root)
    else:
        if fla_root.exists():
            shutil.rmtree(fla_root)
        _run(
            [
                "git",
                "clone",
                "--filter=blob:none",
                FLA_REPO,
                str(fla_root),
            ]
        )
        _run(["git", "fetch", "--depth", "1", "origin", FLA_COMMIT], cwd=fla_root)
        _run(["git", "checkout", "--force", FLA_COMMIT], cwd=fla_root)

    _apply_windows_fla_patches(fla_root, local_dir)
    _pip_install("--no-build-isolation", "--no-deps", "-e", str(fla_root))


def _ensure_fla(local_dir: Path | None) -> None:
    ok, error = _fla_importable()
    if ok:
        return
    print(f"FLA not ready ({error})", flush=True)
    try:
        _install_fla(local_dir)
    except Exception as install_error:  # noqa: BLE001
        raise RuntimeError(
            "Automatic flash-linear-attention install failed.\n"
            f"Original import error: {error}\n"
            f"Install error: {install_error}"
        ) from install_error

    # Drop cached failed imports so the newly installed package is picked up.
    for name in list(sys.modules):
        if name == "fla" or name.startswith("fla."):
            del sys.modules[name]

    ok, error = _fla_importable()
    if not ok:
        raise RuntimeError(
            "flash-linear-attention installed but still failed to import "
            f"GatedDeltaNet2: {error}"
        )
    print("flash-linear-attention ready.", flush=True)


def _ensure_tiny_gdn(local_dir: Path | None) -> Path:
    """Return a directory that contains the tiny_gdn package on sys.path."""
    # Prefer files next to this script (local clone / Hub snapshot).
    here = Path(__file__).resolve().parent
    if (here / "tiny_gdn" / "__init__.py").is_file():
        return here
    if local_dir and (local_dir / "tiny_gdn" / "__init__.py").is_file():
        return local_dir

    # Pull package modules from the Hub into the cache.
    for name in ("tiny_gdn/__init__.py", "tiny_gdn/config.py", "tiny_gdn/model.py"):
        _download(name, local_dir)
    # hf_hub_download with local_dir=None caches under hub/; resolve via download of init.
    init_path = _download("tiny_gdn/__init__.py", local_dir)
    return init_path.parent.parent


def _sample(
    logits: torch.Tensor,
    *,
    temperature: float,
    top_p: float,
    top_k: int,
    generator: torch.Generator,
) -> int:
    logits = logits.float()
    if temperature <= 1e-5:
        return int(torch.argmax(logits).item())
    logits = logits / temperature
    if 0 < top_k < logits.shape[-1]:
        threshold = torch.topk(logits, top_k).values[-1]
        logits = logits.masked_fill(logits < threshold, -torch.inf)
    if top_p < 1.0:
        sorted_logits, sorted_indices = torch.sort(logits, descending=True)
        probs = torch.softmax(sorted_logits, dim=-1)
        remove = torch.cumsum(probs, dim=-1) > top_p
        remove[1:] = remove[:-1].clone()
        remove[0] = False
        sorted_logits = sorted_logits.masked_fill(remove, -torch.inf)
        logits = torch.full_like(logits, -torch.inf)
        logits.scatter_(0, sorted_indices, sorted_logits)
    probs = torch.softmax(logits, dim=-1)
    return int(torch.multinomial(probs, 1, generator=generator).item())


def _apply_repetition_penalty(
    logits: torch.Tensor,
    token_ids: list[int],
    penalty: float,
    window: int,
) -> torch.Tensor:
    if penalty == 1.0 or not token_ids:
        return logits
    recent = token_ids[-window:] if window > 0 else token_ids
    unique = torch.tensor(list(set(recent)), dtype=torch.long, device=logits.device)
    score = logits[unique]
    logits[unique] = torch.where(score > 0, score / penalty, score * penalty)
    return logits


@torch.inference_mode()
def generate(
    model,
    tokenizer: Tokenizer,
    prompt_ids: list[int],
    *,
    max_new_tokens: int,
    context_length: int,
    temperature: float,
    top_p: float,
    top_k: int,
    repetition_penalty: float,
    repetition_window: int,
    seed: int,
    stream: bool,
    device: torch.device,
) -> tuple[str, int, str]:
    eos_id = int(model.config.eos_token_id)
    token_ids = list(prompt_ids[-context_length:])
    generated: list[int] = []
    decoded = ""
    stop_reason = "max_new_tokens"
    generator = torch.Generator(device=device)
    generator.manual_seed(seed)
    started = time.perf_counter()

    for _ in range(max_new_tokens):
        context = token_ids[-context_length:]
        input_ids = torch.tensor([context], dtype=torch.long, device=device)
        output = model(input_ids, return_logits=True, logits_to_keep=1)
        if output.logits is None:
            raise RuntimeError("Model returned no logits")
        next_logits = _apply_repetition_penalty(
            output.logits[0, -1],
            token_ids,
            repetition_penalty,
            repetition_window,
        )
        next_id = _sample(
            next_logits,
            temperature=temperature,
            top_p=top_p,
            top_k=top_k,
            generator=generator,
        )
        if next_id == eos_id:
            stop_reason = "eos"
            break
        token_ids.append(next_id)
        generated.append(next_id)
        current = tokenizer.decode(generated, skip_special_tokens=True)
        delta = (
            current[len(decoded) :]
            if current.startswith(decoded)
            else current
        )
        decoded = current
        if stream and delta:
            print(delta, end="", flush=True)

    if stream:
        print(flush=True)
    elapsed = time.perf_counter() - started
    tps = len(generated) / max(elapsed, 1e-9)
    if stream:
        print(
            f"[done] tokens={len(generated)} stop={stop_reason} "
            f"{tps:.1f} tok/s",
            file=sys.stderr,
            flush=True,
        )
    return decoded, len(generated), stop_reason


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Tercet-base pretrained text continuation"
    )
    parser.add_argument(
        "--prompt",
        default=None,
        help="Single prompt to continue (omit for interactive REPL)",
    )
    parser.add_argument("--max-new-tokens", type=int, default=256)
    parser.add_argument("--temperature", type=float, default=0.8)
    parser.add_argument("--top-p", type=float, default=0.95)
    parser.add_argument("--top-k", type=int, default=50)
    parser.add_argument("--repetition-penalty", type=float, default=1.08)
    parser.add_argument("--repetition-window", type=int, default=256)
    parser.add_argument("--context-length", type=int, default=2048)
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument(
        "--device",
        default="cuda" if torch.cuda.is_available() else "cpu",
        choices=["cuda", "cpu"],
    )
    parser.add_argument(
        "--no-stream",
        action="store_true",
        help="Disable token streaming (print full completion at once)",
    )
    parser.add_argument(
        "--no-bos",
        action="store_true",
        help="Do not prepend <|begin_of_text|> to the prompt",
    )
    parser.add_argument(
        "--local-dir",
        default=None,
        help="Optional local snapshot directory (skips re-download when populated)",
    )
    parser.add_argument(
        "--repo-id",
        default=REPO_ID,
        help=f"Hub repo id (default: {REPO_ID})",
    )
    return parser.parse_args()


def main() -> int:
    _silence_runtime_warnings()
    args = parse_args()
    global REPO_ID
    REPO_ID = args.repo_id
    local_dir = Path(args.local_dir).resolve() if args.local_dir else None

    print(f"Loading Tercet-base from huggingface.co/{REPO_ID} …", flush=True)
    try:
        package_root = _ensure_tiny_gdn(local_dir)
    except Exception as error:  # noqa: BLE001
        print(f"Failed to resolve tiny_gdn package: {error}", file=sys.stderr)
        print(
            "Make sure huggingface_hub is installed and you can reach the Hub.",
            file=sys.stderr,
        )
        return 1

    if str(package_root) not in sys.path:
        sys.path.insert(0, str(package_root))

    try:
        _ensure_fla(local_dir)
    except Exception as error:  # noqa: BLE001
        print(str(error), file=sys.stderr)
        return 1

    try:
        from tiny_gdn import TinyGDNConfig, TinyGDNForCausalLM
    except ImportError as error:
        print(f"Could not import tiny_gdn: {error}", file=sys.stderr)
        return 1

    weights_path = _download("model.safetensors", local_dir)
    tok_path = _download("tokenizer.json", local_dir)
    cfg_path = _download("config.json", local_dir)

    raw = json.loads(cfg_path.read_text(encoding="utf-8"))
    # Config.from_json rejects unknown HF-only keys — filter to dataclass fields.
    from dataclasses import fields

    allowed = {item.name for item in fields(TinyGDNConfig)}
    payload = {key: value for key, value in raw.items() if key in allowed}
    if "shared_layer_indices" in payload:
        payload["shared_layer_indices"] = tuple(payload["shared_layer_indices"])
    config = TinyGDNConfig(**payload)

    device = torch.device(args.device)
    if device.type == "cuda" and not torch.cuda.is_available():
        print("CUDA requested but unavailable; falling back to CPU.", flush=True)
        device = torch.device("cpu")
    dtype = torch.bfloat16 if device.type == "cuda" else torch.float32

    print(
        f"Building TinyGDN ({config.num_hidden_layers}L / {config.hidden_size}d) "
        f"on {device} …",
        flush=True,
    )
    model = TinyGDNForCausalLM(config)
    state = load_file(str(weights_path), device="cpu")
    model.load_state_dict(state, strict=True)
    del state
    model = model.to(device=device, dtype=dtype)
    model.eval()
    model.requires_grad_(False)

    tokenizer = Tokenizer.from_file(str(tok_path))
    bos_id = config.bos_token_id
    context_length = min(args.context_length, config.max_position_embeddings)
    stream = not args.no_stream

    def encode_prompt(text: str) -> list[int]:
        ids = tokenizer.encode(text, add_special_tokens=False).ids
        if not args.no_bos and (not ids or ids[0] != bos_id):
            ids = [bos_id] + ids
        return ids

    def run_once(prompt: str) -> None:
        prompt_ids = encode_prompt(prompt)
        if stream:
            print(prompt, end="", flush=True)
        text, _, _ = generate(
            model,
            tokenizer,
            prompt_ids,
            max_new_tokens=args.max_new_tokens,
            context_length=context_length,
            temperature=args.temperature,
            top_p=args.top_p,
            top_k=args.top_k,
            repetition_penalty=args.repetition_penalty,
            repetition_window=args.repetition_window,
            seed=args.seed,
            stream=stream,
            device=device,
        )
        if not stream:
            print(prompt + text)

    if args.prompt is not None:
        run_once(args.prompt)
        return 0

    print(
        "Interactive pretrain continuation. Type a prompt and press Enter.\n"
        "Commands: /exit  /quit  /reset (clears nothing persistent; just a noop marker)\n"
        "Note: this is the BASE model — raw continuation, not chat.",
        flush=True,
    )
    while True:
        try:
            user_input = input("prompt> ")
        except (EOFError, KeyboardInterrupt):
            print()
            break
        text = user_input.strip()
        if not text:
            continue
        if text.lower() in {"/exit", "/quit"}:
            break
        if text.lower() == "/reset":
            print("(history not kept in pretrain mode)", flush=True)
            continue
        if stream:
            print("completion> ", end="", flush=True)
        run_once(text)
        print(flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())