busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(translate): add translation cache with source_hash dedup + enforce dictionary post-processing

Browse Source 
- Add source_hash column (SHA256 of source_text+source_data+dict_hash+config_hash) to TranslationRecord for cache dedup
- Add Alembic migration b1c2d3e4f5a6 for the new column and composite index (source_hash, status)
- Implement _compute_source_hash / _check_translation_cache in executor.py — before LLM call, check if same source+dict+config combo was already translated successfully
- Implement cache-aware routing in _process_batch: full cache hit → skip LLM, partial hit → route to LLM
- Store source_hash on all new TranslationRecord rows for future cache hits
- Strengthen dictionary prompt instructions from 'use when applicable' to 'MUST use — mandatory'
- Add _enforce_dictionary() post-processing: force-replace dictionary terms in LLM output if LLM ignored them
- Fix CRITICAL: use run.dict_snapshot_hash/config_hash (exist on TranslationRun, not TranslationJob)
- Fix MAJOR: verify cached languages cover ALL target_languages before accepting cache hit
- Fix MAJOR: prevent regex back-reference injection in dictionary enforcement (use lambda)
- Fix: add joinedload to cache lookup to avoid N+1 queries
- Fix: remove redundant single-column index (composite index is sufficient)
- Fix(frontend): parse paginated /translate/dictionaries response correctly (result.items)
This commit is contained in:
busya
2026-05-16 00:42:07 +03:00
parent 30ba70933d
commit b466ac6211
12 changed files with 458 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/src/models/translate.py
View File

@@ -132,12 +132,15 @@ class TranslationRecord(Base):
token_count_input = Column(Integer, nullable=True)
token_count_output = Column(Integer, nullable=True)
translation_duration_ms = Column(Integer, nullable=True)
source_hash = Column(String, nullable=True,
comment="SHA256(source_text+source_data+dict_snapshot_hash+config_hash) for cache dedup")
created_at = Column(DateTime, default=lambda: datetime.now(UTC))
languages = relationship("TranslationLanguage", back_populates="record")
__table_args__ = (
Index("ix_translation_records_run_status", "run_id", "status"),
Index("ix_translation_records_source_hash_status", "source_hash", "status"),
)
# #endregion TranslationRecord

193
backend/src/plugins/translate/executor.py
View File

