diff --git a/CHANGELOG.md b/CHANGELOG.md index f0d66e9a4..195e39db3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,20 @@ and this project adheres to ## [Unreleased] +### Added + +- `EntitySpan` TypedDict (#1363). + Migration notes: + + ```python + # EntitySpan – Before (plain dict) + entity = {"text": ["สมชาย"], "span": [0, 1], "entity_type": "PERSON"} + + # EntitySpan – After (TypedDict) + from pythainlp.tag.named_entity import EntitySpan + entity = EntitySpan(text=["สมชาย"], span=[0, 1], entity_type="PERSON") + ``` + ### Fixed - thai2rom_onnx: fix ONNX encoder model and fix inference bugs (#1349) diff --git a/pythainlp/benchmarks/__init__.py b/pythainlp/benchmarks/__init__.py index 716230cef..773ddd2bc 100644 --- a/pythainlp/benchmarks/__init__.py +++ b/pythainlp/benchmarks/__init__.py @@ -5,6 +5,11 @@ __all__: list[str] = [ "BleuScore", + "CharLevelStat", + "GlobalStat", + "RougeScore", + "TokenizationStat", + "WordLevelStat", "benchmark", "bleu_score", "character_error_rate", @@ -14,9 +19,16 @@ from pythainlp.benchmarks.metrics import ( BleuScore, + RougeScore, bleu_score, character_error_rate, rouge_score, word_error_rate, ) -from pythainlp.benchmarks.word_tokenization import benchmark +from pythainlp.benchmarks.word_tokenization import ( + CharLevelStat, + GlobalStat, + TokenizationStat, + WordLevelStat, + benchmark, +) diff --git a/pythainlp/benchmarks/metrics.py b/pythainlp/benchmarks/metrics.py index b2eb963ee..ae793c181 100644 --- a/pythainlp/benchmarks/metrics.py +++ b/pythainlp/benchmarks/metrics.py @@ -16,7 +16,7 @@ class BleuScore(TypedDict): - """BLEU score""" + """BLEU score components returned by :func:`bleu_score`.""" bleu: float # BLEU score as a percentage (0.0 to 100.0) precisions: list[float] @@ -26,6 +26,14 @@ class BleuScore(TypedDict): ref_length: int +class RougeScore(TypedDict): + """Precision, recall, and F-measure for a single ROUGE type.""" + + precision: float + recall: float + fmeasure: float + + def _get_ngrams(tokens: list[str], n: int) -> list[tuple[str, ...]]: """ Get n-grams from a list of tokens. @@ -249,7 +257,7 @@ def rouge_score( hypothesis: str, tokenize: str = "newmm", rouge_types: Optional[list[str]] = None, -) -> dict[str, tuple[float, float, float]]: +) -> dict[str, RougeScore]: """ Calculate ROUGE scores for Thai text with automatic tokenization. @@ -269,8 +277,9 @@ def rouge_score( :param Optional[list[str]] rouge_types: list of ROUGE types to calculate. Default is ["rouge1", "rouge2", "rougeL"] - :return: dictionary mapping ROUGE type to (precision, recall, fmeasure) - :rtype: dict[str, tuple[float, float, float]] + :return: dictionary mapping ROUGE type to a :class:`RougeScore` typed dict + with ``'precision'``, ``'recall'``, and ``'fmeasure'`` keys. + :rtype: dict[str, RougeScore] :Example: :: @@ -280,9 +289,9 @@ def rouge_score( reference = "สวัสดีครับ วันนี้อากาศดีมาก" hypothesis = "สวัสดีค่ะ วันนี้อากาศดี" scores = rouge_score(reference, hypothesis) - print(f"ROUGE-1 F-measure: {scores['rouge1'][2]:.4f}") - print(f"ROUGE-2 F-measure: {scores['rouge2'][2]:.4f}") - print(f"ROUGE-L F-measure: {scores['rougeL'][2]:.4f}") + print(f"ROUGE-1 F-measure: {scores['rouge1']['fmeasure']:.4f}") + print(f"ROUGE-2 F-measure: {scores['rouge2']['fmeasure']:.4f}") + print(f"ROUGE-L F-measure: {scores['rougeL']['fmeasure']:.4f}") """ from pythainlp.tokenize import word_tokenize @@ -297,7 +306,7 @@ def rouge_score( hypothesis, engine=tokenize, keep_whitespace=False ) - result: dict[str, tuple[float, float, float]] = {} + result: dict[str, RougeScore] = {} for rouge_type in rouge_types: if rouge_type == "rouge1": @@ -309,9 +318,12 @@ def rouge_score( ref_count = len(ref_tokens) hyp_count = len(hyp_tokens) - result[rouge_type] = _calculate_precision_recall_fmeasure( + precision, recall, fmeasure = _calculate_precision_recall_fmeasure( overlap, hyp_count, ref_count ) + result[rouge_type] = RougeScore( + precision=precision, recall=recall, fmeasure=fmeasure + ) elif rouge_type == "rouge2": # Bigram-based @@ -325,9 +337,12 @@ def rouge_score( ref_count = len(ref_bigrams) hyp_count = len(hyp_bigrams) - result[rouge_type] = _calculate_precision_recall_fmeasure( + precision, recall, fmeasure = _calculate_precision_recall_fmeasure( overlap, hyp_count, ref_count ) + result[rouge_type] = RougeScore( + precision=precision, recall=recall, fmeasure=fmeasure + ) elif rouge_type == "rougeL": # Longest Common Subsequence-based @@ -335,9 +350,12 @@ def rouge_score( ref_count = len(ref_tokens) hyp_count = len(hyp_tokens) - result[rouge_type] = _calculate_precision_recall_fmeasure( + precision, recall, fmeasure = _calculate_precision_recall_fmeasure( lcs_len, hyp_count, ref_count ) + result[rouge_type] = RougeScore( + precision=precision, recall=recall, fmeasure=fmeasure + ) return result diff --git a/pythainlp/benchmarks/word_tokenization.py b/pythainlp/benchmarks/word_tokenization.py index 01b67fee4..e0475657c 100644 --- a/pythainlp/benchmarks/word_tokenization.py +++ b/pythainlp/benchmarks/word_tokenization.py @@ -5,7 +5,8 @@ import re import sys -from typing import TYPE_CHECKING, Union +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, TypedDict, Union, overload if TYPE_CHECKING: import numpy as np @@ -29,6 +30,37 @@ TAILING_SEP_RX: re.Pattern[str] = re.compile(f"{re.escape(SEPARATOR)}$") +class CharLevelStat(TypedDict): + """Character-level confusion matrix statistics for tokenization.""" + + tp: int + fp: int + tn: int + fn: int + + +class WordLevelStat(TypedDict): + """Word-level tokenization statistics.""" + + correctly_tokenized_words: int + total_words_in_sample: int + total_words_in_ref_sample: int + + +class GlobalStat(TypedDict): + """Global tokenization indicator as a binary indicator string.""" + + tokenization_indicators: str + + +class TokenizationStat(TypedDict): + """Tokenization quality statistics at character, word, and global level.""" + + char_level: CharLevelStat + word_level: WordLevelStat + global_: GlobalStat + + def _f1(precision: float, recall: float) -> float: """Compute f1. @@ -43,8 +75,21 @@ def _f1(precision: float, recall: float) -> float: return 2 * precision * recall / (precision + recall) +@overload +def _flatten_result( + my_dict: TokenizationStat, sep: str = ... +) -> dict[str, Union[int, str]]: ... + + +@overload +def _flatten_result( + my_dict: Mapping[str, Mapping[str, Union[int, str]]], sep: str = ... +) -> dict[str, Union[int, str]]: ... + + def _flatten_result( - my_dict: dict[str, dict[str, Union[int, str]]], sep: str = ":" + my_dict: Any, + sep: str = ":", ) -> dict[str, Union[int, str]]: """Flatten two-dimension dictionary. @@ -55,8 +100,9 @@ def _flatten_result( { "a:b": 7 } - :param dict[str, dict[str, Union[int, str]]] my_dict: dictionary - containing stats + :param my_dict: dictionary containing stats + :type my_dict: TokenizationStat or + collections.abc.Mapping[str, collections.abc.Mapping[str, Union[int, str]]] :param str sep: separator between the two keys (default: ":") :return: a one-dimension dictionary with keys combined @@ -139,7 +185,7 @@ def preprocessing(txt: str, remove_space: bool = True) -> str: def compute_stats( ref_sample: str, raw_sample: str -) -> dict[str, dict[str, Union[int, str]]]: +) -> TokenizationStat: """Compute statistics for tokenization quality These statistics include: @@ -156,7 +202,7 @@ def compute_stats( :param str samples: samples that we want to evaluate :return: metrics at character- and word-level and indicators of correctly tokenized words - :rtype: dict[str, dict[str, Union[int, str]]] + :rtype: TokenizationStat """ import numpy as np @@ -185,29 +231,29 @@ def compute_stats( # Find correctly tokenized words in the sample ss_boundaries = _find_word_boundaries(sample_arr) - tokenization_indicators = _find_words_correctly_tokenised( + tokenization_indicators = _find_words_correctly_tokenized( word_boundaries, ss_boundaries ) - correctly_tokenised_words: int = int(np.sum(tokenization_indicators)) + correctly_tokenized_words: int = int(np.sum(tokenization_indicators)) tokenization_indicators_str = list(map(str, tokenization_indicators)) return { - "char_level": { - "tp": c_tp, - "fp": c_fp, - "tn": c_tn, - "fn": c_fn, - }, - "word_level": { - "correctly_tokenised_words": correctly_tokenised_words, - "total_words_in_sample": int(np.sum(sample_arr)), - "total_words_in_ref_sample": int(np.sum(ref_sample_arr)), - }, - "global": { - "tokenisation_indicators": "".join(tokenization_indicators_str) - }, + "char_level": CharLevelStat( + tp=c_tp, + fp=c_fp, + tn=c_tn, + fn=c_fn, + ), + "word_level": WordLevelStat( + correctly_tokenized_words=correctly_tokenized_words, + total_words_in_sample=int(np.sum(sample_arr)), + total_words_in_ref_sample=int(np.sum(ref_sample_arr)), + ), + "global_": GlobalStat( + tokenization_indicators="".join(tokenization_indicators_str), + ), } @@ -271,7 +317,7 @@ def _find_word_boundaries( return list(zip(start_idx, end_idx)) -def _find_words_correctly_tokenised( +def _find_words_correctly_tokenized( ref_boundaries: list[tuple[int, int]], predicted_boundaries: list[tuple[int, int]], ) -> tuple[int, ...]: diff --git a/pythainlp/cli/benchmark.py b/pythainlp/cli/benchmark.py index 953bb8181..9eb5a57cd 100644 --- a/pythainlp/cli/benchmark.py +++ b/pythainlp/cli/benchmark.py @@ -108,7 +108,7 @@ def __init__(self, name: str, argv: Sequence[str]) -> None: "char_level:fp", "char_level:tn", "char_level:fn", - "word_level:correctly_tokenised_words", + "word_level:correctly_tokenized_words", "word_level:total_words_in_sample", "word_level:total_words_in_ref_sample", ] @@ -127,12 +127,12 @@ def __init__(self, name: str, argv: Sequence[str]) -> None: ) statistics["word_level:precision"] = ( - statistics["word_level:correctly_tokenised_words"] + statistics["word_level:correctly_tokenized_words"] / statistics["word_level:total_words_in_sample"] ) statistics["word_level:recall"] = ( - statistics["word_level:correctly_tokenised_words"] + statistics["word_level:correctly_tokenized_words"] / statistics["word_level:total_words_in_ref_sample"] ) @@ -146,7 +146,7 @@ def __init__(self, name: str, argv: Sequence[str]) -> None: for c in [ "total_words_in_sample", "total_words_in_ref_sample", - "correctly_tokenised_words", + "correctly_tokenized_words", "precision", "recall", ]: diff --git a/pythainlp/coref/__init__.py b/pythainlp/coref/__init__.py index 883bc970c..2b85347ad 100644 --- a/pythainlp/coref/__init__.py +++ b/pythainlp/coref/__init__.py @@ -3,6 +3,7 @@ # SPDX-License-Identifier: Apache-2.0 """PyThaiNLP Coreference Resolution""" -__all__: list[str] = ["coreference_resolution"] +__all__: list[str] = ["CorefResult", "coreference_resolution"] +from pythainlp.coref._fastcoref import CorefResult from pythainlp.coref.core import coreference_resolution diff --git a/pythainlp/coref/_fastcoref.py b/pythainlp/coref/_fastcoref.py index f1db6bc47..7c431201a 100644 --- a/pythainlp/coref/_fastcoref.py +++ b/pythainlp/coref/_fastcoref.py @@ -6,12 +6,13 @@ from typing import TYPE_CHECKING, Optional, TypedDict if TYPE_CHECKING: - from fastcoref.modeling import CorefModel, CorefResult + from fastcoref.modeling import CorefModel + from fastcoref.modeling import CorefResult as FastCorefResult from spacy.language import Language -class CorefResultDict(TypedDict): - """Dictionary representation of coreference resolution results.""" +class CorefResult(TypedDict): + """Coreference resolution result for a single text.""" text: str clusters_string: list[list[str]] @@ -42,14 +43,14 @@ def __init__( self.model_name, device=device, nlp=self.nlp ) - def _to_json(self, _predict: "CorefResult") -> CorefResultDict: + def _to_json(self, _predict: "FastCorefResult") -> CorefResult: return { "text": _predict.text, "clusters_string": _predict.get_clusters(as_strings=True), "clusters": _predict.get_clusters(as_strings=False), } - def predict(self, texts: list[str]) -> list[CorefResultDict]: + def predict(self, texts: list[str]) -> list[CorefResult]: return [ self._to_json(pred) for pred in self.model.predict(texts=texts) ] diff --git a/pythainlp/coref/core.py b/pythainlp/coref/core.py index 94459e453..2776b5786 100644 --- a/pythainlp/coref/core.py +++ b/pythainlp/coref/core.py @@ -5,6 +5,8 @@ from typing import Any, Union, cast +from pythainlp.coref._fastcoref import CorefResult + _MODEL_CACHE: dict[tuple[str, str], Any] = {} @@ -12,7 +14,7 @@ def coreference_resolution( texts: Union[str, list[str]], model_name: str = "han-coref-v1.0", device: str = "cpu", -) -> list[dict[str, Any]]: +) -> list[CorefResult]: """Coreference Resolution :param Union[str, list[str]] texts: list of texts to apply coreference resolution to @@ -20,7 +22,7 @@ def coreference_resolution( :param str device: device for running coreference resolution model on\ ("cpu", "cuda", and others) :return: List of texts with coreference resolution - :rtype: list[dict[str, Any]] + :rtype: list[CorefResult] :Options for model_name: * *han-coref-v1.0* - (default) Han-Coref: Thai coreference resolution\ @@ -54,8 +56,9 @@ def coreference_resolution( model = _MODEL_CACHE.get(model_key) if model is not None: - return cast(list[dict[str, Any]], model.predict(texts)) + return cast(list[CorefResult], model.predict(texts)) return [ - {"text": text, "clusters_string": [], "clusters": []} for text in texts + CorefResult(text=text, clusters_string=[], clusters=[]) + for text in texts ] diff --git a/pythainlp/tag/named_entity.py b/pythainlp/tag/named_entity.py index e768b789c..c76ab525e 100644 --- a/pythainlp/tag/named_entity.py +++ b/pythainlp/tag/named_entity.py @@ -22,7 +22,8 @@ class EntitySpan(TypedDict): - """Entity span dictionary""" + """Named-entity span with its type, tokens, and position.""" + entity_type: str text: list[str] span: list[int] diff --git a/tests/extra/testx_benchmarks.py b/tests/extra/testx_benchmarks.py index d59611179..5a0857cbe 100644 --- a/tests/extra/testx_benchmarks.py +++ b/tests/extra/testx_benchmarks.py @@ -9,6 +9,11 @@ from pythainlp.benchmarks import ( BleuScore, + CharLevelStat, + GlobalStat, + RougeScore, + TokenizationStat, + WordLevelStat, bleu_score, rouge_score, word_tokenization, @@ -53,6 +58,35 @@ def test_compute_stats(self): self.assertIsNotNone(result) + def test_compute_stats_return_type(self): + """Test that compute_stats returns a TokenizationStat typed dict.""" + ref = word_tokenization.preprocessing("อากาศ|ร้อน|มาก") + act = word_tokenization.preprocessing("อากาศ|ร้อนมาก") + + result: TokenizationStat = word_tokenization.compute_stats(ref, act) + + self.assertIsInstance(result, dict) + self.assertIn("char_level", result) + self.assertIn("word_level", result) + self.assertIn("global_", result) + + char: CharLevelStat = result["char_level"] + self.assertIsInstance(char, dict) + self.assertIsInstance(char["tp"], int) + self.assertIsInstance(char["fp"], int) + self.assertIsInstance(char["tn"], int) + self.assertIsInstance(char["fn"], int) + + word: WordLevelStat = result["word_level"] + self.assertIsInstance(word, dict) + self.assertIsInstance(word["correctly_tokenized_words"], int) + self.assertIsInstance(word["total_words_in_sample"], int) + self.assertIsInstance(word["total_words_in_ref_sample"], int) + + glob: GlobalStat = result["global_"] + self.assertIsInstance(glob, dict) + self.assertIsInstance(glob["tokenization_indicators"], str) + def test_benchmark(self): expected = [] actual = [] @@ -64,7 +98,7 @@ def test_benchmark(self): self.assertIsNotNone(df) - def test_count_correctly_tokenised_words(self): + def test_count_correctly_tokenized_words(self): for d in TEST_DATA["binary_sentences"]: sample = np.array(list(d["actual"])).astype(int) ref_sample = np.array(list(d["expected"])).astype(int) @@ -74,20 +108,20 @@ def test_count_correctly_tokenised_words(self): # in binary [{0, 1}, ...] correctly_tokenized_words = ( - word_tokenization._find_words_correctly_tokenised(rb, sb) + word_tokenization._find_words_correctly_tokenized(rb, sb) ) self.assertEqual( np.sum(correctly_tokenized_words), d["expected_count"] ) - def test_words_correctly_tokenised(self): + def test_words_correctly_tokenized(self): r = [(0, 2), (2, 10), (10, 12)] s = [(0, 10), (10, 12)] expected = "01" - labels = word_tokenization._find_words_correctly_tokenised(r, s) + labels = word_tokenization._find_words_correctly_tokenized(r, s) self.assertEqual(expected, "".join(np.array(labels).astype(str))) def test_flatten_result(self): @@ -195,16 +229,36 @@ def test_rouge_score_basic(self): self.assertIn("rouge2", scores) self.assertIn("rougeL", scores) - # Each score should be a tuple of (precision, recall, fmeasure) + # Each score is a RougeScore dict with precision, recall, fmeasure + for key in ["rouge1", "rouge2", "rougeL"]: + self.assertIsInstance(scores[key], dict) + self.assertIsInstance(scores[key]["precision"], float) + self.assertIsInstance(scores[key]["recall"], float) + self.assertIsInstance(scores[key]["fmeasure"], float) + self.assertGreaterEqual(scores[key]["precision"], 0.0) + self.assertLessEqual(scores[key]["precision"], 1.0) + self.assertGreaterEqual(scores[key]["recall"], 0.0) + self.assertLessEqual(scores[key]["recall"], 1.0) + self.assertGreaterEqual(scores[key]["fmeasure"], 0.0) + self.assertLessEqual(scores[key]["fmeasure"], 1.0) + + def test_rouge_score_return_type(self): + """Test that rouge_score returns dict[str, RougeScore] typed dicts.""" + reference = "สวัสดีครับ วันนี้อากาศดีมาก" + hypothesis = "สวัสดีค่ะ วันนี้อากาศดี" + + scores: dict[str, RougeScore] = rouge_score(reference, hypothesis) + + self.assertIsInstance(scores, dict) for key in ["rouge1", "rouge2", "rougeL"]: - self.assertEqual(len(scores[key]), 3) - precision, recall, fmeasure = scores[key] - self.assertGreaterEqual(precision, 0.0) - self.assertLessEqual(precision, 1.0) - self.assertGreaterEqual(recall, 0.0) - self.assertLessEqual(recall, 1.0) - self.assertGreaterEqual(fmeasure, 0.0) - self.assertLessEqual(fmeasure, 1.0) + score: RougeScore = scores[key] + self.assertIsInstance(score, dict) + self.assertIn("precision", score) + self.assertIn("recall", score) + self.assertIn("fmeasure", score) + self.assertIsInstance(score["precision"], float) + self.assertIsInstance(score["recall"], float) + self.assertIsInstance(score["fmeasure"], float) def test_rouge_score_identical_text(self): """Test ROUGE score when reference and hypothesis are identical.""" @@ -214,10 +268,9 @@ def test_rouge_score_identical_text(self): # All scores should be perfect (1.0) for key in ["rouge1", "rouge2", "rougeL"]: - precision, recall, fmeasure = scores[key] - self.assertEqual(precision, 1.0) - self.assertEqual(recall, 1.0) - self.assertEqual(fmeasure, 1.0) + self.assertEqual(scores[key]["precision"], 1.0) + self.assertEqual(scores[key]["recall"], 1.0) + self.assertEqual(scores[key]["fmeasure"], 1.0) def test_rouge_score_custom_types(self): """Test ROUGE score with custom rouge types.""" @@ -258,10 +311,9 @@ def test_rouge_score_no_overlap(self): # Scores should be 0 or very low since there's no overlap for key in ["rouge1", "rouge2", "rougeL"]: - precision, recall, fmeasure = scores[key] - self.assertGreaterEqual(precision, 0.0) - self.assertGreaterEqual(recall, 0.0) - self.assertGreaterEqual(fmeasure, 0.0) + self.assertGreaterEqual(scores[key]["precision"], 0.0) + self.assertGreaterEqual(scores[key]["recall"], 0.0) + self.assertGreaterEqual(scores[key]["fmeasure"], 0.0) def test_word_error_rate_basic(self): """Test WER with basic Thai text."""