翻译重构

tangwang
1 parent a0a173ae
Showing 21 changed files with 832 additions and 2263 deletions Show diff stats
api/routes/search.py
api/translator_app.py
config/__init__.py
config/config.yaml
config/config_loader.py
config/services_config.py
config/translate_prompts.py
docs/搜索API对接指南.md
indexer/document_transformer.py
indexer/incremental_service.py
indexer/indexing_utils.py
indexer/test_indexing.py
providers/translation.py
query/deepl_provider.py
query/llm_translate.py
query/query_parser.py
query/qwen_mt_translate.py
query/test_translation.py
query/translator.py
services.translation.providers.llm
@@ -472,7 +472,6 @@ async def get_es_raw_document(spu_id: str, http_request: Request):
         index_name = get_tenant_index_name(tenant_id)
         body = {
-            "size": 5,
             "query": {
                 "bool": {
                     "filter": [
@@ -98,7 +98,9 @@ from pydantic import BaseModel, Field
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 from query.qwen_mt_translate import Translator
-from config.env_config import DEEPL_AUTH_KEY, DASHSCOPE_API_KEY, REDIS_CONFIG
+from query.llm_translate import LLMTranslatorProvider
+from query.deepl_provider import DeepLProvider
+from config.services_config import get_translation_config
 # Configure logging
 logging.basicConfig(
@@ -107,23 +109,52 @@ logging.basicConfig(
 )
 logger = logging.getLogger(__name__)
-# Fixed translation prompt
-TRANSLATION_PROMPT = "Translate the original text into an English product SKU name. Requirements: Ensure accurate and complete transmission of the original information, with concise, clear, authentic, and professional language."
-
 # Global translator instances cache (keyed by model)
-_translators: Dict[str, Translator] = {}
+_translators: Dict[str, object] = {}
+
+def _resolve_default_model() -> str:
+    """
+    Resolve translator model from services.translation config first.
-def get_translator(model: str = "qwen") -> Translator:
+    Priority:
+    1) TRANSLATION_MODEL env (explicit runtime override)
+    2) services.translation.provider + providers.<provider>.model
+    3) qwen-mt
+    """
+    env_model = (os.getenv("TRANSLATION_MODEL") or "").strip()
+    if env_model:
+        return env_model
+    try:
+        cfg = get_translation_config()
+        provider = (cfg.provider or "").strip().lower()
+        provider_cfg = cfg.get_provider_cfg() if hasattr(cfg, "get_provider_cfg") else {}
+        model = (provider_cfg.get("model") or "").strip().lower() if isinstance(provider_cfg, dict) else ""
+        if provider == "llm":
+            return "llm"
+        if provider in {"qwen-mt", "direct", "http"}:
+            return model or "qwen-mt"
+        if provider == "deepl":
+            return "deepl"
+    except Exception:
+        pass
+    return "qwen-mt"
+
+
+def get_translator(model: str = "qwen") -> object:
     """Get or create translator instance for the specified model."""
     global _translators
     if model not in _translators:
         logger.info(f"Initializing translator with model: {model}...")
-        _translators[model] = Translator(
-            model=model,
-            use_cache=True,
-            timeout=10
-        )
+        normalized = (model or "qwen").strip().lower()
+        if normalized in {"qwen", "qwen-mt", "qwen-mt-flash", "qwen-mt-flush"}:
+            _translators[model] = Translator(model=normalized, use_cache=True, timeout=10)
+        elif normalized == "deepl":
+            _translators[model] = DeepLProvider(api_key=None, timeout=10.0)
+        elif normalized == "llm":
+            _translators[model] = LLMTranslatorProvider()
+        else:
+            raise ValueError(f"Unsupported model: {model}")
         logger.info(f"Translator initialized with model: {model}")
     return _translators[model]
@@ -134,7 +165,9 @@ class TranslationRequest(BaseModel):
     text: str = Field(..., description="Text to translate")
     target_lang: str = Field(..., description="Target language code (zh, en, ru, etc.)")
     source_lang: Optional[str] = Field(None, description="Source language code (optional, auto-detect if not provided)")
-    model: Optional[str] = Field("qwen", description="Translation model: 'qwen' (default) or 'deepl'")
+    model: Optional[str] = Field(None, description="Translation model: qwen-mt | deepl | llm")
+    context: Optional[str] = Field(None, description="Optional translation scene or context")
+    prompt: Optional[str] = Field(None, description="Optional prompt override")
     class Config:
         json_schema_extra = {
@@ -142,7 +175,8 @@ class TranslationRequest(BaseModel):
                 "text": "商品名称",
                 "target_lang": "en",
                 "source_lang": "zh",
-                "model": "qwen"
+                "model": "llm",
+                "context": "sku_name"
             }
         }
@@ -180,8 +214,7 @@ app.add_middleware(
 async def startup_event():
     """Initialize translator on startup."""
     logger.info("Starting Translation Service API on port 6006")
-    # Get default model from environment variable or use 'qwen'
-    default_model = os.getenv("TRANSLATION_MODEL", "qwen")
+    default_model = _resolve_default_model()
     try:
         get_translator(model=default_model)
         logger.info(f"Translation service ready with default model: {default_model}")
@@ -194,15 +227,17 @@ async def startup_event():
 async def health_check():
     """Health check endpoint."""
     try:
-        default_model = os.getenv("TRANSLATION_MODEL", "qwen")
-        translator = get_translator(model=default_model)
+        # 仅做轻量级本地检查，避免在健康检查中触发潜在的阻塞初始化或外部依赖
+        default_model = _resolve_default_model()
+        # 如果启动事件成功，默认模型通常会已经初始化到缓存中
+        translator = _translators.get(default_model) or next(iter(_translators.values()), None)
         return {
             "status": "healthy",
             "service": "translation",
             "default_model": default_model,
             "available_models": list(_translators.keys()),
             "translator_initialized": translator is not None,
-            "cache_enabled": translator.use_cache if translator else False
+            "cache_enabled": bool(getattr(translator, "use_cache", False))
         }
     except Exception as e:
         logger.error(f"Health check failed: {e}")
@@ -238,11 +273,11 @@ async def translate(request: TranslationRequest):
         )
     # Validate model parameter
-    model = request.model.lower() if request.model else "qwen"
-    if model not in ['qwen', 'deepl']:
+    model = request.model.lower() if request.model else _resolve_default_model().lower()
+    if model not in ["qwen", "qwen-mt", "deepl", "llm"]:
         raise HTTPException(
             status_code=400,
-            detail=f"Invalid model: {model}. Supported models: 'qwen', 'deepl'"
+            detail="Invalid model. Supported models: 'qwen-mt', 'deepl', 'llm'"
         )
     try:
@@ -254,7 +289,8 @@ async def translate(request: TranslationRequest):
             text=request.text,
             target_lang=request.target_lang,
             source_lang=request.source_lang,
-            prompt=TRANSLATION_PROMPT
+            context=request.context,
+            prompt=request.prompt,
         )
         if translated_text is None:
@@ -269,7 +305,7 @@ async def translate(request: TranslationRequest):
             source_lang=request.source_lang,
             translated_text=translated_text,
             status="success",
-            model=translator.model
+            model=str(getattr(translator, "model", model))
         )
     except HTTPException:
@@ -28,6 +28,7 @@ from .services_config import (
     get_translation_base_url,
     get_embedding_base_url,
     get_rerank_service_url,
+    get_translation_cache_config,
     ServiceConfig,
 )
@@ -53,5 +54,6 @@ __all__ = [
     'get_translation_base_url',
     'get_embedding_base_url',
     'get_rerank_service_url',
+    'get_translation_cache_config',
     'ServiceConfig',
 ]
@@ -81,18 +81,6 @@ query_config:
   translation_service: "deepl"
   translation_api_key: null  # 通过环境变量设置
-  # 翻译提示词配置（用于提高翻译质量，作为DeepL API的context参数）
-  translation_prompts:
-    # 商品标题翻译提示词
-    product_title_zh: "请将原文翻译成中文商品SKU名称，要求：确保精确、完整地传达原文信息的基础上，语言简洁清晰、地道、专业。"
-    product_title_en: "Translate the original text into an English product SKU name. Requirements: Ensure accurate and complete transmission of the original information, with concise, clear, authentic, and professional language."
-    # query翻译提示词
-    query_zh: "电商领域"
-    query_en: "e-commerce domain"
-    # 默认翻译用词
-    default_zh: "电商领域"
-    default_en: "e-commerce domain"
-  
   # 返回字段配置（_source includes）
   # null表示返回所有字段，[]表示不返回任何字段，列表表示只返回指定字段
   source_fields: null
@@ -119,16 +107,24 @@ rerank:
 # 可扩展服务/provider 注册表（单一配置源）
 services:
   translation:
-    provider: "llm"  # direct | http | google(reserved)
+    provider: "llm"  # qwen-mt | deepl | http | llm
     base_url: "http://127.0.0.1:6006"
-    model: "qwen"
+    model: "qwen-flash"
     timeout_sec: 10.0
+    cache:
+      enabled: true
+      key_prefix: "trans:v2"
+      ttl_seconds: 62208000
+      sliding_expiration: true
+      key_include_context: true
+      key_include_prompt: true
+      key_include_source_lang: true
     providers:
-      direct:
-        model: "qwen"
+      qwen-mt:
+        model: "qwen-mt-flush"
       http:
         base_url: "http://127.0.0.1:6006"
-        model: "qwen"
+        model: "qwen-mt-flush"
         timeout_sec: 10.0
       llm:
         model: "qwen-flash"
@@ -136,6 +132,11 @@ services:
         # base_url 留空则使用 DASHSCOPE_BASE_URL 或默认地域
         base_url: ""
         timeout_sec: 30.0
+      deepl:
+        model: "deepl"
+        timeout_sec: 10.0
+        # 可选：用于术语表翻译（由 query_config.translation_glossary_id 衔接）
+        glossary_id: ""
       google:
         enabled: false
         project_id: ""
@@ -42,7 +42,6 @@ class QueryConfig:
     translation_api_key: Optional[str] = None
     translation_glossary_id: Optional[str] = None
     translation_context: str = "e-commerce product search"
-    translation_prompts: Dict[str, str] = field(default_factory=dict)
     # Embedding field names
     text_embedding_field: Optional[str] = "title_embedding"
@@ -250,7 +249,6 @@ class ConfigLoader:
             translation_service=query_config_data.get("translation_service") or "deepl",
             translation_glossary_id=query_config_data.get("translation_glossary_id"),
             translation_context=query_config_data.get("translation_context") or "e-commerce product search",
-            translation_prompts=query_config_data.get("translation_prompts", {}),
             text_embedding_field=query_config_data.get("text_embedding_field"),
             image_embedding_field=query_config_data.get("image_embedding_field"),
             source_fields=query_config_data.get("source_fields"),
@@ -72,12 +72,12 @@ def _resolve_translation() -&gt; ServiceConfig:
         config_provider=cfg.get("provider"),
         capability="translation",
     )
-    if provider not in ("direct", "local", "inprocess", "http", "service"):
+    if provider not in ("qwen-mt", "deepl", "direct", "local", "inprocess", "http", "service", "llm"):
         raise ValueError(f"Unsupported translation provider: {provider}")
     # Env override for http base_url
     env_url = os.getenv("TRANSLATION_SERVICE_URL")
-    if env_url and provider == "http":
+    if env_url and provider in ("http", "service"):
         providers = dict(providers)
         providers["http"] = dict(providers.get("http", {}))
         providers["http"]["base_url"] = env_url.rstrip("/")
@@ -206,6 +206,27 @@ def get_translation_base_url() -&gt; str:
     return str(base).rstrip("/")
+def get_translation_cache_config() -> Dict[str, Any]:
+    """
+    Resolve translation cache policy from services.translation.cache.
+
+    All translation cache key/TTL behavior should be configured in config.yaml,
+    not hardcoded in code.
+    """
+    raw = _load_services_raw()
+    cfg = raw.get("translation", {}) if isinstance(raw.get("translation"), dict) else {}
+    cache_cfg = cfg.get("cache", {}) if isinstance(cfg.get("cache"), dict) else {}
+    return {
+        "enabled": bool(cache_cfg.get("enabled", True)),
+        "key_prefix": str(cache_cfg.get("key_prefix", "trans:v2")),
+        "ttl_seconds": int(cache_cfg.get("ttl_seconds", 360 * 24 * 3600)),
+        "sliding_expiration": bool(cache_cfg.get("sliding_expiration", True)),
+        "key_include_context": bool(cache_cfg.get("key_include_context", True)),
+        "key_include_prompt": bool(cache_cfg.get("key_include_prompt", True)),
+        "key_include_source_lang": bool(cache_cfg.get("key_include_source_lang", True)),
+    }
+
+
 def get_embedding_base_url() -> str:
     """Resolve embedding HTTP base URL."""
     base = (
@@ -0,0 +1,82 @@
+SOURCE_LANG_CODE_MAP = {
+    "en": "English",
+    "zh": "Chinese",
+    "zh_tw": "Traditional Chinese",
+    "ru": "Russian",
+    "ja": "Japanese",
+    "ko": "Korean",
+    "es": "Spanish",
+    "fr": "French",
+    "pt": "Portuguese",
+    "de": "German",
+    "it": "Italian",
+    "th": "Thai",
+    "vi": "Vietnamese",
+    "id": "Indonesian",
+    "ms": "Malay",
+    "ar": "Arabic",
+    "hi": "Hindi",
+    "he": "Hebrew",
+    "my": "Burmese",
+    "ta": "Tamil",
+    "ur": "Urdu",
+    "bn": "Bengali",
+    "pl": "Polish",
+    "nl": "Dutch",
+    "ro": "Romanian",
+    "tr": "Turkish",
+    "km": "Khmer",
+    "lo": "Lao",
+    "yue": "Cantonese",
+    "cs": "Czech",
+    "el": "Greek",
+    "sv": "Swedish",
+    "hu": "Hungarian",
+    "da": "Danish",
+    "fi": "Finnish",
+    "uk": "Ukrainian",
+    "bg": "Bulgarian",
+}
+
+TARGET_LANG_CODE_MAP = {v: k for k, v in SOURCE_LANG_CODE_MAP.items()}
+
+TRANSLATION_PROMPTS = {
+    "general": {
+        "zh": "你是一名专业的 {source_lang}（{src_lang_code}）到 {target_lang}（{tgt_lang_code}）翻译专家，请准确传达原文含义并符合{target_lang}语言习惯，只输出翻译结果：{text}",
+        "en": "You are a professional {source_lang} ({src_lang_code}) to {target_lang} ({tgt_lang_code}) translator. Accurately convey the meaning following {target_lang} grammar and usage, output only the translation: {text}",
+        "ru": "Вы профессиональный переводчик с {source_lang} ({src_lang_code}) на {target_lang} ({tgt_lang_code}). Точно передайте смысл текста, соблюдая нормы {target_lang}, выводите только перевод: {text}",
+        "ar": "أنت مترجم محترف من {source_lang} ({src_lang_code}) إلى {target_lang} ({tgt_lang_code}). انقل المعنى بدقة وفق قواعد {target_lang} وأخرج الترجمة فقط: {text}",
+        "ja": "あなたは {source_lang}（{src_lang_code}）から {target_lang}（{tgt_lang_code}）へのプロ翻訳者です。意味を正確に伝え、{target_lang}の表現に従い、翻訳のみ出力してください：{text}",
+        "es": "Eres un traductor profesional de {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Transmite con precisión el significado y devuelve solo la traducción: {text}",
+        "de": "Du bist ein professioneller Übersetzer von {source_lang} ({src_lang_code}) nach {target_lang} ({tgt_lang_code}). Gib die Bedeutung korrekt wieder und gib nur die Übersetzung aus: {text}",
+        "fr": "Vous êtes un traducteur professionnel de {source_lang} ({src_lang_code}) vers {target_lang} ({tgt_lang_code}). Transmettez fidèlement le sens et produisez uniquement la traduction : {text}",
+        "it": "Sei un traduttore professionista da {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Trasmetti accuratamente il significato e restituisci solo la traduzione: {text}",
+        "pt": "Você é um tradutor profissional de {source_lang} ({src_lang_code}) para {target_lang} ({tgt_lang_code}). Transmita o significado com precisão e produza apenas a tradução: {text}"
+    },
+
+    "sku_name": {
+        "zh": "你是一名专业的 {source_lang}（{src_lang_code}）到 {target_lang}（{tgt_lang_code}）电商翻译专家，请将原文翻译为{target_lang}商品SKU名称，要求准确完整、简洁专业，只输出结果：{text}",
+        "en": "You are a professional {source_lang} ({src_lang_code}) to {target_lang} ({tgt_lang_code}) ecommerce translator. Translate into a concise and accurate {target_lang} product SKU name, output only the result: {text}",
+        "ru": "Вы переводчик e-commerce с {source_lang} ({src_lang_code}) на {target_lang} ({tgt_lang_code}). Переведите в краткое и точное название SKU товара на {target_lang}, выводите только результат: {text}",
+        "ar": "أنت مترجم تجارة إلكترونية من {source_lang} ({src_lang_code}) إلى {target_lang} ({tgt_lang_code}). ترجم إلى اسم SKU للمنتج بلغة {target_lang} بدقة واختصار، وأخرج النتيجة فقط: {text}",
+        "ja": "{source_lang}（{src_lang_code}）から {target_lang}（{tgt_lang_code}）へのEC翻訳者として、簡潔で正確な{target_lang}の商品SKU名に翻訳し、結果のみ出力してください：{text}",
+        "es": "Eres un traductor ecommerce de {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Traduce a un nombre SKU de producto en {target_lang}, preciso y conciso, devuelve solo el resultado: {text}",
+        "de": "Du bist ein E-Commerce-Übersetzer von {source_lang} ({src_lang_code}) nach {target_lang} ({tgt_lang_code}). Übersetze in einen präzisen und kurzen {target_lang} Produkt-SKU-Namen, nur Ergebnis ausgeben: {text}",
+        "fr": "Vous êtes un traducteur e-commerce de {source_lang} ({src_lang_code}) vers {target_lang} ({tgt_lang_code}). Traduisez en un nom SKU produit {target_lang} précis et concis, sortie uniquement : {text}",
+        "it": "Sei un traduttore ecommerce da {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Traduce in un nome SKU prodotto {target_lang} conciso e accurato, restituisci solo il risultato: {text}",
+        "pt": "Você é um tradutor de e-commerce de {source_lang} ({src_lang_code}) para {target_lang} ({tgt_lang_code}). Traduza para um nome SKU de produto {target_lang} conciso e preciso, produza apenas o resultado: {text}"
+    },
+
+    "ecommerce_search_query": {
+        "zh": "你是一名专业的 {source_lang}（{src_lang_code}）到 {target_lang}（{tgt_lang_code}）翻译助手，请将电商搜索词准确翻译为{target_lang}并符合搜索习惯，只输出结果：{text}",
+        "en": "You are a professional {source_lang} ({src_lang_code}) to {target_lang} ({tgt_lang_code}) translator. Translate the ecommerce search query accurately following {target_lang} search habits, output only the result: {text}",
+        "ru": "Вы переводчик с {source_lang} ({src_lang_code}) на {target_lang} ({tgt_lang_code}). Переведите поисковый запрос e-commerce с учётом привычек поиска, выводите только результат: {text}",
+        "ar": "أنت مترجم من {source_lang} ({src_lang_code}) إلى {target_lang} ({tgt_lang_code}). ترجم عبارة البحث للتجارة الإلكترونية بما يناسب عادات البحث وأخرج النتيجة فقط: {text}",
+        "ja": "{source_lang}（{src_lang_code}）から {target_lang}（{tgt_lang_code}）への翻訳者として、EC検索キーワードを{target_lang}の検索習慣に合わせて翻訳し、結果のみ出力してください：{text}",
+        "es": "Eres un traductor de {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Traduce la consulta de búsqueda ecommerce según los hábitos de búsqueda y devuelve solo el resultado: {text}",
+        "de": "Du bist ein Übersetzer von {source_lang} ({src_lang_code}) nach {target_lang} ({tgt_lang_code}). Übersetze die E-Commerce-Suchanfrage entsprechend den Suchgewohnheiten, nur Ergebnis ausgeben: {text}",
+        "fr": "Vous êtes un traducteur de {source_lang} ({src_lang_code}) vers {target_lang} ({tgt_lang_code}). Traduisez la requête de recherche e-commerce selon les habitudes de recherche, sortie uniquement : {text}",
+        "it": "Sei un traduttore da {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Traduce la query di ricerca ecommerce secondo le abitudini di ricerca e restituisci solo il risultato: {text}",
+        "pt": "Você é um tradutor de {source_lang} ({src_lang_code}) para {target_lang} ({tgt_lang_code}). Traduza a consulta de busca de ecommerce conforme os hábitos de busca e produza apenas o resultado: {text}"
+    }
+}
@@ -1814,7 +1814,8 @@ curl &quot;http://localhost:6007/health&quot;
   "text": "商品名称",
   "target_lang": "en",
   "source_lang": "zh",
-  "model": "qwen"
+  "model": "qwen",
+  "context": "sku_name"
 }
 ```
@@ -1823,7 +1824,8 @@ curl &quot;http://localhost:6007/health&quot;
 | `text` | string | Y | 待翻译文本 |
 | `target_lang` | string | Y | 目标语言：`zh`、`en`、`ru` 等 |
 | `source_lang` | string | N | 源语言，不传则自动检测 |
-| `model` | string | N | `qwen`（默认）或 `deepl` |
+| `model` | string | N | `qwen`（默认）、`deepl` 或 `llm` |
+| `context` | string | N | 翻译场景参数：商品标题翻译使用 `sku_name`，搜索请求中的 query 翻译使用 `ecommerce_search_query`，其它通用场景可不传或使用 `general` |
 **响应**:
 ```json
@@ -36,7 +36,6 @@ class SPUDocumentTransformer:
         searchable_option_dimensions: List[str],
         tenant_config: Optional[Dict[str, Any]] = None,
         translator: Optional[Any] = None,
-        translation_prompts: Optional[Dict[str, str]] = None,
         encoder: Optional[Any] = None,
         enable_title_embedding: bool = True,
         image_encoder: Optional[Any] = None,
@@ -50,7 +49,6 @@ class SPUDocumentTransformer:
             searchable_option_dimensions: 可搜索的option维度列表
             tenant_config: 租户配置（包含主语言和翻译配置）
             translator: 翻译器实例（可选，如果提供则启用翻译功能）
-            translation_prompts: 翻译提示词配置（可选）
             encoder: 文本编码器实例（可选，用于生成title_embedding）
             enable_title_embedding: 是否启用标题向量化（默认True）
             image_encoder: 图片编码器实例（可选，需实现 encode_image_urls(urls) -> List[Optional[np.ndarray]]）
@@ -60,12 +58,33 @@ class SPUDocumentTransformer:
         self.searchable_option_dimensions = searchable_option_dimensions
         self.tenant_config = tenant_config or {}
         self.translator = translator
-        self.translation_prompts = translation_prompts or {}
         self.encoder = encoder
         self.enable_title_embedding = enable_title_embedding
         self.image_encoder = image_encoder
         self.enable_image_embedding = bool(enable_image_embedding and image_encoder is not None)
+    def _translate_index_languages(
+        self,
+        text: str,
+        source_lang: str,
+        index_languages: List[str],
+        scene: str,
+    ) -> Dict[str, Optional[str]]:
+        translations: Dict[str, Optional[str]] = {}
+        if not self.translator or not text or not str(text).strip():
+            return translations
+        for lang in index_languages:
+            if lang == source_lang:
+                translations[lang] = text
+                continue
+            translations[lang] = self.translator.translate(
+                text=text,
+                target_lang=lang,
+                source_lang=source_lang,
+                context=scene,
+            )
+        return translations
+
     def transform_spu_to_doc(
         self,
         tenant_id: str,
@@ -322,15 +341,12 @@ class SPUDocumentTransformer:
             title_text = str(spu_row['title'])
             translations: Dict[str, Optional[str]] = {}
             if self.translator:
-                prompt_zh = self.translation_prompts.get('product_title_zh') or self.translation_prompts.get('default_zh')
-                prompt_en = self.translation_prompts.get('product_title_en') or self.translation_prompts.get('default_en')
-                translations = self.translator.translate_for_indexing(
-                    title_text,
-                    shop_language=primary_lang,
+                translations = self._translate_index_languages(
+                    text=title_text,
                     source_lang=primary_lang,
-                    prompt=prompt_zh if primary_lang == 'zh' else prompt_en,
                     index_languages=index_langs,
-                ) or {}
+                    scene="product_title",
+                )
             _set_lang_obj("title", title_text, translations)
         # Brief
@@ -338,14 +354,12 @@ class SPUDocumentTransformer:
             brief_text = str(spu_row['brief'])
             translations = {}
             if self.translator:
-                prompt = self.translation_prompts.get('default_zh') or self.translation_prompts.get('default_en')
-                translations = self.translator.translate_for_indexing(
-                    brief_text,
-                    shop_language=primary_lang,
+                translations = self._translate_index_languages(
+                    text=brief_text,
                     source_lang=primary_lang,
-                    prompt=prompt,
                     index_languages=index_langs,
-                ) or {}
+                    scene="default",
+                )
             _set_lang_obj("brief", brief_text, translations)
         # Description
@@ -353,14 +367,12 @@ class SPUDocumentTransformer:
             desc_text = str(spu_row['description'])
             translations = {}
             if self.translator:
-                prompt = self.translation_prompts.get('default_zh') or self.translation_prompts.get('default_en')
-                translations = self.translator.translate_for_indexing(
-                    desc_text,
-                    shop_language=primary_lang,
+                translations = self._translate_index_languages(
+                    text=desc_text,
                     source_lang=primary_lang,
-                    prompt=prompt,
                     index_languages=index_langs,
-                ) or {}
+                    scene="default",
+                )
             _set_lang_obj("description", desc_text, translations)
         # Vendor
@@ -368,14 +380,12 @@ class SPUDocumentTransformer:
             vendor_text = str(spu_row['vendor'])
             translations = {}
             if self.translator:
-                prompt = self.translation_prompts.get('default_zh') or self.translation_prompts.get('default_en')
-                translations = self.translator.translate_for_indexing(
-                    vendor_text,
-                    shop_language=primary_lang,
+                translations = self._translate_index_languages(
+                    text=vendor_text,
                     source_lang=primary_lang,
-                    prompt=prompt,
                     index_languages=index_langs,
-                ) or {}
+                    scene="default",
+                )
             _set_lang_obj("vendor", vendor_text, translations)
     def _fill_category_fields(self, doc: Dict[str, Any], spu_row: pd.Series):
@@ -39,7 +39,6 @@ class IncrementalIndexerService:
         self._config: Optional[Any] = None
         self._config_lock = threading.Lock()
         self._translator: Optional[Any] = None
-        self._translation_prompts: Optional[Dict[str, Any]] = None
         self._searchable_option_dimensions: Optional[List[str]] = None
         self._shared_text_encoder: Optional[Any] = None
         self._shared_image_encoder: Optional[Any] = None
@@ -52,7 +51,6 @@ class IncrementalIndexerService:
     def _eager_init(self) -> None:
         """Strict eager initialization. Any dependency failure should fail fast."""
         self._config = ConfigLoader("config/config.yaml").load_config()
-        self._translation_prompts = getattr(self._config.query_config, "translation_prompts", {}) or {}
         self._searchable_option_dimensions = (
             getattr(self._config.spu_config, "searchable_option_dimensions", None)
             or ["option1", "option2", "option3"]
@@ -110,7 +108,6 @@ class IncrementalIndexerService:
             tenant_id=tenant_id,
             searchable_option_dimensions=self._searchable_option_dimensions,
             translator=self._translator,
-            translation_prompts=self._translation_prompts,
             encoder=encoder,
             enable_title_embedding=False,  # batch fill later
             image_encoder=image_encoder,
@@ -57,7 +57,6 @@ def create_document_transformer(
     tenant_id: str,
     searchable_option_dimensions: Optional[list] = None,
     translator: Optional[Any] = None,
-    translation_prompts: Optional[Dict[str, str]] = None,
     encoder: Optional[Any] = None,
     enable_title_embedding: bool = True,
     image_encoder: Optional[Any] = None,
@@ -72,7 +71,6 @@ def create_document_transformer(
         tenant_id: 租户ID
         searchable_option_dimensions: 可搜索的option维度列表（如果为None则从配置加载）
         translator: 翻译器实例（如果为None则根据配置初始化）
-        translation_prompts: 翻译提示词配置（如果为None则从配置加载）
         encoder: 文本编码器实例（如果为None且enable_title_embedding为True则根据配置初始化）
         enable_title_embedding: 是否启用标题向量化（默认True）
         image_encoder: 图片编码器（可选，需实现 encode_image_urls(urls)）
@@ -89,7 +87,6 @@ def create_document_transformer(
     if (
         searchable_option_dimensions is None
         or translator is None
-        or translation_prompts is None
         or (encoder is None and enable_title_embedding)
         or config is None
     ):
@@ -107,9 +104,6 @@ def create_document_transformer(
             translator = create_translation_provider(config.query_config)
-        if translation_prompts is None:
-            translation_prompts = config.query_config.translation_prompts
-
         # 初始化encoder（如果启用标题向量化且未提供encoder）
         if encoder is None and enable_title_embedding and config.query_config.enable_text_embedding:
             from embeddings.text_encoder import TextEmbeddingEncoder
@@ -122,7 +116,6 @@ def create_document_transformer(
         searchable_option_dimensions=searchable_option_dimensions,
         tenant_config=tenant_config,
         translator=translator,
-        translation_prompts=translation_prompts,
         encoder=encoder,
         enable_title_embedding=enable_title_embedding,
         image_encoder=image_encoder,
@@ -285,7 +285,6 @@ def test_document_transformer():
             searchable_option_dimensions=['option1', 'option2', 'option3'],
             tenant_config=tenant_config,
             translator=translator,
-            translation_prompts=config.query_config.translation_prompts
         )
         # 转换文档
-"""
-Translation provider - direct (in-process) or HTTP service.
-"""
+"""Translation provider factory and HTTP provider implementation."""
 from __future__ import annotations
 import logging
-from typing import Any, Dict, List, Optional, Union
-
-from concurrent.futures import Future, ThreadPoolExecutor
+from typing import Any, Dict, Optional
 import requests
 from config.services_config import get_translation_config, get_translation_base_url
@@ -22,19 +18,18 @@ class HttpTranslationProvider:
         base_url: str,
         model: str = "qwen",
         timeout_sec: float = 10.0,
-        translation_context: Optional[str] = None,
     ):
         self.base_url = (base_url or "").rstrip("/")
         self.model = model or "qwen"
         self.timeout_sec = float(timeout_sec or 10.0)
-        self.translation_context = translation_context or "e-commerce product search"
-        self.executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="http-translator")
     def _translate_once(
         self,
         text: str,
         target_lang: str,
         source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None,
     ) -> Optional[str]:
         if not text or not str(text).strip():
             return text
@@ -46,6 +41,10 @@ class HttpTranslationProvider:
                 "source_lang": source_lang or "auto",
                 "model": self.model,
             }
+            if context:
+                payload["context"] = context
+            if prompt:
+                payload["prompt"] = prompt
             response = requests.post(url, json=payload, timeout=self.timeout_sec)
             if response.status_code != 200:
                 logger.warning(
@@ -69,58 +68,13 @@ class HttpTranslationProvider:
         context: Optional[str] = None,
         prompt: Optional[str] = None,
     ) -> Optional[str]:
-        del context, prompt
-        result = self._translate_once(text=text, target_lang=target_lang, source_lang=source_lang)
-        return result if result is not None else text
-
-    def translate_multi(
-        self,
-        text: str,
-        target_langs: List[str],
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        async_mode: bool = True,
-        prompt: Optional[str] = None,
-    ) -> Dict[str, Optional[str]]:
-        del context, async_mode, prompt
-        out: Dict[str, Optional[str]] = {}
-        for lang in target_langs:
-            out[lang] = self.translate(text, lang, source_lang=source_lang)
-        return out
-
-    def translate_multi_async(
-        self,
-        text: str,
-        target_langs: List[str],
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None,
-    ) -> Dict[str, Union[str, Future]]:
-        del context, prompt
-        out: Dict[str, Union[str, Future]] = {}
-        for lang in target_langs:
-            out[lang] = self.executor.submit(self.translate, text, lang, source_lang)
-        return out
-
-    def translate_for_indexing(
-        self,
-        text: str,
-        shop_language: str,
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None,
-        index_languages: Optional[List[str]] = None,
-    ) -> Dict[str, Optional[str]]:
-        del context, prompt
-        langs = index_languages if index_languages else ["en", "zh"]
-        source = source_lang or shop_language or "auto"
-        out: Dict[str, Optional[str]] = {}
-        for lang in langs:
-            if lang == shop_language:
-                out[lang] = text
-            else:
-                out[lang] = self.translate(text, target_lang=lang, source_lang=source)
-        return out
+        return self._translate_once(
+            text=text,
+            target_lang=target_lang,
+            source_lang=source_lang,
+            context=context,
+            prompt=prompt,
+        )
 def create_translation_provider(query_config: Any = None) -> Any:
@@ -133,9 +87,9 @@ def create_translation_provider(query_config: Any = None) -&gt; Any:
     provider = cfg.provider
     pc = cfg.get_provider_cfg()
-    if provider in ("direct", "local", "inprocess"):
+    if provider in ("qwen-mt", "direct", "local", "inprocess"):
         from query.qwen_mt_translate import Translator
-        model = pc.get("model") or "qwen"
+        model = pc.get("model") or "qwen-mt-flash"
         qc = query_config or _empty_query_config()
         return Translator(
             model=model,
@@ -145,7 +99,7 @@ def create_translation_provider(query_config: Any = None) -&gt; Any:
             translation_context=getattr(qc, "translation_context", "e-commerce product search"),
         )
-    if provider in ("http", "service"):
+    elif provider in ("http", "service"):
         base_url = get_translation_base_url()
         model = pc.get("model") or "qwen"
         timeout = pc.get("timeout_sec", 10.0)
@@ -154,7 +108,26 @@ def create_translation_provider(query_config: Any = None) -&gt; Any:
             base_url=base_url,
             model=model,
             timeout_sec=float(timeout),
-            translation_context=getattr(qc, "translation_context", "e-commerce product search"),
+        )
+
+    elif provider == "llm":
+        from query.llm_translate import LLMTranslatorProvider
+        model = pc.get("model")
+        timeout = float(pc.get("timeout_sec", 30.0))
+        base_url = (pc.get("base_url") or "").strip() or None
+        return LLMTranslatorProvider(
+            model=model,
+            timeout_sec=timeout,
+            base_url=base_url,
+        )
+
+    elif provider == "deepl":
+        from query.deepl_provider import DeepLProvider
+        qc = query_config or _empty_query_config()
+        return DeepLProvider(
+            api_key=getattr(qc, "translation_api_key", None),
+            timeout=float(pc.get("timeout_sec", 10.0)),
+            glossary_id=pc.get("glossary_id") or getattr(qc, "translation_glossary_id", None),
         )
     raise ValueError(f"Unsupported translation provider: {provider}")
@@ -0,0 +1,203 @@
+"""
+DeepL backend provider.
+
+This module only handles network calls to DeepL.
+It does not handle cache, async fanout, or fallback semantics.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import re
+from typing import Dict, Optional, Tuple
+
+import requests
+from config.services_config import get_translation_config
+
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_CONTEXTS: Dict[str, Dict[str, str]] = {
+    "sku_name": {
+        "zh": "商品SKU名称",
+        "en": "product SKU name",
+    },
+    "ecommerce_search_query": {
+        "zh": "电商",
+        "en": "e-commerce",
+    },
+    "general": {
+        "zh": "",
+        "en": "",
+    },
+}
+SCENE_NAMES = frozenset(DEFAULT_CONTEXTS.keys())
+
+
+def _merge_contexts(raw: object) -> Dict[str, Dict[str, str]]:
+    merged: Dict[str, Dict[str, str]] = {
+        scene: dict(lang_map) for scene, lang_map in DEFAULT_CONTEXTS.items()
+    }
+    if not isinstance(raw, dict):
+        return merged
+    for scene, lang_map in raw.items():
+        if not isinstance(lang_map, dict):
+            continue
+        scene_name = str(scene or "").strip()
+        if not scene_name:
+            continue
+        merged.setdefault(scene_name, {})
+        for lang, value in lang_map.items():
+            lang_key = str(lang or "").strip().lower()
+            context_value = str(value or "").strip()
+            if lang_key and context_value:
+                merged[scene_name][lang_key] = context_value
+    return merged
+
+
+class DeepLProvider:
+    API_URL = "https://api.deepl.com/v2/translate"  # Pro tier
+    LANG_CODE_MAP = {
+        "zh": "ZH",
+        "en": "EN",
+        "ru": "RU",
+        "ar": "AR",
+        "ja": "JA",
+        "es": "ES",
+        "de": "DE",
+        "fr": "FR",
+        "it": "IT",
+        "pt": "PT",
+    }
+
+    def __init__(
+        self,
+        api_key: Optional[str],
+        *,
+        timeout: float = 10.0,
+        glossary_id: Optional[str] = None,
+    ) -> None:
+        cfg = get_translation_config()
+        provider_cfg = cfg.providers.get("deepl", {}) if isinstance(cfg.providers, dict) else {}
+        self.api_key = api_key or os.getenv("DEEPL_AUTH_KEY")
+        self.timeout = float(provider_cfg.get("timeout_sec") or timeout or 10.0)
+        self.glossary_id = glossary_id or provider_cfg.get("glossary_id")
+        self.model = "deepl"
+        self.context_presets = _merge_contexts(provider_cfg.get("contexts"))
+        if not self.api_key:
+            logger.warning("DEEPL_AUTH_KEY not set; DeepL translation is unavailable")
+
+    def _resolve_request_context(
+        self,
+        target_lang: str,
+        context: Optional[str],
+        prompt: Optional[str],
+    ) -> Optional[str]:
+        if prompt:
+            return prompt
+        if context in SCENE_NAMES:
+            scene_map = self.context_presets.get(context) or self.context_presets.get("default") or {}
+            tgt = (target_lang or "").strip().lower()
+            return scene_map.get(tgt) or scene_map.get("en")
+        if context:
+            return context
+        scene_map = self.context_presets.get("default") or {}
+        tgt = (target_lang or "").strip().lower()
+        return scene_map.get(tgt) or scene_map.get("en")
+
+    def translate(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None,
+    ) -> Optional[str]:
+        if not self.api_key:
+            return None
+
+        target_code = self.LANG_CODE_MAP.get((target_lang or "").lower(), (target_lang or "").upper())
+        headers = {
+            "Authorization": f"DeepL-Auth-Key {self.api_key}",
+            "Content-Type": "application/json",
+        }
+
+        api_context = self._resolve_request_context(target_lang, context, prompt)
+        text_to_translate, needs_extraction = self._add_ecommerce_context(text, source_lang, api_context)
+
+        payload = {
+            "text": [text_to_translate],
+            "target_lang": target_code,
+        }
+        if source_lang:
+            payload["source_lang"] = self.LANG_CODE_MAP.get(source_lang.lower(), source_lang.upper())
+        if api_context:
+            payload["context"] = api_context
+        if self.glossary_id:
+            payload["glossary_id"] = self.glossary_id
+
+        try:
+            response = requests.post(self.API_URL, headers=headers, json=payload, timeout=self.timeout)
+            if response.status_code != 200:
+                logger.warning(
+                    "[deepl] Failed | status=%s tgt=%s body=%s",
+                    response.status_code,
+                    target_code,
+                    (response.text or "")[:200],
+                )
+                return None
+
+            data = response.json()
+            translations = data.get("translations") or []
+            if not translations:
+                return None
+            translated = translations[0].get("text")
+            if not translated:
+                return None
+            if needs_extraction:
+                translated = self._extract_term_from_translation(translated, text, target_code)
+            return translated
+        except requests.Timeout:
+            logger.warning("[deepl] Timeout | tgt=%s timeout=%.1fs", target_code, self.timeout)
+            return None
+        except Exception as exc:
+            logger.warning("[deepl] Exception | tgt=%s error=%s", target_code, exc, exc_info=True)
+            return None
+
+    def _add_ecommerce_context(
+        self,
+        text: str,
+        source_lang: Optional[str],
+        context: Optional[str],
+    ) -> Tuple[str, bool]:
+        if not context or "e-commerce" not in context.lower():
+            return text, False
+        if (source_lang or "").lower() != "zh":
+            return text, False
+
+        term = (text or "").strip()
+        if len(term.split()) == 1 and len(term) <= 2:
+            return f"购买 {term}", True
+        return text, False
+
+    def _extract_term_from_translation(
+        self,
+        translated_text: str,
+        original_text: str,
+        target_lang_code: str,
+    ) -> str:
+        del original_text
+        if target_lang_code != "EN":
+            return translated_text
+
+        words = translated_text.strip().split()
+        if len(words) <= 1:
+            return translated_text
+        context_words = {"buy", "purchase", "product", "item", "commodity", "goods"}
+        for word in reversed(words):
+            normalized = re.sub(r"[.,!?;:]+$", "", word.lower())
+            if normalized not in context_words:
+                return normalized
+        return re.sub(r"[.,!?;:]+$", "", words[-1].lower())
+
 """
-LLM-based translation helper using Qwen chat model.
+LLM-based translation backend (DashScope-compatible OpenAI API).
-This module provides a thin wrapper around DashScope's `qwen-flash` model
-for high-quality, prompt-controlled translation, independent of the main
-`Translator` (machine translation) pipeline.
-
-Usage example:
-
-    from query.llm_translate import llm_translate
-
-    result = llm_translate(
-        text="我看到这个视频后没有笑",
-        target_lang="en",
-        source_lang="zh",
-        source_lang_label="中文",
-        target_lang_label="英文",
-    )
+Failure semantics are strict:
+- success: translated string
+- failure: None
 """
 from __future__ import annotations
@@ -23,113 +11,159 @@ from __future__ import annotations
 import logging
 import os
 import time
-from typing import Dict, Optional
+from typing import Optional
 from openai import OpenAI
 from config.env_config import DASHSCOPE_API_KEY
 from config.services_config import get_translation_config
+from config.translate_prompts import TRANSLATION_PROMPTS, SOURCE_LANG_CODE_MAP
+
 logger = logging.getLogger(__name__)
-# 华北2（北京）：https://dashscope.aliyuncs.com/compatible-mode/v1
-# 新加坡：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
-# 美国（弗吉尼亚）：https://dashscope-us.aliyuncs.com/compatible-mode/v1
-#
-# 默认保持与现有翻译/索引脚本相同的美国地域，可通过环境变量覆盖：
-#   DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
 DEFAULT_QWEN_BASE_URL = "https://dashscope-us.aliyuncs.com/compatible-mode/v1"
-QWEN_MODEL_NAME = "qwen-flash"
-
-
-# 由调用方提供的语言标签/代码填充，占位符说明：
-# - source_lang: 源语言的人类可读名称（按目标语言本地化，例如 "中文", "English"）
-# - target_lang: 目标语言的人类可读名称
-# - src_lang_code: 源语言代码，例如 "zh"
-# - tgt_lang_code: 目标语言代码，例如 "en"
-TRANSLATION_PROMPTS: Dict[str, str] = {
-    "zh": """你是一名专业的 {source_lang}（{src_lang_code}）到 {target_lang}（{tgt_lang_code}）翻译员。你的目标是在遵循 {target_lang} 的语法、词汇和文化习惯的前提下，准确传达原始 {source_lang} 文本的含义和细微差别。请只输出 {target_lang} 的翻译内容，不要包含任何额外的解释或评论。请将以下 {source_lang} 文本翻译成 {target_lang}：
-
-{text}""",
-    "en": """You are a professional {source_lang} ({src_lang_code}) to {target_lang} ({tgt_lang_code}) translator. Your goal is to accurately convey the meaning and nuances of the original {source_lang} text while adhering to {target_lang} grammar, vocabulary, and cultural sensitivities. Produce only the {target_lang} translation, without any additional explanations or commentary. Please translate the following {source_lang} text into {target_lang}:
-
-{text}""",
-    "ru": """Вы профессиональный переводчик с {source_lang} ({src_lang_code}) на {target_lang} ({tgt_lang_code}). Ваша задача — точно передать смысл и нюансы исходного текста на {source_lang}, соблюдая грамматику, лексику и культурные особенности {target_lang}. Выводите только перевод на {target_lang}, без каких-либо дополнительных объяснений или комментариев. Пожалуйста, переведите следующий текст с {source_lang} на {target_lang}:
-
-{text}""",
-    "ar": """أنت مترجم محترف من {source_lang} ({src_lang_code}) إلى {target_lang} ({tgt_lang_code}). هدفك هو نقل المعنى والدلالات الدقيقة للنص الأصلي بلغة {source_lang} بدقة، مع الالتزام بقواعد اللغة والمفردات والحساسيات الثقافية الخاصة بلغة {target_lang}. قم بإنتاج الترجمة إلى {target_lang} فقط دون أي شروحات أو تعليقات إضافية. يرجى ترجمة النص التالي من {source_lang} إلى {target_lang}:
-
-{text}""",
-    "ja": """あなたは {source_lang}（{src_lang_code}）から {target_lang}（{tgt_lang_code}）へのプロの翻訳者です。{target_lang} の文法、語彙、文化的配慮に従いながら、元の {source_lang} テキストの意味やニュアンスを正確に伝えることが目的です。追加の説明やコメントは一切含めず、{target_lang} の翻訳のみを出力してください。次の {source_lang} テキストを {target_lang} に翻訳してください：
-
-{text}""",
-    "es": """Eres un traductor profesional de {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Tu objetivo es transmitir con precisión el significado y los matices del texto original en {source_lang}, respetando la gramática, el vocabulario y las sensibilidades culturales de {target_lang}. Produce únicamente la traducción en {target_lang}, sin explicaciones ni comentarios adicionales. Por favor, traduce el siguiente texto de {source_lang} a {target_lang}:
-
-{text}""",
-    "de": """Du bist ein professioneller Übersetzer von {source_lang} ({src_lang_code}) nach {target_lang} ({tgt_lang_code}). Dein Ziel ist es, die Bedeutung und Nuancen des ursprünglichen {source_lang}-Textes genau zu vermitteln und dabei die Grammatik, den Wortschatz und die kulturellen Besonderheiten von {target_lang} zu berücksichtigen. Gib ausschließlich die Übersetzung in {target_lang} aus, ohne zusätzliche Erklärungen oder Kommentare. Bitte übersetze den folgenden {source_lang}-Text in {target_lang}:
-
-{text}""",
-    "fr": """Vous êtes un traducteur professionnel de {source_lang} ({src_lang_code}) vers {target_lang} ({tgt_lang_code}). Votre objectif est de transmettre fidèlement le sens et les nuances du texte original en {source_lang}, tout en respectant la grammaire, le vocabulaire et les sensibilités culturelles de {target_lang}. Produisez uniquement la traduction en {target_lang}, sans explications ni commentaires supplémentaires. Veuillez traduire le texte suivant de {source_lang} vers {target_lang} :
-
-{text}""",
-    "it": """Sei un traduttore professionista da {source_lang} ({src_lang_code}) a {target_lang} ({tgt_lang_code}). Il tuo obiettivo è trasmettere con precisione il significato e le sfumature del testo originale in {source_lang}, rispettando la grammatica, il vocabolario e le sensibilità culturali di {target_lang}. Produci solo la traduzione in {target_lang}, senza spiegazioni o commenti aggiuntivi. Per favore traduci il seguente testo da {source_lang} a {target_lang}:
-
-{text}""",
-    "pt": """Você é um tradutor profissional de {source_lang} ({src_lang_code}) para {target_lang} ({tgt_lang_code}). Seu objetivo é transmitir com precisão o significado e as nuances do texto original em {source_lang}, respeitando a gramática, o vocabulário e as sensibilidades culturais de {target_lang}. Produza apenas a tradução em {target_lang}, sem quaisquer explicações ou comentários adicionais. Por favor, traduza o seguinte texto de {source_lang} para {target_lang}:
-
-{text}""",
-}
-
-
-def _get_qwen_client(base_url: Optional[str] = None) -> Optional[OpenAI]:
-    """
-    Lazily construct an OpenAI-compatible client for DashScope.
-
-    Uses DASHSCOPE_API_KEY and base_url (provider config / env) to configure endpoint.
-    """
-    api_key = DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY")
-    if not api_key:
-        logger.warning("DASHSCOPE_API_KEY not set; llm-based translation will be disabled")
-        return None
-
-    # 优先使用显式传入的 base_url，其次环境变量，最后默认地域。
-    base_url = (
-        (base_url or "").strip()
-        or os.getenv("DASHSCOPE_BASE_URL")
-        or DEFAULT_QWEN_BASE_URL
-    )
-
-    try:
-        client = OpenAI(api_key=api_key, base_url=base_url)
-        return client
-    except Exception as exc:
-        logger.error("Failed to initialize DashScope OpenAI client: %s", exc, exc_info=True)
-        return None
+DEFAULT_LLM_MODEL = "qwen-flash"
 def _build_prompt(
     text: str,
+    *,
+    source_lang: Optional[str],
     target_lang: str,
-    source_lang_label: str,
-    target_lang_label: str,
-    src_lang_code: str,
-    tgt_lang_code: str,
+    scene: Optional[str],
 ) -> str:
     """
-    Build translation prompt for given target language, defaulting to English template.
+    从 config.translate_prompts.TRANSLATION_PROMPTS 中构建提示词。
+
+    要求：模板必须包含 {source_lang}（{src_lang_code}）{target_lang}（{tgt_lang_code}）。
+    这里统一使用 code 作为占位的 lang 与 label，外部接口仍然只传语言 code。
     """
-    key = (target_lang or "").lower()
-    template = TRANSLATION_PROMPTS.get(key) or TRANSLATION_PROMPTS["en"]
+    tgt = (target_lang or "").lower() or "en"
+    src = (source_lang or "auto").lower()
+
+    # 将业务上下文 scene 映射为模板分组名
+    normalized_scene = (scene or "").strip() or "general"
+    # 如果出现历史词，则报错，用于发现错误
+    if normalized_scene in {"query", "ecommerce_search", "ecommerce_search_query"}:
+        group_key = "ecommerce_search_query"
+    elif normalized_scene in {"product_title", "sku_name"}:
+        group_key = "sku_name"
+    else:
+        group_key = normalized_scene
+    group = TRANSLATION_PROMPTS.get(group_key) or TRANSLATION_PROMPTS["general"]
+
+    # 先按目标语言 code 取模板，取不到回退到英文
+    template = group.get(tgt) or group.get("en")
+    if not template:
+        # 理论上不会发生，兜底一个简单模板
+        template = (
+            "You are a professional {source_lang} ({src_lang_code}) to "
+            "{target_lang} ({tgt_lang_code}) translator, output only the translation: {text}"
+        )
+
+    # 目前不额外维护语言名称映射，直接使用 code 作为 label
+    source_lang_label = SOURCE_LANG_CODE_MAP.get(src, src)
+    target_lang_label = SOURCE_LANG_CODE_MAP.get(tgt, tgt)
+
     return template.format(
         source_lang=source_lang_label,
+        src_lang_code=src,
         target_lang=target_lang_label,
-        src_lang_code=src_lang_code,
-        tgt_lang_code=tgt_lang_code,
+        tgt_lang_code=tgt,
         text=text,
     )
+class LLMTranslatorProvider:
+    def __init__(
+        self,
+        *,
+        model: Optional[str] = None,
+        timeout_sec: float = 30.0,
+        base_url: Optional[str] = None,
+    ) -> None:
+        cfg = get_translation_config()
+        llm_cfg = cfg.providers.get("llm", {}) if isinstance(cfg.providers, dict) else {}
+        self.model = model or llm_cfg.get("model") or DEFAULT_LLM_MODEL
+        self.timeout_sec = float(llm_cfg.get("timeout_sec") or timeout_sec or 30.0)
+        self.base_url = (
+            (base_url or "").strip()
+            or (llm_cfg.get("base_url") or "").strip()
+            or os.getenv("DASHSCOPE_BASE_URL")
+            or DEFAULT_QWEN_BASE_URL
+        )
+        self.client = self._create_client()
+
+    def _create_client(self) -> Optional[OpenAI]:
+        api_key = DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY")
+        if not api_key:
+            logger.warning("DASHSCOPE_API_KEY not set; llm translation unavailable")
+            return None
+        try:
+            return OpenAI(api_key=api_key, base_url=self.base_url)
+        except Exception as exc:
+            logger.error("Failed to initialize llm translation client: %s", exc, exc_info=True)
+            return None
+
+    def translate(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str] = None,
+        context: Optional[str] = None,
+        prompt: Optional[str] = None,
+    ) -> Optional[str]:
+        if not text or not str(text).strip():
+            return text
+        if not self.client:
+            return None
+
+        tgt = (target_lang or "").lower() or "en"
+        src = (source_lang or "auto").lower()
+        scene = context or "default"
+        user_prompt = prompt or _build_prompt(
+            text=text,
+            source_lang=src,
+            target_lang=tgt,
+            scene=scene,
+        )
+        start = time.time()
+        try:
+            logger.info(
+                "[llm] Request | src=%s tgt=%s model=%s prompt=%s",
+                src,
+                tgt,
+                self.model,
+                user_prompt,
+            )
+            completion = self.client.chat.completions.create(
+                model=self.model,
+                messages=[{"role": "user", "content": user_prompt}],
+                timeout=self.timeout_sec,
+            )
+            content = (completion.choices[0].message.content or "").strip()
+            latency_ms = (time.time() - start) * 1000
+            if not content:
+                logger.warning("[llm] Empty result | src=%s tgt=%s latency=%.1fms", src, tgt, latency_ms)
+                return None
+            logger.info("[llm] Response | src=%s tgt=%s response=%s", src, tgt, content)
+            logger.info("[llm] Success | src=%s tgt=%s latency=%.1fms", src, tgt, latency_ms)
+            return content
+        except Exception as exc:
+            latency_ms = (time.time() - start) * 1000
+            logger.warning(
+                "[llm] Failed | src=%s tgt=%s latency=%.1fms error=%s",
+                src,
+                tgt,
+                latency_ms,
+                exc,
+                exc_info=True,
+            )
+            return None
+
+
 def llm_translate(
     text: str,
     target_lang: str,
@@ -139,100 +173,13 @@ def llm_translate(
     target_lang_label: Optional[str] = None,
     timeout_sec: Optional[float] = None,
 ) -> Optional[str]:
-    """
-    Translate text with Qwen chat model using rich prompts.
-
-    - 根据目标语言选择提示词，如果没匹配到则退回英文模板。
-    - 不对 text 做语言检测或缓存，调用方自行控制。
-
-    Args:
-        text: 原始文本
-        target_lang: 目标语言代码（如 "zh", "en"）
-        source_lang: 源语言代码（可选，不影响提示词选择，仅用于日志）
-        source_lang_label: 源语言展示名称，用于 prompt（默认使用 source_lang）
-        target_lang_label: 目标语言展示名称，用于 prompt（默认使用 target_lang）
-        timeout_sec: 请求超时时间（秒，可选；若未配置则从 config 读取或采用默认）
-
-    Returns:
-        翻译后的文本；如失败则返回 None。
-    """
-    if not text or not str(text).strip():
-        return text
-
-    cfg = get_translation_config()
-    provider_cfg = cfg.providers.get("llm", {}) if isinstance(cfg.providers, dict) else {}
-
-    model_name = provider_cfg.get("model") or QWEN_MODEL_NAME
-    req_timeout = float(provider_cfg.get("timeout_sec") or timeout_sec or 30.0)
-    base_url = (provider_cfg.get("base_url") or "").strip() or None
-
-    client = _get_qwen_client(base_url=base_url)
-    if not client:
-        # 无法调用云端，直接回退
-        logger.warning(
-            "[llm_translate] Client init failed; returning original text. "
-            "text=%r target_lang=%s source_lang=%s",
-            text[:80],
-            target_lang,
-            source_lang or "auto",
-        )
-        return text
-
-    tgt = (target_lang or "").lower() or "en"
-    src = (source_lang or "auto").lower()
-    src_label = source_lang_label or src
-    tgt_label = target_lang_label or tgt
-
-    prompt = _build_prompt(
+    provider = LLMTranslatorProvider(timeout_sec=timeout_sec or 30.0)
+    return provider.translate(
         text=text,
-        target_lang=tgt,
-        source_lang_label=src_label,
-        target_lang_label=tgt_label,
-        src_lang_code=src,
-        tgt_lang_code=tgt,
+        target_lang=target_lang,
+        source_lang=source_lang,
+        context=None,
     )
-    start = time.time()
-    try:
-        completion = client.chat.completions.create(
-            model=model_name,
-            messages=[
-                {
-                    "role": "user",
-                    "content": prompt,
-                }
-            ],
-            timeout=req_timeout,
-        )
-        content = (completion.choices[0].message.content or "").strip()
-        duration_ms = (time.time() - start) * 1000
-        logger.info(
-            "[llm_translate] Success | model=%s src=%s tgt=%s latency=%.1fms text=%r -> %r",
-            model_name,
-            src,
-            tgt,
-            duration_ms,
-            text[:80],
-            content[:80],
-        )
-        return content or text
-    except Exception as exc:
-        duration_ms = (time.time() - start) * 1000
-        logger.warning(
-            "[llm_translate] Failed | model=%s src=%s tgt=%s latency=%.1fms error=%s",
-            model_name,
-            src,
-            tgt,
-            duration_ms,
-            exc,
-            exc_info=True,
-        )
-        # 安全回退：出错时返回原文，避免中断上游流程
-        return text
-
-
-__all__ = [
-    "TRANSLATION_PROMPTS",
-    "llm_translate",
-]
+__all__ = ["LLMTranslatorProvider", "llm_translate"]
@@ -8,7 +8,7 @@ from typing import Dict, List, Optional, Any, Union
 import numpy as np
 import logging
 import re
-from concurrent.futures import Future, ThreadPoolExecutor, as_completed
+from concurrent.futures import ThreadPoolExecutor, as_completed, wait
 from embeddings.text_encoder import TextEmbeddingEncoder
 from config import SearchConfig
@@ -135,6 +135,7 @@ class QueryParser:
             cfg = get_translation_config()
             logger.info("Initializing translator at QueryParser construction (provider=%s)...", cfg.provider)
             self._translator = create_translation_provider(self.config.query_config)
+        self._translation_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="query-translation")
     @property
     def text_encoder(self) -> TextEmbeddingEncoder:
@@ -265,6 +266,7 @@ class QueryParser:
         # Stage 4: Translation (with async support and conditional waiting)
         translations = {}
         translation_futures = {}
+        translation_executor = None
         index_langs = ["en", "zh"]
         try:
             # 根据租户配置的 index_languages 决定翻译目标语言
@@ -287,48 +289,33 @@ class QueryParser:
                 target_langs = target_langs_for_translation
                 if target_langs:
-                    # Use e-commerce context for better disambiguation
-                    translation_context = self.config.query_config.translation_context
-                    # For query translation, we use a general prompt (not language-specific)
-                    query_prompt = (
-                        self.config.query_config.translation_prompts.get(f"query_{detected_lang}")
-                        or self.config.query_config.translation_prompts.get("query_en")
-                        or self.config.query_config.translation_prompts.get("default_en")
-                        or self.config.query_config.translation_prompts.get("default_zh")
-                    )
-                    
                     # Determine if we need to wait for translation results
                     # If detected_lang is not in index_languages, we must wait for translation
                     need_wait_translation = detected_lang not in index_langs
-                    
+
                     if need_wait_translation:
-                        # Use async method that returns Futures, so we can wait for results
-                        translation_results = self.translator.translate_multi_async(
-                            query_text,
-                            target_langs,
-                            source_lang=detected_lang,
-                            context=translation_context,
-                            prompt=query_prompt
+                        translation_executor = ThreadPoolExecutor(
+                            max_workers=max(1, min(len(target_langs), 4)),
+                            thread_name_prefix="query-translation-wait",
                         )
-                        # Separate cached results and futures
-                        for lang, result in translation_results.items():
-                            if isinstance(result, Future):
-                                translation_futures[lang] = result
-                            else:
-                                translations[lang] = result
+                        for lang in target_langs:
+                            translation_futures[lang] = translation_executor.submit(
+                                self.translator.translate,
+                                query_text,
+                                lang,
+                                detected_lang,
+                                "ecommerce_search_query",
+                            )
                     else:
-                        # Use async mode: returns cached translations immediately, missing ones translated in background
-                        translations = self.translator.translate_multi(
-                            query_text,
-                            target_langs,
-                            source_lang=detected_lang,
-                            context=translation_context,
-                            async_mode=True,
-                            prompt=query_prompt
-                        )
-                        # Filter out None values (missing translations that are being processed async)
-                        translations = {k: v for k, v in translations.items() if v is not None}
-                    
+                        for lang in target_langs:
+                            self._translation_executor.submit(
+                                self.translator.translate,
+                                query_text,
+                                lang,
+                                detected_lang,
+                                "ecommerce_search_query",
+                            )
+
                     if translations:
                         log_info(f"Translation completed (cache hit) | Query text: '{query_text}' | Results: {translations}")
                     if translation_futures:
@@ -407,15 +394,18 @@ class QueryParser:
                 all_futures.append(embedding_future)
                 future_to_lang[embedding_future] = ('embedding', None)
-            # Wait for all futures to complete
-            for future in as_completed(all_futures):
+            # Enforce a hard timeout for translation-related work (300ms budget)
+            done, not_done = wait(all_futures, timeout=0.3)
+            for future in done:
                 task_type, lang = future_to_lang[future]
                 try:
                     result = future.result()
                     if task_type == 'translation':
                         if result:
                             translations[lang] = result
-                            log_info(f"Translation completed | Query text: '{query_text}' | Target language: {lang} | Translation result: '{result}'")
+                            log_info(
+                                f"Translation completed | Query text: '{query_text}' | Target language: {lang} | Translation result: '{result}'"
+                            )
                             if context:
                                 context.store_intermediate_result(f'translation_{lang}', result)
                     elif task_type == 'embedding':
@@ -434,10 +424,27 @@ class QueryParser:
                     log_info(error_msg)
                     if context:
                         context.add_warning(error_msg)
-            
+
+            # Log timeouts for any futures that did not finish within 300ms
+            if not_done:
+                for future in not_done:
+                    task_type, lang = future_to_lang[future]
+                    if task_type == 'translation':
+                        timeout_msg = (
+                            f"Translation timeout (>300ms) | Language: {lang} | "
+                            f"Query text: '{query_text}'"
+                        )
+                    else:
+                        timeout_msg = "Query vector generation timeout (>300ms), proceeding without embedding result"
+                    log_info(timeout_msg)
+                    if context:
+                        context.add_warning(timeout_msg)
+
             # Clean up encoding executor
             if encoding_executor:
                 encoding_executor.shutdown(wait=False)
+            if translation_executor:
+                translation_executor.shutdown(wait=False)
             # Update translations in context after all are complete
             if translations and context:
-"""
-Translation service for multi-language query support.
+"""Qwen-MT translation orchestrator with cache and async helpers."""
-Supports multiple translation models:
-- Qwen (default): Alibaba Cloud DashScope API using qwen-mt-flash model
-- DeepL: DeepL API for high-quality translations
-
-重要说明（Qwen 机翻限速）:
-- 当前默认使用的 `qwen-mt-flash` 为云端机翻模型，**官方限速较低，约 RPM=60（每分钟约 60 请求）**
-- 在高并发场景必须依赖 Redis 翻译缓存与批量预热，避免在用户实时请求路径上直接打满 DashScope 限流
-- 若业务侧存在大规模离线翻译或更高吞吐需求，建议评估 DeepL 或自建翻译后端
-
-使用方法 (Usage):
-
-```python
-from query.translator import Translator
-
-# 使用默认的 qwen 模型（推荐）
-translator = Translator()  # 默认使用 qwen 模型
-
-# 或显式指定模型
-translator = Translator(model='qwen')  # 使用 qwen 模型
-translator = Translator(model='deepl')  # 使用 DeepL 模型
-
-# 翻译文本
-result = translator.translate(
-    text="我看到这个视频后没有笑",
-    target_lang="en",
-    source_lang="auto"  # 自动检测源语言
-)
-```
-
-配置说明 (Configuration):
-- Qwen 模型需要设置 DASHSCOPE_API_KEY 环境变量（在 .env 文件中）
-- DeepL 模型需要设置 DEEPL_AUTH_KEY 环境变量（在 .env 文件中）
-
-Qwen 模型参考文档：
-- 官方文档：https://help.aliyun.com/zh/model-studio/get-api-key
-- 模型：qwen-mt-flash（快速翻译模型）
-
-DeepL 官方文档：
-https://developers.deepl.com/api-reference/translate/request-translation
-"""
+from __future__ import annotations
+import hashlib
+import logging
 import os
-import requests
 import re
-import redis
-from concurrent.futures import ThreadPoolExecutor, Future
-from datetime import timedelta
-from typing import Dict, List, Optional, Union
-import logging
 import time
+from typing import Dict, List, Optional
-logger = logging.getLogger(__name__)
-
-from config.env_config import DEEPL_AUTH_KEY, DASHSCOPE_API_KEY, REDIS_CONFIG
+import redis
 from openai import OpenAI
+from config.env_config import DASHSCOPE_API_KEY, REDIS_CONFIG
+from config.services_config import get_translation_cache_config
+from config.translate_prompts import SOURCE_LANG_CODE_MAP
-class Translator:
-    """
-    Multi-language translator supporting Qwen and DeepL APIs.
-    
-    Default model is 'qwen' which uses Alibaba Cloud DashScope API.
-    """
-# 华北2（北京）：https://dashscope.aliyuncs.com/compatible-mode/v1
-# 新加坡：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
-# 美国（弗吉尼亚）：https://dashscope-us.aliyuncs.com/compatible-mode/v1
+logger = logging.getLogger(__name__)
-    DEEPL_API_URL = "https://api.deepl.com/v2/translate"  # Pro tier
-    QWEN_BASE_URL = "https://dashscope-us.aliyuncs.com/compatible-mode/v1"  # 北京地域
-    # QWEN_BASE_URL = "https://dashscope-intl.aliyuncs.com/compatible-mode/v1"  # 新加坡
-    # 如果使用新加坡地域的模型，需要将base_url替换为：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
-    QWEN_MODEL = "qwen-mt-flash"  # 快速翻译模型
-    # Language code mapping
-    LANG_CODE_MAP = {
-        'zh': 'ZH',
-        'en': 'EN',
-        'ru': 'RU',
-        'ar': 'AR',
-        'ja': 'JA',
-        'es': 'ES',
-        'de': 'DE',
-        'fr': 'FR',
-        'it': 'IT',
-        'pt': 'PT',
-    }
+class Translator:
+    QWEN_DEFAULT_BASE_URL = "https://dashscope-us.aliyuncs.com/compatible-mode/v1"
+    QWEN_MODEL = "qwen-mt-flash"
     def __init__(
         self,
@@ -95,77 +30,90 @@ class Translator:
         use_cache: bool = True,
         timeout: int = 10,
         glossary_id: Optional[str] = None,
-        translation_context: Optional[str] = None
+        translation_context: Optional[str] = None,
     ):
-        """
-        Initialize translator.
-
-        Args:
-            model: Translation model to use. Options: 'qwen' (default) or 'deepl'
-            api_key: API key for the selected model (or None to use from config/env)
-            use_cache: Whether to cache translations
-            timeout: Request timeout in seconds
-            glossary_id: DeepL glossary ID for custom terminology (optional, only for DeepL)
-            translation_context: Context hint for translation (e.g., "e-commerce", "product search")
-        """
-        self.model = model.lower()
-        if self.model not in ['qwen', 'deepl']:
-            raise ValueError(f"Unsupported model: {model}. Supported models: 'qwen', 'deepl'")
-        
-        # Get API key from config if not provided
-        if api_key is None:
-            if self.model == 'qwen':
-                api_key = DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY")
-            else:  # deepl
-                api_key = DEEPL_AUTH_KEY or os.getenv("DEEPL_AUTH_KEY")
-
-        self.api_key = api_key
-        self.timeout = timeout
-        self.use_cache = use_cache
+        self.model = self._normalize_model(model)
+        self.timeout = int(timeout)
+        self.use_cache = bool(use_cache)
         self.glossary_id = glossary_id
         self.translation_context = translation_context or "e-commerce product search"
-        
-        # Initialize OpenAI client for Qwen if needed
-        self.qwen_client = None
-        if self.model == 'qwen':
-            if not self.api_key:
-                logger.warning("DASHSCOPE_API_KEY not set. Qwen translation will not work.")
-            else:
-                self.qwen_client = OpenAI(
-                    api_key=self.api_key,
-                    base_url=self.QWEN_BASE_URL,
-                )
-        # Initialize Redis cache if enabled
-        if use_cache:
+        cache_cfg = get_translation_cache_config()
+        self.cache_prefix = str(cache_cfg.get("key_prefix", "trans:v2"))
+        self.expire_seconds = int(cache_cfg.get("ttl_seconds", 360 * 24 * 3600))
+        self.cache_sliding_expiration = bool(cache_cfg.get("sliding_expiration", True))
+        self.cache_include_context = bool(cache_cfg.get("key_include_context", True))
+        self.cache_include_prompt = bool(cache_cfg.get("key_include_prompt", True))
+        self.cache_include_source_lang = bool(cache_cfg.get("key_include_source_lang", True))
+
+        self.qwen_model_name = self._resolve_qwen_model_name(model)
+        self._api_key = api_key or self._default_api_key(self.model)
+        self._qwen_client: Optional[OpenAI] = None
+        base_url = os.getenv("DASHSCOPE_BASE_URL") or self.QWEN_DEFAULT_BASE_URL
+        if self._api_key:
             try:
-                self.redis_client = redis.Redis(
-                    host=REDIS_CONFIG.get('host', 'localhost'),
-                    port=REDIS_CONFIG.get('port', 6479),
-                    password=REDIS_CONFIG.get('password'),
-                    decode_responses=True,  # Return str instead of bytes
-                    socket_timeout=REDIS_CONFIG.get('socket_timeout', 1),
-                    socket_connect_timeout=REDIS_CONFIG.get('socket_connect_timeout', 1),
-                    retry_on_timeout=REDIS_CONFIG.get('retry_on_timeout', False),
-                    health_check_interval=10,  # 避免复用坏连接
-                )
-                # Test connection
-                self.redis_client.ping()
-                expire_days = REDIS_CONFIG.get('translation_cache_expire_days', 360)
-                self.expire_time = timedelta(days=expire_days)
-                self.expire_seconds = int(self.expire_time.total_seconds())  # Redis 需要秒数
-                self.cache_prefix = REDIS_CONFIG.get('translation_cache_prefix', 'trans')
-                logger.info("Redis cache initialized for translations")
-            except Exception as e:
-                logger.warning(f"Failed to initialize Redis cache: {e}, falling back to no cache")
-                self.redis_client = None
-                self.cache = None
+                self._qwen_client = OpenAI(api_key=self._api_key, base_url=base_url)
+            except Exception as exc:
+                logger.warning("Failed to initialize qwen-mt client: %s", exc, exc_info=True)
         else:
-            self.redis_client = None
-            self.cache = None
-        
-        # Thread pool for async translation
-        self.executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="translator")
+            logger.warning("DASHSCOPE_API_KEY not set; qwen-mt translation unavailable")
+
+        self.redis_client = None
+        if self.use_cache and bool(cache_cfg.get("enabled", True)):
+            self.redis_client = self._init_redis_client()
+
+    @staticmethod
+    def _normalize_model(model: str) -> str:
+        m = (model or "qwen").strip().lower()
+        if m.startswith("qwen"):
+            return "qwen-mt"
+        raise ValueError(f"Unsupported model: {model}. Supported models: 'qwen', 'qwen-mt', 'qwen-mt-flash'")
+
+    @staticmethod
+    def _resolve_qwen_model_name(model: str) -> str:
+        m = (model or "qwen").strip().lower()
+        if m in {"qwen", "qwen-mt"}:
+            return "qwen-mt-flash"
+        return m
+
+    @staticmethod
+    def _default_api_key(model: str) -> Optional[str]:
+        del model
+        return DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY")
+
+    def _init_redis_client(self):
+        try:
+            client = redis.Redis(
+                host=REDIS_CONFIG.get("host", "localhost"),
+                port=REDIS_CONFIG.get("port", 6479),
+                password=REDIS_CONFIG.get("password"),
+                decode_responses=True,
+                socket_timeout=REDIS_CONFIG.get("socket_timeout", 1),
+                socket_connect_timeout=REDIS_CONFIG.get("socket_connect_timeout", 1),
+                retry_on_timeout=REDIS_CONFIG.get("retry_on_timeout", False),
+                health_check_interval=10,
+            )
+            client.ping()
+            return client
+        except Exception as exc:
+            logger.warning("Failed to initialize translation redis cache: %s", exc)
+            return None
+
+    def _build_cache_key(
+        self,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str],
+        context: Optional[str],
+        prompt: Optional[str],
+    ) -> str:
+        src = (source_lang or "auto").strip().lower() if self.cache_include_source_lang else "-"
+        tgt = (target_lang or "").strip().lower()
+        ctx = (context or "").strip() if self.cache_include_context else ""
+        prm = (prompt or "").strip() if self.cache_include_prompt else ""
+        payload = f"model={self.model}\nsrc={src}\ntgt={tgt}\nctx={ctx}\nprm={prm}\ntext={text}"
+        digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()
+        return f"{self.cache_prefix}:{self.model}:{src}:{tgt}:{digest}"
     def translate(
         self,
@@ -173,99 +121,27 @@ class Translator:
         target_lang: str,
         source_lang: Optional[str] = None,
         context: Optional[str] = None,
-        prompt: Optional[str] = None
+        prompt: Optional[str] = None,
     ) -> Optional[str]:
-        """
-        Translate text to target language (synchronous mode).
-
-        Args:
-            text: Text to translate
-            target_lang: Target language code ('zh', 'en', 'ru', etc.)
-            source_lang: Source language code (option al, auto-detect if None)
-            context: Additional context for translation (overrides default context)
-            prompt: Translation prompt/instruction (optional, for better translation quality)
-
-        Returns:
-            Translated text or None if translation fails
-        """
         if not text or not text.strip():
             return text
-        # Normalize language codes
-        target_lang = target_lang.lower()
-        if source_lang:
-            source_lang = source_lang.lower()
-
-        # Optimization: Skip translation if not needed
-        if target_lang == 'en' and self._is_english_text(text):
-            logger.info(f"[Translator] Text is already English, skipping translation: '{text[:50]}...'")
+        tgt = (target_lang or "").strip().lower()
+        src = (source_lang or "").strip().lower() or None
+        if tgt == "en" and self._is_english_text(text):
             return text
-        
-        if target_lang == 'zh' and (self._contains_chinese(text) or self._is_pure_number(text)):
-            logger.info(
-                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
-                f"Source language: {source_lang or 'auto'} | Result: Skip translation (contains Chinese or pure number)"
-            )
+        if tgt == "zh" and (self._contains_chinese(text) or self._is_pure_number(text)):
             return text
-        # Use provided context or default context
         translation_context = context or self.translation_context
-        
-        # Build cache key (include prompt in cache key if provided)
-        cache_key_parts = [source_lang or 'auto', target_lang, translation_context]
-        if prompt:
-            cache_key_parts.append(prompt)
-        cache_key_parts.append(text)
-        cache_key = ':'.join(cache_key_parts)
+        cached = self._get_cached_translation_redis(text, tgt, src, translation_context, prompt)
+        if cached is not None:
+            return cached
-        # Check cache (include context and prompt in cache key for accuracy)
-        if self.use_cache and self.redis_client:
-            cached = self._get_cached_translation_redis(text, target_lang, source_lang, translation_context, prompt)
-            if cached:
-                logger.info(
-                    f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
-                    f"Source language: {source_lang or 'auto'} | Result: '{cached}' | Source: Cache hit"
-                )
-                return cached
-
-        # If no API key, return mock translation (for testing)
-        if not self.api_key:
-            logger.info(
-                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
-                f"Source language: {source_lang or 'auto'} | Result: '{text}' | Source: Mock mode (no API key)"
-            )
-            return text
-
-        # Translate using selected model
-        logger.info(
-            f"[Translator] Translation request | Model: {self.model} | Original text: '{text}' | Target language: {target_lang} | "
-            f"Source language: {source_lang or 'auto'} | Context: {translation_context} | "
-            f"Prompt: {'yes' if prompt else 'no'} | Status: Starting translation"
-        )
-        
-        if self.model == 'qwen':
-            result = self._translate_qwen(text, target_lang, source_lang, translation_context, prompt)
-        else:  # deepl
-            result = self._translate_deepl(text, target_lang, source_lang, translation_context, prompt)
-
-        # Surface translation failure to the caller instead of silently
-        # masquerading the source text as a successful translation.
-        if result is None:
-            logger.warning(
-                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
-                f"Source language: {source_lang or 'auto'} | Status: Translation failed"
-            )
-        else:
-            logger.info(
-                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
-                f"Source language: {source_lang or 'auto'} | Result: '{result}' | Status: Translation successful"
-            )
-
-        # Cache only successful translations. Failed attempts must not poison
-        # Redis with the original text.
-        if result is not None and self.use_cache and self.redis_client:
-            self._set_cached_translation_redis(text, target_lang, result, source_lang, translation_context, prompt)
+        result = self._translate_qwen(text, tgt, src)
+        if result is not None:
+            self._set_cached_translation_redis(text, tgt, result, src, translation_context, prompt)
         return result
     def _translate_qwen(
@@ -273,412 +149,63 @@ class Translator:
         text: str,
         target_lang: str,
         source_lang: Optional[str],
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
     ) -> Optional[str]:
-        """
-        Translate using Qwen MT Flash model via Alibaba Cloud DashScope API.
-        
-        Args:
-            text: Text to translate
-            target_lang: Target language code ('zh', 'en', 'ru', etc.)
-            source_lang: Source language code (optional, 'auto' if None)
-            context: Context hint for translation (optional)
-            prompt: Translation prompt/instruction (optional)
-            
-        Returns:
-            Translated text or None if translation fails
-        """
-        if not self.qwen_client:
-            logger.error("[Translator] Qwen client not initialized. Check DASHSCOPE_API_KEY.")
+        if not self._qwen_client:
             return None
-        
-        # Qwen (qwen-mt-plus/flash/turbo) supported languages mapping
-        # 标准来自：你提供的“语言 / 英文名 / 代码”表
-        qwen_lang_map = {
-            "en": "English",
-            "zh": "Chinese",
-            "zh_tw": "Traditional Chinese",
-            "ru": "Russian",
-            "ja": "Japanese",
-            "ko": "Korean",
-            "es": "Spanish",
-            "fr": "French",
-            "pt": "Portuguese",
-            "de": "German",
-            "it": "Italian",
-            "th": "Thai",
-            "vi": "Vietnamese",
-            "id": "Indonesian",
-            "ms": "Malay",
-            "ar": "Arabic",
-            "hi": "Hindi",
-            "he": "Hebrew",
-            "my": "Burmese",
-            "ta": "Tamil",
-            "ur": "Urdu",
-            "bn": "Bengali",
-            "pl": "Polish",
-            "nl": "Dutch",
-            "ro": "Romanian",
-            "tr": "Turkish",
-            "km": "Khmer",
-            "lo": "Lao",
-            "yue": "Cantonese",
-            "cs": "Czech",
-            "el": "Greek",
-            "sv": "Swedish",
-            "hu": "Hungarian",
-            "da": "Danish",
-            "fi": "Finnish",
-            "uk": "Ukrainian",
-            "bg": "Bulgarian",
-        }
-        
-        # Convert target language
-        target_lang_normalized = target_lang.lower()
-        target_lang_qwen = qwen_lang_map.get(target_lang_normalized, target_lang.capitalize())
-
-        # Convert source language
-        source_lang_normalized = (source_lang or "").strip().lower()
-        if not source_lang_normalized or source_lang_normalized == "auto":
-            source_lang_qwen = "auto"
-        else:
-            source_lang_qwen = qwen_lang_map.get(source_lang_normalized, source_lang.capitalize())
-        
-        # Prepare translation options
-        translation_options = {
-            "source_lang": source_lang_qwen,
-            "target_lang": target_lang_qwen,
-        }
-        
-        # Prepare messages
-        messages = [
-            {
-                "role": "user",
-                "content": text
-            }
-        ]
-        
-        start_time = time.time()
+        tgt_norm = (target_lang or "").strip().lower()
+        src_norm = (source_lang or "").strip().lower()
+        tgt_qwen = self.SOURCE_LANG_CODE_MAP.get(tgt_norm, tgt_norm.capitalize())
+        src_qwen = "auto" if not src_norm or src_norm == "auto" else self.SOURCE_LANG_CODE_MAP.get(src_norm, src_norm.capitalize())
+        start = time.time()
         try:
-            completion = self.qwen_client.chat.completions.create(
-                model=self.QWEN_MODEL,
-                messages=messages,
+            completion = self._qwen_client.chat.completions.create(
+                model=self.qwen_model_name,
+                messages=[{"role": "user", "content": text}],
                 extra_body={
-                    "translation_options": translation_options
-                }
-            )
-            
-            translated_text = completion.choices[0].message.content.strip()
-            duration_ms = (time.time() - start_time) * 1000
-            
-            logger.info(
-                f"[Translator] Qwen API response success | Original text: '{text}' | Target language: {target_lang_qwen} | "
-                f"Translation result: '{translated_text}' | Duration: {duration_ms:.2f} ms"
-            )
-            return translated_text
-            
-        except Exception as e:
-            duration_ms = (time.time() - start_time) * 1000
-            logger.error(
-                f"[Translator] Qwen API request exception | Original text: '{text}' | Target language: {target_lang_qwen} | "
-                f"Duration: {duration_ms:.2f} ms | Error: {e}", exc_info=True
-            )
-            return None
-
-    def _translate_deepl(
-        self,
-        text: str,
-        target_lang: str,
-        source_lang: Optional[str],
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ) -> Optional[str]:
-        """
-        Translate using DeepL API with context and glossary support.
-        
-        Args:
-            text: Text to translate
-            target_lang: Target language code
-            source_lang: Source language code (optional)
-            context: Context hint for translation (e.g., "e-commerce product search")
-        """
-        # Map to DeepL language codes
-        target_code = self.LANG_CODE_MAP.get(target_lang, target_lang.upper())
-
-        headers = {
-            "Authorization": f"DeepL-Auth-Key {self.api_key}",
-            "Content-Type": "application/json",
-        }
-
-        # Use prompt as context parameter for DeepL API (not as text prefix)
-        # According to DeepL API: context is "Additional context that can influence a translation but is not translated itself"
-        # If prompt is provided, use it as context; otherwise use the default context
-        api_context = prompt if prompt else context
-        
-        # For e-commerce, add context words to help DeepL understand the domain
-        # This is especially important for single-word ambiguous terms like "车" (car vs rook)
-        text_to_translate, needs_extraction = self._add_ecommerce_context(text, source_lang, api_context)
-
-        payload = {
-            "text": [text_to_translate],
-            "target_lang": target_code,
-        }
-
-        if source_lang:
-            source_code = self.LANG_CODE_MAP.get(source_lang, source_lang.upper())
-            payload["source_lang"] = source_code
-
-        # Add context parameter (prompt or default context)
-        # Context influences translation but is not translated itself
-        if api_context:
-            payload["context"] = api_context
-
-        # Add glossary if configured
-        if self.glossary_id:
-            payload["glossary_id"] = self.glossary_id
-
-        # Note: DeepL API v2 supports "context" parameter for additional context
-        # that influences translation but is not translated itself.
-        # We use prompt as context parameter when provided.
-
-        try:
-            response = requests.post(
-                self.DEEPL_API_URL,
-                headers=headers,
-                json=payload,
-                timeout=self.timeout
+                    "translation_options": {
+                        "source_lang": src_qwen,
+                        "target_lang": tgt_qwen,
+                    }
+                },
+                timeout=self.timeout,
             )
-
-            if response.status_code == 200:
-                data = response.json()
-                if "translations" in data and len(data["translations"]) > 0:
-                    translated_text = data["translations"][0]["text"]
-                    # If we added context, extract just the term from the result
-                    if needs_extraction:
-                        translated_text = self._extract_term_from_translation(
-                            translated_text, text, target_code
-                        )
-                    logger.debug(
-                        f"[Translator] DeepL API response success | Original text: '{text}' | Target language: {target_code} | "
-                        f"Translation result: '{translated_text}'"
-                    )
-                    return translated_text
-            else:
-                logger.error(
-                    f"[Translator] DeepL API error | Original text: '{text}' | Target language: {target_code} | "
-                    f"Status code: {response.status_code} | Error message: {response.text}"
-                )
+            content = (completion.choices[0].message.content or "").strip()
+            if not content:
                 return None
-
-        except requests.Timeout:
+            logger.info("[qwen-mt] Success | src=%s tgt=%s latency=%.1fms", src_qwen, tgt_qwen, (time.time() - start) * 1000)
+            return content
+        except Exception as exc:
             logger.warning(
-                f"[Translator] DeepL API request timeout | Original text: '{text}' | Target language: {target_code} | "
-                f"Timeout: {self.timeout}s"
-            )
-            return None
-        except Exception as e:
-            logger.error(
-                f"[Translator] DeepL API request exception | Original text: '{text}' | Target language: {target_code} | "
-                f"Error: {e}", exc_info=True
+                "[qwen-mt] Failed | src=%s tgt=%s latency=%.1fms error=%s",
+                src_qwen,
+                tgt_qwen,
+                (time.time() - start) * 1000,
+                exc,
+                exc_info=True,
             )
             return None
-    # NOTE: _translate_deepl_free is intentionally not implemented.
-    # We do not support automatic fallback to the free endpoint, to avoid
-    # mixing Pro keys with https://api-free.deepl.com and related 403 errors.
-
-    def translate_multi(
-        self,
-        text: str,
-        target_langs: List[str],
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        async_mode: bool = True,
-        prompt: Optional[str] = None
-    ) -> Dict[str, Optional[str]]:
-        """
-        Translate text to multiple target languages.
-        
-        In async_mode=True (default):
-        - Returns cached translations immediately if available
-        - For translations that can be optimized (e.g., pure numbers, already in target language),
-          returns result immediately via synchronous call
-        - Launches async tasks for other missing translations (non-blocking)
-        - Returns None for missing translations that require async processing
-        
-        In async_mode=False:
-        - Waits for all translations to complete (blocking)
-
-        Args:
-            text: Text to translate
-            target_langs: List of target language codes
-            source_lang: Source language code (optional)
-            context: Context hint for translation (optional)
-            async_mode: If True, return cached results immediately and translate missing ones async
-            prompt: Translation prompt/instruction (optional)
-        Returns:
-            Dictionary mapping language code to translated text (only cached results in async mode)
-        """
-        results = {}
-        missing_langs = []
-        async_langs = []
-        
-        # First, get cached translations
-        for lang in target_langs:
-            cached = self._get_cached_translation(text, lang, source_lang, context, prompt)
-            if cached is not None:
-                results[lang] = cached
-            else:
-                missing_langs.append(lang)
-        
-        # If async mode and there are missing translations
-        if async_mode and missing_langs:
-            # Check if translation can be optimized (immediate return)
-            for lang in missing_langs:
-                target_lang = lang.lower()
-                # Check optimization conditions (same as in translate method)
-                can_optimize = False
-                if target_lang == 'en' and self._is_english_text(text):
-                    can_optimize = True
-                elif target_lang == 'zh' and (self._contains_chinese(text) or self._is_pure_number(text)):
-                    can_optimize = True
-                
-                if can_optimize:
-                    # Can be optimized, call translate synchronously for immediate result
-                    results[lang] = self.translate(text, lang, source_lang, context, prompt)
-                else:
-                    # Requires actual translation, add to async list
-                    async_langs.append(lang)
-            
-            # Launch async tasks for translations that require actual API calls
-            if async_langs:
-                for lang in async_langs:
-                    self._translate_async(text, lang, source_lang, context, prompt)
-                # Return None for async translations
-                for lang in async_langs:
-                    results[lang] = None
-        else:
-            # Synchronous mode: wait for all translations
-            for lang in missing_langs:
-                results[lang] = self.translate(text, lang, source_lang, context, prompt)
-        
-        return results
-    
-    def translate_multi_async(
-        self,
-        text: str,
-        target_langs: List[str],
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ) -> Dict[str, Union[str, Future]]:
-        """
-        Translate text to multiple target languages asynchronously, returning Futures that can be awaited.
-        
-        This method returns a dictionary where:
-        - If translation is cached, the value is the translation string (immediate)
-        - If translation needs to be done, the value is a Future object that can be awaited
-        
-        Args:
-            text: Text to translate
-            target_langs: List of target language codes
-            source_lang: Source language code (optional)
-            context: Context hint for translation (optional)
-            prompt: Translation prompt/instruction (optional)
-
-        Returns:
-            Dictionary mapping language code to either translation string (cached) or Future object
-        """
-        results = {}
-        missing_langs = []
-        
-        # First, get cached translations
-        for lang in target_langs:
-            cached = self._get_cached_translation(text, lang, source_lang, context, prompt)
-            if cached is not None:
-                results[lang] = cached
-            else:
-                missing_langs.append(lang)
-        
-        # For missing translations, submit async tasks and return Futures
-        for lang in missing_langs:
-            future = self.executor.submit(
-                self.translate,
-                text,
-                lang,
-                source_lang,
-                context,
-                prompt
-            )
-            results[lang] = future
-        
-        return results
-    
-    def _get_cached_translation(
-        self,
-        text: str,
-        target_lang: str,
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ) -> Optional[str]:
-        """Get translation from cache if available."""
-        if not self.redis_client:
-            return None
-        return self._get_cached_translation_redis(text, target_lang, source_lang, context, prompt)
-    
     def _get_cached_translation_redis(
         self,
         text: str,
         target_lang: str,
         source_lang: Optional[str] = None,
         context: Optional[str] = None,
-        prompt: Optional[str] = None
+        prompt: Optional[str] = None,
     ) -> Optional[str]:
-        """
-        Get translation from Redis cache with sliding expiration.
-        
-        滑动过期机制：每次访问缓存时，重置过期时间为配置的过期时间（默认720天）。
-        这样缓存会在最后一次访问后的720天才过期，而不是写入后的720天。
-        这确保了常用的翻译缓存不会被过早删除。
-        """
         if not self.redis_client:
             return None
-        
+        key = self._build_cache_key(text, target_lang, source_lang, context, prompt)
         try:
-            # Build cache key: prefix:target_lang:text
-            # For simplicity, we use target_lang and text as key
-            # Context and prompt are not included in key to maximize cache hits
-            cache_key = f"{self.cache_prefix}:{target_lang.upper()}:{text}"
-            value = self.redis_client.get(cache_key)
-            if value:
-                # Sliding expiration: reset expiration time on access
-                # 每次读取缓存时，重置过期时间为配置的过期时间（最后一次访问后的N天才过期）
-                try:
-                    self.redis_client.expire(cache_key, self.expire_seconds)
-                except Exception as expire_error:
-                    # 即使 expire 失败，也返回缓存值（不影响功能）
-                    logger.warning(
-                        f"[Translator] Failed to update cache expiration for key {cache_key}: {expire_error}"
-                    )
-                
-                logger.debug(
-                    f"[Translator] Redis cache hit | Original text: '{text}' | Target language: {target_lang} | "
-                    f"Cache key: {cache_key} | Translation result: '{value}' | TTL reset to {self.expire_seconds}s"
-                )
-                return value
-            logger.debug(
-                f"[Translator] Redis cache miss | Original text: '{text}' | Target language: {target_lang} | "
-                f"Cache key: {cache_key}"
-            )
+            value = self.redis_client.get(key)
+            if value and self.cache_sliding_expiration:
+                self.redis_client.expire(key, self.expire_seconds)
+            return value
+        except Exception as exc:
+            logger.warning("Redis get translation cache failed: %s", exc)
             return None
-        except Exception as e:
-            logger.error(f"[Translator] Redis error during get translation cache | Original text: '{text}' | Target language: {target_lang} | Error: {e}")
-            return None
-    
+
     def _set_cached_translation_redis(
         self,
         text: str,
@@ -686,128 +213,17 @@ class Translator:
         translation: str,
         source_lang: Optional[str] = None,
         context: Optional[str] = None,
-        prompt: Optional[str] = None
+        prompt: Optional[str] = None,
     ) -> None:
-        """Store translation in Redis cache."""
         if not self.redis_client:
             return
-        
+        key = self._build_cache_key(text, target_lang, source_lang, context, prompt)
         try:
-            cache_key = f"{self.cache_prefix}:{target_lang.upper()}:{text}"
-            self.redis_client.setex(cache_key, self.expire_seconds, translation)
-            logger.info(
-                f"[Translator] Redis cache write | Original text: '{text}' | Target language: {target_lang} | "
-                f"Cache key: {cache_key} | Translation result: '{translation}'"
-            )
-        except Exception as e:
-            logger.error(
-                f"[Translator] Redis cache write failed | Original text: '{text}' | Target language: {target_lang} | "
-                f"Error: {e}"
-            )
-    
-    def _translate_async(
-        self,
-        text: str,
-        target_lang: str,
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ):
-        """Launch async translation task."""
-        def _do_translate():
-            try:
-                result = self.translate(text, target_lang, source_lang, context, prompt)
-                if result:
-                    logger.debug(f"Async translation completed: {text} -> {target_lang}: {result}")
-            except Exception as e:
-                logger.warning(f"Async translation failed: {text} -> {target_lang}: {e}")
-        
-        self.executor.submit(_do_translate)
-
-    def _add_ecommerce_context(
-        self,
-        text: str,
-        source_lang: Optional[str],
-        context: Optional[str]
-    ) -> tuple:
-        """
-        Add e-commerce context to text for better disambiguation.
-        
-        For single-word ambiguous Chinese terms, we add context words that help
-        DeepL understand this is an e-commerce/product search context.
-        
-        Args:
-            text: Original text to translate
-            source_lang: Source language code
-            context: Context hint
-            
-        Returns:
-            Tuple of (text_with_context, needs_extraction)
-            - text_with_context: Text to send to DeepL
-            - needs_extraction: Whether we need to extract the term from the result
-        """
-        # Only apply for e-commerce context and Chinese source
-        if not context or "e-commerce" not in context.lower():
-            return text, False
-            
-        if not source_lang or source_lang.lower() != 'zh':
-            return text, False
-            
-        # For single-word queries, add context to help disambiguation
-        text_stripped = text.strip()
-        if len(text_stripped.split()) == 1 and len(text_stripped) <= 2:
-            # Common ambiguous Chinese e-commerce terms like "车" (car vs rook)
-            # We add a context phrase: "购买 [term]" (buy [term]) or "商品 [term]" (product [term])
-            # This helps DeepL understand the e-commerce context
-            # We'll need to extract just the term from the translation result
-            context_phrase = f"购买 {text_stripped}"
-            return context_phrase, True
-        
-        # For multi-word queries, DeepL usually has enough context
-        return text, False
-
-    def _extract_term_from_translation(
-        self,
-        translated_text: str,
-        original_text: str,
-        target_lang_code: str
-    ) -> str:
-        """
-        Extract the actual term from a translation that included context.
-        
-        For example, if we translated "购买 车" (buy car) and got "buy car",
-        we want to extract just "car".
-        
-        Args:
-            translated_text: Full translation result
-            original_text: Original single-word query
-            target_lang_code: Target language code (EN, ZH, etc.)
-            
-        Returns:
-            Extracted term or original translation if extraction fails
-        """
-        # For English target, try to extract the last word (the actual term)
-        if target_lang_code == "EN":
-            words = translated_text.strip().split()
-            if len(words) > 1:
-                # Usually the last word is the term we want
-                # But we need to be smart - if it's "buy car", we want "car"
-                # Common context words to skip: buy, purchase, product, item, etc.
-                context_words = {"buy", "purchase", "product", "item", "commodity", "goods"}
-                # Try to find the term (not a context word)
-                for word in reversed(words):
-                    word_lower = word.lower().rstrip('.,!?;:')
-                    if word_lower not in context_words:
-                        return word_lower
-                # If all words are context words, return the last one
-                return words[-1].lower().rstrip('.,!?;:')
-        
-        # For other languages or if extraction fails, return as-is
-        # The user can configure a glossary for better results
-        return translated_text
+            self.redis_client.setex(key, self.expire_seconds, translation)
+        except Exception as exc:
+            logger.warning("Redis set translation cache failed: %s", exc)
     def _shop_lang_matches(self, shop_lang_lower: str, lang_code: str) -> bool:
-        """True if shop language matches index language (use source, no translate)."""
         if not shop_lang_lower or not lang_code:
             return False
         if shop_lang_lower == lang_code:
@@ -818,146 +234,27 @@ class Translator:
             return True
         return False
-    def translate_for_indexing(
-        self,
-        text: str,
-        shop_language: str,
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None,
-        index_languages: Optional[List[str]] = None,
-    ) -> Dict[str, Optional[str]]:
-        """
-        Translate text for indexing based on shop language and tenant index_languages.
-
-        For each language in index_languages: use source text if shop language matches,
-        otherwise translate to that language.
-
-        Args:
-            text: Text to translate
-            shop_language: Shop primary language (e.g. 'zh', 'en', 'ru')
-            source_lang: Source language code (optional)
-            context: Additional context for translation (optional)
-            prompt: Translation prompt (optional)
-            index_languages: Languages to index (from tenant_config). Default ["en", "zh"].
-
-        Returns:
-            Dict keyed by each index_language with translated or source text (or None).
-        """
-        langs = index_languages if index_languages else ["en", "zh"]
-        results = {lang: None for lang in langs}
-        if not text or not text.strip():
-            return results
-        if re.match(r'^[\d\s_-]+$', text):
-            logger.info(f"[Translator] Skip translation for symbol-only query: '{text}'")
-            return results
-
-        shop_lang_lower = (shop_language or "").strip().lower()
-        targets = []
-        for lang in langs:
-            if self._shop_lang_matches(shop_lang_lower, lang):
-                results[lang] = text
-            else:
-                targets.append(lang)
-
-        for target_lang in targets:
-            cached = self._get_cached_translation_redis(text, target_lang, source_lang, context, prompt)
-            if cached:
-                results[target_lang] = cached
-                logger.debug(f"[Translator] Cache hit for indexing: '{text}' -> {target_lang}: {cached}")
-                continue
-            translated = self.translate(
-                text,
-                target_lang=target_lang,
-                source_lang=source_lang or shop_language,
-                context=context,
-                prompt=prompt,
-            )
-            results[target_lang] = translated
-        return results
-
-    def get_translation_needs(
-        self,
-        detected_lang: str,
-        supported_langs: List[str]
-    ) -> List[str]:
-        """
-        Determine which languages need translation.
-
-        Args:
-            detected_lang: Detected query language
-            supported_langs: List of supported languages
-
-        Returns:
-            List of language codes to translate to
-        """
-        # If detected language is in supported list, translate to others
+    def get_translation_needs(self, detected_lang: str, supported_langs: List[str]) -> List[str]:
         if detected_lang in supported_langs:
-            return [lang for lang in supported_langs if detected_lang != lang]
-
-        # Otherwise, translate to all supported languages
+            return [lang for lang in supported_langs if lang != detected_lang]
         return supported_langs
-    
+
     def _is_english_text(self, text: str) -> bool:
-        """
-        Check if text is primarily English (ASCII letters, numbers, common punctuation).
-        
-        Args:
-            text: Text to check
-            
-        Returns:
-            True if text appears to be English
-        """
         if not text or not text.strip():
             return True
-        
-        # Remove whitespace and common punctuation
-        text_clean = re.sub(r'[\s\.,!?;:\-\'\"\(\)\[\]{}]', '', text)
+        text_clean = re.sub(r"[\s\.,!?;:\-\'\"\(\)\[\]{}]", "", text)
         if not text_clean:
             return True
-        
-        # Check if all remaining characters are ASCII (letters, numbers)
-        # This is a simple heuristic: if most characters are ASCII, it's likely English
         ascii_count = sum(1 for c in text_clean if ord(c) < 128)
-        ratio = ascii_count / len(text_clean) if text_clean else 0
-        
-        # If more than 80% are ASCII characters, consider it English
-        return ratio > 0.8
-    
+        return (ascii_count / len(text_clean)) > 0.8
+
     def _contains_chinese(self, text: str) -> bool:
-        """
-        Check if text contains Chinese characters (Han characters).
-        
-        Args:
-            text: Text to check
-            
-        Returns:
-            True if text contains Chinese characters
-        """
         if not text:
             return False
-        
-        # Check for Chinese characters (Unicode range: \u4e00-\u9fff)
-        chinese_pattern = re.compile(r'[\u4e00-\u9fff]')
-        return bool(chinese_pattern.search(text))
-    
+        return bool(re.search(r"[\u4e00-\u9fff]", text))
+
     def _is_pure_number(self, text: str) -> bool:
-        """
-        Check if text is purely numeric (digits, possibly with spaces, dots, commas).
-        
-        Args:
-            text: Text to check
-            
-        Returns:
-            True if text is purely numeric
-        """
         if not text or not text.strip():
             return False
-        
-        # Remove whitespace, dots, commas (common number separators)
-        text_clean = re.sub(r'[\s\.,]', '', text.strip())
-        if not text_clean:
-            return False
-        
-        # Check if all remaining characters are digits
-        return text_clean.isdigit()
+        text_clean = re.sub(r"[\s\.,]", "", text.strip())
+        return bool(text_clean) and text_clean.isdigit()
@@ -14,6 +14,7 @@ Test content:
 import sys
 import os
 from pathlib import Path
+from concurrent.futures import ThreadPoolExecutor
 # Add parent directory to path
 sys.path.insert(0, str(Path(__file__).parent.parent))
@@ -42,9 +43,6 @@ def test_config_loading():
         print(f"✓ Configuration loaded successfully")
         print(f"  Translation service: {config.query_config.translation_service}")
-        print(f"  Translation prompt configuration:")
-        for key, value in config.query_config.translation_prompts.items():
-            print(f"    {key}: {value[:60]}..." if len(value) > 60 else f"    {key}: {value}")
         return config
     except Exception as e:
@@ -72,34 +70,23 @@ def test_translator_sync(config):
             translation_context=config.query_config.translation_context
         )
-        # 测试商品标题翻译（使用product_title提示词）
+        # 测试商品标题翻译（使用sku_name提示词）
         test_texts = [
-            ("蓝牙耳机", "zh", "en", "product_title"),
-            ("Wireless Headphones", "en", "zh", "product_title"),
+            ("蓝牙耳机", "zh", "en", "sku_name"),
+            ("Wireless Headphones", "en", "zh", "sku_name"),
         ]
-        for text, source_lang, target_lang, prompt_type in test_texts:
-            if prompt_type == "product_title":
-                if target_lang == "zh":
-                    prompt = config.query_config.translation_prompts.get('product_title_zh')
-                else:
-                    prompt = config.query_config.translation_prompts.get('product_title_en')
-            else:
-                if target_lang == "zh":
-                    prompt = config.query_config.translation_prompts.get('default_zh')
-                else:
-                    prompt = config.query_config.translation_prompts.get('default_en')
-            
+        for text, source_lang, target_lang, scene in test_texts:
             print(f"\nTranslation test:")
             print(f"  Original text ({source_lang}): {text}")
             print(f"  Target language: {target_lang}")
-            print(f"  Prompt: {prompt[:50] if prompt else 'None'}...")
+            print(f"  Scene: {scene}")
             result = translator.translate(
                 text,
                 target_lang=target_lang,
                 source_lang=source_lang,
-                prompt=prompt
+                context=scene,
             )
             if result:
@@ -131,43 +118,25 @@ def test_translator_async(config, translator):
         query_text = "手机"
         target_langs = ['en']
         source_lang = 'zh'
-        
-        query_prompt = config.query_config.translation_prompts.get('query_zh')
-        
+
         print(f"Query text: {query_text}")
         print(f"Target languages: {target_langs}")
-        print(f"Prompt: {query_prompt}")
-        
-        # 异步模式（立即返回，后台翻译）
-        results = translator.translate_multi(
-            query_text,
-            target_langs,
-            source_lang=source_lang,
-            context=config.query_config.translation_context,
-            async_mode=True,
-            prompt=query_prompt
-        )
-        
-        print(f"\nAsynchronous translation results:")
-        for lang, translation in results.items():
-            if translation:
-                print(f"  {lang}: {translation} (cache hit)")
-            else:
-                print(f"  {lang}: None (translating in background...)")
-        
-        # 同步模式（等待完成）
-        print(f"\nSynchronous translation (waiting for completion):")
-        results_sync = translator.translate_multi(
-            query_text,
-            target_langs,
-            source_lang=source_lang,
-            context=config.query_config.translation_context,
-            async_mode=False,
-            prompt=query_prompt
-        )
+        print("Scene: ecommerce_search_query")
-        for lang, translation in results_sync.items():
-            print(f"  {lang}: {translation}")
+        print(f"\nConcurrent translation via generic translate():")
+        with ThreadPoolExecutor(max_workers=len(target_langs)) as executor:
+            futures = {
+                lang: executor.submit(
+                    translator.translate,
+                    query_text,
+                    lang,
+                    source_lang,
+                    "ecommerce_search_query",
+                )
+                for lang in target_langs
+            }
+            for lang, future in futures.items():
+                print(f"  {lang}: {future.result()}")
     except Exception as e:
         print(f"✗ Asynchronous translation test failed: {e}")
@@ -193,14 +162,13 @@ def test_cache():
         test_text = "测试文本"
         target_lang = "en"
         source_lang = "zh"
-        prompt = config.query_config.translation_prompts.get('default_zh')
         print(f"First translation (should call API or return mock):")
-        result1 = translator.translate(test_text, target_lang, source_lang, prompt=prompt)
+        result1 = translator.translate(test_text, target_lang, source_lang, context="default")
         print(f"  Result: {result1}")
         print(f"\nSecond translation (should use cache):")
-        result2 = translator.translate(test_text, target_lang, source_lang, prompt=prompt)
+        result2 = translator.translate(test_text, target_lang, source_lang, context="default")
         print(f"  Result: {result2}")
         if result1 == result2:
@@ -231,17 +199,16 @@ def test_context_parameter():
         # 测试带context和不带context的翻译
         text = "手机"
-        prompt = config.query_config.translation_prompts.get('query_zh')
         print(f"Test text: {text}")
-        print(f"Prompt (as context): {prompt}")
+        print("Scene: ecommerce_search_query")
         # 带context的翻译
         result_with_context = translator.translate(
             text,
             target_lang='en',
             source_lang='zh',
-            prompt=prompt
+            context="ecommerce_search_query",
         )
         print(f"\nTranslation result with context: {result_with_context}")
@@ -1,963 +0,0 @@
-"""
-Translation service for multi-language query support.
-
-Supports multiple translation models:
-- Qwen (default): Alibaba Cloud DashScope API using qwen-mt-flash model
-- DeepL: DeepL API for high-quality translations
-
-重要说明（Qwen 机翻限速）:
-- 当前默认使用的 `qwen-mt-flash` 为云端机翻模型，**官方限速较低，约 RPM=60（每分钟约 60 请求）**
-- 在高并发场景必须依赖 Redis 翻译缓存与批量预热，避免在用户实时请求路径上直接打满 DashScope 限流
-- 若业务侧存在大规模离线翻译或更高吞吐需求，建议评估 DeepL 或自建翻译后端
-
-使用方法 (Usage):
-
-```python
-from query.qwen_mt_translate import Translator
-
-# 使用默认的 qwen 模型（推荐）
-translator = Translator()  # 默认使用 qwen 模型
-
-# 或显式指定模型
-translator = Translator(model='qwen')  # 使用 qwen 模型
-translator = Translator(model='deepl')  # 使用 DeepL 模型
-
-# 翻译文本
-result = translator.translate(
-    text="我看到这个视频后没有笑",
-    target_lang="en",
-    source_lang="auto"  # 自动检测源语言
-)
-```
-
-配置说明 (Configuration):
-- Qwen 模型需要设置 DASHSCOPE_API_KEY 环境变量（在 .env 文件中）
-- DeepL 模型需要设置 DEEPL_AUTH_KEY 环境变量（在 .env 文件中）
-
-Qwen 模型参考文档：
-- 官方文档：https://help.aliyun.com/zh/model-studio/get-api-key
-- 模型：qwen-mt-flash（快速翻译模型）
-
-DeepL 官方文档：
-https://developers.deepl.com/api-reference/translate/request-translation
-"""
-
-import os
-import requests
-import re
-import redis
-from concurrent.futures import ThreadPoolExecutor, Future
-from datetime import timedelta
-from typing import Dict, List, Optional, Union
-import logging
-import time
-
-logger = logging.getLogger(__name__)
-
-from config.env_config import DEEPL_AUTH_KEY, DASHSCOPE_API_KEY, REDIS_CONFIG
-from openai import OpenAI
-
-
-class Translator:
-    """
-    Multi-language translator supporting Qwen and DeepL APIs.
-    
-    Default model is 'qwen' which uses Alibaba Cloud DashScope API.
-    """
-# 华北2（北京）：https://dashscope.aliyuncs.com/compatible-mode/v1
-# 新加坡：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
-# 美国（弗吉尼亚）：https://dashscope-us.aliyuncs.com/compatible-mode/v1
-
-    DEEPL_API_URL = "https://api.deepl.com/v2/translate"  # Pro tier
-    QWEN_BASE_URL = "https://dashscope-us.aliyuncs.com/compatible-mode/v1"  # 北京地域
-    # QWEN_BASE_URL = "https://dashscope-intl.aliyuncs.com/compatible-mode/v1"  # 新加坡
-    # 如果使用新加坡地域的模型，需要将base_url替换为：https://dashscope-intl.aliyuncs.com/compatible-mode/v1
-    QWEN_MODEL = "qwen-mt-flash"  # 快速翻译模型
-
-    # Language code mapping
-    LANG_CODE_MAP = {
-        'zh': 'ZH',
-        'en': 'EN',
-        'ru': 'RU',
-        'ar': 'AR',
-        'ja': 'JA',
-        'es': 'ES',
-        'de': 'DE',
-        'fr': 'FR',
-        'it': 'IT',
-        'pt': 'PT',
-    }
-
-    def __init__(
-        self,
-        model: str = "qwen",
-        api_key: Optional[str] = None,
-        use_cache: bool = True,
-        timeout: int = 10,
-        glossary_id: Optional[str] = None,
-        translation_context: Optional[str] = None
-    ):
-        """
-        Initialize translator.
-
-        Args:
-            model: Translation model to use. Options: 'qwen' (default) or 'deepl'
-            api_key: API key for the selected model (or None to use from config/env)
-            use_cache: Whether to cache translations
-            timeout: Request timeout in seconds
-            glossary_id: DeepL glossary ID for custom terminology (optional, only for DeepL)
-            translation_context: Context hint for translation (e.g., "e-commerce", "product search")
-        """
-        self.model = model.lower()
-        if self.model not in ['qwen', 'deepl']:
-            raise ValueError(f"Unsupported model: {model}. Supported models: 'qwen', 'deepl'")
-        
-        # Get API key from config if not provided
-        if api_key is None:
-            if self.model == 'qwen':
-                api_key = DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY")
-            else:  # deepl
-                api_key = DEEPL_AUTH_KEY or os.getenv("DEEPL_AUTH_KEY")
-
-        self.api_key = api_key
-        self.timeout = timeout
-        self.use_cache = use_cache
-        self.glossary_id = glossary_id
-        self.translation_context = translation_context or "e-commerce product search"
-        
-        # Initialize OpenAI client for Qwen if needed
-        self.qwen_client = None
-        if self.model == 'qwen':
-            if not self.api_key:
-                logger.warning("DASHSCOPE_API_KEY not set. Qwen translation will not work.")
-            else:
-                self.qwen_client = OpenAI(
-                    api_key=self.api_key,
-                    base_url=self.QWEN_BASE_URL,
-                )
-
-        # Initialize Redis cache if enabled
-        if use_cache:
-            try:
-                self.redis_client = redis.Redis(
-                    host=REDIS_CONFIG.get('host', 'localhost'),
-                    port=REDIS_CONFIG.get('port', 6479),
-                    password=REDIS_CONFIG.get('password'),
-                    decode_responses=True,  # Return str instead of bytes
-                    socket_timeout=REDIS_CONFIG.get('socket_timeout', 1),
-                    socket_connect_timeout=REDIS_CONFIG.get('socket_connect_timeout', 1),
-                    retry_on_timeout=REDIS_CONFIG.get('retry_on_timeout', False),
-                    health_check_interval=10,  # 避免复用坏连接
-                )
-                # Test connection
-                self.redis_client.ping()
-                expire_days = REDIS_CONFIG.get('translation_cache_expire_days', 360)
-                self.expire_time = timedelta(days=expire_days)
-                self.expire_seconds = int(self.expire_time.total_seconds())  # Redis 需要秒数
-                self.cache_prefix = REDIS_CONFIG.get('translation_cache_prefix', 'trans')
-                logger.info("Redis cache initialized for translations")
-            except Exception as e:
-                logger.warning(f"Failed to initialize Redis cache: {e}, falling back to no cache")
-                self.redis_client = None
-                self.cache = None
-        else:
-            self.redis_client = None
-            self.cache = None
-        
-        # Thread pool for async translation
-        self.executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="translator")
-
-    def translate(
-        self,
-        text: str,
-        target_lang: str,
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ) -> Optional[str]:
-        """
-        Translate text to target language (synchronous mode).
-
-        Args:
-            text: Text to translate
-            target_lang: Target language code ('zh', 'en', 'ru', etc.)
-            source_lang: Source language code (option al, auto-detect if None)
-            context: Additional context for translation (overrides default context)
-            prompt: Translation prompt/instruction (optional, for better translation quality)
-
-        Returns:
-            Translated text or None if translation fails
-        """
-        if not text or not text.strip():
-            return text
-
-        # Normalize language codes
-        target_lang = target_lang.lower()
-        if source_lang:
-            source_lang = source_lang.lower()
-
-        # Optimization: Skip translation if not needed
-        if target_lang == 'en' and self._is_english_text(text):
-            logger.info(f"[Translator] Text is already English, skipping translation: '{text[:50]}...'")
-            return text
-        
-        if target_lang == 'zh' and (self._contains_chinese(text) or self._is_pure_number(text)):
-            logger.info(
-                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
-                f"Source language: {source_lang or 'auto'} | Result: Skip translation (contains Chinese or pure number)"
-            )
-            return text
-
-        # Use provided context or default context
-        translation_context = context or self.translation_context
-        
-        # Build cache key (include prompt in cache key if provided)
-        cache_key_parts = [source_lang or 'auto', target_lang, translation_context]
-        if prompt:
-            cache_key_parts.append(prompt)
-        cache_key_parts.append(text)
-        cache_key = ':'.join(cache_key_parts)
-
-        # Check cache (include context and prompt in cache key for accuracy)
-        if self.use_cache and self.redis_client:
-            cached = self._get_cached_translation_redis(text, target_lang, source_lang, translation_context, prompt)
-            if cached:
-                logger.info(
-                    f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
-                    f"Source language: {source_lang or 'auto'} | Result: '{cached}' | Source: Cache hit"
-                )
-                return cached
-
-        # If no API key, return mock translation (for testing)
-        if not self.api_key:
-            logger.info(
-                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
-                f"Source language: {source_lang or 'auto'} | Result: '{text}' | Source: Mock mode (no API key)"
-            )
-            return text
-
-        # Translate using selected model
-        logger.info(
-            f"[Translator] Translation request | Model: {self.model} | Original text: '{text}' | Target language: {target_lang} | "
-            f"Source language: {source_lang or 'auto'} | Context: {translation_context} | "
-            f"Prompt: {'yes' if prompt else 'no'} | Status: Starting translation"
-        )
-        
-        if self.model == 'qwen':
-            result = self._translate_qwen(text, target_lang, source_lang, translation_context, prompt)
-        else:  # deepl
-            result = self._translate_deepl(text, target_lang, source_lang, translation_context, prompt)
-
-        # Surface translation failure to the caller instead of silently
-        # masquerading the source text as a successful translation.
-        if result is None:
-            logger.warning(
-                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
-                f"Source language: {source_lang or 'auto'} | Status: Translation failed"
-            )
-        else:
-            logger.info(
-                f"[Translator] Translation request | Original text: '{text}' | Target language: {target_lang} | "
-                f"Source language: {source_lang or 'auto'} | Result: '{result}' | Status: Translation successful"
-            )
-
-        # Cache only successful translations. Failed attempts must not poison
-        # Redis with the original text.
-        if result is not None and self.use_cache and self.redis_client:
-            self._set_cached_translation_redis(text, target_lang, result, source_lang, translation_context, prompt)
-
-        return result
-
-    def _translate_qwen(
-        self,
-        text: str,
-        target_lang: str,
-        source_lang: Optional[str],
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ) -> Optional[str]:
-        """
-        Translate using Qwen MT Flash model via Alibaba Cloud DashScope API.
-        
-        Args:
-            text: Text to translate
-            target_lang: Target language code ('zh', 'en', 'ru', etc.)
-            source_lang: Source language code (optional, 'auto' if None)
-            context: Context hint for translation (optional)
-            prompt: Translation prompt/instruction (optional)
-            
-        Returns:
-            Translated text or None if translation fails
-        """
-        if not self.qwen_client:
-            logger.error("[Translator] Qwen client not initialized. Check DASHSCOPE_API_KEY.")
-            return None
-        
-        # Qwen (qwen-mt-plus/flash/turbo) supported languages mapping
-        # 标准来自：你提供的“语言 / 英文名 / 代码”表
-        qwen_lang_map = {
-            "en": "English",
-            "zh": "Chinese",
-            "zh_tw": "Traditional Chinese",
-            "ru": "Russian",
-            "ja": "Japanese",
-            "ko": "Korean",
-            "es": "Spanish",
-            "fr": "French",
-            "pt": "Portuguese",
-            "de": "German",
-            "it": "Italian",
-            "th": "Thai",
-            "vi": "Vietnamese",
-            "id": "Indonesian",
-            "ms": "Malay",
-            "ar": "Arabic",
-            "hi": "Hindi",
-            "he": "Hebrew",
-            "my": "Burmese",
-            "ta": "Tamil",
-            "ur": "Urdu",
-            "bn": "Bengali",
-            "pl": "Polish",
-            "nl": "Dutch",
-            "ro": "Romanian",
-            "tr": "Turkish",
-            "km": "Khmer",
-            "lo": "Lao",
-            "yue": "Cantonese",
-            "cs": "Czech",
-            "el": "Greek",
-            "sv": "Swedish",
-            "hu": "Hungarian",
-            "da": "Danish",
-            "fi": "Finnish",
-            "uk": "Ukrainian",
-            "bg": "Bulgarian",
-        }
-        
-        # Convert target language
-        target_lang_normalized = target_lang.lower()
-        target_lang_qwen = qwen_lang_map.get(target_lang_normalized, target_lang.capitalize())
-
-        # Convert source language
-        source_lang_normalized = (source_lang or "").strip().lower()
-        if not source_lang_normalized or source_lang_normalized == "auto":
-            source_lang_qwen = "auto"
-        else:
-            source_lang_qwen = qwen_lang_map.get(source_lang_normalized, source_lang.capitalize())
-        
-        # Prepare translation options
-        translation_options = {
-            "source_lang": source_lang_qwen,
-            "target_lang": target_lang_qwen,
-        }
-        
-        # Prepare messages
-        messages = [
-            {
-                "role": "user",
-                "content": text
-            }
-        ]
-        
-        start_time = time.time()
-        try:
-            completion = self.qwen_client.chat.completions.create(
-                model=self.QWEN_MODEL,
-                messages=messages,
-                extra_body={
-                    "translation_options": translation_options
-                }
-            )
-            
-            translated_text = completion.choices[0].message.content.strip()
-            duration_ms = (time.time() - start_time) * 1000
-            
-            logger.info(
-                f"[Translator] Qwen API response success | Original text: '{text}' | Target language: {target_lang_qwen} | "
-                f"Translation result: '{translated_text}' | Duration: {duration_ms:.2f} ms"
-            )
-            return translated_text
-            
-        except Exception as e:
-            duration_ms = (time.time() - start_time) * 1000
-            logger.error(
-                f"[Translator] Qwen API request exception | Original text: '{text}' | Target language: {target_lang_qwen} | "
-                f"Duration: {duration_ms:.2f} ms | Error: {e}", exc_info=True
-            )
-            return None
-
-    def _translate_deepl(
-        self,
-        text: str,
-        target_lang: str,
-        source_lang: Optional[str],
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ) -> Optional[str]:
-        """
-        Translate using DeepL API with context and glossary support.
-        
-        Args:
-            text: Text to translate
-            target_lang: Target language code
-            source_lang: Source language code (optional)
-            context: Context hint for translation (e.g., "e-commerce product search")
-        """
-        # Map to DeepL language codes
-        target_code = self.LANG_CODE_MAP.get(target_lang, target_lang.upper())
-
-        headers = {
-            "Authorization": f"DeepL-Auth-Key {self.api_key}",
-            "Content-Type": "application/json",
-        }
-
-        # Use prompt as context parameter for DeepL API (not as text prefix)
-        # According to DeepL API: context is "Additional context that can influence a translation but is not translated itself"
-        # If prompt is provided, use it as context; otherwise use the default context
-        api_context = prompt if prompt else context
-        
-        # For e-commerce, add context words to help DeepL understand the domain
-        # This is especially important for single-word ambiguous terms like "车" (car vs rook)
-        text_to_translate, needs_extraction = self._add_ecommerce_context(text, source_lang, api_context)
-
-        payload = {
-            "text": [text_to_translate],
-            "target_lang": target_code,
-        }
-
-        if source_lang:
-            source_code = self.LANG_CODE_MAP.get(source_lang, source_lang.upper())
-            payload["source_lang"] = source_code
-
-        # Add context parameter (prompt or default context)
-        # Context influences translation but is not translated itself
-        if api_context:
-            payload["context"] = api_context
-
-        # Add glossary if configured
-        if self.glossary_id:
-            payload["glossary_id"] = self.glossary_id
-
-        # Note: DeepL API v2 supports "context" parameter for additional context
-        # that influences translation but is not translated itself.
-        # We use prompt as context parameter when provided.
-
-        try:
-            response = requests.post(
-                self.DEEPL_API_URL,
-                headers=headers,
-                json=payload,
-                timeout=self.timeout
-            )
-
-            if response.status_code == 200:
-                data = response.json()
-                if "translations" in data and len(data["translations"]) > 0:
-                    translated_text = data["translations"][0]["text"]
-                    # If we added context, extract just the term from the result
-                    if needs_extraction:
-                        translated_text = self._extract_term_from_translation(
-                            translated_text, text, target_code
-                        )
-                    logger.debug(
-                        f"[Translator] DeepL API response success | Original text: '{text}' | Target language: {target_code} | "
-                        f"Translation result: '{translated_text}'"
-                    )
-                    return translated_text
-            else:
-                logger.error(
-                    f"[Translator] DeepL API error | Original text: '{text}' | Target language: {target_code} | "
-                    f"Status code: {response.status_code} | Error message: {response.text}"
-                )
-                return None
-
-        except requests.Timeout:
-            logger.warning(
-                f"[Translator] DeepL API request timeout | Original text: '{text}' | Target language: {target_code} | "
-                f"Timeout: {self.timeout}s"
-            )
-            return None
-        except Exception as e:
-            logger.error(
-                f"[Translator] DeepL API request exception | Original text: '{text}' | Target language: {target_code} | "
-                f"Error: {e}", exc_info=True
-            )
-            return None
-
-    # NOTE: _translate_deepl_free is intentionally not implemented.
-    # We do not support automatic fallback to the free endpoint, to avoid
-    # mixing Pro keys with https://api-free.deepl.com and related 403 errors.
-
-    def translate_multi(
-        self,
-        text: str,
-        target_langs: List[str],
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        async_mode: bool = True,
-        prompt: Optional[str] = None
-    ) -> Dict[str, Optional[str]]:
-        """
-        Translate text to multiple target languages.
-        
-        In async_mode=True (default):
-        - Returns cached translations immediately if available
-        - For translations that can be optimized (e.g., pure numbers, already in target language),
-          returns result immediately via synchronous call
-        - Launches async tasks for other missing translations (non-blocking)
-        - Returns None for missing translations that require async processing
-        
-        In async_mode=False:
-        - Waits for all translations to complete (blocking)
-
-        Args:
-            text: Text to translate
-            target_langs: List of target language codes
-            source_lang: Source language code (optional)
-            context: Context hint for translation (optional)
-            async_mode: If True, return cached results immediately and translate missing ones async
-            prompt: Translation prompt/instruction (optional)
-
-        Returns:
-            Dictionary mapping language code to translated text (only cached results in async mode)
-        """
-        results = {}
-        missing_langs = []
-        async_langs = []
-        
-        # First, get cached translations
-        for lang in target_langs:
-            cached = self._get_cached_translation(text, lang, source_lang, context, prompt)
-            if cached is not None:
-                results[lang] = cached
-            else:
-                missing_langs.append(lang)
-        
-        # If async mode and there are missing translations
-        if async_mode and missing_langs:
-            # Check if translation can be optimized (immediate return)
-            for lang in missing_langs:
-                target_lang = lang.lower()
-                # Check optimization conditions (same as in translate method)
-                can_optimize = False
-                if target_lang == 'en' and self._is_english_text(text):
-                    can_optimize = True
-                elif target_lang == 'zh' and (self._contains_chinese(text) or self._is_pure_number(text)):
-                    can_optimize = True
-                
-                if can_optimize:
-                    # Can be optimized, call translate synchronously for immediate result
-                    results[lang] = self.translate(text, lang, source_lang, context, prompt)
-                else:
-                    # Requires actual translation, add to async list
-                    async_langs.append(lang)
-            
-            # Launch async tasks for translations that require actual API calls
-            if async_langs:
-                for lang in async_langs:
-                    self._translate_async(text, lang, source_lang, context, prompt)
-                # Return None for async translations
-                for lang in async_langs:
-                    results[lang] = None
-        else:
-            # Synchronous mode: wait for all translations
-            for lang in missing_langs:
-                results[lang] = self.translate(text, lang, source_lang, context, prompt)
-        
-        return results
-    
-    def translate_multi_async(
-        self,
-        text: str,
-        target_langs: List[str],
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ) -> Dict[str, Union[str, Future]]:
-        """
-        Translate text to multiple target languages asynchronously, returning Futures that can be awaited.
-        
-        This method returns a dictionary where:
-        - If translation is cached, the value is the translation string (immediate)
-        - If translation needs to be done, the value is a Future object that can be awaited
-        
-        Args:
-            text: Text to translate
-            target_langs: List of target language codes
-            source_lang: Source language code (optional)
-            context: Context hint for translation (optional)
-            prompt: Translation prompt/instruction (optional)
-
-        Returns:
-            Dictionary mapping language code to either translation string (cached) or Future object
-        """
-        results = {}
-        missing_langs = []
-        
-        # First, get cached translations
-        for lang in target_langs:
-            cached = self._get_cached_translation(text, lang, source_lang, context, prompt)
-            if cached is not None:
-                results[lang] = cached
-            else:
-                missing_langs.append(lang)
-        
-        # For missing translations, submit async tasks and return Futures
-        for lang in missing_langs:
-            future = self.executor.submit(
-                self.translate,
-                text,
-                lang,
-                source_lang,
-                context,
-                prompt
-            )
-            results[lang] = future
-        
-        return results
-    
-    def _get_cached_translation(
-        self,
-        text: str,
-        target_lang: str,
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ) -> Optional[str]:
-        """Get translation from cache if available."""
-        if not self.redis_client:
-            return None
-        return self._get_cached_translation_redis(text, target_lang, source_lang, context, prompt)
-    
-    def _get_cached_translation_redis(
-        self,
-        text: str,
-        target_lang: str,
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ) -> Optional[str]:
-        """
-        Get translation from Redis cache with sliding expiration.
-        
-        滑动过期机制：每次访问缓存时，重置过期时间为配置的过期时间（默认720天）。
-        这样缓存会在最后一次访问后的720天才过期，而不是写入后的720天。
-        这确保了常用的翻译缓存不会被过早删除。
-        """
-        if not self.redis_client:
-            return None
-        
-        try:
-            # Build cache key: prefix:target_lang:text
-            # For simplicity, we use target_lang and text as key
-            # Context and prompt are not included in key to maximize cache hits
-            cache_key = f"{self.cache_prefix}:{target_lang.upper()}:{text}"
-            value = self.redis_client.get(cache_key)
-            if value:
-                # Sliding expiration: reset expiration time on access
-                # 每次读取缓存时，重置过期时间为配置的过期时间（最后一次访问后的N天才过期）
-                try:
-                    self.redis_client.expire(cache_key, self.expire_seconds)
-                except Exception as expire_error:
-                    # 即使 expire 失败，也返回缓存值（不影响功能）
-                    logger.warning(
-                        f"[Translator] Failed to update cache expiration for key {cache_key}: {expire_error}"
-                    )
-                
-                logger.debug(
-                    f"[Translator] Redis cache hit | Original text: '{text}' | Target language: {target_lang} | "
-                    f"Cache key: {cache_key} | Translation result: '{value}' | TTL reset to {self.expire_seconds}s"
-                )
-                return value
-            logger.debug(
-                f"[Translator] Redis cache miss | Original text: '{text}' | Target language: {target_lang} | "
-                f"Cache key: {cache_key}"
-            )
-            return None
-        except Exception as e:
-            logger.error(f"[Translator] Redis error during get translation cache | Original text: '{text}' | Target language: {target_lang} | Error: {e}")
-            return None
-    
-    def _set_cached_translation_redis(
-        self,
-        text: str,
-        target_lang: str,
-        translation: str,
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ) -> None:
-        """Store translation in Redis cache."""
-        if not self.redis_client:
-            return
-        
-        try:
-            cache_key = f"{self.cache_prefix}:{target_lang.upper()}:{text}"
-            self.redis_client.setex(cache_key, self.expire_seconds, translation)
-            logger.info(
-                f"[Translator] Redis cache write | Original text: '{text}' | Target language: {target_lang} | "
-                f"Cache key: {cache_key} | Translation result: '{translation}'"
-            )
-        except Exception as e:
-            logger.error(
-                f"[Translator] Redis cache write failed | Original text: '{text}' | Target language: {target_lang} | "
-                f"Error: {e}"
-            )
-    
-    def _translate_async(
-        self,
-        text: str,
-        target_lang: str,
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None
-    ):
-        """Launch async translation task."""
-        def _do_translate():
-            try:
-                result = self.translate(text, target_lang, source_lang, context, prompt)
-                if result:
-                    logger.debug(f"Async translation completed: {text} -> {target_lang}: {result}")
-            except Exception as e:
-                logger.warning(f"Async translation failed: {text} -> {target_lang}: {e}")
-        
-        self.executor.submit(_do_translate)
-
-    def _add_ecommerce_context(
-        self,
-        text: str,
-        source_lang: Optional[str],
-        context: Optional[str]
-    ) -> tuple:
-        """
-        Add e-commerce context to text for better disambiguation.
-        
-        For single-word ambiguous Chinese terms, we add context words that help
-        DeepL understand this is an e-commerce/product search context.
-        
-        Args:
-            text: Original text to translate
-            source_lang: Source language code
-            context: Context hint
-            
-        Returns:
-            Tuple of (text_with_context, needs_extraction)
-            - text_with_context: Text to send to DeepL
-            - needs_extraction: Whether we need to extract the term from the result
-        """
-        # Only apply for e-commerce context and Chinese source
-        if not context or "e-commerce" not in context.lower():
-            return text, False
-            
-        if not source_lang or source_lang.lower() != 'zh':
-            return text, False
-            
-        # For single-word queries, add context to help disambiguation
-        text_stripped = text.strip()
-        if len(text_stripped.split()) == 1 and len(text_stripped) <= 2:
-            # Common ambiguous Chinese e-commerce terms like "车" (car vs rook)
-            # We add a context phrase: "购买 [term]" (buy [term]) or "商品 [term]" (product [term])
-            # This helps DeepL understand the e-commerce context
-            # We'll need to extract just the term from the translation result
-            context_phrase = f"购买 {text_stripped}"
-            return context_phrase, True
-        
-        # For multi-word queries, DeepL usually has enough context
-        return text, False
-
-    def _extract_term_from_translation(
-        self,
-        translated_text: str,
-        original_text: str,
-        target_lang_code: str
-    ) -> str:
-        """
-        Extract the actual term from a translation that included context.
-        
-        For example, if we translated "购买 车" (buy car) and got "buy car",
-        we want to extract just "car".
-        
-        Args:
-            translated_text: Full translation result
-            original_text: Original single-word query
-            target_lang_code: Target language code (EN, ZH, etc.)
-            
-        Returns:
-            Extracted term or original translation if extraction fails
-        """
-        # For English target, try to extract the last word (the actual term)
-        if target_lang_code == "EN":
-            words = translated_text.strip().split()
-            if len(words) > 1:
-                # Usually the last word is the term we want
-                # But we need to be smart - if it's "buy car", we want "car"
-                # Common context words to skip: buy, purchase, product, item, etc.
-                context_words = {"buy", "purchase", "product", "item", "commodity", "goods"}
-                # Try to find the term (not a context word)
-                for word in reversed(words):
-                    word_lower = word.lower().rstrip('.,!?;:')
-                    if word_lower not in context_words:
-                        return word_lower
-                # If all words are context words, return the last one
-                return words[-1].lower().rstrip('.,!?;:')
-        
-        # For other languages or if extraction fails, return as-is
-        # The user can configure a glossary for better results
-        return translated_text
-
-    def _shop_lang_matches(self, shop_lang_lower: str, lang_code: str) -> bool:
-        """True if shop language matches index language (use source, no translate)."""
-        if not shop_lang_lower or not lang_code:
-            return False
-        if shop_lang_lower == lang_code:
-            return True
-        if lang_code == "zh" and "zh" in shop_lang_lower:
-            return True
-        if lang_code == "en" and "en" in shop_lang_lower:
-            return True
-        return False
-
-    def translate_for_indexing(
-        self,
-        text: str,
-        shop_language: str,
-        source_lang: Optional[str] = None,
-        context: Optional[str] = None,
-        prompt: Optional[str] = None,
-        index_languages: Optional[List[str]] = None,
-    ) -> Dict[str, Optional[str]]:
-        """
-        Translate text for indexing based on shop language and tenant index_languages.
-
-        For each language in index_languages: use source text if shop language matches,
-        otherwise translate to that language.
-
-        Args:
-            text: Text to translate
-            shop_language: Shop primary language (e.g. 'zh', 'en', 'ru')
-            source_lang: Source language code (optional)
-            context: Additional context for translation (optional)
-            prompt: Translation prompt (optional)
-            index_languages: Languages to index (from tenant_config). Default ["en", "zh"].
-
-        Returns:
-            Dict keyed by each index_language with translated or source text (or None).
-        """
-        langs = index_languages if index_languages else ["en", "zh"]
-        results = {lang: None for lang in langs}
-        if not text or not text.strip():
-            return results
-        if re.match(r'^[\d\s_-]+$', text):
-            logger.info(f"[Translator] Skip translation for symbol-only query: '{text}'")
-            return results
-
-        shop_lang_lower = (shop_language or "").strip().lower()
-        targets = []
-        for lang in langs:
-            if self._shop_lang_matches(shop_lang_lower, lang):
-                results[lang] = text
-            else:
-                targets.append(lang)
-
-        for target_lang in targets:
-            cached = self._get_cached_translation_redis(text, target_lang, source_lang, context, prompt)
-            if cached:
-                results[target_lang] = cached
-                logger.debug(f"[Translator] Cache hit for indexing: '{text}' -> {target_lang}: {cached}")
-                continue
-            translated = self.translate(
-                text,
-                target_lang=target_lang,
-                source_lang=source_lang or shop_language,
-                context=context,
-                prompt=prompt,
-            )
-            results[target_lang] = translated
-        return results
-
-    def get_translation_needs(
-        self,
-        detected_lang: str,
-        supported_langs: List[str]
-    ) -> List[str]:
-        """
-        Determine which languages need translation.
-
-        Args:
-            detected_lang: Detected query language
-            supported_langs: List of supported languages
-
-        Returns:
-            List of language codes to translate to
-        """
-        # If detected language is in supported list, translate to others
-        if detected_lang in supported_langs:
-            return [lang for lang in supported_langs if detected_lang != lang]
-
-        # Otherwise, translate to all supported languages
-        return supported_langs
-    
-    def _is_english_text(self, text: str) -> bool:
-        """
-        Check if text is primarily English (ASCII letters, numbers, common punctuation).
-        
-        Args:
-            text: Text to check
-            
-        Returns:
-            True if text appears to be English
-        """
-        if not text or not text.strip():
-            return True
-        
-        # Remove whitespace and common punctuation
-        text_clean = re.sub(r'[\s\.,!?;:\-\'\"\(\)\[\]{}]', '', text)
-        if not text_clean:
-            return True
-        
-        # Check if all remaining characters are ASCII (letters, numbers)
-        # This is a simple heuristic: if most characters are ASCII, it's likely English
-        ascii_count = sum(1 for c in text_clean if ord(c) < 128)
-        ratio = ascii_count / len(text_clean) if text_clean else 0
-        
-        # If more than 80% are ASCII characters, consider it English
-        return ratio > 0.8
-    
-    def _contains_chinese(self, text: str) -> bool:
-        """
-        Check if text contains Chinese characters (Han characters).
-        
-        Args:
-            text: Text to check
-            
-        Returns:
-            True if text contains Chinese characters
-        """
-        if not text:
-            return False
-        
-        # Check for Chinese characters (Unicode range: \u4e00-\u9fff)
-        chinese_pattern = re.compile(r'[\u4e00-\u9fff]')
-        return bool(chinese_pattern.search(text))
-    
-    def _is_pure_number(self, text: str) -> bool:
-        """
-        Check if text is purely numeric (digits, possibly with spaces, dots, commas).
-        
-        Args:
-            text: Text to check
-            
-        Returns:
-            True if text is purely numeric
-        """
-        if not text or not text.strip():
-            return False
-        
-        # Remove whitespace, dots, commas (common number separators)
-        text_clean = re.sub(r'[\s\.,]', '', text.strip())
-        if not text_clean:
-            return False
-        
-        # Check if all remaining characters are digits
-        return text_clean.isdigit()
@@ -77,12 +77,10 @@ def _build_test_config() -&gt; SearchConfig:
             enable_text_embedding=True,
             enable_query_rewrite=False,
             rewrite_dictionary={},
-            translation_prompts={"query_zh": "e-commerce domain", "query_en": "e-commerce domain"},
             text_embedding_field="title_embedding",
             image_embedding_field=None,
         ),
         function_score=FunctionScoreConfig(),
-        function_score=FunctionScoreConfig(),
         rerank=RerankConfig(),
         spu_config=SPUConfig(enabled=True, spu_field="spu_id", inner_hits_size=3),
         es_index_name="test_products",