diff --git a/CHANGELOG.md b/CHANGELOG.md index d04339fb7..a36cb48b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,9 @@ and this project adheres to ### Changed +- Move `pythainlp.phayathaibert`, `pythainlp.wangchanberta`, and + `pythainlp.ulmfit` to `pythainlp.lm`; deprecate the old import paths + ([#1527]) - `pythainlp.tokenize.deepcut`: built-in ONNX engine replaces the TensorFlow-based `deepcut`; `custom_dict` is no longer applied ([#1372]) - Improve guardrails in `check_sara()` and `nighit()` ([#1453]) @@ -85,6 +88,7 @@ and this project adheres to [#1511]: https://github.com/PyThaiNLP/pythainlp/pull/1511 [#1512]: https://github.com/PyThaiNLP/pythainlp/pull/1512 [#1526]: https://github.com/PyThaiNLP/pythainlp/pull/1526 +[#1527]: https://github.com/PyThaiNLP/pythainlp/pull/1527 [#1529]: https://github.com/PyThaiNLP/pythainlp/pull/1529 [#1541]: https://github.com/PyThaiNLP/pythainlp/pull/1541 [#1542]: https://github.com/PyThaiNLP/pythainlp/pull/1542 diff --git a/docs/api/lm.rst b/docs/api/lm.rst index 063aecb2d..0393b7735 100644 --- a/docs/api/lm.rst +++ b/docs/api/lm.rst @@ -3,8 +3,19 @@ pythainlp.lm ============ +The `pythainlp.lm` package provides language models and language modeling utilities. + Modules ------- .. autofunction:: calculate_ngram_counts -.. autofunction:: remove_repeated_ngrams \ No newline at end of file +.. autofunction:: remove_repeated_ngrams +.. autoclass:: Qwen3 + :members: + +Submodules +---------- + +* :mod:`pythainlp.lm.phayathaibert` +* :mod:`pythainlp.lm.wangchanberta` +* :mod:`pythainlp.lm.ulmfit` \ No newline at end of file diff --git a/docs/api/phayathaibert.rst b/docs/api/phayathaibert.rst index 348ab5a24..e29d8a816 100644 --- a/docs/api/phayathaibert.rst +++ b/docs/api/phayathaibert.rst @@ -1,8 +1,14 @@ -.. currentmodule:: pythainlp.phayathaibert +.. currentmodule:: pythainlp.lm.phayathaibert -pythainlp.phayathaibert -======================= -The `pythainlp.phayathaibert` module is built upon the phayathaibert base model. +pythainlp.lm.phayathaibert +========================== + +.. note:: + :mod:`pythainlp.phayathaibert` has moved to :mod:`pythainlp.lm.phayathaibert`. + Importing from :mod:`pythainlp.phayathaibert` still works but emits a + :class:`DeprecationWarning` and will be removed in 6.0. + +The `pythainlp.lm.phayathaibert` module is built upon the phayathaibert base model. Modules ------- diff --git a/docs/api/ulmfit.rst b/docs/api/ulmfit.rst index 527b99c1b..856673973 100644 --- a/docs/api/ulmfit.rst +++ b/docs/api/ulmfit.rst @@ -1,8 +1,14 @@ -.. currentmodule:: pythainlp.ulmfit +.. currentmodule:: pythainlp.lm.ulmfit -pythainlp.ulmfit -==================================== -Welcome to the `pythainlp.ulmfit` module, where you'll find powerful tools for Universal Language Model Fine-tuning for Text Classification (ULMFiT). ULMFiT is a cutting-edge technique for training deep learning models on large text corpora and then fine-tuning them for specific text classification tasks. +pythainlp.lm.ulmfit +=================== + +.. note:: + :mod:`pythainlp.ulmfit` has moved to :mod:`pythainlp.lm.ulmfit`. + Importing from :mod:`pythainlp.ulmfit` still works but emits a + :class:`DeprecationWarning` and will be removed in 6.0. + +Welcome to the `pythainlp.lm.ulmfit` module, where you'll find powerful tools for Universal Language Model Fine-tuning for Text Classification (ULMFiT). ULMFiT is a cutting-edge technique for training deep learning models on large text corpora and then fine-tuning them for specific text classification tasks. Modules ------- @@ -86,5 +92,3 @@ Modules :noindex: The `ungroup_emoji` function is designed for ungrouping emojis in text data, which can be crucial for emoji recognition and classification tasks. - -.. The `pythainlp.ulmfit` module provides a comprehensive set of tools for ULMFiT-based text classification. Whether you need to preprocess Thai text, tokenize it, compute document vectors, or perform various text cleaning tasks, this module has the utilities you need. ULMFiT is a state-of-the-art technique in NLP, and these tools empower you to use it effectively for text classification. diff --git a/docs/api/wangchanberta.rst b/docs/api/wangchanberta.rst index 6df28fc01..29c33f8c3 100644 --- a/docs/api/wangchanberta.rst +++ b/docs/api/wangchanberta.rst @@ -1,8 +1,14 @@ -.. currentmodule:: pythainlp.wangchanberta +.. currentmodule:: pythainlp.lm.wangchanberta -pythainlp.wangchanberta -======================= -The `pythainlp.wangchanberta` module is built upon the WangchanBERTa base model, specifically the `wangchanberta-base-att-spm-uncased` model, as detailed in the paper by Lowphansirikul et al. [#Lowphansirikul_2021]_. +pythainlp.lm.wangchanberta +========================== + +.. note:: + :mod:`pythainlp.wangchanberta` has moved to :mod:`pythainlp.lm.wangchanberta`. + Importing from :mod:`pythainlp.wangchanberta` still works but emits a + :class:`DeprecationWarning` and will be removed in 6.0. + +The `pythainlp.lm.wangchanberta` module is built upon the WangchanBERTa base model, specifically the `wangchanberta-base-att-spm-uncased` model, as detailed in the paper by Lowphansirikul et al. [#Lowphansirikul_2021]_. This base model is utilized for various natural language processing tasks in the Thai language, including named entity recognition, part-of-speech tagging, and subword tokenization. diff --git a/pythainlp/augment/lm/phayathaibert.py b/pythainlp/augment/lm/phayathaibert.py index 0853a2032..988b3f79b 100644 --- a/pythainlp/augment/lm/phayathaibert.py +++ b/pythainlp/augment/lm/phayathaibert.py @@ -12,7 +12,7 @@ if TYPE_CHECKING: from transformers import AutoModelForMaskedLM, AutoTokenizer, Pipeline -from pythainlp.phayathaibert.core import ThaiTextProcessor +from pythainlp.lm.phayathaibert.core import ThaiTextProcessor _MODEL_NAME: str = "clicknext/phayathaibert" diff --git a/pythainlp/lm/phayathaibert/__init__.py b/pythainlp/lm/phayathaibert/__init__.py new file mode 100644 index 000000000..63a393075 --- /dev/null +++ b/pythainlp/lm/phayathaibert/__init__.py @@ -0,0 +1,20 @@ +# SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project +# SPDX-FileType: SOURCE +# SPDX-License-Identifier: Apache-2.0 +"""PhayaThaiBERT language model.""" + +__all__: list[str] = [ + "NamedEntityTagger", + "PartOfSpeechTagger", + "ThaiTextAugmenter", + "ThaiTextProcessor", + "segment", +] + +from pythainlp.lm.phayathaibert.core import ( + NamedEntityTagger, + PartOfSpeechTagger, + ThaiTextAugmenter, + ThaiTextProcessor, + segment, +) diff --git a/pythainlp/lm/phayathaibert/core.py b/pythainlp/lm/phayathaibert/core.py new file mode 100644 index 000000000..3a91d8606 --- /dev/null +++ b/pythainlp/lm/phayathaibert/core.py @@ -0,0 +1,531 @@ +# SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project +# SPDX-FileType: SOURCE +# SPDX-License-Identifier: Apache-2.0 +"""Provide text processing, augmentation, and tagging with PhayaThaiBERT.""" + +from __future__ import annotations + +import random +import re +import warnings +from typing import TYPE_CHECKING, Optional, Union, cast + +if TYPE_CHECKING: + from collections.abc import Callable + + from transformers import ( + AutoModelForMaskedLM, + AutoModelForTokenClassification, + CamembertTokenizer, + Pipeline, + PreTrainedTokenizerBase, + ) + +from pythainlp.tokenize import word_tokenize + +_PAT_URL: str = ( + r"(?:https?|ftp)://[\w-]+(?:\.[\w-]+)+" + r"(?:[\w.,@?^=%&:/~+#-]*[\w@?^=%&/~+#-])?" +) + +_model_name: str = "clicknext/phayathaibert" +_tokenizer: Optional[CamembertTokenizer] = None + + +def _get_tokenizer() -> CamembertTokenizer: + """Get the tokenizer, initializing it if necessary.""" + global _tokenizer + if _tokenizer is None: + from transformers import CamembertTokenizer + + _tokenizer = CamembertTokenizer.from_pretrained( + _model_name # nosec B615 + ) + return _tokenizer + + +class ThaiTextProcessor: + """Preprocess Thai text for PhayaThaiBERT.""" + + def __init__(self) -> None: + """Initialize the text processor.""" + ( + self._TK_UNK, + self._TK_REP, + self._TK_WREP, + self._TK_URL, + self._TK_END, + ) = ["", "", "", "", ""] + self.SPACE_SPECIAL_TOKEN: str = "<_>" # noqa: S105 + + def replace_url(self, text: str) -> str: + """ + Replace URLs in text with the URL token. + + See https://stackoverflow.com/a/6041965 + + :param str text: text to replace url + :return: text where urls are replaced + :rtype: str + :Example: + + >>> replace_url("go to https://github.com") + 'go to ' + """ + return re.sub(_PAT_URL, self._TK_URL, text) + + def rm_brackets(self, text: str) -> str: + """ + Remove empty brackets and artifacts within brackets from text. + + :param str text: text to remove useless brackets + :return: text where all useless brackets are removed + :rtype: str + :Example: + + >>> rm_brackets("hey() whats[;] up{*&} man(hey)") + 'hey whats up man(hey)' + """ + # remove empty brackets + new_line = text.replace("()", "").replace("{}", "").replace("[]", "") + # brackets with only punctuations + new_line = re.sub(r"\([^a-zA-Z0-9ก-๙]+\)", "", new_line) + new_line = re.sub(r"\{[^a-zA-Z0-9ก-๙]+\}", "", new_line) + new_line = re.sub(r"\[[^a-zA-Z0-9ก-๙]+\]", "", new_line) + # artifiacts after ( + new_line = re.sub( + r"(?<=\()[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line + ) + new_line = re.sub( + r"(?<=\{)[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line + ) + new_line = re.sub( + r"(?<=\[)[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line + ) + # artifacts before ) + new_line = re.sub( + r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\))", "", new_line + ) + new_line = re.sub( + r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\})", "", new_line + ) + new_line = re.sub( + r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\])", "", new_line + ) + return new_line + + def replace_newlines(self, text: str) -> str: + """ + Replace newlines in text with spaces. + + :param str text: text to replace all newlines with spaces + :return: text where all newlines are replaced with spaces + :rtype: str + :Example: + + >>> rm_useless_spaces("hey whats\n\nup") + hey whats up + """ + return text.strip().replace("\n", " ") + + def rm_useless_spaces(self, text: str) -> str: + """ + Collapse repeated spaces in text (code from `fastai`). + + :param str text: text to replace useless spaces + :return: text where all spaces are reduced to one + :rtype: str + :Example: + + >>> rm_useless_spaces("oh no") + oh no + """ + return re.sub(" {2,}", " ", text) + + def replace_spaces(self, text: str, space_token: str = "<_>") -> str: # noqa: S107 # nosec B107 + """ + Replace spaces in text with a space token. + + :param str text: text to replace spaces + :return: text where all spaces replaced with _ + :rtype: str + :Example: + + >>> replace_spaces("oh no") + oh_no + """ + return text.replace(" ", space_token) + + def replace_rep_after(self, text: str) -> str: + """ + Remove character repetitions in text. + + :param str text: input text to replace character repetition + :return: text with repetitive tokens removed. + :rtype: str + :Example: + + >>> text = "กาาาาาาา" + >>> replace_rep_after(text) + 'กา' + """ + + def _replace_rep(m: re.Match[str]) -> str: + c, _ = m.groups() + return f"{c}" + + re_rep = re.compile(r"(\S)(\1{3,})") + return re_rep.sub(_replace_rep, text) + + def replace_wrep_post(self, toks: list[str]) -> list[str]: + """ + Remove repeated words after tokenization. + + The `replace_wrep` function of `fastai` does not work well with Thai. + + :param list[str] toks: list of tokens + :return: list of tokens where repetitive words are removed. + :rtype: list[str] + :Example: + + >>> toks = ["กา", "น้ำ", "น้ำ", "น้ำ", "น้ำ"] + >>> replace_wrep_post(toks) + ['กา', 'น้ำ'] + """ + previous_word = "" + rep_count = 0 + res = [] + for current_word in [*toks, self._TK_END]: + if current_word == previous_word: + rep_count += 1 + elif (current_word != previous_word) & (rep_count > 0): + res += [previous_word] + rep_count = 0 + else: + res.append(previous_word) + previous_word = current_word + + return res[1:] + + def remove_space(self, toks: list[str]) -> list[str]: + """ + Remove spaces from a list of words for bag-of-words models. + + :param list[str] toks: list of tokens + :return: List of tokens where space tokens (" ") are filtered out + :rtype: list[str] + :Example: + + >>> toks = ["ฉัน", "เดิน", " ", "กลับ", "บ้าน"] + >>> remove_space(toks) + ['ฉัน', 'เดิน', 'กลับ', 'บ้าน'] + """ + res = [] + for token in toks: + stripped = token.strip() + if stripped: + res.append(stripped) + + return res + + # combine them together + def preprocess( + self, + text: str, + pre_rules: Optional[list[Callable[..., str]]] = None, + tok_func: Callable[..., list[str]] = word_tokenize, + ) -> str: + """Apply preprocessing rules, tokenize text, and join the words.""" + if pre_rules is None: + pre_rules = [ + self.rm_brackets, + self.replace_newlines, + self.rm_useless_spaces, + self.replace_spaces, + self.replace_rep_after, + ] + text = text.lower() + for rule in pre_rules: + text = rule(text) + toks = tok_func(text) + + return "".join(toks) + + +class ThaiTextAugmenter: + """Generate and augment text with PhayaThaiBERT.""" + + def __init__(self) -> None: + """Initialize the text augmenter.""" + from transformers import ( + AutoModelForMaskedLM, + AutoTokenizer, + pipeline, + ) + + self.tokenizer: PreTrainedTokenizerBase = ( + AutoTokenizer.from_pretrained(_model_name) # nosec B615 + ) + self.model_for_masked_lm: AutoModelForMaskedLM = ( + AutoModelForMaskedLM.from_pretrained(_model_name) # nosec B615 + ) + self.model: Pipeline = pipeline( # transformers.Pipeline + "fill-mask", + tokenizer=self.tokenizer, + model=self.model_for_masked_lm, + ) + self.processor: ThaiTextProcessor = ThaiTextProcessor() + + def generate( + self, + sample_text: str, + word_rank: int, + max_length: int = 3, + sample: bool = False, + ) -> str: + """Generate text from PhayaThaiBERT.""" + sample_txt = sample_text + final_text = "" + for _ in range(max_length): + input_text = self.processor.preprocess(sample_txt) + if sample: + # Non-cryptographic use, pseudo-random generator is acceptable here + random_word_idx = random.randint(0, 4) # noqa: S311 # nosec B311 # NOSONAR + output = self.model(input_text)[random_word_idx]["sequence"] + else: + output = self.model(input_text)[word_rank]["sequence"] + sample_txt = output + "" + final_text = sample_txt + + gen_txt = final_text.replace("", "") + + return gen_txt + + def augment( + self, + text: str, + num_augs: int = 3, + sample: bool = False, + ) -> list[str]: + """ + Text augmentation from PhayaThaiBERT. + + :param str text: Thai text + :param int num_augs: an amount of augmentation text needed as an output + :param bool sample: whether to sample the text as an output or not,\ + true if more word diversity is needed + + :return: list of text augment + :rtype: list[str] + + :Example: + + >>> from pythainlp.augment.lm import ThaiTextAugmenter # doctest: +SKIP + + >>> aug = ThaiTextAugmenter() # doctest: +SKIP + >>> aug.augment("ช้างมีทั้งหมด 50 ตัว บน", num_args=5) # doctest: +SKIP + + ['ช้างมีทั้งหมด 50 ตัว บนโลกใบนี้ครับ.', + 'ช้างมีทั้งหมด 50 ตัว บนพื้นดินครับ...', + 'ช้างมีทั้งหมด 50 ตัว บนท้องฟ้าครับ...', + 'ช้างมีทั้งหมด 50 ตัว บนดวงจันทร์.‼', + 'ช้างมีทั้งหมด 50 ตัว บนเขาค่ะ😁'] + """ + MAX_NUM_AUGS = 5 + augment_list = [] + + if num_augs <= MAX_NUM_AUGS: + for rank in range(num_augs): + gen_text = self.generate( + text, + rank, + sample=sample, + ) + processed_text = self.processor.preprocess(gen_text).replace( + "<_>", " " + ) + augment_list.append(processed_text) + else: + raise ValueError( + f"augmentation of more than {num_augs} is exceeded \ + the default limit: {MAX_NUM_AUGS}" + ) + + return augment_list + + +class PartOfSpeechTagger: + """Tag Thai text with part-of-speech labels using PhayaThaiBERT.""" + + def __init__( + self, + model: str = "lunarlist/pos_thai_phayathai", + revision: Optional[str] = None, + ) -> None: + """Initialize the part-of-speech tagger.""" + # Load model directly + from transformers import ( + AutoModelForTokenClassification, + AutoTokenizer, + ) + + self.tokenizer: PreTrainedTokenizerBase = ( + AutoTokenizer.from_pretrained(model, revision=revision) + ) + self.model: AutoModelForTokenClassification = ( + AutoModelForTokenClassification.from_pretrained( + model, revision=revision + ) + ) + + def get_tag( + self, sentence: str, strategy: str = "simple" + ) -> list[list[tuple[str, str]]]: + """ + Tag sentences with part-of-speech (POS) labels. + + :param str sentence: a list of lists of tokenized words + :return: a list of lists of tuples (word, POS tag) + :rtype: list[list[tuple[str, str]]] + + :Example: + + Labels POS for given sentence: + + >>> from pythainlp.lm.phayathaibert.core import ( + ... PartOfSpeechTagger, + ... ) # doctest: +SKIP + + >>> tagger = PartOfSpeechTagger() # doctest: +SKIP + >>> tagger.get_tag("แมวทำอะไรตอนห้าโมงเช้า") # doctest: +SKIP + [[('แมว', 'NOUN'), ('ทําอะไร', 'VERB'), ('ตอนห้าโมงเช้า', 'NOUN')]] + """ + from transformers import TokenClassificationPipeline + + pipeline = TokenClassificationPipeline( + model=self.model, + tokenizer=self.tokenizer, + aggregation_strategy=strategy, + ) + outputs = pipeline(sentence) + word_tags = [[(tag["word"], tag["entity_group"]) for tag in outputs]] + + return word_tags + + +class NamedEntityTagger: + """Tag named entities with PhayaThaiBERT.""" + + def __init__( + self, + model: str = "Pavarissy/phayathaibert-thainer", + revision: Optional[str] = None, + ) -> None: + """Initialize the named entity tagger.""" + from transformers import ( + AutoModelForTokenClassification, + AutoTokenizer, + ) + + self.tokenizer: PreTrainedTokenizerBase = ( + AutoTokenizer.from_pretrained(model, revision=revision) + ) + self.model: AutoModelForTokenClassification = ( + AutoModelForTokenClassification.from_pretrained( + model, revision=revision + ) + ) + + def get_ner( + self, + text: str, + tag: bool = False, + pos: bool = False, + strategy: str = "simple", + ) -> Union[list[tuple[str, str]], list[tuple[str, str, str]], str]: + """ + Tag named entities in text. + + :param str text: text in Thai to be tagged + :param bool pos: output with part-of-speech tags.\ + (PhayaThaiBERT is supported in PartOfSpeechTagger) + :return: a list of tuples associated with tokenized words, NER tags, + POS tags (if the parameter `pos` is specified as `True`), + and output HTML-like tags (if the parameter `tag` is + specified as `True`). + Otherwise, return a list of tuples associated with tokenized + words and NER tags + :rtype: Union[list[tuple[str, str]], list[tuple[str, str, str]], str] + :Example: + + >>> from pythainlp.lm.phayathaibert.core import NamedEntityTagger + >>> + >>> tagger = NamedEntityTagger() + >>> tagger.get_ner("ทดสอบนายปวริศ เรืองจุติโพธิ์พานจากประเทศไทย") + [('นายปวริศ เรืองจุติโพธิ์พานจากประเทศไทย', 'PERSON'), + ('จาก', 'LOCATION'), + ('ประเทศไทย', 'LOCATION')] + >>> ner.tag("ทดสอบนายปวริศ เรืองจุติโพธิ์พานจากประเทศไทย", tag=True) + 'ทดสอบนายปวริศ เรืองจุติโพธิ์พาน\ + จากประเทศไทย' + """ + from transformers import TokenClassificationPipeline + + if pos: + warnings.warn( + "This model does not support POS tag output.", + UserWarning, + stacklevel=2, + ) + + sample_output = [] + tag_text_list = [] + current_pos = 0 + pipeline = TokenClassificationPipeline( + model=self.model, + tokenizer=self.tokenizer, + aggregation_strategy=strategy, + ) + outputs = pipeline(text) + + for token in outputs: + ner_tag = token["entity_group"] + begin_pos, end_pos = token["start"], token["end"] + if current_pos == 0: + text_tag = ( + text[:begin_pos] + + f"<{ner_tag}>" + + text[begin_pos:end_pos] + + f"" + ) + else: + text_tag = ( + text[current_pos:begin_pos] + + f"<{ner_tag}>" + + text[begin_pos:end_pos] + + f"" + ) + tag_text_list.append(text_tag) + sample_output.append((token["word"], token["entity_group"])) + current_pos = end_pos + + if tag: + return str("".join(tag_text_list)) + + return sample_output + + +def segment(sentence: str) -> list[str]: + """ + Tokenize text into subwords with the PhayaThaiBERT tokenizer. + + The tokenizer is a SentencePiece model with vocabulary expansion. + + :param str sentence: text to be tokenized + :return: list of subwords + :rtype: list[str] + """ + if not sentence or not isinstance(sentence, str): + return [] + + return cast("list[str]", _get_tokenizer().tokenize(sentence)) diff --git a/pythainlp/lm/ulmfit/__init__.py b/pythainlp/lm/ulmfit/__init__.py new file mode 100644 index 000000000..bfaebbb4f --- /dev/null +++ b/pythainlp/lm/ulmfit/__init__.py @@ -0,0 +1,69 @@ +# SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project +# SPDX-FileType: SOURCE +# SPDX-License-Identifier: Apache-2.0 +""" +Universal Language Model Fine-tuning for Text Classification (ULMFiT). + +Code by Charin Polpanumas +https://github.com/cstorm125/thai2fit/ + +Some pre-processing functions are from fastai (Apache 2.0) +https://github.com/fastai/fastai/blob/master/fastai/text/transform.py + +Universal Language Model Fine-tuning for Text Classification +https://arxiv.org/abs/1801.06146 +""" + +__all__: list[str] = [ + "THWIKI_LSTM", + "ThaiTokenizer", + "document_vector", + "fix_html", + "get_thwiki_lstm", + "lowercase_all", + "merge_wgts", + "post_rules_th", + "post_rules_th_sparse", + "pre_rules_th", + "pre_rules_th_sparse", + "process_thai", + "remove_space", + "replace_rep_after", + "replace_rep_nonum", + "replace_url", + "replace_wrep_post", + "replace_wrep_post_nonum", + "rm_brackets", + "rm_useless_newlines", + "rm_useless_spaces", + "spec_add_spaces", + "ungroup_emoji", +] + +from pythainlp.lm.ulmfit.core import ( + THWIKI_LSTM, + document_vector, + get_thwiki_lstm, + merge_wgts, + post_rules_th, + post_rules_th_sparse, + pre_rules_th, + pre_rules_th_sparse, + process_thai, +) +from pythainlp.lm.ulmfit.preprocess import ( + fix_html, + lowercase_all, + remove_space, + replace_rep_after, + replace_rep_nonum, + replace_url, + replace_wrep_post, + replace_wrep_post_nonum, + rm_brackets, + rm_useless_newlines, + rm_useless_spaces, + spec_add_spaces, + ungroup_emoji, +) +from pythainlp.lm.ulmfit.tokenizer import ThaiTokenizer diff --git a/pythainlp/lm/ulmfit/core.py b/pythainlp/lm/ulmfit/core.py new file mode 100644 index 000000000..26c051043 --- /dev/null +++ b/pythainlp/lm/ulmfit/core.py @@ -0,0 +1,322 @@ +# SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project +# SPDX-FileType: SOURCE +# SPDX-License-Identifier: Apache-2.0 +"""Universal Language Model Fine-tuning for Text Classification (ULMFiT).""" + +from __future__ import annotations + +import collections +from typing import TYPE_CHECKING, Any, Optional, cast + +import torch + +if TYPE_CHECKING: + from collections.abc import Callable, Collection + + import numpy as np + from numpy.typing import NDArray + +from pythainlp.corpus import get_corpus_path +from pythainlp.lm.ulmfit.preprocess import ( + fix_html, + lowercase_all, + remove_space, + replace_rep_after, + replace_rep_nonum, + replace_url, + replace_wrep_post, + replace_wrep_post_nonum, + rm_brackets, + rm_useless_newlines, + rm_useless_spaces, + spec_add_spaces, + ungroup_emoji, +) +from pythainlp.tokenize import thai2fit_tokenizer +from pythainlp.util import reorder_vowels + +device: torch.device = torch.device( + "cuda" if torch.cuda.is_available() else "cpu" +) + +_MODEL_NAME_LSTM: str = "wiki_lm_lstm" +_ITOS_NAME_LSTM: str = "wiki_itos_lstm" + + +# Pretrained model paths +# Note: These may be None if corpus is not downloaded. +# Access via get_thwiki_lstm() for proper validation or use directly +# if you've already verified the corpus is downloaded. +THWIKI_LSTM: dict[str, Optional[str]] = { + "wgts_fname": get_corpus_path(_MODEL_NAME_LSTM), + "itos_fname": get_corpus_path(_ITOS_NAME_LSTM), + "json_itos_fname": get_corpus_path("wiki_itos_lstm_json"), +} + + +def get_thwiki_lstm() -> dict[str, str]: + """ + Get THWIKI LSTM model paths with validation. + + :return: dictionary with ``wgts_fname`` and ``itos_fname`` keys + :rtype: dict[str, str] + :raises FileNotFoundError: if corpus files are not found + """ + wgts_fname = THWIKI_LSTM["wgts_fname"] + itos_fname = THWIKI_LSTM["itos_fname"] + + if not wgts_fname or not itos_fname: + raise FileNotFoundError( + "corpus-not-found names=['wiki_lm_lstm', 'wiki_itos_lstm']\n" + " ULMFiT model files not found.\n" + " Python: pythainlp.corpus.download('wiki_lm_lstm')\n" + " CLI: thainlp data get wiki_lm_lstm\n" + " Python: pythainlp.corpus.download('wiki_itos_lstm')\n" + " CLI: thainlp data get wiki_itos_lstm" + ) + + return {"wgts_fname": wgts_fname, "itos_fname": itos_fname} + + +# Preprocessing rules for Thai text +# dense features +pre_rules_th: list[Callable[[str], str]] = [ + replace_rep_after, + fix_html, + reorder_vowels, + spec_add_spaces, + rm_useless_spaces, + rm_useless_newlines, + rm_brackets, + replace_url, +] +post_rules_th: list[Callable[[Collection[str]], list[str]]] = [ + replace_wrep_post, + ungroup_emoji, + lowercase_all, +] + +# sparse features +pre_rules_th_sparse: list[Callable[[str], str]] = [ + *pre_rules_th[1:], + replace_rep_nonum, +] +post_rules_th_sparse: list[Callable[[Collection[str]], list[str]]] = [ + *post_rules_th[1:], + replace_wrep_post_nonum, + remove_space, +] + + +def process_thai( + text: str, + pre_rules: Optional[Collection[Callable[[str], str]]] = None, + tok_func: Optional[Callable[[str], list[str]]] = None, + post_rules: Optional[Collection[Callable[[list[str]], list[str]]]] = None, +) -> list[str]: + """ + Process Thai text for models, with sparse features by default. + + :param str text: text to be cleaned + :param Optional[Collection[Callable[[str], str]]] pre_rules: rules to + apply before tokenization. If None, use the default sparse pre-rules. + :param Optional[Callable[[str], list[str]]] tok_func: tokenization + function. By default, **tok_func** is + :func:`pythainlp.tokenize.word_tokenize`. + + :param Optional[Collection[Callable[[list[str]], list[str]]]] post_rules: + rules to apply after tokenization. If None, use the default sparse + post-rules. + + :return: a list of cleaned tokenized texts + :rtype: list[str] + + + :Note: + - The default **pre-rules** consists of :func:`fix_html`, + :func:`pythainlp.util.normalize`, + :func:`spec_add_spaces`, + :func:`rm_useless_spaces`, + :func:`rm_useless_newlines`, + :func:`rm_brackets` + and :func:`replace_rep_nonum`. + + - The default **post-rules** consists of :func:`ungroup_emoji`, + :func:`lowercase_all`, :func:`replace_wrep_post_nonum`, + and :func:`remove_space`. + + :Example: + + 1. Use default pre-rules and post-rules: + + >>> from pythainlp.lm.ulmfit import process_thai + >>> text = "บ้านนนนน () อยู่นานนานนาน 😂🤣😃😄😅 PyThaiNLP amp; " + >>> process_thai(text) + ['บ้าน', 'xxrep', ' ', 'อยู่', 'xxwrep', 'นาน', '😂', '🤣', + '😃', '😄', '😅', 'pythainlp', '&'] + + 2. Modify pre_rules and post_rules arguments with + rules provided in :mod:`pythainlp.lm.ulmfit`: + + >>> from pythainlp.lm.ulmfit import ( + process_thai, + replace_rep_after, + fix_html, + ungroup_emoji, + replace_wrep_post, + remove_space) + >>> + >>> text = "บ้านนนนน () อยู่นานนานนาน 😂🤣😃😄😅 PyThaiNLP amp; " + >>> process_thai(text, + pre_rules=[replace_rep_after, fix_html], + post_rules=[ungroup_emoji, + replace_wrep_post, + remove_space] + ) + ['บ้าน', 'xxrep', '5', '()', 'อยู่', 'xxwrep', '2', 'นาน', '😂', '🤣', + '😃', '😄', '😅', 'PyThaiNLP', '&'] + + + """ + processed_text = text + if pre_rules is None: + pre_rules = pre_rules_th_sparse + if post_rules is None: + post_rules = cast( + "Collection[Callable[[list[str]], list[str]]]", + post_rules_th_sparse, + ) + + if tok_func is None: + tok_func = thai2fit_tokenizer().word_tokenize + + for pre_rule in pre_rules: + processed_text = pre_rule(processed_text) + tokens = tok_func(processed_text) + for post_rule in post_rules: + tokens = post_rule(tokens) + + return tokens + + +def document_vector( + text: str, learn: Any, data: Any, agg: str = "mean" +) -> NDArray[np.float32]: + """ + Vectorize a Thai sentence into a 400-dimension vector. + + Uses a :class:`fastai` language model and data bunch. + Word vectors are aggregated by mean or summation. + + :param str text: text to vectorize + :param learn: :class:`fastai` language model learner + :param data: :class:`fastai` data bunch + :param str agg: aggregation method; ``"mean"`` or ``"sum"`` + + :return: document vector of shape ``(1, 400)`` + :rtype: numpy.typing.NDArray[numpy.float32] + + :Example: + + >>> from pythainlp.lm.ulmfit import document_vector + >>> from fastai.text import load_data, language_model_learner, AWD_LSTM + >>> + >>> # Load Data Bunch + >>> data = load_data(MODEL_PATH, 'thwiki_lm_data.pkl') + >>> + >>> # Initialize language_model_learner + >>> config = dict(emb_sz=400, n_hid=1550, n_layers=4, pad_token=1, + qrnn=False, tie_weights=True, out_bias=True, output_p=0.25, + hidden_p=0.1, input_p=0.2, embed_p=0.02, weight_p=0.15) + >>> trn_args = dict(drop_mult=0.9, clip=0.12, alpha=2, beta=1) + >>> learn = language_model_learner(data, AWD_LSTM, config=config, + pretrained=False, **trn_args) + >>> document_vector('วันนี้วันดีปีใหม่', learn, data) + + :See Also: + * A notebook showing how to train `ulmfit` language model and its + usage, `Jupyter Notebook \ + `_ + + """ + s = thai2fit_tokenizer().word_tokenize(text) + t = torch.tensor(data.vocab.numericalize(s), requires_grad=False).to( + device + ) + m = learn.model[0].encoder.to(device) + res = m(t).cpu().detach().numpy().astype("float32", copy=False) + if agg == "mean": + res = res.mean(0, dtype="float32") + elif agg == "sum": + res = res.sum(0, dtype="float32") + else: + raise ValueError("Aggregate by mean or sum") + + return cast("NDArray[np.float32]", res) + + +def merge_wgts( + em_sz: int, + wgts: dict[str, torch.Tensor], + itos_pre: list[str], + itos_new: list[str], +) -> dict[str, torch.Tensor]: + """ + Insert new vocab into an existing model and update weights. + + New vocab weights are initialised with the average embedding + when not found in the pretrained vocab. + + :param int em_sz: embedding size + :param wgts: torch model weights + :param list[str] itos_pre: pretrained list of vocab + :param list[str] itos_new: list of new vocab + + :return: merged torch model weights + :rtype: dict[str, torch.Tensor] + + :Example: + + >>> from pythainlp.lm.ulmfit import merge_wgts # doctest: +SKIP + >>> import torch # doctest: +SKIP + + >>> wgts = {"0.encoder.weight": torch.randn(5, 3)} # doctest: +SKIP + >>> itos_pre = ["แมว", "คน", "หนู"] # doctest: +SKIP + >>> itos_new = ["ปลา", "เต่า", "นก"] # doctest: +SKIP + >>> em_sz = 3 # doctest: +SKIP + + >>> merge_wgts(em_sz, wgts, itos_pre, itos_new) # doctest: +SKIP + {'0.encoder.weight': tensor([[0.5952, 0.4453, 0.0011], + [0.5952, 0.4453, 0.0011], + [0.5952, 0.4453, 0.0011]]), + '0.encoder_dp.emb.weight': tensor([[0.5952, 0.4453, 0.0011], + [0.5952, 0.4453, 0.0011], + [0.5952, 0.4453, 0.0011]]), + '1.decoder.weight': tensor([[0.5952, 0.4453, 0.0011], + [0.5952, 0.4453, 0.0011], + [0.5952, 0.4453, 0.0011]])} + """ + vocab_size = len(itos_new) + enc_wgts = wgts["0.encoder.weight"].numpy().astype("float32", copy=False) + + # Average weight of encoding + row_m = enc_wgts.mean(0, dtype="float32") + stoi_pre = collections.defaultdict( + lambda: -1, {v: k for k, v in enumerate(itos_pre)} + ) + + # New embedding based on classification dataset + import numpy as np + + new_w = np.zeros((vocab_size, em_sz), dtype=np.float32) + + for i, w in enumerate(itos_new): + r = stoi_pre[w] + # Use pretrianed embedding if present; else use the average + new_w[i] = enc_wgts[r] if r >= 0 else row_m + + wgts["0.encoder.weight"] = torch.tensor(new_w) + wgts["0.encoder_dp.emb.weight"] = torch.tensor(np.copy(new_w)) + wgts["1.decoder.weight"] = torch.tensor(np.copy(new_w)) + + return wgts diff --git a/pythainlp/lm/ulmfit/preprocess.py b/pythainlp/lm/ulmfit/preprocess.py new file mode 100644 index 000000000..8c89cee0e --- /dev/null +++ b/pythainlp/lm/ulmfit/preprocess.py @@ -0,0 +1,296 @@ +# SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project +# SPDX-FileType: SOURCE +# SPDX-License-Identifier: Apache-2.0 +"""Preprocessing for ULMFiT.""" + +from __future__ import annotations + +import html +import re +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from collections.abc import Collection + +import emoji + +_TK_UNK: str = "xxunk" +_TK_REP: str = "xxrep" +_TK_WREP: str = "xxwrep" +_TK_END: str = "xxend" +_TK_URL: str = "xxurl" + + +def replace_url(text: str) -> str: + """ + Replace URL in `text` with TK_URL. + + :param str text: text to replace URL in + + :return: text with URLs replaced + :rtype: str + + :Example: + + >>> from pythainlp.lm.ulmfit import replace_url + >>> replace_url("go to github.com") + 'go to xxurl' + """ + URL_PATTERN = r"""(?i)\b((?:https?:(?:/{1,3}|[a-z0-9%])|[a-z0-9.-]+\.(?:com|net|org|edu|gov|mil|aero|asia|biz|cat|coop|info|int|jobs|mobi|museum|name|post|pro|tel|travel|xxx|ac|ad|ae|af|ag|ai|al|am|an|ao|aq|ar|as|at|au|aw|ax|az|ba|bb|bd|be|bf|bg|bh|bi|bj|bm|bn|bo|br|bs|bt|bv|bw|by|bz|ca|cc|cd|cf|cg|ch|ci|ck|cl|cm|cn|co|cr|cs|cu|cv|cx|cy|cz|dd|de|dj|dk|dm|do|dz|ec|ee|eg|eh|er|es|et|eu|fi|fj|fk|fm|fo|fr|ga|gb|gd|ge|gf|gg|gh|gi|gl|gm|gn|gp|gq|gr|gs|gt|gu|gw|gy|hk|hm|hn|hr|ht|hu|id|ie|il|im|in|io|iq|ir|is|it|je|jm|jo|jp|ke|kg|kh|ki|km|kn|kp|kr|kw|ky|kz|la|lb|lc|li|lk|lr|ls|lt|lu|lv|ly|ma|mc|md|me|mg|mh|mk|ml|mm|mn|mo|mp|mq|mr|ms|mt|mu|mv|mw|mx|my|mz|na|nc|ne|nf|ng|ni|nl|no|np|nr|nu|nz|om|pa|pe|pf|pg|ph|pk|pl|pm|pn|pr|ps|pt|pw|py|qa|re|ro|rs|ru|rw|sa|sb|sc|sd|se|sg|sh|si|sj|Ja|sk|sl|sm|sn|so|sr|ss|st|su|sv|sx|sy|sz|tc|td|tf|tg|th|tj|tk|tl|tm|tn|to|tp|tr|tt|tv|tw|tz|ua|ug|uk|us|uy|uz|va|vc|ve|vg|vi|vn|vu|wf|ws|ye|yt|yu|za|zm|zw)/)(?:[^\s()<>{}\[\]]|\([^\s()]*?\([^\s()]+\)[^\s()]*?\)|\([^\s]+?\))+(?:\([^\s()]*?\([^\s()]+\)[^\s()]*?\)|\([^\s]+?\)|[^\s`!()\[\]{};:'".,<>?«»“”‘’])|(?:(? str: + """ + Replace HTML strings in `test`. (codes from `fastai`). + + :param str text: text to replace HTML strings in + + :return: text with HTML strings replaced + :rtype: str + + :Example: + + >>> from pythainlp.lm.ulmfit import fix_html + >>> fix_html("Anbsp;amp;nbsp;B @.@ ") + 'A & B.' + """ + re1 = re.compile(r" +") + text = ( + text.replace("#39;", "'") + .replace("amp;", "&") + .replace("#146;", "'") + .replace("nbsp;", " ") + .replace("#36;", "$") + .replace("\\n", "\n") + .replace("quot;", "'") + .replace("
", "\n") + .replace('\\"', '"') + .replace("", _TK_UNK) + .replace(" @.@ ", ".") + .replace(" @-@ ", "-") + .replace(" @,@ ", ",") + .replace("\\", " \\ ") + ) + return re1.sub(" ", html.unescape(text)) + + +def rm_useless_spaces(text: str) -> str: + """Remove multiple spaces in `text`. (codes from `fastai`).""" + return re.sub(" {2,}", " ", text) + + +def spec_add_spaces(text: str) -> str: + """Add spaces around / and # in `text`. \n (codes from `fastai`).""" + return re.sub(r"([/#\n])", r" \1 ", text) + + +def replace_rep_after(text: str) -> str: + """ + Replace repetitions at the character level in `text` after the repeated character. + + This is to prevent cases such as 'น้อยยยยยยยย' becomes 'น้อ xxrep 8 ย' + ; instead it will retain the word as 'น้อย xxrep 8'. + + :param str text: input text to replace character repetitions in + + :return: text with repetitive token **xxrep** and the counter + after the repeated character + + :rtype: str + :Example: + + >>> from pythainlp.lm.ulmfit import replace_rep_after + >>> + >>> text = "กาาาาาาา" + >>> replace_rep_after(text) + 'กาxxrep7 ' + """ + + def _replace_rep(m: re.Match[str]) -> str: + c, cc = m.groups() + return f"{c}{_TK_REP}{len(cc) + 1} " + + re_rep = re.compile(r"(\S)(\1{3,})") + + return re_rep.sub(_replace_rep, text) + + +def replace_wrep_post(toks: Collection[str]) -> list[str]: + """ + Replace repetitive words after tokenization. + + fastai `replace_wrep` does not work well with Thai. + + :param list[str] toks: list of tokens + + :return: list of tokens where **xxwrep** token and the counter + is added before repetitive words. + :rtype: list[str] + + :Example: + + >>> from pythainlp.lm.ulmfit import replace_wrep_post_nonum + >>> + >>> toks = ["กา", "น้ำ", "น้ำ", "น้ำ", "น้ำ"] + >>> replace_wrep_post(toks) + ['กา', 'xxwrep', '3', 'น้ำ'] + + """ + previous_word: Optional[str] = None + rep_count = 0 + res: list[Optional[str]] = [] + for current_word in [*list(toks), _TK_END]: + if current_word == previous_word: + rep_count += 1 + elif (current_word != previous_word) & (rep_count > 0): + res += [_TK_WREP, str(rep_count), previous_word] + rep_count = 0 + else: + res.append(previous_word) + previous_word = current_word + return list(filter(None, res[1:])) + + +def rm_useless_newlines(text: str) -> str: + """Remove multiple newlines in `text`.""" + return re.sub(r"\n{2,}", " ", text) + + +def rm_brackets(text: str) -> str: + """Remove all empty brackets and artifacts within brackets from `text`.""" + # remove empty brackets + new_line = text.replace("()", "").replace("{}", "").replace("[]", "") + # brackets with only punctuation marks + new_line = re.sub(r"\([^a-zA-Z0-9ก-๙]+\)", "", new_line) + new_line = re.sub(r"\{[^a-zA-Z0-9ก-๙]+\}", "", new_line) + new_line = re.sub(r"\[[^a-zA-Z0-9ก-๙]+\]", "", new_line) + # artifacts after ( + new_line = re.sub( + r"(?<=\()[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line + ) + new_line = re.sub( + r"(?<=\{)[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line + ) + new_line = re.sub( + r"(?<=\[)[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line + ) + # artifacts before ) + new_line = re.sub( + r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\))", "", new_line + ) + new_line = re.sub( + r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\})", "", new_line + ) + new_line = re.sub( + r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\])", "", new_line + ) + return new_line + + +def ungroup_emoji(toks: Collection[str]) -> list[str]: + """ + Ungroup Zero Width Joiner (ZVJ) Emojis. + + See https://emojipedia.org/emoji-zwj-sequence/ + """ + res = [] + for tok in toks: + if emoji.emoji_count(tok) == len(tok): + res.extend(list(tok)) + else: + res.append(tok) + return res + + +def lowercase_all(toks: Collection[str]) -> list[str]: + """ + Lowercase all English words. + + English words in Thai texts don't usually have nuances of capitalization. + """ + return list(map(str.lower, toks)) + + +def replace_rep_nonum(text: str) -> str: + """ + Replace repetitions at the character level in `text` after the repetition. + + This is done to prevent such case as 'น้อยยยยยยยย' becoming 'น้อ xxrep ย'; + instead it will retain the word as 'น้อย xxrep '. + + :param str text: input text to replace character repetition + + :return: text with repetitive token **xxrep** after + character repetition + :rtype: str + + :Example: + + >>> from pythainlp.lm.ulmfit import replace_rep_nonum + >>> + >>> text = "กาาาาาาา" + >>> replace_rep_nonum(text) + 'กา xxrep ' + + """ + + def _replace_rep(m: re.Match[str]) -> str: + c, _ = m.groups() + return f"{c} {_TK_REP} " + + re_rep = re.compile(r"(\S)(\1{3,})") + return re_rep.sub(_replace_rep, text) + + +def replace_wrep_post_nonum(toks: Collection[str]) -> list[str]: + """ + Replace repeated words after tokenization. + + fastai `replace_wrep` does not work well with Thai. + + :param list[str] toks: list of tokens + + :return: list of tokens where **xxwrep** token is added in front of + repetitive words. + :rtype: list[str] + + :Example: + + >>> from pythainlp.lm.ulmfit import replace_wrep_post_nonum + >>> + >>> toks = ["กา", "น้ำ", "น้ำ", "น้ำ", "น้ำ"] + >>> replace_wrep_post_nonum(toks) + ['กา', 'xxwrep', 'น้ำ'] + + """ + previous_word: Optional[str] = None + rep_count = 0 + res: list[Optional[str]] = [] + for current_word in [*list(toks), _TK_END]: + if current_word == previous_word: + rep_count += 1 + elif (current_word != previous_word) & (rep_count > 0): + res += [_TK_WREP, previous_word] + rep_count = 0 + else: + res.append(previous_word) + previous_word = current_word + return list(filter(None, res[1:])) + + +def remove_space(toks: Collection[str]) -> list[str]: + """ + Do not include space for bag-of-word models. + + :param list[str] toks: list of tokens + + :return: list of tokens where space tokens (" ") are filtered out + :rtype: list[str] + """ + res = [] + for token in toks: + stripped = token.strip() + if stripped: + res.append(stripped) + return res diff --git a/pythainlp/lm/ulmfit/tokenizer.py b/pythainlp/lm/ulmfit/tokenizer.py new file mode 100644 index 000000000..d7896a4b5 --- /dev/null +++ b/pythainlp/lm/ulmfit/tokenizer.py @@ -0,0 +1,77 @@ +# SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project +# SPDX-FileType: SOURCE +# SPDX-License-Identifier: Apache-2.0 +"""Provide tokenizer classes for ULMFiT.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Collection + +from pythainlp.tokenize import thai2fit_tokenizer + + +class BaseTokenizer: + """Provide a basic tokenizer interface (code from `fastai`).""" + + lang: str + + def __init__(self, lang: str) -> None: + """Initialize tokenizer with a language code.""" + self.lang: str = lang + + def tokenizer(self, t: str) -> list[str]: + """Split text on spaces.""" + return t.split(" ") + + def add_special_cases(self, toks: Collection[str]) -> None: + """ + Accept special cases without changing this tokenizer. + + The fastai interface requires this hook, but this tokenizer has + no custom special cases. + """ + # The fastai interface requires this hook, but this tokenizer has + # no custom special cases. + + +class ThaiTokenizer(BaseTokenizer): + """Wrap the newmm tokenizer as a fastai tokenizer.""" + + lang: str + + def __init__(self, lang: str = "th") -> None: + """Initialize the tokenizer with a language code.""" + self.lang: str = lang + + @staticmethod + def tokenizer(text: str) -> list[str]: + """ + Tokenize text using the newmm engine and the thai2fit dictionary. + + :param str text: text to tokenize + :return: tokenized text + :rtype: list[str] + + :Example: + + Using :func:`ThaiTokenizer.tokenizer` is similar to + :func:`pythainlp.tokenize.word_tokenize` with the + ``"ulmfit"`` engine. + + >>> from pythainlp.lm.ulmfit import ThaiTokenizer + >>> from pythainlp.tokenize import word_tokenize + >>> + >>> text = "อาภรณ์, จินตมยปัญญา ภาวนามยปัญญา" + >>> ThaiTokenizer.tokenizer(text) + ['อาภรณ์', ',', ' ', 'จิน', 'ตม', 'ย', 'ปัญญา', + ' ', 'ภาวนามยปัญญา'] + >>> + >>> word_tokenize(text, engine="ulmfit") + ['อาภรณ์', ',', ' ', 'จิน', 'ตม', 'ย', 'ปัญญา', + ' ', 'ภาวนามยปัญญา'] + + """ + return thai2fit_tokenizer().word_tokenize(text) diff --git a/pythainlp/lm/wangchanberta/__init__.py b/pythainlp/lm/wangchanberta/__init__.py new file mode 100644 index 000000000..edbd41dee --- /dev/null +++ b/pythainlp/lm/wangchanberta/__init__.py @@ -0,0 +1,16 @@ +# SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project +# SPDX-FileType: SOURCE +# SPDX-License-Identifier: Apache-2.0 +"""Named entity recognition and tokenization using WangchanBERTa.""" + +__all__: list[str] = [ + "NamedEntityRecognition", + "ThaiNameTagger", + "segment", +] + +from pythainlp.lm.wangchanberta.core import ( + NamedEntityRecognition, + ThaiNameTagger, + segment, +) diff --git a/pythainlp/lm/wangchanberta/core.py b/pythainlp/lm/wangchanberta/core.py new file mode 100644 index 000000000..ce501283f --- /dev/null +++ b/pythainlp/lm/wangchanberta/core.py @@ -0,0 +1,283 @@ +# SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project +# SPDX-FileType: SOURCE +# SPDX-License-Identifier: Apache-2.0 +"""Recognize named entities and tokenize subwords with WangchanBERTa.""" + +from __future__ import annotations + +import warnings +from typing import TYPE_CHECKING, Optional, Union, cast + +if TYPE_CHECKING: + from transformers import ( + CamembertTokenizer, + PreTrainedModel, + PreTrainedTokenizerBase, + ) + from transformers.pipelines import TokenClassificationPipeline + +from pythainlp.tokenize import word_tokenize + +_model_name: str = "wangchanberta-base-att-spm-uncased" +_tokenizer: Optional[CamembertTokenizer] = None + + +def _get_tokenizer() -> CamembertTokenizer: + """Get the tokenizer, initializing it if necessary.""" + global _tokenizer + if _tokenizer is None: + from transformers import CamembertTokenizer + + _tokenizer = CamembertTokenizer.from_pretrained( + f"airesearch/{_model_name}", + revision="main", # nosec B615 + ) + if _model_name == "wangchanberta-base-att-spm-uncased": + _tokenizer.additional_special_tokens = [ + "NOTUSED", + "NOTUSED", + "<_>", + ] + return _tokenizer + + +def _format_ner_tags(entities: list[tuple[str, str]]) -> str: + """Wrap named entities in HTML-like tags.""" + tagged_text = [] + active_entity = "" + for word, ner in entities: + if ner.startswith("B-"): + if active_entity: + tagged_text.append(f"") + active_entity = ner[2:] + tagged_text.append(f"<{active_entity}>") + elif ner == "O" and active_entity: + tagged_text.append(f"") + active_entity = "" + tagged_text.append(word) + if active_entity: + tagged_text.append(f"") + return "".join(tagged_text) + + +class ThaiNameTagger: + """Tag named entities with the WangchanBERTa pipeline.""" + + dataset_name: str + grouped_entities: bool + classify_tokens: TokenClassificationPipeline + json_ner: list[dict[str, str]] + output: str + sent_ner: list[tuple[str, str]] + + def __init__( + self, dataset_name: str = "thainer", grouped_entities: bool = True + ) -> None: + """ + Initialize a named entity tagger in IOB format. + + Use WangchanBERTa from the VISTEC-depa AI Research Institute + of Thailand. + + :param str dataset_name: dataset the model is fine-tuned on + * *thainer* - ThaiNER dataset (default) + :param bool grouped_entities: whether to group word pieces of the + same entity + """ + from transformers import pipeline + + self.dataset_name = dataset_name + self.grouped_entities = grouped_entities + self.classify_tokens = pipeline( + task="ner", + tokenizer=_get_tokenizer(), + model=f"airesearch/{_model_name}", + revision=f"finetuned@{self.dataset_name}-ner", + ignore_labels=[], + grouped_entities=self.grouped_entities, + ) + + def _iob(self, tag: str) -> str: + if tag != "O": + return "B-" + tag + return "O" + + def _clear_tag(self, tag: str) -> str: + return tag.replace("B-", "").replace("I-", "") + + def _prepare_ner( + self, entities: list[dict[str, str]] + ) -> list[tuple[str, str]]: + if self.grouped_entities and self.dataset_name == "thainer": + return [ + ( + item["word"].replace("<_>", " ").replace("▁", ""), + self._iob(item["entity_group"]), + ) + for item in entities + ] + if self.dataset_name == "thainer": + return [ + ( + item["word"].replace("<_>", " ").replace("▁", ""), + item["entity"], + ) + for item in entities + if item["word"] != "▁" + ] + return [ + ( + item["word"].replace("<_>", " ").replace("▁", ""), + item["entity"].replace("_", "-").replace("E-", "I-"), + ) + for item in entities + ] + + def _fix_consecutive_begin_tags(self) -> None: + for idx in range(1, len(self.sent_ner)): + word, ner = self.sent_ner[idx] + previous_ner = self.sent_ner[idx - 1][1] + if ner.startswith("B-") and self._clear_tag( + ner + ) == self._clear_tag(previous_ner): + self.sent_ner[idx] = (word, ner.replace("B-", "I-")) + + def get_ner( + self, text: str, pos: bool = False, tag: bool = False + ) -> Union[list[tuple[str, str]], str]: + """ + Tag named entities in Thai text. + + Use WangchanBERTa from the VISTEC-depa AI Research Institute + of Thailand. + + :param str text: Thai text to be tagged + :param bool pos: whether to request part-of-speech output; the + model does not support it, so a warning is raised + :param bool tag: whether to return HTML-like tags instead of tuples + :return: tagged words, or an HTML-like tagged string if *tag* is True + :rtype: Union[list[tuple[str, str]], str] + """ + if pos: + warnings.warn( + "This model does not support POS tag output.", + UserWarning, + stacklevel=2, + ) + text = text.replace(" ", "<_>") + self.json_ner: list[dict[str, str]] = self.classify_tokens(text) + self.output: str = "" + self.sent_ner = self._prepare_ner(self.json_ner) + if ( + self.sent_ner + and self.sent_ner[0][0] == "" + and len(self.sent_ner) > 1 + ): + self.sent_ner = self.sent_ner[1:] + self._fix_consecutive_begin_tags() + if tag: + return _format_ner_tags(self.sent_ner) + return self.sent_ner + + +class NamedEntityRecognition: + """Recognize Thai named entities with a WangchanBERTa model.""" + + tokenizer: PreTrainedTokenizerBase + model: PreTrainedModel + + def __init__( + self, + model: str = "pythainlp/thainer-corpus-v2-base-model", + revision: Optional[str] = None, + ) -> None: + """ + Initialize a named entity recognizer. + + Use a model pretrained with WangchanBERTa. + + :param str model: pretrained model name + :param Optional[str] revision: git revision ID (branch, tag, or + commit hash). Pin to a full commit hash for secure downloads. + """ + from transformers import AutoModelForTokenClassification, AutoTokenizer + + self.tokenizer: PreTrainedTokenizerBase = ( + AutoTokenizer.from_pretrained(model, revision=revision) + ) + self.model: PreTrainedModel = ( + AutoModelForTokenClassification.from_pretrained( + model, revision=revision + ) + ) + + def _fix_span_error( + self, words: list[int], ner: list[str] + ) -> list[tuple[str, str]]: + _ner = [] + _ner = ner + _new_tag = [] + for i, j in zip(words, _ner): + i_decoded = self.tokenizer.decode(i) + tag = "O" if i_decoded.isspace() and j.startswith("B-") else j + if i_decoded in ("", "", ""): + continue + if i_decoded == "<_>": + i_decoded = " " + _new_tag.append((i_decoded, tag)) + return _new_tag + + def get_ner( + self, text: str, pos: bool = False, tag: bool = False + ) -> Union[list[tuple[str, str]], str]: + """ + Tag named entities in Thai text. + + :param str text: Thai text to be tagged + :param bool pos: whether to request part-of-speech output; the + model does not support it, so a warning is raised + :param bool tag: whether to return HTML-like tags instead of tuples + :return: tagged words, or an HTML-like tagged string if *tag* is True + :rtype: Union[list[tuple[str, str]], str] + """ + import torch + + if pos: + warnings.warn( + "This model does not support POS tag output.", + UserWarning, + stacklevel=2, + ) + words_token = word_tokenize(text.replace(" ", "<_>")) + inputs = self.tokenizer( + words_token, is_split_into_words=True, return_tensors="pt" + ) + ids = inputs["input_ids"] + mask = inputs["attention_mask"] + # forward pass + outputs = self.model(ids, attention_mask=mask) + logits = outputs[0] + predictions = torch.argmax(logits, dim=2) + predicted_token_class = [ + self.model.config.id2label[t.item()] for t in predictions[0] + ] + ner_tag = self._fix_span_error( + inputs["input_ids"][0], predicted_token_class + ) + if tag: + return _format_ner_tags(ner_tag) + return ner_tag + + +def segment(text: str) -> list[str]: + """ + Tokenize text into subwords with the WangchanBERTa tokenizer. + + :param str text: text to be tokenized + :return: list of subwords + :rtype: list[str] + """ + if not text or not isinstance(text, str): + return [] + + return cast("list[str]", _get_tokenizer().tokenize(text)) diff --git a/pythainlp/phayathaibert/__init__.py b/pythainlp/phayathaibert/__init__.py index 8f664cb8c..74204d54b 100644 --- a/pythainlp/phayathaibert/__init__.py +++ b/pythainlp/phayathaibert/__init__.py @@ -1,20 +1,35 @@ # SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project # SPDX-FileType: SOURCE # SPDX-License-Identifier: Apache-2.0 -"""PhayaThaiBERT.""" +""" +Deprecated. Use :mod:`pythainlp.lm.phayathaibert` instead. -__all__: list[str] = [ - "NamedEntityTagger", - "PartOfSpeechTagger", - "ThaiTextAugmenter", - "ThaiTextProcessor", - "segment", -] +.. deprecated:: 5.3.9 + :mod:`pythainlp.phayathaibert` has moved to :mod:`pythainlp.lm.phayathaibert`. +""" -from pythainlp.phayathaibert.core import ( +from __future__ import annotations + +from pythainlp.lm.phayathaibert import ( NamedEntityTagger, PartOfSpeechTagger, ThaiTextAugmenter, ThaiTextProcessor, segment, ) +from pythainlp.tools import warn_deprecation + +warn_deprecation( + "pythainlp.phayathaibert", + "pythainlp.lm.phayathaibert", + "5.3.9", + "6.0", +) + +__all__: list[str] = [ + "NamedEntityTagger", + "PartOfSpeechTagger", + "ThaiTextAugmenter", + "ThaiTextProcessor", + "segment", +] diff --git a/pythainlp/phayathaibert/core.py b/pythainlp/phayathaibert/core.py index 794d7e0c0..48875bafb 100644 --- a/pythainlp/phayathaibert/core.py +++ b/pythainlp/phayathaibert/core.py @@ -1,573 +1,35 @@ # SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project # SPDX-FileType: SOURCE # SPDX-License-Identifier: Apache-2.0 -"""PhayaThaiBERT text processing, augmentation, and tagging.""" +""" +Deprecated. Use :mod:`pythainlp.lm.phayathaibert.core` instead. -from __future__ import annotations - -import random -import re -import warnings -from typing import TYPE_CHECKING, Optional, Union, cast +.. deprecated:: 5.3.9 + :mod:`pythainlp.phayathaibert.core` has moved to :mod:`pythainlp.lm.phayathaibert.core`. +""" -if TYPE_CHECKING: - from collections.abc import Callable - - from transformers import ( - AutoModelForMaskedLM, - AutoModelForTokenClassification, - CamembertTokenizer, - Pipeline, - PreTrainedTokenizerBase, - ) +from __future__ import annotations -from transformers import ( - CamembertTokenizer, +from pythainlp.lm.phayathaibert.core import ( + NamedEntityTagger, + PartOfSpeechTagger, + ThaiTextAugmenter, + ThaiTextProcessor, + segment, ) +from pythainlp.tools import warn_deprecation -from pythainlp.tokenize import word_tokenize - -_PAT_URL: str = r"(http|ftp|https)://([\w_-]+(?:(?:\.[\w_-]+)+))([\w.,@?^=%&:/~+#-]*[\w@?^=%&/~+#-])?" - -_model_name: str = "clicknext/phayathaibert" -_tokenizer: CamembertTokenizer = CamembertTokenizer.from_pretrained( - _model_name # nosec B615 +warn_deprecation( + "pythainlp.phayathaibert.core", + "pythainlp.lm.phayathaibert.core", + "5.3.9", + "6.0", ) - -class ThaiTextProcessor: - """Preprocess Thai text for PhayaThaiBERT.""" - - def __init__(self) -> None: - """Initialize the text processor.""" - ( - self._TK_UNK, - self._TK_REP, - self._TK_WREP, - self._TK_URL, - self._TK_END, - ) = ["", "", "", "", ""] - self.SPACE_SPECIAL_TOKEN: str = "<_>" # noqa: S105 - - def replace_url(self, text: str) -> str: - """ - Replace URLs in text with the URL token. - - See https://stackoverflow.com/a/6041965 - - :param str text: text to be processed - :return: text with URLs replaced - :rtype: str - - :Example: - - >>> replace_url("go to https://github.com") - 'go to ' - """ - return re.sub(_PAT_URL, self._TK_URL, text) - - def rm_brackets(self, text: str) -> str: - """ - Remove empty brackets and artifacts within brackets from text. - - :param str text: text to be processed - :return: text with useless brackets removed - :rtype: str - - :Example: - - >>> rm_brackets("hey() whats[;] up{*&} man(hey)") - 'hey whats up man(hey)' - """ - # remove empty brackets - new_line = re.sub(r"\(\)", "", text) - new_line = re.sub(r"\{\}", "", new_line) - new_line = re.sub(r"\[\]", "", new_line) - # brackets with only punctuations - new_line = re.sub(r"\([^a-zA-Z0-9ก-๙]+\)", "", new_line) - new_line = re.sub(r"\{[^a-zA-Z0-9ก-๙]+\}", "", new_line) - new_line = re.sub(r"\[[^a-zA-Z0-9ก-๙]+\]", "", new_line) - # artifiacts after ( - new_line = re.sub( - r"(?<=\()[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line - ) - new_line = re.sub( - r"(?<=\{)[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line - ) - new_line = re.sub( - r"(?<=\[)[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line - ) - # artifacts before ) - new_line = re.sub( - r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\))", "", new_line - ) - new_line = re.sub( - r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\})", "", new_line - ) - new_line = re.sub( - r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\])", "", new_line - ) - return new_line - - def replace_newlines(self, text: str) -> str: - """ - Replace newlines in text with spaces. - - :param str text: text to be processed - :return: text with newlines replaced with spaces - :rtype: str - - :Example: - - >>> rm_useless_spaces("hey whats\n\nup") - hey whats up - """ - return re.sub(r"[\n]", " ", text.strip()) - - def rm_useless_spaces(self, text: str) -> str: - """ - Collapse repeated spaces in text (code from `fastai`). - - :param str text: text to be processed - :return: text with repeated spaces reduced to one - :rtype: str - - :Example: - - >>> rm_useless_spaces("oh no") - oh no - """ - return re.sub(" {2,}", " ", text) - - def replace_spaces(self, text: str, space_token: str = "<_>") -> str: # noqa: S107 # nosec B107 - """ - Replace spaces in text with a space token. - - :param str text: text to be processed - :param str space_token: token to replace spaces with - :return: text with spaces replaced with the space token - :rtype: str - - :Example: - - >>> replace_spaces("oh no") - oh_no - """ - return re.sub(" ", space_token, text) - - def replace_rep_after(self, text: str) -> str: - """ - Remove character repetitions in text. - - :param str text: text to be processed - :return: text with repeated characters removed - :rtype: str - - :Example: - - >>> text = "กาาาาาาา" - >>> replace_rep_after(text) - 'กา' - """ - - def _replace_rep(m: re.Match[str]) -> str: - c, cc = m.groups() - return f"{c}" - - re_rep = re.compile(r"(\S)(\1{3,})") - return re_rep.sub(_replace_rep, text) - - def replace_wrep_post(self, toks: list[str]) -> list[str]: - """ - Remove repeated words after tokenization. - - The `replace_wrep` function of `fastai` does not work well - with Thai. - - :param list[str] toks: list of words - :return: list of words with repeated words removed - :rtype: list[str] - - :Example: - - >>> toks = ["กา", "น้ำ", "น้ำ", "น้ำ", "น้ำ"] - >>> replace_wrep_post(toks) - ['กา', 'น้ำ'] - """ - previous_word = "" - rep_count = 0 - res = [] - for current_word in [*toks, self._TK_END]: - if current_word == previous_word: - rep_count += 1 - elif (current_word != previous_word) & (rep_count > 0): - res += [previous_word] - rep_count = 0 - else: - res.append(previous_word) - previous_word = current_word - - return res[1:] - - def remove_space(self, toks: list[str]) -> list[str]: - """ - Remove spaces from a list of words, for bag-of-words models. - - :param list[str] toks: list of words - :return: list of words with space tokens (" ") filtered out - :rtype: list[str] - - :Example: - - >>> toks = ["ฉัน", "เดิน", " ", "กลับ", "บ้าน"] - >>> remove_space(toks) - ['ฉัน', 'เดิน', 'กลับ', 'บ้าน'] - """ - res = [] - for t in toks: - stripped = t.strip() - if stripped: - res.append(stripped) - - return res - - # combine them together - def preprocess( - self, - text: str, - pre_rules: Optional[list[Callable[..., str]]] = None, - tok_func: Callable[..., list[str]] = word_tokenize, - ) -> str: - """ - Preprocess text: apply the rules, then tokenize and join. - - :param str text: text to be preprocessed - :param Optional[list[Callable[..., str]]] pre_rules: rules to - apply in order after lowercasing. If None, use the text - cleaning methods of this class. - :param Callable[..., list[str]] tok_func: function to tokenize - text - :return: preprocessed text - :rtype: str - """ - if pre_rules is None: - pre_rules = [ - self.rm_brackets, - self.replace_newlines, - self.rm_useless_spaces, - self.replace_spaces, - self.replace_rep_after, - ] - text = text.lower() - for rule in pre_rules: - text = rule(text) - toks = tok_func(text) - - return "".join(toks) - - -class ThaiTextAugmenter: - """Augment Thai text using PhayaThaiBERT masked language model.""" - - def __init__(self) -> None: - """Initialize the text augmenter.""" - from transformers import ( - AutoModelForMaskedLM, - AutoTokenizer, - pipeline, - ) - - self.tokenizer: PreTrainedTokenizerBase = ( - AutoTokenizer.from_pretrained(_model_name) # nosec B615 - ) - self.model_for_masked_lm: AutoModelForMaskedLM = ( - AutoModelForMaskedLM.from_pretrained(_model_name) # nosec B615 - ) - self.model: Pipeline = pipeline( # transformers.Pipeline - "fill-mask", - tokenizer=self.tokenizer, - model=self.model_for_masked_lm, - ) - self.processor: ThaiTextProcessor = ThaiTextProcessor() - - def generate( - self, - sample_text: str, - word_rank: int, - max_length: int = 3, - sample: bool = False, - ) -> str: - """ - Generate text from PhayaThaiBERT. - - :param str sample_text: text to continue from - :param int word_rank: rank of the predicted word to select - :param int max_length: number of words to generate - :param bool sample: whether to select a random word among the - top five predictions - :return: generated text - :rtype: str - """ - sample_txt = sample_text - final_text = "" - for _ in range(max_length): - input_text = self.processor.preprocess(sample_txt) - if sample: - # Non-cryptographic use, pseudo-random generator is acceptable here - random_word_idx = random.randint(0, 4) # noqa: S311 # nosec B311 # NOSONAR - output = self.model(input_text)[random_word_idx]["sequence"] - else: - output = self.model(input_text)[word_rank]["sequence"] - sample_txt = output + "" - final_text = sample_txt - - gen_txt = re.sub("", "", final_text) - - return gen_txt - - def augment( - self, - text: str, - num_augs: int = 3, - sample: bool = False, - ) -> list[str]: - """ - Augment text with PhayaThaiBERT. - - :param str text: Thai text to be augmented - :param int num_augs: number of augmented texts to return - :param bool sample: whether to sample words randomly, for more - word diversity - :return: list of augmented texts - :rtype: list[str] - :raises ValueError: if **num_augs** exceeds the limit of five - - :Example: - - >>> from pythainlp.augment.lm import ( - ... ThaiTextAugmenter, - ... ) # doctest: +SKIP - - >>> aug = ThaiTextAugmenter() # doctest: +SKIP - >>> aug.augment("ช้างมีทั้งหมด 50 ตัว บน", num_args=5) # doctest: +SKIP - - ['ช้างมีทั้งหมด 50 ตัว บนโลกใบนี้ครับ.', - 'ช้างมีทั้งหมด 50 ตัว บนพื้นดินครับ...', - 'ช้างมีทั้งหมด 50 ตัว บนท้องฟ้าครับ...', - 'ช้างมีทั้งหมด 50 ตัว บนดวงจันทร์.‼', - 'ช้างมีทั้งหมด 50 ตัว บนเขาค่ะ😁'] - """ - MAX_NUM_AUGS = 5 - augment_list = [] - - if num_augs <= MAX_NUM_AUGS: - for rank in range(num_augs): - gen_text = self.generate( - text, - rank, - sample=sample, - ) - processed_text = re.sub( - "<_>", " ", self.processor.preprocess(gen_text) - ) - augment_list.append(processed_text) - else: - raise ValueError( - f"augmentation of more than {num_augs} is exceeded \ - the default limit: {MAX_NUM_AUGS}" - ) - - return augment_list - - -class PartOfSpeechTagger: - """Tag part-of-speech (POS) in Thai text using PhayaThaiBERT.""" - - def __init__( - self, - model: str = "lunarlist/pos_thai_phayathai", - revision: Optional[str] = None, - ) -> None: - """ - Initialize the POS tagger. - - :param str model: name of the model to load - :param Optional[str] revision: git revision id (branch, tag, or - commit hash) - """ - # Load model directly - from transformers import ( - AutoModelForTokenClassification, - AutoTokenizer, - ) - - self.tokenizer: PreTrainedTokenizerBase = ( - AutoTokenizer.from_pretrained(model, revision=revision) - ) - self.model: AutoModelForTokenClassification = ( - AutoModelForTokenClassification.from_pretrained( - model, revision=revision - ) - ) - - def get_tag( - self, sentence: str, strategy: str = "simple" - ) -> list[list[tuple[str, str]]]: - """ - Tag text with part-of-speech (POS) tags. - - :param str sentence: text to be tagged - :param str strategy: aggregation strategy of the token - classification pipeline - :return: list of lists of tuples (word, POS tag) - :rtype: list[list[tuple[str, str]]] - - :Example: - - Label POS for the given text: - - >>> from pythainlp.phayathaibert.core import ( - ... PartOfSpeechTagger, - ... ) # doctest: +SKIP - - >>> tagger = PartOfSpeechTagger() # doctest: +SKIP - >>> tagger.get_tag("แมวทำอะไรตอนห้าโมงเช้า") # doctest: +SKIP - [[('แมว', 'NOUN'), ('ทําอะไร', 'VERB'), ('ตอนห้าโมงเช้า', 'NOUN')]] - """ - from transformers import TokenClassificationPipeline - - pipeline = TokenClassificationPipeline( - model=self.model, - tokenizer=self.tokenizer, - aggregation_strategy=strategy, - ) - outputs = pipeline(sentence) - word_tags = [[(tag["word"], tag["entity_group"]) for tag in outputs]] - - return word_tags - - -class NamedEntityTagger: - """Tag named entities in Thai text using PhayaThaiBERT.""" - - def __init__( - self, - model: str = "Pavarissy/phayathaibert-thainer", - revision: Optional[str] = None, - ) -> None: - """ - Initialize the named entity tagger. - - :param str model: name of the model to load - :param Optional[str] revision: git revision id (branch, tag, or - commit hash) - """ - from transformers import ( - AutoModelForTokenClassification, - AutoTokenizer, - ) - - self.tokenizer: PreTrainedTokenizerBase = ( - AutoTokenizer.from_pretrained(model, revision=revision) - ) - self.model: AutoModelForTokenClassification = ( - AutoModelForTokenClassification.from_pretrained( - model, revision=revision - ) - ) - - def get_ner( - self, - text: str, - tag: bool = False, - pos: bool = False, - strategy: str = "simple", - ) -> Union[list[tuple[str, str]], list[tuple[str, str, str]], str]: - """ - Tag named entities in text. - - :param str text: Thai text to be tagged - :param bool tag: return HTML-like tags in a string instead of a - list of tuples - :param bool pos: output part-of-speech tags. This model does not - support them (use :class:`PartOfSpeechTagger` instead), so - a warning is raised. - :param str strategy: aggregation strategy of the token - classification pipeline - :return: list of tuples (word, named entity tag), or a string with - HTML-like tags if **tag** is True - :rtype: Union[list[tuple[str, str]], list[tuple[str, str, str]], str] - - :Example: - - >>> from pythainlp.phayathaibert.core import NamedEntityTagger - >>> - >>> tagger = NamedEntityTagger() - >>> tagger.get_ner("ทดสอบนายปวริศ เรืองจุติโพธิ์พานจากประเทศไทย") - [('นายปวริศ เรืองจุติโพธิ์พานจากประเทศไทย', 'PERSON'), - ('จาก', 'LOCATION'), - ('ประเทศไทย', 'LOCATION')] - >>> ner.tag("ทดสอบนายปวริศ เรืองจุติโพธิ์พานจากประเทศไทย", tag=True) - 'ทดสอบนายปวริศ เรืองจุติโพธิ์พานจากประเทศไทย' - """ - from transformers import TokenClassificationPipeline - - if pos: - warnings.warn( - "This model does not support POS tag output.", - UserWarning, - stacklevel=2, - ) - - sample_output = [] - tag_text_list = [] - current_pos = 0 - pipeline = TokenClassificationPipeline( - model=self.model, - tokenizer=self.tokenizer, - aggregation_strategy=strategy, - ) - outputs = pipeline(text) - - for token in outputs: - ner_tag = token["entity_group"] - begin_pos, end_pos = token["start"], token["end"] - if current_pos == 0: - text_tag = ( - text[:begin_pos] - + f"<{ner_tag}>" - + text[begin_pos:end_pos] - + f"" - ) - else: - text_tag = ( - text[current_pos:begin_pos] - + f"<{ner_tag}>" - + text[begin_pos:end_pos] - + f"" - ) - tag_text_list.append(text_tag) - sample_output.append((token["word"], token["entity_group"])) - current_pos = end_pos - - if tag: - return str("".join(tag_text_list)) - - return sample_output - - -def segment(sentence: str) -> list[str]: - """ - Tokenize text into subwords with the PhayaThaiBERT tokenizer. - - The tokenizer is the sentencepiece model of WangchanBERTa, with - vocabulary expansion. - - :param str sentence: text to be tokenized - :return: list of subwords - :rtype: list[str] - """ - if not sentence or not isinstance(sentence, str): - return [] - - return cast("list[str]", _tokenizer.tokenize(sentence)) +__all__: list[str] = [ + "NamedEntityTagger", + "PartOfSpeechTagger", + "ThaiTextAugmenter", + "ThaiTextProcessor", + "segment", +] diff --git a/pythainlp/tag/named_entity.py b/pythainlp/tag/named_entity.py index 2a3736850..e6a0bca2c 100644 --- a/pythainlp/tag/named_entity.py +++ b/pythainlp/tag/named_entity.py @@ -10,15 +10,15 @@ if TYPE_CHECKING: from types import ModuleType - from pythainlp.phayathaibert.core import NamedEntityTagger - from pythainlp.tag.thai_nner import ThaiNNER - from pythainlp.tag.thainer import ThaiNameTagger - from pythainlp.wangchanberta.core import ( + from pythainlp.lm.phayathaibert.core import NamedEntityTagger + from pythainlp.lm.wangchanberta.core import ( NamedEntityRecognition, ) - from pythainlp.wangchanberta.core import ( + from pythainlp.lm.wangchanberta.core import ( ThaiNameTagger as WangchanbertaThaiNameTagger, ) + from pythainlp.tag.thai_nner import ThaiNNER + from pythainlp.tag.thainer import ThaiNameTagger class EntitySpan(TypedDict): @@ -105,20 +105,20 @@ def load_engine(self, engine: str, corpus: str) -> None: self.engine = ThaiNameTagger() elif engine == "thainer-v2": - from pythainlp.wangchanberta import NamedEntityRecognition + from pythainlp.lm.wangchanberta import NamedEntityRecognition self.engine = NamedEntityRecognition( model="pythainlp/thainer-corpus-v2-base-model" ) elif engine == "wangchanberta": - from pythainlp.wangchanberta import ( + from pythainlp.lm.wangchanberta import ( ThaiNameTagger as WangchanbertaThaiNameTagger, ) self.engine = WangchanbertaThaiNameTagger(dataset_name=corpus) elif corpus == "thainer-v2": if engine == "phayathaibert": - from pythainlp.phayathaibert.core import NamedEntityTagger + from pythainlp.lm.phayathaibert.core import NamedEntityTagger self.engine = NamedEntityTagger() diff --git a/pythainlp/tokenize/_registry.py b/pythainlp/tokenize/_registry.py index 6fdbadcd3..9e01c5120 100644 --- a/pythainlp/tokenize/_registry.py +++ b/pythainlp/tokenize/_registry.py @@ -279,7 +279,7 @@ def _etcc(text: str) -> list[str]: def _wangchanberta(text: str) -> list[str]: - from pythainlp.wangchanberta import segment + from pythainlp.lm.wangchanberta import segment return segment(text) @@ -316,7 +316,7 @@ def _han_solo(text: str) -> list[str]: def _phayathai(text: str) -> list[str]: - from pythainlp.phayathaibert import segment + from pythainlp.lm.phayathaibert import segment return segment(text) diff --git a/pythainlp/ulmfit/__init__.py b/pythainlp/ulmfit/__init__.py index e4239612b..7fe199f50 100644 --- a/pythainlp/ulmfit/__init__.py +++ b/pythainlp/ulmfit/__init__.py @@ -2,17 +2,47 @@ # SPDX-FileType: SOURCE # SPDX-License-Identifier: Apache-2.0 """ -Universal Language Model Fine-tuning for Text Classification (ULMFiT). +Deprecated. Use :mod:`pythainlp.lm.ulmfit` instead. -Code by Charin Polpanumas -https://github.com/cstorm125/thai2fit/ +.. deprecated:: 5.3.9 + :mod:`pythainlp.ulmfit` has moved to :mod:`pythainlp.lm.ulmfit`. +""" -Some pre-processing functions are from fastai (Apache 2.0) -https://github.com/fastai/fastai/blob/master/fastai/text/transform.py +from __future__ import annotations -Universal Language Model Fine-tuning for Text Classification -https://arxiv.org/abs/1801.06146 -""" +from pythainlp.lm.ulmfit import ( + THWIKI_LSTM, + ThaiTokenizer, + document_vector, + fix_html, + get_thwiki_lstm, + lowercase_all, + merge_wgts, + post_rules_th, + post_rules_th_sparse, + pre_rules_th, + pre_rules_th_sparse, + process_thai, + remove_space, + replace_rep_after, + replace_rep_nonum, + replace_url, + replace_wrep_post, + replace_wrep_post_nonum, + rm_brackets, + rm_useless_newlines, + rm_useless_spaces, + spec_add_spaces, + ungroup_emoji, +) +from pythainlp.tools import warn_deprecation + +warn_deprecation( + "pythainlp.ulmfit", + "pythainlp.lm.ulmfit", + "5.3.9", + "6.0", +) __all__: list[str] = [ "THWIKI_LSTM", @@ -39,31 +69,3 @@ "spec_add_spaces", "ungroup_emoji", ] - -from pythainlp.ulmfit.core import ( - THWIKI_LSTM, - document_vector, - get_thwiki_lstm, - merge_wgts, - post_rules_th, - post_rules_th_sparse, - pre_rules_th, - pre_rules_th_sparse, - process_thai, -) -from pythainlp.ulmfit.preprocess import ( - fix_html, - lowercase_all, - remove_space, - replace_rep_after, - replace_rep_nonum, - replace_url, - replace_wrep_post, - replace_wrep_post_nonum, - rm_brackets, - rm_useless_newlines, - rm_useless_spaces, - spec_add_spaces, - ungroup_emoji, -) -from pythainlp.ulmfit.tokenizer import ThaiTokenizer diff --git a/pythainlp/ulmfit/core.py b/pythainlp/ulmfit/core.py index f5aef2038..a41adca78 100644 --- a/pythainlp/ulmfit/core.py +++ b/pythainlp/ulmfit/core.py @@ -1,327 +1,43 @@ # SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project # SPDX-FileType: SOURCE # SPDX-License-Identifier: Apache-2.0 -"""Universal Language Model Fine-tuning for Text Classification (ULMFiT).""" +""" +Deprecated. Use :mod:`pythainlp.lm.ulmfit.core` instead. -from __future__ import annotations - -import collections -from typing import TYPE_CHECKING, Any, Optional, cast - -import torch - -if TYPE_CHECKING: - from collections.abc import Callable, Collection +.. deprecated:: 5.3.9 + :mod:`pythainlp.ulmfit.core` has moved to :mod:`pythainlp.lm.ulmfit.core`. +""" - import numpy as np - from numpy.typing import NDArray +from __future__ import annotations -from pythainlp.corpus import get_corpus_path -from pythainlp.tokenize import thai2fit_tokenizer -from pythainlp.ulmfit.preprocess import ( - fix_html, - lowercase_all, - remove_space, - replace_rep_after, - replace_rep_nonum, - replace_url, - replace_wrep_post, - replace_wrep_post_nonum, - rm_brackets, - rm_useless_newlines, - rm_useless_spaces, - spec_add_spaces, - ungroup_emoji, +from pythainlp.lm.ulmfit.core import ( + THWIKI_LSTM, + document_vector, + get_thwiki_lstm, + merge_wgts, + post_rules_th, + post_rules_th_sparse, + pre_rules_th, + pre_rules_th_sparse, + process_thai, ) -from pythainlp.util import reorder_vowels +from pythainlp.tools import warn_deprecation -device: torch.device = torch.device( - "cuda" if torch.cuda.is_available() else "cpu" +warn_deprecation( + "pythainlp.ulmfit.core", + "pythainlp.lm.ulmfit.core", + "5.3.9", + "6.0", ) -_MODEL_NAME_LSTM: str = "wiki_lm_lstm" -_ITOS_NAME_LSTM: str = "wiki_itos_lstm" - - -# Pretrained model paths -# Note: These may be None if corpus is not downloaded. -# Access via get_thwiki_lstm() for proper validation or use directly -# if you've already verified the corpus is downloaded. -THWIKI_LSTM: dict[str, Optional[str]] = { - "wgts_fname": get_corpus_path(_MODEL_NAME_LSTM), - "itos_fname": get_corpus_path(_ITOS_NAME_LSTM), - "json_itos_fname": get_corpus_path("wiki_itos_lstm_json"), -} - - -def get_thwiki_lstm() -> dict[str, str]: - """ - Return the Thai Wikipedia (THWIKI) LSTM model paths, with validation. - - :return: dictionary with ``wgts_fname`` and ``itos_fname`` keys - :rtype: dict[str, str] - :raises FileNotFoundError: if corpus files are not found - """ - wgts_fname = THWIKI_LSTM["wgts_fname"] - itos_fname = THWIKI_LSTM["itos_fname"] - - if not wgts_fname or not itos_fname: - raise FileNotFoundError( - "corpus-not-found names=['wiki_lm_lstm', 'wiki_itos_lstm']\n" - " ULMFiT model files not found.\n" - " Python: pythainlp.corpus.download('wiki_lm_lstm')\n" - " CLI: thainlp data get wiki_lm_lstm\n" - " Python: pythainlp.corpus.download('wiki_itos_lstm')\n" - " CLI: thainlp data get wiki_itos_lstm" - ) - - return {"wgts_fname": wgts_fname, "itos_fname": itos_fname} - - -# Preprocessing rules for Thai text -# dense features -pre_rules_th: list[Callable[[str], str]] = [ - replace_rep_after, - fix_html, - reorder_vowels, - spec_add_spaces, - rm_useless_spaces, - rm_useless_newlines, - rm_brackets, - replace_url, -] -post_rules_th: list[Callable[[Collection[str]], list[str]]] = [ - replace_wrep_post, - ungroup_emoji, - lowercase_all, -] - -# sparse features -pre_rules_th_sparse: list[Callable[[str], str]] = [ - *pre_rules_th[1:], - replace_rep_nonum, +__all__: list[str] = [ + "THWIKI_LSTM", + "document_vector", + "get_thwiki_lstm", + "merge_wgts", + "post_rules_th", + "post_rules_th_sparse", + "pre_rules_th", + "pre_rules_th_sparse", + "process_thai", ] -post_rules_th_sparse: list[Callable[[Collection[str]], list[str]]] = [ - *post_rules_th[1:], - replace_wrep_post_nonum, - remove_space, -] - - -def process_thai( - text: str, - pre_rules: Optional[Collection[Callable[[str], str]]] = None, - tok_func: Optional[Callable[[str], list[str]]] = None, - post_rules: Optional[Collection[Callable[[list[str]], list[str]]]] = None, -) -> list[str]: - """ - Process Thai text for models, with sparse features as default. - - :param str text: text to be processed - :param Optional[Collection[Callable[[str], str]]] pre_rules: rules to - apply before tokenization. If None, use the default sparse - pre-rules. - :param Optional[Callable[[str], list[str]]] tok_func: function to - tokenize text. If None, use the ``word_tokenize`` method of - :func:`pythainlp.tokenize.thai2fit_tokenizer`. - :param Optional[Collection[Callable[[list[str]], list[str]]]] post_rules: - rules to apply after tokenization. If None, use the default sparse - post-rules. - :return: list of processed words - :rtype: list[str] - - :Note: - - The default **pre-rules** consist of :func:`fix_html`, - :func:`pythainlp.util.normalize`, - :func:`spec_add_spaces`, - :func:`rm_useless_spaces`, - :func:`rm_useless_newlines`, - :func:`rm_brackets`, - and :func:`replace_rep_nonum`. - - - The default **post-rules** consist of :func:`ungroup_emoji`, - :func:`lowercase_all`, :func:`replace_wrep_post_nonum`, - and :func:`remove_space`. - - :Example: - - 1. Use default pre-rules and post-rules: - - >>> from pythainlp.ulmfit import process_thai - >>> text = "บ้านนนนน () อยู่นานนานนาน 😂🤣😃😄😅 PyThaiNLP amp; " - >>> process_thai(text) - ['บ้าน', 'xxrep', ' ', 'อยู่', 'xxwrep', 'นาน', '😂', '🤣', - '😃', '😄', '😅', 'pythainlp', '&'] - - 2. Modify pre_rules and post_rules arguments with - rules provided in :mod:`pythainlp.ulmfit`: - - >>> from pythainlp.ulmfit import ( - process_thai, - replace_rep_after, - fix_html, - ungroup_emoji, - replace_wrep_post, - remove_space) - >>> - >>> text = "บ้านนนนน () อยู่นานนานนาน 😂🤣😃😄😅 PyThaiNLP amp; " - >>> process_thai(text, - pre_rules=[replace_rep_after, fix_html], - post_rules=[ungroup_emoji, - replace_wrep_post, - remove_space] - ) - ['บ้าน', 'xxrep', '5', '()', 'อยู่', 'xxwrep', '2', 'นาน', '😂', '🤣', - '😃', '😄', '😅', 'PyThaiNLP', '&'] - - - """ - processed_text = text - if pre_rules is None: - pre_rules = pre_rules_th_sparse - if post_rules is None: - post_rules = cast( - "Collection[Callable[[list[str]], list[str]]]", - post_rules_th_sparse, - ) - - if tok_func is None: - tok_func = thai2fit_tokenizer().word_tokenize - - for pre_rule in pre_rules: - processed_text = pre_rule(processed_text) - tokens = tok_func(processed_text) - for post_rule in post_rules: - tokens = post_rule(tokens) - - return tokens - - -def document_vector( - text: str, learn: Any, data: Any, agg: str = "mean" -) -> NDArray[np.float32]: - """ - Vectorize Thai text into a 400-dimension vector. - - Use a :class:`fastai` language model and data bunch. - Aggregate word vectors by mean or sum. - - :param str text: text to be vectorized - :param learn: :class:`fastai` language model learner - (``fastai.basic_train.Learner``) - :param data: :class:`fastai` data bunch - (``fastai.basic_data.DataBunch``) - :param str agg: aggregation method - - * *mean* - average of word vectors (default) - * *sum* - sum of word vectors - - :return: document vector of shape ``(1, 400)`` - :rtype: numpy.typing.NDArray[numpy.float32] - - :Example: - - >>> from pythainlp.ulmfit import document_vector - >>> from fastai.text import load_data, language_model_learner, AWD_LSTM - >>> - >>> # Load Data Bunch - >>> data = load_data(MODEL_PATH, "thwiki_lm_data.pkl") - >>> - >>> # Initialize language_model_learner - >>> config = dict(emb_sz=400, n_hid=1550, n_layers=4, pad_token=1, - qrnn=False, tie_weights=True, out_bias=True, output_p=0.25, - hidden_p=0.1, input_p=0.2, embed_p=0.02, weight_p=0.15) - >>> trn_args = dict(drop_mult=0.9, clip=0.12, alpha=2, beta=1) - >>> learn = language_model_learner(data, AWD_LSTM, config=config, - pretrained=False, **trn_args) - >>> document_vector("วันนี้วันดีปีใหม่", learn, data) - - :See Also: - * A notebook showing how to train a `ulmfit` language model and - how to use it: `Jupyter Notebook`_ - - .. _Jupyter Notebook: - https://github.com/cstorm125/thai2fit/blob/master/thwiki_lm/word2vec_examples.ipynb - - """ - s = thai2fit_tokenizer().word_tokenize(text) - t = torch.tensor(data.vocab.numericalize(s), requires_grad=False).to( - device - ) - m = learn.model[0].encoder.to(device) - res = m(t).cpu().detach().numpy().astype("float32", copy=False) - if agg == "mean": - res = res.mean(0, dtype="float32") - elif agg == "sum": - res = res.sum(0, dtype="float32") - else: - raise ValueError("Aggregate by mean or sum") - - return cast("NDArray[np.float32]", res) - - -def merge_wgts( - em_sz: int, - wgts: dict[str, torch.Tensor], - itos_pre: list[str], - itos_new: list[str], -) -> dict[str, torch.Tensor]: - """ - Insert new vocabulary into an existing model and update weights. - - Initialize weights of new words with the average embedding - when the words are not in the pretrained vocabulary. - - :param int em_sz: embedding size - :param dict[str, torch.Tensor] wgts: torch model weights - :param list[str] itos_pre: list of words in the pretrained vocabulary - :param list[str] itos_new: list of words in the new vocabulary - - :return: merged torch model weights - :rtype: dict[str, torch.Tensor] - - :Example: - - >>> from pythainlp.ulmfit import merge_wgts # doctest: +SKIP - >>> import torch # doctest: +SKIP - - >>> wgts = {"0.encoder.weight": torch.randn(5, 3)} # doctest: +SKIP - >>> itos_pre = ["แมว", "คน", "หนู"] # doctest: +SKIP - >>> itos_new = ["ปลา", "เต่า", "นก"] # doctest: +SKIP - >>> em_sz = 3 # doctest: +SKIP - - >>> merge_wgts(em_sz, wgts, itos_pre, itos_new) # doctest: +SKIP - {'0.encoder.weight': tensor([[0.5952, 0.4453, 0.0011], - [0.5952, 0.4453, 0.0011], - [0.5952, 0.4453, 0.0011]]), - '0.encoder_dp.emb.weight': tensor([[0.5952, 0.4453, 0.0011], - [0.5952, 0.4453, 0.0011], - [0.5952, 0.4453, 0.0011]]), - '1.decoder.weight': tensor([[0.5952, 0.4453, 0.0011], - [0.5952, 0.4453, 0.0011], - [0.5952, 0.4453, 0.0011]])} - """ - vocab_size = len(itos_new) - enc_wgts = wgts["0.encoder.weight"].numpy().astype("float32", copy=False) - - # Average weight of encoding - row_m = enc_wgts.mean(0, dtype="float32") - stoi_pre = collections.defaultdict( - lambda: -1, {v: k for k, v in enumerate(itos_pre)} - ) - - # New embedding based on classification dataset - import numpy as np - - new_w = np.zeros((vocab_size, em_sz), dtype=np.float32) - - for i, w in enumerate(itos_new): - r = stoi_pre[w] - # Use pretrianed embedding if present; else use the average - new_w[i] = enc_wgts[r] if r >= 0 else row_m - - wgts["0.encoder.weight"] = torch.tensor(new_w) - wgts["0.encoder_dp.emb.weight"] = torch.tensor(np.copy(new_w)) - wgts["1.decoder.weight"] = torch.tensor(np.copy(new_w)) - - return wgts diff --git a/pythainlp/ulmfit/preprocess.py b/pythainlp/ulmfit/preprocess.py index 9eb15c931..0e13ff971 100644 --- a/pythainlp/ulmfit/preprocess.py +++ b/pythainlp/ulmfit/preprocess.py @@ -1,305 +1,51 @@ # SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project # SPDX-FileType: SOURCE # SPDX-License-Identifier: Apache-2.0 -"""Preprocessing functions for ULMFiT.""" +""" +Deprecated. Use :mod:`pythainlp.lm.ulmfit.preprocess` instead. -from __future__ import annotations - -import html -import re -from typing import TYPE_CHECKING, Optional - -if TYPE_CHECKING: - from collections.abc import Collection - -import emoji - -_TK_UNK: str = "xxunk" -_TK_REP: str = "xxrep" -_TK_WREP: str = "xxwrep" -_TK_END: str = "xxend" -_TK_URL: str = "xxurl" - - -def replace_url(text: str) -> str: - """ - Replace URLs in text with the **xxurl** token. - - :param str text: text to be processed - - :return: text with URLs replaced - :rtype: str - - :Example: - - >>> from pythainlp.ulmfit import replace_url - >>> replace_url("go to github.com") - 'go to xxurl' - """ - URL_PATTERN = r"""(?i)\b((?:https?:(?:/{1,3}|[a-z0-9%])|[a-z0-9.\-]+[.](?:com|net|org|edu|gov|mil|aero|asia|biz|cat|coop|info|int|jobs|mobi|museum|name|post|pro|tel|travel|xxx|ac|ad|ae|af|ag|ai|al|am|an|ao|aq|ar|as|at|au|aw|ax|az|ba|bb|bd|be|bf|bg|bh|bi|bj|bm|bn|bo|br|bs|bt|bv|bw|by|bz|ca|cc|cd|cf|cg|ch|ci|ck|cl|cm|cn|co|cr|cs|cu|cv|cx|cy|cz|dd|de|dj|dk|dm|do|dz|ec|ee|eg|eh|er|es|et|eu|fi|fj|fk|fm|fo|fr|ga|gb|gd|ge|gf|gg|gh|gi|gl|gm|gn|gp|gq|gr|gs|gt|gu|gw|gy|hk|hm|hn|hr|ht|hu|id|ie|il|im|in|io|iq|ir|is|it|je|jm|jo|jp|ke|kg|kh|ki|km|kn|kp|kr|kw|ky|kz|la|lb|lc|li|lk|lr|ls|lt|lu|lv|ly|ma|mc|md|me|mg|mh|mk|ml|mm|mn|mo|mp|mq|mr|ms|mt|mu|mv|mw|mx|my|mz|na|nc|ne|nf|ng|ni|nl|no|np|nr|nu|nz|om|pa|pe|pf|pg|ph|pk|pl|pm|pn|pr|ps|pt|pw|py|qa|re|ro|rs|ru|rw|sa|sb|sc|sd|se|sg|sh|si|sj|Ja|sk|sl|sm|sn|so|sr|ss|st|su|sv|sx|sy|sz|tc|td|tf|tg|th|tj|tk|tl|tm|tn|to|tp|tr|tt|tv|tw|tz|ua|ug|uk|us|uy|uz|va|vc|ve|vg|vi|vn|vu|wf|ws|ye|yt|yu|za|zm|zw)/)(?:[^\s()<>{}\[\]]|\([^\s()]*?\([^\s()]+\)[^\s()]*?\)|\([^\s]+?\))+(?:\([^\s()]*?\([^\s()]+\)[^\s()]*?\)|\([^\s]+?\)|[^\s`!()\[\]{};:'".,<>?«»“”‘’])|(?:(? str: - """ - Replace HTML strings in text (code from `fastai`). - - :param str text: text to be processed - - :return: text with HTML strings replaced - :rtype: str - - :Example: - - >>> from pythainlp.ulmfit import fix_html - >>> fix_html("Anbsp;amp;nbsp;B @.@ ") - 'A & B.' - """ - re1 = re.compile(r" +") - text = ( - text.replace("#39;", "'") - .replace("amp;", "&") - .replace("#146;", "'") - .replace("nbsp;", " ") - .replace("#36;", "$") - .replace("\\n", "\n") - .replace("quot;", "'") - .replace("
", "\n") - .replace('\\"', '"') - .replace("", _TK_UNK) - .replace(" @.@ ", ".") - .replace(" @-@ ", "-") - .replace(" @,@ ", ",") - .replace("\\", " \\ ") - ) - return re1.sub(" ", html.unescape(text)) - - -def rm_useless_spaces(text: str) -> str: - """Collapse repeated spaces in text (code from `fastai`).""" - return re.sub(" {2,}", " ", text) - - -def spec_add_spaces(text: str) -> str: - """Add spaces around slashes, hash signs, and newlines in text.""" - return re.sub(r"([/#\n])", r" \1 ", text) - - -def replace_rep_after(text: str) -> str: - """ - Replace character repetitions in text, after the repeated character. - - This prevents 'น้อยยยยยยยย' from becoming 'น้อ xxrep 8 ย'. - Instead, the word stays as 'น้อย xxrep 8'. - - :param str text: text to be processed - - :return: text with the **xxrep** token and the counter - after the repeated character - :rtype: str - - :Example: - - >>> from pythainlp.ulmfit import replace_rep_after - >>> - >>> text = "กาาาาาาา" - >>> replace_rep_after(text) - 'กาxxrep7 ' - """ - - def _replace_rep(m: re.Match[str]) -> str: - c, cc = m.groups() - return f"{c}{_TK_REP}{len(cc) + 1} " - - re_rep = re.compile(r"(\S)(\1{3,})") - - return re_rep.sub(_replace_rep, text) - - -def replace_wrep_post(toks: Collection[str]) -> list[str]: - """ - Replace repeated words after tokenization. - - The `replace_wrep` function of `fastai` does not work well with Thai. - - :param Collection[str] toks: list of words - - :return: list of words with the **xxwrep** token and the counter - added before repeated words - :rtype: list[str] +.. deprecated:: 5.3.9 + :mod:`pythainlp.ulmfit.preprocess` has moved to :mod:`pythainlp.lm.ulmfit.preprocess`. +""" - :Example: - - >>> from pythainlp.ulmfit import replace_wrep_post_nonum - >>> - >>> toks = ["กา", "น้ำ", "น้ำ", "น้ำ", "น้ำ"] - >>> replace_wrep_post(toks) - ['กา', 'xxwrep', '3', 'น้ำ'] - - """ - previous_word: Optional[str] = None - rep_count = 0 - res: list[Optional[str]] = [] - for current_word in [*list(toks), _TK_END]: - if current_word == previous_word: - rep_count += 1 - elif (current_word != previous_word) & (rep_count > 0): - res += [_TK_WREP, str(rep_count), previous_word] - rep_count = 0 - else: - res.append(previous_word) - previous_word = current_word - return list(filter(None, res[1:])) - - -def rm_useless_newlines(text: str) -> str: - """Replace repeated newlines in text with a space.""" - return re.sub(r"[\n]{2,}", " ", text) - - -def rm_brackets(text: str) -> str: - """Remove empty brackets and artifacts within brackets from text.""" - # remove empty brackets - new_line = re.sub(r"\(\)", "", text) - new_line = re.sub(r"\{\}", "", new_line) - new_line = re.sub(r"\[\]", "", new_line) - # brackets with only punctuation marks - new_line = re.sub(r"\([^a-zA-Z0-9ก-๙]+\)", "", new_line) - new_line = re.sub(r"\{[^a-zA-Z0-9ก-๙]+\}", "", new_line) - new_line = re.sub(r"\[[^a-zA-Z0-9ก-๙]+\]", "", new_line) - # artifacts after ( - new_line = re.sub( - r"(?<=\()[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line - ) - new_line = re.sub( - r"(?<=\{)[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line - ) - new_line = re.sub( - r"(?<=\[)[^a-zA-Z0-9ก-๙]+(?=[a-zA-Z0-9ก-๙])", "", new_line - ) - # artifacts before ) - new_line = re.sub( - r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\))", "", new_line - ) - new_line = re.sub( - r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\})", "", new_line - ) - new_line = re.sub( - r"(?<=[a-zA-Z0-9ก-๙])[^a-zA-Z0-9ก-๙]+(?=\])", "", new_line - ) - return new_line - - -def ungroup_emoji(toks: Collection[str]) -> list[str]: - """ - Ungroup emojis joined by Zero Width Joiner (ZWJ). - - See https://emojipedia.org/emoji-zwj-sequence/ - - :param Collection[str] toks: list of words - :return: list of words with ZWJ emoji sequences split - :rtype: list[str] - """ - res = [] - for tok in toks: - if emoji.emoji_count(tok) == len(tok): - res.extend(list(tok)) - else: - res.append(tok) - return res - - -def lowercase_all(toks: Collection[str]) -> list[str]: - """ - Lowercase all English words. - - English words in Thai text usually do not carry capitalization nuances. - - :param Collection[str] toks: list of words - :return: list of lowercased words - :rtype: list[str] - """ - return list(map(str.lower, toks)) - - -def replace_rep_nonum(text: str) -> str: - """ - Replace character repetitions in text, without a counter. - - This prevents 'น้อยยยยยยยย' from becoming 'น้อ xxrep ย'. - Instead, the word stays as 'น้อย xxrep '. - - :param str text: text to be processed - - :return: text with the **xxrep** token after the repeated character - :rtype: str - - :Example: - - >>> from pythainlp.ulmfit import replace_rep_nonum - >>> - >>> text = "กาาาาาาา" - >>> replace_rep_nonum(text) - 'กา xxrep ' - - """ - - def _replace_rep(m: re.Match[str]) -> str: - c, _ = m.groups() - return f"{c} {_TK_REP} " - - re_rep = re.compile(r"(\S)(\1{3,})") - return re_rep.sub(_replace_rep, text) - - -def replace_wrep_post_nonum(toks: Collection[str]) -> list[str]: - """ - Replace repeated words after tokenization, without a counter. - - The `replace_wrep` function of `fastai` does not work well with Thai. - - :param Collection[str] toks: list of words - - :return: list of words with the **xxwrep** token added before - repeated words - :rtype: list[str] - - :Example: - - >>> from pythainlp.ulmfit import replace_wrep_post_nonum - >>> - >>> toks = ["กา", "น้ำ", "น้ำ", "น้ำ", "น้ำ"] - >>> replace_wrep_post_nonum(toks) - ['กา', 'xxwrep', 'น้ำ'] - - """ - previous_word: Optional[str] = None - rep_count = 0 - res: list[Optional[str]] = [] - for current_word in [*list(toks), _TK_END]: - if current_word == previous_word: - rep_count += 1 - elif (current_word != previous_word) & (rep_count > 0): - res += [_TK_WREP, previous_word] - rep_count = 0 - else: - res.append(previous_word) - previous_word = current_word - return list(filter(None, res[1:])) - - -def remove_space(toks: Collection[str]) -> list[str]: - """ - Remove spaces from a list of words, for bag-of-words models. - - :param Collection[str] toks: list of words +from __future__ import annotations - :return: list of words with space tokens (" ") filtered out - :rtype: list[str] - """ - res = [] - for t in toks: - stripped = t.strip() - if stripped: - res.append(stripped) - return res +from pythainlp.lm.ulmfit.preprocess import ( + fix_html, + lowercase_all, + remove_space, + replace_rep_after, + replace_rep_nonum, + replace_url, + replace_wrep_post, + replace_wrep_post_nonum, + rm_brackets, + rm_useless_newlines, + rm_useless_spaces, + spec_add_spaces, + ungroup_emoji, +) +from pythainlp.tools import warn_deprecation + +warn_deprecation( + "pythainlp.ulmfit.preprocess", + "pythainlp.lm.ulmfit.preprocess", + "5.3.9", + "6.0", +) + +__all__: list[str] = [ + "fix_html", + "lowercase_all", + "remove_space", + "replace_rep_after", + "replace_rep_nonum", + "replace_url", + "replace_wrep_post", + "replace_wrep_post_nonum", + "rm_brackets", + "rm_useless_newlines", + "rm_useless_spaces", + "spec_add_spaces", + "ungroup_emoji", +] diff --git a/pythainlp/ulmfit/tokenizer.py b/pythainlp/ulmfit/tokenizer.py index 28f87fe7c..5881ed6cf 100644 --- a/pythainlp/ulmfit/tokenizer.py +++ b/pythainlp/ulmfit/tokenizer.py @@ -1,99 +1,29 @@ # SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project # SPDX-FileType: SOURCE # SPDX-License-Identifier: Apache-2.0 -"""Tokenizer classes for ULMFiT.""" +""" +Deprecated. Use :mod:`pythainlp.lm.ulmfit.tokenizer` instead. -from __future__ import annotations - -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - from collections.abc import Collection - -from pythainlp.tokenize import thai2fit_tokenizer - - -class BaseTokenizer: - """Provide a basic tokenizer class (code from `fastai`).""" - - lang: str - - def __init__(self, lang: str) -> None: - """ - Initialize the tokenizer. - - :param str lang: language code - """ - self.lang: str = lang - - def tokenizer(self, t: str) -> list[str]: - """ - Tokenize text by splitting on spaces. - - :param str t: text to be tokenized - :return: list of words - :rtype: list[str] - """ - return t.split(" ") - - def add_special_cases(self, toks: Collection[str]) -> None: - """ - Add special cases to the tokenizer; do nothing. - - :param Collection[str] toks: special cases to be added - """ +.. deprecated:: 5.3.9 + :mod:`pythainlp.ulmfit.tokenizer` has moved to :mod:`pythainlp.lm.ulmfit.tokenizer`. +""" +from __future__ import annotations -class ThaiTokenizer(BaseTokenizer): - """ - Wrap a frozen newmm tokenizer as a :class:`fastai.BaseTokenizer`. - - See https://docs.fast.ai/text.transform#BaseTokenizer - """ - - lang: str - - def __init__(self, lang: str = "th") -> None: - """ - Initialize the tokenizer. - - :param str lang: language code - """ - self.lang: str = lang - - @staticmethod - def tokenizer(text: str) -> list[str]: - """ - Tokenize text using the newmm engine and the thai2fit dictionary. - - :param str text: text to be tokenized - :return: list of words - :rtype: list[str] - - :Example: - - Using :func:`ThaiTokenizer.tokenizer` is similar to - :func:`pythainlp.tokenize.word_tokenize` with the - ``"ulmfit"`` engine. - - >>> from pythainlp.ulmfit import ThaiTokenizer - >>> from pythainlp.tokenize import word_tokenize - >>> - >>> text = "อาภรณ์, จินตมยปัญญา ภาวนามยปัญญา" - >>> ThaiTokenizer.tokenizer(text) - ['อาภรณ์', ',', ' ', 'จิน', 'ตม', 'ย', 'ปัญญา', - ' ', 'ภาวนามยปัญญา'] - >>> - >>> word_tokenize(text, engine="ulmfit") - ['อาภรณ์', ',', ' ', 'จิน', 'ตม', 'ย', 'ปัญญา', - ' ', 'ภาวนามยปัญญา'] - - """ - return thai2fit_tokenizer().word_tokenize(text) - - def add_special_cases(self, toks: Collection[str]) -> None: - """ - Add special cases to the tokenizer; do nothing. - - :param Collection[str] toks: special cases to be added - """ +from pythainlp.lm.ulmfit.tokenizer import ( + BaseTokenizer, + ThaiTokenizer, +) +from pythainlp.tools import warn_deprecation + +warn_deprecation( + "pythainlp.ulmfit.tokenizer", + "pythainlp.lm.ulmfit.tokenizer", + "5.3.9", + "6.0", +) + +__all__: list[str] = [ + "BaseTokenizer", + "ThaiTokenizer", +] diff --git a/pythainlp/wangchanberta/__init__.py b/pythainlp/wangchanberta/__init__.py index 9f134da19..2d45ca3c6 100644 --- a/pythainlp/wangchanberta/__init__.py +++ b/pythainlp/wangchanberta/__init__.py @@ -1,16 +1,31 @@ # SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project # SPDX-FileType: SOURCE # SPDX-License-Identifier: Apache-2.0 -"""Named entity recognition using WangchanBERTa.""" +""" +Deprecated. Use :mod:`pythainlp.lm.wangchanberta` instead. -__all__: list[str] = [ - "NamedEntityRecognition", - "ThaiNameTagger", - "segment", -] +.. deprecated:: 5.3.9 + :mod:`pythainlp.wangchanberta` has moved to :mod:`pythainlp.lm.wangchanberta`. +""" + +from __future__ import annotations -from pythainlp.wangchanberta.core import ( +from pythainlp.lm.wangchanberta import ( NamedEntityRecognition, ThaiNameTagger, segment, ) +from pythainlp.tools import warn_deprecation + +warn_deprecation( + "pythainlp.wangchanberta", + "pythainlp.lm.wangchanberta", + "5.3.9", + "6.0", +) + +__all__: list[str] = [ + "NamedEntityRecognition", + "ThaiNameTagger", + "segment", +] diff --git a/pythainlp/wangchanberta/core.py b/pythainlp/wangchanberta/core.py index f9fa357cf..2e0d0e77e 100644 --- a/pythainlp/wangchanberta/core.py +++ b/pythainlp/wangchanberta/core.py @@ -1,300 +1,31 @@ # SPDX-FileCopyrightText: 2016-2026 PyThaiNLP Project # SPDX-FileType: SOURCE # SPDX-License-Identifier: Apache-2.0 -"""Named entity recognition using WangchanBERTa.""" +""" +Deprecated. Use :mod:`pythainlp.lm.wangchanberta.core` instead. -from __future__ import annotations - -import re -import warnings -from typing import TYPE_CHECKING, Optional, Union, cast - -if TYPE_CHECKING: - from transformers import ( - CamembertTokenizer, - PreTrainedModel, - PreTrainedTokenizerBase, - ) - from transformers.pipelines import TokenClassificationPipeline - -from pythainlp.tokenize import word_tokenize - -_model_name: str = "wangchanberta-base-att-spm-uncased" -_tokenizer: Optional[CamembertTokenizer] = None - - -def _get_tokenizer() -> CamembertTokenizer: - """Return the tokenizer, initializing it if necessary.""" - global _tokenizer - if _tokenizer is None: - from transformers import CamembertTokenizer - - _tokenizer = CamembertTokenizer.from_pretrained( - f"airesearch/{_model_name}", - revision="main", # nosec B615 - ) - if _model_name == "wangchanberta-base-att-spm-uncased": - _tokenizer.additional_special_tokens = [ - "NOTUSED", - "NOTUSED", - "<_>", - ] - return _tokenizer - - -class ThaiNameTagger: - """Tag named entities in Thai text using WangchanBERTa.""" - - dataset_name: str - grouped_entities: bool - classify_tokens: TokenClassificationPipeline - json_ner: list[dict[str, str]] - output: str - sent_ner: list[tuple[str, str]] - - def __init__( - self, dataset_name: str = "thainer", grouped_entities: bool = True - ) -> None: - """ - Initialize a named entity tagger in IOB format. - - Powered by WangchanBERTa from the VISTEC-depa AI Research - Institute of Thailand. - - :param str dataset_name: dataset the model is fine-tuned on - - * *thainer* - ThaiNER dataset (default) - - :param bool grouped_entities: whether to group word pieces of - the same entity - """ - from transformers import pipeline - - self.dataset_name = dataset_name - self.grouped_entities = grouped_entities - self.classify_tokens = pipeline( - task="ner", - tokenizer=_get_tokenizer(), - model=f"airesearch/{_model_name}", - revision=f"finetuned@{self.dataset_name}-ner", - ignore_labels=[], - grouped_entities=self.grouped_entities, - ) - - def _IOB(self, tag: str) -> str: - if tag != "O": - return "B-" + tag - return "O" - - def _clear_tag(self, tag: str) -> str: - return tag.replace("B-", "").replace("I-", "") - - def get_ner( # noqa: C901, CCR001 # phase2-todo - self, text: str, pos: bool = False, tag: bool = False - ) -> Union[list[tuple[str, str]], str]: - """ - Tag named entities in text in IOB format. - - Powered by WangchanBERTa from the VISTEC-depa AI Research - Institute of Thailand. +.. deprecated:: 5.3.9 + :mod:`pythainlp.wangchanberta.core` has moved to :mod:`pythainlp.lm.wangchanberta.core`. +""" - :param str text: Thai text to be tagged - :param bool pos: output part-of-speech tags. This model does not - support them, so a warning is raised. - :param bool tag: return HTML-like tags in a string instead of a - list of tuples - :return: list of tuples (word group, named entity tag), or a string - with HTML-like tags if **tag** is True - :rtype: Union[list[tuple[str, str]], str] - """ - if pos: - warnings.warn( - "This model does not support POS tag output.", - UserWarning, - stacklevel=2, - ) - text = re.sub(" ", "<_>", text) - self.json_ner: list[dict[str, str]] = self.classify_tokens(text) - self.output: str = "" - if self.grouped_entities and self.dataset_name == "thainer": - self.sent_ner: list[tuple[str, str]] = [ - ( - i["word"].replace("<_>", " ").replace("▁", ""), - self._IOB(i["entity_group"]), - ) - for i in self.json_ner - ] - elif self.dataset_name == "thainer": - self.sent_ner = [ - (i["word"].replace("<_>", " ").replace("▁", ""), i["entity"]) - for i in self.json_ner - if i["word"] != "▁" - ] - else: - self.sent_ner = [ - ( - i["word"].replace("<_>", " ").replace("▁", ""), - i["entity"].replace("_", "-").replace("E-", "I-"), - ) - for i in self.json_ner - ] - if self.sent_ner[0][0] == "" and len(self.sent_ner) > 1: - self.sent_ner = self.sent_ner[1:] - for idx, (word, ner) in enumerate(self.sent_ner): - if ( - idx > 0 - and ner.startswith("B-") - and self._clear_tag(ner) - == self._clear_tag(self.sent_ner[idx - 1][1]) - ): - self.sent_ner[idx] = (word, ner.replace("B-", "I-")) - if tag: - temp = "" - sent = "" - for idx, (word, ner) in enumerate(self.sent_ner): - if ner.startswith("B-") and temp != "": - sent += "" - temp = ner[2:] - sent += "<" + temp + ">" - elif ner.startswith("B-"): - temp = ner[2:] - sent += "<" + temp + ">" - elif ner == "O" and temp != "": - sent += "" - temp = "" - sent += word - - if idx == len(self.sent_ner) - 1 and temp != "": - sent += "" - - return sent - return self.sent_ner - - -class NamedEntityRecognition: - """Recognize named entities in Thai text using WangchanBERTa.""" - - tokenizer: PreTrainedTokenizerBase - model: PreTrainedModel - - def __init__( - self, - model: str = "pythainlp/thainer-corpus-v2-base-model", - revision: Optional[str] = None, - ) -> None: - """ - Initialize a named entity tagger in IOB format. - - Powered by WangchanBERTa from the VISTEC-depa AI Research - Institute of Thailand. - - :param str model: name of a model pretrained from WangchanBERTa - :param Optional[str] revision: git revision id (branch, tag, or - commit hash). Pin to a full commit hash for secure downloads. - """ - from transformers import AutoModelForTokenClassification, AutoTokenizer - - self.tokenizer: PreTrainedTokenizerBase = ( - AutoTokenizer.from_pretrained(model, revision=revision) - ) - self.model: PreTrainedModel = ( - AutoModelForTokenClassification.from_pretrained( - model, revision=revision - ) - ) - - def _fix_span_error( - self, words: list[int], ner: list[str] - ) -> list[tuple[str, str]]: - _ner = [] - _ner = ner - _new_tag = [] - for i, j in zip(words, _ner): - i_decoded = self.tokenizer.decode(i) - tag = "O" if i_decoded.isspace() and j.startswith("B-") else j - if i_decoded in ("", "", ""): - continue - if i_decoded == "<_>": - i_decoded = " " - _new_tag.append((i_decoded, tag)) - return _new_tag - - def get_ner( # noqa: CCR001 # phase2-todo - self, text: str, pos: bool = False, tag: bool = False - ) -> Union[list[tuple[str, str]], str]: - """ - Tag named entities in text in IOB format. - - Powered by WangchanBERTa from the VISTEC-depa AI Research - Institute of Thailand. - - :param str text: Thai text to be tagged - :param bool pos: output part-of-speech tags. This model does not - support them, so a warning is raised. - :param bool tag: return HTML-like tags in a string instead of a - list of tuples - :return: list of tuples (word group, named entity tag), or a string - with HTML-like tags if **tag** is True - :rtype: Union[list[tuple[str, str]], str] - """ - import torch - - if pos: - warnings.warn( - "This model does not support POS tag output.", - UserWarning, - stacklevel=2, - ) - words_token = word_tokenize(text.replace(" ", "<_>")) - inputs = self.tokenizer( - words_token, is_split_into_words=True, return_tensors="pt" - ) - ids = inputs["input_ids"] - mask = inputs["attention_mask"] - # forward pass - outputs = self.model(ids, attention_mask=mask) - logits = outputs[0] - predictions = torch.argmax(logits, dim=2) - predicted_token_class = [ - self.model.config.id2label[t.item()] for t in predictions[0] - ] - ner_tag = self._fix_span_error( - inputs["input_ids"][0], predicted_token_class - ) - if tag: - temp = "" - sent = "" - for idx, (word, ner) in enumerate(ner_tag): - if ner.startswith("B-") and temp != "": - sent += "" - temp = ner[2:] - sent += "<" + temp + ">" - elif ner.startswith("B-"): - temp = ner[2:] - sent += "<" + temp + ">" - elif ner == "O" and temp != "": - sent += "" - temp = "" - sent += word - - if idx == len(ner_tag) - 1 and temp != "": - sent += "" - - return sent - return ner_tag - - -def segment(text: str) -> list[str]: - """ - Tokenize text into subwords with the WangchanBERTa tokenizer. - - The tokenizer is a SentencePiece model. - - :param str text: text to be tokenized - :return: list of subwords - :rtype: list[str] - """ - if not text or not isinstance(text, str): - return [] +from __future__ import annotations - return cast("list[str]", _get_tokenizer().tokenize(text)) +from pythainlp.lm.wangchanberta.core import ( + NamedEntityRecognition, + ThaiNameTagger, + segment, +) +from pythainlp.tools import warn_deprecation + +warn_deprecation( + "pythainlp.wangchanberta.core", + "pythainlp.lm.wangchanberta.core", + "5.3.9", + "6.0", +) + +__all__: list[str] = [ + "NamedEntityRecognition", + "ThaiNameTagger", + "segment", +] diff --git a/tests/core/test_lm.py b/tests/core/test_lm.py index cec4da133..182300119 100644 --- a/tests/core/test_lm.py +++ b/tests/core/test_lm.py @@ -30,3 +30,142 @@ def test_remove_repeated_ngrams(self): remove_repeated_ngrams(texts, n=2), ["เอา", "เอา", "แบบ", "แบบ", "ไหน"], ) + + def test_lm_phayathaibert_segment_empty(self): + from pythainlp.lm.phayathaibert import segment + + self.assertEqual(segment(""), []) + self.assertEqual(segment(None), []) # type: ignore[arg-type] + + def test_lm_phayathaibert_normalization(self): + from pythainlp.lm.phayathaibert.core import ThaiTextProcessor + + processor = ThaiTextProcessor() + self.assertEqual( + processor.replace_url("https://github.com ftp://a-b.example.org"), + " ", + ) + self.assertEqual(processor.rm_brackets("a()b{}c[]d"), "abcd") + + def test_lm_wangchanberta_segment_empty(self): + from pythainlp.lm.wangchanberta import segment + + self.assertEqual(segment(""), []) + self.assertEqual(segment(None), []) # type: ignore[arg-type] + + def test_lm_wangchanberta_fix_span_error(self): + from pythainlp.lm.wangchanberta.core import NamedEntityRecognition + + class Tokenizer: + def decode(self, token: int) -> str: + return {1: " ", 2: "word", 3: ""}[token] + + tagger = object.__new__(NamedEntityRecognition) + tagger.tokenizer = Tokenizer() + self.assertEqual( + tagger._fix_span_error([1, 2, 3], ["B-PER", "B-PER", "O"]), + [(" ", "O"), ("word", "B-PER")], + ) + + def test_wangchanberta_prepare_ner_variants(self): + from pythainlp.lm.wangchanberta.core import ThaiNameTagger + + tagger = object.__new__(ThaiNameTagger) + self.assertEqual(tagger._iob("O"), "O") + self.assertEqual(tagger._iob("PER"), "B-PER") + tagger.dataset_name = "thainer" + tagger.grouped_entities = True + self.assertEqual( + tagger._prepare_ner( + [{"word": "▁John", "entity_group": "PER", "entity": "B-PER"}] + ), + [("John", "B-PER")], + ) + tagger.grouped_entities = False + self.assertEqual( + tagger._prepare_ner( + [ + {"word": "▁", "entity_group": "O", "entity": "O"}, + {"word": "John", "entity_group": "PER", "entity": "B-PER"}, + ] + ), + [("John", "B-PER")], + ) + tagger.dataset_name = "other" + self.assertEqual( + tagger._prepare_ner( + [{"word": "John", "entity_group": "PER", "entity": "E_PER"}] + ), + [("John", "I-PER")], + ) + + def test_wangchanberta_fix_tags_and_format_tags(self): + from pythainlp.lm.wangchanberta.core import ( + ThaiNameTagger, + _format_ner_tags, + ) + + tagger = object.__new__(ThaiNameTagger) + tagger.sent_ner = [ + ("Mary", "B-PER"), + ("Jane", "B-PER"), + ("works", "O"), + ("Paris", "B-LOC"), + ] + tagger._fix_consecutive_begin_tags() + self.assertEqual( + tagger.sent_ner, + [ + ("Mary", "B-PER"), + ("Jane", "I-PER"), + ("works", "O"), + ("Paris", "B-LOC"), + ], + ) + self.assertEqual( + _format_ner_tags(tagger.sent_ner), + "MaryJaneworksParis", + ) + self.assertEqual(_format_ner_tags([]), "") + + def test_wangchanberta_empty_prediction(self): + from pythainlp.lm.wangchanberta.core import ThaiNameTagger + + tagger = object.__new__(ThaiNameTagger) + tagger.dataset_name = "thainer" + tagger.grouped_entities = True + tagger.classify_tokens = lambda text: [] + self.assertEqual(tagger.get_ner(""), []) + self.assertEqual(tagger.get_ner("", tag=True), "") + + def test_deprecated_phayathaibert(self): + import importlib + import warnings + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + import pythainlp.phayathaibert + + importlib.reload(pythainlp.phayathaibert) + self.assertTrue( + any( + issubclass(warning.category, DeprecationWarning) + for warning in w + ) + ) + + def test_deprecated_wangchanberta(self): + import importlib + import warnings + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + import pythainlp.wangchanberta + + importlib.reload(pythainlp.wangchanberta) + self.assertTrue( + any( + issubclass(warning.category, DeprecationWarning) + for warning in w + ) + ) diff --git a/tests/core/test_phayathaibert_preprocess.py b/tests/core/test_phayathaibert_preprocess.py index 64f55ec0c..88d4417da 100644 --- a/tests/core/test_phayathaibert_preprocess.py +++ b/tests/core/test_phayathaibert_preprocess.py @@ -21,15 +21,15 @@ def _import_phayathaibert() -> Any: """ - Load phayathaibert/core.py with a fake ``transformers``. + Load lm/phayathaibert/core.py with a fake ``transformers``. - Neither ``sys.modules`` nor the ``pythainlp.phayathaibert`` package + Neither ``sys.modules`` nor the ``pythainlp.lm.phayathaibert`` package attribute is changed. """ fake = types.ModuleType("transformers") tokenizer_class = mock.Mock() fake.CamembertTokenizer = tokenizer_class # type: ignore[attr-defined] - path = Path(pythainlp.__file__).parent / "phayathaibert" / "core.py" + path = Path(pythainlp.__file__).parent / "lm" / "phayathaibert" / "core.py" spec = importlib.util.spec_from_file_location( "phayathaibert_under_test", path ) diff --git a/tests/core/test_tokenize_core.py b/tests/core/test_tokenize_core.py index 7000c90e2..77eb7f7de 100644 --- a/tests/core/test_tokenize_core.py +++ b/tests/core/test_tokenize_core.py @@ -77,11 +77,11 @@ "tcc": ("pythainlp.tokenize.tcc", "segment"), "tcc_p": ("pythainlp.tokenize.tcc_p", "segment"), "etcc": ("pythainlp.tokenize.etcc", "segment"), - "wangchanberta": ("pythainlp.wangchanberta", "segment"), + "wangchanberta": ("pythainlp.lm.wangchanberta", "segment"), "ssg": ("pythainlp.tokenize.ssg", "segment"), "tltk": ("pythainlp.tokenize.tltk", "syllable_tokenize"), "han_solo": ("pythainlp.tokenize.han_solo", "segment"), - "phayathai": ("pythainlp.phayathaibert", "segment"), + "phayathai": ("pythainlp.lm.phayathaibert", "segment"), } # engine: model size passed to pythainlp.tokenize.wtsplit.tokenize diff --git a/tests/diff-cover-noauto.txt b/tests/diff-cover-noauto.txt index 61529e05d..cd6cc3a81 100644 --- a/tests/diff-cover-noauto.txt +++ b/tests/diff-cover-noauto.txt @@ -16,13 +16,20 @@ pythainlp/augment/lm/wangchanberta.py pythainlp/coref/_fastcoref.py pythainlp/coref/han_coref.py pythainlp/el/_multiel.py +pythainlp/lm/phayathaibert/__init__.py +pythainlp/lm/phayathaibert/core.py pythainlp/lm/qwen3.py +pythainlp/lm/ulmfit/__init__.py +pythainlp/lm/ulmfit/core.py +pythainlp/lm/ulmfit/preprocess.py +pythainlp/lm/ulmfit/tokenizer.py +pythainlp/lm/wangchanberta/__init__.py +pythainlp/lm/wangchanberta/core.py pythainlp/parse/attaparse_engine.py pythainlp/parse/esupar_engine.py pythainlp/parse/spacy_thai_engine.py pythainlp/parse/transformers_ud.py pythainlp/parse/ud_goeswith.py -pythainlp/phayathaibert/core.py pythainlp/soundex/sound.py pythainlp/spell/phunspell.py pythainlp/spell/wanchanberta_thai_grammarly.py @@ -49,7 +56,4 @@ pythainlp/transliterate/thaig2p.py pythainlp/transliterate/thaig2p_v2.py pythainlp/transliterate/thaig2p_v4.py pythainlp/transliterate/umt5_thaig2p.py -pythainlp/ulmfit/core.py -pythainlp/ulmfit/preprocess.py -pythainlp/wangchanberta/core.py pythainlp/wsd/core.py diff --git a/tests/extra/testx_ulmfit.py b/tests/extra/testx_ulmfit.py index 4e370a59b..73f62cdd7 100644 --- a/tests/extra/testx_ulmfit.py +++ b/tests/extra/testx_ulmfit.py @@ -20,8 +20,7 @@ untar_data, ) -from pythainlp.tokenize import thai2fit_tokenizer -from pythainlp.ulmfit import ( +from pythainlp.lm.ulmfit import ( THWIKI_LSTM, ThaiTokenizer, document_vector, @@ -32,7 +31,7 @@ pre_rules_th_sparse, process_thai, ) -from pythainlp.ulmfit.preprocess import ( +from pythainlp.lm.ulmfit.preprocess import ( fix_html, lowercase_all, remove_space, @@ -47,7 +46,8 @@ spec_add_spaces, ungroup_emoji, ) -from pythainlp.ulmfit.tokenizer import BaseTokenizer as base_tokenizer +from pythainlp.lm.ulmfit.tokenizer import BaseTokenizer as base_tokenizer +from pythainlp.tokenize import thai2fit_tokenizer class UlmfitTestCaseX(unittest.TestCase): @@ -284,3 +284,19 @@ def test_merge_wgts(self): itos_new = ["ปลา", "เต่า", "นก"] em_sz = 3 self.assertIsNotNone(merge_wgts(em_sz, wgts, itos_pre, itos_new)) + + def test_deprecated_ulmfit(self): + import importlib + import warnings + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + import pythainlp.ulmfit + + importlib.reload(pythainlp.ulmfit) + self.assertTrue( + any( + issubclass(warning.category, DeprecationWarning) + for warning in w + ) + ) diff --git a/tests/extra/testx_wangchanberta.py b/tests/extra/testx_wangchanberta.py index 5cd6030ae..b0887c021 100644 --- a/tests/extra/testx_wangchanberta.py +++ b/tests/extra/testx_wangchanberta.py @@ -3,8 +3,9 @@ # SPDX-License-Identifier: Apache-2.0 import unittest +import warnings -from pythainlp.wangchanberta import ThaiNameTagger, segment +from pythainlp.lm.wangchanberta import ThaiNameTagger, segment class WangchanbertaTestCaseX(unittest.TestCase): @@ -25,3 +26,18 @@ def test_segment_wangchanberta(self): self.assertIsNotNone( segment([]) # type: ignore[arg-type] ) + + def test_deprecated_wangchanberta(self): + import importlib + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + import pythainlp.wangchanberta + + importlib.reload(pythainlp.wangchanberta) + self.assertTrue( + any( + issubclass(warning.category, DeprecationWarning) + for warning in w + ) + )