sepehr/office_translator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(qualite): retenter automatiquement les traductions livrees dans la mauvaise ecriture

Browse Source 
Le defaut le plus sournois du pipeline etait silencieux : de l'arabe
livre pour une cible persane (meme ecriture, mauvaise langue) passe
inapercu et part chez le lecteur. Desormais chaque lot traduit est
verifie par le detecteur d'ecritures, et chaque segment fautif est
redemande une fois au moteur avec une consigne renforcee (nom de la
langue + lettres specifiques, ex. persan پ چ ژ گ). La seconde tentative
ne remplace la premiere que si elle passe le meme controle.

- services/quality/script_detector.py : extraction d'un controle
  script_issue() reutilisable ; detect_arabic_variant signale des
 ormais un long texte en ecriture arabe sans aucune lettre specifique
  de la langue cible (arabe pur livre pour du persan/ourdou/pachto)
- translators/segments.py : retry_wrong_script() + construction de la
  consigne renforcee, sans jamais faire echouer le travail
- Word, Excel, PDF : branchement apres la memoire de traduction et les
  validations humaines ; le texte inchange (chiffres, noms propres) ne
  declenche jamais de retentative
- 14 tests nouveaux (tests/test_translators/test_script_retry.py)
This commit is contained in:
sepehr
2026-09-01 20:52:23 +02:00
parent ffbd85a7b6
commit d92bbf0fa6
6 changed files with 504 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
services/quality/script_detector.py
View File

