Spaces:
Sleeping
Sleeping
File size: 6,044 Bytes
1d46eb9 a7fd3ba 5dcfc82 1d46eb9 5dcfc82 a7fd3ba 5dcfc82 1d46eb9 a7fd3ba 1d46eb9 5dcfc82 a7fd3ba 5dcfc82 1d46eb9 95a3120 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 |
"""
Processing utilities for Vietnamese translation integration
"""
import logging
from typing import Dict, Any, List, Optional, Callable
logger = logging.getLogger(__name__)
def _vi_sanitize_text(s: str) -> str:
"""Light Vietnamese sanitization for finetuning and RAG: strip extra spaces, limit repetition, preserve numbers/units."""
if not isinstance(s, str):
return s
t = s.strip()
# Collapse repeated punctuation and whitespace
import re
t = re.sub(r"\s+", " ", t)
t = re.sub(r"([.?!]){3,}", r"..", t)
# Remove obvious repetition chunks (very heuristic)
parts = t.split()
if len(parts) > 20:
window = 6
seen = set()
filtered = []
for i in range(len(parts)):
ngram = " ".join(parts[max(0, i-window):i+1])
if ngram in seen:
continue
seen.add(ngram)
filtered.append(parts[i])
t = " ".join(filtered)
return t
def _validate_vi_translation(original: str, translated: str) -> bool:
"""Validate Vietnamese translation quality"""
if not translated or not isinstance(translated, str):
return False
# Check if translation is too short or too different in length
if len(translated.strip()) < 3:
return False
# Check if translation contains too much English (should be mostly Vietnamese)
import re
english_chars = len(re.findall(r'[a-zA-Z]', translated))
total_chars = len(re.sub(r'\s', '', translated))
if total_chars > 0 and english_chars / total_chars > 0.7:
return False
# Check for common translation failure patterns
failure_patterns = [
"translation", "error", "failed", "unable", "cannot",
"not available", "not found", "invalid", "error"
]
translated_lower = translated.lower()
for pattern in failure_patterns:
if pattern in translated_lower:
return False
return True
def translate_sft_row(row: Dict[str, Any], translator, text_fields: List[str] = None) -> Dict[str, Any]:
"""
Translate specific text fields in an SFT row from English to Vietnamese.
Args:
row: SFT row dictionary
translator: VietnameseTranslator instance
text_fields: List of field names to translate. If None, uses default fields.
Returns:
Translated SFT row dictionary
"""
if not translator or not translator.is_loaded():
logger.warning("Translator not available, skipping translation")
return row
if text_fields is None:
# Default fields to translate in SFT format
text_fields = ["instruction", "input", "output"]
try:
translated_row = translator.translate_dict(row, text_fields)
# Validate and sanitize translated fields
for f in text_fields:
if f in translated_row.get("sft", {}):
original = row.get("sft", {}).get(f, "")
translated = translated_row["sft"][f]
if _validate_vi_translation(original, translated):
translated_row["sft"][f] = _vi_sanitize_text(translated)
else:
logger.warning(f"Invalid Vietnamese translation for field {f}, keeping original")
translated_row["sft"][f] = original
logger.debug(f"Translated SFT row with fields: {text_fields}")
return translated_row
except Exception as e:
logger.error(f"Failed to translate SFT row: {e}")
return row
def translate_rag_row(row: Dict[str, Any], translator, text_fields: List[str] = None) -> Dict[str, Any]:
"""
Translate specific text fields in a RAG row from English to Vietnamese.
Args:
row: RAG row dictionary
translator: VietnameseTranslator instance
text_fields: List of field names to translate. If None, uses default fields.
Returns:
Translated RAG row dictionary
"""
if not translator or not translator.is_loaded():
logger.warning("Translator not available, skipping translation")
return row
if text_fields is None:
# Default fields to translate in RAG format (Q, A, C)
text_fields = ["question", "answer", "context"]
try:
translated_row = translator.translate_dict(row, text_fields)
# Validate and sanitize translated fields
for f in text_fields:
if f in translated_row:
original = row.get(f, "")
translated = translated_row[f]
if _validate_vi_translation(original, translated):
translated_row[f] = _vi_sanitize_text(translated)
else:
logger.warning(f"Invalid Vietnamese translation for field {f}, keeping original")
translated_row[f] = original
logger.debug(f"Translated RAG row with fields: {text_fields}")
return translated_row
except Exception as e:
logger.error(f"Failed to translate RAG row: {e}")
return row
def should_translate(vietnamese_translation: bool, translator) -> bool:
"""
Check if translation should be performed.
Args:
vietnamese_translation: Flag from user input
translator: VietnameseTranslator instance
Returns:
True if translation should be performed
"""
if not vietnamese_translation:
return False
if not translator or not translator.is_loaded():
logger.warning("Vietnamese translation requested but translator not available")
return False
return True
def log_translation_stats(stats: Dict[str, Any], translated_count: int) -> None:
"""
Log translation statistics.
Args:
stats: Statistics dictionary to update
translated_count: Number of items translated
"""
stats["vietnamese_translated"] = translated_count
logger.info(f"Vietnamese translation completed: {translated_count} items translated") |