inference.py · KerzgrAI/Senryu-2-1B at main

Upload Senryu-2-1B (EMA weights)

0a4bb1e verified about 10 hours ago

21.2 kB

#!/usr/bin/env python3
"""Standalone chat inference for KerzgrAI/Senryu-2-1B.
Streams a live reasoning trace (`<think>` / `<|think|>`) and runs Hermes JSON
tool-call rounds. Built-in `web-search`, `python`, and `calculator` tools
execute automatically. Other observations are sent as `<|tool_response|>`.
Examples:
  python inference.py --prompt "Explain recursion in two sentences."
  python inference.py --tools web-search,python,calculator
  python inference.py --no-think --prompt "Reply in one sentence."
  python inference.py --ngram-device cpu   # keep the 4.3 GB n-gram tables in host RAM
"""

from __future__ import annotations

import argparse
import importlib
import json
import os
import platform
import shutil
import subprocess
import sys
import time
import warnings
from pathlib import Path
from types import SimpleNamespace
from typing import Any

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


def _silence_runtime_warnings() -> None:
    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 = "KerzgrAI/Senryu-2-1B"
MODEL_NAME = "Senryu-2-1B"
FLA_COMMIT = "cbb0a72efb55c18ca0ef4f298298317573ad2cb3"
FLA_REPO = "https://github.com/fla-org/flash-linear-attention.git"
FLA_RUNTIME_REQUIREMENT = "transformers>=4.45.0"
# Parallax attention is not in a flash-linear-attention release yet, so fla
# installs from git at the commit the model was trained with.
FLA_MODULES = (
    "fla.layers.gdn2",
    "fla.models.utils",
    "fla.modules.rotary",
    "fla.ops.parallax.parallel",
    "fla.ops.parallax.decode",
)
PATCH_FILES = (
    "fla/__init__.py",
    "fla/ops/__init__.py",
    "fla/layers/__init__.py",
    "fla/ops/simple_gla/__init__.py",
)
TINY_GDN_FILES = (
    "tiny_gdn/__init__.py",
    "tiny_gdn/config.py",
    "tiny_gdn/model.py",
    "tiny_gdn/haiku_layers.py",
    "tiny_gdn/nn_common.py",
    "tiny_gdn/fused_ops.py",
    "tiny_gdn/senryu2.py",
    "tiny_gdn/parallax_ops.py",
    "tiny_gdn/gated_residual_ops.py",
    "tiny_gdn/ngram.py",
    "tiny_gdn/tools.py",
    "tiny_gdn/smoltalk_chat.py",
    "tiny_gdn/chatml.py",
    "tiny_gdn/cli_chat.py",
    "tiny_gdn/detokenize.py",
    "tiny_gdn/code_exec.py",
    "tiny_gdn/web_search.py",
)
NGRAM_TABLES_FILE = "ngram_tables.safetensors"


def _download(filename: str, local_dir: Path | None) -> Path:
    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)
    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:
        for name in FLA_MODULES:
            importlib.import_module(name)
    except Exception as error:  # noqa: BLE001
        return False, str(error)
    return True, ""


def _cache_root() -> Path:
    override = os.environ.get("SENRYU_CACHE")
    if override:
        path = Path(override).expanduser().resolve()
    else:
        path = Path.home() / ".cache" / "senryu"
    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", FLA_RUNTIME_REQUIREMENT)
    if platform.system() != "Windows":
        _pip_install("--no-deps", "--force-reinstall", f"git+{FLA_REPO}@{FLA_COMMIT}")
        return

    _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

    for name in list(sys.modules):
        if name == "fla" or name.startswith("fla."):
            del sys.modules[name]
    importlib.invalidate_caches()

    ok, error = _fla_importable()
    if not ok:
        raise RuntimeError(
            "flash-linear-attention installed but still failed to import "
            f"Gated DeltaNet-2, Parallax attention and the decode cache: {error}"
        )
    print("flash-linear-attention ready.", flush=True)


def _has_tiny_gdn(root: Path) -> bool:
    return all((root / name).is_file() for name in TINY_GDN_FILES)


