sepehr
4d466699fd
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m26s
L1 quality layer — uses a cheap LLM via the OpenAI-compatible API to
validate translation quality. Designed to be the SECOND line of defense
after L0 (script detection, length, pattern).
Architecture:
- sampler.py — picks 5 representative chunks per job (longest first,
skips L0-failed indices, skips too-short or identical pairs)
- llm_judge.py — OpenAI-compatible client, binary verdict per chunk
(accurate / fluent / correct_language / no_leaks), JSON output,
hard timeout, defensive (never raises), cost estimation built in
- pipeline.py — defensive wrapper that integrates both, never breaks
a translation job, always logs a structured event
Integration:
- 5 feature flags in config.py (QUALITY_L1_ENABLED, _LOG_ONLY, etc.)
- QUALITY_L1_LOG_ONLY=true by default: log-only mode, verdict NEVER
blocks or retries a job
- Reuses the chunks extracted by L0 (no double work)
- Passes the set of L0-failed indices so L1 doesn't re-judge them
- Wrapped in try/except so a misconfigured L1 NEVER breaks a job
Default config: deepseek-chat via DeepSeek API
- Cost: ~0.0003 USD per job (5 chunks)
- Speed: typically 1-2s per call, hard ceiling at 8s
- Easy to swap: just set L1_JUDGE_BASE_URL and L1_JUDGE_MODEL
LLM judge is intentionally a SEPARATE model from the translator
(self-evaluation bias mitigation — Meta/Stanford papers 2024-2025).
Tests:
test_sampler.py — 9 tests covering the sampling strategy
test_llm_judge.py — 22 tests covering init, parsing, mocked API,
cost estimation, env factory
test_l1_pipeline.py — 6 tests covering the wrapper
Total new: 37 tests, all pass
Grand total quality+format: 264 tests passing (0 regression)
All 36 new tests + 111 L0 tests + 117 existing translator tests = 264
Phase 1 (observation) for 2 weeks. Then QUALITY_L1_LOG_ONLY=false
to enable auto-retry via the fallback chain.
80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
"""
|
|
Stratified sampler for the L1 quality layer.
|
|
|
|
The L1 layer sends a small number of (source, translation) pairs to a
|
|
cheap LLM for a quality verdict. We don't want to send ALL chunks
|
|
(cost, latency) and we don't want to send random chunks (might miss
|
|
the problematic ones). We use a simple but effective strategy:
|
|
|
|
- Prefer the LONGEST chunks (they contain the most diagnostic
|
|
information per call).
|
|
- Skip chunks that the L0 already flagged (we already know they're
|
|
bad; we don't need the LLM to confirm).
|
|
- Never sample more than `max_samples` chunks.
|
|
- If `min_chunks` chunks aren't available, skip the L1 entirely
|
|
(small documents don't need it).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import List, Tuple
|
|
|
|
|
|
def sample_chunks_for_l1(
|
|
source_chunks: List[str],
|
|
translated_chunks: List[str],
|
|
failed_indices: set,
|
|
max_samples: int = 5,
|
|
min_chunks: int = 10,
|
|
) -> List[Tuple[str, str]]:
|
|
"""
|
|
Select a sample of (source, translation) pairs to send to the L1 judge.
|
|
|
|
Args:
|
|
source_chunks: Original texts.
|
|
translated_chunks: Translated texts (same length as source_chunks).
|
|
failed_indices: Set of indices that L0 already flagged as bad.
|
|
These are SKIPPED — we want fresh signal, not
|
|
confirmation of an already-detected failure.
|
|
max_samples: Maximum number of pairs to return.
|
|
min_chunks: If the document has fewer than this many chunks,
|
|
return an empty list (not enough signal to bother
|
|
calling the LLM).
|
|
|
|
Returns:
|
|
A list of (source, translated) tuples, ready to send to the LLM
|
|
judge. Order: longest chunks first.
|
|
"""
|
|
n = min(len(source_chunks), len(translated_chunks))
|
|
|
|
if n < min_chunks:
|
|
return []
|
|
|
|
# Build candidate list, excluding L0 failures
|
|
candidates: List[Tuple[int, int, str, str]] = []
|
|
for i in range(n):
|
|
if i in failed_indices:
|
|
continue
|
|
src = (source_chunks[i] or "").strip()
|
|
trans = (translated_chunks[i] or "").strip()
|
|
# Skip very short pairs (not enough context for the LLM to judge)
|
|
if len(src) < 5 and len(trans) < 5:
|
|
continue
|
|
# Skip pairs where source and translation are identical
|
|
# (probably a non-translated cell like a number, code, or brand)
|
|
if src == trans:
|
|
continue
|
|
# Rank by length of the translation (longer = more diagnostic)
|
|
rank = len(trans) + len(src) // 2
|
|
candidates.append((rank, i, src, trans))
|
|
|
|
# Sort by length descending and take top N
|
|
candidates.sort(key=lambda c: c[0], reverse=True)
|
|
selected = candidates[:max_samples]
|
|
|
|
# Return in original document order (helps the LLM judge maintain
|
|
# context, and makes the verdict easier to interpret)
|
|
selected.sort(key=lambda c: c[1])
|
|
|
|
return [(src, trans) for _rank, _i, src, trans in selected]
|