@@ -15,7 +15,9 @@
# @RATIONALE: Batch processing with retry — independent batches allow partial recovery.
# @REJECTED: Single monolithic LLM call — would lose all progress on any failure.
import hashlib
import json
import re
import time
import uuid
from collections.abc import Callable
@@ -56,6 +58,127 @@ MAX_RETRIES_PER_BATCH = 3
MAX_ROWS_PER_RUN = 10000
# #endregion MAX_ROWS_PER_RUN
# #region _enforce_dictionary [TYPE Function]
# @BRIEF Post-processing: enforce dictionary entries in LLM output.
# If a source term from the dictionary appears in the original text,
# but the target term is missing from the translation, replace occurrences
# of the source term in the translated output with the target term.
# @PRE: dict_matches is a list of dict entries with source_term/target_term.
# @POST: per_lang_values may be mutated to include forced dictionary replacements.
# @SIDE_EFFECT: Logs when enforcement is applied.
def _enforce_dictionary(
source_text: str,
per_lang_values: dict[str, str],
dict_matches: list[dict[str, Any]],
batch_id: str,
row_id: str,
) -> None:
"""Post-processing: enforce dictionary terms in LLM translation output.
For each dictionary entry whose source_term appears in the source_text,
verify that the target_term is present in each language's translation.
If missing, replace any occurrence of the source term (which the LLM
may have left untranslated) with the dictionary target term.
"""
if not dict_matches or not source_text:
return
text_lower = source_text.lower()
for dm in dict_matches:
src_term = dm.get("source_term", "")
tgt_term = dm.get("target_term", "")
if not src_term or not tgt_term:
continue
# Only process if source term actually appears in the original text
if src_term.lower() not in text_lower:
continue
# Try to replace in each language
for lang_code in list(per_lang_values.keys()):
val = per_lang_values[lang_code]
if not val:
continue
# Check if target term is already present (case-insensitive)
if tgt_term.lower() in val.lower():
continue
# Try to find the source term (or a close variant) in the output
# This catches cases where the LLM left the source term untranslated
# Use lambda to prevent regex back-reference injection from tgt_term
src_pattern = re.compile(re.escape(src_term), re.IGNORECASE)
if src_pattern.search(val):
new_val = src_pattern.sub(lambda _: tgt_term, val)
if new_val != val:
logger.reason("Dictionary enforcement applied", {
"batch_id": batch_id,
"row_id": row_id,
"language_code": lang_code,
"source_term": src_term,
"target_term": tgt_term,
"before": val[:200],
"after": new_val[:200],
})
per_lang_values[lang_code] = new_val
# #endregion _enforce_dictionary
# #region _compute_source_hash [TYPE Function]
# @BRIEF Compute deterministic cache key for a source row.
# SHA256 of (source_text + source_data + dict_config_ctx) so that changing
# the text, context, dictionary snapshot, or job config produces a new key.
def _compute_source_hash(
source_text: str,
source_data: dict | None,
dict_snapshot_hash: str | None,
config_hash: str | None,
) -> str:
"""Deterministic cache key for a translation source row."""
payload = json.dumps({
"text": source_text,
"data": source_data,
"dict_hash": dict_snapshot_hash or "",
"config_hash": config_hash or "",
}, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
# #endregion _compute_source_hash
# #region _check_translation_cache [TYPE Function]
# @BRIEF Look up a previously successful translation by source_hash.
# Returns per-language dict {lang_code: final_value} or None.
def _check_translation_cache(
db: Session,
source_hash: str,
) -> dict[str, str] | None:
"""Check if this source_hash was already translated successfully."""
from sqlalchemy.orm import joinedload
from ...models.translate import TranslationRecord, TranslationLanguage
cached = (
db.query(TranslationRecord)
.options(joinedload(TranslationRecord.languages))
.filter(
TranslationRecord.source_hash == source_hash,
TranslationRecord.status == "SUCCESS",
)
.order_by(TranslationRecord.created_at.desc())
.first()
)
if not cached:
return None
lang_values: dict[str, str] = {}
for lang in cached.languages:
if lang.status == "translated" and lang.final_value:
lang_values[lang.language_code] = lang.final_value
return lang_values if lang_values else None
# #endregion _check_translation_cache
# #region TranslationExecutor [C:4] [TYPE Class]
# @BRIEF Process translation batches: fetch source rows, filter dict, call LLM, persist results.
# @PRE: DB session and config manager available.
@@ -161,6 +284,8 @@ class TranslationExecutor:
run_id=run.id,
batch_index=batch_idx,
batch_rows=batch_rows,
dict_snapshot_hash=run.dict_snapshot_hash,
config_hash=run.config_hash,
)
successful_records += batch_result["successful"]
failed_records += batch_result["failed"]
@@ -551,6 +676,8 @@ class TranslationExecutor:
run_id: str,
batch_index: int,
batch_rows: list[dict[str, Any]],
dict_snapshot_hash: str | None = None,
config_hash: str | None = None,
) -> dict[str, int]:
with belief_scope("TranslationExecutor._process_batch"):
batch_start = time.monotonic()
@@ -584,7 +711,28 @@ class TranslationExecutor:
row_context=row_context,
)
# For each row, determine if we need LLM translation or can use approved/preview edit
# ── Translation cache check: skip LLM for rows translated before ──
for row in batch_rows:
if row.get("approved_translation"):
continue
source_text = row.get("source_text", "")
if not source_text:
continue
source_data = row.get("source_data")
source_hash = _compute_source_hash(
source_text, source_data,
dict_snapshot_hash, config_hash,
)
row["_source_hash"] = source_hash
cached = _check_translation_cache(self.db, source_hash)
if cached:
row["_cached_lang_values"] = cached
logger.reason("Translation cache hit", {
"source_hash": source_hash[:12],
"langs": list(cached.keys()),
})
# For each row, determine if we need LLM translation or can use approved/preview edit/cache
rows_for_llm = []
pre_translated = []
@@ -593,6 +741,26 @@ class TranslationExecutor:
pre_translated.append(row)
continue
# Translation cache hit → treat as pre-translated
# BUT only if cached langs cover ALL target languages
cached_langs = row.get("_cached_lang_values")
if cached_langs:
target_langs_for_check = job.target_languages or [job.target_dialect or "en"]
if not isinstance(target_langs_for_check, list):
target_langs_for_check = [str(target_langs_for_check)]
cached_covers_all = all(
lang_code in cached_langs
for lang_code in target_langs_for_check
)
if cached_covers_all:
pre_translated.append(row)
continue
else:
logger.reason("Partial cache hit — missing languages, routing to LLM", {
"cached_langs": list(cached_langs.keys()),
"target_langs": target_langs_for_check,
})
# Check for preview edits carry-forward
source_data = row.get("source_data") or {}
if source_data and self._preview_edits_cache:
@@ -618,29 +786,35 @@ class TranslationExecutor:
target_languages = [str(target_languages)]
for row in pre_translated:
cached_langs = row.get("_cached_lang_values") # None for approved, dict for cache hits
record = TranslationRecord(
id=str(uuid.uuid4()),
batch_id=batch_id,
run_id=run_id,
source_sql=row.get("source_text", ""),
target_sql=row.get("approved_translation"),
target_sql=row.get("approved_translation", ""),
source_object_type="table_row",
source_object_id=row.get("row_index"),
source_object_name=row.get("source_object_name", ""),
source_data=row.get("source_data"),
source_hash=row.get("_source_hash"),
status="SUCCESS",
)
self.db.add(record)
# Create per-language entry for each target language (pre-approved)
# Create per-language entry for each target language
for lang_code in target_languages:
if cached_langs and lang_code in cached_langs:
final_val = cached_langs[lang_code]
else:
final_val = row.get("approved_translation", "")
lang_entry = TranslationLanguage(
id=str(uuid.uuid4()),
record_id=record.id,
language_code=lang_code,
source_language_detected="und",
translated_value=row.get("approved_translation"),
final_value=row.get("approved_translation"),
translated_value=final_val,
final_value=final_val,
status="translated",
needs_review=False,
)
@@ -1130,6 +1304,14 @@ class TranslationExecutor:
per_lang_values[target_languages[0]] = translation_text
has_any_translation = True
# ── Dictionary post-processing enforcement ──
# If a dictionary entry matches the source text but the target term
# is missing from the LLM output, apply forced replacement.
if dict_matches and source_text:
_enforce_dictionary(source_text, per_lang_values, dict_matches,
batch_id, row_id)
# ── End dictionary enforcement ──
if not has_any_translation:
# Empty/all-empty translations — skip
skipped += 1
@@ -1163,6 +1345,7 @@ class TranslationExecutor:
source_object_id=row.get("row_index"),
source_object_name=row.get("source_object_name", ""),
source_data=row.get("source_data"),
source_hash=row.get("_source_hash"),
status="SUCCESS",
)
self.db.add(record)

