#!/usr/bin/env python3 """Pomiar samych tokenizerów z publicznych, przypiętych plików Hugging Face. Instalacja: python -m pip install -r requirements.txt Uruchomienie: python mierz_tokenizery.py Powtórzenie bez sieci: python mierz_tokenizery.py --offline --output wynik2.json Licencja: CC BY 4.0, Robert Marczyński, ekspertodsztucznejinteligencji.pl. """ import argparse import hashlib import importlib.metadata import json import os from pathlib import Path import platform import re import sys import unicodedata KATALOG = Path(__file__).resolve().parent def sha256(data): return hashlib.sha256(data).hexdigest() def sprawdz_sha(data, expected, name): actual = sha256(data) if actual != expected: raise ValueError(f"Niezgodne SHA-256 {name}: {actual}, oczekiwano {expected}") return actual def struktura_prawa(text): """Sprawdza artykuły, numerowane ustępy i litery punktów bez zmiany tekstu.""" parts = re.split(r"(?m)^(?:Artykuł|Article|Artikel)\s+(\d+)\s*$", text) articles = [] for index in range(1, len(parts), 2): number, body = int(parts[index]), parts[index + 1] paragraphs = re.findall(r"(?m)^(?:([0-9]+)\.|\(([0-9]+)\))\s", body) paragraph_numbers = [int(a or b) for a, b in paragraphs] items = re.findall(r"(?m)^\(?([a-z])\)\s*$", body) articles.append({ "artykul": number, "numery_ustepow": paragraph_numbers, "ustepy_nienumerowane": 1 if number == 4 and not paragraph_numbers else 0, "punkty": items, }) expected = [ {"artykul": 4, "numery_ustepow": [], "ustepy_nienumerowane": 1, "punkty": []}, {"artykul": 50, "numery_ustepow": list(range(1, 8)), "ustepy_nienumerowane": 0, "punkty": []}, {"artykul": 73, "numery_ustepow": list(range(1, 12)), "ustepy_nienumerowane": 0, "punkty": []}, {"artykul": 33, "numery_ustepow": list(range(1, 6)), "ustepy_nienumerowane": 0, "punkty": list("abcd")}, {"artykul": 34, "numery_ustepow": list(range(1, 5)), "ustepy_nienumerowane": 0, "punkty": list("abc")}, ] if articles != expected: raise ValueError(f"Nieoczekiwana struktura korpusu prawnego: {articles}") return articles def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", type=Path, default=KATALOG / "wyniki.json") parser.add_argument("--cache-dir", type=Path, help="Cache plików HF; domyślnie HF_HOME/hub.") parser.add_argument("--offline", action="store_true", help="Użyj wyłącznie już pobranych plików.") args = parser.parse_args() # Ustawione przed importem HF. token=False wyklucza także token zapisany na komputerze. os.environ["HF_HUB_DISABLE_IMPLICIT_TOKEN"] = "1" os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1" os.environ["HF_HUB_DISABLE_XET"] = "1" os.environ["TOKENIZERS_PARALLELISM"] = "false" if args.offline: os.environ["HF_HUB_OFFLINE"] = "1" from huggingface_hub import hf_hub_download from tokenizers import Tokenizer manifest_bytes = (KATALOG / "manifest.json").read_bytes() manifest = json.loads(manifest_bytes) texts, corpus = {}, [] for entry in manifest["korpus"]: filename = entry["plik"] raw = (KATALOG / "korpus" / filename).read_bytes() sprawdz_sha(raw, entry["sha256"], filename) original = raw.decode("utf-8", errors="strict") text = unicodedata.normalize("NFC", original) texts[filename] = text info = { **entry, "bajty": len(raw), "znaki": len(text), "sha256_utf8_po_nfc": sha256(text.encode("utf-8")), "nfc_zmienilo_tekst": text != original, } if filename.startswith("prawo-"): info["struktura"] = struktura_prawa(text) corpus.append(info) results = [] for model in manifest["modele"]: repo, revision = model["repozytorium"], model["rewizja"] if not re.fullmatch(r"[0-9a-f]{40}", revision): raise ValueError(f"Wymagany pełny SHA commita: {repo}: {revision}") if model["gated"] is not False: raise ValueError(f"Repozytorium za bramką w liście pomiarowej: {repo}") print(f"{repo}@{revision}", file=sys.stderr, flush=True) path = hf_hub_download( repo_id=repo, filename="tokenizer.json", revision=revision, token=False, local_files_only=args.offline, cache_dir=args.cache_dir, ) raw = Path(path).read_bytes() sprawdz_sha(raw, model["tokenizer_sha256"], f"{repo}/tokenizer.json") configuration = json.loads(raw) if configuration.get("model", {}).get("dropout") not in (None, 0): raise ValueError(f"Losowy dropout w tokenizerze: {repo}") tokenizer = Tokenizer.from_file(str(path)) tokenizer.no_padding() tokenizer.no_truncation() counts = {} for filename, text in texts.items(): encoded = tokenizer.encode(text, add_special_tokens=False) if encoded.overflowing or not all(encoded.attention_mask): raise ValueError(f"Obcięcie lub padding: {repo}, {filename}") tokens, characters = len(encoded.ids), len(text) counts[filename] = { "znaki": characters, "tokeny": tokens, "tokeny_na_1000_znakow": 1000 * tokens / characters, } pl, en, de = (counts[f"prawo-{lang}.txt"]["tokeny"] for lang in ("PL", "EN", "DE")) results.append({ **model, "vocab_size": tokenizer.get_vocab_size(with_added_tokens=True), "vocab_size_bez_dodanych": tokenizer.get_vocab_size(with_added_tokens=False), "pliki": counts, "stosunki_tokenow_prawo": {"PL_EN": pl / en, "DE_EN": de / en, "PL_DE": pl / de}, }) output = { "wersja_formatu": 1, "data_zamrozenia_zbioru": manifest["data_zamrozenia_zbioru"], "manifest_sha256": sha256(manifest_bytes), "srodowisko": { "python": platform.python_version(), "tokenizers": importlib.metadata.version("tokenizers"), "huggingface_hub": importlib.metadata.version("huggingface-hub"), "unicode": unicodedata.unidata_version, }, "metoda": { "wejscie": "UTF-8, strict, zachowane białe znaki i końcowy LF", "normalizacja": "NFC przed liczeniem znaków i tokenów", "znaki": "len(str) po NFC, punkty kodowe Unicode", "tokenizacja": "tokenizers.Tokenizer.from_file(tokenizer.json)", "add_special_tokens": False, "szablon_czatu": False, "padding": False, "truncation": False, "normalizer_tokenizera": "Zachowany z tokenizer.json, działa po wejściowej normalizacji NFC", "vocab_size": "get_vocab_size(with_added_tokens=True); osobno bez dodanych tokenów", "tokeny_na_1000_znakow": "1000 * tokeny / znaki", "stosunki": "Ilorazy całkowitych liczb tokenów plików prawa, nie gęstości tokenizacji", "daty_pobrania": "Daty oryginalnego pobrania w manifest.json, nie data tego uruchomienia", }, "korpus": corpus, "modele": results, "nie_zmierzono": manifest["nie_zmierzono"], } # Zapis dopiero po udanym pomiarze całości. Bez zegara, ścieżek cache i czasu wykonania. args.output.write_bytes((json.dumps(output, ensure_ascii=False, indent=2) + "\n").encode("utf-8")) print(f"Zapisano {args.output}; SHA-256 {sha256(args.output.read_bytes())}", file=sys.stderr) if __name__ == "__main__": try: main() except Exception as error: print(f"Błąd: {error}", file=sys.stderr) sys.exit(1)