""" moe_fixes.py ============ รวม fix จาก Adversarial + Multi-turn Testing Fix ล่าสุด (Round 7): 1. เพิ่ม is_degenerate_query() + DEGENERATE_FALLBACK_RESPONSE — ตรวจจับ Query ที่ไม่มีตัวอักษรที่มีความหมายเพียงพอ (ว่างเปล่า, มีแต่ Emoji, มีแต่สัญลักษณ์เช่น "???!!!") ก่อนเข้า Pipeline ทั้งหมด แก้ปัญหา: จาก Adversarial Test พบว่า Query ประเภทนี้ (Empty, Emoji-only, Symbol-only) ถูกส่งเข้า LLM Classifier + Reasoning Expert + Translation Layer เต็มขั้นตอนทุกครั้ง ทำให้ Latency สูงถึง 4-14 วินาทีโดยไม่จำเป็น และเสี่ยง Hallucination (พบคำตอบเพี้ยนเรื่อง "crab" ที่ไม่เกี่ยวข้อง) หรือ Language Leakage (พบคำว่า "riêng" หลุดเข้ามาในคำตอบภาษาไทย) ใช้คู่กับ Fast-Path Guard ใน moe_orchestrator.py ที่ตอบกลับทันทีโดย ไม่ต้อง Generate เลย (Latency ≈ 0s) Fix รอบก่อนหน้า (Round 6 — คงเดิม): 2. แก้ Bug: GREETING_CLOSING_WORDS ใช้ Substring Match ทำให้ "บาย" (bye) จับ False Positive กับคำว่า "อธิบาย" (explain) — แยกเป็น 2 กลุ่มตาม ความเสี่ยง (Safe Substring / ต้อง Exact Match) Fix รอบก่อนหน้านั้น (Round 5 — คงเดิม): 3. route_query() รับ previous_expert — History Continuation แทนการผ่าน LLM Classifier ทุกครั้งที่ Query ไม่มี Keyword ชัดเจน 4. is_code_refusal() — จับ Pattern ปฏิเสธงานที่ไม่สมเหตุสมผลของ Python/TypeScript Expert คู่กับ Retry Logic ก่อนบันทึกเข้า History Fix รอบก่อนหน้านั้น (คงเดิมทั้งหมด): 5. ALLOWED_PATTERN รองรับ Emoji Unicode Range 6. strip_translation_preamble() ใช้ตำแหน่งตัวอักษรไทยตัวแรก 7. enforce_persona() บังคับ ผม/ครับ แบบ Deterministic """ import re from dataclasses import dataclass, field from typing import Optional # ====================================================================== # 0) DEGENERATE QUERY GUARD (Round 7: Fast-Path ก่อนเข้า Pipeline ทั้งหมด) # ====================================================================== # จำนวนตัวอักษรที่มีความหมาย (ตัวหนังสือ/ตัวเลข/underscore ทุกภาษา) ขั้นต่ำ # ที่ต้องมีถึงจะถือว่า Query นี้ "มีเนื้อหาให้ตอบจริง" — ต่ำกว่านี้ถือว่า # Degenerate (ว่าง/มีแต่ Emoji/มีแต่สัญลักษณ์) ไม่ควรส่งเข้า Model เลย MIN_MEANINGFUL_LENGTH = 2 # \w ใน Python 3 regex ครอบคลุม Unicode Word Character ทุกภาษาโดย Default # (รวมภาษาไทย, อังกฤษ, ตัวเลข, underscore) ส่วน Emoji/สัญลักษณ์ทั้งหมด # (Category So, Po, Sm ฯลฯ) จะถูกจับด้วย \W เพราะไม่ใช่ Word Character _NON_MEANINGFUL_REGEX = re.compile(r'\W', flags=re.UNICODE) def is_degenerate_query(query: str) -> bool: """ คืนค่า True ถ้า Query ไม่มีตัวอักษรที่มีความหมายเพียงพอให้ตอบ เช่น "", " ", "\\n\\n\\n", "🐍🐍🐍", "???!!!...", "?" คืนค่า False ถ้ามีเนื้อหาจริง เช่น "help", "ทำไง", "def foo(): pass" """ stripped = query.strip() if len(stripped) == 0: return True meaningful_chars = _NON_MEANINGFUL_REGEX.sub('', stripped) return len(meaningful_chars) < MIN_MEANINGFUL_LENGTH DEGENERATE_FALLBACK_RESPONSE = ( "ขอโทษครับ ผมไม่แน่ใจว่าคุณต้องการถามอะไรครับ " "ลองพิมพ์คำถามหรือรายละเอียดเพิ่มเติมให้ผมช่วยได้ไหมครับ" ) # ====================================================================== # 0b) GREETING / CLOSING QUICK-REPLY (Round 8: ลด Latency ของ Small-talk) # ====================================================================== # ต่างจาก is_degenerate_query() ตรงที่นี่ใช้ Exact-Match แบบเข้มงวด # เพื่อป้องกันไม่ให้ Query ที่มีเนื้อหาจริงแซมมา (เช่น "ขอบคุณสำหรับ # function นี้นะครับ") ถูก Quick-Reply ผิดพลาดจนพลาดเนื้อหาที่ User # ต้องการคุยต่อจริงๆ — ใช้เฉพาะ Query ที่เป็น Greeting/Closing ล้วนๆ # แก้ปัญหา: จาก Multi-turn Test พบว่า "สวัสดี" และ "ขอบคุณครับ" เสีย # Latency 4.48s และ 10.06s เพราะผ่าน llm_classifier + reasoning + # Translation Layer เต็มขั้นตอน ทั้งที่ไม่มีเนื้อหาให้ Generate จริง _POLITE_SUFFIXES = ["ครับ", "ค่ะ", "จ้า", "นะครับ", "นะคะ", "น้า"] GREETING_KEYWORDS = {"สวัสดี", "หวัดดี", "หวัดดีจ้า", "hi", "hello", "hey"} CLOSING_KEYWORDS = {"บาย", "บ๊ายบาย", "ลาก่อน", "แล้วเจอกัน", "bye", "goodbye"} THANKS_KEYWORDS = {"ขอบคุณ", "ขอบคุณมาก", "ขอบใจ", "thanks", "thank you", "thx"} GREETING_QUICK_REPLY_TEXT = "สวัสดีครับ มีอะไรให้ผมช่วยไหมครับ" CLOSING_QUICK_REPLY_TEXT = "ลาก่อนครับ แล้วพบกันใหม่นะครับ" THANKS_QUICK_REPLY_TEXT = "ด้วยความยินดีครับ มีอะไรให้ผมช่วยเพิ่มอีกไหมครับ" def _normalize_for_greeting_match(query: str) -> str: text = query.strip() text = re.sub(r'[!.?~๏๛]+$', '', text).strip() for suffix in _POLITE_SUFFIXES: if text.endswith(suffix): text = text[: -len(suffix)].strip() break return text.lower() def get_greeting_quick_reply(query: str) -> Optional[str]: """ คืนค่า Canned Response ถ้า query เป็น Greeting/Closing/Thanks ล้วนๆ (Exact Match หลังตัดคำลงท้ายสุภาพ) คืนค่า None ถ้ามีเนื้อหาอื่นแซมอยู่ หรือไม่เข้าเงื่อนไขใดเลย — เพื่อให้ระบบ Generate ตามปกติ """ normalized = _normalize_for_greeting_match(query) if not normalized: return None if normalized in GREETING_KEYWORDS: return GREETING_QUICK_REPLY_TEXT if normalized in CLOSING_KEYWORDS: return CLOSING_QUICK_REPLY_TEXT if normalized in THANKS_KEYWORDS: return THANKS_QUICK_REPLY_TEXT return None # ====================================================================== # 0c) REPETITION LOOP DETECTOR (Round 9: ป้องกัน Generation Collapse) # ====================================================================== # แก้ปัญหา: จาก Multi-turn Test พบว่าทั้ง Python Expert และ Reasoning # Expert สามารถหลุดเข้า "Repetition Trap" ได้ — วนพูดวลีเดิมซ้ำๆไม่รู้จบ # (เช่น "ผมขออภัยที่ทำให้คุณสับสน" ซ้ำ 50+ ครั้ง หรือรายชื่อ Google Service # ซ้ำวนไม่จบ) จนชน max_tokens ทำให้ Latency พุ่งสูงสุดถึง 39.6 วินาที # และคำตอบไม่มีเนื้อหาที่ต้องการเลย เกิดข้าม Expert คนละตัว ยืนยันว่าเป็น # ปัญหาระดับ Sampler Configuration ไม่ใช่ปัญหาเฉพาะ LoRA Adapter ตัวใด _REPETITION_LOOP_REGEX = re.compile(r'(.{6,50}?)\1{3,}', re.DOTALL) def is_repetitive_response(text: str) -> bool: """ คืนค่า True ถ้าพบ Substring ความยาว 6-50 ตัวอักษรที่ซ้ำติดกัน ตั้งแต่ 4 ครั้งขึ้นไป (Repetition Collapse) — threshold ถูกตั้งให้ ไม่ False Positive กับโค้ดที่มี Pattern ซ้ำแบบปกติ (เช่น Code Block ตัวอย่าง 2 อันที่คล้ายกัน) เพราะต้องซ้ำ 4 ครั้งขึ้นไปถึงจะ Flag """ return bool(_REPETITION_LOOP_REGEX.search(text)) # ====================================================================== # 1) WEIGHTED ROUTER # ====================================================================== @dataclass class ExpertKeywords: name: str keywords: list = field(default_factory=list) weight_multiplier: float = 1.0 EXPERT_REGISTRY = [ ExpertKeywords("python", ["python", "ไพธอน", "def ", ".py", "pip", "pandas", "numpy"]), ExpertKeywords("typescript", ["typescript", ".ts", "interface", "npm", "const ", "javascript", "react", "node.js"]), ExpertKeywords("reasoning", [ "คำนวณ", "อธิบายทีละขั้นตอน", "วิเคราะห์เหตุผล", "โจทย์คณิตศาสตร์", "แก้สมการ", "หาค่า", "step by step" ]), ] CODE_DEBUG_SIGNALS = ["โค้ด", "code", "error", "บัค", "bug", "พัง", "ฟังก์ชัน", "function"] OUTPUT_SIGNAL_WORDS = ["output เป็น", "แปลงเป็น", "เขียนเป็น", "ทำเป็น", "output as", "convert to"] AMBIGUOUS_THRESHOLD = 0.15 GREETING_SAFE_SUBSTRING_WORDS = ["สวัสดี", "ขอบคุณ", "หวัดดี", "ลาก่อน"] GREETING_EXACT_MATCH_WORDS = ["บาย", "bye", "บ๊ายบาย"] def _has_greeting_or_closing(query: str) -> bool: if any(w in query for w in GREETING_SAFE_SUBSTRING_WORDS): return True normalized = re.sub(r'[^\w\u0E00-\u0E7F]', '', query).strip().lower() return normalized in GREETING_EXACT_MATCH_WORDS def _keyword_score(query: str, expert: "ExpertKeywords") -> float: q_lower = query.lower() score = 0.0 for kw in expert.keywords: count = q_lower.count(kw.lower()) if count > 0: position = q_lower.find(kw.lower()) position_boost = 1.0 + max(0, (30 - position) / 30) * 0.3 score += count * position_boost return score * expert.weight_multiplier def route_query(query: str, previous_expert: Optional[str] = None) -> dict: """ previous_expert: Expert ที่ใช้ใน Turn ก่อนหน้า (สำหรับ Multi-turn Conversation) ถ้าไม่ส่งมา (None) ทำงานแบบ Stateless เหมือนเดิม — Backward Compatible 100% """ q_lower = query.lower() has_debug_signal = any(sig in q_lower for sig in CODE_DEBUG_SIGNALS) has_output_signal = any(sig in q_lower for sig in OUTPUT_SIGNAL_WORDS) has_greeting = _has_greeting_or_closing(query) scores = {e.name: _keyword_score(query, e) for e in EXPERT_REGISTRY} sorted_scores = sorted(scores.items(), key=lambda x: x[1], reverse=True) top_expert, top_score = sorted_scores[0] second_expert, second_score = sorted_scores[1] if len(sorted_scores) > 1 else (None, 0.0) lang_score = scores.get("python", 0.0) + scores.get("typescript", 0.0) if has_debug_signal and lang_score == 0.0: if previous_expert in ("python", "typescript") and not has_greeting: return { "expert": previous_expert, "method": "history_continuation_debug", "scores": scores, "is_ambiguous": False, } return {"expert": None, "method": "none", "scores": scores, "is_ambiguous": True} if top_score == 0.0: if previous_expert is not None and not has_greeting: return { "expert": previous_expert, "method": "history_continuation", "scores": scores, "is_ambiguous": False, } return {"expert": None, "method": "none", "scores": scores, "is_ambiguous": True} diff_ratio = (top_score - second_score) / max(top_score, 1e-6) is_ambiguous = diff_ratio < AMBIGUOUS_THRESHOLD if has_output_signal: is_ambiguous = True return { "expert": top_expert, "method": "keyword_match_weighted", "scores": scores, "is_ambiguous": is_ambiguous, } # ====================================================================== # 2) LANGUAGE PURITY FILTER (Emoji-Aware) # ====================================================================== ALLOWED_PATTERN = re.compile( r'[^\u0E00-\u0E7F' r'a-zA-Z0-9' r'\s' r'.,!?;:\'"()\[\]{}<>@#$%^&*_+=\-/\\|`~๐-๙×÷°≈≠±' r'\U0001F1E6-\U0001F1FF' r'\U0001F300-\U0001F5FF' r'\U0001F600-\U0001F64F' r'\U0001F680-\U0001F6FF' r'\U0001F700-\U0001F77F' r'\U0001F780-\U0001F7FF' r'\U0001F800-\U0001F8FF' r'\U0001F900-\U0001F9FF' r'\U0001FA00-\U0001FA6F' r'\U0001FA70-\U0001FAFF' r'\u2600-\u26FF' r'\u2700-\u27BF' r'\uFE0F' r'\u200D]' ) def find_foreign_chars(text: str) -> list: return ALLOWED_PATTERN.findall(text) def detect_language_leakage_words(text: str) -> list: words = re.findall(r'\S+', text) suspicious_words = [] for word in words: foreign = find_foreign_chars(word) if foreign: suspicious_words.append({"word": word, "foreign_chars": foreign}) return suspicious_words def clean_or_flag_response(text: str, auto_retry_callback=None) -> dict: suspicious = detect_language_leakage_words(text) if not suspicious: return {"text": text, "is_clean": True, "flagged_words": []} print(f"[WARNING] พบคำที่อาจเป็น language leakage: {suspicious}") if auto_retry_callback is not None: print("[INFO] กำลัง retry เพื่อแก้ language leakage...") new_text = auto_retry_callback() recheck = detect_language_leakage_words(new_text) if not recheck: return {"text": new_text, "is_clean": True, "flagged_words": []} print("[WARNING] Retry แล้วยังพบ leakage อยู่ — คืนคำตอบเดิมพร้อม flag") return {"text": text, "is_clean": False, "flagged_words": suspicious} # ====================================================================== # 3) HEDGE-PHRASE LIMITER # ====================================================================== HEDGE_PHRASE_PATTERN = re.compile(r'ผม(?:คิด|เชื่อ|เข้าใจ)?ว่า') def limit_hedge_phrases(text: str, max_allowed: int = 1) -> str: matches = list(HEDGE_PHRASE_PATTERN.finditer(text)) if len(matches) <= max_allowed: return text result_parts = [] last_end = 0 count = 0 for m in matches: if count < max_allowed: result_parts.append(text[last_end:m.end()]) else: result_parts.append(text[last_end:m.start()]) last_end = m.end() count += 1 result_parts.append(text[last_end:]) cleaned = ''.join(result_parts) cleaned = re.sub(r'\s{2,}', ' ', cleaned) cleaned = re.sub(r'\s+([.,!?ครับค่ะ])', r'\1', cleaned) cleaned = re.sub(r',\s*,', ',', cleaned) return cleaned.strip() # ====================================================================== # 4) REFUSAL DETECTORS # ====================================================================== REFUSAL_PATTERNS = [ "ไม่สามารถแปล", "ไม่สามารถที่จะแปล", "ยากต่อการเข้าใจถ้าแปล", "ขออภัยที่ไม่สามารถ", "ยังไม่สามารถแปลเป็นภาษาไทยได้", "ไม่ได้เป็นข้อความภาษาไทย", "cannot translate", ] def is_refusal(text: str) -> bool: return any(p in text for p in REFUSAL_PATTERNS) CODE_REFUSAL_PATTERNS = [ "ไม่สามารถช่วยคุณเขียนโค้ด", "ไม่สามารถเขียนโค้ด", "ผมไม่สามารถเขียน", "ลองเขียนเองนะ", "ไม่สามารถช่วยเขียน", "ขอแนะนำให้คุณลองเขียนเอง", "ไม่สามารถสร้างโค้ด", ] def is_code_refusal(text: str) -> bool: return any(p in text for p in CODE_REFUSAL_PATTERNS) # ====================================================================== # 5) PREAMBLE STRIPPER # ====================================================================== _FALLBACK_PREAMBLE_PATTERNS = [ r"^here'?s\s+(the\s+)?translation.*?:\s*", r"^translation\s*:\s*", r"^คำแปล\s*:\s*", r"^แปลว่า\s*:\s*", ] _FALLBACK_PREAMBLE_REGEX = re.compile("|".join(_FALLBACK_PREAMBLE_PATTERNS), flags=re.IGNORECASE) def _find_first_thai_char_index(text: str) -> Optional[int]: for i, ch in enumerate(text): if '\u0E00' <= ch <= '\u0E7F': return i return None def strip_translation_preamble(text: str) -> str: text = text.strip() first_thai_idx = _find_first_thai_char_index(text) if first_thai_idx is not None and first_thai_idx > 0: preamble = text[:first_thai_idx] if re.search(r'translat', preamble, re.IGNORECASE): remainder = text[first_thai_idx:] remainder = re.sub(r'^[\s\-:>*]+', '', remainder) return remainder.strip() stripped = _FALLBACK_PREAMBLE_REGEX.sub("", text, count=1) return stripped.strip() # ====================================================================== # 6) PERSONA ENFORCER # ====================================================================== def enforce_persona(text: str) -> str: text = text.replace("ดิฉัน", "ผม") text = text.replace("ฉัน", "ผม") text = text.replace("ค่ะ", "ครับ") return text # ====================================================================== # 7) PER-EXPERT CONFIG # ====================================================================== @dataclass class ExpertConfig: name: str max_tokens: int = 512 temp: float = 0.7 top_p: float = 0.9 EXPERT_CONFIGS = { "python": ExpertConfig(name="python", max_tokens=800, temp=0.3, top_p=0.9), "typescript": ExpertConfig(name="typescript", max_tokens=800, temp=0.3, top_p=0.9), "reasoning": ExpertConfig(name="reasoning", max_tokens=1000, temp=0.6, top_p=0.9), } def get_config(expert_name: str) -> ExpertConfig: return EXPERT_CONFIGS.get(expert_name, ExpertConfig(name=expert_name)) # ====================================================================== # 8) PERSONA CONSISTENCY (System Prompt Injection) # ====================================================================== PERSONA_MALE_SUFFIX = ( "\n\nกฎการตอบ: ให้ใช้คำลงท้าย ครับ และแทนตัวเองว่า ผม " "อย่างสม่ำเสมอตลอดคำตอบ ห้ามใช้ ค่ะ หรือ ฉัน เด็ดขาด" ) def inject_persona(system_prompt: str = "") -> str: return system_prompt + PERSONA_MALE_SUFFIX # ====================================================================== # SELF-TEST # ====================================================================== if __name__ == "__main__": print("=" * 70) print("TEST: is_degenerate_query (Round 7 — Fast-Path Guard)") print("=" * 70) degenerate_cases = ["", " ", "\n\n\n", "🐍🐍🐍", "???!!!...", "?", "{{}} }}"] for q in degenerate_cases: result = is_degenerate_query(q) print(f"Query: {q!r} -> is_degenerate={result}") assert result is True, f"❌ FAIL: {q!r} ควรถูก Flag เป็น Degenerate!" print("✅ PASS: Query ว่าง/Emoji-only/Symbol-only ถูก Flag ถูกต้องหมด\n") meaningful_cases = ["help", "ทำไง", "def foo(): pass", "เขียน Python function ที่บวกเลขสองตัว"] for q in meaningful_cases: result = is_degenerate_query(q) print(f"Query: {q!r} -> is_degenerate={result}") assert result is False, f"❌ FAIL: {q!r} มีเนื้อหาจริง ไม่ควรถูก Flag!" print("✅ PASS: Query ที่มีเนื้อหาจริงไม่ถูก Flag ผิดเลย\n") print("=" * 70) print("TEST: get_greeting_quick_reply (Round 8)") print("=" * 70) greeting_true_cases = ["สวัสดี", "สวัสดีครับ", "หวัดดี", "hi", "hello"] for q in greeting_true_cases: result = get_greeting_quick_reply(q) assert result == GREETING_QUICK_REPLY_TEXT, f"❌ FAIL: {q!r} ควรได้ Greeting Reply!" print("✅ PASS: Greeting คำล้วนๆ ได้ Quick-Reply ถูกต้อง") closing_true_cases = ["บาย", "bye", "ลาก่อน"] for q in closing_true_cases: result = get_greeting_quick_reply(q) assert result == CLOSING_QUICK_REPLY_TEXT, f"❌ FAIL: {q!r} ควรได้ Closing Reply!" print("✅ PASS: Closing คำล้วนๆ ได้ Quick-Reply ถูกต้อง") thanks_true_cases = ["ขอบคุณ", "ขอบคุณครับ", "ขอบคุณมากครับ", "thanks"] for q in thanks_true_cases: result = get_greeting_quick_reply(q) assert result == THANKS_QUICK_REPLY_TEXT, f"❌ FAIL: {q!r} ควรได้ Thanks Reply!" print("✅ PASS: Thanks คำล้วนๆ ได้ Quick-Reply ถูกต้อง") false_positive_cases = [ "อธิบายเพิ่มอีกหน่อย", "คุณช่วยอะไรได้บ้าง", "ขอบคุณสำหรับ function นี้นะครับ", "แก้ให้หน่อย", ] for q in false_positive_cases: result = get_greeting_quick_reply(q) assert result is None, f"❌ FAIL: {q!r} มีเนื้อหาจริง ไม่ควรถูก Quick-Reply!" print("✅ PASS: Query ที่มีเนื้อหาจริงแซมอยู่ไม่ถูก Quick-Reply ผิดเลย\n") print("=" * 70) print("TEST: is_repetitive_response (Round 9)") print("=" * 70) looped_text = "ผมขออภัยที่ทำให้คุณสับสน " * 15 assert is_repetitive_response(looped_text) is True, "❌ FAIL: ควรจับ Loop ได้!" print("✅ PASS: จับ Repetition Loop ภาษาไทยได้ถูกต้อง") looped_list = "ดูข้อมูลจาก Google Maps, " * 20 assert is_repetitive_response(looped_list) is True, "❌ FAIL: ควรจับ Loop แบบ List ได้!" print("✅ PASS: จับ Repetition Loop แบบรายการซ้ำได้ถูกต้อง") normal_code_response = ( "แน่นอนครับ นี่คือฟังก์ชันที่เพิ่ม docstring:\n\n" "```python\ndef add_numbers(a, b):\n" ' """บวกเลขสองตัวเข้าด้วยกัน"""\n' " return a + b\n```\n\n" "ใน docstring นี้ ผมอธิบายว่าฟังก์ชันนี้จะบวกเลขสองตัวเข้าด้วยกัน " "และระบุประเภทของพารามิเตอร์ที่ใช้ครับ" ) assert is_repetitive_response(normal_code_response) is False, "❌ FAIL: คำตอบปกติไม่ควรถูก Flag!" print("✅ PASS: คำตอบปกติที่มีโค้ดซ้ำเล็กน้อยไม่ถูก Flag ผิด\n") print("=" * 70) print("TEST: Weighted Router (Stateless)") print("=" * 70) for q in ["เขียน Python function ที่บวกเลขสองตัว", "def foo(): pass"]: result = route_query(q) print(f"Query: {q!r} -> {result}\n") print("=" * 70) print("TEST: Router — History Continuation") print("=" * 70) r1 = route_query("แก้ให้หน่อย", previous_expert="python") assert r1["expert"] == "python" and r1["method"] == "history_continuation" print("✅ PASS: 'แก้ให้หน่อย' ใช้ previous_expert ต่อ\n") r2 = route_query("อธิบายเพิ่มอีกหน่อย", previous_expert="python") assert r2["expert"] == "python", "❌ FAIL: ควรใช้ previous_expert ต่อ!" print("✅ PASS: 'อธิบายเพิ่มอีกหน่อย' ไม่ถูกเข้าใจผิดว่าเป็นการลาก่อนแล้ว\n") r2b = route_query("อธิบายว่าแก้ตรงไหน", previous_expert="python") assert r2b["expert"] == "python", "❌ FAIL: Bug 'บาย' ใน 'อธิบาย' กลับมาอีกแล้ว!" print("✅ PASS: 'อธิบายว่าแก้ตรงไหน' ไม่ถูก Flag ผิดเป็น Greeting แล้ว\n") r3 = route_query("แก้ให้หน่อย", previous_expert=None) assert r3["expert"] is None print("✅ PASS: ไม่มี previous_expert ยังทำงานแบบเดิม\n") r4 = route_query("ขอบคุณครับ", previous_expert="typescript") assert r4["expert"] is None, "❌ FAIL: คำขอบคุณจริงๆต้องยัง Flag เป็น Greeting!" print("✅ PASS: 'ขอบคุณครับ' ยังถูก Flag เป็น Greeting ถูกต้อง\n") r5 = route_query("บาย", previous_expert="python") assert r5["expert"] is None, "❌ FAIL: คำว่า 'บาย' โดดๆต้องยัง Flag เป็น Greeting!" print("✅ PASS: 'บาย' คำเดียวโดดๆ ยังถูก Flag เป็น Greeting ถูกต้อง\n") print("=" * 70) print("TEST: is_code_refusal") print("=" * 70) assert is_code_refusal("ขออภัยครับ ผมไม่สามารถช่วยคุณเขียนโค้ด Python ที่บวกเลขสองตัวได้ครับ") is True assert is_code_refusal("```python\ndef add(a, b):\n return a + b\n```\nครับ") is False print("✅ PASS: Code Refusal Detector ทำงานถูกต้อง\n") print("=" * 70) print("TEST: Language purity filter — Emoji") print("=" * 70) result_emoji = clean_or_flag_response("ผมไม่สามารถทำให้🐍🐍🐍เป็นคำว่า python ได้ครับ 😊👍") assert result_emoji["is_clean"] is True print("✅ PASS: Emoji ไม่ถูก Flag ผิด\n") print("=" * 70) print("TEST: Persona enforcer") print("=" * 70) fixed = enforce_persona("ขออภัยค่ะ ฉันไม่เข้าใจคำถามนี้ค่ะ") assert "ค่ะ" not in fixed and "ฉัน" not in fixed print("✅ PASS\n") print("=" * 70) print("🎉 SELF-TEST ทั้งหมดผ่านเรียบร้อย") print("=" * 70)