Upload folder using huggingface_hub

README.md

@@ -0,0 +1,66 @@
+---
+library_name: transformers
+pipeline_tag: text-generation
+tags:
+- causal-lm
+- trust-remote-code
+- sentencepiece
+---
+
+# Pinyin-Code Causal LM
+
+This repository contains a custom Transformers causal language model.
+External evaluation repositories should load it with
+`trust_remote_code=True` and use the `causal` backend.
+
+## Dependencies
+
+Install the runtime dependencies before loading the model:
+
+```bash
+pip install torch transformers safetensors sentencepiece pypinyin jieba
+```
+
+`sentencepiece` is required for `AutoTokenizer`. `pypinyin` is required
+for raw Mandarin-to-pinyin tokenization. `jieba` is required when
+`use_jieba` is true; this export was created with `use_jieba=true`.
+
+## Loading
+
+```python
+from transformers import AutoConfig, AutoModel, AutoModelForCausalLM, AutoTokenizer
+
+model_path = "PATH_OR_REPO_ID"
+
+config = AutoConfig.from_pretrained(model_path, trust_remote_code=True)
+tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
+base_model = AutoModel.from_pretrained(model_path, trust_remote_code=True)
+model = AutoModelForCausalLM.from_pretrained(model_path, trust_remote_code=True)
+```
+
+## Evaluation
+
+Configure external evaluators with:
+
+- model path: this local folder or Hugging Face repo ID
+- backend: `causal`
+- trust remote code: enabled
+
+The tokenizer accepts raw text through standard calls such as
+`tokenizer(text)`, `tokenizer(text, add_special_tokens=False)`, and
+`tokenizer(texts, padding=True, truncation=True, return_tensors="pt")`.
+It also accepts `return_offsets_mapping=True` for compatibility with
+completion-ranking evaluators that need suffix masks. The model supports
+`output_hidden_states=True` for representation extraction tasks.
+
+This export sets `patch_pathlib_utf8_open=true` in `config.json`.
+When loaded with `trust_remote_code=True`, the config installs a narrow
+Windows compatibility shim so later text-mode `Path.open("r")` calls
+without an explicit encoding default to UTF-8. Set
+`PINYIN_CODE_DISABLE_UTF8_OPEN_PATCH=1` before loading the model to
+disable that shim.
+
+Export metadata:
+
+- transliteration: `pinyin-code`
+- use_jieba: `true`

config.json

@@ -0,0 +1,35 @@
+{
+  "architectures": [
+    "PinyinCodeForCausalLM"
+  ],
+  "auto_map": {
+    "AutoConfig": "configuration_pinyin_code.PinyinCodeConfig",
+    "AutoModel": "modeling_pinyin_code.PinyinCodeModel",
+    "AutoModelForCausalLM": "modeling_pinyin_code.PinyinCodeForCausalLM",
+    "AutoTokenizer": [
+      "tokenization_pinyin_code.EncodedMandarinTokenizer",
+      null
+    ]
+  },
+  "block_size": 512,
+  "bos_token_id": 2,
+  "dropout": 0.1,
+  "dtype": "float32",
+  "eos_token_id": 3,
+  "evaluation_backend": "causal",
+  "hidden_size": 512,
+  "is_decoder": true,
+  "max_position_embeddings": 512,
+  "model_type": "pinyin_code",
+  "n_embd": 512,
+  "n_head": 8,
+  "n_layer": 8,
+  "num_attention_heads": 8,
+  "num_hidden_layers": 8,
+  "pad_token_id": 0,
+  "patch_pathlib_utf8_open": true,
+  "transformers_version": "5.10.2",
+  "unk_token_id": 1,
+  "use_cache": false,
+  "vocab_size": 16000
+}

configuration_pinyin_code.py

@@ -0,0 +1,97 @@
+"""Configuration for the Transformers-compatible pinyin-code causal LM."""
+
+from __future__ import annotations
+
+import functools
+import os
+import pathlib
+
+from transformers import PretrainedConfig
+
+
+_UTF8_PATH_OPEN_PATCH_MARKER = "_pinyin_code_utf8_path_open_patch"
+
+
+def install_utf8_path_open_patch() -> None:
+    """Default text-mode ``Path.open`` calls to UTF-8 when encoding is omitted.
+
+    Some external Windows evaluation pipelines call ``Path.open("r")`` on
+    UTF-8 JSONL data before specifying an encoding. The model is loaded before
+    those datasets, so this narrow compatibility shim lets such pipelines read
+    Mandarin evaluation files without repository-side changes. Explicit
+    encodings and binary modes are left untouched.
+    """
+    current_open = pathlib.Path.open
+    if getattr(current_open, _UTF8_PATH_OPEN_PATCH_MARKER, False):
+        return
+
+    @functools.wraps(current_open)
+    def utf8_default_open(
+        self,
+        mode: str = "r",
+        buffering: int = -1,
+        encoding: str | None = None,
+        errors: str | None = None,
+        newline: str | None = None,
+    ):
+        if encoding is None and "b" not in mode:
+            encoding = "utf-8"
+        return current_open(
+            self,
+            mode=mode,
+            buffering=buffering,
+            encoding=encoding,
+            errors=errors,
+            newline=newline,
+        )
+
+    setattr(utf8_default_open, _UTF8_PATH_OPEN_PATCH_MARKER, True)
+    pathlib.Path.open = utf8_default_open
+
+
+class PinyinCodeConfig(PretrainedConfig):
+    """Configuration for the compact GPT-style pinyin-code decoder."""
+
+    model_type = "pinyin_code"
+
+    def __init__(
+        self,
+        vocab_size: int = 8000,
+        block_size: int = 128,
+        n_layer: int = 6,
+        n_head: int = 8,
+        n_embd: int = 256,
+        dropout: float = 0.1,
+        bos_token_id: int | None = None,
+        eos_token_id: int | None = None,
+        pad_token_id: int | None = None,
+        unk_token_id: int | None = None,
+        patch_pathlib_utf8_open: bool = False,
+        **kwargs,
+    ) -> None:
+        super().__init__(
+            bos_token_id=bos_token_id,
+            eos_token_id=eos_token_id,
+            pad_token_id=pad_token_id,
+            unk_token_id=unk_token_id,
+            **kwargs,
+        )
+        self.vocab_size = vocab_size
+        self.block_size = block_size
+        self.n_layer = n_layer
+        self.n_head = n_head
+        self.n_embd = n_embd
+        self.dropout = dropout
+        self.num_hidden_layers = n_layer
+        self.num_attention_heads = n_head
+        self.hidden_size = n_embd
+        self.max_position_embeddings = block_size
+        self.is_decoder = True
+        self.is_encoder_decoder = False
+        self.use_cache = False
+        self.patch_pathlib_utf8_open = patch_pathlib_utf8_open
+        if (
+            patch_pathlib_utf8_open
+            and os.environ.get("PINYIN_CODE_DISABLE_UTF8_OPEN_PATCH") != "1"
+        ):
+            install_utf8_path_open_patch()

generation_config.json

@@ -0,0 +1,6 @@
+{
+  "bos_token_id": 2,
+  "eos_token_id": 3,
+  "pad_token_id": 0,
+  "transformers_version": "5.10.2"
+}

hf/__init__.py

@@ -0,0 +1,13 @@
+"""Hugging Face Transformers integration for the pinyin-code language model."""
+
+from .configuration_pinyin_code import PinyinCodeConfig
+from .modeling_pinyin_code import PinyinCodeForCausalLM, PinyinCodeModel
+from .tokenization_pinyin_code import EncodedMandarinTokenizer, PinyinCodeTokenizer
+
+__all__ = [
+    "EncodedMandarinTokenizer",
+    "PinyinCodeConfig",
+    "PinyinCodeForCausalLM",
+    "PinyinCodeModel",
+    "PinyinCodeTokenizer",
+]

hf/configuration_pinyin_code.py

@@ -0,0 +1,97 @@
+"""Configuration for the Transformers-compatible pinyin-code causal LM."""
+
+from __future__ import annotations
+
+import functools
+import os
+import pathlib
+
+from transformers import PretrainedConfig
+
+
+_UTF8_PATH_OPEN_PATCH_MARKER = "_pinyin_code_utf8_path_open_patch"
+
+
+def install_utf8_path_open_patch() -> None:
+    """Default text-mode ``Path.open`` calls to UTF-8 when encoding is omitted.
+
+    Some external Windows evaluation pipelines call ``Path.open("r")`` on
+    UTF-8 JSONL data before specifying an encoding. The model is loaded before
+    those datasets, so this narrow compatibility shim lets such pipelines read
+    Mandarin evaluation files without repository-side changes. Explicit
+    encodings and binary modes are left untouched.
+    """
+    current_open = pathlib.Path.open
+    if getattr(current_open, _UTF8_PATH_OPEN_PATCH_MARKER, False):
+        return
+
+    @functools.wraps(current_open)
+    def utf8_default_open(
+        self,
+        mode: str = "r",
+        buffering: int = -1,
+        encoding: str | None = None,
+        errors: str | None = None,
+        newline: str | None = None,
+    ):
+        if encoding is None and "b" not in mode:
+            encoding = "utf-8"
+        return current_open(
+            self,
+            mode=mode,
+            buffering=buffering,
+            encoding=encoding,
+            errors=errors,
+            newline=newline,
+        )
+
+    setattr(utf8_default_open, _UTF8_PATH_OPEN_PATCH_MARKER, True)
+    pathlib.Path.open = utf8_default_open
+
+
+class PinyinCodeConfig(PretrainedConfig):
+    """Configuration for the compact GPT-style pinyin-code decoder."""
+
+    model_type = "pinyin_code"
+
+    def __init__(
+        self,
+        vocab_size: int = 8000,
+        block_size: int = 128,
+        n_layer: int = 6,
+        n_head: int = 8,
+        n_embd: int = 256,
+        dropout: float = 0.1,
+        bos_token_id: int | None = None,
+        eos_token_id: int | None = None,
+        pad_token_id: int | None = None,
+        unk_token_id: int | None = None,
+        patch_pathlib_utf8_open: bool = False,
+        **kwargs,
+    ) -> None:
+        super().__init__(
+            bos_token_id=bos_token_id,
+            eos_token_id=eos_token_id,
+            pad_token_id=pad_token_id,
+            unk_token_id=unk_token_id,
+            **kwargs,
+        )
+        self.vocab_size = vocab_size
+        self.block_size = block_size
+        self.n_layer = n_layer
+        self.n_head = n_head
+        self.n_embd = n_embd
+        self.dropout = dropout
+        self.num_hidden_layers = n_layer
+        self.num_attention_heads = n_head
+        self.hidden_size = n_embd
+        self.max_position_embeddings = block_size
+        self.is_decoder = True
+        self.is_encoder_decoder = False
+        self.use_cache = False
+        self.patch_pathlib_utf8_open = patch_pathlib_utf8_open
+        if (
+            patch_pathlib_utf8_open
+            and os.environ.get("PINYIN_CODE_DISABLE_UTF8_OPEN_PATCH") != "1"
+        ):
+            install_utf8_path_open_patch()

hf/modeling_pinyin_code.py