@@ -162,6 +162,31 @@ def detect_arabic_variant(
),
}
# No discriminating character of ANY Arabic-script variant was found.
# For a long text this is itself a strong signal: a Persian/Urdu/Pashto
# text of this length almost always contains at least one of its
# specific letters (Persian پ چ ژ گ are common letters), while plain
# Arabic never does. Short texts stay tolerant — the signal is too
# weak to tell plain Arabic from e.g. Persian without those letters.
target_chars = _config.get_discriminating_chars(claimed_lang)
if (
not detected
and claimed_lang
and claimed_lang.lower() != "ar"
and target_chars
and letters >= 25
):
return {
"verdict": "fail",
"claimed_lang": claimed_lang,
"detected_variants": [],
"arabic_ratio": round(arabic_ratio, 3),
"reason": (
f"target={claimed_lang} but none of its specific letters appear "
f"in a {letters}-letter Arabic-script text — likely plain Arabic"
),
}
return {
"verdict": "pass",
"claimed_lang": claimed_lang,
@@ -173,31 +198,14 @@ def detect_arabic_variant(
# ---------- Per-chunk evaluation ----------
def evaluate_chunk(
source_text: str,
translated_text: str,
target_lang: Optional[str],
) -> QualityCheckResult:
def _script_checks(text: str, target_lang: Optional[str]) -> tuple:
"""
Run the L0 checks on a single (source, translation) pair.
Script checks shared by evaluate_chunk (QA) and script_issue (retry).
Returns a QualityCheckResult. The function is purely defensive — it
never raises; any internal error results in a "skip" result.
Returns (issues, details): issues is a list of
"wrong_script"/"wrong_arabic_variant" strings, details carries the
diagnostics evaluate_chunk logs.
"""
if translated_text is None:
return QualityCheckResult(
passed=True, score=0.0, issues=["empty_translation"],
details={"reason": "translation is None"},
)
text = translated_text.strip()
if not text:
return QualityCheckResult(
passed=True, score=0.0, issues=["empty_translation"],
details={"reason": "translation is empty or whitespace-only"},
)
target_lang = (target_lang or "").lower() or None
issues: List[str] = []
details: Dict = {}
@@ -262,6 +270,68 @@ def evaluate_chunk(
if variant_result["verdict"] == "fail":
issues.append("wrong_arabic_variant")
return issues, details
def script_issue(translated_text: str, target_lang: Optional[str]) -> Optional[str]:
"""
Standalone wrong-script check, usable by the translators to decide
whether a delivered translation must be retried.
Returns a short human-readable reason when the text is NOT written
in the script expected for target_lang (wrong script, or wrong
Arabic-script variant such as Arabic delivered for Persian), and
None when the script is acceptable. Purely heuristic, never raises.
"""
try:
text = (translated_text or "").strip()
if not text:
return None
issues, details = _script_checks(text, (target_lang or "").lower() or None)
if not issues:
return None
if "wrong_arabic_variant" in issues:
variant = details.get("arabic_variant") or {}
return variant.get("reason") or "wrong_arabic_variant"
return details.get("reason") or issues[0]
except Exception as e: # defensive: retry decisions must never crash a job
logger.warning("script_issue_error", error=str(e)[:200])
return None
def evaluate_chunk(
source_text: str,
translated_text: str,
target_lang: Optional[str],
) -> QualityCheckResult:
"""
Run the L0 checks on a single (source, translation) pair.
Returns a QualityCheckResult. The function is purely defensive — it
never raises; any internal error results in a "skip" result.
"""
if translated_text is None:
return QualityCheckResult(
passed=True, score=0.0, issues=["empty_translation"],
details={"reason": "translation is None"},
)
text = translated_text.strip()
if not text:
return QualityCheckResult(
passed=True, score=0.0, issues=["empty_translation"],
details={"reason": "translation is empty or whitespace-only"},
)
target_lang = (target_lang or "").lower() or None
issues: List[str] = []
details: Dict = {}
# --- Script detection + Arabic variant (shared with script_issue) ---
script_issues, script_details = _script_checks(text, target_lang)
issues.extend(script_issues)
details.update(script_details)
# --- Length sanity ---
length_result = length_checker.check(source_text, text)
details["length"] = length_result
@@ -291,8 +361,8 @@ def evaluate_chunk(
passed=passed,
score=round(score, 3),
issues=issues,
detected_script=detected_script,
expected_script=expected_script,
detected_script=script_details.get("detected_script"),
expected_script=script_details.get("expected_script"),
details=details,
)

212
tests/test_translators/test_script_retry.py Normal file
View File

@@ -0,0 +1,212 @@
"""
Tests du garde-fou d'écriture (wrong-script retry).
Scénario principal : un LLM livre de l'arabe alors que la cible est le
persan — même écriture, mauvaise langue, indétectable à l'œil d'un
non-lecteur. Le détecteur doit le signaler et le traducteur doit
redemander une fois avec une consigne renforcée, sans jamais risquer
de dégrader le document (la 2e tentative ne remplace que si elle passe
le même contrôle).
"""
from docx import Document
from services.providers.base import TranslationProvider
from services.providers.schemas import TranslationRequest, TranslationResponse
from services.quality.script_detector import script_issue
from translators.segments import retry_wrong_script
from translators.word_translator import WordTranslator
# Arabe pur (aucune lettre persane spécifique), bien plus de 25 lettres.
ARABIC_FOR_PERSIAN = "مرحبا بكم في صفحة المشاريع الجديدة الخاصة بنا"
# Persan correct : contient پ et ژ (lettres spécifiques du persan).
GOOD_PERSIAN = "به صفحه پروژه های جدید ما خوش آمدید"
SOURCE_EN = "Welcome to our new projects page"
class SequencedMockProvider(TranslationProvider):
"""Renvoie une mauvaise traduction, puis la bonne dès que la requête
porte la consigne de retentative (détectée via le custom prompt)."""
def __init__(self, wrong: dict, corrected: dict):
self._wrong = wrong
self._corrected = corrected
self.calls = [] # (texte, porte_la_consigne_de_retry)
def get_name(self) -> str:
return "sequenced-mock"
def is_available(self) -> bool:
return True
def translate_text(self, request: TranslationRequest) -> TranslationResponse:
custom = (request.metadata or {}).get("custom_prompt") or ""
is_retry = "CRITICAL RETRY" in custom
self.calls.append((request.text, is_retry))
table = self._corrected if is_retry else self._wrong
return TranslationResponse(
translated_text=table.get(request.text, request.text),
provider_name="sequenced-mock",
source_language=request.source_language,
)
def translate_batch(self, requests: list) -> list:
return [self.translate_text(req) for req in requests]
# ---------------------------------------------------------------------------
# script_issue — détection seule
# ---------------------------------------------------------------------------
class TestScriptIssue:
def test_pure_arabic_for_persian_flagged(self):
reason = script_issue(ARABIC_FOR_PERSIAN, "fa")
assert reason is not None
assert "fa" in reason
def test_correct_persian_passes(self):
assert script_issue(GOOD_PERSIAN, "fa") is None
def test_latin_for_persian_flagged(self):
assert script_issue("Bienvenue sur notre page", "fa") is not None
def test_cyrillic_for_french_flagged(self):
reason = script_issue("Добро пожаловать на новую страницу", "fr")
assert reason is not None
def test_short_pure_arabic_stays_tolerated(self):
# Texte court : pas assez de signal pour distinguer arabe et
# persan sans lettre discriminante — comportement historique.
assert script_issue("السلام عليكم", "fa") is None
def test_identity_and_empty_are_never_flagged(self):
assert script_issue("", "fa") is None
assert script_issue("2024", "fa") is None
# ---------------------------------------------------------------------------
# retry_wrong_script — mécanisme seul
# ---------------------------------------------------------------------------
class TestRetryWrongScript:
def test_retry_replaces_with_valid_translation(self):
def translate_fn(texts, hint):
assert "CRITICAL RETRY" in hint
assert texts == [SOURCE_EN]
return [GOOD_PERSIAN]
result, fixed = retry_wrong_script(
[SOURCE_EN], [ARABIC_FOR_PERSIAN], "fa", translate_fn
)
assert fixed == 1
assert result[0] == GOOD_PERSIAN
def test_keeps_first_attempt_when_retry_still_wrong(self):
def translate_fn(texts, hint):
return [ARABIC_FOR_PERSIAN] # la 2e tentative est tout aussi fausse
result, fixed = retry_wrong_script(
[SOURCE_EN], [ARABIC_FOR_PERSIAN], "fa", translate_fn
)
assert fixed == 0
assert result[0] == ARABIC_FOR_PERSIAN
def test_identity_output_never_triggers_retry(self):
def translate_fn(texts, hint): # pragma: no cover — ne doit pas servir
raise AssertionError("no retry expected for identity output")
result, fixed = retry_wrong_script(
["Microsoft 2024"], ["Microsoft 2024"], "fa", translate_fn
)
assert fixed == 0
assert result == ["Microsoft 2024"]
def test_correct_translation_never_triggers_retry(self):
def translate_fn(texts, hint): # pragma: no cover
raise AssertionError("no retry expected for a correct translation")
result, fixed = retry_wrong_script(
[SOURCE_EN], [GOOD_PERSIAN], "fa", translate_fn
)
assert fixed == 0
assert result == [GOOD_PERSIAN]
def test_provider_error_keeps_originals(self):
def translate_fn(texts, hint):
raise RuntimeError("boom")
result, fixed = retry_wrong_script(
[SOURCE_EN], [ARABIC_FOR_PERSIAN], "fa", translate_fn
)
assert fixed == 0
assert result == [ARABIC_FOR_PERSIAN]
def test_only_offenders_are_resent(self):
sent = []
def translate_fn(texts, hint):
sent.extend(texts)
return [GOOD_PERSIAN] * len(texts)
sources = [SOURCE_EN, "Second text", "Third text"]
translations = [ARABIC_FOR_PERSIAN, "متنی که کاملاً درست است", "سلام دوست من"]
# La 2e contient ک ? non — vérifions plutôt avec un persan valide
# pour montrer que seules les traductions fautives repartent.
translations[1] = "این متن کاملاً درست است"
translations[2] = "این هم درست است"
result, fixed = retry_wrong_script(sources, translations, "fa", translate_fn)
assert sent == [SOURCE_EN]
assert fixed == 1
# ---------------------------------------------------------------------------
# Intégration — le traducteur Word corrige un lot entier
# ---------------------------------------------------------------------------
class TestWordIntegration:
def test_word_retries_arabic_delivered_for_persian(self, tmp_path):
provider = SequencedMockProvider(
{SOURCE_EN: ARABIC_FOR_PERSIAN},
{SOURCE_EN: GOOD_PERSIAN},
)
translator = WordTranslator(provider=provider)
doc = Document()
doc.add_paragraph(SOURCE_EN)
input_file = tmp_path / "input.docx"
doc.save(input_file)
output_file = tmp_path / "output.docx"
translator.translate_file(input_file, output_file, "fa")
# Une demande initiale + une retentative avec consigne renforcée
non_retry = [c for c in provider.calls if not c[1]]
retry = [c for c in provider.calls if c[1]]
assert len(non_retry) == 1
assert len(retry) == 1
assert retry[0][0] == SOURCE_EN
doc_out = Document(output_file)
assert GOOD_PERSIAN in doc_out.paragraphs[0].text
def test_word_no_retry_for_latin_target(self, tmp_path):
provider = SequencedMockProvider(
{SOURCE_EN: "Bienvenue sur notre page"}, {}
)
translator = WordTranslator(provider=provider)
doc = Document()
doc.add_paragraph(SOURCE_EN)
input_file = tmp_path / "input.docx"
doc.save(input_file)
output_file = tmp_path / "output.docx"
translator.translate_file(input_file, output_file, "fr")
# Traduction latine correcte pour une cible latine : une seule demande
assert len(provider.calls) == 1
assert not provider.calls[0][1]

25
translators/excel_translator.py
View File

@@ -526,6 +526,24 @@ class ExcelTranslator:
)
miss_pos += 1
# Wrong-script guard: re-ask once (with a reinforced instruction)
# for translations delivered in the wrong alphabet — e.g. Arabic
# delivered for a Persian target.
from translators.segments import retry_wrong_script
def _retry_translate(retry_texts, hint):
if self._provider is not None:
return self._translate_with_provider(
retry_texts, target_language, source_language, extra_prompt=hint
)
return self._translate_with_legacy(
retry_texts, target_language, source_language
)
translated, _script_fixed = retry_wrong_script(
texts, translated, target_language, _retry_translate
)
recorder = getattr(self, "_segment_recorder", None)
if recorder is not None:
recorder.record_pairs(zip(texts, translated))
@@ -539,7 +557,8 @@ class ExcelTranslator:
return dict(self._translation_stats)
def _translate_with_provider(
self, texts: List[str], target_language: str, source_language: str
self, texts: List[str], target_language: str, source_language: str,
extra_prompt: Optional[str] = None,
) -> List[str]:
"""Translate using the TranslationProvider.translate_batch() interface."""
from services.providers.base import TranslationProvider as NewTranslationProvider
@@ -557,6 +576,10 @@ class ExcelTranslator:
if is_new_style:
from services.providers.schemas import TranslationRequest
custom_prompt = getattr(self, "_custom_prompt", None)
if extra_prompt:
custom_prompt = (
f"{custom_prompt}\n\n{extra_prompt}" if custom_prompt else extra_prompt
)
metadata = {"custom_prompt": custom_prompt} if custom_prompt else None
requests = [

26
translators/pdf_translator.py
View File

@@ -1702,7 +1702,8 @@ class PDFTranslator:
# ------------------------------------------------------------------ #
def _translate_with_provider(
self, texts: List[str], target_language: str, source_language: str
self, texts: List[str], target_language: str, source_language: str,
extra_prompt: Optional[str] = None,
) -> List[str]:
"""Translate using the TranslationProvider interface (handles old & new styles)."""
from services.providers.base import TranslationProvider as NewTranslationProvider
@@ -1720,6 +1721,10 @@ class PDFTranslator:
if is_new_style:
from services.providers.schemas import TranslationRequest
custom_prompt = getattr(self, "_custom_prompt", None)
if extra_prompt:
custom_prompt = (
f"{custom_prompt}\n\n{extra_prompt}" if custom_prompt else extra_prompt
)
metadata = {"custom_prompt": custom_prompt} if custom_prompt else None
requests = [
@@ -1812,6 +1817,25 @@ class PDFTranslator:
logger.warning("legacy_translate_failed", error=str(e))
translated = texts
# Wrong-script guard: re-ask once (with a reinforced instruction)
# for translations delivered in the wrong alphabet — e.g. Arabic
# delivered for a Persian target.
from translators.segments import retry_wrong_script
def _retry_translate(retry_texts, hint):
if self._provider is not None:
return self._translate_with_provider(
retry_texts, target_language, source_language, extra_prompt=hint
)
from services.translation_service import translation_service
return translation_service.translate_batch(
retry_texts, target_language, source_language
)
translated, _script_fixed = retry_wrong_script(
texts, translated, target_language, _retry_translate
)
changed = sum(1 for orig, trans in zip(texts, translated) if orig != trans and trans.strip())
self._translation_stats["changed"] += changed

114
translators/segments.py
View File

@@ -72,3 +72,117 @@ def apply_overrides(
else:
misses.append(i)
return hits, misses
# ---------------------------------------------------------------------------
# Wrong-script retry — turn the script detector into an actionable guard
# ---------------------------------------------------------------------------
def _script_retry_hint(target_language: str, reason: str) -> str:
"""Extra instructions appended to the retry request's custom prompt.
Tells the model, in its own working language (English), that the
previous attempt was rejected for the wrong script/variant and what
the output must look like this time.
"""
from core.languages import language_name
from services.quality import config as _qa_config
name = language_name(target_language) or target_language
lines = [
"CRITICAL RETRY: your previous attempt was rejected because the "
f"output was not written in {name}.",
f"Rejected output reason: {reason}",
f"Write the translation strictly in {name}.",
]
target_chars = _qa_config.get_discriminating_chars(
(target_language or "").lower().split("-")[0]
)
if target_chars:
sample = " ".join(sorted(target_chars))
lines.append(
f"It must be {name}, not plain Arabic: use the letters specific "
f"to {name} ({sample}) wherever the words require them."
)
return " ".join(lines)
def retry_wrong_script(
sources: List[str],
translations: List[str],
target_language: str,
translate_fn,
) -> Tuple[List[str], int]:
"""Retry the translations delivered in the wrong script, once each.
The most damaging silent failure of a translation pipeline is the
right language family with the wrong alphabet: Arabic delivered for
Persian looks almost correct to a non-reader and ships unnoticed.
This guard checks every delivered translation against the target
language's expected script (and, for Arabic-script languages, the
expected variant), and re-asks the model once for each offender with
an explicit instruction. A retry only replaces the first attempt when
it passes the same check; otherwise the first attempt is kept.
translate_fn(texts, extra_prompt) must translate a list of texts in
one call, appending extra_prompt to the request's custom prompt.
Returns (translations, number_of_replacements). Never raises: any
error in detection or retry keeps the original translations.
"""
if not translations or not target_language:
return translations, 0
try:
from services.quality.script_detector import script_issue
except Exception:
return translations, 0
# Only genuinely translated segments can be "wrong": identity output
# (numbers, proper names kept as-is) must not trigger a retry.
offenders = []
reasons = []
for i, (src, trans) in enumerate(zip(sources, translations)):
if not trans or trans.strip() == (src or "").strip():
continue
try:
reason = script_issue(trans, target_language)
except Exception:
continue
if reason:
offenders.append(i)
reasons.append(reason)
if not offenders:
return translations, 0
hint = _script_retry_hint(target_language, reasons[0])
try:
retried = translate_fn([sources[i] for i in offenders], hint)
except Exception as e:
logger.warning("script_retry_failed", error=str(e)[:200], count=len(offenders))
return translations, 0
fixed = 0
result = list(translations)
for pos, i in enumerate(offenders):
if pos >= len(retried):
break
candidate = retried[pos]
if not candidate or not candidate.strip():
continue
try:
still_wrong = script_issue(candidate, target_language)
except Exception:
still_wrong = True # do not swap in an unverifiable retry
if not still_wrong:
result[i] = candidate
fixed += 1
logger.info(
"script_retry",
target=target_language,
offenders=len(offenders),
fixed=fixed,
)
return result, fixed

40
translators/word_translator.py
View File

@@ -763,6 +763,25 @@ class WordTranslator:
)
miss_pos += 1
# Wrong-script guard: re-ask once (with a reinforced instruction)
# for translations delivered in the wrong alphabet — e.g. Arabic
# delivered for a Persian target, which looks almost right to a
# non-reader and ships unnoticed.
from translators.segments import retry_wrong_script
def _retry_translate(retry_texts, hint):
if self._provider is not None:
return self._translate_with_provider(
retry_texts, target_language, source_language, extra_prompt=hint
)
return self._translate_with_legacy(
retry_texts, target_language, source_language, extra_prompt=hint
)
translated, _script_fixed = retry_wrong_script(
texts, translated, target_language, _retry_translate
)
recorder = getattr(self, "_segment_recorder", None)
if recorder is not None:
recorder.record_pairs(zip(texts, translated))
@@ -776,7 +795,8 @@ class WordTranslator:
return dict(self._translation_stats)
def _translate_with_provider(
self, texts: List[str], target_language: str, source_language: str
self, texts: List[str], target_language: str, source_language: str,
extra_prompt: Optional[str] = None,
) -> List[str]:
"""Translate using the TranslationProvider.translate_batch() interface."""
from services.providers.base import TranslationProvider as NewTranslationProvider
@@ -794,8 +814,12 @@ class WordTranslator:
if is_new_style:
from services.providers.schemas import TranslationRequest
custom_prompt = getattr(self, "_custom_prompt", None)
if extra_prompt:
custom_prompt = (
f"{custom_prompt}\n\n{extra_prompt}" if custom_prompt else extra_prompt
)
metadata = {"custom_prompt": custom_prompt} if custom_prompt else None
requests = [
TranslationRequest(
text=t,
@@ -809,7 +833,7 @@ class WordTranslator:
translated = [resp.translated_text for resp in responses]
else:
translated = self._provider.translate_batch(texts, target_language, source_language)
# Fallback: keep original text for any empty/failed result
return [
t if (t and t.strip()) else orig
@@ -817,9 +841,15 @@ class WordTranslator:
]
def _translate_with_legacy(
self, texts: List[str], target_language: str, source_language: str
self, texts: List[str], target_language: str, source_language: str,
extra_prompt: Optional[str] = None,
) -> List[str]:
"""Fallback to legacy translation_service for backward compatibility."""
"""Fallback to legacy translation_service for backward compatibility.
The legacy service has no custom-prompt parameter: extra_prompt is
accepted for signature parity but cannot be forwarded — the retry
still happens, just without the reinforced instruction.
"""
from services.translation_service import translation_service
_log_info(