From c794eff823d8c994fe1f4f946f9cfd8cd29b6446 Mon Sep 17 00:00:00 2001 From: sepehr Date: Tue, 14 Jul 2026 16:56:04 +0200 Subject: [PATCH] =?UTF-8?q?feat(quality):=20A4=20=E2=80=94=20L2=20Pro=20pr?= =?UTF-8?q?emium=20judge=20(8=20dims,=20gpt-4o,=20Pro-gated,=20opt-in)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config.py | 13 + middleware/metrics.py | 45 +++ routes/translate_routes.py | 43 +++ services/quality/l2_judge.py | 395 +++++++++++++++++++ services/quality/pipeline.py | 153 ++++++++ tests/services/quality/test_l2_judge.py | 494 ++++++++++++++++++++++++ 6 files changed, 1143 insertions(+) create mode 100644 services/quality/l2_judge.py create mode 100644 tests/services/quality/test_l2_judge.py diff --git a/config.py b/config.py index 53aa68b..c405080 100644 --- a/config.py +++ b/config.py @@ -91,6 +91,19 @@ class Config: # 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")) + # ============== Quality Layer (L2 — Pro tier) ============== + # Track A4 of the dev plan — STRONGER LLM judge (8 dimensions). + # Gated to Pro+ plans in the route. Default off everywhere. + # Cost: ~$0.005–$0.02/job (gpt-4o) or ~$0.001/job (gpt-4o-mini). + # Set QUALITY_L2_TIER_GATE=false to allow L2 for free tier too. + QUALITY_L2_ENABLED = os.getenv("QUALITY_L2_ENABLED", "false").lower() == "true" + QUALITY_L2_LOG_ONLY = os.getenv("QUALITY_L2_LOG_ONLY", "true").lower() == "true" + QUALITY_L2_SAMPLE_SIZE = int(os.getenv("QUALITY_L2_SAMPLE_SIZE", "15")) + QUALITY_L2_MIN_CHUNKS = int(os.getenv("QUALITY_L2_MIN_CHUNKS", "20")) + QUALITY_L2_TIMEOUT_SEC = float(os.getenv("QUALITY_L2_TIMEOUT_SEC", "20.0")) + # When true, only Pro+ plans can use L2. Otherwise, all plans can. + QUALITY_L2_TIER_GATE = os.getenv("QUALITY_L2_TIER_GATE", "true").lower() == "true" + # ============== API Configuration ============== API_TITLE = "Document Translation API" diff --git a/middleware/metrics.py b/middleware/metrics.py index ee36f5f..38a669f 100644 --- a/middleware/metrics.py +++ b/middleware/metrics.py @@ -70,6 +70,28 @@ quality_l1_judge_cost_usd = Histogram( buckets=(0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1), ) +# ---- L2 Pro premium judge (Track A4) ---- + +quality_l2_judge_total = Counter( + "quality_l2_judge_total", + "Total L2 (Pro premium judge, 8-dim) verdicts", + ["verdict", "model", "tier"], # verdict: pass | fail | skip | error +) + +quality_l2_judge_duration_seconds = Histogram( + "quality_l2_judge_duration_seconds", + "L2 Pro judge call duration in seconds", + ["model"], + buckets=(0.5, 1, 2, 5, 10, 20, 30, 60), +) + +quality_l2_judge_cost_usd = Histogram( + "quality_l2_judge_cost_usd", + "L2 Pro judge estimated cost in USD", + ["model"], + buckets=(0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0), +) + # ---- Retry metrics ---- translation_retry_total = Counter( @@ -138,6 +160,29 @@ def record_l1_verdict( quality_l1_judge_cost_usd.labels(model=model).observe(cost_usd) +def record_l2_verdict( + verdict: str, + model: str = "unknown", + tier: str = "pro", + duration_seconds: float = None, + cost_usd: float = None, +): + """Record an L2 Pro judge verdict. + + Args: + verdict: pass | fail | skip | error + model: model name (e.g. gpt-4o) + tier: pro | business | enterprise + duration_seconds: optional, observed in histogram + cost_usd: optional, observed in histogram + """ + quality_l2_judge_total.labels(verdict=verdict, model=model, tier=tier).inc() + if duration_seconds is not None: + quality_l2_judge_duration_seconds.labels(model=model).observe(duration_seconds) + if cost_usd is not None: + quality_l2_judge_cost_usd.labels(model=model).observe(cost_usd) + + def record_translation_retry(reason: str, tier: str = "free"): """Record a translation retry triggered by a quality issue. diff --git a/routes/translate_routes.py b/routes/translate_routes.py index 642e86a..2539bdf 100644 --- a/routes/translate_routes.py +++ b/routes/translate_routes.py @@ -1455,6 +1455,49 @@ async def _run_translation_job( except Exception: pass + # ------------------------------------------------------------------ + # Quality L2 layer (Track A4 — Pro premium judge) + # Stronger LLM (gpt-4o default), 8 dimensions, 15 samples. + # Gated to Pro+ plans (configurable via QUALITY_L2_TIER_GATE). + # Default OFF everywhere — observation first. + # Cost: ~$0.005–$0.02/job (gpt-4o), ~$0.001/job (gpt-4o-mini). + # ------------------------------------------------------------------ + if getattr(config, "QUALITY_L2_ENABLED", False) and quality_samples: + try: + from services.quality import run_l2_check + + # Tier gate: Pro+ plans only (unless gate is disabled) + user_tier = ( + _tier_for_quota(current_user.plan) if current_user else "free" + ) + tier_gate_on = getattr(config, "QUALITY_L2_TIER_GATE", True) + if not tier_gate_on or user_tier in ("pro", "business", "enterprise"): + translated_chunks_for_l2 = [s["translated"] for s in quality_samples] + l2_result = await run_l2_check( + source_chunks=[""] * len(translated_chunks_for_l2), + translated_chunks=translated_chunks_for_l2, + target_lang=target_lang, + l0_failed_indices=l0_failed_indices, + job_id=job_id, + file_extension=file_extension, + max_samples=getattr(config, "QUALITY_L2_SAMPLE_SIZE", 15), + min_chunks=getattr(config, "QUALITY_L2_MIN_CHUNKS", 20), + log_only=getattr(config, "QUALITY_L2_LOG_ONLY", True), + ) + else: + # Free/Starter user — skip L2 silently (gated) + logger.info( + "quality_l2_check_skipped", + job_id=job_id, + reason="tier_gated", + tier=user_tier, + ) + except Exception as l2_err: + # L2 must NEVER break a job. Log and continue. + logger.warning( + f"Job {job_id}: quality L2 layer failed: {l2_err}" + ) + if user_id: # Determine cost factor based on selected provider and model cost_factor = 1 diff --git a/services/quality/l2_judge.py b/services/quality/l2_judge.py new file mode 100644 index 0000000..49efeb0 --- /dev/null +++ b/services/quality/l2_judge.py @@ -0,0 +1,395 @@ +""" +L2 Pro Premium Judge — stronger model, more dimensions, Pro-tier only. + +Why a SEPARATE module from L1? + - L1 is fast + cheap (deepseek-chat, 4 dimensions, 5 samples) + - L2 is slow + expensive (gpt-4o, 8 dimensions, 15 samples) + - Different defaults, different config, different metrics + - L2 is gated to the Pro plan; L1 is universal + +L2 dimensions (8): + 1. accurate — meaning preserved + 2. fluent — natural in target language + 3. correct_lang — in the target language (not source leakage) + 4. no_leaks — no prompt artifacts + 5. terminology — domain terms correctly handled + 6. style — appropriate register (formal/informal/technical) + 7. completeness — no content dropped or added + 8. formatting — codes, numbers, units preserved + +L1 was a binary pass/fail. L2 returns per-dimension scores (0/1) so the +caller can decide which dimensions matter for a given job. + +DESIGN CONSTRAINTS (same as L1): + - 100% API-based. No local models, no GPU. + - Async, with a hard timeout. + - Defensive: never raises. + - Output is structured JSON. +""" + +from __future__ import annotations + +import asyncio +import json +import re +import time +from dataclasses import dataclass, field, asdict +from typing import List, Optional, Tuple, Dict, Any + +from core.logging import get_logger + +logger = get_logger(__name__) + + +# ---------- Result dataclasses ---------- + +@dataclass +class L2DimensionVerdict: + """8-dimension verdict for a single chunk.""" + accurate: bool = False + fluent: bool = False + correct_lang: bool = False + no_leaks: bool = False + terminology: bool = False + style: bool = False + completeness: bool = False + formatting: bool = False + reason: str = "" + + @property + def passed_count(self) -> int: + return sum([ + self.accurate, self.fluent, self.correct_lang, self.no_leaks, + self.terminology, self.style, self.completeness, self.formatting, + ]) + + @property + def total(self) -> int: + return 8 + + @property + def score(self) -> float: + return self.passed_count / self.total + + @property + def passed(self) -> bool: + # L2 is conservative: any single fail = chunk fails + return self.passed_count == self.total + + def to_log_dict(self) -> dict: + return asdict(self) + + +@dataclass +class L2Result: + """Aggregate result of an L2 check on a sample of chunks.""" + verdict: str # "pass", "fail", "skip" + chunks_evaluated: int = 0 + chunks_passed: int = 0 + chunks_failed: int = 0 + failure_rate: float = 0.0 + average_score: float = 0.0 # mean of per-chunk scores (0.0 to 1.0) + dimension_pass_rates: Dict[str, float] = field(default_factory=dict) + 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 (8 dimensions) ---------- + +L2_JUDGE_SYSTEM_PROMPT = """You are an expert translation quality evaluator using MQM-inspired criteria. + +For each (SOURCE, TRANSLATION) pair, check these 8 criteria (yes/no for each): + +1. ACCURATE — Does the translation preserve the meaning of the source? +2. FLUENT — Is the translation natural and grammatical in {target_lang_name}? +3. CORRECT_LANG — Is the translation actually in {target_lang_name} (ISO: {target_lang})? +4. NO_LEAKS — Is the translation free of prompt artifacts, source-language text, or meta-commentary? +5. TERMINOLOGY — Are domain-specific terms (technical, legal, medical, etc.) correctly translated? +6. STYLE — Is the register/tone appropriate (formal/informal/technical matching the source)? +7. COMPLETENESS — Is all content present, with nothing added or dropped? +8. FORMATTING — Are codes, numbers, dates, and units preserved exactly? + +A translation FAILS if ANY criterion is "no". The "reason" must be in English and ≤ 20 words. + +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_lang": "yes"|"no", "no_leaks": "yes"|"no", "terminology": "yes"|"no", "style": "yes"|"no", "completeness": "yes"|"no", "formatting": "yes"|"no", "reason": "short justification"}} +] +""" + + +# ---------- LLM client ---------- + +class L2ProJudge: + """ + Calls a STRONG LLM via the OpenAI-compatible API to judge translation + quality across 8 dimensions. Pro-tier only. + + Default model: gpt-4o (strongest general judge we can afford at scale). + For sub-$0.01/job cost, we limit to 15 samples per job and 8 dimensions. + """ + + def __init__( + self, + api_key: str, + base_url: str = "https://api.openai.com/v1", + model: str = "gpt-4o", + timeout_seconds: float = 20.0, + max_retries: int = 1, + ): + if not api_key: + raise ValueError("api_key is required for L2ProJudge") + self._api_key = api_key + self._base_url = base_url.rstrip("/") + self._model = model + self._timeout = timeout_seconds + self._max_retries = max_retries + 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("l2_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 = "", + ) -> L2Result: + """ + Judge a batch of (source, translation) pairs across 8 dimensions. + + Returns an L2Result with verdict="skip" on any internal error. + Never raises. + """ + start = time.time() + empty = L2Result(verdict="skip", error="not_run") + + if not pairs: + return L2Result(verdict="skip", error="empty pairs", + elapsed_ms=round((time.time() - start) * 1000, 2)) + + client = self._get_client() + if client is None: + return L2Result(verdict="skip", error="client unavailable", + elapsed_ms=round((time.time() - start) * 1000, 2)) + + 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 = L2_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 + 5.0, + ) + except asyncio.TimeoutError: + elapsed_ms = round((time.time() - start) * 1000, 2) + logger.warning("l2_judge_timeout", + timeout_s=self._timeout, elapsed_ms=elapsed_ms) + return L2Result(verdict="skip", error="timeout", elapsed_ms=elapsed_ms) + except Exception as e: + elapsed_ms = round((time.time() - start) * 1000, 2) + logger.warning("l2_judge_error", + error=str(e)[:200], elapsed_ms=elapsed_ms) + return L2Result(verdict="skip", error=str(e)[:200], + elapsed_ms=elapsed_ms) + + verdicts = self._parse_response(response, len(pairs)) + + if not verdicts: + elapsed_ms = round((time.time() - start) * 1000, 2) + return L2Result(verdict="skip", error="parse_failed", + elapsed_ms=elapsed_ms) + + # Aggregate + passed = sum(1 for v in verdicts if v.passed) + failed = len(verdicts) - passed + failure_rate = failed / len(verdicts) if verdicts else 0.0 + average_score = sum(v.score for v in verdicts) / len(verdicts) + + # Per-dimension pass rates + dimensions = [ + "accurate", "fluent", "correct_lang", "no_leaks", + "terminology", "style", "completeness", "formatting", + ] + dim_pass_rates = {} + for dim in dimensions: + count = sum(1 for v in verdicts if getattr(v, dim)) + dim_pass_rates[dim] = round(count / len(verdicts), 3) if verdicts else 0.0 + + # L2 verdict: strict — any chunk fail = overall fail + verdict = "pass" if failed == 0 else "fail" + + elapsed_ms = round((time.time() - start) * 1000, 2) + cost_estimate = self._estimate_cost(len(pairs)) + + return L2Result( + verdict=verdict, + chunks_evaluated=len(verdicts), + chunks_passed=passed, + chunks_failed=failed, + failure_rate=round(failure_rate, 3), + average_score=round(average_score, 3), + dimension_pass_rates=dim_pass_rates, + 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, + max_tokens=1200, # larger than L1 (more dimensions) + response_format={"type": "json_object"}, + ) + return response + except Exception as e: + last_exc = e + if attempt < self._max_retries: + await asyncio.sleep(0.8) + raise last_exc + + def _parse_response(self, response, expected_count: int) -> List[L2DimensionVerdict]: + """Parse the LLM response into a list of 8-dimension verdicts.""" + try: + content = response.choices[0].message.content or "" + except (AttributeError, IndexError) as e: + logger.warning("l2_judge_bad_response", error=str(e)) + return [] + + content = content.strip() + if content.startswith("```"): + content = re.sub(r"^```(?:json)?\s*\n?", "", content) + content = re.sub(r"\n?```\s*$", "", content) + + try: + data = json.loads(content) + except json.JSONDecodeError as e: + logger.warning("l2_judge_json_parse_error", + error=str(e), content_preview=content[:200]) + return [] + + if isinstance(data, dict): + 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: + for v in data.values(): + if isinstance(v, list): + items = v + break + if items is None: + logger.warning("l2_judge_no_list_in_response") + return [] + elif isinstance(data, list): + items = data + else: + logger.warning("l2_judge_unexpected_response_type", + type_=type(data).__name__) + return [] + + verdicts: List[L2DimensionVerdict] = [] + for item in items: + try: + v = L2DimensionVerdict( + accurate=str(item.get("accurate", "")).lower() == "yes", + fluent=str(item.get("fluent", "")).lower() == "yes", + correct_lang=str(item.get("correct_lang", "")).lower() == "yes", + no_leaks=str(item.get("no_leaks", "")).lower() == "yes", + terminology=str(item.get("terminology", "")).lower() == "yes", + style=str(item.get("style", "")).lower() == "yes", + completeness=str(item.get("completeness", "")).lower() == "yes", + formatting=str(item.get("formatting", "")).lower() == "yes", + reason=str(item.get("reason", ""))[:300], + ) + verdicts.append(v) + except Exception as e: + logger.warning("l2_judge_item_parse_error", + error=str(e), item=str(item)[:200]) + + return verdicts + + def _estimate_cost(self, num_pairs: int) -> float: + """Rough USD cost estimate for the call.""" + # L2 has more dimensions = longer output + input_tokens = 250 + (num_pairs * 280) + output_tokens = num_pairs * 110 + model_lower = self._model.lower() + # IMPORTANT: check 'mini' BEFORE full 'gpt-4o' because + # 'gpt-4o-mini' contains 'gpt-4o'. + if "gpt-4o-mini" in model_lower: + input_cost = input_tokens / 1_000_000 * 0.15 + output_cost = output_tokens / 1_000_000 * 0.60 + elif "gpt-4o" in model_lower: + input_cost = input_tokens / 1_000_000 * 2.50 + output_cost = output_tokens / 1_000_000 * 10.00 + elif "claude" in model_lower: + input_cost = input_tokens / 1_000_000 * 3.00 + output_cost = output_tokens / 1_000_000 * 15.00 + else: + # Generic conservative + input_cost = input_tokens / 1_000_000 * 1.00 + output_cost = output_tokens / 1_000_000 * 3.00 + return round(input_cost + output_cost, 6) + + +# ---------- Convenience factory ---------- + +def make_l2_judge_from_env() -> Optional[L2ProJudge]: + """ + Build an L2ProJudge from environment variables. Returns None if + no API key is configured. + + Reads: + - L2_JUDGE_API_KEY (required) + - L2_JUDGE_BASE_URL (default: OpenAI) + - L2_JUDGE_MODEL (default: gpt-4o) + - L2_JUDGE_TIMEOUT (default: 20.0) + """ + import os + api_key = os.getenv("L2_JUDGE_API_KEY", "").strip() + if not api_key: + return None + return L2ProJudge( + api_key=api_key, + base_url=os.getenv("L2_JUDGE_BASE_URL", "https://api.openai.com/v1"), + model=os.getenv("L2_JUDGE_MODEL", "gpt-4o"), + timeout_seconds=float(os.getenv("L2_JUDGE_TIMEOUT", "20.0")), + ) diff --git a/services/quality/pipeline.py b/services/quality/pipeline.py index 83a3eb6..9823c24 100644 --- a/services/quality/pipeline.py +++ b/services/quality/pipeline.py @@ -23,6 +23,7 @@ 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 +from .l2_judge import L2Result, L2ProJudge logger = get_logger(__name__) @@ -271,3 +272,155 @@ def make_judge_from_env_safe() -> Optional[LLMJudge]: except Exception as e: logger.warning("l1_judge_init_failed", error=str(e)[:200]) return None + + +# ---------- L2 (Pro tier) ---------- + +async def run_l2_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 = 15, + min_chunks: int = 20, + judge: Optional[L2ProJudge] = None, + log_only: bool = True, +) -> L2Result: + """ + Run the L2 Pro premium judge (8 dimensions, gpt-4o default). + + 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 L2ProJudge 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 L2Result. verdict="skip" on any internal error. + Never raises — defensive wrapper. + """ + skip = L2Result(verdict="skip", error="not_run") + + if l0_failed_indices is None: + l0_failed_indices = set() + + # Sample (reuse the L1 sampler — it's just chunk selection, model-agnostic) + 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_l2_check_skipped", + job_id=job_id, + reason="insufficient_chunks_or_all_flagged", + chunk_count=len(source_chunks), + ) + _record_l2_metric(verdict="skip", model="none") + return skip + + # Get the judge + if judge is None: + judge = make_l2_judge_from_env_safe() + + if judge is None: + logger.info( + "quality_l2_check_skipped", + job_id=job_id, + reason="no_l2_judge_configured", + ) + _record_l2_metric(verdict="skip", model="none") + 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: + logger.warning( + "quality_l2_check_failed", + job_id=job_id, + error=str(e)[:200], + error_type=type(e).__name__, + ) + _record_l2_metric(verdict="error", model="unknown") + return L2Result(verdict="skip", error=str(e)[:200]) + + # Log (always) — caller decides what to do + logger.info( + "quality_l2_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, + average_score=result.average_score, + dimension_pass_rates=result.dimension_pass_rates, + model=result.model_used, + elapsed_ms=result.elapsed_ms, + cost_estimate_usd=result.cost_estimate_usd, + log_only=log_only, + ) + + # Record Prometheus metric + duration_s = None + if result.elapsed_ms is not None: + try: + duration_s = float(result.elapsed_ms) / 1000.0 + except Exception: + duration_s = None + _record_l2_metric( + verdict=result.verdict or "skip", + model=result.model_used or "unknown", + duration_seconds=duration_s, + cost_usd=result.cost_estimate_usd, + ) + return result + + +def _record_l2_metric( + verdict: str, + model: str = "unknown", + duration_seconds: float = None, + cost_usd: float = None, +) -> None: + """Best-effort Prometheus metric emission for L2. + + Never raises. + """ + try: + from middleware.metrics import record_l2_verdict + record_l2_verdict( + verdict=verdict, + model=model, + duration_seconds=duration_seconds, + cost_usd=cost_usd, + ) + except Exception: + pass + + +def make_l2_judge_from_env_safe() -> Optional[L2ProJudge]: + """Read env vars and build an L2 judge, or return None if not configured. + + Defensive wrapper — a misconfigured L2 environment NEVER breaks a job. + """ + try: + from .l2_judge import make_l2_judge_from_env + return make_l2_judge_from_env() + except Exception as e: + logger.warning("l2_judge_init_failed", error=str(e)[:200]) + return None diff --git a/tests/services/quality/test_l2_judge.py b/tests/services/quality/test_l2_judge.py new file mode 100644 index 0000000..ef7df7b --- /dev/null +++ b/tests/services/quality/test_l2_judge.py @@ -0,0 +1,494 @@ +""" +Tests for Track A4 — L2 Pro premium judge. + +Covers: + - L2DimensionVerdict: 8 dimensions + scoring + - L2Result: aggregation + dimension pass rates + - L2ProJudge: construction, missing api_key, cost estimation + - make_l2_judge_from_env: env-var-driven factory + - run_l2_check: defensive wrapper +""" +import os +import json +import pytest +from unittest.mock import MagicMock, AsyncMock, patch + +from services.quality.l2_judge import ( + L2DimensionVerdict, + L2Result, + L2ProJudge, + make_l2_judge_from_env, + L2_JUDGE_SYSTEM_PROMPT, +) + + +# ============================================================================ +# L2DimensionVerdict +# ============================================================================ + +class TestL2DimensionVerdict: + def test_all_pass(self): + v = L2DimensionVerdict( + accurate=True, fluent=True, correct_lang=True, no_leaks=True, + terminology=True, style=True, completeness=True, formatting=True, + ) + assert v.passed_count == 8 + assert v.total == 8 + assert v.score == 1.0 + assert v.passed is True + + def test_one_fails(self): + v = L2DimensionVerdict( + accurate=True, fluent=True, correct_lang=True, no_leaks=True, + terminology=False, # <-- fails + style=True, completeness=True, formatting=True, + ) + assert v.passed_count == 7 + assert v.total == 8 + assert v.score == pytest.approx(0.875) + # L2 is strict: one fail = chunk fails + assert v.passed is False + + def test_all_fail(self): + v = L2DimensionVerdict() # all default False + assert v.passed_count == 0 + assert v.score == 0.0 + assert v.passed is False + + def test_default_construction(self): + v = L2DimensionVerdict() + assert v.accurate is False + assert v.reason == "" + + +# ============================================================================ +# L2Result +# ============================================================================ + +class TestL2Result: + def test_default_construction(self): + r = L2Result(verdict="pass") + assert r.verdict == "pass" + assert r.chunks_evaluated == 0 + assert r.dimension_pass_rates == {} + assert r.error == "" + + def test_to_log_dict(self): + r = L2Result( + verdict="pass", + chunks_evaluated=10, + chunks_passed=8, + chunks_failed=2, + failure_rate=0.2, + average_score=0.85, + model_used="gpt-4o", + cost_estimate_usd=0.012, + ) + d = r.to_log_dict() + assert d["verdict"] == "pass" + assert d["chunks_evaluated"] == 10 + assert d["model_used"] == "gpt-4o" + assert d["cost_estimate_usd"] == 0.012 + + +# ============================================================================ +# L2ProJudge construction +# ============================================================================ + +class TestL2ProJudgeConstruction: + def test_requires_api_key(self): + with pytest.raises(ValueError, match="api_key is required"): + L2ProJudge(api_key="") + + def test_basic_construction(self): + judge = L2ProJudge(api_key="sk-test") + assert judge._api_key == "sk-test" + assert judge._model == "gpt-4o" + assert judge._base_url == "https://api.openai.com/v1" + + def test_custom_model(self): + judge = L2ProJudge( + api_key="sk-test", + model="gpt-4o-mini", + base_url="https://api.openai.com/v1", + ) + assert judge._model == "gpt-4o-mini" + + def test_strips_trailing_slash_from_base_url(self): + judge = L2ProJudge( + api_key="sk-test", + base_url="https://api.example.com/v1/", + ) + assert judge._base_url == "https://api.example.com/v1" + + +# ============================================================================ +# Cost estimation +# ============================================================================ + +class TestL2CostEstimation: + def test_gpt4o_cost(self): + judge = L2ProJudge(api_key="sk", model="gpt-4o") + cost = judge._estimate_cost(15) # 15 samples default + # Should be in the $0.01–$0.05 range for 15 chunks + assert 0.001 < cost < 0.10 + + def test_gpt4o_mini_cheaper(self): + judge_mini = L2ProJudge(api_key="sk", model="gpt-4o-mini") + judge_full = L2ProJudge(api_key="sk", model="gpt-4o") + cost_mini = judge_mini._estimate_cost(15) + cost_full = judge_full._estimate_cost(15) + # gpt-4o-mini should be cheaper than gpt-4o + assert cost_mini < cost_full + + def test_zero_pairs(self): + judge = L2ProJudge(api_key="sk", model="gpt-4o") + cost = judge._estimate_cost(0) + # Even with 0 pairs, the system prompt has some cost + assert cost >= 0 + + +# ============================================================================ +# judge_batch — defensive +# ============================================================================ + +class TestL2JudgeBatch: + @pytest.mark.asyncio + async def test_empty_pairs_skips(self): + judge = L2ProJudge(api_key="sk") + result = await judge.judge_batch([], "fr", "French") + assert result.verdict == "skip" + assert result.error == "empty pairs" + + @pytest.mark.asyncio + async def test_client_unavailable_skips(self): + judge = L2ProJudge(api_key="sk") + # Simulate a client init failure + with patch.object(judge, "_get_client", return_value=None): + result = await judge.judge_batch( + [("Hello", "Bonjour")], "fr", "French" + ) + assert result.verdict == "skip" + assert "unavailable" in result.error or "client" in result.error.lower() + + @pytest.mark.asyncio + async def test_successful_judgement(self): + """A mock client that returns a well-formed JSON response should + produce a passing L2Result.""" + judge = L2ProJudge(api_key="sk", model="gpt-4o") + + # Mock the client + mock_response = MagicMock() + mock_response.choices = [MagicMock()] + mock_response.choices[0].message.content = json.dumps([ + { + "accurate": "yes", "fluent": "yes", "correct_lang": "yes", + "no_leaks": "yes", "terminology": "yes", "style": "yes", + "completeness": "yes", "formatting": "yes", + "reason": "Perfect translation", + } + ]) + + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock(return_value=mock_response) + + with patch.object(judge, "_get_client", return_value=mock_client): + result = await judge.judge_batch( + [("Hello", "Bonjour")], "fr", "French" + ) + + assert result.verdict == "pass" + assert result.chunks_evaluated == 1 + assert result.chunks_passed == 1 + assert result.chunks_failed == 0 + assert result.failure_rate == 0.0 + assert result.average_score == 1.0 + # All 8 dimensions should have pass rate 1.0 + for dim in ["accurate", "fluent", "terminology", "style"]: + assert result.dimension_pass_rates[dim] == 1.0 + + @pytest.mark.asyncio + async def test_partial_failure(self): + """If one of 8 dimensions fails, the chunk should fail.""" + judge = L2ProJudge(api_key="sk", model="gpt-4o") + + mock_response = MagicMock() + mock_response.choices = [MagicMock()] + # fluent = no, others yes + mock_response.choices[0].message.content = json.dumps([ + { + "accurate": "yes", "fluent": "no", "correct_lang": "yes", + "no_leaks": "yes", "terminology": "yes", "style": "yes", + "completeness": "yes", "formatting": "yes", + "reason": "Awkward phrasing", + } + ]) + + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock(return_value=mock_response) + + with patch.object(judge, "_get_client", return_value=mock_client): + result = await judge.judge_batch( + [("Hello", "Bonjour")], "fr", "French" + ) + + # L2 is strict: one fail = overall fail + assert result.verdict == "fail" + assert result.chunks_evaluated == 1 + assert result.chunks_passed == 0 + assert result.chunks_failed == 1 + assert result.dimension_pass_rates["fluent"] == 0.0 + assert result.dimension_pass_rates["accurate"] == 1.0 + + @pytest.mark.asyncio + async def test_handles_markdown_fences(self): + """The judge should strip markdown code fences from responses.""" + judge = L2ProJudge(api_key="sk", model="gpt-4o") + + mock_response = MagicMock() + mock_response.choices = [MagicMock()] + mock_response.choices[0].message.content = ( + "```json\n" + + json.dumps([{ + "accurate": "yes", "fluent": "yes", "correct_lang": "yes", + "no_leaks": "yes", "terminology": "yes", "style": "yes", + "completeness": "yes", "formatting": "yes", + "reason": "ok", + }]) + + "\n```" + ) + + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock(return_value=mock_response) + + with patch.object(judge, "_get_client", return_value=mock_client): + result = await judge.judge_batch( + [("Hello", "Bonjour")], "fr", "French" + ) + + assert result.verdict == "pass" + assert result.chunks_evaluated == 1 + + @pytest.mark.asyncio + async def test_handles_dict_with_list(self): + """Some LLMs return {"verdicts": [...]} instead of a raw list.""" + judge = L2ProJudge(api_key="sk", model="gpt-4o") + + mock_response = MagicMock() + mock_response.choices = [MagicMock()] + mock_response.choices[0].message.content = json.dumps({ + "verdicts": [ + { + "accurate": "yes", "fluent": "yes", "correct_lang": "yes", + "no_leaks": "yes", "terminology": "yes", "style": "yes", + "completeness": "yes", "formatting": "yes", + "reason": "good", + } + ] + }) + + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock(return_value=mock_response) + + with patch.object(judge, "_get_client", return_value=mock_client): + result = await judge.judge_batch( + [("Hello", "Bonjour")], "fr", "French" + ) + + assert result.verdict == "pass" + + @pytest.mark.asyncio + async def test_timeout_returns_skip(self): + import asyncio + judge = L2ProJudge(api_key="sk", model="gpt-4o", timeout_seconds=0.1) + + # Mock client that raises TimeoutError + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock( + side_effect=asyncio.TimeoutError() + ) + + with patch.object(judge, "_get_client", return_value=mock_client): + result = await judge.judge_batch( + [("Hello", "Bonjour")], "fr", "French" + ) + + assert result.verdict == "skip" + assert "timeout" in result.error.lower() + + @pytest.mark.asyncio + async def test_never_raises(self): + """Even on unexpected error, the judge should return a skip, not raise.""" + judge = L2ProJudge(api_key="sk", model="gpt-4o") + + # Mock client that raises a generic exception + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock( + side_effect=RuntimeError("something unexpected") + ) + + with patch.object(judge, "_get_client", return_value=mock_client): + # Should NOT raise + result = await judge.judge_batch( + [("Hello", "Bonjour")], "fr", "French" + ) + assert result.verdict == "skip" + + @pytest.mark.asyncio + async def test_dimension_pass_rates_aggregated(self): + """Per-dimension pass rates should aggregate across chunks.""" + judge = L2ProJudge(api_key="sk", model="gpt-4o") + + mock_response = MagicMock() + mock_response.choices = [MagicMock()] + # 2 chunks: chunk 1 all-pass, chunk 2 fluent-fail + mock_response.choices[0].message.content = json.dumps([ + { + "accurate": "yes", "fluent": "yes", "correct_lang": "yes", + "no_leaks": "yes", "terminology": "yes", "style": "yes", + "completeness": "yes", "formatting": "yes", + "reason": "ok", + }, + { + "accurate": "yes", "fluent": "no", "correct_lang": "yes", + "no_leaks": "yes", "terminology": "yes", "style": "yes", + "completeness": "yes", "formatting": "yes", + "reason": "awkward", + }, + ]) + + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock(return_value=mock_response) + + with patch.object(judge, "_get_client", return_value=mock_client): + result = await judge.judge_batch( + [("Hello", "Bonjour"), ("Goodbye", "Au revoir")], + "fr", "French" + ) + + assert result.chunks_evaluated == 2 + # fluent: 1/2 = 0.5 + assert result.dimension_pass_rates["fluent"] == 0.5 + # accurate: 2/2 = 1.0 + assert result.dimension_pass_rates["accurate"] == 1.0 + + +# ============================================================================ +# Factory +# ============================================================================ + +class TestL2JudgeFactory: + def test_no_api_key_returns_none(self, monkeypatch): + monkeypatch.delenv("L2_JUDGE_API_KEY", raising=False) + assert make_l2_judge_from_env() is None + + def test_api_key_creates_judge(self, monkeypatch): + monkeypatch.setenv("L2_JUDGE_API_KEY", "sk-test") + monkeypatch.setenv("L2_JUDGE_MODEL", "gpt-4o") + monkeypatch.setenv("L2_JUDGE_BASE_URL", "https://api.openai.com/v1") + judge = make_l2_judge_from_env() + assert judge is not None + assert judge._api_key == "sk-test" + assert judge._model == "gpt-4o" + + def test_api_key_with_default_model(self, monkeypatch): + monkeypatch.setenv("L2_JUDGE_API_KEY", "sk-test") + # Clear other vars to test defaults + monkeypatch.delenv("L2_JUDGE_MODEL", raising=False) + monkeypatch.delenv("L2_JUDGE_BASE_URL", raising=False) + judge = make_l2_judge_from_env() + assert judge._model == "gpt-4o" + assert judge._base_url == "https://api.openai.com/v1" + + +# ============================================================================ +# Pipeline integration +# ============================================================================ + +class TestL2PipelineIntegration: + @pytest.mark.asyncio + async def test_run_l2_check_no_judge_skips(self): + from services.quality.pipeline import run_l2_check + + result = await run_l2_check( + source_chunks=["Hello"] * 25, + translated_chunks=["Bonjour"] * 25, + target_lang="fr", + file_extension="docx", + judge=None, # Will try to load from env, but env has no key + ) + # Should return skip, not raise + assert result.verdict == "skip" + + @pytest.mark.asyncio + async def test_run_l2_check_with_mock_judge(self): + from services.quality.pipeline import run_l2_check + + # Build a judge that always returns pass + judge = L2ProJudge(api_key="sk") + + mock_response = MagicMock() + mock_response.choices = [MagicMock()] + mock_response.choices[0].message.content = json.dumps([ + { + "accurate": "yes", "fluent": "yes", "correct_lang": "yes", + "no_leaks": "yes", "terminology": "yes", "style": "yes", + "completeness": "yes", "formatting": "yes", + "reason": "ok", + } + ] * 5) + + mock_client = MagicMock() + mock_client.chat.completions.create = AsyncMock(return_value=mock_response) + judge._client = mock_client + + # 25 chunks > default min_chunks of 20 + result = await run_l2_check( + source_chunks=["Hello world"] * 25, + translated_chunks=["Bonjour le monde"] * 25, + target_lang="fr", + file_extension="docx", + judge=judge, + max_samples=5, + ) + + assert result.verdict == "pass" + assert result.chunks_evaluated >= 1 + assert result.chunks_passed >= 1 + + @pytest.mark.asyncio + async def test_run_l2_check_too_few_chunks_skips(self): + from services.quality.pipeline import run_l2_check + + # 5 chunks < default min_chunks of 20 + result = await run_l2_check( + source_chunks=["a"] * 5, + translated_chunks=["b"] * 5, + target_lang="fr", + min_chunks=20, + ) + # Should skip due to insufficient chunks + assert result.verdict == "skip" + + +# ============================================================================ +# 8-dimension coverage +# ============================================================================ + +class TestL28DimensionCoverage: + """Sanity check: the L2 prompt template actually mentions all 8 dimensions.""" + + def test_prompt_has_all_8_dimensions(self): + for dim in [ + "ACCURATE", "FLUENT", "CORRECT_LANG", "NO_LEAKS", + "TERMINOLOGY", "STYLE", "COMPLETENESS", "FORMATTING", + ]: + assert dim in L2_JUDGE_SYSTEM_PROMPT, ( + f"Dimension {dim!r} missing from L2 prompt template" + ) + + def test_prompt_has_format_hint(self): + # Should tell the model to respond with a JSON array + assert "JSON" in L2_JUDGE_SYSTEM_PROMPT + assert "yes" in L2_JUDGE_SYSTEM_PROMPT + assert "no" in L2_JUDGE_SYSTEM_PROMPT