@@ -0,0 +1,287 @@
+"""Transformers-compatible implementation of the pinyin-code causal LM."""
+
+from __future__ import annotations
+
+import torch
+from torch import nn
+from torch.nn import functional as F
+from transformers import PreTrainedModel
+from transformers.generation import GenerationMixin
+from transformers.modeling_outputs import BaseModelOutput, CausalLMOutput
+
+from .configuration_pinyin_code import PinyinCodeConfig
+
+
+class CausalSelfAttention(nn.Module):
+    """Multi-head masked self-attention matching the original training module."""
+
+    def __init__(self, config: PinyinCodeConfig) -> None:
+        super().__init__()
+        if config.n_embd % config.n_head != 0:
+            raise ValueError("n_embd must be divisible by n_head")
+
+        self.n_head = config.n_head
+        self.head_dim = config.n_embd // config.n_head
+        self.dropout_p = config.dropout
+        self.qkv = nn.Linear(config.n_embd, 3 * config.n_embd)
+        self.proj = nn.Linear(config.n_embd, config.n_embd)
+        self.resid_dropout = nn.Dropout(config.dropout)
+
+    def forward(self, x: torch.Tensor, attention_mask: torch.Tensor | None = None) -> torch.Tensor:
+        batch_size, seq_len, embd = x.shape
+        q, k, v = self.qkv(x).split(embd, dim=2)
+
+        q = q.view(batch_size, seq_len, self.n_head, self.head_dim).transpose(1, 2)
+        k = k.view(batch_size, seq_len, self.n_head, self.head_dim).transpose(1, 2)
+        v = v.view(batch_size, seq_len, self.n_head, self.head_dim).transpose(1, 2)
+
+        dropout_p = self.dropout_p if self.training else 0.0
+        if attention_mask is not None:
+            causal_mask = torch.ones(
+                seq_len,
+                seq_len,
+                device=x.device,
+                dtype=torch.bool,
+            ).tril()
+            key_mask = attention_mask[:, None, None, :seq_len].to(dtype=torch.bool)
+            attn_mask = causal_mask.view(1, 1, seq_len, seq_len) & key_mask
+            y = F.scaled_dot_product_attention(
+                q,
+                k,
+                v,
+                attn_mask=attn_mask,
+                dropout_p=dropout_p,
+                is_causal=False,
+            )
+        else:
+            y = F.scaled_dot_product_attention(
+                q,
+                k,
+                v,
+                dropout_p=dropout_p,
+                is_causal=True,
+            )
+
+        y = y.transpose(1, 2).contiguous().view(batch_size, seq_len, embd)
+        return self.resid_dropout(self.proj(y))
+
+
+class FeedForward(nn.Module):
+    """Transformer MLP block."""
+
+    def __init__(self, config: PinyinCodeConfig) -> None:
+        super().__init__()
+        self.net = nn.Sequential(
+            nn.Linear(config.n_embd, 4 * config.n_embd),
+            nn.GELU(),
+            nn.Linear(4 * config.n_embd, config.n_embd),
+            nn.Dropout(config.dropout),
+        )
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return self.net(x)
+
+
+class TransformerBlock(nn.Module):
+    """Pre-norm Transformer block."""
+
+    def __init__(self, config: PinyinCodeConfig) -> None:
+        super().__init__()
+        self.ln_1 = nn.LayerNorm(config.n_embd)
+        self.attn = CausalSelfAttention(config)
+        self.ln_2 = nn.LayerNorm(config.n_embd)
+        self.mlp = FeedForward(config)
+
+    def forward(self, x: torch.Tensor, attention_mask: torch.Tensor | None = None) -> torch.Tensor:
+        x = x + self.attn(self.ln_1(x), attention_mask=attention_mask)
+        x = x + self.mlp(self.ln_2(x))
+        return x
+
+
+class PinyinCodePreTrainedModel(PreTrainedModel):
+    """Base class for pinyin-code Transformers models."""
+
+    config_class = PinyinCodeConfig
+    base_model_prefix = "pinyin_code"
+    supports_gradient_checkpointing = False
+
+    def _init_weights(self, module: nn.Module) -> None:
+        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)
+
+
+class PinyinCodeModel(PinyinCodePreTrainedModel):
+    """Base decoder model returned by ``AutoModel``."""
+
+    def __init__(self, config: PinyinCodeConfig, init_weights: bool = True) -> None:
+        super().__init__(config)
+        self.token_embedding = nn.Embedding(config.vocab_size, config.n_embd)
+        self.position_embedding = nn.Embedding(config.block_size, config.n_embd)
+        self.dropout = nn.Dropout(config.dropout)
+        self.blocks = nn.ModuleList(TransformerBlock(config) for _ in range(config.n_layer))
+        self.ln_f = nn.LayerNorm(config.n_embd)
+        if init_weights:
+            self.post_init()
+
+    def get_input_embeddings(self) -> nn.Embedding:
+        return self.token_embedding
+
+    def set_input_embeddings(self, value: nn.Embedding) -> None:
+        self.token_embedding = value
+
+    def forward(
+        self,
+        input_ids: torch.Tensor | None = None,
+        attention_mask: torch.Tensor | None = None,
+        inputs_embeds: torch.Tensor | None = None,
+        position_ids: torch.Tensor | None = None,
+        output_hidden_states: bool | None = None,
+        return_dict: bool | None = None,
+        **kwargs,
+    ) -> BaseModelOutput | tuple:
+        return_dict = True if return_dict is None else return_dict
+        output_hidden_states = (
+            self.config.output_hidden_states
+            if output_hidden_states is None
+            else output_hidden_states
+        )
+
+        if input_ids is None and inputs_embeds is None:
+            raise ValueError("You must provide either input_ids or inputs_embeds")
+        if input_ids is not None and inputs_embeds is not None:
+            raise ValueError("You cannot provide both input_ids and inputs_embeds")
+
+        if inputs_embeds is None:
+            _, seq_len = input_ids.shape
+            if seq_len > self.config.block_size:
+                raise ValueError(
+                    f"Sequence length {seq_len} exceeds block size {self.config.block_size}"
+                )
+            inputs_embeds = self.token_embedding(input_ids)
+        else:
+            seq_len = inputs_embeds.shape[1]
+            if seq_len > self.config.block_size:
+                raise ValueError(
+                    f"Sequence length {seq_len} exceeds block size {self.config.block_size}"
+                )
+
+        if position_ids is None:
+            if attention_mask is not None:
+                position_ids = attention_mask.long().cumsum(dim=-1) - 1
+                position_ids = position_ids.clamp_min(0)
+            else:
+                position_ids = torch.arange(seq_len, device=inputs_embeds.device)
+        position_ids = position_ids[:, -seq_len:] if position_ids.ndim == 2 else position_ids
+
+        x = inputs_embeds + self.position_embedding(position_ids)
+        x = self.dropout(x)
+        all_hidden_states = (x,) if output_hidden_states else None
+        for block in self.blocks:
+            x = block(x, attention_mask=attention_mask)
+            if output_hidden_states:
+                all_hidden_states = all_hidden_states + (x,)
+        hidden_states = self.ln_f(x)
+        if output_hidden_states:
+            all_hidden_states = all_hidden_states + (hidden_states,)
+
+        if not return_dict:
+            output = (hidden_states,)
+            if output_hidden_states:
+                output = output + (all_hidden_states,)
+            return output
+
+        return BaseModelOutput(
+            last_hidden_state=hidden_states,
+            hidden_states=all_hidden_states,
+        )
+
+
+class PinyinCodeForCausalLM(PinyinCodeModel, GenerationMixin):
+    """Compact GPT-style causal language model using the original architecture."""
+
+    _tied_weights_keys = {"lm_head.weight": "token_embedding.weight"}
+    _keys_to_ignore_on_load_missing = [r"lm_head\.weight"]
+
+    def __init__(self, config: PinyinCodeConfig) -> None:
+        super().__init__(config, init_weights=False)
+        self.lm_head = nn.Linear(config.n_embd, config.vocab_size, bias=False)
+        self.post_init()
+        self.tie_weights()
+
+    def get_output_embeddings(self) -> nn.Linear:
+        return self.lm_head
+
+    def set_output_embeddings(self, new_embeddings: nn.Linear) -> None:
+        self.lm_head = new_embeddings
+
+    def tie_weights(self, *args, **kwargs) -> None:
+        self.lm_head.weight = self.token_embedding.weight
+
+    def prepare_inputs_for_generation(
+        self,
+        input_ids: torch.Tensor,
+        past_key_values=None,
+        attention_mask: torch.Tensor | None = None,
+        **kwargs,
+    ) -> dict:
+        if input_ids.shape[1] > self.config.block_size:
+            input_ids = input_ids[:, -self.config.block_size :]
+            if attention_mask is not None:
+                attention_mask = attention_mask[:, -self.config.block_size :]
+        position_ids = None
+        if attention_mask is not None:
+            position_ids = attention_mask.long().cumsum(dim=-1) - 1
+            position_ids = position_ids.clamp_min(0)
+        return {
+            "input_ids": input_ids,
+            "attention_mask": attention_mask,
+            "position_ids": position_ids,
+        }
+
+    def forward(
+        self,
+        input_ids: torch.Tensor | None = None,
+        attention_mask: torch.Tensor | None = None,
+        labels: torch.Tensor | None = None,
+        inputs_embeds: torch.Tensor | None = None,
+        position_ids: torch.Tensor | None = None,
+        output_hidden_states: bool | None = None,
+        return_dict: bool | None = None,
+        **kwargs,
+    ) -> CausalLMOutput | tuple:
+        return_dict = True if return_dict is None else return_dict
+
+        decoder_outputs = PinyinCodeModel.forward(
+            self,
+            input_ids=input_ids,
+            attention_mask=attention_mask,
+            inputs_embeds=inputs_embeds,
+            position_ids=position_ids,
+            output_hidden_states=output_hidden_states,
+            return_dict=True,
+        )
+        logits = self.lm_head(decoder_outputs.last_hidden_state)
+
+        loss = None
+        if labels is not None:
+            loss = F.cross_entropy(
+                logits[:, :-1, :].contiguous().view(-1, logits.size(-1)),
+                labels[:, 1:].contiguous().view(-1),
+                ignore_index=-100,
+            )
+
+        if not return_dict:
+            output = (logits,)
+            if decoder_outputs.hidden_states is not None:
+                output = output + (decoder_outputs.hidden_states,)
+            return ((loss,) + output) if loss is not None else output
+
+        return CausalLMOutput(
+            loss=loss,
+            logits=logits,
+            hidden_states=decoder_outputs.hidden_states,
+        )

hf/tokenization_pinyin_code.py

