diff --git a/.env.example b/.env.example index 9ec936d..8409717 100644 --- a/.env.example +++ b/.env.example @@ -104,6 +104,30 @@ MAX_CONCURRENT_TRANSLATIONS=5 QUALITY_L0_ENABLED=false QUALITY_L0_SAMPLE_SIZE=20 +# ============== Quality Layer (L1) ============== +# Track A3 of the dev plan — API-based LLM judge. +# Sends 5 sampled chunks per job to a cheap LLM (deepseek-chat by default) +# and logs the verdict (pass/fail) but does NOT modify the file or job +# status. Log-only by default for the first 2 weeks of observation. +# After that, set QUALITY_L1_LOG_ONLY=false to enable auto-retry. +# +# Cost: ~$0.0003 per job with deepseek-chat. ~$0.001 with gpt-4o-mini. +QUALITY_L1_ENABLED=false +QUALITY_L1_LOG_ONLY=true +QUALITY_L1_SAMPLE_SIZE=5 +QUALITY_L1_MIN_CHUNKS=10 +QUALITY_L1_TIMEOUT_SEC=8.0 + +# L1 judge configuration (any OpenAI-compatible endpoint). +# DeepSeek is the default (cheapest, ~$0.14/M input tokens). +L1_JUDGE_API_KEY= +L1_JUDGE_BASE_URL=https://api.deepseek.com/v1 +L1_JUDGE_MODEL=deepseek-chat +# L1_JUDGE_MODEL=gpt-4o-mini +# L1_JUDGE_BASE_URL=https://api.openai.com/v1 +# L1_JUDGE_MODEL=google/gemini-2.5-flash-lite +# L1_JUDGE_BASE_URL=https://openrouter.ai/api/v1 + # ============== Cleanup Service ============== # Enable automatic file cleanup CLEANUP_ENABLED=true diff --git a/config.py b/config.py index 07bad69..53aa68b 100644 --- a/config.py +++ b/config.py @@ -77,6 +77,21 @@ class Config: QUALITY_L0_SAMPLE_SIZE = int(os.getenv("QUALITY_L0_SAMPLE_SIZE", "20")) + # ============== Quality Layer (L1) ============== + # Track A3 of the dev plan — API-based LLM judge. + # Observability first; the verdict is logged but never used to retry + # or block a job (QUALITY_L1_LOG_ONLY=true). After 2 weeks of + # monitoring, set QUALITY_L1_LOG_ONLY=false to enable auto-retry. + QUALITY_L1_ENABLED = os.getenv("QUALITY_L1_ENABLED", "false").lower() == "true" + QUALITY_L1_LOG_ONLY = os.getenv("QUALITY_L1_LOG_ONLY", "true").lower() == "true" + # Number of chunks sampled per job. 5 is the sweet spot (cost vs coverage). + QUALITY_L1_SAMPLE_SIZE = int(os.getenv("QUALITY_L1_SAMPLE_SIZE", "5")) + # Skip the check if the document has fewer than this many chunks. + QUALITY_L1_MIN_CHUNKS = int(os.getenv("QUALITY_L1_MIN_CHUNKS", "10")) + # Hard ceiling on the L1 call (seconds). Anything longer is a skip. + QUALITY_L1_TIMEOUT_SEC = float(os.getenv("QUALITY_L1_TIMEOUT_SEC", "8.0")) + + # ============== API Configuration ============== API_TITLE = "Document Translation API" API_VERSION = "1.0.0" diff --git a/routes/translate_routes.py b/routes/translate_routes.py index a26444f..5aa2b98 100644 --- a/routes/translate_routes.py +++ b/routes/translate_routes.py @@ -1360,28 +1360,70 @@ async def _run_translation_job( # L0 quality checks. NEVER blocks the job, NEVER modifies the file. # Enabled by feature flag QUALITY_L0_ENABLED (default: false). # ------------------------------------------------------------------ + quality_samples: list = [] # captured here for L1 to reuse + l0_failed_indices: set = set() # captured for L1 sampling if getattr(config, "QUALITY_L0_ENABLED", False): try: from services.quality import run_l0_check, extract_sample - samples = extract_sample( + quality_samples = extract_sample( output_path, file_extension, max_samples=getattr(config, "QUALITY_L0_SAMPLE_SIZE", 20), ) - translated_chunks = [s["translated"] for s in samples] - run_l0_check( + translated_chunks = [s["translated"] for s in quality_samples] + l0_result = run_l0_check( source_chunks=[""] * len(translated_chunks), # L0 doesn't need source translated_chunks=translated_chunks, target_lang=target_lang, job_id=job_id, file_extension=file_extension, ) + # Build the set of indices that L0 flagged as bad + if not l0_result.passed and l0_result.samples: + l0_failed_indices = {s["index"] for s in l0_result.samples} except Exception as l0_err: # Quality L0 must NEVER break a job. Log and continue. logger.warning( f"Job {job_id}: quality L0 layer failed: {l0_err}" ) + # ------------------------------------------------------------------ + # Quality L1 layer (Track A3 — API-based LLM judge, observability first) + # Sends a small sample of chunks to a cheap LLM (deepseek-chat by + # default) and logs a binary verdict (pass/fail). Log-only by + # default — the verdict NEVER blocks a job, NEVER triggers a retry. + # After 2 weeks of observation, set QUALITY_L1_LOG_ONLY=false to + # enable auto-retry. + # Cost: ~$0.0003/job (deepseek) or ~$0.001/job (gpt-4o-mini). + # ------------------------------------------------------------------ + if getattr(config, "QUALITY_L1_ENABLED", False) and quality_samples: + try: + from services.quality import run_l1_check + translated_chunks_for_l1 = [s["translated"] for s in quality_samples] + # The samples are extracted from the OUTPUT, not the source — + # we don't have the source here. L1 still works because the + # judge can spot language confusion, gibberish, repetition, + # and prompt leaks without the source. (Source IS used for + # the "accurate" check, so the verdict on that dimension is + # conservative when no source is available — we expect + # the judge to be honest about what it can verify.) + l1_result = await run_l1_check( + source_chunks=[""] * len(translated_chunks_for_l1), + translated_chunks=translated_chunks_for_l1, + target_lang=target_lang, + l0_failed_indices=l0_failed_indices, + job_id=job_id, + file_extension=file_extension, + max_samples=getattr(config, "QUALITY_L1_SAMPLE_SIZE", 5), + min_chunks=getattr(config, "QUALITY_L1_MIN_CHUNKS", 10), + log_only=getattr(config, "QUALITY_L1_LOG_ONLY", True), + ) + except Exception as l1_err: + # L1 must NEVER break a job. Log and continue. + logger.warning( + f"Job {job_id}: quality L1 layer failed: {l1_err}" + ) + if user_id: # Determine cost factor based on selected provider and model cost_factor = 1 diff --git a/services/quality/__init__.py b/services/quality/__init__.py index 24c83fc..dd07445 100644 --- a/services/quality/__init__.py +++ b/services/quality/__init__.py @@ -1,17 +1,28 @@ """ Quality check layer for translations. -Track A1 — L0 backend (observation only). -Pure Python, no new dependencies, no network calls. +Tracks: + A1 — L0 backend (pure Python, no API calls). Always available. + A2 — L0 frontend JavaScript mirror. Browser/Node compatible. + A3 — L1 LLM judge (API-based, opt-in). Cheap model, sampled. + Designed to be ADDITIVE: existing translation flow is untouched. Public API: + L0: QualityCheckResult — per-chunk result dataclass DocumentQualityResult — aggregated result dataclass evaluate_chunk(...) — score a single (source, translation) pair evaluate_document(...) — score a list of pairs and aggregate run_l0_check(...) — defensive wrapper used by the route extract_sample(...) — extract text from a finished file + L1: + L1Result — verdict dataclass + LLMJudge — judge client + run_l1_check(...) — async wrapper used by the route + sample_chunks_for_l1(...) — pick representative chunks + Helpers: + get_script, isArabicScriptLang """ from .script_detector import ( @@ -21,10 +32,13 @@ from .script_detector import ( evaluate_document, detect_arabic_variant, ) -from .pipeline import run_l0_check +from .pipeline import run_l0_check, run_l1_check from .file_extractor import extract_sample +from .sampler import sample_chunks_for_l1 +from .llm_judge import L1Result, L1ChunkVerdict, LLMJudge __all__ = [ + # L0 "QualityCheckResult", "DocumentQualityResult", "evaluate_chunk", @@ -32,4 +46,10 @@ __all__ = [ "detect_arabic_variant", "run_l0_check", "extract_sample", + # L1 + "L1Result", + "L1ChunkVerdict", + "LLMJudge", + "run_l1_check", + "sample_chunks_for_l1", ] diff --git a/services/quality/llm_judge.py b/services/quality/llm_judge.py new file mode 100644 index 0000000..9a9d427 --- /dev/null +++ b/services/quality/llm_judge.py @@ -0,0 +1,365 @@ +""" +L1 LLM Judge — uses a cheap LLM via the OpenAI-compatible API to validate +the quality of a translation. + +Why a SEPARATE LLM (not the one that did the translation)? + - Self-evaluation bias: a model tends to rate its own output as good + (Meta/Stanford papers on LLM-as-judge bias, 2024-2025). + - Independence gives a more reliable signal. + +Design constraints: + - 100% API-based. No local models, no GPU. + - Async, with a hard timeout. + - Defensive: never raises. Returns a "skip" verdict on any internal error. + - Output is structured JSON for reliable parsing. + - Sampled (we never judge all chunks, just a small subset). + +The LLM is asked a binary question per chunk: "Is this translation +accurate and fluent, in the correct language?" with a short reason. +We do NOT ask for nuanced MQM scores — binary is more reliable and +cheaper. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import re +import time +from dataclasses import dataclass, field, asdict +from typing import List, Optional, Tuple + +from core.logging import get_logger + +logger = get_logger(__name__) + + +# ---------- Result dataclasses ---------- + +@dataclass +class L1ChunkVerdict: + """Verdict for a single (source, translation) pair.""" + accurate: bool + fluent: bool + correct_language: bool + no_leaks: bool + reason: str = "" + + @property + def passed(self) -> bool: + return self.accurate and self.fluent and self.correct_language and self.no_leaks + + def to_log_dict(self) -> dict: + return asdict(self) + + +@dataclass +class L1Result: + """Aggregate result of an L1 check on a sample of chunks.""" + verdict: str # "pass", "fail", "skip" + chunks_evaluated: int + chunks_passed: int + chunks_failed: int + failure_rate: float # 0.0 to 1.0 + samples: List[dict] = field(default_factory=list) + model_used: str = "" + elapsed_ms: float = 0.0 + cost_estimate_usd: float = 0.0 + error: str = "" + + def to_log_dict(self) -> dict: + return asdict(self) + + +# ---------- Prompt template ---------- + +JUDGE_SYSTEM_PROMPT = """You are a strict translation quality evaluator. Your job is to detect translation failures. + +For each (SOURCE, TRANSLATION) pair, check these 4 criteria: + +1. ACCURATE — Does the translation preserve the meaning of the source? (yes/no) +2. FLUENT — Is the translation grammatically correct in the target language? (yes/no) +3. CORRECT_LANGUAGE — Is the translation actually in {target_lang_name} (ISO code: {target_lang})? (yes/no) +4. NO_LEAKS — Is the translation free of prompt artifacts, source-language text, or meta-commentary? (yes/no) + +A translation FAILS if ANY of the 4 checks is "no". + +Respond with a JSON array, one object per pair, in the same order. NO other text, NO markdown fences: +[ + {{"accurate": "yes"|"no", "fluent": "yes"|"no", "correct_language": "yes"|"no", "no_leaks": "yes"|"no", "reason": "one short sentence"}} +] + +Be honest: if the translation looks suspicious, say "no". The "reason" must be in English and ≤ 12 words. +""" + + +# ---------- LLM client ---------- + +class LLMJudge: + """ + Calls a cheap LLM via the OpenAI-compatible API to judge translation quality. + + Default configuration targets deepseek-chat via the OpenAI-compatible + DeepSeek API. The judge can be reconfigured to use any OpenAI-compatible + endpoint (OpenAI, OpenRouter, etc.) by passing different params. + """ + + def __init__( + self, + api_key: str, + base_url: str = "https://api.deepseek.com/v1", + model: str = "deepseek-chat", + timeout_seconds: float = 8.0, + max_retries: int = 1, + ): + if not api_key: + raise ValueError("api_key is required for LLMJudge") + self._api_key = api_key + self._base_url = base_url.rstrip("/") + self._model = model + self._timeout = timeout_seconds + self._max_retries = max_retries + # Lazy import — keep services/quality free of the openai dep at import time + self._client = None + + def _get_client(self): + if self._client is None: + try: + import openai + self._client = openai.AsyncOpenAI( + api_key=self._api_key, + base_url=self._base_url, + timeout=self._timeout, + ) + except Exception as e: + logger.warning("llm_judge_client_init_failed", error=str(e)) + return None + return self._client + + async def judge_batch( + self, + pairs: List[Tuple[str, str]], + target_lang: str, + target_lang_name: str = "", + ) -> L1Result: + """ + Judge a batch of (source, translation) pairs. + + Returns an L1Result with verdict="skip" on any internal error. + Never raises. + """ + start = time.time() + if not pairs: + return L1Result(verdict="skip", chunks_evaluated=0, chunks_passed=0, + chunks_failed=0, failure_rate=0.0, + error="empty pairs") + + client = self._get_client() + if client is None: + return L1Result(verdict="skip", chunks_evaluated=0, chunks_passed=0, + chunks_failed=0, failure_rate=0.0, + error="client unavailable") + + # Build the user message + user_lines = [f"Target language: {target_lang} ({target_lang_name})\n"] + for i, (src, trans) in enumerate(pairs, 1): + user_lines.append(f"\n--- Pair {i} ---") + user_lines.append(f"SOURCE: {src}") + user_lines.append(f"TRANSLATION: {trans}") + user_msg = "\n".join(user_lines) + + system_prompt = JUDGE_SYSTEM_PROMPT.format( + target_lang=target_lang, + target_lang_name=target_lang_name or target_lang, + ) + + try: + response = await asyncio.wait_for( + self._call_with_retries(client, system_prompt, user_msg), + timeout=self._timeout + 2.0, # hard ceiling + ) + except asyncio.TimeoutError: + elapsed_ms = round((time.time() - start) * 1000, 2) + logger.warning("llm_judge_timeout", timeout_s=self._timeout, elapsed_ms=elapsed_ms) + return L1Result(verdict="skip", chunks_evaluated=0, chunks_passed=0, + chunks_failed=0, failure_rate=0.0, + error="timeout", elapsed_ms=elapsed_ms) + except Exception as e: + elapsed_ms = round((time.time() - start) * 1000, 2) + logger.warning("llm_judge_error", error=str(e)[:200], elapsed_ms=elapsed_ms) + return L1Result(verdict="skip", chunks_evaluated=0, chunks_passed=0, + chunks_failed=0, failure_rate=0.0, + error=str(e)[:200], elapsed_ms=elapsed_ms) + + # Parse the response + verdicts = self._parse_response(response, len(pairs)) + + passed = sum(1 for v in verdicts if v.passed) + failed = len(verdicts) - passed + failure_rate = (failed / len(verdicts)) if verdicts else 0.0 + + # Aggregate verdict + if failed == 0: + verdict = "pass" + elif failure_rate >= 0.5: + verdict = "fail" + else: + # Some pass, some fail — degraded but not catastrophic. + # Caller can decide what to do. + verdict = "fail" # conservative: any failure = fail + + elapsed_ms = round((time.time() - start) * 1000, 2) + + # Cost estimate: deepseek-chat at $0.14/M in, $0.28/M out + # Rough estimate: 500 input tokens + 100 output tokens per call + cost_estimate = self._estimate_cost(len(pairs)) + + return L1Result( + verdict=verdict, + chunks_evaluated=len(verdicts), + chunks_passed=passed, + chunks_failed=failed, + failure_rate=round(failure_rate, 3), + samples=[v.to_log_dict() for v in verdicts], + model_used=self._model, + elapsed_ms=elapsed_ms, + cost_estimate_usd=cost_estimate, + ) + + async def _call_with_retries(self, client, system_prompt: str, user_msg: str): + """Call the LLM with retry on transient errors.""" + last_exc = None + for attempt in range(self._max_retries + 1): + try: + response = await client.chat.completions.create( + model=self._model, + messages=[ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_msg}, + ], + temperature=0.0, # deterministic + max_tokens=500, # enough for ~5 verdicts + response_format={"type": "json_object"}, # if supported + ) + return response + except Exception as e: + last_exc = e + if attempt < self._max_retries: + await asyncio.sleep(0.5) + raise last_exc + + def _parse_response(self, response, expected_count: int) -> List[L1ChunkVerdict]: + """ + Parse the LLM response into a list of verdicts. + Robust to JSON wrapped in code fences or with extra commentary. + """ + # Extract the message content + try: + content = response.choices[0].message.content or "" + except (AttributeError, IndexError) as e: + logger.warning("llm_judge_bad_response", error=str(e)) + return [] + + content = content.strip() + + # Strip markdown code fences if present + if content.startswith("```"): + content = re.sub(r"^```(?:json)?\s*\n?", "", content) + content = re.sub(r"\n?```\s*$", "", content) + + # Try to parse as JSON object (deepseek with json_object format) or array + try: + data = json.loads(content) + except json.JSONDecodeError as e: + logger.warning("llm_judge_json_parse_error", error=str(e), content_preview=content[:200]) + return [] + + # Handle both {"verdicts": [...]} and direct [...] + if isinstance(data, dict): + # Look for a list-typed value + items = None + for key in ("verdicts", "results", "translations", "data"): + if key in data and isinstance(data[key], list): + items = data[key] + break + if items is None: + # Take the first list-typed value + for v in data.values(): + if isinstance(v, list): + items = v + break + if items is None: + logger.warning("llm_judge_no_list_in_response") + return [] + elif isinstance(data, list): + items = data + else: + logger.warning("llm_judge_unexpected_response_type", type_=type(data).__name__) + return [] + + # Parse each item + verdicts: List[L1ChunkVerdict] = [] + for item in items: + try: + v = L1ChunkVerdict( + accurate=str(item.get("accurate", "")).lower() == "yes", + fluent=str(item.get("fluent", "")).lower() == "yes", + correct_language=str(item.get("correct_language", "")).lower() == "yes", + no_leaks=str(item.get("no_leaks", "")).lower() == "yes", + reason=str(item.get("reason", ""))[:200], + ) + verdicts.append(v) + except Exception as e: + logger.warning("llm_judge_item_parse_error", error=str(e), item=str(item)[:200]) + # Continue with what we have + + return verdicts + + def _estimate_cost(self, num_pairs: int) -> float: + """Rough USD cost estimate for the call. Conservative (rounded up).""" + # Approximation: 250 input tokens per pair + 50 output tokens per pair + # (rough based on the prompt template + JSON output) + input_tokens = 200 + (num_pairs * 250) + output_tokens = num_pairs * 50 + # deepseek-chat pricing + if "deepseek" in self._model.lower(): + input_cost = input_tokens / 1_000_000 * 0.14 + output_cost = output_tokens / 1_000_000 * 0.28 + elif "gpt-4o-mini" in self._model.lower(): + input_cost = input_tokens / 1_000_000 * 0.15 + output_cost = output_tokens / 1_000_000 * 0.60 + elif "gemini" in self._model.lower() and "flash" in self._model.lower(): + # gemini-2.5-flash-lite ~ $0.10/M in, $0.40/M out + input_cost = input_tokens / 1_000_000 * 0.10 + output_cost = output_tokens / 1_000_000 * 0.40 + else: + # Generic conservative estimate + input_cost = input_tokens / 1_000_000 * 0.50 + output_cost = output_tokens / 1_000_000 * 1.00 + return round(input_cost + output_cost, 6) + + +# ---------- Convenience factory ---------- + +def make_judge_from_env() -> Optional[LLMJudge]: + """ + Build an LLMJudge from environment variables. Returns None if no + API key is configured. + + Reads: + - L1_JUDGE_API_KEY (required) + - L1_JUDGE_BASE_URL (default: DeepSeek) + - L1_JUDGE_MODEL (default: deepseek-chat) + - L1_JUDGE_TIMEOUT (default: 8.0) + """ + import os + api_key = os.getenv("L1_JUDGE_API_KEY", "").strip() + if not api_key: + return None + return LLMJudge( + api_key=api_key, + base_url=os.getenv("L1_JUDGE_BASE_URL", "https://api.deepseek.com/v1"), + model=os.getenv("L1_JUDGE_MODEL", "deepseek-chat"), + timeout_seconds=float(os.getenv("L1_JUDGE_TIMEOUT", "8.0")), + ) diff --git a/services/quality/pipeline.py b/services/quality/pipeline.py index e9bb6f6..11cb57b 100644 --- a/services/quality/pipeline.py +++ b/services/quality/pipeline.py @@ -1,27 +1,34 @@ """ -Quality pipeline — defensive wrapper around the L0 checks. +Quality pipeline — defensive wrapper around the L0 and L1 checks. The pipeline is the integration point for the route. It: - 1. Catches all exceptions (L0 must NEVER break a translation job) + 1. Catches all exceptions (quality checks must NEVER break a translation job) 2. Adds timing 3. Emits a single structured log line per job + 4. For L1, reads configuration from environment and is opt-in via + a flag passed in by the route (the route reads config.py). -The actual checks live in `script_detector`, `length_checker`, `pattern_leak`. -This module is the orchestration / safety layer. +L0 = pure-Python script + length + pattern checks. No API calls. +L1 = cheap LLM judge via API. Sampled (5 chunks per job by default). """ from __future__ import annotations +import os import time -from typing import List, Optional +from typing import List, Optional, Set from core.logging import get_logger from .script_detector import evaluate_document, DocumentQualityResult +from .sampler import sample_chunks_for_l1 +from .llm_judge import L1Result, LLMJudge logger = get_logger(__name__) +# ---------- L0 ---------- + def run_l0_check( source_chunks: List[str], translated_chunks: List[str], @@ -31,10 +38,6 @@ def run_l0_check( ) -> DocumentQualityResult: """ Run the L0 quality checks defensively. Never raises. - - Returns an empty/neutral DocumentQualityResult on internal error - so the calling route can log and continue without affecting the - translation job outcome. """ start = time.time() empty = DocumentQualityResult( @@ -73,3 +76,138 @@ def run_l0_check( elapsed_ms=elapsed_ms, ) return result + + +# ---------- L1 ---------- + +# Common ISO codes → language name for the prompt +_LANG_NAMES = { + "en": "English", "fr": "French", "es": "Spanish", "de": "German", + "it": "Italian", "pt": "Portuguese", "nl": "Dutch", "ru": "Russian", + "ja": "Japanese", "ko": "Korean", "zh": "Chinese", "ar": "Arabic", + "fa": "Persian", "hi": "Hindi", "tr": "Turkish", "pl": "Polish", + "vi": "Vietnamese", "th": "Thai", "id": "Indonesian", "ms": "Malay", + "uk": "Ukrainian", "cs": "Czech", "sv": "Swedish", "ro": "Romanian", + "hu": "Hungarian", "el": "Greek", "he": "Hebrew", "da": "Danish", + "fi": "Finnish", "no": "Norwegian", "bg": "Bulgarian", "hr": "Croatian", +} + + +async def run_l1_check( + source_chunks: List[str], + translated_chunks: List[str], + target_lang: Optional[str], + l0_failed_indices: Optional[Set[int]] = None, + job_id: Optional[str] = None, + file_extension: Optional[str] = None, + max_samples: int = 5, + min_chunks: int = 10, + judge: Optional[LLMJudge] = None, + log_only: bool = True, +) -> L1Result: + """ + Run the L1 LLM judge check on a sample of chunks. + + Args: + source_chunks: Original texts. + translated_chunks: Translated texts. + target_lang: Target language code (e.g. "fr", "en"). + l0_failed_indices: Indices that L0 flagged as bad — skipped. + job_id: For logging. + file_extension: For logging. + max_samples: How many chunks to send to the LLM. + min_chunks: Skip the check if document has fewer chunks. + judge: An LLMJudge instance. If None, created from env vars. + log_only: If True, never propagate the verdict (observation mode). + If False, the caller can decide what to do with the verdict. + + Returns an L1Result. verdict="skip" on any internal error. + Never raises — defensive wrapper. + """ + skip = L1Result( + verdict="skip", + chunks_evaluated=0, chunks_passed=0, chunks_failed=0, + failure_rate=0.0, error="not_run", + ) + + if l0_failed_indices is None: + l0_failed_indices = set() + + # Sample + sample = sample_chunks_for_l1( + source_chunks, translated_chunks, l0_failed_indices, + max_samples=max_samples, min_chunks=min_chunks, + ) + if not sample: + logger.info( + "quality_l1_check_skipped", + job_id=job_id, + reason="insufficient_chunks_or_all_flagged", + chunk_count=len(source_chunks), + ) + return skip + + # Get the judge + if judge is None: + judge = make_judge_from_env_safe() + + if judge is None: + logger.info( + "quality_l1_check_skipped", + job_id=job_id, + reason="no_judge_configured", + ) + return skip + + # Get the language name for the prompt + target_lang_name = _LANG_NAMES.get((target_lang or "").lower(), target_lang or "auto") + + # Call the LLM + try: + result = await judge.judge_batch(sample, target_lang or "auto", target_lang_name) + except Exception as e: + elapsed_ms = 0.0 + logger.warning( + "quality_l1_check_failed", + job_id=job_id, + error=str(e)[:200], + error_type=type(e).__name__, + ) + return L1Result( + verdict="skip", + chunks_evaluated=0, chunks_passed=0, chunks_failed=0, + failure_rate=0.0, error=str(e)[:200], elapsed_ms=elapsed_ms, + ) + + # Log (always) — caller decides what to do + logger.info( + "quality_l1_check", + job_id=job_id, + file_extension=file_extension, + target_lang=target_lang, + verdict=result.verdict, + chunks_evaluated=result.chunks_evaluated, + chunks_passed=result.chunks_passed, + chunks_failed=result.chunks_failed, + failure_rate=result.failure_rate, + model=result.model_used, + elapsed_ms=result.elapsed_ms, + cost_estimate_usd=result.cost_estimate_usd, + log_only=log_only, + ) + return result + + +def make_judge_from_env_safe() -> Optional[LLMJudge]: + """Read env vars and build a judge, or return None if not configured. + + This is a thin wrapper around `llm_judge.make_judge_from_env` that + catches any import-time or init-time error and returns None — so + a misconfigured L1 environment NEVER breaks a translation job. + """ + try: + from .llm_judge import make_judge_from_env + return make_judge_from_env() + except Exception as e: + logger.warning("l1_judge_init_failed", error=str(e)[:200]) + return None diff --git a/services/quality/sampler.py b/services/quality/sampler.py new file mode 100644 index 0000000..a91113d --- /dev/null +++ b/services/quality/sampler.py @@ -0,0 +1,79 @@ +""" +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] diff --git a/tests/services/quality/test_l1_pipeline.py b/tests/services/quality/test_l1_pipeline.py new file mode 100644 index 0000000..6d59ff8 --- /dev/null +++ b/tests/services/quality/test_l1_pipeline.py @@ -0,0 +1,127 @@ +""" +Tests for services/quality/pipeline.py — the L1 wrapper. +""" +import asyncio +import os +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from services.quality.pipeline import run_l1_check +from services.quality.llm_judge import L1Result, LLMJudge + + +class TestRunL1Check: + """Test the defensive wrapper around the LLM judge.""" + + def test_too_few_chunks_returns_skip(self): + sources = ["a", "b", "c"] # only 3 chunks + translations = ["x", "y", "z"] + result = asyncio.run(run_l1_check( + sources, translations, "fr", + max_samples=5, min_chunks=10, + )) + assert result.verdict == "skip" + + def test_no_judge_configured_returns_skip(self, monkeypatch): + monkeypatch.delenv("L1_JUDGE_API_KEY", raising=False) + sources = [f"source {i}" for i in range(15)] + translations = [f"translation {i}" for i in range(15)] + result = asyncio.run(run_l1_check( + sources, translations, "fr", + max_samples=5, min_chunks=10, + )) + assert result.verdict == "skip" + assert "judge" in result.error.lower() or "configured" in result.error.lower() or result.error == "not_run" + + def test_skips_l0_failures(self): + """Chunks that L0 already flagged should be excluded from the sample.""" + # All chunks are long enough to be sampled. Use unique markers per + # index so substring checks don't have false positives (e.g. + # "source1" being a substring of "source10"). + sources = [f"srcidx{i:02d} with enough length" for i in range(15)] + translations = [f"transidx{i:02d} avec longueur suffisante" for i in range(15)] + + # Mock judge that records what it received + mock_judge = MagicMock(spec=LLMJudge) + mock_judge.judge_batch = AsyncMock(return_value=L1Result( + verdict="pass", chunks_evaluated=5, chunks_passed=5, + chunks_failed=0, failure_rate=0.0, + )) + + # Mark indices 01, 03, 05, 07, 09, 11, 13 as L0 failures + l0_failures = {1, 3, 5, 7, 9, 11, 13} + + result = asyncio.run(run_l1_check( + sources, translations, "fr", + l0_failed_indices=l0_failures, + max_samples=5, min_chunks=10, + judge=mock_judge, + )) + + # The mock was called with 5 pairs + call_args = mock_judge.judge_batch.call_args + pairs = call_args[0][0] + # The pairs should NOT include any source from the L0-failed indices + for src, _trans in pairs: + for failed_idx in l0_failures: + marker = f"srcidx{failed_idx:02d}" + assert marker not in src, ( + f"L0-failed source {failed_idx} was sent to L1: {src}" + ) + + def test_passes_log_only_flag(self): + """The log_only flag should be passed through to the LLMJudge result.""" + mock_judge = MagicMock(spec=LLMJudge) + mock_judge.judge_batch = AsyncMock(return_value=L1Result( + verdict="pass", chunks_evaluated=1, chunks_passed=1, + chunks_failed=0, failure_rate=0.0, + )) + + sources = [f"long enough source {i} " * 3 for i in range(15)] + translations = [f"long enough translation {i} " * 3 for i in range(15)] + + asyncio.run(run_l1_check( + sources, translations, "fr", + max_samples=5, min_chunks=10, + judge=mock_judge, + log_only=True, + )) + # log_only doesn't affect the call, just the downstream decision + + def test_never_raises_on_judge_error(self): + """If the judge itself raises, run_l1_check should catch it.""" + mock_judge = MagicMock(spec=LLMJudge) + mock_judge.judge_batch = AsyncMock(side_effect=Exception("boom")) + + sources = [f"long enough source {i} " * 3 for i in range(15)] + translations = [f"long enough translation {i} " * 3 for i in range(15)] + + # Should NOT raise + result = asyncio.run(run_l1_check( + sources, translations, "fr", + max_samples=5, min_chunks=10, + judge=mock_judge, + )) + assert result is not None + # verdict is either "skip" or an error one — never crashes the call + + def test_target_lang_name_for_known_lang(self): + """Verify the LANG_NAMES mapping is used.""" + mock_judge = MagicMock(spec=LLMJudge) + mock_judge.judge_batch = AsyncMock(return_value=L1Result( + verdict="pass", chunks_evaluated=1, chunks_passed=1, + chunks_failed=0, failure_rate=0.0, + )) + + sources = [f"long enough source {i} " * 3 for i in range(15)] + translations = [f"long enough translation {i} " * 3 for i in range(15)] + + asyncio.run(run_l1_check( + sources, translations, "fr", + max_samples=5, min_chunks=10, + judge=mock_judge, + )) + call_args = mock_judge.judge_batch.call_args + # target_lang_name is the 3rd positional arg + target_lang_name = call_args[0][2] + assert target_lang_name == "French" diff --git a/tests/services/quality/test_llm_judge.py b/tests/services/quality/test_llm_judge.py new file mode 100644 index 0000000..efc61e0 --- /dev/null +++ b/tests/services/quality/test_llm_judge.py @@ -0,0 +1,258 @@ +""" +Tests for services/quality/llm_judge.py +Uses mocks for the OpenAI client — no actual API calls. +""" +import asyncio +import json +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from services.quality.llm_judge import ( + LLMJudge, + L1Result, + L1ChunkVerdict, + JUDGE_SYSTEM_PROMPT, + make_judge_from_env, +) + + +# ---------- Judge init ---------- + +class TestLLMJudgeInit: + def test_requires_api_key(self): + with pytest.raises(ValueError): + LLMJudge(api_key="") + + def test_init_with_defaults(self): + judge = LLMJudge(api_key="test-key") + assert judge._api_key == "test-key" + assert judge._model == "deepseek-chat" + assert judge._base_url == "https://api.deepseek.com/v1" + + def test_init_with_custom_params(self): + judge = LLMJudge( + api_key="k", + base_url="https://api.openai.com/v1/", + model="gpt-4o-mini", + timeout_seconds=15.0, + ) + assert judge._base_url == "https://api.openai.com/v1" # trailing slash stripped + assert judge._model == "gpt-4o-mini" + assert judge._timeout == 15.0 + + +# ---------- Response parsing ---------- + +class TestParseResponse: + def setup_method(self): + self.judge = LLMJudge(api_key="test-key") + + def _make_response(self, content: str): + response = MagicMock() + response.choices = [MagicMock()] + response.choices[0].message.content = content + return response + + def test_parse_pure_json_array(self): + content = json.dumps([ + {"accurate": "yes", "fluent": "yes", "correct_language": "yes", + "no_leaks": "yes", "reason": "good"} + ]) + verdicts = self.judge._parse_response(self._make_response(content), 1) + assert len(verdicts) == 1 + assert verdicts[0].passed is True + assert verdicts[0].accurate is True + assert verdicts[0].reason == "good" + + def test_parse_json_object_with_verdicts_key(self): + content = json.dumps({"verdicts": [ + {"accurate": "no", "fluent": "yes", "correct_language": "yes", + "no_leaks": "yes", "reason": "lost meaning"} + ]}) + verdicts = self.judge._parse_response(self._make_response(content), 1) + assert len(verdicts) == 1 + assert verdicts[0].passed is False + assert verdicts[0].accurate is False + assert verdicts[0].fluent is True + assert verdicts[0].reason == "lost meaning" + + def test_parse_json_with_markdown_fences(self): + content = "```json\n" + json.dumps([ + {"accurate": "yes", "fluent": "yes", "correct_language": "yes", + "no_leaks": "yes", "reason": "ok"} + ]) + "\n```" + verdicts = self.judge._parse_response(self._make_response(content), 1) + assert len(verdicts) == 1 + assert verdicts[0].passed is True + + def test_parse_invalid_json_returns_empty(self): + content = "not json at all" + verdicts = self.judge._parse_response(self._make_response(content), 1) + assert verdicts == [] + + def test_parse_partial_verdict_defaults_to_false(self): + content = json.dumps([{"accurate": "yes"}]) # missing other fields + verdicts = self.judge._parse_response(self._make_response(content), 1) + assert len(verdicts) == 1 + # All fields default to False when missing + assert verdicts[0].passed is False + assert verdicts[0].accurate is True + assert verdicts[0].fluent is False + assert verdicts[0].correct_language is False + assert verdicts[0].no_leaks is False + + def test_parse_mixed_pass_fail(self): + content = json.dumps([ + {"accurate": "yes", "fluent": "yes", "correct_language": "yes", + "no_leaks": "yes", "reason": "good"}, + {"accurate": "no", "fluent": "yes", "correct_language": "yes", + "no_leaks": "yes", "reason": "mistranslation"}, + {"accurate": "yes", "fluent": "no", "correct_language": "yes", + "no_leaks": "yes", "reason": "awkward"}, + ]) + verdicts = self.judge._parse_response(self._make_response(content), 3) + assert len(verdicts) == 3 + assert verdicts[0].passed is True + assert verdicts[1].passed is False + assert verdicts[2].passed is False + + +# ---------- L1Result ---------- + +class TestL1Result: + def test_passed_property(self): + v = L1ChunkVerdict(accurate=True, fluent=True, correct_language=True, no_leaks=True) + assert v.passed is True + + v_fail = L1ChunkVerdict(accurate=True, fluent=True, correct_language=True, no_leaks=False) + assert v_fail.passed is False + + +# ---------- Cost estimation ---------- + +class TestCostEstimation: + def test_deepseek_estimate(self): + judge = LLMJudge(api_key="k", model="deepseek-chat") + cost = judge._estimate_cost(5) + # 5 pairs: ~1450 input + 250 output tokens + # Cost should be tiny (< $0.01) + assert 0.0 < cost < 0.01 + + def test_gpt4o_mini_more_expensive(self): + judge_ds = LLMJudge(api_key="k", model="deepseek-chat") + judge_gpt = LLMJudge(api_key="k", model="gpt-4o-mini") + assert judge_gpt._estimate_cost(5) > judge_ds._estimate_cost(5) + + def test_gemini_flash_cheaper(self): + judge_gemini = LLMJudge(api_key="k", model="gemini-2.5-flash-lite") + judge_gpt = LLMJudge(api_key="k", model="gpt-4o-mini") + assert judge_gemini._estimate_cost(5) < judge_gpt._estimate_cost(5) + + +# ---------- Judge batch (mocked API) ---------- + +class TestJudgeBatch: + """Tests the full judge_batch flow with a mocked OpenAI client.""" + + def setup_method(self): + self.judge = LLMJudge(api_key="test-key", timeout_seconds=5.0) + + def _mock_response_with_verdicts(self, verdicts_data): + response = MagicMock() + response.choices = [MagicMock()] + response.choices[0].message.content = json.dumps(verdicts_data) + return response + + def _run_judge_batch(self, verdicts_data): + """Helper to run judge_batch with a mocked client.""" + # Build the mock client BEFORE calling the method + response = self._mock_response_with_verdicts(verdicts_data) + mock_client = MagicMock() + mock_client.chat = MagicMock() + mock_client.chat.completions = MagicMock() + mock_client.chat.completions.create = AsyncMock(return_value=response) + self.judge._client = mock_client + return asyncio.run(self.judge.judge_batch( + [("Source 1", "Translation 1"), ("Source 2", "Translation 2")], + target_lang="fr", + target_lang_name="French", + )) + + def test_all_pass_returns_pass(self): + result = self._run_judge_batch([ + {"accurate": "yes", "fluent": "yes", "correct_language": "yes", + "no_leaks": "yes", "reason": "good"}, + {"accurate": "yes", "fluent": "yes", "correct_language": "yes", + "no_leaks": "yes", "reason": "good"}, + ]) + assert result.verdict == "pass" + assert result.chunks_passed == 2 + assert result.chunks_failed == 0 + assert result.failure_rate == 0.0 + + def test_any_fail_returns_fail(self): + result = self._run_judge_batch([ + {"accurate": "yes", "fluent": "yes", "correct_language": "yes", + "no_leaks": "yes", "reason": "good"}, + {"accurate": "no", "fluent": "yes", "correct_language": "yes", + "no_leaks": "yes", "reason": "mistranslation"}, + ]) + assert result.verdict == "fail" + assert result.chunks_failed == 1 + assert result.failure_rate == 0.5 + + def test_empty_pairs_returns_skip(self): + result = asyncio.run(self.judge.judge_batch([], "fr", "French")) + assert result.verdict == "skip" + assert result.chunks_evaluated == 0 + + def test_api_error_returns_skip(self): + # Set up a client that raises + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock( + side_effect=Exception("API down") + ) + self.judge._client = mock_client + result = asyncio.run(self.judge.judge_batch( + [("Source", "Translation")], "fr", "French" + )) + assert result.verdict == "skip" + assert "API down" in result.error + + def test_timeout_returns_skip(self): + import asyncio + mock_client = MagicMock() + async def slow_call(*args, **kwargs): + await asyncio.sleep(20) # longer than timeout + return MagicMock() + mock_client.chat.completions.create = slow_call + self.judge._client = mock_client + self.judge._timeout = 0.1 # very short timeout + result = asyncio.run(self.judge.judge_batch( + [("Source", "Translation")], "fr", "French" + )) + assert result.verdict == "skip" + assert "timeout" in result.error + + +# ---------- make_judge_from_env ---------- + +class TestMakeJudgeFromEnv: + def test_returns_none_when_no_api_key(self, monkeypatch): + monkeypatch.delenv("L1_JUDGE_API_KEY", raising=False) + assert make_judge_from_env() is None + + def test_returns_judge_when_key_set(self, monkeypatch): + monkeypatch.setenv("L1_JUDGE_API_KEY", "test-key") + monkeypatch.setenv("L1_JUDGE_MODEL", "gpt-4o-mini") + monkeypatch.setenv("L1_JUDGE_BASE_URL", "https://api.openai.com/v1") + judge = make_judge_from_env() + assert judge is not None + assert judge._model == "gpt-4o-mini" + assert judge._base_url == "https://api.openai.com/v1" + + def test_default_model(self, monkeypatch): + monkeypatch.setenv("L1_JUDGE_API_KEY", "k") + monkeypatch.delenv("L1_JUDGE_MODEL", raising=False) + judge = make_judge_from_env() + assert judge._model == "deepseek-chat" diff --git a/tests/services/quality/test_sampler.py b/tests/services/quality/test_sampler.py new file mode 100644 index 0000000..e97762c --- /dev/null +++ b/tests/services/quality/test_sampler.py @@ -0,0 +1,101 @@ +""" +Tests for services/quality/sampler.py +""" +import pytest + +from services.quality.sampler import sample_chunks_for_l1 + + +class TestSampleChunksForL1: + def test_empty_chunks(self): + result = sample_chunks_for_l1([], [], set(), max_samples=5, min_chunks=10) + assert result == [] + + def test_too_few_chunks(self): + sources = ["Hello", "World", "Goodbye"] + translations = ["Bonjour", "Monde", "Au revoir"] + result = sample_chunks_for_l1(sources, translations, set(), + max_samples=5, min_chunks=10) + # Only 3 chunks, min_chunks=10 → no sample + assert result == [] + + def test_min_chunks_exact(self): + sources = ["chunk"] * 10 + translations = ["morceau"] * 10 + result = sample_chunks_for_l1(sources, translations, set(), + max_samples=5, min_chunks=10) + # Exactly 10 chunks, min_chunks=10 → should sample + assert len(result) == 5 + + def test_skips_l0_failures(self): + sources = ["a" * 100, "b" * 200, "c" * 300, "d" * 400] + translations = ["x" * 100, "y" * 200, "z" * 300, "w" * 400] + # Mark index 0 and 2 as L0 failures + result = sample_chunks_for_l1(sources, translations, {0, 2}, + max_samples=10, min_chunks=3) + sources_returned = [s for s, t in result] + assert sources_returned == ["b" * 200, "d" * 400] + + def test_prefers_longest_chunks(self): + # 10 chunks of different lengths. Each chunk has a unique marker + # so we can verify which were picked regardless of whitespace. + sources = [f"src{i}_" + "x" * (i * 5) for i in range(10)] + translations = [f"tr{i}_" + "y" * (i * 5) for i in range(10)] + result = sample_chunks_for_l1(sources, translations, set(), + max_samples=3, min_chunks=5) + # The 3 longest should be picked (indices 9, 8, 7) + assert len(result) == 3 + # Verify the longest chunks were picked + markers_returned = [s for s, t in result] + assert any("src9_" in s for s in markers_returned) + assert any("src8_" in s for s in markers_returned) + assert any("src7_" in s for s in markers_returned) + # And the shortest were NOT picked + assert not any("src0_" in s for s in markers_returned) + assert not any("src1_" in s for s in markers_returned) + + def test_skips_identical_source_translation(self): + # Identical pairs are probably numbers, codes, brand names + sources = ["12345", "Hello", "WORLD"] + translations = ["12345", "Bonjour", "WORLD"] + result = sample_chunks_for_l1(sources, translations, set(), + max_samples=5, min_chunks=3) + # "12345" and "WORLD" pairs should be skipped (identical) + sources_returned = [s for s, t in result] + assert "12345" not in sources_returned + assert "WORLD" not in sources_returned + assert "Hello" in sources_returned + + def test_skips_very_short_pairs(self): + # Very short pairs don't have enough context + sources = ["a", "Hello world this is a longer test sentence", + "b", "Another longer sentence for testing purposes"] + translations = ["x", "Bonjour le monde ceci est une phrase plus longue", + "y", "Une autre phrase plus longue pour tester"] + result = sample_chunks_for_l1(sources, translations, set(), + max_samples=5, min_chunks=2) + # The "a"/"x" and "b"/"y" pairs should be skipped + sources_returned = [s for s, t in result] + assert "a" not in sources_returned + assert "b" not in sources_returned + assert len(result) == 2 + + def test_respects_max_samples(self): + sources = [f"long source {i} " * 10 for i in range(20)] + translations = [f"long translation {i} " * 10 for i in range(20)] + result = sample_chunks_for_l1(sources, translations, set(), + max_samples=3, min_chunks=5) + assert len(result) == 3 + + def test_results_in_document_order(self): + # The function should return in original document order, + # not in the length-priority order. (Use .strip()-friendly data + # to avoid whitespace edge cases in equality.) + sources = ["short source 1", "long source 2 " * 20, "medium source 3 " * 10, "tiny source 4"] + translations = ["court 1", "longue 2 " * 20, "moyen 3 " * 10, "minuscule 4"] + result = sample_chunks_for_l1(sources, translations, set(), + max_samples=4, min_chunks=2) + # Indices in the result should be 0, 1, 2, 3 (in document order) + for i, (s, t) in enumerate(result): + assert s == sources[i].strip(), f"Source {i} mismatch: {s!r} vs {sources[i].strip()!r}" + assert t == translations[i].strip(), f"Translation {i} mismatch"