umbragit/UmbraTrans
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Verify that a translated paragraph still renders its own source

Browse Source 
Measured on the real Chinese edition: the drafter sometimes re-flows the
passage across paragraph ids. Pass 0 splits the original passage at sentence
boundaries, so many paragraphs are sentence fragments ("The chair in the
middle has", ""operations""). A fragment is exactly where an intent-based
drafter is told to write natural target-language prose, and once it does
that, the id-to-text correspondence drifts: the Chinese is good, but it is
attached to the wrong paragraph.

In ch-15 the drift runs from p0095 to p0104 - ten paragraphs - and the
Chinese for p0102 ("Yes, Captain.") is the translation of p0101. The audio
would say the right words at the wrong place.

The alignment pass could not see it. A one-to-one paragraph is a pure copy,
so there is nothing to divide and nothing to disagree; the drift is invisible
to an alignment check by construction.

--verify now checks the paragraphs where the drift is possible: it starts at
a sentence fragment and follows the drift until the passage re-synchronises.
The check asks two questions - is content missing, and is content extra - and
a paragraph that carries content its source does not have is marked
"mismatch" and dropped from the delivery. A visible gap is better than a
confident wrong line in the audiobook.

6 new tests, 51 total, ruff clean.
This commit is contained in:
Umbra Tools
2026-11-06 20:31:24 -08:00
parent f0c2317ca0
commit 5a50b04749
2 changed files with 315 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
tools/translation/align.py
View File

@@ -64,6 +64,7 @@ second contract.
from __future__ import annotations
import argparse
import hashlib
import json
import re
import sys
@@ -319,7 +320,7 @@ class ParagraphAlignment:
@property
def needs_review(self) -> bool:
return bool(self.problems) or self.method in {"mechanical", "failed"}
return bool(self.problems) or self.method in {"mechanical", "failed", "mismatch"}
def source_utterances(paragraph: dict, bundle: dict[str, dict]) -> list[SourceUtterance]:
@@ -551,14 +552,144 @@ def align_paragraph(
)
# --- drift verification ----------------------------------------------------
SYSTEM_VERIFIER = """你是一位双语校对，负责检查一段译文是否真的对应它的英文原文。
译文是按段落 id 存储的；如果写译文时把内容写到了相邻段落，id 和内容就会错位。
只输出 JSON，不要解释，不要 Markdown 代码围栏。"""
# A paragraph whose English does not end on a sentence boundary is a fragment:
# Pass 0 split the original passage mid-sentence. A fragment is where the
# drafter is most likely to re-flow the passage into natural target-language
# prose and lose the id-to-text correspondence, so that is where the check
# starts.
_SENTENCE_END = ".!?…"
def is_fragment(text: str) -> bool:
"""True when *text* does not end on a sentence boundary.
Trailing closing quotes and brackets are ignored: a quoted label such as
``"operations"`` is a complete quotation but not a complete sentence.
"""
stripped = text.strip().rstrip("”’\"')]】》")
if not stripped:
return False
return stripped[-1] not in _SENTENCE_END
def verification_prompt(source_text: str, translation: str) -> str:
"""Ask whether *translation* renders *source_text* and nothing else."""
return f"""英文原文：
{source_text}
中文译文：
{translation}
请检查这段译文与原文的对应关系：
1. 原文里有没有意思在译文里找不到的内容？（missing）
2. 译文里有没有原文没有讲的内容？例如译文里混进了相邻段落的话。（extra）
只输出 JSON：{{"missing": true 或 false, "extra": true 或 false, "reason": "一句话理由"}}"""
def verify_translation(
source_text: str,
translation: str,
llm: LLM,
model: str = DEFAULT_ALIGN_MODEL,
) -> tuple[bool, str]:
"""Does *translation* render *source_text* exactly?
Returns ``(ok, reason)``. ``ok`` is False when the translation is missing
content the source has, or carries content the source does not - which is
what a paragraph-boundary drift looks like from the inside.
"""
try:
data, _usage = llm.chat_json(
SYSTEM_VERIFIER,
verification_prompt(source_text, translation),
model=model,
temperature=0.0,
cache_key=f"verify-{hashlib.sha256((source_text + translation).encode('utf-8')).hexdigest()[:16]}",
)
except LLMError as exc:
return False, f"verification model failed: {exc}"
if not isinstance(data, dict):
return False, "verification answer was not an object"
missing = bool(data.get("missing"))
extra = bool(data.get("extra"))
reason = str(data.get("reason") or "").strip()
if missing or extra:
parts = []
if missing:
parts.append("the translation is missing content the source has")
if extra:
parts.append("the translation carries content the source does not have")
if reason:
parts.append(reason)
return False, "; ".join(parts)
return True, reason
def _apply_drift_verification(
alignments: list[ParagraphAlignment],
chapter: dict,
translated: dict[str, str],
bundle: dict[str, dict],
llm: LLM,
model: str,
) -> None:
"""Flag the paragraphs whose translation does not render their own source.
The check starts at a sentence fragment and follows the drift until the
passage re-synchronises. A drifted paragraph is marked ``mismatch`` and
its segments are dropped: the delivery must not carry text that renders
the wrong source utterance, and a visible gap is better than a confident
wrong line in the audiobook.
"""
by_id = {alignment.paragraph_id: alignment for alignment in alignments}
drift_active = False
for paragraph in chapter.get("paragraphs") or []:
paragraph_id = str(paragraph.get("id") or "")
alignment = by_id.get(paragraph_id)
if alignment is None:
continue
source_text = str(paragraph.get("text") or "").strip()
translation = str(translated.get(paragraph_id) or "").strip()
if not source_text or not translation:
continue
if not (drift_active or is_fragment(source_text)):
continue
ok, reason = verify_translation(source_text, translation, llm, model)
if ok:
drift_active = False
continue
drift_active = True
alignment.method = "mismatch"
alignment.problems = [f"translation does not render its own source: {reason}"]
alignment.segments = []
def align_chapter(
chapter: dict,
translated: dict[str, str],
bundle: dict[str, dict],
llm: LLM | None = None,
model: str = DEFAULT_ALIGN_MODEL,
verify: bool = False,
) -> tuple[list[ParagraphAlignment], dict[str, Any]]:
"""Align every translated paragraph of one chapter."""
"""Align every translated paragraph of one chapter.
With *verify*, the paragraphs where the drafter could have re-flowed the
passage across paragraph ids are checked as well - see
:func:`verify_drift_run`.
"""
alignments: list[ParagraphAlignment] = []
for paragraph in chapter.get("paragraphs") or []:
paragraph_id = str(paragraph.get("id") or "")
@@ -576,6 +707,9 @@ def align_chapter(
continue
alignments.append(align_paragraph(paragraph, translation, bundle, llm=llm, model=model))
if verify and llm is not None:
_apply_drift_verification(alignments, chapter, translated, bundle, llm, model)
methods: dict[str, int] = {}
problems: list[str] = []
for alignment in alignments:
@@ -678,6 +812,15 @@ def main(argv: list[str] | None = None) -> int:
parser.add_argument("--model", default=DEFAULT_ALIGN_MODEL)
parser.add_argument("--base-url", default=None)
parser.add_argument("--no-llm", action="store_true", help="use the mechanical split only")
parser.add_argument(
"--verify",
action="store_true",
help=(
"check the paragraphs where the drafter could have re-flowed the "
"passage across paragraph ids, and drop the ones that do not "
"render their own source"
),
)
parser.add_argument("--source-ref", default="")
args = parser.parse_args(argv)
@@ -707,7 +850,14 @@ def main(argv: list[str] | None = None) -> int:
for chapter in selected:
chapter_id = str(chapter.get("id") or "")
translated = load_translations(args.run_dir, chapter_id)
alignments, report = align_chapter(chapter, translated, bundle, llm=llm, model=args.model)
alignments, report = align_chapter(
chapter,
translated,
bundle,
llm=llm,
model=args.model,
verify=args.verify,
)
chapters.append((chapter, alignments))
reports.append(report)
print(

162
tools/translation/tests/test_align.py
View File

@@ -20,8 +20,10 @@ sys.path.insert(0, str(HERE.parent))
from align import ( # noqa: E402
Segment,
align_chapter,
align_paragraph,
build_delivery,
is_fragment,
mechanical_split,
normalise_target_language,
parse_row_id,
@@ -29,6 +31,7 @@ from align import ( # noqa: E402
repartition,
source_utterances,
split_target_sentences,
verify_translation,
)
@@ -397,5 +400,164 @@ class CommandTests(unittest.TestCase):
self.assertTrue(Path(cache_dir).is_dir())
class DriftVerificationTests(unittest.TestCase):
"""The paragraph-boundary drift the production translation actually has."""
def test_a_fragment_is_an_english_text_without_a_sentence_end(self) -> None:
self.assertTrue(is_fragment("The chair in the middle has"))
self.assertTrue(is_fragment('"operations"'))
self.assertTrue(is_fragment("on the console. She looks around, and she asks,"))
self.assertFalse(is_fragment("The ship was large."))
self.assertFalse(is_fragment('"Yes, Captain."'))
self.assertFalse(is_fragment("“The last time was thirty-four years ago?”"))
def test_a_matching_translation_verifies(self) -> None:
class StubLLM:
def __init__(self) -> None:
self.calls = 0
def chat_json(self, system: str, user: str, **kwargs: object) -> tuple[dict, object]:
self.calls += 1
return ({"missing": False, "extra": False, "reason": ""}, None)
llm = StubLLM()
ok, _reason = verify_translation("The chair in the middle has", "中间的椅子旁", llm)
self.assertTrue(ok)
self.assertEqual(llm.calls, 1)
def test_a_drifted_translation_is_caught(self) -> None:
"""The Chinese is good - it just belongs to the next paragraph."""
class StubLLM:
def __init__(self) -> None:
self.calls = 0
def chat_json(self, system: str, user: str, **kwargs: object) -> tuple[dict, object]:
self.calls += 1
return (
{"missing": False, "extra": True, "reason": "the text is about another paragraph"},
None,
)
llm = StubLLM()
ok, reason = verify_translation(
'"Yes, Captain."',
"“上一次有人登上这艘舰桥还是三十四年前，那时是西顿船长在任？”",
llm,
)
self.assertFalse(ok)
self.assertIn("another paragraph", reason)
def test_a_drifted_paragraph_is_dropped_from_the_delivery(self) -> None:
"""A visible gap is better than a confident wrong line in the audio."""
class StubLLM:
def chat_json(self, system: str, user: str, **kwargs: object) -> tuple[dict, object]:
return ({"missing": False, "extra": True, "reason": "belongs to another paragraph"}, None)
chapter = {
"id": "ch-1",
"paragraphs": [
{
"id": "ch-1-p0000",
"speaker": "narrator",
"type": "narration",
"text": "The chair in the middle has",
"source_indices": [0],
"source_ids": ["utt-100"],
}
],
}
translated = {"ch-1-p0000": "“Yes, Captain.”"}
alignments, report = align_chapter(
chapter,
translated,
bundle_fixture(),
llm=StubLLM(),
verify=True,
)
self.assertEqual(alignments[0].method, "mismatch")
self.assertTrue(alignments[0].needs_review)
self.assertEqual(alignments[0].segments, [])
self.assertEqual(report["needs_review"], 1)
def test_the_check_only_starts_at_a_fragment(self) -> None:
"""A complete paragraph needs no check: the drafter had no reason to re-flow it."""
class CountingLLM:
def __init__(self) -> None:
self.calls = 0
def chat_json(self, system: str, user: str, **kwargs: object) -> tuple[dict, object]:
self.calls += 1
return ({"missing": False, "extra": False}, None)
chapter = {
"id": "ch-1",
"paragraphs": [
{
"id": "ch-1-p0000",
"speaker": "narrator",
"type": "narration",
"text": "The ship was large.",
"source_indices": [0],
"source_ids": ["utt-100"],
}
],
}
llm = CountingLLM()
align_chapter(chapter, {"ch-1-p0000": "这艘船很大。"}, bundle_fixture(), llm=llm, verify=True)
self.assertEqual(llm.calls, 0)
def test_the_check_follows_the_drift_until_the_passage_resynchronises(self) -> None:
"""A drift runs on: the next paragraphs carry the wrong content too."""
class DriftingLLM:
def __init__(self) -> None:
self.calls = 0
def chat_json(self, system: str, user: str, **kwargs: object) -> tuple[dict, object]:
self.calls += 1
if self.calls <= 3:
return ({"missing": False, "extra": True, "reason": "drifted"}, None)
return ({"missing": False, "extra": False}, None)
def para(pid: str, text: str, sid: str, index: int) -> dict:
return {
"id": pid,
"speaker": "narrator",
"type": "narration",
"text": text,
"source_indices": [index],
"source_ids": [sid],
}
chapter = {
"id": "ch-1",
"paragraphs": [
para("ch-1-p0000", "The chair in the middle has", "utt-100", 0),
para("ch-1-p0001", '"Captain"', "utt-101", 1),
para("ch-1-p0002", "on the console.", "utt-102", 2),
para("ch-1-p0003", "She looked around and saw dust everywhere.", "utt-103", 3),
],
}
llm = DriftingLLM()
alignments, _report = align_chapter(
chapter,
{p["id"]: "译文" for p in chapter["paragraphs"]},
bundle_fixture(),
llm=llm,
verify=True,
)
methods = [a.method for a in alignments]
self.assertEqual(methods[0], "mismatch")
self.assertEqual(methods[1], "mismatch")
self.assertEqual(methods[2], "mismatch")
self.assertEqual(methods[3], "one_to_one")
# The drift follows the run: the third paragraph was not a fragment, but
# it was checked because the passage had not re-synchronised yet.
self.assertEqual(llm.calls, 4)
if __name__ == "__main__":
unittest.main()