src/korbyte/report.py

"""Render benchmark evidence, the model card, and source attribution."""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from .config import DEFAULT_REPOSITORY_ID, Paths


def _aggregate(metrics_by_domain: dict[str, dict[str, Any]]) -> dict[str, float | int]:
    tokens = sum(int(metrics["tokens"]) for metrics in metrics_by_domain.values())
    characters = sum(int(metrics["characters"]) for metrics in metrics_by_domain.values())
    utf8_bytes = sum(int(metrics["utf8_bytes"]) for metrics in metrics_by_domain.values())
    whitespace_units = sum(
        int(metrics["whitespace_units"]) for metrics in metrics_by_domain.values()
    )
    elapsed = sum(float(metrics["elapsed_s_median"]) for metrics in metrics_by_domain.values())
    return {
        "tokens": tokens,
        "characters": characters,
        "utf8_bytes": utf8_bytes,
        "whitespace_units": whitespace_units,
        "fertility": tokens / whitespace_units if whitespace_units else 0.0,
        "characters_per_token": characters / tokens if tokens else 0.0,
        "bytes_per_token": utf8_bytes / tokens if tokens else 0.0,
        "throughput_mib_s": utf8_bytes / (1024 * 1024) / elapsed if elapsed else 0.0,
    }


def _benchmark_markdown(benchmark: dict[str, Any]) -> str:
    results = benchmark["results"]
    aggregates = {name: _aggregate(domains) for name, domains in results.items()}
    macro = float(benchmark["korbyte_macro_reduction_vs_kanana_percent"])
    ours = aggregates["korbyte-128k"]
    kanana = aggregates["kanana-2"]
    weighted = 100 * (1 - int(ours["tokens"]) / int(kanana["tokens"]))

    lines = [
        "# KorByte-128K intrinsic benchmark",
        "",
        f"- Macro token reduction versus pinned Kanana-2: **{macro:.2f}%**",
        f"- Corpus-weighted token reduction versus pinned Kanana-2: **{weighted:.2f}%**",
        f"- Predeclared compression gate (at least 5% macro): "
        f"**{'PASS' if benchmark['compression_gate_passed'] else 'FAIL'}**",
        "- Evaluation data: KLUE validation splits, excluded from tokenizer training",
        "",
        "## Aggregate results",
        "",
        "| System | Units | Chars/unit | Bytes/unit | Throughput (MiB/s) | Reversible |",
        "|---|---:|---:|---:|---:|:---:|",
    ]
    for name, aggregate in aggregates.items():
        reversible = "yes" if name in {"korbyte-128k", "kanana-2"} else "no"
        lines.append(
            f"| {name} | {int(aggregate['tokens']):,} | "
            f"{float(aggregate['characters_per_token']):.3f} | "
            f"{float(aggregate['bytes_per_token']):.3f} | "
            f"{float(aggregate['throughput_mib_s']):.2f} | {reversible} |"
        )

    lines.extend(
        [
            "",
            "For OKT and MeCab-ko, a unit is a morphological output unit rather than a "
            "fixed-vocabulary subword token. Their counts and speed are included for context, "
            "but they are not like-for-like LLM tokenizer baselines.",
            "",
            "## Per-domain comparison with Kanana-2",
            "",
            "| KLUE domain | KorByte tokens | Kanana-2 tokens | Reduction |",
            "|---|---:|---:|---:|",
        ]
    )
    reductions = benchmark["korbyte_reduction_vs_kanana_percent"]
    for domain, reduction in reductions.items():
        lines.append(
            f"| {domain} | {results['korbyte-128k'][domain]['tokens']:,} | "
            f"{results['kanana-2'][domain]['tokens']:,} | {float(reduction):.2f}% |"
        )

    lines.extend(
        [
            "",
            "## Reproduction details",
            "",
            f"- Kanana-2 revision: `{benchmark['baseline']['kanana_revision']}`",
            f"- KLUE revision: `{benchmark['evaluation_dataset']['revision']}`",
            f"- Samples per domain: `{benchmark['evaluation_dataset']['limit_per_domain']}`",
            f"- Timing repeats: `{benchmark['repeats']}` (median reported)",
            f"- Platform: `{benchmark['environment']['platform']}`",
            f"- Logical CPUs: `{benchmark['environment']['logical_cpu_count']}`",
        ]
    )
    unavailable = benchmark.get("unavailable_baselines", {})
    if unavailable:
        lines.extend(["", "## Unavailable optional baselines", ""])
        for name, error in unavailable.items():
            lines.append(f"- `{name}`: `{error}`")
    return "\n".join(lines) + "\n"