8
backend/src/plugins/translate/preview.py
View File

@@ -49,6 +49,10 @@ DEFAULT_EXECUTION_PROMPT_TEMPLATE: str = (
"Target dialect(s): {target_dialect}\n"
"Column to translate: {translation_column}\n\n"
"{dictionary_section}"
"IMPORTANT — You MUST use the terminology dictionary above. "
"For any source term listed in the dictionary, you MUST use its exact target translation. "
"Do not translate dictionary terms differently. "
"This is mandatory, not optional.\n\n"
"For each row, provide an accurate translation of the text into each target language.\n\n"
"Rows to translate:\n{rows_json}\n\n"
"Respond with a JSON object in this exact format:\n"
@@ -67,6 +71,10 @@ DEFAULT_PREVIEW_PROMPT_TEMPLATE: str = (
"Source dialect: {source_dialect}\n"
"Column to translate: {translation_column}\n\n"
"{dictionary_section}"
"IMPORTANT — You MUST use the terminology dictionary above. "
"For any source term listed in the dictionary, you MUST use its exact target translation. "
"Do not translate dictionary terms differently. "
"This is mandatory, not optional.\n\n"
"Translate to the following languages: {target_languages}\n\n"
"For each row, provide an accurate translation of the '{translation_column}' value into each language.\n"
"Consider the context columns when determining the meaning of the text.\n\n"