@@ -0,0 +1,548 @@
+"""SentencePiece tokenizer wrapper for pinyin-code Transformers models."""
+
+from __future__ import annotations
+
+import logging
+import math
+import re
+import shutil
+import unicodedata
+from pathlib import Path
+from typing import Any
+
+import sentencepiece as spm
+from transformers import PreTrainedTokenizer
+
+
+CHINESE_RE = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]")
+CHINESE_SPAN_RE = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]+")
+PINYIN_CODE_TOKEN_RE = re.compile(
+    r"(?<![A-Za-z0-9])[A-Za-z]\d(?:[A-Za-z]\d)*(?![A-Za-z0-9])"
+)
+SPECIAL_MARKER_RE = re.compile(r"<[A-Z_]+>")
+PUNCTUATION = set(
+    "\u3002\uff0c\u3001\uff1f\uff01\uff1a\uff1b.,?!:;()[]{}<>\u300a\u300b"
+    "\u3010\u3011\u201c\u201d\"'\u2018\u2019\u300c\u300d\u300e\u300f"
+    "\u2014-~\u2026/\\"
+)
+LATIN_LETTER = (
+    r"A-Za-z\u00c0-\u00d6\u00d8-\u00f6\u00f8-\u00ff"
+    r"\u0100-\u017f\u0180-\u024f\u0250-\u02af"
+)
+LATIN_ALNUM_PATTERN = (
+    rf"(?:[{LATIN_LETTER}][{LATIN_LETTER}0-9]*"
+    rf"(?:[-_][{LATIN_LETTER}0-9]+)*|"
+    rf"[0-9]+[{LATIN_LETTER}][{LATIN_LETTER}0-9]*"
+    rf"(?:[-_][{LATIN_LETTER}0-9]+)*)"
+)
+LATIN_ALNUM_RE = re.compile(LATIN_ALNUM_PATTERN)
+URL_RE = re.compile(r"\b(?:https?://\S*|www\.\S+)", flags=re.I)
+DISCARDED_UNICODE_CATEGORIES = {"Cc", "Cf", "Co", "Cs", "Cn"}
+TOKEN_RE = re.compile(
+    r"<[A-Z_]+>|"
+    r"[\u3400-\u4dbf\u4e00-\u9fff]+|"
+    rf"{LATIN_ALNUM_PATTERN}|"
+    r"\S"
+)
+LABELS = {
+    "\u9898\u5e72": "<QUESTION>",
+    "\u9009\u9879": "<OPTIONS>",
+    "\u7b54\u6848": "<ANSWER>",
+    "\u89e3\u6790": "<EXPLANATION>",
+}
+PINYIN_FORMAT_ALIASES = {
+    "code": "pinyin-code",
+    "codes": "pinyin-code",
+    "pinyin-code": "pinyin-code",
+    "initial": "pinyin-initial",
+    "initials": "pinyin-initial",
+    "pinyin-initial": "pinyin-initial",
+    "hanzi": "hanzi",
+}
+
+
+def latin_token_to_model_token(token: str) -> str:
+    upper = token.upper()
+    return upper if upper in {"A", "B", "C", "D"} else token.lower()
+
+
+def should_preserve_fallback_token(token: str) -> bool:
+    if token == "\ufffd":
+        return False
+    for char in token:
+        category = unicodedata.category(char)
+        if category in DISCARDED_UNICODE_CATEGORIES:
+            return False
+        if category[0] not in {"L", "P", "S"}:
+            return False
+    return True
+
+
+class PinyinCodeTokenizer(PreTrainedTokenizer):
+    """Slow tokenizer that preserves the existing SentencePiece model."""
+
+    vocab_files_names = {"vocab_file": "tokenizer.model"}
+    model_input_names = ["input_ids", "attention_mask"]
+
+    def __init__(
+        self,
+        vocab_file: str,
+        add_bos_token: bool = False,
+        add_eos_token: bool = False,
+        transliteration: str = "pinyin-code",
+        pinyin_format: str | None = None,
+        use_jieba: bool = True,
+        jieba: bool | None = None,
+        **kwargs,
+    ) -> None:
+        self.vocab_file = vocab_file
+        self.sp_model = spm.SentencePieceProcessor(model_file=vocab_file)
+        self.add_bos_token = add_bos_token
+        self.add_eos_token = add_eos_token
+        self.transliteration = self._normalize_transliteration(
+            pinyin_format or transliteration
+        )
+        self.use_jieba = use_jieba if jieba is None else jieba
+
+        kwargs.setdefault("unk_token", self._piece_or_none(self.sp_model.unk_id()))
+        kwargs.setdefault("bos_token", self._piece_or_none(self.sp_model.bos_id()))
+        kwargs.setdefault("eos_token", self._piece_or_none(self.sp_model.eos_id()))
+        kwargs.setdefault("pad_token", self._piece_or_none(self.sp_model.pad_id()))
+        kwargs.setdefault("transliteration", self.transliteration)
+        kwargs.setdefault("pinyin_format", self.transliteration)
+        kwargs.setdefault("use_jieba", self.use_jieba)
+        kwargs.setdefault("jieba", self.use_jieba)
+        super().__init__(**kwargs)
+
+    def _normalize_transliteration(self, value: str) -> str:
+        normalized = PINYIN_FORMAT_ALIASES.get(value.lower())
+        if normalized is None:
+            allowed = ", ".join(sorted(set(PINYIN_FORMAT_ALIASES.values())))
+            raise ValueError(f"Unsupported transliteration {value!r}; choose from {allowed}")
+        return normalized
+
+    def _piece_or_none(self, token_id: int) -> str | None:
+        if token_id is None or token_id < 0:
+            return None
+        return self.sp_model.id_to_piece(token_id)
+
+    def _looks_preprocessed(self, text: str) -> bool:
+        if SPECIAL_MARKER_RE.search(text):
+            return True
+        if self.transliteration == "pinyin-code" and PINYIN_CODE_TOKEN_RE.search(text):
+            return True
+        return False
+
+    def _preprocess_raw_text(self, text: str) -> str:
+        if not CHINESE_RE.search(text) and self._looks_preprocessed(text):
+            return text
+        try:
+            from preprocessing.preprocess import (
+                hanzi_to_encoded,
+                process_text,
+                require_dependencies,
+            )
+        except ImportError:
+            return self._fallback_process_text(text)
+
+        require_dependencies()
+        if self.transliteration == "pinyin-code":
+            return hanzi_to_encoded(text, self.use_jieba)
+        return process_text(text, self.transliteration, self.use_jieba)
+
+    def _fallback_process_text(self, text: str) -> str:
+        if self.use_jieba:
+            try:
+                import jieba
+            except ImportError as exc:
+                raise ImportError(
+                    "Tokenizing raw Mandarin benchmark text with jieba segmentation "
+                    "requires jieba. Install the model dependencies before running "
+                    "lm_eval."
+                ) from exc
+
+            jieba.setLogLevel(logging.WARNING)
+        else:
+            jieba = None
+
+        if self.transliteration != "hanzi":
+            try:
+                from pypinyin import Style, pinyin
+            except ImportError as exc:
+                raise ImportError(
+                    "Tokenizing raw Mandarin benchmark text as pinyin requires pypinyin. "
+                    "Install the model dependencies before running lm_eval."
+                ) from exc
+
+
+        def normalize_text(value: str) -> str:
+            value = unicodedata.normalize("NFKC", value)
+            value = URL_RE.sub(" <URL> ", value)
+            value = re.sub(r"\$\$.*?\$\$", " <MATH> ", value, flags=re.DOTALL)
+            value = re.sub(r"[\uff08(]\s*[\uff09)]", " <BLANK> ", value)
+            for label, marker in LABELS.items():
+                value = re.sub(rf"{label}\s*[:\uff1a]", f" {marker} ", value)
+            value = re.sub(
+                rf"(?<![{LATIN_LETTER}])yes(?![{LATIN_LETTER}])",
+                " <YES> ",
+                value,
+                flags=re.I,
+            )
+            value = re.sub(
+                rf"(?<![{LATIN_LETTER}])no(?![{LATIN_LETTER}])",
+                " <NO> ",
+                value,
+                flags=re.I,
+            )
+            value = re.sub(
+                rf"(?<![{LATIN_LETTER}])[ABCD](?=\s*[:\uff1a.\uff0e\u3001\)])",
+                r" \g<0> ",
+                value,
+            )
+            value = re.sub(
+                rf"(?<![{LATIN_LETTER}0-9])[-+]?\d+(?:[.,]\d+)*(?:%|\uff05)?"
+                rf"(?![{LATIN_LETTER}0-9])",
+                " <NUM> ",
+                value,
+            )
+            value = value.replace("\uff08", "(").replace("\uff09", ")")
+            return re.sub(r"\s+", " ", value).strip()
+
+        def split_tone3_syllable(syllable: str) -> tuple[str, int]:
+            match = re.fullmatch(r"([a-z\u00fcv]+)([1-5]?)", syllable.lower())
+            if not match:
+                return syllable, 5
+            plain, tone = match.groups()
+            return plain, int(tone or "5")
+
+        def length_digit_offset(syllable: str) -> int:
+            return min(max(len(syllable), 1), 5) - 1
+
+        def syllable_to_initial_code(syllable: str) -> str:
+            plain, tone = split_tone3_syllable(syllable)
+            if not plain:
+                return ""
+            tone_offset = 5 if tone in {3, 4, 5} else 0
+            digit = tone_offset + length_digit_offset(plain)
+            initial = plain[0].upper() if tone in {1, 3, 5} else plain[0].lower()
+            return f"{initial}{digit}"
+
+        def syllable_to_initial_letter(syllable: str) -> str:
+            plain, _ = split_tone3_syllable(syllable)
+            return plain[:1].lower()
+
+        def convert_word(word: str) -> str:
+            if self.transliteration == "hanzi":
+                return word
+            syllables = pinyin(word, style=Style.TONE3, heteronym=False, errors="ignore")
+            if self.transliteration == "pinyin-code":
+                codes = [
+                    syllable_to_initial_code(item[0])
+                    for item in syllables
+                    if item and item[0]
+                ]
+                return "".join(code for code in codes if code)
+            initials = [
+                syllable_to_initial_letter(item[0])
+                for item in syllables
+                if item and item[0]
+            ]
+            return "".join(initial for initial in initials if initial)
+
+        def tokenize_chinese_span(value: str) -> list[str]:
+            tokens = []
+            words = jieba.cut(value, cut_all=False) if self.use_jieba else value
+            for word in words:
+                word = word.strip()
+                if word and CHINESE_SPAN_RE.search(word):
+                    token = convert_word(word)
+                    if token:
+                        tokens.append(token)
+            return tokens
+
+        tokens = []
+        for part in TOKEN_RE.findall(normalize_text(text)):
+            if part.startswith("<") and part.endswith(">"):
+                tokens.append(part)
+            elif CHINESE_SPAN_RE.fullmatch(part):
+                tokens.extend(tokenize_chinese_span(part))
+            elif part in PUNCTUATION:
+                tokens.append(part)
+            elif LATIN_ALNUM_RE.fullmatch(part):
+                tokens.append(latin_token_to_model_token(part))
+            elif part.isdigit():
+                tokens.append("<NUM>")
+            elif should_preserve_fallback_token(part):
+                tokens.append(part.lower())
+
+        return " ".join(tokens)
+
+    def _preprocess_tokenizer_input(self, value: Any) -> Any:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return self._preprocess_raw_text(value)
+        if isinstance(value, tuple):
+            return tuple(self._preprocess_tokenizer_input(item) for item in value)
+        if isinstance(value, list):
+            return [self._preprocess_tokenizer_input(item) for item in value]
+        return value
+
+    def _non_content_token_ids(self) -> set[int]:
+        return {
+            token_id
+            for token_id in (
+                self.pad_token_id,
+                self.bos_token_id,
+                self.eos_token_id,
+                self.cls_token_id,
+                self.sep_token_id,
+                self.mask_token_id,
+            )
+            if token_id is not None
+        }
+
+    def _offset_source_text(self, value: Any, is_split_into_words: bool = False) -> str:
+        if value is None:
+            return ""
+        if isinstance(value, str):
+            return value
+        if isinstance(value, tuple):
+            return " ".join(self._offset_source_text(item) for item in value)
+        if isinstance(value, list):
+            separator = " " if is_split_into_words else ""
+            return separator.join(self._offset_source_text(item) for item in value)
+        return str(value)
+
+    def _synthetic_offset_mapping(self, text: Any, input_ids: Any, is_split_into_words: bool = False) -> list[tuple[int, int]]:
+        """Return slow-tokenizer-compatible offsets for evaluators that require them.
+
+        SentencePiece offsets are not available for this Python tokenizer because
+        raw Mandarin text is preprocessed into pinyin-code before encoding. These
+        spans conservatively distribute non-special tokens across the original
+        text so suffix/completion masking code can run without requiring a fast
+        tokenizer.
+        """
+        ids = input_ids.tolist() if hasattr(input_ids, "tolist") else list(input_ids)
+        source = self._offset_source_text(text, is_split_into_words=is_split_into_words)
+        source_length = len(source)
+        if not ids:
+            return []
+        if source_length == 0:
+            return [(0, 0) for _ in ids]
+
+        non_content_ids = self._non_content_token_ids()
+        content_positions = [
+            index for index, token_id in enumerate(ids) if int(token_id) not in non_content_ids
+        ]
+        if not content_positions:
+            return [(0, 0) for _ in ids]
+
+        offsets = [(0, 0) for _ in ids]
+        count = len(content_positions)
+        for ordinal, position in enumerate(content_positions):
+            start = math.floor(ordinal * source_length / count)
+            end = math.ceil((ordinal + 1) * source_length / count)
+            if end <= start:
+                end = min(source_length, start + 1)
+            offsets[position] = (start, end)
+        return offsets
+
+    def _with_optional_offsets(
+        self,
+        encoding,
+        original_text: Any,
+        return_offsets_mapping: bool,
+        is_split_into_words: bool = False,
+        return_tensors: str | None = None,
+    ):
+        if not return_offsets_mapping:
+            return encoding
+
+        input_ids = encoding["input_ids"]
+        tensor_input = hasattr(input_ids, "ndim")
+        input_ids_list = input_ids.tolist() if tensor_input else input_ids
+
+        is_batched = False
+        if tensor_input:
+            is_batched = input_ids.ndim > 1
+        elif input_ids_list and isinstance(input_ids_list[0], list):
+            is_batched = True
+
+        if is_batched:
+            if isinstance(original_text, list) and not is_split_into_words:
+                texts = original_text
+            else:
+                texts = [original_text] * len(input_ids_list)
+            offsets = [
+                self._synthetic_offset_mapping(text, ids, is_split_into_words=is_split_into_words)
+                for text, ids in zip(texts, input_ids_list)
+            ]
+        else:
+            offsets = self._synthetic_offset_mapping(
+                original_text,
+                input_ids_list,
+                is_split_into_words=is_split_into_words,
+            )
+
+        if return_tensors == "pt" or tensor_input:
+            try:
+                import torch
+
+                offsets = torch.tensor(offsets, dtype=torch.long)
+            except ImportError:
+                pass
+        encoding["offset_mapping"] = offsets
+        return encoding
+
+    def __call__(self, text=None, text_pair=None, *args, **kwargs):
+        original_text = text
+        return_offsets_mapping = bool(kwargs.pop("return_offsets_mapping", False))
+        is_split_into_words = bool(kwargs.get("is_split_into_words", False))
+        return_tensors = kwargs.get("return_tensors")
+
+        if "text_target" in kwargs:
+            kwargs["text_target"] = self._preprocess_tokenizer_input(kwargs["text_target"])
+        if "text_pair_target" in kwargs:
+            kwargs["text_pair_target"] = self._preprocess_tokenizer_input(
+                kwargs["text_pair_target"]
+            )
+
+        text = self._preprocess_tokenizer_input(text)
+        text_pair = self._preprocess_tokenizer_input(text_pair)
+        if text_pair is None:
+            encoding = super().__call__(text, *args, **kwargs)
+        else:
+            encoding = super().__call__(text, text_pair, *args, **kwargs)
+        return self._with_optional_offsets(
+            encoding,
+            original_text,
+            return_offsets_mapping,
+            is_split_into_words=is_split_into_words,
+            return_tensors=return_tensors,
+        )
+
+    def encode(self, text, text_pair=None, add_special_tokens=True, *args, **kwargs):
+        kwargs["add_special_tokens"] = add_special_tokens
+        text = self._preprocess_tokenizer_input(text)
+        text_pair = self._preprocess_tokenizer_input(text_pair)
+        if text_pair is None:
+            return super().encode(text, *args, **kwargs)
+        return super().encode(text, text_pair, *args, **kwargs)
+
+    def encode_plus(self, text, text_pair=None, *args, **kwargs):
+        original_text = text
+        return_offsets_mapping = bool(kwargs.pop("return_offsets_mapping", False))
+        is_split_into_words = bool(kwargs.get("is_split_into_words", False))
+        return_tensors = kwargs.get("return_tensors")
+
+        text = self._preprocess_tokenizer_input(text)
+        text_pair = self._preprocess_tokenizer_input(text_pair)
+        if text_pair is None:
+            encoding = super().encode_plus(text, *args, **kwargs)
+        else:
+            encoding = super().encode_plus(text, text_pair, *args, **kwargs)
+        return self._with_optional_offsets(
+            encoding,
+            original_text,
+            return_offsets_mapping,
+            is_split_into_words=is_split_into_words,
+            return_tensors=return_tensors,
+        )
+
+    def batch_encode_plus(self, batch_text_or_text_pairs, *args, **kwargs):
+        original_batch = batch_text_or_text_pairs
+        return_offsets_mapping = bool(kwargs.pop("return_offsets_mapping", False))
+        is_split_into_words = bool(kwargs.get("is_split_into_words", False))
+        return_tensors = kwargs.get("return_tensors")
+
+        batch_text_or_text_pairs = self._preprocess_tokenizer_input(
+            batch_text_or_text_pairs
+        )
+        encoding = super().batch_encode_plus(batch_text_or_text_pairs, *args, **kwargs)
+        return self._with_optional_offsets(
+            encoding,
+            original_batch,
+            return_offsets_mapping,
+            is_split_into_words=is_split_into_words,
+            return_tensors=return_tensors,
+        )
+
+    @property
+    def vocab_size(self) -> int:
+        return self.sp_model.get_piece_size()
+
+    def get_vocab(self) -> dict[str, int]:
+        vocab = {self.sp_model.id_to_piece(i): i for i in range(self.vocab_size)}
+        vocab.update(self.added_tokens_encoder)
+        return vocab
+
+    def _tokenize(self, text: str) -> list[str]:
+        text = self._preprocess_raw_text(text)
+        return self.sp_model.encode(text, out_type=str)
+
+    def _convert_token_to_id(self, token: str) -> int:
+        return self.sp_model.piece_to_id(token)
+
+    def _convert_id_to_token(self, index: int) -> str:
+        return self.sp_model.id_to_piece(index)
+
+    def convert_tokens_to_string(self, tokens: list[str]) -> str:
+        return self.sp_model.decode(tokens)
+
+    def build_inputs_with_special_tokens(
+        self,
+        token_ids_0: list[int],
+        token_ids_1: list[int] | None = None,
+    ) -> list[int]:
+        output = list(token_ids_0)
+        if self.add_bos_token and self.bos_token_id is not None:
+            output = [self.bos_token_id] + output
+        if self.add_eos_token and self.eos_token_id is not None:
+            output = output + [self.eos_token_id]
+        if token_ids_1 is not None:
+            output += list(token_ids_1)
+            if self.add_eos_token and self.eos_token_id is not None:
+                output.append(self.eos_token_id)
+        return output
+
+    def get_special_tokens_mask(
+        self,
+        token_ids_0: list[int],
+        token_ids_1: list[int] | None = None,
+        already_has_special_tokens: bool = False,
+    ) -> list[int]:
+        if already_has_special_tokens:
+            special_ids = set(self.all_special_ids)
+            return [1 if token_id in special_ids else 0 for token_id in token_ids_0]
+
+        mask = [0] * len(token_ids_0)
+        if self.add_bos_token and self.bos_token_id is not None:
+            mask = [1] + mask
+        if self.add_eos_token and self.eos_token_id is not None:
+            mask = mask + [1]
+        if token_ids_1 is not None:
+            mask += [0] * len(token_ids_1)
+            if self.add_eos_token and self.eos_token_id is not None:
+                mask.append(1)
+        return mask
+
+    def create_token_type_ids_from_sequences(
+        self,
+        token_ids_0: list[int],
+        token_ids_1: list[int] | None = None,
+    ) -> list[int]:
+        return [0] * len(self.build_inputs_with_special_tokens(token_ids_0, token_ids_1))
+
+    def save_vocabulary(self, save_directory: str, filename_prefix: str | None = None) -> tuple[str]:
+        output_name = "tokenizer.model"
+        if filename_prefix:
+            output_name = f"{filename_prefix}-{output_name}"
+        output_path = Path(save_directory) / output_name
+        if Path(self.vocab_file).resolve() != output_path.resolve():
+            shutil.copyfile(self.vocab_file, output_path)
+        return (str(output_path),)
+
+
+class EncodedMandarinTokenizer(PinyinCodeTokenizer):
+    """Tokenizer wrapper that hides Hanzi-to-encoded-Mandarin preprocessing."""

