trans logs

tangwang
1 parent c90f80ed
Showing 14 changed files with 657 additions and 239 deletions Show diff stats
api/translator_app.py
config/config.yaml
config/env_config.py
docs/工作总结-微服务性能优化与架构.md
docs/缓存与Redis使用说明.md
scripts/redis/redis_cache_health_check.py
tests/ci/test_service_api_contracts.py
tests/test_translation_local_backends.py
tests/test_translator_failure_semantics.py
translation/README.md
translation/backends/qwen_mt.py
translation/cache.py
translation/service.py
translation/settings.py
@@ -2,8 +2,12 @@
 import argparse
 import logging
+import os
+import pathlib
+import time
 from contextlib import asynccontextmanager
 from functools import lru_cache
+from logging.handlers import TimedRotatingFileHandler
 from typing import List, Optional, Union
 import uvicorn
@@ -20,12 +24,57 @@ from translation.settings import (
     normalize_translation_scene,
 )
-# Configure logging
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-)
+
+def configure_translator_logging() -> None:
+    log_dir = pathlib.Path("logs")
+    verbose_dir = log_dir / "verbose"
+    log_dir.mkdir(exist_ok=True)
+    verbose_dir.mkdir(parents=True, exist_ok=True)
+
+    log_level = os.getenv("LOG_LEVEL", "INFO").upper()
+    numeric_level = getattr(logging, log_level, logging.INFO)
+    formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+
+    root_logger = logging.getLogger()
+    root_logger.setLevel(numeric_level)
+    root_logger.handlers.clear()
+
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(numeric_level)
+    console_handler.setFormatter(formatter)
+    root_logger.addHandler(console_handler)
+
+    file_handler = TimedRotatingFileHandler(
+        filename=log_dir / "translator_api.log",
+        when="midnight",
+        interval=1,
+        backupCount=30,
+        encoding="utf-8",
+    )
+    file_handler.setLevel(numeric_level)
+    file_handler.setFormatter(formatter)
+    root_logger.addHandler(file_handler)
+
+    verbose_logger = logging.getLogger("translator.verbose")
+    verbose_logger.setLevel(numeric_level)
+    verbose_logger.handlers.clear()
+    verbose_logger.propagate = False
+
+    verbose_handler = TimedRotatingFileHandler(
+        filename=verbose_dir / "translator_verbose.log",
+        when="midnight",
+        interval=1,
+        backupCount=30,
+        encoding="utf-8",
+    )
+    verbose_handler.setLevel(numeric_level)
+    verbose_handler.setFormatter(formatter)
+    verbose_logger.addHandler(verbose_handler)
+
+
+configure_translator_logging()
 logger = logging.getLogger(__name__)
+verbose_logger = logging.getLogger("translator.verbose")
 @lru_cache(maxsize=1)
