"""LLM-powered term translation service.""" import asyncio import json import time import structlog from openai import AsyncAzureOpenAI, AzureOpenAI from src.core.config import settings from src.llm import get_async_chat_client, get_chat_client, get_chat_model_name, strip_thinking logger = structlog.get_logger(__name__) TRANSLATION_PROMPT = """Translate the following domain term to Japanese, English, and Vietnamese. Term: {term} Source language: {source_lang} Rules: - If the term is already in a target language, copy it as-is - If the term cannot be accurately translated, return null for that language - Return ONLY a JSON object with keys: ja, en, vi Output ONLY valid JSON: {{"ja": "...", "en": "...", "vi": "..."}} """ class TranslationService: """LLM-powered term translation service.""" def __init__(self, client: AzureOpenAI | None = None): self._client = client self._client_initialized = client is not None def _get_client(self): """Lazy client initialization.""" if not self._client_initialized: self._client = get_chat_client() self._client_initialized = True return self._client def translate_term( self, term: str, source_lang: str, max_retries: int = 3 ) -> tuple[dict[str, str | None], bool]: """Translate term to Japanese, English, Vietnamese. Args: term: Term to translate source_lang: Source language code (ja, en, vi) max_retries: Number of retries on failure Returns: Tuple of (translations dict, success bool) translations: {"ja": "...", "en": "...", "vi": "..."} success: True if translation succeeded, False if failed (for translation_failed column) """ for attempt in range(max_retries + 1): try: client = self._get_client() prompt = TRANSLATION_PROMPT.format(term=term, source_lang=source_lang) response = client.chat.completions.create( model=get_chat_model_name(), messages=[{"role": "user", "content": prompt}], temperature=getattr(settings, "translation_temperature", 0.0), max_tokens=getattr(settings, "translation_max_tokens", 200), ) stripped = strip_thinking(response.choices[0].message.content).strip() if not stripped: logger.warning("translation_empty_after_think_strip", term=term) return term # Return original term if LLM gives nothing result = json.loads(stripped) # Validate result is a dictionary if not isinstance(result, dict): logger.warning( "translation_invalid_response_type", term=term, expected_type="dict", actual_type=type(result).__name__, result=str(result)[:100] ) raise ValueError(f"Expected dict, got {type(result).__name__}") # Extract translations with validation return { "ja": result.get("ja"), "en": result.get("en"), "vi": result.get("vi"), }, True # Success except json.JSONDecodeError as e: # LLM returned invalid JSON - likely formatting issue if attempt < max_retries: backoff = 2**attempt logger.warning( "translation_json_parse_error", term=term, attempt=attempt + 1, error=str(e), error_type="JSONDecodeError", retry_in_seconds=backoff, likely_transient=False ) time.sleep(backoff) else: logger.error( "translation_failed_json_parse", term=term, attempts=max_retries + 1, error=str(e), error_type="JSONDecodeError", action_required="check_llm_response_format" ) return {"ja": None, "en": None, "vi": None}, False except (ConnectionError, TimeoutError) as e: # Network/timeout errors - likely transient if attempt < max_retries: backoff = 2**attempt logger.warning( "translation_network_error", term=term, attempt=attempt + 1, error=str(e), error_type=type(e).__name__, retry_in_seconds=backoff, likely_transient=True ) time.sleep(backoff) else: logger.error( "translation_failed_network", term=term, attempts=max_retries + 1, error=str(e), error_type=type(e).__name__, action_required="check_network_or_service_availability" ) return {"ja": None, "en": None, "vi": None}, False except Exception as e: # Catch-all for other errors (API auth, quota, unknown) error_type = type(e).__name__ error_str = str(e).lower() # Categorize error for better diagnostics if "auth" in error_str or "unauthorized" in error_str or "forbidden" in error_str: likely_transient = False action = "check_azure_openai_credentials" elif "quota" in error_str or "rate" in error_str or "limit" in error_str: likely_transient = True action = "check_quota_or_wait_for_rate_limit_reset" else: likely_transient = None action = "investigate_error_details" if attempt < max_retries: backoff = 2**attempt logger.warning( "translation_error", term=term, attempt=attempt + 1, error=str(e), error_type=error_type, retry_in_seconds=backoff, likely_transient=likely_transient ) time.sleep(backoff) else: logger.error( "translation_failed", term=term, attempts=max_retries + 1, error=str(e), error_type=error_type, likely_transient=likely_transient, action_required=action ) return {"ja": None, "en": None, "vi": None}, False # Defensive return (unreachable but ensures type safety) return {"ja": None, "en": None, "vi": None}, False def _get_async_client(self): """Lazy async client initialization.""" if not hasattr(self, '_async_client') or self._async_client is None: self._async_client = get_async_chat_client() return self._async_client async def translate_term_async( self, term: str, source_lang: str, max_retries: int = 3 ) -> tuple[dict[str, str | None], bool]: """Async version of translate_term.""" for attempt in range(max_retries + 1): try: client = self._get_async_client() prompt = TRANSLATION_PROMPT.format(term=term, source_lang=source_lang) response = await client.chat.completions.create( model=get_chat_model_name(), messages=[{"role": "user", "content": prompt}], temperature=getattr(settings, "translation_temperature", 0.0), max_tokens=getattr(settings, "translation_max_tokens", 200), ) content = response.choices[0].message.content or "" stripped = strip_thinking(content).strip() if not stripped: return {"ja": term, "en": term, "vi": term}, False result = json.loads(stripped) if not isinstance(result, dict): raise ValueError(f"Expected dict, got {type(result).__name__}") return { "ja": result.get("ja"), "en": result.get("en"), "vi": result.get("vi"), }, True except Exception as e: if attempt < max_retries: await asyncio.sleep(2 ** attempt) else: logger.error("translation_failed", term=term, error=str(e)) return {"ja": None, "en": None, "vi": None}, False return {"ja": None, "en": None, "vi": None}, False