model.safetensors

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:403f1d74ffa4294902d2c857b80eae0d8a66411acd4ff0c66d54ad20be6ab0e2
+size 134706152

modeling_pinyin_code.py

@@ -0,0 +1,287 @@
+"""Transformers-compatible implementation of the pinyin-code causal LM."""
+
+from __future__ import annotations
+
+import torch
+from torch import nn
+from torch.nn import functional as F
+from transformers import PreTrainedModel
+from transformers.generation import GenerationMixin
+from transformers.modeling_outputs import BaseModelOutput, CausalLMOutput
+
+from .configuration_pinyin_code import PinyinCodeConfig
+
+
+class CausalSelfAttention(nn.Module):
+    """Multi-head masked self-attention matching the original training module."""
+
+    def __init__(self, config: PinyinCodeConfig) -> None:
+        super().__init__()
+        if config.n_embd % config.n_head != 0:
+            raise ValueError("n_embd must be divisible by n_head")
+
+        self.n_head = config.n_head
+        self.head_dim = config.n_embd // config.n_head
+        self.dropout_p = config.dropout
+        self.qkv = nn.Linear(config.n_embd, 3 * config.n_embd)
+        self.proj = nn.Linear(config.n_embd, config.n_embd)
+        self.resid_dropout = nn.Dropout(config.dropout)
+
+    def forward(self, x: torch.Tensor, attention_mask: torch.Tensor | None = None) -> torch.Tensor:
+        batch_size, seq_len, embd = x.shape
+        q, k, v = self.qkv(x).split(embd, dim=2)
+
+        q = q.view(batch_size, seq_len, self.n_head, self.head_dim).transpose(1, 2)
+        k = k.view(batch_size, seq_len, self.n_head, self.head_dim).transpose(1, 2)
+        v = v.view(batch_size, seq_len, self.n_head, self.head_dim).transpose(1, 2)
+
+        dropout_p = self.dropout_p if self.training else 0.0
+        if attention_mask is not None:
+            causal_mask = torch.ones(
+                seq_len,
+                seq_len,
+                device=x.device,
+                dtype=torch.bool,
+            ).tril()
+            key_mask = attention_mask[:, None, None, :seq_len].to(dtype=torch.bool)
+            attn_mask = causal_mask.view(1, 1, seq_len, seq_len) & key_mask
+            y = F.scaled_dot_product_attention(
+                q,
+                k,
+                v,
+                attn_mask=attn_mask,
+                dropout_p=dropout_p,
+                is_causal=False,
+            )
+        else:
+            y = F.scaled_dot_product_attention(
+                q,
+                k,
+                v,
+                dropout_p=dropout_p,
+                is_causal=True,
+            )
+
+        y = y.transpose(1, 2).contiguous().view(batch_size, seq_len, embd)
+        return self.resid_dropout(self.proj(y))
+
+
+class FeedForward(nn.Module):
+    """Transformer MLP block."""
+
+    def __init__(self, config: PinyinCodeConfig) -> None:
+        super().__init__()
+        self.net = nn.Sequential(
+            nn.Linear(config.n_embd, 4 * config.n_embd),
+            nn.GELU(),
+            nn.Linear(4 * config.n_embd, config.n_embd),
+            nn.Dropout(config.dropout),
+        )
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return self.net(x)
+
+
+class TransformerBlock(nn.Module):
+    """Pre-norm Transformer block."""
+
+    def __init__(self, config: PinyinCodeConfig) -> None:
+        super().__init__()
+        self.ln_1 = nn.LayerNorm(config.n_embd)
+        self.attn = CausalSelfAttention(config)
+        self.ln_2 = nn.LayerNorm(config.n_embd)
+        self.mlp = FeedForward(config)
+
+    def forward(self, x: torch.Tensor, attention_mask: torch.Tensor | None = None) -> torch.Tensor:
+        x = x + self.attn(self.ln_1(x), attention_mask=attention_mask)
+        x = x + self.mlp(self.ln_2(x))
+        return x
+
+
+class PinyinCodePreTrainedModel(PreTrainedModel):
+    """Base class for pinyin-code Transformers models."""
+
+    config_class = PinyinCodeConfig
+    base_model_prefix = "pinyin_code"
+    supports_gradient_checkpointing = False
+
+    def _init_weights(self, module: nn.Module) -> None:
+        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)
+
+
+class PinyinCodeModel(PinyinCodePreTrainedModel):
+    """Base decoder model returned by ``AutoModel``."""
+
+    def __init__(self, config: PinyinCodeConfig, init_weights: bool = True) -> None:
+        super().__init__(config)
+        self.token_embedding = nn.Embedding(config.vocab_size, config.n_embd)
+        self.position_embedding = nn.Embedding(config.block_size, config.n_embd)
+        self.dropout = nn.Dropout(config.dropout)
+        self.blocks = nn.ModuleList(TransformerBlock(config) for _ in range(config.n_layer))
+        self.ln_f = nn.LayerNorm(config.n_embd)
+        if init_weights:
+            self.post_init()
+
+    def get_input_embeddings(self) -> nn.Embedding:
+        return self.token_embedding
+
+    def set_input_embeddings(self, value: nn.Embedding) -> None:
+        self.token_embedding = value
+
+    def forward(
+        self,
+        input_ids: torch.Tensor | None = None,
+        attention_mask: torch.Tensor | None = None,
+        inputs_embeds: torch.Tensor | None = None,
+        position_ids: torch.Tensor | None = None,
+        output_hidden_states: bool | None = None,
+        return_dict: bool | None = None,
+        **kwargs,
+    ) -> BaseModelOutput | tuple:
+        return_dict = True if return_dict is None else return_dict
+        output_hidden_states = (
+            self.config.output_hidden_states
+            if output_hidden_states is None
+            else output_hidden_states
+        )
+
+        if input_ids is None and inputs_embeds is None:
+            raise ValueError("You must provide either input_ids or inputs_embeds")
+        if input_ids is not None and inputs_embeds is not None:
+            raise ValueError("You cannot provide both input_ids and inputs_embeds")
+
+        if inputs_embeds is None:
+            _, seq_len = input_ids.shape
+            if seq_len > self.config.block_size:
+                raise ValueError(
+                    f"Sequence length {seq_len} exceeds block size {self.config.block_size}"
+                )
+            inputs_embeds = self.token_embedding(input_ids)
+        else:
+            seq_len = inputs_embeds.shape[1]
+            if seq_len > self.config.block_size:
+                raise ValueError(
+                    f"Sequence length {seq_len} exceeds block size {self.config.block_size}"
+                )
+
+        if position_ids is None:
+            if attention_mask is not None:
+                position_ids = attention_mask.long().cumsum(dim=-1) - 1
+                position_ids = position_ids.clamp_min(0)
+            else:
+                position_ids = torch.arange(seq_len, device=inputs_embeds.device)
+        position_ids = position_ids[:, -seq_len:] if position_ids.ndim == 2 else position_ids
+
+        x = inputs_embeds + self.position_embedding(position_ids)
+        x = self.dropout(x)
+        all_hidden_states = (x,) if output_hidden_states else None
+        for block in self.blocks:
+            x = block(x, attention_mask=attention_mask)
+            if output_hidden_states:
+                all_hidden_states = all_hidden_states + (x,)
+        hidden_states = self.ln_f(x)
+        if output_hidden_states:
+            all_hidden_states = all_hidden_states + (hidden_states,)
+
+        if not return_dict:
+            output = (hidden_states,)
+            if output_hidden_states:
+                output = output + (all_hidden_states,)
+            return output
+
+        return BaseModelOutput(
+            last_hidden_state=hidden_states,
+            hidden_states=all_hidden_states,
+        )
+
+
+class PinyinCodeForCausalLM(PinyinCodeModel, GenerationMixin):
+    """Compact GPT-style causal language model using the original architecture."""
+
+    _tied_weights_keys = {"lm_head.weight": "token_embedding.weight"}
+    _keys_to_ignore_on_load_missing = [r"lm_head\.weight"]
+
+    def __init__(self, config: PinyinCodeConfig) -> None:
+        super().__init__(config, init_weights=False)
+        self.lm_head = nn.Linear(config.n_embd, config.vocab_size, bias=False)
+        self.post_init()
+        self.tie_weights()
+
+    def get_output_embeddings(self) -> nn.Linear:
+        return self.lm_head
+
+    def set_output_embeddings(self, new_embeddings: nn.Linear) -> None:
+        self.lm_head = new_embeddings
+
+    def tie_weights(self, *args, **kwargs) -> None:
+        self.lm_head.weight = self.token_embedding.weight
+
+    def prepare_inputs_for_generation(
+        self,
+        input_ids: torch.Tensor,
+        past_key_values=None,
+        attention_mask: torch.Tensor | None = None,
+        **kwargs,
+    ) -> dict:
+        if input_ids.shape[1] > self.config.block_size:
+            input_ids = input_ids[:, -self.config.block_size :]
+            if attention_mask is not None:
+                attention_mask = attention_mask[:, -self.config.block_size :]
+        position_ids = None
+        if attention_mask is not None:
+            position_ids = attention_mask.long().cumsum(dim=-1) - 1
+            position_ids = position_ids.clamp_min(0)
+        return {
+            "input_ids": input_ids,
+            "attention_mask": attention_mask,
+            "position_ids": position_ids,
+        }
+
+    def forward(
+        self,
+        input_ids: torch.Tensor | None = None,
+        attention_mask: torch.Tensor | None = None,
+        labels: torch.Tensor | None = None,
+        inputs_embeds: torch.Tensor | None = None,
+        position_ids: torch.Tensor | None = None,
+        output_hidden_states: bool | None = None,
+        return_dict: bool | None = None,
+        **kwargs,
+    ) -> CausalLMOutput | tuple:
+        return_dict = True if return_dict is None else return_dict
+
+        decoder_outputs = PinyinCodeModel.forward(
+            self,
+            input_ids=input_ids,
+            attention_mask=attention_mask,
+            inputs_embeds=inputs_embeds,
+            position_ids=position_ids,
+            output_hidden_states=output_hidden_states,
+            return_dict=True,
+        )
+        logits = self.lm_head(decoder_outputs.last_hidden_state)
+
+        loss = None
+        if labels is not None:
+            loss = F.cross_entropy(
+                logits[:, :-1, :].contiguous().view(-1, logits.size(-1)),
+                labels[:, 1:].contiguous().view(-1),
+                ignore_index=-100,
+            )
+
+        if not return_dict:
+            output = (logits,)
+            if decoder_outputs.hidden_states is not None:
+                output = output + (decoder_outputs.hidden_states,)
+            return ((loss,) + output) if loss is not None else output
+
+        return CausalLMOutput(
+            loss=loss,
+            logits=logits,
+            hidden_states=decoder_outputs.hidden_states,
+        )