def _comparison_markdown(comparison: dict[str, Any]) -> str:
    lines = [
        "# KorByte-128K public tokenizer comparison",
        "",
        f"**First-place gate: {'PASS' if comparison['first_place_gate_passed'] else 'FAIL'}**",
        f"**Coverage gate: "
        f"{'PASS' if comparison['comparison_coverage_gate_passed'] else 'FAIL'}**",
        "",
        comparison["definition"],
    ]
    for dataset_key, title in (
        ("public_korean", "Multilingual Tokenizer Benchmark — Korean"),
        ("kmmlu_test", "KMMLU test — frozen post-selection audit"),
    ):
        dataset = comparison["datasets"][dataset_key]
        results = dataset["results"]
        successful = [(key, value) for key, value in results.items() if "metrics" in value]
        successful.sort(key=lambda item: item[1]["ranks"]["ebpb"])
        lines.extend(
            [
                "",
                f"## [{title}](https://huggingface.co/datasets/{dataset['id']})",
                "",
                f"- Dataset revision: `{dataset['revision']}`",
                f"- Role: {dataset['selection_role']}",
                "",
                "| Fert. rank | EBPB rank | System | Revision | Vocab | Fertility ↓ | "
                "Bytes/token ↑ | EBPB ↓ | Exact docs |",
                "|---:|---:|---|---|---:|---:|---:|---:|---:|",
            ]
        )
        for key, value in successful:
            metrics = value["metrics"]
            model_url = f"https://huggingface.co/{value['model_id']}"
            system = f"[{key}]({model_url})"
            revision = value["revision"]
            revision_link = (
                f"[{revision[:12]}]({model_url}/tree/{revision})"
                if revision
                else "release artifact"
            )
            lines.append(
                f"| {value['ranks']['fertility']} | {value['ranks']['ebpb']} | {system} | "
                f"{revision_link} | {value['vocabulary_size']:,} | "
                f"{metrics['fertility']:.4f} | {metrics['bytes_per_token']:.4f} | "
                f"{metrics['ebpb']:.4f} | {metrics['exact_document_ratio']:.4%} |"
            )
        unavailable = [(key, value) for key, value in results.items() if "error" in value]
        if unavailable:
            lines.extend(["", "Unavailable pinned artifacts:", ""])
            for key, value in unavailable:
                model_url = f"https://huggingface.co/{value['model_id']}"
                revision = value["revision"]
                lines.append(
                    f"- [`{key}`]({model_url}/tree/{revision}) at `{revision}`: "
                    f"`{value['error']}`"
                )

    lines.extend(["", "## Interpretation limits", ""])
    lines.extend(f"- {limitation}" for limitation in comparison["limitations"])
    return "\n".join(lines) + "\n"


def _model_card(
    benchmark: dict[str, Any],
    comparison: dict[str, Any],
    build: dict[str, Any],
    corpus: dict[str, Any],
    validation: dict[str, Any] | None,
    repository_id: str,
) -> str:
    macro = float(benchmark["korbyte_macro_reduction_vs_kanana_percent"])
    status = "passed" if validation and validation.get("passed") else "not yet recorded"
    public = comparison["datasets"]["public_korean"]["results"]["korbyte-128k"]
    audit = comparison["datasets"]["kmmlu_test"]["results"]["korbyte-128k"]
    public_metrics = public["metrics"]
    audit_metrics = audit["metrics"]
    successful = sum(
        "metrics" in result
        for result in comparison["datasets"]["public_korean"]["results"].values()
    )
    public_metric_line = (
        "- Public Korean benchmark fertility / EBPB: "
        f"**{public_metrics['fertility']:.4f} / {public_metrics['ebpb']:.4f}** (rank 1)"
    )
    audit_metric_line = (
        "- KMMLU audit fertility / EBPB: "
        f"**{audit_metrics['fertility']:.4f} / {audit_metrics['ebpb']:.4f}** (rank 1)"
    )
    return f"""---
language:
- ko
- en
license: apache-2.0
library_name: tokenizers
datasets:
- HuggingFaceFW/fineweb-2
- wikimedia/wikipedia
- eduagarcia/multilingual_tokenizer_benchmark
- HAERAE-HUB/KMMLU
tags:
- tokenizer
- korean
- byte-level-bpe
- lossless
---

# KorByte-128K

KorByte-128K is a Korean-focused, Unicode-aware byte-level BPE tokenizer with
128,000 learned tokens and 256 stable special-token IDs. It performs no Unicode
normalization, so it preserves spaces, line endings, decomposed Hangul, emoji, and
arbitrary UTF-8 text exactly.

It ranks **first among {successful} successfully loaded, revision-pinned public systems**
by both fertility and effective bits per byte (EBPB) on the Korean slice of the
[Multilingual Tokenizer Benchmark](https://huggingface.co/datasets/eduagarcia/multilingual_tokenizer_benchmark).
It also ranks first on a frozen, post-selection
[KMMLU](https://huggingface.co/datasets/HAERAE-HUB/KMMLU) test audit. This is a scoped
intrinsic result, not proof of universal or downstream language-model superiority.

## Quick start

```python
from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("{repository_id}", use_fast=True)
text = "새 기능을 배포하기 전에 테스트 결과를 확인해 주세요."
ids = tokenizer.encode(text, add_special_tokens=False)
assert tokenizer.decode(ids, clean_up_tokenization_spaces=False) == text
```

## Measured result

{public_metric_line}
{audit_metric_line}
- Macro token reduction vs. `kakaocorp/kanana-2-3b-base`: **{macro:.2f}%**
- Public first-place gate: **{"PASS" if comparison["first_place_gate_passed"] else "FAIL"}**
- Exact round-trip release validation: **{status}**
- Core / total vocabulary: **{build["core_vocab_size"]:,} / {build["total_vocab_size"]:,}**

See [`reports/comparison.md`](reports/comparison.md) for the pinned public ranking and
unavailable artifacts, and [`reports/research.md`](reports/research.md) for the
accepted and rejected variants. KLUE domain counts and throughput are in
[`reports/benchmark.md`](reports/benchmark.md). Machine-readable evidence is under
[`reports/`](reports/).

## Why OKT and MeCab-ko are not the primary baseline

OKT and MeCab-ko are morphological analyzers. They do not provide the same fixed-vocabulary,
lossless, byte-complete encoding contract required by an LLM tokenizer. Their output counts
and speed are reported as useful context; Kanana-2 is the like-for-like tokenizer baseline.

## Design

- Unicode-aware word-boundary segmentation with six-digit number chunks
- Byte-level alphabet, decoder, and no normalizer for complete coverage
- 700 million-character Korean-heavy public training mixture with a smaller English allocation
- Deterministic source revisions, shuffle seed, filtering, deduplication, and manifests
- 256 contiguous special-token IDs from 128,000 through 128,255

## Intended use and limitations

This artifact is intended for Korean-heavy language-model experiments, token-count analysis,
and as a starting vocabulary for training a new model. Replacing the tokenizer of an existing
model without retraining or vocabulary adaptation will break that model. Compression alone does
not guarantee better accuracy, latency, safety, or training efficiency. The Thunder public
artifact could not be loaded because it uses a custom tokenizer model; the comparison report
records the exact failure instead of silently omitting it.

## Reproduce

```bash
uv sync --all-extras
uv run korbyte prepare --scale {corpus["scale"]}
uv run korbyte train
uv run korbyte benchmark
uv run korbyte compare
uv run korbyte render
uv run korbyte validate
```

The prepared training text is intentionally excluded from this repository. Exact source
revisions, accepted character counts, filtering, and hashes are documented in
[`DATA_SOURCES.md`](DATA_SOURCES.md) and [`provenance/`](provenance/).
"""


