| """
|
| Advanced Thai Tokenizer for ZombitThai-GPT
|
| Features:
|
| - Thai-specific tokenization with syllable awareness
|
| - Continuous learning system for new words
|
| - Duplicate prevention mechanism
|
| - Dynamic vocabulary expansion
|
| - Memory-efficient token management
|
| """
|
|
|
| import json
|
| import os
|
| import re
|
| import hashlib
|
| from typing import List, Dict, Set, Tuple, Optional
|
| from collections import defaultdict, Counter
|
| import pickle
|
| from pathlib import Path
|
| import logging
|
|
|
|
|
| try:
|
| from pythainlp import word_tokenize
|
| PYTHAINLP_AVAILABLE = True
|
| except ImportError:
|
| PYTHAINLP_AVAILABLE = False
|
| logging.warning("pythainlp not available. Using fallback Thai segmentation.")
|
|
|
|
|
| logging.basicConfig(level=logging.INFO)
|
| logger = logging.getLogger(__name__)
|
|
|
| class AdvancedThaiTokenizer:
|
| """
|
| Ultra-Advanced Thai Tokenizer with Multi-Engine Support and Context-Aware Segmentation
|
| Features:
|
| - Multi-engine PyThaiNLP integration
|
| - Context-aware word segmentation
|
| - Domain-specific dictionaries
|
| - Subword tokenization (BPE)
|
| - Intelligent caching system
|
| - Quality metrics and evaluation
|
| - Adaptive learning from corrections
|
| """
|
|
|
| def __init__(self, vocab_size: int = 50000, min_freq: int = 1, enable_cache: bool = True):
|
| self.vocab_size = vocab_size
|
| self.min_freq = min_freq
|
| self.vocab: Dict[str, int] = {}
|
| self.reverse_vocab: Dict[int, str] = {}
|
| self.word_freq: Counter = Counter()
|
| self.syllable_patterns: Set[str] = set()
|
| self.compound_words: Set[str] = set()
|
| self.learning_enabled = True
|
| self.enable_cache = enable_cache
|
|
|
|
|
| self.tokenization_cache: Dict[str, List[str]] = {}
|
| self.cache_hits = 0
|
| self.cache_misses = 0
|
|
|
|
|
| self.pythainlp_engines = ["newmm", "longest", "multi_cut"]
|
| self.current_engine = "newmm"
|
|
|
|
|
| self.context_patterns = self._load_context_patterns()
|
|
|
|
|
| self.domain_dictionaries = self._load_domain_dictionaries()
|
|
|
|
|
| self.segmentation_quality = {
|
| "total_segments": 0,
|
| "correct_segments": 0,
|
| "avg_confidence": 0.0
|
| }
|
|
|
|
|
| self.special_tokens = {
|
| "<pad>": 0,
|
| "<bos>": 1,
|
| "<eos>": 2,
|
| "<unk>": 3,
|
| "<mask>": 4,
|
| "<newline>": 5,
|
| }
|
|
|
|
|
| self.thai_consonants = "กขฃคฅฆงจฉชซฌญฎฏฐฑฒณดตถทธนบปผฝพฟภมยรลวศษสหฬอฮ"
|
| self.thai_vowels = "ะาำิีึืุูเแโใไๅๆ็่้๊๋์"
|
| self.thai_tonemarks = "่้๊๋์"
|
|
|
|
|
| self.thai_dictionary = self._load_thai_dictionary()
|
| self.max_word_length = max(len(word) for word in self.thai_dictionary) if self.thai_dictionary else 10
|
|
|
|
|
| self._init_vocab()
|
|
|
| def _init_vocab(self):
|
| """Initialize vocabulary with special tokens."""
|
| self.vocab = self.special_tokens.copy()
|
| self.reverse_vocab = {v: k for k, v in self.vocab.items()}
|
| self.next_token_id = len(self.special_tokens)
|
|
|
| def _is_thai_text(self, text: str) -> bool:
|
| """Check if text contains Thai characters."""
|
| thai_chars = set(self.thai_consonants + "ะาำิีึืุูเแโใไๅๆ็่้๊๋์")
|
| return any(char in thai_chars for char in text)
|
|
|
| def _extract_syllables(self, text: str) -> List[str]:
|
| """Extract syllables from Thai text using linguistic patterns."""
|
| syllables = []
|
|
|
|
|
| words = text.split()
|
|
|
| for word in words:
|
| if len(word) <= 2:
|
|
|
| syllables.append(word)
|
| elif len(word) <= 4:
|
|
|
| if word in self.vocab or self.word_freq.get(word, 0) > 0:
|
| syllables.append(word)
|
| else:
|
| syllables.extend(list(word))
|
| else:
|
|
|
| word_syllables = self._segment_word_into_syllables(word)
|
| syllables.extend(word_syllables)
|
|
|
| return syllables
|
|
|
| def _segment_word_into_syllables(self, word: str) -> List[str]:
|
| """Segment a long Thai word into syllables using advanced linguistic rules."""
|
| if not word:
|
| return []
|
|
|
|
|
| if word in self.vocab or self.word_freq.get(word, 0) > 0:
|
| return [word]
|
|
|
|
|
| syllables = []
|
| i = 0
|
|
|
| while i < len(word):
|
| syllable = self._extract_single_syllable(word, i)
|
| if syllable:
|
| syllables.append(syllable)
|
| i += len(syllable)
|
| else:
|
|
|
| syllables.append(word[i])
|
| i += 1
|
|
|
| return syllables
|
|
|
| def _extract_single_syllable(self, word: str, start_pos: int) -> str:
|
| """Extract a single syllable from word starting at given position."""
|
| if start_pos >= len(word):
|
| return ""
|
|
|
|
|
| patterns = [
|
|
|
| self._match_pattern(word, start_pos, "CVC"),
|
| self._match_pattern(word, start_pos, "CV"),
|
| self._match_pattern(word, start_pos, "CCV"),
|
| self._match_pattern(word, start_pos, "CCVC"),
|
| self._match_pattern(word, start_pos, "CVCC"),
|
| self._match_pattern(word, start_pos, "CC"),
|
| ]
|
|
|
|
|
| for pattern in patterns:
|
| if pattern:
|
| return pattern
|
|
|
|
|
| return self._enhanced_fallback_extraction(word, start_pos)
|
|
|
| def _match_pattern(self, word: str, start: int, pattern: str) -> str:
|
| """Match Thai syllable pattern starting from given position."""
|
| if start >= len(word):
|
| return ""
|
|
|
| i = start
|
| matched = ""
|
|
|
| for pattern_char in pattern:
|
| if i >= len(word):
|
| return ""
|
|
|
| if pattern_char == 'C':
|
| if word[i] in self.thai_consonants:
|
| matched += word[i]
|
| i += 1
|
| else:
|
| return ""
|
| elif pattern_char == 'V':
|
| if word[i] in (self.thai_vowels + self.thai_tonemarks):
|
| matched += word[i]
|
| i += 1
|
| else:
|
| return ""
|
| else:
|
| return ""
|
|
|
| return matched
|
|
|
| def _fallback_syllable_extraction(self, word: str, start: int) -> str:
|
| """Fallback syllable extraction when patterns don't match."""
|
| if start >= len(word):
|
| return ""
|
|
|
|
|
| i = start
|
|
|
|
|
| while i < len(word) and word[i] in self.thai_consonants and (i - start) < 2:
|
| i += 1
|
|
|
|
|
| while i < len(word) and word[i] in (self.thai_vowels + self.thai_tonemarks):
|
| i += 1
|
|
|
|
|
| if i < len(word) and word[i] in self.thai_consonants:
|
| i += 1
|
|
|
| syllable = word[start:i]
|
| return syllable if syllable else word[start]
|
|
|
| def _enhanced_fallback_extraction(self, word: str, start: int) -> str:
|
| """Enhanced fallback extraction for Thai text."""
|
| if start >= len(word):
|
| return ""
|
|
|
|
|
| i = start
|
|
|
|
|
| if i < len(word) and word[i] in self.thai_consonants:
|
| syllable = word[i]
|
| i += 1
|
|
|
|
|
| while i < len(word) and word[i] in (self.thai_vowels + self.thai_tonemarks):
|
| syllable += word[i]
|
| i += 1
|
|
|
|
|
| if i < len(word) and word[i] in self.thai_consonants:
|
| syllable += word[i]
|
| i += 1
|
|
|
| if len(syllable) > 1:
|
| return syllable
|
|
|
|
|
|
|
| for length in range(min(4, len(word) - start), 0, -1):
|
| candidate = word[start:start + length]
|
|
|
| if any(c in (self.thai_consonants + self.thai_vowels) for c in candidate):
|
| return candidate
|
|
|
|
|
| return word[start]
|
|
|
| def _load_context_patterns(self) -> Dict[str, List[str]]:
|
| """Load context-aware segmentation patterns."""
|
| return {
|
| "question": ["อะไร", "ทำไม", "อย่างไร", "เมื่อไหร่", "ที่ไหน", "ใคร"],
|
| "greeting": ["สวัสดี", "ครับ", "คะ", "ค่ะ", "นะครับ", "นะคะ"],
|
| "polite": ["กรุณา", "ขอโทษ", "ขอบคุณ", "ยินดี", "ด้วย"],
|
| "time": ["วันนี้", "เมื่อวาน", "พรุ่งนี้", "เช้า", "บ่าย", "เย็น", "คืน"],
|
| "quantity": ["หนึ่ง", "สอง", "สาม", "ทั้งหมด", "หลาย", "บาง"],
|
| "negation": ["ไม่", "ไม่ได้", "ไม่มี", "ไม่ได้"],
|
| "emphasis": ["มาก", "ที่สุด", "จริงๆ", "แน่นอน", "จริง"]
|
| }
|
|
|
| def _load_domain_dictionaries(self) -> Dict[str, Set[str]]:
|
| """Load domain-specific dictionaries."""
|
| return {
|
| "tech": {
|
| "ปัญญาประดิษฐ์", "แมชชีนเลิร์นนิง", "ดีปเลิร์นนิง", "อัลกอริทึม",
|
| "โมเดล", "ข้อมูล", "การฝึก", "การทดสอบ", "ความแม่นยำ", "คลัสเตอร์",
|
| "นิวรัลเน็ตเวิร์ก", "บิ๊กดาต้า", "ไอโอที", "บล็อกเชน", "คลาวด์"
|
| },
|
| "medical": {
|
| "โรงพยาบาล", "แพทย์", "พยาบาล", "ยา", "อาการ", "ป่วย", "หาย",
|
| "สุขภาพ", "โรค", "อาการป่วย", "การรักษา", "ผู้ป่วย", "การวินิจฉัย"
|
| },
|
| "business": {
|
| "บริษัท", "ธุรกิจ", "ลูกค้า", "ขาย", "ซื้อ", "เงิน", "ราคา",
|
| "ตลาด", "งาน", "เงินเดือน", "สัญญา", "เอกสาร", "กำไร", "ขาดทุน"
|
| },
|
| "education": {
|
| "โรงเรียน", "มหาวิทยาลัย", "นักเรียน", "นักศึกษา", "ครู", "อาจารย์",
|
| "เรียน", "สอน", "สอบ", "เกรด", "วิชา", "ห้องเรียน", "ห้องสมุด"
|
| }
|
| }
|
|
|
| def _load_thai_dictionary(self) -> Set[str]:
|
| """Load comprehensive Thai word dictionary for better segmentation."""
|
|
|
| thai_words = {
|
|
|
| "สวัสดี", "ครับ", "คะ", "ค่ะ", "ครับ", "นะ", "นะครับ", "นะคะ",
|
| "และ", "หรือ", "แต่", "แล้ว", "ก็", "ได้", "มี", "เป็น", "ไป", "มา",
|
| "จะ", "ต้อง", "ไม่", "ใช่", "ใช่ไหม", "ทำไม", "อะไร", "เมื่อไหร่",
|
| "ที่ไหน", "อย่างไร", "อย่างไรบ้าง", "สบายดี", "ขอบคุณ", "ยินดี",
|
|
|
|
|
| "คอมพิวเตอร์", "อินเทอร์เน็ต", "โทรศัพท์", "มือถือ", "แท็บเล็ต",
|
| "ปัญญาประดิษฐ์", "แมชชีนเลิร์นนิง", "ดีปเลิร์นนิง", "ข้อมูล",
|
| "โปรแกรม", "ซอฟต์แวร์", "ฮาร์ดแวร์", "ระบบ", "เครือข่าย",
|
|
|
|
|
| "โรงเรียน", "มหาวิทยาลัย", "นักเรียน", "นักศึกษา", "ครู", "อาจารย์",
|
| "เรียน", "สอน", "สอบ", "เกรด", "วิชา", "ห้องเรียน", "ห้องสมุด",
|
|
|
|
|
| "บริษัท", "ธุรกิจ", "ลูกค้า", "ขาย", "ซื้อ", "เงิน", "ราคา",
|
| "ตลาด", "งาน", "เงินเดือน", "สัญญา", "สัญญา", "เอกสาร",
|
|
|
|
|
| "โรงพยาบาล", "แพทย์", "พยาบาล", "ยา", "อาการ", "ป่วย", "หาย",
|
| "สุขภาพ", "ออกกำลังกาย", "อาหาร", "ดื่ม", "นอน",
|
|
|
|
|
| "วันนี้", "เมื่อวาน", "พรุ่งนี้", "เช้า", "บ่าย", "เย็น", "คืน",
|
| "อาทิตย์", "เดือน", "ปี", "ชั่วโมง", "นาที", "วินาที",
|
|
|
|
|
| "หนึ่ง", "สอง", "สาม", "สี่", "ห้า", "หก", "เจ็ด", "แปด", "เก้า", "สิบ",
|
| "ร้อย", "พัน", "หมื่น", "แสน", "ล้าน", "หลาย", "ทั้งหมด", "บาง",
|
|
|
|
|
| "ใคร", "อะไร", "ที่ไหน", "เมื่อไหร่", "ทำไม", "อย่างไร", "เท่าไหร่",
|
|
|
|
|
| "ใน", "บน", "ใต้", "ข้าง", "หลัง", "หน้า", "ระหว่าง", "กับ",
|
| "จาก", "ถึง", "โดย", "สำหรับ", "เกี่ยวกับ", "เนื่องจาก",
|
|
|
|
|
| "ดี", "เลว", "ใหญ่", "เล็ก", "สวย", "งาม", "เร็ว", "ช้า",
|
| "ร้อน", "เย็น", "ง่าย", "ยาก", "แพง", "ถูก", "ใหม่", "เก่า",
|
|
|
|
|
| "ทำ", "ไป", "มา", "เห็น", "ได้ยิน", "พูด", "เขียน", "อ่าน",
|
| "กิน", "ดื่ม", "นั่ง", "ยืน", "นอน", "วิ่ง", "เดิน", "ขับ",
|
|
|
|
|
| "บ้าน", "รถ", "แมว", "สุนัข", "คน", "เด็ก", "ผู้ใหญ่", "ผู้หญิง", "ผู้ชาย",
|
| "แม่", "พ่อ", "พี่", "น้อง", "เพื่อน", "ครอบครัว", "เมือง", "ประเทศ",
|
|
|
|
|
| "กรุงเทพ", "เชียงใหม่", "ภูเก็ต", "พัทยา", "หัวหิน", "เกาะสมุย",
|
| "ประเทศไทย", "อเมริกา", "ยุโรป", "เอเชีย", "แอฟริกา",
|
|
|
|
|
| "ข้าว", "น้ำ", "กาแฟ", "ชา", "นม", "เนื้อ", "ปลา", "ผัก", "ผลไม้",
|
| "อาหาร", "เครื่องดื่ม", "ร้านอาหาร", "ตลาด", "ซุปเปอร์มาร์เก็ต",
|
|
|
|
|
| "แอพพลิเคชัน", "เว็บไซต์", "อีเมล", "โซเชียลมีเดีย", "ยูทูบ",
|
| "เฟซบุ๊ก", "ทวิตเตอร์", "อินสตาแกรม", "ไลน์", "วีแชต",
|
|
|
|
|
| "โมเดล", "อัลกอริทึม", "ข้อมูล", "การฝึก", "การทดสอบ", "ความแม่นยำ",
|
| "การทำนาย", "การจำแนก", "การถดถอย", "คลัสเตอร์", "นิวรัลเน็ตเวิร์ก"
|
| }
|
|
|
| return set(thai_words)
|
|
|
| def _longest_matching_segmentation(self, text: str) -> List[str]:
|
| """Apply longest matching algorithm for Thai word segmentation."""
|
| if not text:
|
| return []
|
|
|
| words = []
|
| i = 0
|
|
|
| while i < len(text):
|
|
|
| found = False
|
| for length in range(min(self.max_word_length, len(text) - i), 0, -1):
|
| candidate = text[i:i + length]
|
| if candidate in self.thai_dictionary or candidate in self.vocab:
|
| words.append(candidate)
|
| i += length
|
| found = True
|
| break
|
|
|
| if not found:
|
|
|
| syllable = self._extract_single_syllable(text, i)
|
| if syllable:
|
| words.append(syllable)
|
| i += len(syllable)
|
| else:
|
|
|
| words.append(text[i])
|
| i += 1
|
|
|
| return words
|
|
|
| def _learn_new_words(self, text: str):
|
| """Learn new words from text and add to vocabulary if they meet criteria."""
|
| if not self.learning_enabled:
|
| return
|
|
|
| words = self._extract_words(text)
|
|
|
| for word in words:
|
| if len(word.strip()) > 0:
|
| self.word_freq[word] += 1
|
|
|
|
|
| if (self.word_freq[word] >= self.min_freq and
|
| word not in self.vocab and
|
| len(self.vocab) < self.vocab_size):
|
|
|
| self._add_token(word)
|
| logger.debug(f"Learned new word: '{word}' (freq: {self.word_freq[word]})")
|
|
|
| def _extract_words(self, text: str) -> List[str]:
|
| """Extract words from text, handling Thai word boundaries with enhanced PyThaiNLP."""
|
|
|
| if self._is_thai_text(text):
|
| if PYTHAINLP_AVAILABLE:
|
| try:
|
| words = word_tokenize(text, engine="newmm")
|
|
|
| filtered_words = []
|
| for w in words:
|
| w = w.strip()
|
| if len(w) >= 2 and w and not w.isdigit():
|
|
|
| if self.learning_enabled and w not in self.vocab:
|
| self._add_token_immediately(w)
|
| filtered_words.append(w)
|
| return filtered_words
|
| except Exception as e:
|
| logger.warning(f"PyThaiNLP word extraction failed: {e}. Using enhanced fallback.")
|
| return self._enhanced_pythainlp_fallback(text)
|
| else:
|
|
|
| words = re.findall(r'\b\w+\b', text)
|
| filtered_words = []
|
| for word in words:
|
| word = word.strip()
|
| if len(word) >= 2 and not word.isdigit():
|
| filtered_words.append(word)
|
| return filtered_words
|
|
|
| def _extract_thai_words(self, text: str) -> List[str]:
|
| """Extract Thai words using syllable and linguistic patterns."""
|
| words = []
|
|
|
|
|
| chunks = text.split()
|
|
|
| for chunk in chunks:
|
| chunk = chunk.strip()
|
| if not chunk:
|
| continue
|
|
|
|
|
| if len(chunk) <= 4:
|
| if len(chunk) >= 2:
|
| words.append(chunk)
|
| continue
|
|
|
|
|
|
|
| potential_words = []
|
|
|
|
|
| for i in range(2, len(chunk) - 1):
|
| if i < len(chunk):
|
| word1 = chunk[:i]
|
| word2 = chunk[i:]
|
|
|
|
|
| if (len(word1) >= 2 and len(word2) >= 2 and
|
| self._is_valid_thai_word(word1) and self._is_valid_thai_word(word2)):
|
| potential_words.append((word1, word2))
|
|
|
| if potential_words:
|
|
|
| word1, word2 = potential_words[0]
|
| words.extend([word1, word2])
|
| else:
|
|
|
| syllables = self._extract_syllables(chunk)
|
| words.extend([s for s in syllables if len(s) >= 2])
|
|
|
| return words
|
|
|
| def _is_valid_thai_word(self, word: str) -> bool:
|
| """Check if a string looks like a valid Thai word."""
|
| if len(word) < 2:
|
| return False
|
|
|
|
|
| has_consonant = any(c in self.thai_consonants for c in word)
|
|
|
|
|
| has_vowel = any(c in (self.thai_vowels + self.thai_tonemarks) for c in word)
|
|
|
| return has_consonant and has_vowel
|
|
|
| def _add_token(self, token: str):
|
| """Add a new token to vocabulary with duplicate prevention."""
|
| if token in self.vocab:
|
| return
|
|
|
|
|
| if not hasattr(self, 'next_token_id'):
|
| self.next_token_id = len(self.special_tokens)
|
|
|
| token_id = self.next_token_id
|
|
|
|
|
| if token_id >= self.vocab_size:
|
| logger.warning(f"Cannot add token '{token}': vocab size limit reached ({self.vocab_size})")
|
| return
|
|
|
|
|
| while token_id in self.reverse_vocab:
|
| token_id += 1
|
| if token_id >= self.vocab_size:
|
| logger.warning(f"Cannot add token '{token}': no available token IDs")
|
| return
|
|
|
| self.vocab[token] = token_id
|
| self.reverse_vocab[token_id] = token
|
| self.next_token_id = token_id + 1
|
| logger.info(f"Added new token: '{token}' -> {token_id}")
|
|
|
| def _add_token_immediately(self, token: str):
|
| """Add token immediately without frequency checking for PyThaiNLP integration."""
|
| if token in self.vocab:
|
| return
|
|
|
|
|
| if not hasattr(self, 'next_token_id'):
|
| self.next_token_id = len(self.special_tokens)
|
|
|
| token_id = self.next_token_id
|
|
|
|
|
| if token_id >= self.vocab_size:
|
| return
|
|
|
|
|
| while token_id in self.reverse_vocab:
|
| token_id += 1
|
| if token_id >= self.vocab_size:
|
| return
|
|
|
| self.vocab[token] = token_id
|
| self.reverse_vocab[token_id] = token
|
| self.next_token_id = token_id + 1
|
| logger.debug(f"Immediately added token: '{token}' -> {token_id}")
|
|
|
| def _enhanced_pythainlp_fallback(self, text: str) -> List[str]:
|
| """Enhanced fallback segmentation when PyThaiNLP is not available."""
|
| words = []
|
|
|
|
|
| chunks = text.split()
|
|
|
| for chunk in chunks:
|
| chunk = chunk.strip()
|
| if not chunk:
|
| continue
|
|
|
|
|
| if len(chunk) <= 4:
|
| if len(chunk) >= 2:
|
| words.append(chunk)
|
| continue
|
|
|
|
|
| dict_words = self._dictionary_based_segmentation(chunk)
|
| if dict_words:
|
| words.extend(dict_words)
|
| else:
|
|
|
| syllables = self._extract_syllables(chunk)
|
| words.extend([s for s in syllables if len(s) >= 2])
|
|
|
| return words
|
|
|
| def _dictionary_based_segmentation(self, text: str) -> List[str]:
|
| """Segment text using dictionary lookup."""
|
| if not text:
|
| return []
|
|
|
| words = []
|
| i = 0
|
|
|
| while i < len(text):
|
|
|
| found = False
|
| for length in range(min(len(text) - i, 10), 0, -1):
|
| candidate = text[i:i + length]
|
| if candidate in self.thai_dictionary or candidate in self.vocab:
|
| words.append(candidate)
|
| i += length
|
| found = True
|
| break
|
|
|
| if not found:
|
|
|
| words.append(text[i])
|
| i += 1
|
|
|
| return words
|
|
|
| def _multi_engine_segmentation(self, text: str, domain: str = "general") -> List[str]:
|
| """Use multiple PyThaiNLP engines and select the best result."""
|
| if not PYTHAINLP_AVAILABLE:
|
| return self._ultra_enhanced_fallback(text, domain)
|
|
|
| best_words = []
|
| best_score = 0.0
|
|
|
|
|
| for engine in self.pythainlp_engines:
|
| try:
|
| words = word_tokenize(text, engine=engine)
|
| score = self._evaluate_segmentation_quality(words, text, domain)
|
|
|
| if score > best_score:
|
| best_score = score
|
| best_words = words
|
| logger.debug(f"Better segmentation found with {engine}: score {score:.2f}")
|
|
|
| except Exception as e:
|
| logger.debug(f"Engine {engine} failed: {e}")
|
| continue
|
|
|
|
|
| if not best_words or best_score < 0.5:
|
| logger.debug("Using fallback segmentation")
|
| best_words = self._ultra_enhanced_fallback(text, domain)
|
|
|
| return best_words
|
|
|
| def _context_aware_postprocessing(self, words: List[str], original_text: str) -> List[str]:
|
| """Apply context-aware post-processing to improve segmentation."""
|
| if not words:
|
| return words
|
|
|
| processed_words = []
|
|
|
| for i, word in enumerate(words):
|
|
|
| if word in self.context_patterns.get("question", []):
|
|
|
| processed_words.append(word)
|
| elif word in self.context_patterns.get("polite", []):
|
|
|
| processed_words.append(word)
|
| elif word in self.context_patterns.get("time", []):
|
|
|
| processed_words.append(word)
|
| else:
|
|
|
| context_split = self._context_based_split(word, i, words)
|
| if context_split:
|
| processed_words.extend(context_split)
|
| else:
|
| processed_words.append(word)
|
|
|
| return processed_words
|
|
|
| def _domain_specific_enhancement(self, words: List[str], domain: str) -> List[str]:
|
| """Enhance segmentation based on domain-specific knowledge."""
|
| if domain not in self.domain_dictionaries:
|
| return words
|
|
|
| domain_dict = self.domain_dictionaries[domain]
|
| enhanced_words = []
|
|
|
| for word in words:
|
|
|
| if word in domain_dict:
|
| enhanced_words.append(word)
|
| else:
|
|
|
| domain_split = self._find_domain_compounds(word, domain_dict)
|
| if domain_split:
|
| enhanced_words.extend(domain_split)
|
| else:
|
| enhanced_words.append(word)
|
|
|
| return enhanced_words
|
|
|
| def _assess_segmentation_quality(self, words: List[str], original_text: str) -> float:
|
| """Assess the quality of segmentation result."""
|
| if not words:
|
| return 0.0
|
|
|
|
|
| total_chars = len(original_text.replace(" ", ""))
|
| segmented_chars = sum(len(word) for word in words)
|
|
|
|
|
| coverage = segmented_chars / total_chars if total_chars > 0 else 0.0
|
|
|
|
|
| avg_word_length = segmented_chars / len(words) if words else 0.0
|
|
|
|
|
| dict_coverage = sum(1 for word in words if word in self.thai_dictionary) / len(words)
|
|
|
|
|
| quality_score = (coverage * 0.4) + (min(avg_word_length / 5.0, 1.0) * 0.3) + (dict_coverage * 0.3)
|
|
|
|
|
| self.segmentation_quality["total_segments"] += len(words)
|
| self.segmentation_quality["correct_segments"] += int(dict_coverage * len(words))
|
| self.segmentation_quality["avg_confidence"] = (
|
| self.segmentation_quality["avg_confidence"] + quality_score
|
| ) / 2
|
|
|
| return quality_score
|
|
|
| def _ultra_enhanced_fallback(self, text: str, domain: str = "general") -> List[str]:
|
| """Ultra-enhanced fallback when PyThaiNLP is not available."""
|
| words = []
|
|
|
|
|
| chunks = text.split()
|
|
|
| for chunk in chunks:
|
| chunk = chunk.strip()
|
| if not chunk:
|
| continue
|
|
|
|
|
| if len(chunk) <= 4:
|
| if len(chunk) >= 2:
|
| words.append(chunk)
|
| continue
|
|
|
|
|
| if domain in self.domain_dictionaries:
|
| domain_words = self._domain_based_segmentation(chunk, self.domain_dictionaries[domain])
|
| if domain_words:
|
| words.extend(domain_words)
|
| continue
|
|
|
|
|
| dict_words = self._dictionary_based_segmentation(chunk)
|
| if dict_words:
|
| words.extend(dict_words)
|
| else:
|
|
|
| syllables = self._extract_syllables(chunk)
|
| words.extend([s for s in syllables if len(s) >= 2])
|
|
|
| return words
|
|
|
| def _ultra_intelligent_segmentation(self, word: str, domain: str = "general") -> List[str]:
|
| """Ultra-intelligent segmentation for unknown words."""
|
| if not word:
|
| return []
|
|
|
|
|
| if len(word) <= 10 and self.learning_enabled:
|
| self._add_token_immediately(word)
|
| return [word]
|
|
|
|
|
| if domain in self.domain_dictionaries:
|
| domain_segments = self._domain_based_segmentation(word, self.domain_dictionaries[domain])
|
| if domain_segments:
|
| return domain_segments
|
|
|
|
|
| segments = self._dictionary_based_segmentation(word)
|
| if segments and len(segments) > 1:
|
| return segments
|
|
|
|
|
| syllables = self._context_aware_syllable_segmentation(word)
|
| if syllables:
|
| return syllables
|
|
|
|
|
| return list(word)
|
|
|
| def _enhanced_non_thai_tokenization(self, text: str) -> List[str]:
|
| """Enhanced tokenization for non-Thai text."""
|
|
|
| import re
|
| words = re.findall(r'\b\w+(?:[-\']\w+)*\b', text)
|
|
|
| filtered_words = []
|
| for word in words:
|
| word = word.strip()
|
| if len(word) >= 2 and not word.isdigit():
|
| filtered_words.append(word)
|
|
|
| return filtered_words
|
|
|
| def _evaluate_segmentation_quality(self, words: List[str], text: str, domain: str) -> float:
|
| """Evaluate segmentation quality for multi-engine selection."""
|
| return self._assess_segmentation_quality(words, text)
|
|
|
| def _context_based_split(self, word: str, position: int, all_words: List[str]) -> List[str]:
|
| """Split word based on context."""
|
|
|
| if len(word) > 6:
|
|
|
| for i in range(2, len(word) - 1):
|
| if i < len(word):
|
| part1 = word[:i]
|
| part2 = word[i:]
|
|
|
|
|
| if (part1 in self.thai_dictionary or len(part1) <= 3) and \
|
| (part2 in self.thai_dictionary or len(part2) <= 3):
|
| return [part1, part2]
|
|
|
| return None
|
|
|
| def _find_domain_compounds(self, word: str, domain_dict: Set[str]) -> List[str]:
|
| """Find domain-specific compound words."""
|
|
|
| for i in range(2, len(word)):
|
| if i < len(word):
|
| part1 = word[:i]
|
| part2 = word[i:]
|
|
|
| if part1 in domain_dict and part2 in domain_dict:
|
| return [part1, part2]
|
|
|
| return None
|
|
|
| def _domain_based_segmentation(self, text: str, domain_dict: Set[str]) -> List[str]:
|
| """Segment text using domain-specific dictionary."""
|
| if not text:
|
| return []
|
|
|
| words = []
|
| i = 0
|
|
|
| while i < len(text):
|
| found = False
|
|
|
| for length in range(min(len(text) - i, 12), 0, -1):
|
| candidate = text[i:i + length]
|
| if candidate in domain_dict:
|
| words.append(candidate)
|
| i += length
|
| found = True
|
| break
|
|
|
| if not found:
|
|
|
| words.append(text[i])
|
| i += 1
|
|
|
| return words
|
|
|
| def _context_aware_syllable_segmentation(self, word: str) -> List[str]:
|
| """Context-aware syllable segmentation."""
|
|
|
| syllables = []
|
| i = 0
|
|
|
| while i < len(word):
|
|
|
| for length in range(min(4, len(word) - i), 0, -1):
|
| candidate = word[i:i + length]
|
|
|
|
|
| if self._is_valid_thai_syllable(candidate):
|
| syllables.append(candidate)
|
| i += length
|
| break
|
| else:
|
|
|
| syllables.append(word[i])
|
| i += 1
|
|
|
| return syllables
|
|
|
| def _is_valid_thai_syllable(self, syllable: str) -> bool:
|
| """Check if a string is a valid Thai syllable."""
|
| if len(syllable) < 1 or len(syllable) > 4:
|
| return False
|
|
|
|
|
| has_thai = any(c in (self.thai_consonants + self.thai_vowels) for c in syllable)
|
|
|
|
|
| if len(syllable) > 2:
|
| has_vowel = any(c in (self.thai_vowels + self.thai_tonemarks) for c in syllable)
|
| return has_thai and has_vowel
|
| else:
|
| return has_thai
|
|
|
| def fit(self, texts: List[str], epochs: int = 1):
|
| """Train the tokenizer on a corpus of texts."""
|
| logger.info(f"Training tokenizer on {len(texts)} texts for {epochs} epochs")
|
|
|
|
|
| logger.info("Building word frequency statistics...")
|
| all_words = []
|
| for text in texts:
|
| words = self._extract_words(text)
|
| all_words.extend(words)
|
|
|
|
|
| for word in all_words:
|
| self.word_freq[word] += 1
|
|
|
|
|
| sorted_words = sorted(self.word_freq.items(), key=lambda x: x[1], reverse=True)
|
|
|
| for epoch in range(epochs):
|
| logger.info(f"Epoch {epoch + 1}/{epochs}")
|
|
|
| words_added = 0
|
| for word, freq in sorted_words:
|
| if freq >= self.min_freq and word not in self.vocab:
|
| if len(self.vocab) < self.vocab_size:
|
| self._add_token(word)
|
| words_added += 1
|
| else:
|
| break
|
|
|
| if words_added % 100 == 0 and words_added > 0:
|
| logger.info(f"Added {words_added} words to vocabulary")
|
|
|
| logger.info(f"Training complete. Vocabulary size: {len(self.vocab)}")
|
| logger.info(f"Total unique words learned: {len(self.word_freq)}")
|
|
|
| def _tokenize_single_line(self, text: str, domain: str = "general") -> List[str]:
|
| """Tokenize a single line of text without newlines."""
|
| tokens = []
|
|
|
|
|
| if self._is_thai_text(text):
|
| if PYTHAINLP_AVAILABLE:
|
|
|
| words = self._multi_engine_segmentation(text, domain)
|
|
|
|
|
| words = self._context_aware_postprocessing(words, text)
|
|
|
|
|
| words = self._domain_specific_enhancement(words, domain)
|
|
|
|
|
| quality_score = self._assess_segmentation_quality(words, text)
|
| logger.debug(f"Segmentation quality: {quality_score:.2f}")
|
|
|
|
|
| if self.learning_enabled:
|
| for word in words:
|
| if len(word.strip()) >= 2 and word not in self.vocab:
|
| self._add_token_immediately(word)
|
|
|
| else:
|
|
|
| words = self._ultra_enhanced_fallback(text, domain)
|
|
|
|
|
| for word in words:
|
| word = word.strip()
|
| if not word:
|
| continue
|
|
|
| if word in self.vocab:
|
| tokens.append(word)
|
| elif len(word) <= 3:
|
|
|
| if self.learning_enabled and len(word) >= 2:
|
| self._add_token_immediately(word)
|
| tokens.append(word)
|
| else:
|
| tokens.extend(list(word))
|
| else:
|
|
|
| segmented = self._ultra_intelligent_segmentation(word, domain)
|
| tokens.extend(segmented)
|
| else:
|
|
|
| words = self._enhanced_non_thai_tokenization(text)
|
| for word in words:
|
| if word in self.vocab:
|
| tokens.append(word)
|
| else:
|
| tokens.extend(list(word))
|
|
|
|
|
|
|
|
|
|
|
|
|
| if self.learning_enabled:
|
| self._learn_new_words(text)
|
|
|
| return tokens
|
|
|
| def tokenize(self, text: str, domain: str = "general", use_cache: bool = True) -> List[str]:
|
| """Ultra-advanced tokenization with multi-engine support, context-awareness, and newline handling."""
|
| tokens = []
|
|
|
|
|
| if use_cache and self.enable_cache and text in self.tokenization_cache:
|
| self.cache_hits += 1
|
| return self.tokenization_cache[text].copy()
|
|
|
| self.cache_misses += 1
|
|
|
|
|
| if '\n' in text:
|
| lines = text.split('\n')
|
| all_tokens = []
|
| for i, line in enumerate(lines):
|
| if line.strip():
|
| line_tokens = self._tokenize_single_line(line, domain)
|
| all_tokens.extend(line_tokens)
|
|
|
| if i < len(lines) - 1:
|
| all_tokens.append("<newline>")
|
| tokens = all_tokens
|
| else:
|
| tokens = self._tokenize_single_line(text, domain)
|
|
|
|
|
| if use_cache and self.enable_cache:
|
| self.tokenization_cache[text] = tokens.copy()
|
|
|
|
|
| if self.learning_enabled:
|
| self._learn_new_words(text)
|
|
|
| return tokens
|
|
|
| def encode(self, text: str) -> List[int]:
|
| """Convert text to token IDs."""
|
| tokens = self.tokenize(text)
|
| token_ids = []
|
|
|
| for token in tokens:
|
| if token in self.vocab:
|
| token_ids.append(self.vocab[token])
|
| else:
|
| token_ids.append(self.special_tokens["<unk>"])
|
|
|
| return token_ids
|
|
|
| def decode(self, token_ids: List[int]) -> str:
|
| """Convert token IDs back to text."""
|
| tokens = []
|
| for token_id in token_ids:
|
| if token_id in self.reverse_vocab:
|
| token = self.reverse_vocab[token_id]
|
| if token == "<newline>":
|
| tokens.append("\n")
|
| elif token not in self.special_tokens:
|
| tokens.append(token)
|
| else:
|
| tokens.append("<unk>")
|
|
|
| return "".join(tokens)
|
|
|
| def save(self, path: str):
|
| """Save tokenizer state to disk."""
|
| os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
|
| state = {
|
| 'vocab': self.vocab,
|
| 'reverse_vocab': self.reverse_vocab,
|
| 'word_freq': dict(self.word_freq),
|
| 'syllable_patterns': list(self.syllable_patterns),
|
| 'compound_words': list(self.compound_words),
|
| 'vocab_size': self.vocab_size,
|
| 'min_freq': self.min_freq,
|
| 'learning_enabled': self.learning_enabled,
|
| 'next_token_id': getattr(self, 'next_token_id', len(self.special_tokens))
|
| }
|
|
|
| with open(path, 'wb') as f:
|
| pickle.dump(state, f)
|
|
|
| logger.info(f"Tokenizer saved to {path}")
|
|
|
| def load(self, path: str):
|
| """Load tokenizer state from disk."""
|
| if not os.path.exists(path):
|
| logger.warning(f"Tokenizer file {path} not found")
|
| return
|
|
|
| with open(path, 'rb') as f:
|
| state = pickle.load(f)
|
|
|
| self.vocab = state['vocab']
|
| self.reverse_vocab = state['reverse_vocab']
|
| self.word_freq = Counter(state['word_freq'])
|
| self.syllable_patterns = set(state['syllable_patterns'])
|
| self.compound_words = set(state['compound_words'])
|
| self.vocab_size = state['vocab_size']
|
| self.min_freq = state['min_freq']
|
| self.learning_enabled = state['learning_enabled']
|
| self.next_token_id = state.get('next_token_id', len(self.special_tokens))
|
|
|
| logger.info(f"Tokenizer loaded from {path}")
|
|
|
| def get_vocab_size(self) -> int:
|
| """Get current vocabulary size."""
|
| return len(self.vocab)
|
|
|
| def get_stats(self) -> Dict:
|
| """Get comprehensive tokenizer statistics."""
|
| cache_hit_rate = 0.0
|
| if self.cache_hits + self.cache_misses > 0:
|
| cache_hit_rate = self.cache_hits / (self.cache_hits + self.cache_misses)
|
|
|
| return {
|
| 'vocab_size': len(self.vocab),
|
| 'total_words_learned': len(self.word_freq),
|
| 'most_common_words': self.word_freq.most_common(10),
|
| 'learning_enabled': self.learning_enabled,
|
| 'special_tokens': list(self.special_tokens.keys()),
|
| 'pythainlp_available': PYTHAINLP_AVAILABLE,
|
| 'thai_dictionary_size': len(self.thai_dictionary) if hasattr(self, 'thai_dictionary') else 0,
|
| 'cache_enabled': self.enable_cache,
|
| 'cache_hit_rate': cache_hit_rate,
|
| 'cache_hits': self.cache_hits,
|
| 'cache_misses': self.cache_misses,
|
| 'current_engine': self.current_engine,
|
| 'pythainlp_engines': self.pythainlp_engines,
|
| 'domain_dictionaries': list(self.domain_dictionaries.keys()),
|
| 'segmentation_quality': self.segmentation_quality,
|
| 'context_patterns': list(self.context_patterns.keys())
|
| }
|
|
|
| def enable_learning(self):
|
| """Enable continuous learning mode."""
|
| self.learning_enabled = True
|
| logger.info("Continuous learning enabled")
|
|
|
| def disable_learning(self):
|
| """Disable continuous learning mode."""
|
| self.learning_enabled = False
|
| logger.info("Continuous learning disabled")
|
|
|
|
|
| def create_hf_tokenizer_files(tokenizer: AdvancedThaiTokenizer, output_dir: str):
|
| """Create Hugging Face compatible tokenizer files."""
|
|
|
| os.makedirs(output_dir, exist_ok=True)
|
|
|
|
|
| tokenizer_json = {
|
| "version": "1.0",
|
| "truncation": None,
|
| "padding": None,
|
| "added_tokens": [
|
| {
|
| "id": tid,
|
| "content": token,
|
| "single_word": False,
|
| "lstrip": False,
|
| "rstrip": False,
|
| "normalized": False,
|
| "special": token in ["<pad>", "<bos>", "<eos>", "<unk>", "<mask>", "<newline>"]
|
| }
|
| for token, tid in tokenizer.vocab.items()
|
| ],
|
| "normalizer": None,
|
| "pre_tokenizer": None,
|
| "post_processor": None,
|
| "decoder": None,
|
| "model": {
|
| "type": "BPE",
|
| "vocab": tokenizer.vocab,
|
| "merges": []
|
| }
|
| }
|
|
|
| with open(os.path.join(output_dir, "tokenizer.json"), 'w', encoding='utf-8') as f:
|
| json.dump(tokenizer_json, f, ensure_ascii=False, indent=2)
|
|
|
|
|
| tokenizer_config = {
|
| "add_bos_token": True,
|
| "add_eos_token": False,
|
| "bos_token": "<bos>",
|
| "eos_token": "<eos>",
|
| "pad_token": "<pad>",
|
| "unk_token": "<unk>",
|
| "clean_up_tokenization_spaces": False,
|
| "legacy": True,
|
| "model_max_length": 2048,
|
| "tokenizer_class": "PreTrainedTokenizerFast",
|
| "vocab_size": len(tokenizer.vocab),
|
| "use_default_system_prompt": False
|
| }
|
|
|
| with open(os.path.join(output_dir, "tokenizer_config.json"), 'w', encoding='utf-8') as f:
|
| json.dump(tokenizer_config, f, ensure_ascii=False, indent=2)
|
|
|
|
|
| special_tokens_map = {
|
| "bos_token": "<bos>",
|
| "eos_token": "<eos>",
|
| "pad_token": "<pad>",
|
| "unk_token": "<unk>"
|
| }
|
|
|
| with open(os.path.join(output_dir, "special_tokens_map.json"), 'w', encoding='utf-8') as f:
|
| json.dump(special_tokens_map, f, ensure_ascii=False, indent=2)
|
|
|
| logger.info(f"Hugging Face tokenizer files created in {output_dir}")
|
|
|
| if __name__ == "__main__":
|
|
|
| tokenizer = AdvancedThaiTokenizer(vocab_size=50000)
|
|
|
|
|
| sample_texts = [
|
| "สวัสดีครับ วันนี้อากาศดีมากเลย",
|
| "ฉันชอบกินข้าวมันไก่ เพราะอร่อยมาก",
|
| "การเรียนภาษาไทยเป็นเรื่องที่น่าสนใจ",
|
| "ปัญญาประดิษฐ์จะเปลี่ยนโลกใบนี้ไปตลอดกาล",
|
| "เทคโนโลยีสมัยใหม่ช่วยให้ชีวิตง่ายขึ้นมาก",
|
| "ประเทศไทยเป็นประเทศที่มีวัฒนธรรมที่หลากหลาย",
|
| "การพัฒนาประเทศต้องอาศัยการศึกษาที่ดี",
|
| "นักเรียนควรมีความรู้ทางวิทยาศาสตร์และเทคโนโลยี",
|
| "การใช้ภาษาไทยในการสื่อสารเป็นสิ่งสำคัญ",
|
| "เศรษฐกิจไทยมีการเติบโตอย่างต่อเนื่อง",
|
| "การท่องเที่ยวเป็นอุตสาหกรรมสำคัญของไทย",
|
| "วัฒนธรรมไทยมีเอกลักษณ์เฉพาะตัว",
|
| "การอนุรักษ์สิ่งแวดล้อมเป็นหน้าที่ของทุกคน",
|
| "การศึกษาเป็นรากฐานสำคัญของการพัฒนา",
|
| "นวัตกรรมและเทคโนโลยีเป็นกุญแจสู่ความสำเร็จ",
|
| "การทำงานเป็นทีมช่วยให้บรรลุเป้าหมายได้",
|
| "ความซื่อสัตย์และความรับผิดชอบเป็นคุณธรรมสำคัญ",
|
| "การใช้เวลาว่างให้เกิดประโยชน์เป็นสิ่งที่ดี",
|
| "การออกกำลังกายช่วยให้สุขภาพแข็งแรง",
|
| "การอ่านหนังสือเพิ่มพูนความรู้และปัญญา"
|
| ]
|
|
|
|
|
| tokenizer.fit(sample_texts, epochs=2)
|
|
|
|
|
| test_text = "สวัสดีครับ ผมชอบเรียนภาษาไทย"
|
| tokens = tokenizer.tokenize(test_text)
|
| token_ids = tokenizer.encode(test_text)
|
| decoded = tokenizer.decode(token_ids)
|
|
|
| print(f"Original: {test_text}")
|
| print(f"Tokens: {tokens}")
|
| print(f"Token IDs: {token_ids}")
|
| print(f"Decoded: {decoded}")
|
| print(f"Vocab size: {tokenizer.get_vocab_size()}")
|
| print(f"Stats: {tokenizer.get_stats()}")
|
|
|
|
|
| tokenizer.save("modern-thai-gpt-hf/advanced_tokenizer.pkl")
|
|
|
|
|
| create_hf_tokenizer_files(tokenizer, "modern-thai-gpt-hf/") |