@@ -98,6 +147,37 @@ def _normalize_batch_result(
     return [translated[idx] if idx < len(translated) else None for idx, _ in enumerate(original)]
+def _text_preview(text: Optional[str], limit: int = 20) -> str:
+    normalized = str(text or "").replace("\n", "\\n")
+    return normalized[:limit]
+
+
+def _request_metrics(text: Union[str, List[str]]) -> dict:
+    if isinstance(text, list):
+        lengths = [len(str(item or "")) for item in text]
+        return {
+            "request_count": len(text),
+            "lengths": lengths,
+            "first_preview": _text_preview(text[0] if text else ""),
+        }
+    return {
+        "request_count": 1,
+        "lengths": [len(str(text or ""))],
+        "first_preview": _text_preview(str(text or "")),
+    }
+
+
+def _result_preview(translated: Union[str, List[Optional[str]], None]) -> str:
+    if isinstance(translated, list):
+        if not translated:
+            return ""
+        first = translated[0]
+        return _text_preview("" if first is None else str(first))
+    if translated is None:
+        return ""
+    return _text_preview(str(translated))
+
+
 def _translate_batch(
     service: TranslationService,
     raw_text: List[str],
@@ -108,6 +188,17 @@ def _translate_batch(
     scene: str,
 ) -> List[Optional[str]]:
     backend = service.get_backend(model)
+    logger.info(
+        "Translation batch dispatch | model=%s scene=%s target_lang=%s source_lang=%s count=%s lengths=%s first_preview=%s supports_batch=%s",
+        model,
+        scene,
+        target_lang,
+        source_lang or "auto",
+        len(raw_text),
+        [len(str(item or "")) for item in raw_text],
+        _text_preview(raw_text[0] if raw_text else ""),
+        bool(getattr(backend, "supports_batch", False)),
+    )
     if getattr(backend, "supports_batch", False):
         try:
             translated = service.translate(
@@ -117,6 +208,13 @@ def _translate_batch(
                 model=model,
                 scene=scene,
             )
+            verbose_logger.info(
+                "Translation batch result | model=%s scene=%s count=%s first_result=%s",
+                model,
+                scene,
+                len(raw_text),
+                _result_preview(translated),
+            )
             return _normalize_batch_result(raw_text, translated)
         except ValueError:
             raise
@@ -139,7 +237,17 @@ def _translate_batch(
         except ValueError:
             raise
         except Exception as exc:
-            logger.warning("Per-item translation failed: %s", exc, exc_info=True)
+            logger.warning(
+                "Per-item translation failed | model=%s scene=%s target_lang=%s source_lang=%s item_len=%s item_preview=%s error=%s",
+                model,
+                scene,
+                target_lang,
+                source_lang or "auto",
+                len(str(item or "")),
+                _text_preview(str(item or "")),
+                exc,
+                exc_info=True,
+            )
             out = None
         results.append(out)
     return results
@@ -147,19 +255,25 @@ def _translate_batch(
 @asynccontextmanager
 async def lifespan(_: FastAPI):
-    """Warm the default backend on process startup."""
+    """Initialize all enabled translation backends on process startup."""
     logger.info("Starting Translation Service API")
     service = get_translation_service()
-    default_backend = service.get_backend(service.config["default_model"])
     logger.info(
-        "Translation service ready | default_model=%s available_models=%s loaded_models=%s",
+        "Translation service ready | default_model=%s default_scene=%s available_models=%s loaded_models=%s",
         service.config["default_model"],
+        service.config["default_scene"],
         service.available_models,
         service.loaded_models,
     )
     logger.info(
-        "Default translation backend warmed up | model=%s",
-        getattr(default_backend, "model", service.config["default_model"]),
+        "Translation backends initialized on startup | models=%s",
+        service.loaded_models,
+    )
+    verbose_logger.info(
+        "Translation startup detail | capabilities=%s cache_ttl_seconds=%s cache_sliding_expiration=%s",
+        service.available_models,
+        service.config["cache"]["ttl_seconds"],
+        service.config["cache"]["sliding_expiration"],
     )
     yield
@@ -189,6 +303,12 @@ async def health_check():
     """Health check endpoint."""
     try:
         service = get_translation_service()
+        logger.info(
+            "Health check | default_model=%s default_scene=%s loaded_models=%s",
+            service.config["default_model"],
+            service.config["default_scene"],
+            service.loaded_models,
+        )
         return {
             "status": "healthy",
             "service": "translation",
@@ -216,12 +336,33 @@ async def translate(request: TranslationRequest):
     if not request.target_lang:
         raise HTTPException(status_code=400, detail="target_lang is required")
+    request_started = time.perf_counter()
     try:
         service = get_translation_service()
         scene = _normalize_scene(service, request.scene)
         model = _normalize_model(service, request.model)
         translator = service.get_backend(model)
         raw_text = request.text
+        metrics = _request_metrics(raw_text)
+        logger.info(
+            "Translation request | model=%s scene=%s target_lang=%s source_lang=%s count=%s lengths=%s first_preview=%s backend=%s",
+            model,
+            scene,
+            request.target_lang,
+            request.source_lang or "auto",
+            metrics["request_count"],
+            metrics["lengths"],
+            metrics["first_preview"],
+            getattr(translator, "model", model),
+        )
+        verbose_logger.info(
+            "Translation request detail | model=%s scene=%s target_lang=%s source_lang=%s payload=%s",
+            model,
+            scene,
+            request.target_lang,
+            request.source_lang or "auto",
+            raw_text,
+        )
         if isinstance(raw_text, list):
             results = _translate_batch(
@@ -232,6 +373,22 @@ async def translate(request: TranslationRequest):
                 model=model,
                 scene=scene,
             )
+            latency_ms = (time.perf_counter() - request_started) * 1000
+            logger.info(
+                "Translation response | model=%s scene=%s count=%s first_result=%s latency_ms=%.2f",
+                model,
+                scene,
+                len(raw_text),
+                _result_preview(results),
+                latency_ms,
+            )
+            verbose_logger.info(
+                "Translation response detail | model=%s scene=%s translated=%s latency_ms=%.2f",
+                model,
+                scene,
+                results,
+                latency_ms,
+            )
             return TranslationResponse(
                 text=raw_text,
                 target_lang=request.target_lang,
@@ -253,6 +410,22 @@ async def translate(request: TranslationRequest):
         if translated_text is None:
             raise HTTPException(status_code=500, detail="Translation failed")
+        latency_ms = (time.perf_counter() - request_started) * 1000
+        logger.info(
+            "Translation response | model=%s scene=%s count=1 first_result=%s latency_ms=%.2f",
+            model,
+            scene,
+            _result_preview(translated_text),
+            latency_ms,
+        )
+        verbose_logger.info(
+            "Translation response detail | model=%s scene=%s translated=%s latency_ms=%.2f",
+            model,
+            scene,
+            translated_text,
+            latency_ms,
+        )
+
         return TranslationResponse(
             text=raw_text,
             target_lang=request.target_lang,
@@ -263,12 +436,22 @@ async def translate(request: TranslationRequest):
             scene=scene,
         )
-    except HTTPException:
+    except HTTPException as exc:
+        latency_ms = (time.perf_counter() - request_started) * 1000
+        logger.warning(
+            "Translation request failed | status_code=%s detail=%s latency_ms=%.2f",
+            exc.status_code,
+            exc.detail,
+            latency_ms,
+        )
         raise
     except ValueError as e:
+        latency_ms = (time.perf_counter() - request_started) * 1000
+        logger.warning("Translation validation error | error=%s latency_ms=%.2f", e, latency_ms, exc_info=True)
         raise HTTPException(status_code=400, detail=str(e)) from e
     except Exception as e:
-        logger.error(f"Translation error: {e}", exc_info=True)
+        latency_ms = (time.perf_counter() - request_started) * 1000
+        logger.error("Translation error | error=%s latency_ms=%.2f", e, latency_ms, exc_info=True)
         raise HTTPException(status_code=500, detail=f"Translation error: {str(e)}")
@@ -106,12 +106,8 @@ services:
     default_scene: "general"
     timeout_sec: 10.0
     cache:
-      enabled: true
-      key_prefix: "trans:v2"
       ttl_seconds: 62208000
       sliding_expiration: true
-      key_include_scene: true
-      key_include_source_lang: true
     capabilities:
       qwen-mt:
         enabled: true
@@ -126,12 +122,14 @@ services:
         model: "qwen-flash"
         base_url: "https://dashscope-us.aliyuncs.com/compatible-mode/v1"
         timeout_sec: 30.0
+        use_cache: true
       deepl:
-        enabled: false
+        enabled: true
         backend: "deepl"
         api_url: "https://api.deepl.com/v2/translate"
         timeout_sec: 10.0
         glossary_id: ""
+        use_cache: true
       nllb-200-distilled-600m:
         enabled: true
         backend: "local_nllb"
@@ -144,6 +142,7 @@ services:
         max_new_tokens: 64
         num_beams: 1
         attn_implementation: "sdpa"
+        use_cache: true
       opus-mt-zh-en:
         enabled: true
         backend: "local_marian"
@@ -155,6 +154,7 @@ services:
         max_input_length: 256
         max_new_tokens: 256
         num_beams: 1
+        use_cache: true
       opus-mt-en-zh:
         enabled: true
         backend: "local_marian"
@@ -166,6 +166,7 @@ services:
         max_input_length: 256
         max_new_tokens: 256
         num_beams: 1
+        use_cache: true
   embedding:
     provider: "http"  # http
     base_url: "http://127.0.0.1:6005"
@@ -42,8 +42,6 @@ REDIS_CONFIG = {
     'socket_connect_timeout': int(os.getenv('REDIS_SOCKET_CONNECT_TIMEOUT', 1)),
     'retry_on_timeout': os.getenv('REDIS_RETRY_ON_TIMEOUT', 'False').lower() == 'true',
     'cache_expire_days': int(os.getenv('REDIS_CACHE_EXPIRE_DAYS', 360*2)),  # 6 months
-    'translation_cache_expire_days': int(os.getenv('REDIS_TRANSLATION_CACHE_EXPIRE_DAYS', 360*2)),
-    'translation_cache_prefix': os.getenv('REDIS_TRANSLATION_CACHE_PREFIX', 'trans'),
     # Embedding 缓存 key 前缀，例如 "embedding"
     'embedding_cache_prefix': os.getenv('REDIS_EMBEDDING_CACHE_PREFIX', 'embedding'),
 }
@@ -88,7 +88,7 @@ instruction: &quot;Given a shopping query, rank product titles by relevance&quot;
 - **配置入口**：`config/config.yaml` → `services.translation`，显式声明 `service_url`、`default_model`、`default_scene`、各 capability 的 `backend`、`base_url/api_url`、timeout 与本地模型运行参数。
 - **内部规则收口**：scene 集合、语言码映射、LLM prompt 模板、本地模型方向约束统一放在 `translation/` 内部，不再散落在 `config/`、`query/` 等位置。
 - **调用位置**：QueryParser 与 Indexer 均通过 `translation.create_translation_client()` 获取客户端，不写死 URL 或模型名。
-- **缓存**：`services.translation.cache` 支持 `key_prefix: "trans:v2"`、`ttl_seconds`、`sliding_expiration` 等，翻译结果写 Redis，减轻重复请求对限速的影响。
+- **缓存**：translator service 对所有 translation capability 统一接入 Redis 缓存；每个 capability 通过 `use_cache` 控制开关，key 格式固定为 `trans:{model}:{target_lang}:{source_text[:4]}{sha256}`。
 - **场景支撑**：在线索引（indexer）与 query 请求（QueryParser）共用同一套 provider 配置；可按环境或租户通过修改 `config.yaml` 或环境变量切换 provider/model。
 - **待配合**：**金伟侧对索引侧翻译调用做流量控制**（限流/排队/批量聚合），避免索引高峰打满 qwen 限速，影响在线 query 翻译。
@@ -12,7 +12,6 @@
 - **Password**：`REDIS_PASSWORD`
 - **Socket & 超时**：`REDIS_SOCKET_TIMEOUT` / `REDIS_SOCKET_CONNECT_TIMEOUT` / `REDIS_RETRY_ON_TIMEOUT`
 - **通用缓存 TTL**：`REDIS_CACHE_EXPIRE_DAYS`（默认 `360*2` 天，代码注释为 “6 months”）
-- **翻译缓存 TTL & 前缀**：`REDIS_TRANSLATION_CACHE_EXPIRE_DAYS`、`REDIS_TRANSLATION_CACHE_PREFIX`
 ---
@@ -21,7 +20,7 @@
 | 模块 / 场景 | Key 模板 | Value 内容示例 | 过期策略 | 备注 |
 |------------|----------|----------------|----------|------|
 | 向量缓存（text/image embedding） | `{EMBEDDING_CACHE_PREFIX}:{query_or_url}` / `{EMBEDDING_CACHE_PREFIX}:image:{url_or_path}` | **BF16 bytes**（每维 2 字节大端存储），读取后恢复为 `np.float32` | TTL=`REDIS_CONFIG["cache_expire_days"]` 天；访问时滑动过期 | 见 `embeddings/text_encoder.py`（文本）与 `embeddings/image_encoder.py`（图片）；前缀由 `REDIS_CONFIG["embedding_cache_prefix"]` 控制 |
-| 翻译结果缓存（Qwen-MT 翻译） | `{cache_prefix}:{model}:{src}:{tgt}:{sha256(payload)}` | 机翻后的单条字符串 | TTL=`services.translation.cache.ttl_seconds` 秒；可配置滑动过期 | 见 `translation/backends/qwen_mt.py` + `config/config.yaml` |
+| 翻译结果缓存（translator service） | `trans:{model}:{target_lang}:{source_text[:4]}{sha256(source_text)}` | 机翻后的单条字符串 | TTL=`services.translation.cache.ttl_seconds` 秒；可配置滑动过期 | 见 `translation/service.py` + `config/config.yaml` |
 | 商品内容理解缓存（anchors / 语义属性 / tags） | `{ANCHOR_CACHE_PREFIX}:{tenant_or_global}:{target_lang}:{md5(title)}` | `json.dumps(dict)`，包含 id/title/category/tags/anchor_text 等 | TTL=`ANCHOR_CACHE_EXPIRE_DAYS` 天 | 见 `indexer/product_enrich.py` |
 下面按模块详细说明。
@@ -71,34 +70,29 @@
 ---
-## 3. 翻译结果缓存（translation/backends/qwen_mt.py）
+## 3. 翻译结果缓存（translation/service.py）
-- **代码位置**：`translation/backends/qwen_mt.py` 中 `QwenMTTranslationBackend`
-- **用途**：缓存 Qwen-MT 翻译（及 translator service 复用的翻译）结果，减少云端请求，遵守限速。
-- **配置入口**：`config/config.yaml -> services.translation.cache`，统一由 `config/services_config.get_translation_cache_config()` 解析。
+- **代码位置**：`translation/service.py`
+- **用途**：统一缓存所有 translation capability 的翻译结果。
+- **配置入口**：
+  - `config/config.yaml -> services.translation.cache`
+  - `config/config.yaml -> services.translation.capabilities.*.use_cache`
 ### 3.1 Key 设计
-- 内部构造函数：`_build_cache_key(...)`
+- 内部构造函数：`TranslationCache.build_key(...)`
 - 模板：
 ```text
-{cache_prefix}:{model}:{src}:{tgt}:{sha256(payload)}
+trans:{model}:{target_lang}:{source_text[:4]}{sha256(source_text)}
 ```
 其中：
-- `cache_prefix`：来自 `services.translation.cache.key_prefix`，默认 `trans:v2`；
-- `model`：如 `"qwen-mt"`；
-- `src`：源语言（如 `zh` / `en` / `auto`），是否包含在 key 中由 `key_include_source_lang` 控制；
-- `tgt`：目标语言，如 `en` / `zh`；
-- `sha256(payload)`：对以下内容整体做 SHA-256：
-  - `model`
-  - `src` / `tgt`
-  - `scene`（受 `key_include_scene` 控制）
-  - 原始 `text`
-
-> 注意：所有 key 设计集中在 `_build_cache_key`，**不要在其他位置手动拼翻译缓存 key**。
+- `model`：capability 名称，如 `qwen-mt`、`llm`、`opus-mt-zh-en`
+- `target_lang`：目标语言，如 `en` / `zh`
+- `source_text[:4]`：原文前 4 个字符
+- `sha256(source_text)`：对完整原文做 SHA-256
 ### 3.2 Value 与类型
@@ -115,20 +109,25 @@
 services:
   translation:
     cache:
-      enabled: true
-      key_prefix: "trans:v2"
       ttl_seconds: 62208000        # 默认约 720 天
       sliding_expiration: true
-      key_include_scene: true
-      key_include_source_lang: true
+    capabilities:
+      qwen-mt:
+        use_cache: true
+      llm:
+        use_cache: true
+      deepl:
+        use_cache: true
+      nllb-200-distilled-600m:
+        use_cache: true
+      opus-mt-zh-en:
+        use_cache: true
+      opus-mt-en-zh:
+        use_cache: true
 ```
 - 运行时行为：
-  - 创建 `Translator` 时，从 `cache_cfg` 读取：
-    - `self.cache_prefix`
-    - `self.expire_seconds`
-    - `self.cache_sliding_expiration`
-    - `self.cache_include_*` 一系列布尔开关；
+  - translator service 启动时初始化共享 Redis cache；
   - **读缓存**：
     - 命中后，若 `sliding_expiration=True`，会调用 `redis.expire(key, expire_seconds)`；
   - **写缓存**：
@@ -136,8 +135,8 @@ services:
 ### 3.4 关联模块
-- `api/translator_app.py` 会通过 `translation.backends.qwen_mt.QwenMTTranslationBackend` 复用同一套缓存逻辑；
-- 文档说明：`docs/翻译模块说明.md` 中提到“推荐通过 Redis 翻译缓存复用结果”。
+- `api/translator_app.py` 通过 `TranslationService` 统一复用同一套缓存逻辑；
+- 所有翻译后端都通过 `TranslationService` 接入缓存。
 ---
@@ -43,7 +43,6 @@ PROJECT_ROOT = Path(__file__).parent.parent.parent
 sys.path.insert(0, str(PROJECT_ROOT))
 from config.env_config import REDIS_CONFIG  # type: ignore
-from config.services_config import get_translation_cache_config  # type: ignore
 from embeddings.bf16 import decode_embedding_from_redis  # type: ignore
@@ -66,13 +65,11 @@ def _load_known_cache_types() -&gt; Dict[str, CacheTypeConfig]:
         description="文本向量缓存（embeddings/text_encoder.py）",
     )
-    # translation 缓存：prefix 来自 services.translation.cache.key_prefix
-    cache_cfg = get_translation_cache_config()
-    trans_prefix = cache_cfg.get("key_prefix", "trans:v2")
+    # translation 缓存：统一前缀 trans
     cache_types["translation"] = CacheTypeConfig(
         name="translation",
-        pattern=f"{trans_prefix}:*",
-        description="翻译结果缓存（query/qwen_mt_translate.Translator）",
+        pattern="trans:*",
+        description="翻译结果缓存（translation/service.py）",
     )
     # anchors 缓存：prefix 来自 REDIS_CONFIG['anchor_cache_prefix']（若存在），否则 product_anchors
@@ -400,4 +397,3 @@ def main() -&gt; None:
 if __name__ == "__main__":
     main()
-
@@ -625,12 +625,8 @@ def translator_client(monkeypatch):
                     }
                 },
                 "cache": {
-                    "enabled": True,
-                    "key_prefix": "trans:v2",
                     "ttl_seconds": 60,
                     "sliding_expiration": True,
-                    "key_include_scene": True,
-                    "key_include_source_lang": True,
                 },
             }
             self.available_models = ["qwen-mt"]
@@ -681,12 +677,8 @@ def test_translator_api_failure_returns_500(monkeypatch):
                     }
                 },
                 "cache": {
-                    "enabled": True,
-                    "key_prefix": "trans:v2",
                     "ttl_seconds": 60,
                     "sliding_expiration": True,
-                    "key_include_scene": True,
-                    "key_include_source_lang": True,
                 },
             }
             self.available_models = ["qwen-mt"]
@@ -96,7 +96,7 @@ def test_nllb_uses_src_lang_and_forced_bos(monkeypatch):
     assert backend.seq2seq_model.last_generate_kwargs["forced_bos_token_id"] == 202
-def test_translation_service_lazy_loads_enabled_backends(monkeypatch):
+def test_translation_service_preloads_enabled_backends(monkeypatch):
     created = []
     def _fake_create_backend(self, *, name, backend_type, cfg):
@@ -126,6 +126,7 @@ def test_translation_service_lazy_loads_enabled_backends(monkeypatch):
             "opus-mt-en-zh": {
                 "enabled": True,
                 "backend": "local_marian",
+                "use_cache": True,
                 "model_id": "dummy",
                 "model_dir": "dummy",
                 "device": "cpu",
@@ -138,6 +139,7 @@ def test_translation_service_lazy_loads_enabled_backends(monkeypatch):
             "nllb-200-distilled-600m": {
                 "enabled": True,
                 "backend": "local_nllb",
+                "use_cache": True,
                 "model_id": "dummy",
                 "model_dir": "dummy",
                 "device": "cpu",
@@ -149,22 +151,19 @@ def test_translation_service_lazy_loads_enabled_backends(monkeypatch):
             },
         },
         "cache": {
-            "enabled": True,
-            "key_prefix": "trans:v2",
             "ttl_seconds": 60,
             "sliding_expiration": True,
-            "key_include_scene": True,
-            "key_include_source_lang": True,
         },
     }
     service = TranslationService(config)
     assert service.available_models == ["opus-mt-en-zh", "nllb-200-distilled-600m"]
-    assert service.loaded_models == []
+    assert service.loaded_models == ["opus-mt-en-zh", "nllb-200-distilled-600m"]
+    assert created == [
+        ("opus-mt-en-zh", "local_marian"),
+        ("nllb-200-distilled-600m", "local_nllb"),
+    ]
     backend = service.get_backend("opus-mt-en-zh")
-
     assert backend.model == "opus-mt-en-zh"
-    assert created == [("opus-mt-en-zh", "local_marian")]
-    assert service.loaded_models == ["opus-mt-en-zh"]
-from translation.backends.qwen_mt import QwenMTTranslationBackend
+from translation.cache import TranslationCache
+from translation.service import TranslationService
-class _RecordingRedis:
+class _FakeCache:
     def __init__(self):
-        self.setex_calls = []
-
-    def setex(self, key, ttl, value):
-        self.setex_calls.append((key, ttl, value))
-
-
-def test_translate_failure_returns_none_and_skips_cache(monkeypatch):
-    translator = QwenMTTranslationBackend(
-        capability_name="qwen-mt",
-        model="qwen-mt-flash",
-        base_url="https://dashscope-us.aliyuncs.com/compatible-mode/v1",
-        api_key="dummy-key",
-        use_cache=False,
-    )
-    fake_redis = _RecordingRedis()
-    translator.use_cache = True
-    translator.redis_client = fake_redis
-    translator.cache_prefix = "trans"
-    translator.expire_seconds = 60
-
-    monkeypatch.setattr(translator, "_translate_qwen", lambda *args, **kwargs: None)
-
-    result = translator.translate(
-        text="商品标题",
-        target_lang="en",
-        source_lang="zh",
-        scene="sku_name",
-    )
-
-    assert result is None
-    assert fake_redis.setex_calls == []
+        self.available = True
+        self.storage = {}
+        self.get_calls = []
+        self.set_calls = []
+
+    def get(self, *, model, target_lang, source_text):
+        self.get_calls.append((model, target_lang, source_text))
+        return self.storage.get((model, target_lang, source_text))
+
+    def set(self, *, model, target_lang, source_text, translated_text):
+        self.set_calls.append((model, target_lang, source_text, translated_text))
+        self.storage[(model, target_lang, source_text)] = translated_text
+
+
+def test_translation_cache_key_format(monkeypatch):
+    monkeypatch.setattr(TranslationCache, "_init_redis_client", staticmethod(lambda: None))
+    cache = TranslationCache({"ttl_seconds": 60, "sliding_expiration": True})
+    key = cache.build_key(model="llm", target_lang="en", source_text="商品标题")
+    assert key.startswith("trans:llm:en:商品标题")
+    assert len(key) == len("trans:llm:en:商品标题") + 64
+
+
+def test_service_caches_all_capabilities(monkeypatch):
+    monkeypatch.setattr(TranslationCache, "_init_redis_client", staticmethod(lambda: None))
+    created = {}
+
+    def _fake_create_backend(self, *, name, backend_type, cfg):
+        del self, backend_type, cfg
+
+        class _Backend:
+            model = name
+
+            @property
+            def supports_batch(self):
+                return True
+
+            def translate(self, text, target_lang, source_lang=None, scene=None):
+                del target_lang, source_lang, scene
+                if isinstance(text, list):
+                    return [f"{name}:{item}" for item in text]
+                return f"{name}:{text}"
+
+        backend = _Backend()
+        created[name] = backend
+        return backend
+
+    monkeypatch.setattr(TranslationService, "_create_backend", _fake_create_backend)
+    config = {
+        "service_url": "http://127.0.0.1:6006",
+        "timeout_sec": 10.0,
+        "default_model": "llm",
+        "default_scene": "general",
+        "capabilities": {
+            "llm": {
+                "enabled": True,
+                "backend": "llm",
+                "model": "dummy-llm",
+                "base_url": "https://example.com",
+                "timeout_sec": 10.0,
+                "use_cache": True,
+            },
+            "opus-mt-zh-en": {
+                "enabled": True,
+                "backend": "local_marian",
+                "model_id": "dummy",
+                "model_dir": "dummy",
+                "device": "cpu",
+                "torch_dtype": "float32",
+                "batch_size": 8,
+                "max_input_length": 16,
+                "max_new_tokens": 16,
+                "num_beams": 1,
+                "use_cache": True,
+            },
+        },
+        "cache": {
+            "ttl_seconds": 60,
+            "sliding_expiration": True,
+        },
+    }
+
+    service = TranslationService(config)
+    fake_cache = _FakeCache()
+    service._translation_cache = fake_cache
+
+    first = service.translate("商品标题", target_lang="en", source_lang="zh", model="llm")
+    second = service.translate("商品标题", target_lang="en", source_lang="zh", model="llm")
+    batch = service.translate(["连衣裙", "衬衫"], target_lang="en", source_lang="zh", model="opus-mt-zh-en")
+
+    assert first == "llm:商品标题"
+    assert second == "llm:商品标题"
+    assert batch == ["opus-mt-zh-en:连衣裙", "opus-mt-zh-en:衬衫"]
+    assert fake_cache.get_calls == [
+        ("llm", "en", "商品标题"),
+        ("llm", "en", "商品标题"),
+        ("opus-mt-zh-en", "en", "连衣裙"),
+        ("opus-mt-zh-en", "en", "衬衫"),
+    ]
+    assert fake_cache.set_calls == [
+        ("llm", "en", "商品标题", "llm:商品标题"),
+        ("opus-mt-zh-en", "en", "连衣裙", "opus-mt-zh-en:连衣裙"),
+        ("opus-mt-zh-en", "en", "衬衫", "opus-mt-zh-en:衬衫"),
+    ]
@@ -75,12 +75,8 @@ services:
     default_scene: "general"
     timeout_sec: 10.0
     cache:
-      enabled: true
-      key_prefix: "trans:v2"
       ttl_seconds: 62208000
       sliding_expiration: true
-      key_include_scene: true
-      key_include_source_lang: true
     capabilities:
       qwen-mt:
         enabled: true
@@ -95,11 +91,13 @@ services:
         model: "qwen-flash"
         base_url: "https://dashscope-us.aliyuncs.com/compatible-mode/v1"
         timeout_sec: 30.0
+        use_cache: true
       deepl:
         enabled: false
         backend: "deepl"
         api_url: "https://api.deepl.com/v2/translate"
         timeout_sec: 10.0
+        use_cache: true
       nllb-200-distilled-600m:
         enabled: true
         backend: "local_nllb"
@@ -112,6 +110,7 @@ services:
         max_new_tokens: 64
         num_beams: 1
         attn_implementation: "sdpa"
+        use_cache: true
       opus-mt-zh-en:
         enabled: true
         backend: "local_marian"
@@ -123,6 +122,7 @@ services:
         max_input_length: 256
         max_new_tokens: 256
         num_beams: 1
+        use_cache: true
       opus-mt-en-zh:
         enabled: true
         backend: "local_marian"
@@ -134,6 +134,7 @@ services:
         max_input_length: 256
         max_new_tokens: 256
         num_beams: 1
+        use_cache: true
 ```
 配置边界：
@@ -247,16 +248,20 @@ TRANSLATION_PORT=6006
 ```json
 {
-  "status": "healthy",
+ "status": "healthy",
   "service": "translation",
   "default_model": "llm",
   "default_scene": "general",
   "available_models": ["qwen-mt", "llm", "nllb-200-distilled-600m", "opus-mt-zh-en", "opus-mt-en-zh"],
   "enabled_capabilities": ["qwen-mt", "llm", "nllb-200-distilled-600m", "opus-mt-zh-en", "opus-mt-en-zh"],
-  "loaded_models": ["llm"]
+  "loaded_models": ["qwen-mt", "llm", "nllb-200-distilled-600m", "opus-mt-zh-en", "opus-mt-en-zh"]
 }
 ```
+说明：
+- translator service 进程启动时会一次性初始化全部已启用 capability
+- 因此本地模型加载失败、依赖缺失、配置错误会在启动阶段直接暴露，而不是拖到首个在线请求
+
 ## 7. 代码调用方式
 业务侧统一这样调用：
@@ -317,6 +322,7 @@ results = translator.translate(
 - 通用大模型翻译
 - 根据 `scene` 生成内部 prompt
 - 更灵活，但成本和稳定性取决于上游模型
+- 支持 Redis 翻译缓存
 ### 8.3 DeepL
@@ -327,6 +333,7 @@ results = translator.translate(
 - 商业翻译 API
 - scene 会映射到内部上下文
 - 当前默认关闭
+- 支持 Redis 翻译缓存
 ### 8.4 `facebook/nllb-200-distilled-600M`
@@ -338,6 +345,7 @@ results = translator.translate(
 - 简介：多语种翻译：覆盖约 200 种语言。作为NLLB-200系列的蒸馏版本，该模型通过知识蒸馏技术将原130亿参数模型压缩至600M，同时保持了80%以上的翻译质量。
 - 本地目录：`models/translation/facebook/nllb-200-distilled-600M`
 - 当前磁盘占用：约 `2.4G`
+- 支持 Redis 翻译缓存
 - 模型类型：多语种 Seq2Seq 机器翻译模型
 - 来源：Meta NLLB（No Language Left Behind）系列的 600M 蒸馏版
 - 结构特点：
@@ -424,6 +432,7 @@ results = translator.translate(
 - encoder-decoder Seq2Seq
 - 聚焦特定语言对
 - 模型更小、加载更轻、吞吐更高
+- 支持 Redis 翻译缓存
 ### 8.6 `opus-mt-en-zh`
@@ -441,6 +450,13 @@ results = translator.translate(
 - encoder-decoder Seq2Seq
 - 双语定向模型
 - 更适合中英双向拆分部署
+- 支持 Redis 翻译缓存
+
+### 8.7 翻译缓存
+
+- 所有 translation capability 都使用统一的 Redis 缓存层
+- 每个 capability 通过各自的 `use_cache` 控制是否启用缓存
+- 缓存 key 格式固定为 `trans:{model}:{target_lang}:{source_text[:4]}{sha256}`
 ## 9. 本地模型安装与部署
-"""Qwen-MT translation backend with cache support."""
+"""Qwen-MT translation backend."""
 from __future__ import annotations
-import hashlib
 import logging
 import os
 import re
 import time
 from typing import List, Optional, Sequence, Union
-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.env_config import DASHSCOPE_API_KEY
 from translation.languages import QWEN_LANGUAGE_CODES
 logger = logging.getLogger(__name__)
@@ -26,7 +23,6 @@ class QwenMTTranslationBackend:
         model: str,
         base_url: str,
         api_key: Optional[str] = None,
-        use_cache: bool = True,
         timeout: int = 10,
         glossary_id: Optional[str] = None,
     ):
@@ -35,16 +31,8 @@ class QwenMTTranslationBackend:
         self.qwen_model_name = self._normalize_model_name(model)
         self.base_url = base_url
         self.timeout = int(timeout)
-        self.use_cache = bool(use_cache)
         self.glossary_id = glossary_id
-        cache_cfg = get_translation_cache_config()
-        self.cache_prefix = str(cache_cfg["key_prefix"])
-        self.expire_seconds = int(cache_cfg["ttl_seconds"])
-        self.cache_sliding_expiration = bool(cache_cfg["sliding_expiration"])
-        self.cache_include_scene = bool(cache_cfg["key_include_scene"])
-        self.cache_include_source_lang = bool(cache_cfg["key_include_source_lang"])
-
         self._api_key = api_key or self._default_api_key(self.model)
         self._qwen_client: Optional[OpenAI] = None
         if self._api_key:
@@ -55,10 +43,6 @@ class QwenMTTranslationBackend:
         else:
             logger.warning("DASHSCOPE_API_KEY not set; qwen-mt translation unavailable")
-        self.redis_client = None
-        if self.use_cache and bool(cache_cfg["enabled"]):
-            self.redis_client = self._init_redis_client()
-
     @property
     def supports_batch(self) -> bool:
         return True
@@ -82,38 +66,6 @@ class QwenMTTranslationBackend:
         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],
-        scene: Optional[str],
-    ) -> str:
-        src = (source_lang or "auto").strip().lower() if self.cache_include_source_lang else "-"
-        tgt = (target_lang or "").strip().lower()
-        scn = (scene or "").strip() if self.cache_include_scene else ""
-        payload = f"model={self.model}\nsrc={src}\ntgt={tgt}\nscene={scn}\ntext={text}"
-        digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()
-        return f"{self.cache_prefix}:{self.model}:{src}:{tgt}:{digest}"
-
     def translate(
         self,
         text: Union[str, Sequence[str]],
@@ -146,14 +98,7 @@ class QwenMTTranslationBackend:
         if tgt == "zh" and (self._contains_chinese(text) or self._is_pure_number(text)):
             return text
-        cached = self._get_cached_translation_redis(text, tgt, src, scene)
-        if cached is not None:
-            return cached
-
         result = self._translate_qwen(text, tgt, src)
-
-        if result is not None:
-            self._set_cached_translation_redis(text, tgt, result, src, scene)
         return result
     def _translate_qwen(
@@ -197,41 +142,6 @@ class QwenMTTranslationBackend:
             )
             return None
-    def _get_cached_translation_redis(
-        self,
-        text: str,
-        target_lang: str,
-        source_lang: Optional[str] = None,
-        scene: Optional[str] = None,
-    ) -> Optional[str]:
-        if not self.redis_client:
-            return None
-        key = self._build_cache_key(text, target_lang, source_lang, scene)
-        try:
-            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
-
-    def _set_cached_translation_redis(
-        self,
-        text: str,
-        target_lang: str,
-        translation: str,
-        source_lang: Optional[str] = None,
-        scene: Optional[str] = None,
-    ) -> None:
-        if not self.redis_client:
-            return
-        key = self._build_cache_key(text, target_lang, source_lang, scene)
-        try:
-            self.redis_client.setex(key, self.expire_seconds, translation)
-        except Exception as exc:
-            logger.warning("Redis set translation cache failed: %s", exc)
-
     @staticmethod
     def _contains_chinese(text: str) -> bool:
         return bool(re.search(r"[\u4e00-\u9fff]", text or ""))
@@ -0,0 +1,92 @@
+"""Shared translation cache utilities."""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+from typing import Mapping, Optional
+
+import redis
+
+from config.env_config import REDIS_CONFIG
+
+logger = logging.getLogger(__name__)
+
+
+class TranslationCache:
+    """Redis-backed cache shared by all translation capabilities."""
+
+    def __init__(self, config: Mapping[str, object]) -> None:
+        self.ttl_seconds = int(config["ttl_seconds"])
+        self.sliding_expiration = bool(config["sliding_expiration"])
+        self.redis_client = self._init_redis_client()
+
+    @property
+    def available(self) -> bool:
+        return self.redis_client is not None
+
+    def build_key(self, *, model: str, target_lang: str, source_text: str) -> str:
+        normalized_model = str(model or "").strip().lower()
+        normalized_target_lang = str(target_lang or "").strip().lower()
+        text = str(source_text or "")
+        text_prefix = text[:4]
+        digest = hashlib.sha256(text.encode("utf-8")).hexdigest()
+        return f"trans:{normalized_model}:{normalized_target_lang}:{text_prefix}{digest}"
+
+    def get(self, *, model: str, target_lang: str, source_text: str) -> Optional[str]:
+        if self.redis_client is None:
+            return None
+        key = self.build_key(model=model, target_lang=target_lang, source_text=source_text)
+        try:
+            value = self.redis_client.get(key)
+            logger.info(
+                "Translation cache %s | model=%s target_lang=%s text_len=%s key=%s",
+                "hit" if value is not None else "miss",
+                model,
+                target_lang,
+                len(str(source_text or "")),
+                key,
+            )
+            if value and self.sliding_expiration:
+                self.redis_client.expire(key, self.ttl_seconds)
+            return value
+        except Exception as exc:
+            logger.warning("Redis get translation cache failed: %s", exc)
+            return None
+
+    def set(self, *, model: str, target_lang: str, source_text: str, translated_text: str) -> None:
+        if self.redis_client is None:
+            return
+        key = self.build_key(model=model, target_lang=target_lang, source_text=source_text)
+        try:
+            self.redis_client.setex(key, self.ttl_seconds, translated_text)
+            logger.info(
+                "Translation cache write | model=%s target_lang=%s text_len=%s result_len=%s ttl_seconds=%s key=%s",
+                model,
+                target_lang,
+                len(str(source_text or "")),
+                len(str(translated_text or "")),
+                self.ttl_seconds,
+                key,
+            )
+        except Exception as exc:
+            logger.warning("Redis set translation cache failed: %s", exc)
+
+    @staticmethod
+    def _init_redis_client() -> Optional[redis.Redis]:
+        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
@@ -3,10 +3,10 @@
 from __future__ import annotations
 import logging
-import threading
 from typing import Dict, List, Optional
 from config.services_config import get_translation_config
+from translation.cache import TranslationCache
 from translation.protocols import TranslateInput, TranslateOutput, TranslationBackendProtocol
 from translation.settings import (
     TranslationConfig,
@@ -25,10 +25,10 @@ class TranslationService:
     def __init__(self, config: Optional[TranslationConfig] = None) -> None:
         self.config = config or get_translation_config()
         self._enabled_capabilities = self._collect_enabled_capabilities()
-        self._backends: Dict[str, TranslationBackendProtocol] = {}
-        self._backend_lock = threading.Lock()
         if not self._enabled_capabilities:
             raise ValueError("No enabled translation backends found in services.translation.capabilities")
+        self._translation_cache = TranslationCache(self.config["cache"])
+        self._backends = self._initialize_backends()
     def _collect_enabled_capabilities(self) -> Dict[str, Dict[str, object]]:
         enabled: Dict[str, Dict[str, object]] = {}
@@ -59,6 +59,25 @@ class TranslationService:
             raise ValueError(f"Unsupported translation backend '{backend_type}' for capability '{name}'")
         return factory(name=name, cfg=cfg)
+    def _initialize_backends(self) -> Dict[str, TranslationBackendProtocol]:
+        backends: Dict[str, TranslationBackendProtocol] = {}
+        for name, capability_cfg in self._enabled_capabilities.items():
+            backend_type = str(capability_cfg["backend"])
+            logger.info("Initializing translation backend | model=%s backend=%s", name, backend_type)
+            backends[name] = self._create_backend(
+                name=name,
+                backend_type=backend_type,
+                cfg=capability_cfg,
+            )
+            logger.info(
+                "Translation backend initialized | model=%s backend=%s use_cache=%s backend_model=%s",
+                name,
+                backend_type,
+                bool(capability_cfg.get("use_cache")),
+                getattr(backends[name], "model", name),
+            )
+        return backends
+
     def _create_qwen_mt_backend(self, *, name: str, cfg: Dict[str, object]) -> TranslationBackendProtocol:
         from translation.backends.qwen_mt import QwenMTTranslationBackend
@@ -67,7 +86,6 @@ class TranslationService:
             model=str(cfg["model"]).strip(),
             base_url=str(cfg["base_url"]).strip(),
             api_key=cfg.get("api_key"),
-            use_cache=bool(cfg["use_cache"]),
             timeout=int(cfg["timeout_sec"]),
             glossary_id=cfg.get("glossary_id"),
         )
@@ -138,26 +156,12 @@ class TranslationService:
     def get_backend(self, model: Optional[str] = None) -> TranslationBackendProtocol:
         normalized = normalize_translation_model(self.config, model)
-        capability_cfg = self._enabled_capabilities.get(normalized)
-        if capability_cfg is None:
+        backend = self._backends.get(normalized)
+        if backend is None:
             raise ValueError(
                 f"Translation model '{normalized}' is not enabled. "
                 f"Available models: {', '.join(self.available_models) or 'none'}"
             )
-        backend = self._backends.get(normalized)
-        if backend is not None:
-            return backend
-        with self._backend_lock:
-            backend = self._backends.get(normalized)
-            if backend is None:
-                backend_type = str(capability_cfg["backend"])
-                logger.info("Initializing translation backend | model=%s backend=%s", normalized, backend_type)
-                backend = self._create_backend(
-                    name=normalized,
-                    backend_type=backend_type,
-                    cfg=capability_cfg,
-                )
-                self._backends[normalized] = backend
         return backend
     def translate(
@@ -169,11 +173,176 @@ class TranslationService:
         model: Optional[str] = None,
         scene: Optional[str] = None,
     ) -> TranslateOutput:
-        backend = self.get_backend(model)
+        normalized_model = normalize_translation_model(self.config, model)
+        backend = self.get_backend(normalized_model)
         active_scene = normalize_translation_scene(self.config, scene)
-        return backend.translate(
+        capability_cfg = self._enabled_capabilities[normalized_model]
+        use_cache = bool(capability_cfg.get("use_cache"))
+        text_count = 1 if isinstance(text, str) else len(list(text))
+        logger.info(
+            "Translation route | model=%s backend=%s scene=%s target_lang=%s source_lang=%s count=%s use_cache=%s cache_available=%s",
+            normalized_model,
+            getattr(backend, "model", normalized_model),
+            active_scene,
+            target_lang,
+            source_lang or "auto",
+            text_count,
+            use_cache,
+            self._translation_cache.available,
+        )
+        if not use_cache or not self._translation_cache.available:
+            return backend.translate(
+                text=text,
+                target_lang=target_lang,
+                source_lang=source_lang,
+                scene=active_scene,
+            )
+
+        if isinstance(text, str):
+            return self._translate_with_cache(
+                backend,
+                text=text,
+                target_lang=target_lang,
+                source_lang=source_lang,
+                scene=active_scene,
+                model=normalized_model,
+            )
+
+        return self._translate_batch_with_cache(
             text=text,
             target_lang=target_lang,
             source_lang=source_lang,
+            backend=backend,
             scene=active_scene,
+            model=normalized_model,
+        )
+
+    def _translate_with_cache(
+        self,
+        backend: TranslationBackendProtocol,
+        *,
+        text: str,
+        target_lang: str,
+        source_lang: Optional[str],
+        scene: str,
+        model: str,
+    ) -> Optional[str]:
+        if not text.strip():
+            return text
+        cached = self._translation_cache.get(model=model, target_lang=target_lang, source_text=text)
+        if cached is not None:
+            logger.info(
+                "Translation cache served | model=%s scene=%s target_lang=%s source_lang=%s text_len=%s",
+                model,
+                scene,
+                target_lang,
+                source_lang or "auto",
+                len(text),
+            )
+            return cached
+        translated = backend.translate(
+            text=text,
+            target_lang=target_lang,
+            source_lang=source_lang,
+            scene=scene,
         )
+        if translated is not None:
+            self._translation_cache.set(
+                model=model,
+                target_lang=target_lang,
+                source_text=text,
+                translated_text=translated,
+            )
+            logger.info(
+                "Translation backend result cached | model=%s scene=%s target_lang=%s source_lang=%s text_len=%s result_len=%s",
+                model,
+                scene,
+                target_lang,
+                source_lang or "auto",
+                len(text),
+                len(str(translated)),
+            )
+        else:
+            logger.warning(
+                "Translation backend returned empty result | model=%s scene=%s target_lang=%s source_lang=%s text_len=%s",
+                model,
+                scene,
+                target_lang,
+                source_lang or "auto",
+                len(text),
+            )
+        return translated
+
+    def _translate_batch_with_cache(
+        self,
+        *,
+        text: TranslateInput,
+        target_lang: str,
+        source_lang: Optional[str],
+        backend: TranslationBackendProtocol,
+        scene: str,
+        model: str,
+    ) -> List[Optional[str]]:
+        texts = list(text)
+        results: List[Optional[str]] = [None] * len(texts)
+        misses: List[str] = []
+        miss_indices: List[int] = []
+        cache_hits = 0
+
+        for idx, item in enumerate(texts):
+            normalized_text = "" if item is None else str(item)
+            if not normalized_text.strip():
+                results[idx] = normalized_text
+                continue
+            cached = self._translation_cache.get(
+                model=model,
+                target_lang=target_lang,
+                source_text=normalized_text,
+            )
+            if cached is not None:
+                results[idx] = cached
+                cache_hits += 1
+                continue
+            misses.append(normalized_text)
+            miss_indices.append(idx)
+
+        logger.info(
+            "Translation batch cache summary | model=%s scene=%s target_lang=%s source_lang=%s total=%s cache_hits=%s cache_misses=%s",
+            model,
+            scene,
+            target_lang,
+            source_lang or "auto",
+            len(texts),
+            cache_hits,
+            len(misses),
+        )
+
+        if misses:
+            translated = backend.translate(
+                text=misses,
+                target_lang=target_lang,
+                source_lang=source_lang,
+                scene=scene,
+            )
+            translated_list = translated if isinstance(translated, list) else [translated]
+            for idx, original_text, translated_text in zip(miss_indices, misses, translated_list):
+                results[idx] = translated_text
+                if translated_text is not None:
+                    self._translation_cache.set(
+                        model=model,
+                        target_lang=target_lang,
+                        source_text=original_text,
+                        translated_text=translated_text,
+                    )
+                else:
+                    logger.warning(
+                        "Translation batch item returned empty result | model=%s scene=%s target_lang=%s source_lang=%s item_index=%s text_len=%s",
+                        model,
+                        scene,
+                        target_lang,
+                        source_lang or "auto",
+                        idx,
+                        len(original_text),
+                    )
+
+        return results
@@ -90,21 +90,11 @@ def _build_cache_config(raw_cache: Any) -&gt; Dict[str, Any]:
     if not isinstance(raw_cache, Mapping):
         raise ValueError("services.translation.cache must be a mapping")
     return {
-        "enabled": _require_bool(raw_cache.get("enabled"), "services.translation.cache.enabled"),
-        "key_prefix": _require_string(raw_cache.get("key_prefix"), "services.translation.cache.key_prefix"),
         "ttl_seconds": _require_positive_int(raw_cache.get("ttl_seconds"), "services.translation.cache.ttl_seconds"),
         "sliding_expiration": _require_bool(
             raw_cache.get("sliding_expiration"),
             "services.translation.cache.sliding_expiration",
         ),
-        "key_include_scene": _require_bool(
-            raw_cache.get("key_include_scene"),
-            "services.translation.cache.key_include_scene",
-        ),
-        "key_include_source_lang": _require_bool(
-            raw_cache.get("key_include_source_lang"),
-            "services.translation.cache.key_include_source_lang",
-        ),
     }
@@ -131,12 +121,12 @@ def _build_capabilities(raw_capabilities: Any) -&gt; Dict[str, Dict[str, Any]]:
 def _validate_capability(name: str, capability: Mapping[str, Any]) -> None:
     prefix = f"services.translation.capabilities.{name}"
     backend = capability.get("backend")
+    _require_bool(capability.get("use_cache"), f"{prefix}.use_cache")
     if backend == "qwen_mt":
         _require_string(capability.get("model"), f"{prefix}.model")
         _require_http_url(capability.get("base_url"), f"{prefix}.base_url")
         _require_positive_float(capability.get("timeout_sec"), f"{prefix}.timeout_sec")
-        _require_bool(capability.get("use_cache"), f"{prefix}.use_cache")
         return
     if backend == "llm":