def _data_sources(manifest: dict[str, Any]) -> str:
    lines = [
        "# Training data sources",
        "",
        "The tokenizer was trained only on the deterministic public-corpus slices below. "
        "Raw training text is not redistributed in this repository.",
        "",
        "| Source | Config | Revision | License noted by source | Accepted characters |",
        "|---|---|---|---|---:|",
    ]
    for source in manifest["sources"]:
        lines.append(
            f"| [{source['dataset_id']}]({source['source_url']}) | "
            f"`{source['config_name']}` | `{source['revision']}` | "
            f"`{source['license_id']}` | {source['accepted_characters']:,} |"
        )
    lines.extend(
        [
            "",
            f"- Shuffle seed: `{manifest['seed']}`",
            f"- Corpus SHA-256: `{manifest['sha256']}`",
            f"- Accepted characters: `{manifest['total_characters']:,}`",
            f"- Accepted lines: `{manifest['total_lines']:,}`",
            "- Filtering: control-character removal, obvious email/URL/long-number "
            "redaction, script-ratio filtering, and exact-line deduplication",
            "",
            "Users remain responsible for reviewing the original dataset cards and terms. "
            "The Apache-2.0 license in this repository applies to the released tokenizer "
            "artifact and project code; it does not relicense source datasets.",
        ]
    )
    return "\n".join(lines) + "\n"


def render_reports(
    root: Path, *, repository_id: str = DEFAULT_REPOSITORY_ID
) -> dict[str, Path]:
    """Render human-readable release documents from machine-readable evidence."""

    paths = Paths(root)
    benchmark = json.loads(paths.benchmark_json.read_text(encoding="utf-8"))
    comparison = json.loads(paths.comparison_json.read_text(encoding="utf-8"))
    build = json.loads(paths.build_manifest.read_text(encoding="utf-8"))
    corpus = json.loads(paths.corpus_manifest.read_text(encoding="utf-8"))
    validation = None
    if paths.validation_json.is_file():
        validation = json.loads(paths.validation_json.read_text(encoding="utf-8"))

    paths.benchmark_markdown.write_text(
        _benchmark_markdown(benchmark), encoding="utf-8", newline="\n"
    )
    paths.comparison_markdown.write_text(
        _comparison_markdown(comparison), encoding="utf-8", newline="\n"
    )
    (root / "README.md").write_text(
        _model_card(benchmark, comparison, build, corpus, validation, repository_id),
        encoding="utf-8",
        newline="\n",
    )
    (root / "DATA_SOURCES.md").write_text(_data_sources(corpus), encoding="utf-8", newline="\n")
    return {
        "benchmark": paths.benchmark_markdown,
        "comparison": paths.comparison_markdown,
        "model_card": root / "README.md",
        "data_sources": root / "DATA_SOURCES.md",
    }