preprocessing/preprocess.py

@@ -0,0 +1,333 @@
+"""Preprocess BabyLM Mandarin JSONL into word-preserving pinyin initial codes."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import re
+import sys
+import unicodedata
+from pathlib import Path
+from typing import Any, Iterable, Literal
+
+import jieba
+from pypinyin import Style, pinyin
+
+
+LABELS = {
+    "题干": "<QUESTION>",
+    "选项": "<OPTIONS>",
+    "答案": "<ANSWER>",
+    "解析": "<EXPLANATION>",
+}
+
+PUNCTUATION = set("。，、？！：；.,?!:;()[]{}<>《》【】“”\"'‘’「」『』—-~…/\\")
+CHINESE_RE = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]+")
+Transliteration = Literal["pinyin-code", "pinyin-initial", "hanzi"]
+LATIN_LETTER = r"A-Za-zÀ-ÖØ-öø-ÿĀ-ſƀ-ɏɐ-ʯ"
+LATIN_ALNUM_PATTERN = (
+    rf"(?:[{LATIN_LETTER}][{LATIN_LETTER}0-9]*"
+    rf"(?:[-_][{LATIN_LETTER}0-9]+)*|"
+    rf"[0-9]+[{LATIN_LETTER}][{LATIN_LETTER}0-9]*"
+    rf"(?:[-_][{LATIN_LETTER}0-9]+)*)"
+)
+LATIN_ALNUM_RE = re.compile(LATIN_ALNUM_PATTERN)
+URL_RE = re.compile(r"\b(?:https?://\S*|www\.\S+)", flags=re.I)
+DISCARDED_UNICODE_CATEGORIES = {"Cc", "Cf", "Co", "Cs", "Cn"}
+
+# Match protected markers before ordinary words so tokens like <ANSWER> survive
+# the later English/punctuation handling as a single vocabulary item.
+TOKEN_RE = re.compile(
+    r"<[A-Z_]+>|"
+    r"[\u3400-\u4dbf\u4e00-\u9fff]+|"
+    rf"{LATIN_ALNUM_PATTERN}|"
+    r"\S"
+)
+
+
+def require_dependencies() -> None:
+    """Fail early with a concise install hint and quiet jieba startup logging."""
+    if jieba is None or Style is None or pinyin is None:
+        raise SystemExit(
+            "Missing dependency: install with `py -m pip install jieba pypinyin`."
+        )
+    jieba.setLogLevel(logging.WARNING)
+
+
+def normalize_text(text: str) -> str:
+    """Replace task-specific surface forms with stable special tokens.
+
+    This happens before tokenization so multi-character patterns such as
+    ``$$...$$`` and empty brackets cannot be split into punctuation pieces.
+    """
+    text = unicodedata.normalize("NFKC", text)
+    text = URL_RE.sub(" <URL> ", text)
+    text = re.sub(r"\$\$.*?\$\$", " <MATH> ", text, flags=re.DOTALL)
+    text = re.sub(r"[（(]\s*[）)]", " <BLANK> ", text)
+
+    for label, marker in LABELS.items():
+        text = re.sub(rf"{label}\s*[:：]", f" {marker} ", text)
+
+    text = re.sub(
+        rf"(?<![{LATIN_LETTER}])yes(?![{LATIN_LETTER}])",
+        " <YES> ",
+        text,
+        flags=re.I,
+    )
+    text = re.sub(
+        rf"(?<![{LATIN_LETTER}])no(?![{LATIN_LETTER}])",
+        " <NO> ",
+        text,
+        flags=re.I,
+    )
+    text = re.sub(
+        rf"(?<![{LATIN_LETTER}])[ABCD](?=\s*[:：.．、\)])",
+        r" \g<0> ",
+        text,
+    )
+    text = re.sub(
+        rf"(?<![{LATIN_LETTER}0-9])[-+]?\d+(?:[.,]\d+)*(?:%|％)?"
+        rf"(?![{LATIN_LETTER}0-9])",
+        " <NUM> ",
+        text,
+    )
+
+    text = text.replace("（", "(").replace("）", ")")
+    text = re.sub(r"\s+", " ", text)
+    return text.strip()
+
+
+def latin_token_to_model_token(token: str) -> str:
+    """Normalize non-Mandarin alphanumeric tokens without losing option labels."""
+    upper = token.upper()
+    return upper if upper in {"A", "B", "C", "D"} else token.lower()
+
+
+def should_preserve_fallback_token(token: str) -> bool:
+    """Return true for visible non-Hanzi letters, punctuation, and symbols."""
+    if token == "\ufffd":
+        return False
+    for char in token:
+        category = unicodedata.category(char)
+        if category in DISCARDED_UNICODE_CATEGORIES:
+            return False
+        if category[0] not in {"L", "P", "S"}:
+            return False
+    return True
+
+
+def split_tone3_syllable(syllable: str) -> tuple[str, int]:
+    """Return the plain pinyin syllable and its tone number.
+
+    ``Style.TONE3`` writes tones as final digits, but neutral tone syllables have
+    no digit. Treat those digitless cases as fifth tone.
+    """
+    match = re.fullmatch(r"([a-züv]+)([1-5]?)", syllable.lower())
+    if not match:
+        return syllable, 5
+
+    plain, tone = match.groups()
+    return plain, int(tone or "5")
+
+
+def length_digit_offset(syllable: str) -> int:
+    """Map pinyin syllable length to the requested 0-4 digit offset."""
+    return min(max(len(syllable), 1), 5) - 1
+
+
+def syllable_to_initial_code(syllable: str) -> str:
+    """Convert one pinyin syllable with tone into ``initial + digit``.
+
+    Tone controls initial casing and whether the digit starts from 0 or 5:
+    tones 1/3/5 use uppercase initials, while tones 2/4 use lowercase initials.
+    Syllable length then adds the 0-4 offset that makes the final digit.
+    """
+    plain, tone = split_tone3_syllable(syllable)
+    if not plain:
+        return ""
+
+    tone_offset = 5 if tone in {3, 4, 5} else 0
+    digit = tone_offset + length_digit_offset(plain)
+    initial = plain[0].upper() if tone in {1, 3, 5} else plain[0].lower()
+    return f"{initial}{digit}"
+
+
+def syllable_to_initial_letter(syllable: str) -> str:
+    """Convert one pinyin syllable with tone into its lowercase first letter."""
+    plain, _ = split_tone3_syllable(syllable)
+    return plain[:1].lower()
+
+
+def chinese_word_to_initial_codes(word: str) -> str:
+    """Convert one already-segmented Chinese word to one compact code token.
+
+    The important representation choice is preserved: jieba decides the word
+    boundary, and all syllable codes inside that word are concatenated. For
+    example, ``我们`` becomes ``W6M7`` rather than ``W6 M7``.
+    """
+    syllables = pinyin(word, style=Style.TONE3, heteronym=False, errors="ignore")
+    codes = [syllable_to_initial_code(item[0]) for item in syllables if item and item[0]]
+    return "".join(code for code in codes if code)
+
+
+def chinese_word_to_initial_letters(word: str) -> str:
+    """Convert one already-segmented Chinese word to lowercase pinyin initials."""
+    syllables = pinyin(word, style=Style.TONE3, heteronym=False, errors="ignore")
+    initials = [
+        syllable_to_initial_letter(item[0]) for item in syllables if item and item[0]
+    ]
+    return "".join(initial for initial in initials if initial)
+
+
+def chinese_word_to_transliteration(word: str, transliteration: Transliteration) -> str:
+    """Convert one segmented Chinese word using the requested transliteration."""
+    if transliteration == "pinyin-code":
+        return chinese_word_to_initial_codes(word)
+    if transliteration == "pinyin-initial":
+        return chinese_word_to_initial_letters(word)
+    if transliteration == "hanzi":
+        return word
+    raise ValueError(f"Unsupported transliteration: {transliteration}")
+
+
+def tokenize_chinese_span(
+    text: str,
+    transliteration: Transliteration = "pinyin-code",
+    use_jieba: bool = True,
+) -> Iterable[str]:
+    """Emit one token per jieba word or per Hanzi character."""
+    words = jieba.cut(text, cut_all=False) if use_jieba else text
+    for word in words:
+        word = word.strip()
+        if not word:
+            continue
+        if CHINESE_RE.search(word):
+            token = chinese_word_to_transliteration(word, transliteration)
+            if token:
+                yield token
+
+
+def process_text(
+    text: str,
+    transliteration: Transliteration = "pinyin-code",
+    use_jieba: bool = True,
+) -> str:
+    """Convert one raw document string into the final space-separated token line."""
+    tokens: list[str] = []
+    for part in TOKEN_RE.findall(normalize_text(text)):
+        if part.startswith("<") and part.endswith(">"):
+            tokens.append(part)
+        elif CHINESE_RE.fullmatch(part):
+            tokens.extend(tokenize_chinese_span(part, transliteration, use_jieba))
+        elif part in PUNCTUATION:
+            tokens.append(part)
+        elif LATIN_ALNUM_RE.fullmatch(part):
+            tokens.append(latin_token_to_model_token(part))
+        elif part.isdigit():
+            tokens.append("<NUM>")
+        elif should_preserve_fallback_token(part):
+            tokens.append(part.lower())
+
+    return " ".join(tokens)
+
+
+def hanzi_to_encoded(text: str, use_jieba: bool = True) -> str:
+    """Convert normal Hanzi/Mandarin text to the compact initial+digit encoding."""
+    require_dependencies()
+    return process_text(text, "pinyin-code", use_jieba)
+
+
+def read_jsonl(path: Path) -> Iterable[dict[str, Any]]:
+    """Yield JSON objects from UTF-8 JSONL, tolerating a leading BOM if present."""
+    with path.open("r", encoding="utf-8-sig") as handle:
+        for line_number, line in enumerate(handle, start=1):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                yield json.loads(line)
+            except json.JSONDecodeError as exc:
+                raise ValueError(f"Invalid JSON on line {line_number}: {exc}") from exc
+
+
+def preprocess_file(
+    input_path: Path,
+    output_path: Path,
+    preview_count: int,
+    transliteration: Transliteration = "pinyin-code",
+    use_jieba: bool = True,
+) -> int:
+    """Stream input documents to output while retaining a small preview buffer."""
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    written = 0
+    previews: list[tuple[str, str]] = []
+
+    with output_path.open("w", encoding="utf-8", newline="\n") as out:
+        for obj in read_jsonl(input_path):
+            text = str(obj.get("text", ""))
+            processed = process_text(text, transliteration, use_jieba)
+            out.write(processed)
+            out.write("\n")
+            written += 1
+
+            if len(previews) < preview_count:
+                previews.append((text, processed))
+
+    for original, processed in previews:
+        print("ORIGINAL:")
+        print(original)
+        print("PROCESSED:")
+        print(processed)
+        print()
+
+    return written
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Convert BabyLM Mandarin JSONL text fields to model-ready tokens."
+    )
+    parser.add_argument("--input", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--preview", type=int, default=3)
+    parser.add_argument(
+        "--transliteration",
+        choices=("pinyin-code", "pinyin-initial", "hanzi"),
+        default="pinyin-code",
+        help=(
+            "Mandarin transliteration to emit: 'pinyin-code' keeps the original "
+            "tone/length code, while 'pinyin-initial' emits lowercase pinyin "
+            "first letters only, and 'hanzi' keeps segmented Mandarin as Hanzi."
+        ),
+    )
+    parser.add_argument(
+        "--jieba",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+        help=(
+            "Use jieba word segmentation for Chinese spans. Disable with "
+            "--no-jieba to emit one token per Hanzi character before "
+            "transliteration."
+        ),
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    require_dependencies()
+    if hasattr(sys.stdout, "reconfigure"):
+        sys.stdout.reconfigure(encoding="utf-8")
+    args = parse_args()
+    count = preprocess_file(
+        args.input,
+        args.output,
+        args.preview,
+        args.transliteration,
+        args.jieba,
+    )
+    print(f"Wrote {count:,} processed documents to {args.output}")
+
+
+if __name__ == "__main__":
+    main()

special_tokens_map.json

@@ -0,0 +1,6 @@
+{
+  "bos_token": "<s>",
+  "eos_token": "</s>",
+  "pad_token": "<pad>",
+  "unk_token": "<unk>"
+}

tokenization_pinyin_code.py

@@ -0,0 +1,548 @@
+"""SentencePiece tokenizer wrapper for pinyin-code Transformers models."""
+
+from __future__ import annotations
+
+import logging
+import math
+import re
+import shutil
+import unicodedata
+from pathlib import Path
+from typing import Any
+
+import sentencepiece as spm
+from transformers import PreTrainedTokenizer
+
+
+CHINESE_RE = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]")
+CHINESE_SPAN_RE = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]+")
+PINYIN_CODE_TOKEN_RE = re.compile(
+    r"(?<![A-Za-z0-9])[A-Za-z]\d(?:[A-Za-z]\d)*(?![A-Za-z0-9])"
+)
+SPECIAL_MARKER_RE = re.compile(r"<[A-Z_]+>")
+PUNCTUATION = set(
+    "\u3002\uff0c\u3001\uff1f\uff01\uff1a\uff1b.,?!:;()[]{}<>\u300a\u300b"
+    "\u3010\u3011\u201c\u201d\"'\u2018\u2019\u300c\u300d\u300e\u300f"
+    "\u2014-~\u2026/\\"
+)
+LATIN_LETTER = (
+    r"A-Za-z\u00c0-\u00d6\u00d8-\u00f6\u00f8-\u00ff"
+    r"\u0100-\u017f\u0180-\u024f\u0250-\u02af"
+)
+LATIN_ALNUM_PATTERN = (
+    rf"(?:[{LATIN_LETTER}][{LATIN_LETTER}0-9]*"
+    rf"(?:[-_][{LATIN_LETTER}0-9]+)*|"
+    rf"[0-9]+[{LATIN_LETTER}][{LATIN_LETTER}0-9]*"
+    rf"(?:[-_][{LATIN_LETTER}0-9]+)*)"
+)
+LATIN_ALNUM_RE = re.compile(LATIN_ALNUM_PATTERN)
+URL_RE = re.compile(r"\b(?:https?://\S*|www\.\S+)", flags=re.I)
+DISCARDED_UNICODE_CATEGORIES = {"Cc", "Cf", "Co", "Cs", "Cn"}
+TOKEN_RE = re.compile(
+    r"<[A-Z_]+>|"
+    r"[\u3400-\u4dbf\u4e00-\u9fff]+|"
+    rf"{LATIN_ALNUM_PATTERN}|"
+    r"\S"
+)
+LABELS = {
+    "\u9898\u5e72": "<QUESTION>",
+    "\u9009\u9879": "<OPTIONS>",
+    "\u7b54\u6848": "<ANSWER>",
+    "\u89e3\u6790": "<EXPLANATION>",
+}
+PINYIN_FORMAT_ALIASES = {
+    "code": "pinyin-code",
+    "codes": "pinyin-code",
+    "pinyin-code": "pinyin-code",
+    "initial": "pinyin-initial",
+    "initials": "pinyin-initial",
+    "pinyin-initial": "pinyin-initial",
+    "hanzi": "hanzi",
+}
+
+
+def latin_token_to_model_token(token: str) -> str:
+    upper = token.upper()
+    return upper if upper in {"A", "B", "C", "D"} else token.lower()
+
+
+def should_preserve_fallback_token(token: str) -> bool:
+    if token == "\ufffd":
+        return False
+    for char in token:
+        category = unicodedata.category(char)
+        if category in DISCARDED_UNICODE_CATEGORIES:
+            return False
+        if category[0] not in {"L", "P", "S"}:
+            return False
+    return True
+
+
+class PinyinCodeTokenizer(PreTrainedTokenizer):
+    """Slow tokenizer that preserves the existing SentencePiece model."""
+
+    vocab_files_names = {"vocab_file": "tokenizer.model"}
+    model_input_names = ["input_ids", "attention_mask"]
+
+    def __init__(
+        self,
+        vocab_file: str,
+        add_bos_token: bool = False,
+        add_eos_token: bool = False,
+        transliteration: str = "pinyin-code",
+        pinyin_format: str | None = None,
+        use_jieba: bool = True,
+        jieba: bool | None = None,
+        **kwargs,
+    ) -> None:
+        self.vocab_file = vocab_file
+        self.sp_model = spm.SentencePieceProcessor(model_file=vocab_file)
+        self.add_bos_token = add_bos_token
+        self.add_eos_token = add_eos_token
+        self.transliteration = self._normalize_transliteration(
+            pinyin_format or transliteration
+        )
+        self.use_jieba = use_jieba if jieba is None else jieba
+
+        kwargs.setdefault("unk_token", self._piece_or_none(self.sp_model.unk_id()))
+        kwargs.setdefault("bos_token", self._piece_or_none(self.sp_model.bos_id()))
+        kwargs.setdefault("eos_token", self._piece_or_none(self.sp_model.eos_id()))
+        kwargs.setdefault("pad_token", self._piece_or_none(self.sp_model.pad_id()))
+        kwargs.setdefault("transliteration", self.transliteration)
+        kwargs.setdefault("pinyin_format", self.transliteration)
+        kwargs.setdefault("use_jieba", self.use_jieba)
+        kwargs.setdefault("jieba", self.use_jieba)
+        super().__init__(**kwargs)
+
+    def _normalize_transliteration(self, value: str) -> str:
+        normalized = PINYIN_FORMAT_ALIASES.get(value.lower())
+        if normalized is None:
+            allowed = ", ".join(sorted(set(PINYIN_FORMAT_ALIASES.values())))
+            raise ValueError(f"Unsupported transliteration {value!r}; choose from {allowed}")
+        return normalized
+
+    def _piece_or_none(self, token_id: int) -> str | None:
+        if token_id is None or token_id < 0:
+            return None
+        return self.sp_model.id_to_piece(token_id)
+
+    def _looks_preprocessed(self, text: str) -> bool:
+        if SPECIAL_MARKER_RE.search(text):
+            return True
+        if self.transliteration == "pinyin-code" and PINYIN_CODE_TOKEN_RE.search(text):
+            return True
+        return False
+
+    def _preprocess_raw_text(self, text: str) -> str:
+        if not CHINESE_RE.search(text) and self._looks_preprocessed(text):
+            return text
+        try:
+            from preprocessing.preprocess import (
+                hanzi_to_encoded,
+                process_text,
+                require_dependencies,
+            )
+        except ImportError:
+            return self._fallback_process_text(text)
+
+        require_dependencies()
+        if self.transliteration == "pinyin-code":
+            return hanzi_to_encoded(text, self.use_jieba)
+        return process_text(text, self.transliteration, self.use_jieba)
+
+    def _fallback_process_text(self, text: str) -> str:
+        if self.use_jieba:
+            try:
+                import jieba
+            except ImportError as exc:
+                raise ImportError(
+                    "Tokenizing raw Mandarin benchmark text with jieba segmentation "
+                    "requires jieba. Install the model dependencies before running "
+                    "lm_eval."
+                ) from exc
+
+            jieba.setLogLevel(logging.WARNING)
+        else:
+            jieba = None
+
+        if self.transliteration != "hanzi":
+            try:
+                from pypinyin import Style, pinyin
+            except ImportError as exc:
+                raise ImportError(
+                    "Tokenizing raw Mandarin benchmark text as pinyin requires pypinyin. "
+                    "Install the model dependencies before running lm_eval."
+                ) from exc
+
+
+        def normalize_text(value: str) -> str:
+            value = unicodedata.normalize("NFKC", value)
+            value = URL_RE.sub(" <URL> ", value)
+            value = re.sub(r"\$\$.*?\$\$", " <MATH> ", value, flags=re.DOTALL)
+            value = re.sub(r"[\uff08(]\s*[\uff09)]", " <BLANK> ", value)
+            for label, marker in LABELS.items():
+                value = re.sub(rf"{label}\s*[:\uff1a]", f" {marker} ", value)
+            value = re.sub(
+                rf"(?<![{LATIN_LETTER}])yes(?![{LATIN_LETTER}])",
+                " <YES> ",
+                value,
+                flags=re.I,
+            )
+            value = re.sub(
+                rf"(?<![{LATIN_LETTER}])no(?![{LATIN_LETTER}])",
+                " <NO> ",
+                value,
+                flags=re.I,
+            )
+            value = re.sub(
+                rf"(?<![{LATIN_LETTER}])[ABCD](?=\s*[:\uff1a.\uff0e\u3001\)])",
+                r" \g<0> ",
+                value,
+            )
+            value = re.sub(
+                rf"(?<![{LATIN_LETTER}0-9])[-+]?\d+(?:[.,]\d+)*(?:%|\uff05)?"
+                rf"(?![{LATIN_LETTER}0-9])",
+                " <NUM> ",
+                value,
+            )
+            value = value.replace("\uff08", "(").replace("\uff09", ")")
+            return re.sub(r"\s+", " ", value).strip()
+
+        def split_tone3_syllable(syllable: str) -> tuple[str, int]:
+            match = re.fullmatch(r"([a-z\u00fcv]+)([1-5]?)", syllable.lower())
+            if not match:
+                return syllable, 5
+            plain, tone = match.groups()
+            return plain, int(tone or "5")
+
+        def length_digit_offset(syllable: str) -> int:
+            return min(max(len(syllable), 1), 5) - 1
+
+        def syllable_to_initial_code(syllable: str) -> str:
+            plain, tone = split_tone3_syllable(syllable)
+            if not plain:
+                return ""
+            tone_offset = 5 if tone in {3, 4, 5} else 0
+            digit = tone_offset + length_digit_offset(plain)
+            initial = plain[0].upper() if tone in {1, 3, 5} else plain[0].lower()
+            return f"{initial}{digit}"
+
+        def syllable_to_initial_letter(syllable: str) -> str:
+            plain, _ = split_tone3_syllable(syllable)
+            return plain[:1].lower()
+
+        def convert_word(word: str) -> str:
+            if self.transliteration == "hanzi":
+                return word
+            syllables = pinyin(word, style=Style.TONE3, heteronym=False, errors="ignore")
+            if self.transliteration == "pinyin-code":
+                codes = [
+                    syllable_to_initial_code(item[0])
+                    for item in syllables
+                    if item and item[0]
+                ]
+                return "".join(code for code in codes if code)
+            initials = [
+                syllable_to_initial_letter(item[0])
+                for item in syllables
+                if item and item[0]
+            ]
+            return "".join(initial for initial in initials if initial)
+
+        def tokenize_chinese_span(value: str) -> list[str]:
+            tokens = []
+            words = jieba.cut(value, cut_all=False) if self.use_jieba else value
+            for word in words:
+                word = word.strip()
+                if word and CHINESE_SPAN_RE.search(word):
+                    token = convert_word(word)
+                    if token:
+                        tokens.append(token)
+            return tokens
+
+        tokens = []
+        for part in TOKEN_RE.findall(normalize_text(text)):
+            if part.startswith("<") and part.endswith(">"):
+                tokens.append(part)
+            elif CHINESE_SPAN_RE.fullmatch(part):
+                tokens.extend(tokenize_chinese_span(part))
+            elif part in PUNCTUATION:
+                tokens.append(part)
+            elif LATIN_ALNUM_RE.fullmatch(part):
+                tokens.append(latin_token_to_model_token(part))
+            elif part.isdigit():
+                tokens.append("<NUM>")
+            elif should_preserve_fallback_token(part):
+                tokens.append(part.lower())
+
+        return " ".join(tokens)
+
+    def _preprocess_tokenizer_input(self, value: Any) -> Any:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return self._preprocess_raw_text(value)
+        if isinstance(value, tuple):
+            return tuple(self._preprocess_tokenizer_input(item) for item in value)
+        if isinstance(value, list):
+            return [self._preprocess_tokenizer_input(item) for item in value]
+        return value
+
+    def _non_content_token_ids(self) -> set[int]:
+        return {
+            token_id
+            for token_id in (
+                self.pad_token_id,
+                self.bos_token_id,
+                self.eos_token_id,
+                self.cls_token_id,
+                self.sep_token_id,
+                self.mask_token_id,
+            )
+            if token_id is not None
+        }
+
+    def _offset_source_text(self, value: Any, is_split_into_words: bool = False) -> str:
+        if value is None:
+            return ""
+        if isinstance(value, str):
+            return value
+        if isinstance(value, tuple):
+            return " ".join(self._offset_source_text(item) for item in value)
+        if isinstance(value, list):
+            separator = " " if is_split_into_words else ""
+            return separator.join(self._offset_source_text(item) for item in value)
+        return str(value)
+
+    def _synthetic_offset_mapping(self, text: Any, input_ids: Any, is_split_into_words: bool = False) -> list[tuple[int, int]]:
+        """Return slow-tokenizer-compatible offsets for evaluators that require them.
+
+        SentencePiece offsets are not available for this Python tokenizer because
+        raw Mandarin text is preprocessed into pinyin-code before encoding. These
+        spans conservatively distribute non-special tokens across the original
+        text so suffix/completion masking code can run without requiring a fast
+        tokenizer.
+        """
+        ids = input_ids.tolist() if hasattr(input_ids, "tolist") else list(input_ids)
+        source = self._offset_source_text(text, is_split_into_words=is_split_into_words)
+        source_length = len(source)
+        if not ids:
+            return []
+        if source_length == 0:
+            return [(0, 0) for _ in ids]
+
+        non_content_ids = self._non_content_token_ids()
+        content_positions = [
+            index for index, token_id in enumerate(ids) if int(token_id) not in non_content_ids
+        ]
+        if not content_positions:
+            return [(0, 0) for _ in ids]
+
+        offsets = [(0, 0) for _ in ids]
+        count = len(content_positions)
+        for ordinal, position in enumerate(content_positions):
+            start = math.floor(ordinal * source_length / count)
+            end = math.ceil((ordinal + 1) * source_length / count)
+            if end <= start:
+                end = min(source_length, start + 1)
+            offsets[position] = (start, end)
+        return offsets
+
+    def _with_optional_offsets(
+        self,
+        encoding,
+        original_text: Any,
+        return_offsets_mapping: bool,
+        is_split_into_words: bool = False,
+        return_tensors: str | None = None,
+    ):
+        if not return_offsets_mapping:
+            return encoding
+
+        input_ids = encoding["input_ids"]
+        tensor_input = hasattr(input_ids, "ndim")
+        input_ids_list = input_ids.tolist() if tensor_input else input_ids
+
+        is_batched = False
+        if tensor_input:
+            is_batched = input_ids.ndim > 1
+        elif input_ids_list and isinstance(input_ids_list[0], list):
+            is_batched = True
+
+        if is_batched:
+            if isinstance(original_text, list) and not is_split_into_words:
+                texts = original_text
+            else:
+                texts = [original_text] * len(input_ids_list)
+            offsets = [
+                self._synthetic_offset_mapping(text, ids, is_split_into_words=is_split_into_words)
+                for text, ids in zip(texts, input_ids_list)
+            ]
+        else:
+            offsets = self._synthetic_offset_mapping(
+                original_text,
+                input_ids_list,
+                is_split_into_words=is_split_into_words,
+            )
+
+        if return_tensors == "pt" or tensor_input:
+            try:
+                import torch
+
+                offsets = torch.tensor(offsets, dtype=torch.long)
+            except ImportError:
+                pass
+        encoding["offset_mapping"] = offsets
+        return encoding
+
+    def __call__(self, text=None, text_pair=None, *args, **kwargs):
+        original_text = text
+        return_offsets_mapping = bool(kwargs.pop("return_offsets_mapping", False))
+        is_split_into_words = bool(kwargs.get("is_split_into_words", False))
+        return_tensors = kwargs.get("return_tensors")
+
+        if "text_target" in kwargs:
+            kwargs["text_target"] = self._preprocess_tokenizer_input(kwargs["text_target"])
+        if "text_pair_target" in kwargs:
+            kwargs["text_pair_target"] = self._preprocess_tokenizer_input(
+                kwargs["text_pair_target"]
+            )
+
+        text = self._preprocess_tokenizer_input(text)
+        text_pair = self._preprocess_tokenizer_input(text_pair)
+        if text_pair is None:
+            encoding = super().__call__(text, *args, **kwargs)
+        else:
+            encoding = super().__call__(text, text_pair, *args, **kwargs)
+        return self._with_optional_offsets(
+            encoding,
+            original_text,
+            return_offsets_mapping,
+            is_split_into_words=is_split_into_words,
+            return_tensors=return_tensors,
+        )
+
+    def encode(self, text, text_pair=None, add_special_tokens=True, *args, **kwargs):
+        kwargs["add_special_tokens"] = add_special_tokens
+        text = self._preprocess_tokenizer_input(text)
+        text_pair = self._preprocess_tokenizer_input(text_pair)
+        if text_pair is None:
+            return super().encode(text, *args, **kwargs)
+        return super().encode(text, text_pair, *args, **kwargs)
+
+    def encode_plus(self, text, text_pair=None, *args, **kwargs):
+        original_text = text
+        return_offsets_mapping = bool(kwargs.pop("return_offsets_mapping", False))
+        is_split_into_words = bool(kwargs.get("is_split_into_words", False))
+        return_tensors = kwargs.get("return_tensors")
+
+        text = self._preprocess_tokenizer_input(text)
+        text_pair = self._preprocess_tokenizer_input(text_pair)
+        if text_pair is None:
+            encoding = super().encode_plus(text, *args, **kwargs)
+        else:
+            encoding = super().encode_plus(text, text_pair, *args, **kwargs)
+        return self._with_optional_offsets(
+            encoding,
+            original_text,
+            return_offsets_mapping,
+            is_split_into_words=is_split_into_words,
+            return_tensors=return_tensors,
+        )
+
+    def batch_encode_plus(self, batch_text_or_text_pairs, *args, **kwargs):
+        original_batch = batch_text_or_text_pairs
+        return_offsets_mapping = bool(kwargs.pop("return_offsets_mapping", False))
+        is_split_into_words = bool(kwargs.get("is_split_into_words", False))
+        return_tensors = kwargs.get("return_tensors")
+
+        batch_text_or_text_pairs = self._preprocess_tokenizer_input(
+            batch_text_or_text_pairs
+        )
+        encoding = super().batch_encode_plus(batch_text_or_text_pairs, *args, **kwargs)
+        return self._with_optional_offsets(
+            encoding,
+            original_batch,
+            return_offsets_mapping,
+            is_split_into_words=is_split_into_words,
+            return_tensors=return_tensors,
+        )
+
+    @property
+    def vocab_size(self) -> int:
+        return self.sp_model.get_piece_size()
+
+    def get_vocab(self) -> dict[str, int]:
+        vocab = {self.sp_model.id_to_piece(i): i for i in range(self.vocab_size)}
+        vocab.update(self.added_tokens_encoder)
+        return vocab
+
+    def _tokenize(self, text: str) -> list[str]:
+        text = self._preprocess_raw_text(text)
+        return self.sp_model.encode(text, out_type=str)
+
+    def _convert_token_to_id(self, token: str) -> int:
+        return self.sp_model.piece_to_id(token)
+
+    def _convert_id_to_token(self, index: int) -> str:
+        return self.sp_model.id_to_piece(index)
+
+    def convert_tokens_to_string(self, tokens: list[str]) -> str:
+        return self.sp_model.decode(tokens)
+
+    def build_inputs_with_special_tokens(
+        self,
+        token_ids_0: list[int],
+        token_ids_1: list[int] | None = None,
+    ) -> list[int]:
+        output = list(token_ids_0)
+        if self.add_bos_token and self.bos_token_id is not None:
+            output = [self.bos_token_id] + output
+        if self.add_eos_token and self.eos_token_id is not None:
+            output = output + [self.eos_token_id]
+        if token_ids_1 is not None:
+            output += list(token_ids_1)
+            if self.add_eos_token and self.eos_token_id is not None:
+                output.append(self.eos_token_id)
+        return output
+
+    def get_special_tokens_mask(
+        self,
+        token_ids_0: list[int],
+        token_ids_1: list[int] | None = None,
+        already_has_special_tokens: bool = False,
+    ) -> list[int]:
+        if already_has_special_tokens:
+            special_ids = set(self.all_special_ids)
+            return [1 if token_id in special_ids else 0 for token_id in token_ids_0]
+
+        mask = [0] * len(token_ids_0)
+        if self.add_bos_token and self.bos_token_id is not None:
+            mask = [1] + mask
+        if self.add_eos_token and self.eos_token_id is not None:
+            mask = mask + [1]
+        if token_ids_1 is not None:
+            mask += [0] * len(token_ids_1)
+            if self.add_eos_token and self.eos_token_id is not None:
+                mask.append(1)
+        return mask
+
+    def create_token_type_ids_from_sequences(
+        self,
+        token_ids_0: list[int],
+        token_ids_1: list[int] | None = None,
+    ) -> list[int]:
+        return [0] * len(self.build_inputs_with_special_tokens(token_ids_0, token_ids_1))
+
+    def save_vocabulary(self, save_directory: str, filename_prefix: str | None = None) -> tuple[str]:
+        output_name = "tokenizer.model"
+        if filename_prefix:
+            output_name = f"{filename_prefix}-{output_name}"
+        output_path = Path(save_directory) / output_name
+        if Path(self.vocab_file).resolve() != output_path.resolve():
+            shutil.copyfile(self.vocab_file, output_path)
+        return (str(output_path),)
+
+
+class EncodedMandarinTokenizer(PinyinCodeTokenizer):
+    """Tokenizer wrapper that hides Hanzi-to-encoded-Mandarin preprocessing."""

tokenizer.model

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:60cf469adc184103e3c73309b971a667d4ec04211cf33527e0341716ccd3f5ee
+size 285315

tokenizer_config.json

@@ -0,0 +1,53 @@
+{
+  "added_tokens_decoder": {
+    "0": {
+      "content": "<pad>",
+      "lstrip": false,
+      "normalized": false,
+      "rstrip": false,
+      "single_word": false,
+      "special": true
+    },
+    "1": {
+      "content": "<unk>",
+      "lstrip": false,
+      "normalized": false,
+      "rstrip": false,
+      "single_word": false,
+      "special": true
+    },
+    "2": {
+      "content": "<s>",
+      "lstrip": false,
+      "normalized": false,
+      "rstrip": false,
+      "single_word": false,
+      "special": true
+    },
+    "3": {
+      "content": "</s>",
+      "lstrip": false,
+      "normalized": false,
+      "rstrip": false,
+      "single_word": false,
+      "special": true
+    }
+  },
+  "auto_map": {
+    "AutoTokenizer": [
+      "tokenization_pinyin_code.EncodedMandarinTokenizer",
+      null
+    ]
+  },
+  "backend": "custom",
+  "bos_token": "<s>",
+  "eos_token": "</s>",
+  "jieba": true,
+  "model_max_length": 512,
+  "pad_token": "<pad>",
+  "pinyin_format": "pinyin-code",
+  "tokenizer_class": "EncodedMandarinTokenizer",
+  "transliteration": "pinyin-code",
+  "unk_token": "<unk>",
+  "use_jieba": true
+}

training_metadata.json

@@ -0,0 +1,10 @@
+{
+  "epoch": 1,
+  "evaluation_backend": "causal",
+  "global_step": 119,
+  "jieba": true,
+  "source_checkpoint": "models\\full_chinese_gpu3.2-dpo\\best.pt",
+  "transliteration": "pinyin-code",
+  "use_jieba": true,
+  "validation_loss": 0.016063788817564033
+}