def _ensure_tiny_gdn(local_dir: Path | None) -> Path:
    here = Path(__file__).resolve().parent
    if _has_tiny_gdn(here):
        return here
    if local_dir and _has_tiny_gdn(local_dir):
        return local_dir
    for name in TINY_GDN_FILES:
        _download(name, local_dir)
    return _download("tiny_gdn/__init__.py", local_dir).parent.parent


def _import_chat_stack() -> SimpleNamespace:
    # tiny_gdn is fetched from the Hub and FLA may be installed at runtime, so
    # these modules only become importable after _ensure_tiny_gdn/_ensure_fla.
    config = importlib.import_module("tiny_gdn.config")
    senryu2 = importlib.import_module("tiny_gdn.senryu2")
    chatml = importlib.import_module("tiny_gdn.chatml")
    cli_chat = importlib.import_module("tiny_gdn.cli_chat")
    detokenize = importlib.import_module("tiny_gdn.detokenize")
    tools = importlib.import_module("tiny_gdn.tools")
    return SimpleNamespace(
        TinyGDNConfig=config.TinyGDNConfig,
        Senryu2ForCausalLM=senryu2.Senryu2ForCausalLM,
        encode_chatml_messages=chatml.encode_chatml_messages,
        ChatLoopState=cli_chat.ChatLoopState,
        LiveReasoningStreamer=cli_chat.LiveReasoningStreamer,
        append_tool_round=cli_chat.append_tool_round,
        apply_slash_command=cli_chat.apply_slash_command,
        prompt_tool_results=cli_chat.prompt_tool_results,
        resolve_cli_tools=cli_chat.resolve_cli_tools,
        IncrementalUtf8Decoder=detokenize.IncrementalUtf8Decoder,
        parse_tool_calls=tools.parse_tool_calls,
    )


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).cpu()
    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_turn(
    model,
    tokenizer: Tokenizer,
    stack: SimpleNamespace,
    streamer,
    messages: list[dict[str, Any]],
    *,
    tools: Any | None,
    enable_thinking: bool,
    max_new_tokens: int,
    context_length: int,
    temperature: float,
    top_p: float,
    top_k: int,
    repetition_penalty: float,
    repetition_window: int,
    seed: int,
    device: torch.device,
) -> tuple[str, tuple]:
    prompt_ids = stack.encode_chatml_messages(
        tokenizer,
        messages,
        tools=tools,
        enable_thinking=enable_thinking,
    )
    if len(prompt_ids) > context_length:
        print(
            f"[warn] conversation is {len(prompt_ids)} tokens; "
            f"keeping the last {context_length} (--context-length)",
            file=sys.stderr,
            flush=True,
        )
    eos_id = int(model.config.eos_token_id)
    im_end = tokenizer.token_to_id("<|im_end|>")
    stop = {eos_id}
    if im_end is not None:
        stop.add(im_end)

    token_ids = list(prompt_ids[-context_length:])
    generated: list[int] = []
    decoder = stack.IncrementalUtf8Decoder(tokenizer)
    stop_reason = "max_new_tokens"
    generator = torch.Generator(device="cpu")
    generator.manual_seed(seed)
    started = time.perf_counter()

    try:
        input_ids = torch.tensor([token_ids], dtype=torch.long, device=device)
        output = model(
            input_ids,
            return_logits=True,
            logits_to_keep=1,
            use_cache=True,
        )

        for step in range(max_new_tokens):
            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 in stop:
                stop_reason = "stop"
                break
            token_ids.append(next_id)
            generated.append(next_id)
            decoder.push(next_id)
            if streamer is not None:
                streamer.update(decoder.text)
            if step + 1 < max_new_tokens:
                if output.past_key_values is None:
                    raise RuntimeError("Model returned no decode cache")
                if output.past_key_values.length + 1 > model.config.max_position_embeddings:
                    stop_reason = "context_full"
                    break
                input_ids = torch.tensor([[next_id]], dtype=torch.long, device=device)
                output = model(
                    input_ids,
                    return_logits=True,
                    logits_to_keep=1,
                    past_key_values=output.past_key_values,
                    use_cache=True,
                )
    except KeyboardInterrupt:
        stop_reason = "interrupted"

    decoded = decoder.finalize()
    if streamer is not None:
        streamer.finish()
    elapsed = time.perf_counter() - started
    tps = len(generated) / max(elapsed, 1e-9)
    print(
        f"[done] tokens={len(generated)} stop={stop_reason} {tps:.1f} tok/s",
        file=sys.stderr,
        flush=True,
    )
    if stop_reason == "interrupted":
        return decoded, ()
    return decoded, tuple(stack.parse_tool_calls(decoded))


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=f"{MODEL_NAME} chat inference")
    parser.add_argument("--prompt", default=None, help="Single user prompt")
    parser.add_argument("--system", default="", help="Optional system prompt")
    parser.add_argument(
        "--think",
        dest="enable_thinking",
        action="store_true",
        default=True,
    )
    parser.add_argument(
        "--no-think",
        dest="enable_thinking",
        action="store_false",
    )
    parser.add_argument(
        "--tools",
        default="",
        help="Built-in tools: web-search, python, calculator (comma-separated)",
    )
    parser.add_argument("--tools-json", default=None)
    parser.add_argument("--max-tool-rounds", type=int, default=8)
    parser.add_argument("--max-new-tokens", type=int, default=4096)
    parser.add_argument("--temperature", type=float, default=0.6)
    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=32768)
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument(
        "--device",
        default="cuda",
        help=f"CUDA device (cuda, cuda:1, …); {MODEL_NAME} needs a GPU",
    )
    parser.add_argument(
        "--ngram-device",
        default=None,
        help="Where the n-gram tables live: the GPU (default, --device) or cpu (host RAM, slower decode)",
    )
    parser.add_argument("--no-stream", action="store_true")
    parser.add_argument("--no-color", action="store_true")
    parser.add_argument("--local-dir", default=None)
    parser.add_argument("--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

    try:
        device = torch.device(args.device)
        ngram_device = torch.device(args.ngram_device) if args.ngram_device else device
    except RuntimeError as error:
        print(f"Invalid device: {error}", file=sys.stderr)
        return 1
    if device.type != "cuda" or not torch.cuda.is_available():
        print(
            f"{MODEL_NAME} needs a CUDA GPU: its Gated DeltaNet-2 and Parallax "
            "layers run on flash-linear-attention Triton kernels. "
            f"torch {torch.__version__} reports CUDA available={torch.cuda.is_available()}.",
            file=sys.stderr,
        )
        return 1
    if ngram_device.type not in {"cuda", "cpu"}:
        print("--ngram-device must be a CUDA device or cpu", file=sys.stderr)
        return 1

    print(f"Loading {MODEL_NAME} 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)
        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:
        stack = _import_chat_stack()
    except ImportError as error:
        print(f"Could not import tiny_gdn chat stack: {error}", file=sys.stderr)
        return 1

    try:
        raw_json = (
            Path(args.tools_json).read_text(encoding="utf-8")
            if args.tools_json
            else None
        )
        tools = stack.resolve_cli_tools(args.tools, raw_json)
    except (OSError, ValueError, json.JSONDecodeError) as error:
        print(error, file=sys.stderr)
        return 1

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

    config = stack.TinyGDNConfig.from_json(cfg_path)
    print(
        f"Building {MODEL_NAME} ({config.num_hidden_layers}L / {config.hidden_size}d) "
        f"on {device}, n-gram tables on {ngram_device} …",
        flush=True,
    )
    with torch.device(device):
        model = stack.Senryu2ForCausalLM(config)
    model = model.to(dtype=torch.bfloat16)
    model.load_state_dict(load_file(str(weights_path), device=str(device)), strict=True)
    if model.ngram is not None:
        model.ngram.load_tables(tables_path, device=ngram_device, dtype=torch.bfloat16)
    model.eval()
    model.requires_grad_(False)

    tokenizer = Tokenizer.from_file(str(tok_path))
    context_length = min(args.context_length, config.max_position_embeddings)
    color = not args.no_color and sys.stdout.isatty()

    def read_line(prompt: str) -> str:
        return input(prompt)

    def run_turn(state) -> None:
        for round_index in range(max(1, args.max_tool_rounds + 1)):
            streamer = (
                None
                if args.no_stream
                else stack.LiveReasoningStreamer(sys.stdout, color=color)
            )
            if streamer is not None:
                print("assistant> ", end="", flush=True)
            text, calls = generate_turn(
                model,
                tokenizer,
                stack,
                streamer,
                state.messages,
                tools=state.tools,
                enable_thinking=state.enable_thinking,
                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,
                device=device,
            )
            if args.no_stream:
                print(f"assistant> {text}")
            if not calls or round_index >= args.max_tool_rounds:
                state.messages.append({"role": "assistant", "content": text})
                return
            results = stack.prompt_tool_results(calls, read_line=read_line)
            stack.append_tool_round(state.messages, text, results)

    state = stack.ChatLoopState(
        system=args.system,
        enable_thinking=args.enable_thinking,
        tools=tools,
    )
    state.reset()

    if args.prompt is not None:
        state.add_user(args.prompt)
        try:
            run_turn(state)
        except KeyboardInterrupt:
            print("\n(interrupted)", file=sys.stderr)
            return 130
        return 0

    print(
        f"Interactive {MODEL_NAME} chat. "
        "Commands: /think  /no_think  /system …  /reset  /exit\n"
        "web-search, python, and calculator run automatically when enabled. "
        "Other <tool_call> observations can be pasted at the prompt. "
        "Ctrl+C stops a reply.",
        flush=True,
    )
    while True:
        try:
            user_input = input("user> ")
        except (EOFError, KeyboardInterrupt):
            print()
            return 0
        text = user_input.strip()
        if not text:
            continue
        status = stack.apply_slash_command(state, text)
        if status == "exit":
            return 0
        if status is not None:
            print(f"({status})", flush=True)
            continue
        state.add_user(text)
        try:
            run_turn(state)
        except KeyboardInterrupt:
            print("\n(interrupted)", flush=True)
        print(flush=True)
    return 0


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

	#!/usr/bin/env python3
	"""Standalone chat inference for KerzgrAI/Senryu-2-1B.

	Streams a live reasoning trace (`<think>` / `<\|think\|>`) and runs Hermes JSON
	tool-call rounds. Built-in `web-search`, `python`, and `calculator` tools
	execute automatically. Other observations are sent as `<\|tool_response\|>`.

	Examples:
	python inference.py --prompt "Explain recursion in two sentences."
	python inference.py --tools web-search,python,calculator
	python inference.py --no-think --prompt "Reply in one sentence."
	python inference.py --ngram-device cpu # keep the 4.3 GB n-gram tables in host RAM
	"""

	from __future__ import annotations

	import argparse
	import importlib
	import json
	import os
	import platform
	import shutil
	import subprocess
	import sys
	import time
	import warnings
	from pathlib import Path
	from types import SimpleNamespace
	from typing import Any

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


	def _silence_runtime_warnings() -> None:
	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 = "KerzgrAI/Senryu-2-1B"
	MODEL_NAME = "Senryu-2-1B"
	FLA_COMMIT = "cbb0a72efb55c18ca0ef4f298298317573ad2cb3"
	FLA_REPO = "https://github.com/fla-org/flash-linear-attention.git"
	FLA_RUNTIME_REQUIREMENT = "transformers>=4.45.0"
	# Parallax attention is not in a flash-linear-attention release yet, so fla
	# installs from git at the commit the model was trained with.
	FLA_MODULES = (
	"fla.layers.gdn2",
	"fla.models.utils",
	"fla.modules.rotary",
	"fla.ops.parallax.parallel",
	"fla.ops.parallax.decode",
	)
	PATCH_FILES = (
	"fla/__init__.py",
	"fla/ops/__init__.py",
	"fla/layers/__init__.py",
	"fla/ops/simple_gla/__init__.py",
	)
	TINY_GDN_FILES = (
	"tiny_gdn/__init__.py",
	"tiny_gdn/config.py",
	"tiny_gdn/model.py",
	"tiny_gdn/haiku_layers.py",
	"tiny_gdn/nn_common.py",
	"tiny_gdn/fused_ops.py",
	"tiny_gdn/senryu2.py",
	"tiny_gdn/parallax_ops.py",
	"tiny_gdn/gated_residual_ops.py",
	"tiny_gdn/ngram.py",
	"tiny_gdn/tools.py",
	"tiny_gdn/smoltalk_chat.py",
	"tiny_gdn/chatml.py",
	"tiny_gdn/cli_chat.py",
	"tiny_gdn/detokenize.py",
	"tiny_gdn/code_exec.py",
	"tiny_gdn/web_search.py",
	)
	NGRAM_TABLES_FILE = "ngram_tables.safetensors"


	def _download(filename: str, local_dir: Path \| None) -> Path:
	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)
	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:
	for name in FLA_MODULES:
	importlib.import_module(name)
	except Exception as error: # noqa: BLE001
	return False, str(error)
	return True, ""


	def _cache_root() -> Path:
	override = os.environ.get("SENRYU_CACHE")
	if override:
	path = Path(override).expanduser().resolve()
	else:
	path = Path.home() / ".cache" / "senryu"
	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", FLA_RUNTIME_REQUIREMENT)
	if platform.system() != "Windows":
	_pip_install("--no-deps", "--force-reinstall", f"git+{FLA_REPO}@{FLA_COMMIT}")
	return

	_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

	for name in list(sys.modules):
	if name == "fla" or name.startswith("fla."):
	del sys.modules[name]
	importlib.invalidate_caches()

	ok, error = _fla_importable()
	if not ok:
	raise RuntimeError(
	"flash-linear-attention installed but still failed to import "
	f"Gated DeltaNet-2, Parallax attention and the decode cache: {error}"
	)
	print("flash-linear-attention ready.", flush=True)


	def _has_tiny_gdn(root: Path) -> bool:
	return all((root / name).is_file() for name in TINY_GDN_FILES)


	def _ensure_tiny_gdn(local_dir: Path \| None) -> Path:
	here = Path(__file__).resolve().parent
	if _has_tiny_gdn(here):
	return here
	if local_dir and _has_tiny_gdn(local_dir):
	return local_dir
	for name in TINY_GDN_FILES:
	_download(name, local_dir)
	return _download("tiny_gdn/__init__.py", local_dir).parent.parent


	def _import_chat_stack() -> SimpleNamespace:
	# tiny_gdn is fetched from the Hub and FLA may be installed at runtime, so
	# these modules only become importable after _ensure_tiny_gdn/_ensure_fla.
	config = importlib.import_module("tiny_gdn.config")
	senryu2 = importlib.import_module("tiny_gdn.senryu2")
	chatml = importlib.import_module("tiny_gdn.chatml")
	cli_chat = importlib.import_module("tiny_gdn.cli_chat")
	detokenize = importlib.import_module("tiny_gdn.detokenize")
	tools = importlib.import_module("tiny_gdn.tools")
	return SimpleNamespace(
	TinyGDNConfig=config.TinyGDNConfig,
	Senryu2ForCausalLM=senryu2.Senryu2ForCausalLM,
	encode_chatml_messages=chatml.encode_chatml_messages,
	ChatLoopState=cli_chat.ChatLoopState,
	LiveReasoningStreamer=cli_chat.LiveReasoningStreamer,
	append_tool_round=cli_chat.append_tool_round,
	apply_slash_command=cli_chat.apply_slash_command,
	prompt_tool_results=cli_chat.prompt_tool_results,
	resolve_cli_tools=cli_chat.resolve_cli_tools,
	IncrementalUtf8Decoder=detokenize.IncrementalUtf8Decoder,
	parse_tool_calls=tools.parse_tool_calls,
	)


	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).cpu()
	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_turn(
	model,
	tokenizer: Tokenizer,
	stack: SimpleNamespace,
	streamer,
	messages: list[dict[str, Any]],
	*,
	tools: Any \| None,
	enable_thinking: bool,
	max_new_tokens: int,
	context_length: int,
	temperature: float,
	top_p: float,
	top_k: int,
	repetition_penalty: float,
	repetition_window: int,
	seed: int,
	device: torch.device,
	) -> tuple[str, tuple]:
	prompt_ids = stack.encode_chatml_messages(
	tokenizer,
	messages,
	tools=tools,
	enable_thinking=enable_thinking,
	)
	if len(prompt_ids) > context_length:
	print(
	f"[warn] conversation is {len(prompt_ids)} tokens; "
	f"keeping the last {context_length} (--context-length)",
	file=sys.stderr,
	flush=True,
	)
	eos_id = int(model.config.eos_token_id)
	im_end = tokenizer.token_to_id("<\|im_end\|>")
	stop = {eos_id}
	if im_end is not None:
	stop.add(im_end)

	token_ids = list(prompt_ids[-context_length:])
	generated: list[int] = []
	decoder = stack.IncrementalUtf8Decoder(tokenizer)
	stop_reason = "max_new_tokens"
	generator = torch.Generator(device="cpu")
	generator.manual_seed(seed)
	started = time.perf_counter()

	try:
	input_ids = torch.tensor([token_ids], dtype=torch.long, device=device)
	output = model(
	input_ids,
	return_logits=True,
	logits_to_keep=1,
	use_cache=True,
	)

	for step in range(max_new_tokens):
	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 in stop:
	stop_reason = "stop"
	break
	token_ids.append(next_id)
	generated.append(next_id)
	decoder.push(next_id)
	if streamer is not None:
	streamer.update(decoder.text)
	if step + 1 < max_new_tokens:
	if output.past_key_values is None:
	raise RuntimeError("Model returned no decode cache")
	if output.past_key_values.length + 1 > model.config.max_position_embeddings:
	stop_reason = "context_full"
	break
	input_ids = torch.tensor([[next_id]], dtype=torch.long, device=device)
	output = model(
	input_ids,
	return_logits=True,
	logits_to_keep=1,
	past_key_values=output.past_key_values,
	use_cache=True,
	)
	except KeyboardInterrupt:
	stop_reason = "interrupted"

	decoded = decoder.finalize()
	if streamer is not None:
	streamer.finish()
	elapsed = time.perf_counter() - started
	tps = len(generated) / max(elapsed, 1e-9)
	print(
	f"[done] tokens={len(generated)} stop={stop_reason} {tps:.1f} tok/s",
	file=sys.stderr,
	flush=True,
	)
	if stop_reason == "interrupted":
	return decoded, ()
	return decoded, tuple(stack.parse_tool_calls(decoded))


	def parse_args() -> argparse.Namespace:
	parser = argparse.ArgumentParser(description=f"{MODEL_NAME} chat inference")
	parser.add_argument("--prompt", default=None, help="Single user prompt")
	parser.add_argument("--system", default="", help="Optional system prompt")
	parser.add_argument(
	"--think",
	dest="enable_thinking",
	action="store_true",
	default=True,
	)
	parser.add_argument(
	"--no-think",
	dest="enable_thinking",
	action="store_false",
	)
	parser.add_argument(
	"--tools",
	default="",
	help="Built-in tools: web-search, python, calculator (comma-separated)",
	)
	parser.add_argument("--tools-json", default=None)
	parser.add_argument("--max-tool-rounds", type=int, default=8)
	parser.add_argument("--max-new-tokens", type=int, default=4096)
	parser.add_argument("--temperature", type=float, default=0.6)
	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=32768)
	parser.add_argument("--seed", type=int, default=42)
	parser.add_argument(
	"--device",
	default="cuda",
	help=f"CUDA device (cuda, cuda:1, …); {MODEL_NAME} needs a GPU",
	)
	parser.add_argument(
	"--ngram-device",
	default=None,
	help="Where the n-gram tables live: the GPU (default, --device) or cpu (host RAM, slower decode)",
	)
	parser.add_argument("--no-stream", action="store_true")
	parser.add_argument("--no-color", action="store_true")
	parser.add_argument("--local-dir", default=None)
	parser.add_argument("--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

	try:
	device = torch.device(args.device)
	ngram_device = torch.device(args.ngram_device) if args.ngram_device else device
	except RuntimeError as error:
	print(f"Invalid device: {error}", file=sys.stderr)
	return 1
	if device.type != "cuda" or not torch.cuda.is_available():
	print(
	f"{MODEL_NAME} needs a CUDA GPU: its Gated DeltaNet-2 and Parallax "
	"layers run on flash-linear-attention Triton kernels. "
	f"torch {torch.__version__} reports CUDA available={torch.cuda.is_available()}.",
	file=sys.stderr,
	)
	return 1
	if ngram_device.type not in {"cuda", "cpu"}:
	print("--ngram-device must be a CUDA device or cpu", file=sys.stderr)
	return 1

	print(f"Loading {MODEL_NAME} 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)
	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:
	stack = _import_chat_stack()
	except ImportError as error:
	print(f"Could not import tiny_gdn chat stack: {error}", file=sys.stderr)
	return 1

	try:
	raw_json = (
	Path(args.tools_json).read_text(encoding="utf-8")
	if args.tools_json
	else None
	)
	tools = stack.resolve_cli_tools(args.tools, raw_json)
	except (OSError, ValueError, json.JSONDecodeError) as error:
	print(error, file=sys.stderr)
	return 1

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

	config = stack.TinyGDNConfig.from_json(cfg_path)
	print(
	f"Building {MODEL_NAME} ({config.num_hidden_layers}L / {config.hidden_size}d) "
	f"on {device}, n-gram tables on {ngram_device} …",
	flush=True,
	)
	with torch.device(device):
	model = stack.Senryu2ForCausalLM(config)
	model = model.to(dtype=torch.bfloat16)
	model.load_state_dict(load_file(str(weights_path), device=str(device)), strict=True)
	if model.ngram is not None:
	model.ngram.load_tables(tables_path, device=ngram_device, dtype=torch.bfloat16)
	model.eval()
	model.requires_grad_(False)

	tokenizer = Tokenizer.from_file(str(tok_path))
	context_length = min(args.context_length, config.max_position_embeddings)
	color = not args.no_color and sys.stdout.isatty()

	def read_line(prompt: str) -> str:
	return input(prompt)

	def run_turn(state) -> None:
	for round_index in range(max(1, args.max_tool_rounds + 1)):
	streamer = (
	None
	if args.no_stream
	else stack.LiveReasoningStreamer(sys.stdout, color=color)
	)
	if streamer is not None:
	print("assistant> ", end="", flush=True)
	text, calls = generate_turn(
	model,
	tokenizer,
	stack,
	streamer,
	state.messages,
	tools=state.tools,
	enable_thinking=state.enable_thinking,
	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,
	device=device,
	)
	if args.no_stream:
	print(f"assistant> {text}")
	if not calls or round_index >= args.max_tool_rounds:
	state.messages.append({"role": "assistant", "content": text})
	return
	results = stack.prompt_tool_results(calls, read_line=read_line)
	stack.append_tool_round(state.messages, text, results)

	state = stack.ChatLoopState(
	system=args.system,
	enable_thinking=args.enable_thinking,
	tools=tools,
	)
	state.reset()

	if args.prompt is not None:
	state.add_user(args.prompt)
	try:
	run_turn(state)
	except KeyboardInterrupt:
	print("\n(interrupted)", file=sys.stderr)
	return 130
	return 0

	print(
	f"Interactive {MODEL_NAME} chat. "
	"Commands: /think /no_think /system … /reset /exit\n"
	"web-search, python, and calculator run automatically when enabled. "
	"Other <tool_call> observations can be pasted at the prompt. "
	"Ctrl+C stops a reply.",
	flush=True,
	)
	while True:
	try:
	user_input = input("user> ")
	except (EOFError, KeyboardInterrupt):
	print()
	return 0
	text = user_input.strip()
	if not text:
	continue
	status = stack.apply_slash_command(state, text)
	if status == "exit":
	return 0
	if status is not None:
	print(f"({status})", flush=True)
	continue
	state.add_user(text)
	try:
	run_turn(state)
	except KeyboardInterrupt:
	print("\n(interrupted)", flush=True)
	print(flush=True)
	return 0


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