diff --git a/CHANGELOG.md b/CHANGELOG.md index 1cc80cacb..4579c0cd1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,8 @@ and this project adheres to ([#1499], [#1509]) - `evaluate_word_tokenization`, `word_eval_function`, `char_eval_function`, and `TokenizationScore` in `pythainlp.benchmarks` ([#1512]) +- PhayaThaiBERT POS tagger: `pos_tag()` engine `phayathaibert` (ONNX) and + `pos_tag_transformers()` corpus `tud` ([#1550]) - Laya Multilingual model (`LayaModel`, `Laya`) to `pythainlp.classify` for zero-shot classification ([#1528]) - `CRFTagger`, CRF weight conversion, and bundled SSG weights ([#1538]) @@ -126,6 +128,7 @@ and this project adheres to [#1546]: https://github.com/PyThaiNLP/pythainlp/pull/1546 [#1547]: https://github.com/PyThaiNLP/pythainlp/pull/1547 [#1548]: https://github.com/PyThaiNLP/pythainlp/pull/1548 +[#1550]: https://github.com/PyThaiNLP/pythainlp/pull/1550 [#1551]: https://github.com/PyThaiNLP/pythainlp/pull/1551 [#1552]: https://github.com/PyThaiNLP/pythainlp/pull/1552 [#1565]: https://github.com/PyThaiNLP/pythainlp/pull/1565 diff --git a/docs/api/tag.rst b/docs/api/tag.rst index ae5f74972..3b86330f4 100644 --- a/docs/api/tag.rst +++ b/docs/api/tag.rst @@ -228,6 +228,7 @@ Modules .. autofunction:: pos_tag .. autofunction:: pos_tag_sents +.. autofunction:: pos_tag_transformers .. autofunction:: tag_provinces .. autofunction:: chunk_parse .. autoclass:: CRFTagger @@ -253,6 +254,23 @@ unigram Unigram tagger doesn't take the ordering of words in the list into account. +phayathaibert ++++++++++++++ + +PhayaThaiBERT tagger [#Sriwirote_2025]_ uses +`nlp-chula/phayathaibert-thai-pos-tagger `_, +fine-tuned on the TUD treebank, and tags words with Universal POS tags. + +- :func:`pos_tag` with ``engine="phayathaibert"`` runs an ONNX export of the + model with ONNX Runtime and does not need PyTorch. + Install the dependencies with ``pip install "pythainlp[phayathaibert_onnx]"``. + The model (about 530 MB) is downloaded from the Hugging Face Hub on first use. +- :func:`pos_tag_transformers` with ``engine="phayathaibert"`` and + ``corpus="tud"`` runs the original model with Transformers. + +.. autoclass:: pythainlp.tag.phayathaibert_onnx.PhayaThaiBERTTagger + :members: tag + References ---------- @@ -263,3 +281,7 @@ References .. [#Prachya_2020] Prachya Boonkwan and Vorapon Luantangsrisuk and Sitthaa Phaholphinyo and Kanyanat Kriengket and Dhanon Leenoi and Charun Phrombut and Monthika Boriboon and Krit Kosawat and Thepchai Supnithi. (2020). The Annotation Guideline of LST20 Corpus. arXiv:2008.05055 +.. [#Sriwirote_2025] Panyut Sriwirote, Attapol T. Rutherford, Jalinee Thapiang and Vasan Timtong. (2025). + PhayaThaiBERT: Enhancing a Pretrained Thai Language Model with Unassimilated Loanwords. + ACM Transactions on Asian and Low-Resource Language Information Processing, Vol. 24, No. 11, pp 1-17. + doi:10.1145/3765962 diff --git a/docs/notes/installation.rst b/docs/notes/installation.rst index 23e5c3dac..c8cf45e71 100644 --- a/docs/notes/installation.rst +++ b/docs/notes/installation.rst @@ -32,6 +32,7 @@ The extras can include: - ``nlpo3`` — nlpo3 Thai word tokenization support - ``onnx`` - ONNX model support - ``oskut`` — OSKUT support (via LEKCut ONNX) + - ``phayathaibert_onnx`` — PhayaThaiBERT part-of-speech tagger with ONNX Runtime - ``sefr_cut`` — SEFR CUT Thai word tokenization support (via LEKCut ONNX) - ``spacy_thai`` — spaCy Thai language support - ``spell`` — support for more spell-checkers (phunspell & symspellpy) diff --git a/pyproject.toml b/pyproject.toml index e15495f80..b0e8b479d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -149,6 +149,13 @@ onnx = ["numpy>=1.22", "onnxruntime>=1.10.0", "sentencepiece>=0.1.91"] oskut = ["lekcut>=1.0.0"] +phayathaibert_onnx = [ + "huggingface-hub>=0.16.0", + "numpy>=1.22", + "onnxruntime>=1.12.0", + "tokenizers>=0.19.0", +] + qwen3 = ["torch>=1.9.0", "transformers>=4.22.1"] sefr_cut = ["lekcut>=1.0.0"] diff --git a/pythainlp/tag/phayathaibert_onnx.py b/pythainlp/tag/phayathaibert_onnx.py new file mode 100644 index 000000000..9d88ae4e4 --- /dev/null +++ b/pythainlp/tag/phayathaibert_onnx.py @@ -0,0 +1,266 @@ +# SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project +# SPDX-FileType: SOURCE +# SPDX-License-Identifier: Apache-2.0 +"""PhayaThaiBERT part-of-speech tagger with ONNX Runtime backend.""" + +from __future__ import annotations + +import json +import threading +import unicodedata +from collections import Counter +from importlib.util import find_spec +from typing import TYPE_CHECKING, Optional + +from pythainlp.corpus import get_hf_hub + +if TYPE_CHECKING: + from collections.abc import Mapping, Sequence + + from onnxruntime import InferenceSession + from tokenizers import Encoding, Tokenizer + +_REPO_ID = "wiriyabot/phayathaibert-thai-pos-tagger-onnx" +# Pinned for reproducible and secure downloads. +_REVISION = "59e3502bd06e2222d32f11b51b4fb9569235a020" + +# Maximum number of subword tokens per model call, including and . +# The model runs on up to 510 tokens, but its accuracy drops sharply from +# position 415 on, which suggests it was pretrained on shorter sequences. +_MAX_SEQUENCE_LENGTH = 415 + +# Tag for words the model never sees (whitespace-only words), matching +# the perceptron tagger with Universal POS corpora. +_WHITESPACE_TAG = "PUNCT" + +# Universal POS tag for words that produce no subword token. +_UNKNOWN_TAG = "X" + +# Modules the tagger needs at run time. +_DEPENDENCIES = ("huggingface_hub", "numpy", "onnxruntime", "tokenizers") + + +def _is_blank(word: str) -> bool: + """ + Return True if a word has only whitespace or format characters. + + Format characters (Unicode category Cf) include the zero-width space, + zero-width non-joiner and byte order mark, which carry no text. + + :param str word: word to check + :return: whether the word is empty, whitespace or format characters + :rtype: bool + """ + return all(ch.isspace() or unicodedata.category(ch) == "Cf" for ch in word) + + +def _first_subword_labels( + word_ids: Sequence[Optional[int]], + label_ids: Sequence[int], + n_words: int, + id2label: Mapping[int, str], +) -> list[str]: + """ + Pick each word's tag from the prediction for its first subword. + + This is the alignment the model was trained with. + + :param Sequence[Optional[int]] word_ids: word index of each subword + token; ``None`` for special tokens + :param Sequence[int] label_ids: predicted label id of each subword token + :param int n_words: number of input words + :param Mapping[int, str] id2label: mapping from label id to tag name + :return: one tag per word; ``"X"`` for a word that produced no subword + :rtype: list[str] + """ + tags = [_UNKNOWN_TAG] * n_words + seen: set[int] = set() + for word_id, label_id in zip(word_ids, label_ids): + if word_id is None or word_id in seen: + continue + seen.add(word_id) + tags[word_id] = id2label[label_id] + return tags + + +def _chunk_spans( + subword_counts: Sequence[int], max_subwords: int +) -> list[tuple[int, int]]: + """ + Group consecutive words into spans that fit the model's input length. + + A word longer than ``max_subwords`` gets a span of its own; its subwords + are truncated at encoding time, which keeps its first subword. + + :param Sequence[int] subword_counts: number of subword tokens of each + word + :param int max_subwords: maximum number of subword tokens per span + :return: ``(start, end)`` word index pairs, end exclusive + :rtype: list[tuple[int, int]] + """ + spans: list[tuple[int, int]] = [] + start = 0 + total = 0 + for i, count in enumerate(subword_counts): + if i > start and total + count > max_subwords: + spans.append((start, i)) + start = i + total = 0 + total += count + if subword_counts: + spans.append((start, len(subword_counts))) + return spans + + +class PhayaThaiBERTTagger: + """ + Universal POS tagger using PhayaThaiBERT with ONNX Runtime. + + The model is an ONNX export of + `nlp-chula/phayathaibert-thai-pos-tagger + `_, + fine-tuned on the UD Thai-TUD treebank. + + Requires ``numpy``, ``onnxruntime``, ``tokenizers`` and + ``huggingface-hub``. The model (about 530 MB) is downloaded from + the Hugging Face Hub on first use. + + :param str repo_id: Hugging Face Hub repository of the ONNX model + :param Optional[str] revision: git revision of the repository. + The default repository is pinned to a commit when this is ``None``. + """ + + session: InferenceSession + tokenizer: Tokenizer + id2label: dict[int, str] + + def __init__( + self, repo_id: str = _REPO_ID, revision: Optional[str] = None + ) -> None: + """ + Download the model and load it with ONNX Runtime. + + :param str repo_id: Hugging Face Hub repository of the ONNX model + :param Optional[str] revision: git revision of the repository. + The default repository is pinned to a commit when this is + ``None``. + :raises ImportError: if a required dependency is not installed + """ + # Check every dependency before the download starts. + if any(find_spec(name) is None for name in _DEPENDENCIES): + raise ImportError( + "PhayaThaiBERT POS tagger requires numpy, onnxruntime," + " tokenizers and huggingface-hub." + ' Install them with: pip install "pythainlp[phayathaibert_onnx]"' + ) + from onnxruntime import InferenceSession + from tokenizers import Tokenizer + + if revision is None and repo_id == _REPO_ID: + revision = _REVISION + model_path = get_hf_hub(repo_id, "model.onnx", revision=revision) + tokenizer_path = get_hf_hub( + repo_id, "tokenizer.json", revision=revision + ) + config_path = get_hf_hub(repo_id, "config.json", revision=revision) + + self.session = InferenceSession( + model_path, providers=["CPUExecutionProvider"] + ) + self.tokenizer = Tokenizer.from_file(tokenizer_path) + # tokenizer.json ships with truncation at 510 tokens, which would + # silently drop words; chunking is done here instead. + self.tokenizer.no_truncation() + self.tokenizer.no_padding() + with open(config_path, encoding="utf-8") as f: + self.id2label = { + int(k): v for k, v in json.load(f)["id2label"].items() + } + + def tag(self, words: list[str]) -> list[tuple[str, str]]: + """ + Tag a list of words with Universal POS tags. + + Words with only whitespace or format characters (such as the + zero-width space) are tagged ``PUNCT`` without being passed to the + model. Input longer than the model limit is tagged in consecutive + chunks of whole words. + + :param list[str] words: list of words to be tagged + :return: list of tuples (word, POS tag) + :rtype: list[tuple[str, str]] + """ + tags = [_WHITESPACE_TAG] * len(words) + positions = [i for i, word in enumerate(words) if not _is_blank(word)] + content = [words[i] for i in positions] + if content: + for position, word_tag in zip( + positions, self._tag_content(content) + ): + tags[position] = word_tag + return list(zip(words, tags)) + + def _tag_content(self, words: list[str]) -> list[str]: + encoding = self.tokenizer.encode(words, is_pretokenized=True) + if len(encoding.ids) <= _MAX_SEQUENCE_LENGTH: + return self._predict(encoding, len(words)) + + counts = Counter(i for i in encoding.word_ids if i is not None) + subword_counts = [counts.get(i, 0) for i in range(len(words))] + labels: list[str] = [] + # Two positions are taken by and . + for start, end in _chunk_spans( + subword_counts, _MAX_SEQUENCE_LENGTH - 2 + ): + chunk = words[start:end] + chunk_encoding = self.tokenizer.encode(chunk, is_pretokenized=True) + labels.extend(self._predict(chunk_encoding, len(chunk))) + return labels + + def _predict(self, encoding: Encoding, n_words: int) -> list[str]: + import numpy as np + + ids = encoding.ids + mask = encoding.attention_mask + word_ids = encoding.word_ids + if len(ids) > _MAX_SEQUENCE_LENGTH: + # Only a single over-long word gets here; its first subword stays. + ids = ids[: _MAX_SEQUENCE_LENGTH - 1] + ids[-1:] + mask = mask[:_MAX_SEQUENCE_LENGTH] + word_ids = [*word_ids[: _MAX_SEQUENCE_LENGTH - 1], None] + logits = self.session.run( + None, + { + "input_ids": np.array([ids], dtype=np.int64), + "attention_mask": np.array([mask], dtype=np.int64), + }, + )[0][0] + return _first_subword_labels( + word_ids, logits.argmax(axis=-1).tolist(), n_words, self.id2label + ) + + +_TAGGER: Optional[PhayaThaiBERTTagger] = None +_TAGGER_LOCK = threading.Lock() + + +def tag( + words: list[str], + corpus: str = "tud", # NOSONAR: unused, kept for the pos_tag interface +) -> list[tuple[str, str]]: + """ + Tag words with Universal POS tags using PhayaThaiBERT (ONNX). + + :param list[str] words: list of words to be tagged + :param str corpus: ignored; the model is trained on UD Thai-TUD only + :return: list of tuples (word, POS tag) + :rtype: list[tuple[str, str]] + """ + global _TAGGER + if not words: + return [] + if _TAGGER is None: + with _TAGGER_LOCK: + if _TAGGER is None: + _TAGGER = PhayaThaiBERTTagger() + return _TAGGER.tag(words) diff --git a/pythainlp/tag/pos_tag.py b/pythainlp/tag/pos_tag.py index 6dd69843e..cbdb5525d 100644 --- a/pythainlp/tag/pos_tag.py +++ b/pythainlp/tag/pos_tag.py @@ -21,6 +21,11 @@ def pos_tag( * *unigram* - unigram tagger * *tltk* - TLTK: Thai Language Toolkit (supports the TNC corpus only; other corpora are converted to the TNC corpus) + * *phayathaibert* - `PhayaThaiBERT POS tagger + `_ + fine-tuned on the TUD corpus, run with ONNX Runtime (supports + the TUD corpus only; other corpora are ignored). Requires + ``pip install "pythainlp[phayathaibert_onnx]"`` :param str corpus: corpus used to train the tagger model. Options are: * *orchid* - `ORCHID @@ -125,6 +130,10 @@ def pos_tag( corpus = "tnc" elif engine == "unigram" and corpus in _support_corpus: # default from pythainlp.tag.unigram import tag as tag_ + elif engine == "phayathaibert": + from pythainlp.tag.phayathaibert_onnx import tag as tag_ + + corpus = "tud" else: raise ValueError( f"pos_tag not support {engine} engine or {corpus} corpus." @@ -150,6 +159,11 @@ def pos_tag_sents( * *unigram* - unigram tagger * *tltk* - TLTK: Thai Language Toolkit (supports the TNC corpus only; other corpora are converted to the TNC corpus) + * *phayathaibert* - `PhayaThaiBERT POS tagger + `_ + fine-tuned on the TUD corpus, run with ONNX Runtime (supports + the TUD corpus only; other corpora are ignored). Requires + ``pip install "pythainlp[phayathaibert_onnx]"`` :param str corpus: corpus used to train the tagger model. Options are: * *orchid* - `ORCHID @@ -206,6 +220,10 @@ def pos_tag_transformers( (supports the PUD corpus only) * *phayathai* - fine-tuned version of clicknext/phayathaibert on the blackboard corpus (supports the blackboard corpus only) + * *phayathaibert* - `nlp-chula/phayathaibert-thai-pos-tagger + `_, + fine-tuned version of clicknext/phayathaibert on the TUD corpus + (supports the TUD corpus only) * *mdeberta* - mDeBERTa: Multilingual Decoding-enhanced BERT with disentangled attention (supports the PUD corpus only) :param str corpus: corpus used to train the tagger model. Options are: @@ -217,6 +235,9 @@ def pos_tag_transformers( `_ treebank, natively uses Universal POS tags (supports the wangchanberta and mdeberta engines) + * *tud* - `Thai Universal Dependency Treebank (TUD) + `_, natively uses Universal + POS tags (supports the phayathaibert engine) :param Optional[str] revision: git revision (branch, tag, or commit hash) of the model. Pin to a full commit hash for secure downloads :return: list of lists of tuples (word, POS tag) @@ -248,36 +269,29 @@ def pos_tag_transformers( if not sentence: return [] - _blackboard_support_engine = { - "bert": "lunarlist/pos_thai", - "phayathai": "lunarlist/pos_thai_phayathai", + _support_engine: dict[str, dict[str, str]] = { + "blackboard": { + "bert": "lunarlist/pos_thai", + "phayathai": "lunarlist/pos_thai_phayathai", + }, + "pud": { + "mdeberta": "Pavarissy/mdeberta-v3-ud-thai-pud-upos", + "wangchanberta": "Pavarissy/wangchanberta-ud-thai-pud-upos", + }, + "tud": { + "phayathaibert": "nlp-chula/phayathaibert-thai-pos-tagger", + }, } - _pud_support_engine = { - "wangchanberta": "Pavarissy/wangchanberta-ud-thai-pud-upos", - "mdeberta": "Pavarissy/mdeberta-v3-ud-thai-pud-upos", - } - - if corpus == "blackboard" and engine in _blackboard_support_engine: - base_model = _blackboard_support_engine.get(engine) - model = AutoModelForTokenClassification.from_pretrained( - base_model, revision=revision - ) - tokenizer = AutoTokenizer.from_pretrained( - base_model, revision=revision - ) - elif corpus == "pud" and engine in _pud_support_engine: - base_model = _pud_support_engine.get(engine) - model = AutoModelForTokenClassification.from_pretrained( - base_model, revision=revision - ) - tokenizer = AutoTokenizer.from_pretrained( - base_model, revision=revision - ) - else: + base_model = _support_engine.get(corpus, {}).get(engine) + if base_model is None: raise ValueError( f"pos_tag_transformers not support {engine} engine or {corpus} corpus." ) + model = AutoModelForTokenClassification.from_pretrained( + base_model, revision=revision + ) + tokenizer = AutoTokenizer.from_pretrained(base_model, revision=revision) pipeline = TokenClassificationPipeline( model=model, diff --git a/tests/README.md b/tests/README.md index 0feae66df..fd55d1310 100644 --- a/tests/README.md +++ b/tests/README.md @@ -142,6 +142,7 @@ By separating tests by dependency group, we can: - Need dependencies from `pip install "pythainlp[noauto-onnx]"` - Tests requiring ONNX Runtime: - attacut, deepcut, oskut, sefr_cut tokenizers (via LEKCut / ONNX) + - phayathaibert POS tagger (downloads a ~530 MB model on first run) - Dependencies: ~200-500 MB - Test case class suffix: `TestCaseN` diff --git a/tests/core/test_tag.py b/tests/core/test_tag.py index dd49f9337..c34caa1e7 100644 --- a/tests/core/test_tag.py +++ b/tests/core/test_tag.py @@ -2,8 +2,13 @@ # SPDX-FileType: SOURCE # SPDX-License-Identifier: Apache-2.0 +from __future__ import annotations + import unittest +from importlib.util import find_spec from os import path +from typing import TYPE_CHECKING, Any, Optional +from unittest import mock from pythainlp.corpus import download from pythainlp.tag import ( @@ -19,8 +24,28 @@ ) from pythainlp.tag._utils import _iob_to_markup +if TYPE_CHECKING: + from pythainlp.tag.phayathaibert_onnx import PhayaThaiBERTTagger + TEST_TOKENS = ["ผม", "รัก", "คุณ"] +_PHAYATHAIBERT_DEPENDENCIES = ( + "huggingface_hub", + "numpy", + "onnxruntime", + "tokenizers", +) + + +def _fake_module(name: str) -> mock.MagicMock: + """Return a stand-in for an installed module, for patching sys.modules.""" + from importlib.machinery import ModuleSpec + + module = mock.MagicMock() + # importlib.util.find_spec() reads __spec__ of modules in sys.modules. + module.__spec__ = ModuleSpec(name, None) + return module + class TagTestCase(unittest.TestCase): """Test pythainlp.tag.pos_tag.""" @@ -523,3 +548,337 @@ def test_iob_to_markup(self): def test_iob_to_markup_long_input(self): tagged = [("ก", "B-A")] + [("ก", "I-A")] * 10000 self.assertEqual(_iob_to_markup(tagged), "" + "ก" * 10001 + "") + + +class PhayaThaiBERTHelperTestCase(unittest.TestCase): + """Test the pure-Python helpers of pythainlp.tag.phayathaibert_onnx.""" + + def test_first_subword_labels(self): + from pythainlp.tag.phayathaibert_onnx import _first_subword_labels + + id2label = {0: "NOUN", 1: "VERB", 2: "PRON"} + # word0-a word0-b word1 + word_ids = [None, 0, 0, 1, None] + label_ids = [1, 2, 1, 0, 1] + self.assertEqual( + _first_subword_labels(word_ids, label_ids, 2, id2label), + ["PRON", "NOUN"], + ) + + def test_first_subword_labels_word_without_subword(self): + from pythainlp.tag.phayathaibert_onnx import _first_subword_labels + + id2label = {0: "NOUN", 1: "VERB"} + # word 1 produced no subword at all + word_ids = [None, 0, 2, None] + label_ids = [0, 0, 1, 0] + self.assertEqual( + _first_subword_labels(word_ids, label_ids, 3, id2label), + ["NOUN", "X", "VERB"], + ) + + def test_chunk_spans(self): + from pythainlp.tag.phayathaibert_onnx import _chunk_spans + + cases = [ + ("empty", [], 5, []), + ("fits in one", [3, 4, 5], 20, [(0, 3)]), + ("exact fit", [3, 4], 7, [(0, 2)]), + ("split at word boundaries", [3, 4, 5], 7, [(0, 2), (2, 3)]), + ( + "oversized word gets own span", + [3, 10, 2], + 5, + [(0, 1), (1, 2), (2, 3)], + ), + ("oversized first word", [10, 1], 5, [(0, 1), (1, 2)]), + ("word with no subword", [3, 0, 4], 7, [(0, 3)]), + ] + for name, counts, limit, expected in cases: + with self.subTest(name): + self.assertEqual(_chunk_spans(counts, limit), expected) + + def test_chunk_spans_exact_fit(self): + from pythainlp.tag.phayathaibert_onnx import _chunk_spans + + self.assertEqual(_chunk_spans([3, 4], 7), [(0, 2)]) + + def test_chunk_spans_zero_count_words(self): + from pythainlp.tag.phayathaibert_onnx import _chunk_spans + + self.assertEqual(_chunk_spans([0, 3, 0, 4], 7), [(0, 4)]) + + def test_is_blank(self): + from pythainlp.tag.phayathaibert_onnx import _is_blank + + for word in ("", " ", "\n", "\u00a0", "\u3000", "\u200b", "\ufeff"): + self.assertTrue(_is_blank(word), repr(word)) + for word in ("แมว", " แมว ", "a", "1", "\u200bแมว"): + self.assertFalse(_is_blank(word), repr(word)) + + def test_tag_empty_list(self): + from pythainlp.tag.phayathaibert_onnx import tag + + self.assertEqual(tag([]), []) + + def test_missing_dependency_error_before_download(self): + import sys + + from pythainlp.tag import phayathaibert_onnx + + for module in _PHAYATHAIBERT_DEPENDENCIES: + with self.subTest(module): + # Fake every dependency, so none is really imported (and + # then unloaded) inside patch.dict. + modules: dict[str, Optional[mock.MagicMock]] = { + m: _fake_module(m) for m in _PHAYATHAIBERT_DEPENDENCIES + } + modules[module] = None + with mock.patch.dict(sys.modules, modules): + with mock.patch( + "pythainlp.tag.phayathaibert_onnx.get_hf_hub" + ) as get_hf_hub: + with self.assertRaises(ImportError) as ctx: + phayathaibert_onnx.PhayaThaiBERTTagger() + self.assertIn( + "pythainlp[phayathaibert_onnx]", str(ctx.exception) + ) + get_hf_hub.assert_not_called() + + def _first_download_revision(self, **kwargs: str) -> object: + """Return the revision PhayaThaiBERTTagger passes to get_hf_hub.""" + import sys + + from pythainlp.tag import phayathaibert_onnx + + fake_modules = { + m: _fake_module(m) for m in _PHAYATHAIBERT_DEPENDENCIES + } + with mock.patch.dict(sys.modules, fake_modules): + with mock.patch( + "pythainlp.tag.phayathaibert_onnx.get_hf_hub", + side_effect=RuntimeError("stop"), + ) as get_hf_hub: + with self.assertRaises(RuntimeError): + phayathaibert_onnx.PhayaThaiBERTTagger(**kwargs) + return get_hf_hub.call_args.kwargs["revision"] + + def test_default_repo_is_pinned(self): + from pythainlp.tag import phayathaibert_onnx + + self.assertEqual( + self._first_download_revision(), phayathaibert_onnx._REVISION + ) + + def test_custom_repo_is_not_pinned_to_default_revision(self): + self.assertIsNone( + self._first_download_revision(repo_id="someone/other-model") + ) + self.assertEqual( + self._first_download_revision( + repo_id="someone/other-model", revision="abc123" + ), + "abc123", + ) + + +class _FakeEncoding: + """ + Encode each word as one subword per character. + + Only the first subword carries the word's label id (1 for words that + start with "ก", 2 otherwise); the others carry 0, so tagging with any + subword but the first gives the wrong tag. + """ + + def __init__(self, words: list[str]) -> None: + self.ids: list[int] = [0] + self.word_ids: list[Optional[int]] = [None] + for i, word in enumerate(words): + label_id = 1 if word.startswith("ก") else 2 + self.ids += [label_id] + [0] * (len(word) - 1) + self.word_ids += [i] * len(word) + self.ids.append(0) + self.word_ids.append(None) + self.attention_mask = [1] * len(self.ids) + + +class _FakeTokenizer: + def encode( + self, words: list[str], is_pretokenized: bool = False + ) -> _FakeEncoding: + return _FakeEncoding(words) + + +class _FakeSession: + """Predict the label id equal to each input id (one-hot logits).""" + + def __init__(self) -> None: + self.lengths: list[int] = [] + + def run( + self, output_names: Optional[list[str]], feeds: dict[str, Any] + ) -> list[Any]: + import numpy as np + + ids = feeds["input_ids"] + self.lengths.append(ids.shape[1]) + return [np.eye(3)[ids]] + + +class PhayaThaiBERTTaggerTestCase(unittest.TestCase): + """Test pythainlp.tag.phayathaibert_onnx with a fake model.""" + + def _tagger(self) -> PhayaThaiBERTTagger: + from pythainlp.tag.phayathaibert_onnx import PhayaThaiBERTTagger + + tagger = PhayaThaiBERTTagger.__new__(PhayaThaiBERTTagger) + tagger.session = _FakeSession() + tagger.tokenizer = _FakeTokenizer() + tagger.id2label = {0: "SCONJ", 1: "NOUN", 2: "VERB"} + return tagger + + @unittest.skipUnless(find_spec("numpy"), "numpy is not installed") + def test_tag(self): + + from pythainlp.tag import phayathaibert_onnx + + cases = [ + ("short", ["กา", "ขาว"], 20, ["NOUN", "VERB"]), + ( + "blank words skipped", + ["กา", " ", "ขา", "", "\u200b"], + 20, + ["NOUN", "PUNCT", "VERB", "PUNCT", "PUNCT"], + ), + ("all blank", [" ", "\ufeff"], 20, ["PUNCT", "PUNCT"]), + ( + "chunked", + ["กา", "ขา", "ก", "ขาว", "กก", "ขข"], + 6, + ["NOUN", "VERB", "NOUN", "VERB", "NOUN", "VERB"], + ), + ("overlong word", ["ก" * 20, "ขา"], 6, ["NOUN", "VERB"]), + ("overlong last word", ["ขา", "ก" * 20], 6, ["VERB", "NOUN"]), + ] + for name, words, limit, expected in cases: + with self.subTest(name): + tagger = self._tagger() + with mock.patch.object( + phayathaibert_onnx, "_MAX_SEQUENCE_LENGTH", limit + ): + result = tagger.tag(words) + self.assertEqual(result, list(zip(words, expected))) + self.assertTrue( + all(n <= limit for n in tagger.session.lengths) + ) + + def test_init(self): + import json + import sys + import tempfile + + from pythainlp.tag import phayathaibert_onnx + + fake_modules = { + m: _fake_module(m) for m in _PHAYATHAIBERT_DEPENDENCIES + } + with tempfile.TemporaryDirectory() as tmp: + config_path = path.join(tmp, "config.json") + with open(config_path, "w", encoding="utf-8") as f: + json.dump({"id2label": {"0": "NOUN", "1": "VERB"}}, f) + + def fake_get_hf_hub(repo_id, filename, revision=None): + return config_path if filename == "config.json" else filename + + with mock.patch.dict(sys.modules, fake_modules): + with mock.patch.object( + phayathaibert_onnx, + "get_hf_hub", + side_effect=fake_get_hf_hub, + ) as get_hf_hub: + tagger = phayathaibert_onnx.PhayaThaiBERTTagger() + self.assertEqual( + [c.args[1] for c in get_hf_hub.call_args_list], + ["model.onnx", "tokenizer.json", "config.json"], + ) + self.assertEqual(tagger.id2label, {0: "NOUN", 1: "VERB"}) + # tokenizer.json truncates at 510 tokens; the tagger must undo it. + tagger.tokenizer.no_truncation.assert_called_once_with() + tagger.tokenizer.no_padding.assert_called_once_with() + + def test_tag_function_loads_tagger_once(self): + + from pythainlp.tag import phayathaibert_onnx + + with mock.patch.object(phayathaibert_onnx, "_TAGGER", None): + with mock.patch.object( + phayathaibert_onnx, "PhayaThaiBERTTagger" + ) as tagger_class: + tagger_class.return_value.tag.return_value = [("ก", "NOUN")] + for _ in range(2): + self.assertEqual( + phayathaibert_onnx.tag(["ก"], corpus="orchid"), + [("ก", "NOUN")], + ) + tagger_class.assert_called_once_with() + + def test_pos_tag_phayathaibert_uses_tud(self): + + with mock.patch( + "pythainlp.tag.phayathaibert_onnx.tag", + return_value=[("ก", "NOUN")], + ) as tag: + self.assertEqual( + pos_tag(["ก"], engine="phayathaibert", corpus="orchid"), + [("ก", "NOUN")], + ) + tag.assert_called_once_with(["ก"], corpus="tud") + + def test_pos_tag_transformers_engine_lookup(self): + import sys + + from pythainlp.tag import pos_tag_transformers + + transformers = mock.MagicMock() + pipeline = transformers.TokenClassificationPipeline.return_value + pipeline.return_value = [{"word": "กิน", "entity_group": "VERB"}] + supported = [ + ("bert", "blackboard", "lunarlist/pos_thai"), + ("phayathai", "blackboard", "lunarlist/pos_thai_phayathai"), + ( + "wangchanberta", + "pud", + "Pavarissy/wangchanberta-ud-thai-pud-upos", + ), + ("mdeberta", "pud", "Pavarissy/mdeberta-v3-ud-thai-pud-upos"), + ( + "phayathaibert", + "tud", + "nlp-chula/phayathaibert-thai-pos-tagger", + ), + ] + unsupported = [ + ("phayathaibert", "pud"), + ("bert", "tud"), + ("bert", "non-existing corpus"), + ] + auto_tokenizer = transformers.AutoTokenizer.from_pretrained + with mock.patch.dict(sys.modules, {"transformers": transformers}): + for engine, corpus, model_name in supported: + with self.subTest(engine=engine, corpus=corpus): + self.assertEqual( + pos_tag_transformers( + "กิน", engine=engine, corpus=corpus + ), + [[("กิน", "VERB")]], + ) + auto_tokenizer.assert_called_with( + model_name, revision=None + ) + for engine, corpus in unsupported: + with self.subTest(engine=engine, corpus=corpus): + with self.assertRaises(ValueError): + pos_tag_transformers( + "กิน", engine=engine, corpus=corpus + ) diff --git a/tests/noauto_onnx/__init__.py b/tests/noauto_onnx/__init__.py index 14a4b2ba3..f75c14128 100644 --- a/tests/noauto_onnx/__init__.py +++ b/tests/noauto_onnx/__init__.py @@ -11,6 +11,7 @@ - onnxruntime - oskut - sefr_cut +- tokenizers and huggingface-hub (PhayaThaiBERT POS tagger) These tests are NOT run in automated CI workflows due to: - Large dependencies diff --git a/tests/noauto_onnx/testn_tag_onnx.py b/tests/noauto_onnx/testn_tag_onnx.py index 6e6fced1c..f2174f25c 100644 --- a/tests/noauto_onnx/testn_tag_onnx.py +++ b/tests/noauto_onnx/testn_tag_onnx.py @@ -4,11 +4,12 @@ # Tests for POS tagging functions that require ONNX Runtime # These tests are NOT run in automated CI workflows due to: -# - Large dependencies (onnxruntime) +# - Large dependencies (onnxruntime, tokenizers, huggingface-hub) # - Platform-specific compatibility issues # - Version constraints import unittest +from typing import ClassVar class TagONNXTestCaseN(unittest.TestCase): @@ -51,3 +52,82 @@ def test_pos_tag_wangchanberta_onnx_empty_list(self): result = pos_tag([], engine="wangchanberta_onnx") self.assertIsInstance(result, list) self.assertEqual(len(result), 0) + + +class TagPhayaThaiBERTONNXTestCaseN(unittest.TestCase): + """Test the PhayaThaiBERT ONNX POS tagger.""" + + WORDS: ClassVar[list[str]] = ["ฉัน", "กิน", "ข้าว", "ที่", "ร้านอาหาร"] + + def test_pos_tag_phayathaibert(self): + from pythainlp.tag import pos_tag + + self.assertEqual( + pos_tag(self.WORDS, engine="phayathaibert"), + [ + ("ฉัน", "PRON"), + ("กิน", "VERB"), + ("ข้าว", "NOUN"), + ("ที่", "ADP"), + ("ร้านอาหาร", "NOUN"), + ], + ) + + def test_pos_tag_phayathaibert_ignores_corpus(self): + from pythainlp.tag import pos_tag + + self.assertEqual( + pos_tag(self.WORDS, engine="phayathaibert", corpus="orchid"), + pos_tag(self.WORDS, engine="phayathaibert", corpus="tud"), + ) + + def test_pos_tag_phayathaibert_whitespace_words(self): + from pythainlp.tag import pos_tag + + result = pos_tag(["แมว", " ", "กิน", ""], engine="phayathaibert") + self.assertEqual(len(result), 4) + self.assertEqual(result[1], (" ", "PUNCT")) + self.assertEqual(result[3], ("", "PUNCT")) + self.assertEqual( + pos_tag(["แมว", "กิน"], engine="phayathaibert"), + [result[0], result[2]], + ) + + def test_pos_tag_phayathaibert_long_input(self): + from pythainlp.tag import pos_tag + + sentence = pos_tag(self.WORDS, engine="phayathaibert") + # Far over the usable length of 415 tokens, so the input is chunked. + result = pos_tag(self.WORDS * 300, engine="phayathaibert") + # Words late in a chunk must be tagged as well as early ones; the + # model is unreliable past position 414 even though it runs on 510. + self.assertEqual(result, sentence * 300) + + def test_pos_tag_phayathaibert_overlong_word(self): + from pythainlp.tag import pos_tag + + # A single word of far more than 415 subwords, between normal words. + words = ["ฉัน", "กิน", "ก" * 5000, "ที่", "ร้านอาหาร"] + result = pos_tag(words, engine="phayathaibert") + tags = [t for _, t in result] + self.assertEqual(len(result), 5) + self.assertNotIn("X", tags) + self.assertEqual(tags[:2], ["PRON", "VERB"]) + self.assertEqual(tags[3:], ["ADP", "NOUN"]) + + def test_pos_tag_phayathaibert_format_characters(self): + from pythainlp.tag import pos_tag + + # Zero-width space, zero-width non-joiner and BOM carry no text. + for char in ("\u200b", "\u200c", "\ufeff"): + result = pos_tag(["แมว", char, "กิน"], engine="phayathaibert") + self.assertEqual(result[1], (char, "PUNCT")) + + def test_pos_tag_sents_phayathaibert(self): + from pythainlp.tag import pos_tag_sents + + result = pos_tag_sents( + [self.WORDS, ["แมว", "กิน", "ปลา"]], engine="phayathaibert" + ) + self.assertEqual(len(result), 2) + self.assertEqual(len(result[1]), 3) diff --git a/tests/noauto_torch/testn_tag_torch.py b/tests/noauto_torch/testn_tag_torch.py index 8e2c05251..97f4ddfb6 100644 --- a/tests/noauto_torch/testn_tag_torch.py +++ b/tests/noauto_torch/testn_tag_torch.py @@ -99,3 +99,37 @@ def test_pos_tag_transformers(self): engine="bert", corpus="non-existing corpus", ) + + def test_pos_tag_transformers_phayathaibert_tud(self): + upos = { + "ADJ", + "ADP", + "ADV", + "AUX", + "CCONJ", + "DET", + "NOUN", + "NUM", + "PART", + "PRON", + "PROPN", + "PUNCT", + "SCONJ", + "SYM", + "VERB", + } + result = pos_tag_transformers( + sentence="ฉันกินข้าวที่ร้านอาหาร", + engine="phayathaibert", + corpus="tud", + ) + self.assertEqual(len(result), 1) + self.assertGreater(len(result[0]), 0) + tags = {tag for _, tag in result[0]} + self.assertLessEqual(tags, upos, tags - upos) + self.assertIn("VERB", tags) + + with self.assertRaises(ValueError): + pos_tag_transformers( + sentence="แมวทำอะไร", engine="phayathaibert", corpus="pud" + )