1. 翻译性能优化

1）接口层 translator_app.py 支持并发调用翻译后端（http接口 /translate 处理函数调用results = _translate_batch 改为 results = await _run_translation_blocking） 2）translation/backends/llm.py 支持batch翻译 2. 缓存清理脚本： scripts/redis/purge_caches.py 删除 embedding_prefix:*、embed:*、anchor_prefix:*，以及 trans:* 但会跳过 trans:deepl* 1）dry run方法： source activate.sh && python scripts/redis/purge_caches.py --dry-run 2）真正的清空缓存：python scripts/redis/purge_caches.py

1. 翻译性能优化
1）接口层 translator_app.py 支持并发调用翻译后端（http接口 /translate 处理函数调用results = _translate_batch 改为 results = await _run_translation_blocking） 2）translation/backends/llm.py 支持batch翻译 2. 缓存清理脚本： scripts/redis/purge_caches.py 删除 embedding_prefix:*、embed:*、anchor_prefix:*，以及 trans:* 但会跳过 trans:deepl* 1）dry run方法： source activate.sh && python scripts/redis/purge_caches.py --dry-run 2）真正的清空缓存：python scripts/redis/purge_caches.py
tangwang
1 parent dba57642
Showing 6 changed files with 720 additions and 24 deletions Show diff stats
api/translator_app.py
scripts/redis/purge_caches.py
tests/test_translation_llm_backend.py
tests/test_translator_failure_semantics.py
translation/backends/llm.py
translation/prompts.py
 """Translator service HTTP app."""
 import argparse
+import asyncio
 import logging
 import os
 import pathlib
 import time
 import uuid
 from contextlib import asynccontextmanager
-from functools import lru_cache
+from functools import lru_cache, partial
 from logging.handlers import TimedRotatingFileHandler
 from typing import List, Optional, Union
+import anyio
 import uvicorn
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
@@ -92,6 +94,17 @@ def get_translation_service() -&gt; TranslationService:
     return TranslationService(get_translation_config())
+_TRANSLATION_MAX_INFLIGHT = int(os.getenv("TRANSLATION_MAX_INFLIGHT", "4"))
+_translation_inflight_semaphore = asyncio.Semaphore(max(1, _TRANSLATION_MAX_INFLIGHT))
+
+
+async def _run_translation_blocking(fn, /, *args, **kwargs):
+    """Run blocking translation work in a thread with bounded concurrency."""
+    bound = partial(fn, *args, **kwargs)
+    async with _translation_inflight_semaphore:
+        return await anyio.to_thread.run_sync(bound)
+
+
 # Request/Response models
 class TranslationRequest(BaseModel):
     """Translation request model."""
@@ -332,6 +345,7 @@ async def health_check():
         return {
             "status": "healthy",
             "service": "translation",
+            "max_inflight": _TRANSLATION_MAX_INFLIGHT,
             "default_model": service.config["default_model"],
             "default_scene": service.config["default_scene"],
             "available_models": service.available_models,
@@ -388,7 +402,8 @@ async def translate(request: TranslationRequest, http_request: Request):
         )
         if isinstance(raw_text, list):
-            results = _translate_batch(
+            results = await _run_translation_blocking(
+                _translate_batch,
                 service,
                 raw_text,
                 target_lang=request.target_lang,
@@ -422,7 +437,8 @@ async def translate(request: TranslationRequest, http_request: Request):
                 scene=scene,
             )
-        translated_text = service.translate(
+        translated_text = await _run_translation_blocking(
+            service.translate,
             text=raw_text,
             target_lang=request.target_lang,
             source_lang=request.source_lang,
@@ -0,0 +1,194 @@
+#!/usr/bin/env python3
+"""
+Purge Redis caches used by this repo (exclude trans:deepl*).
+
+Default behavior (db=0):
+- Delete embedding cache keys: {embedding_cache_prefix}:*
+  (includes :image: and :clip_text: namespaces)
+- Delete legacy embedding keys: embed:*  (older deployments wrote raw logical keys)
+- Delete anchors cache keys: {anchor_cache_prefix}:*
+- Delete translation cache keys: trans:* EXCEPT those starting with "trans:deepl"
+
+Usage:
+  source activate.sh
+  python scripts/redis/purge_caches.py --dry-run
+  python scripts/redis/purge_caches.py
+  python scripts/redis/purge_caches.py --db 1
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+from typing import Iterable, List, Optional
+
+import redis
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from config.env_config import REDIS_CONFIG  # type: ignore
+
+
+def get_redis_client(db: int) -> redis.Redis:
+    return redis.Redis(
+        host=REDIS_CONFIG.get("host", "localhost"),
+        port=REDIS_CONFIG.get("port", 6479),
+        password=REDIS_CONFIG.get("password"),
+        db=db,
+        decode_responses=True,
+        socket_timeout=10,
+        socket_connect_timeout=10,
+    )
+
+
+def iter_scan_keys(client: redis.Redis, pattern: str, scan_count: int = 2000) -> Iterable[str]:
+    cursor = 0
+    while True:
+        cursor, batch = client.scan(cursor=cursor, match=pattern, count=scan_count)
+        for k in batch:
+            yield k
+        if cursor == 0:
+            break
+
+
+def delete_keys(
+    *,
+    client: redis.Redis,
+    keys: Iterable[str],
+    dry_run: bool,
+    batch_size: int = 1000,
+) -> int:
+    deleted = 0
+    buf: List[str] = []
+
+    def flush() -> int:
+        nonlocal buf
+        if not buf:
+            return 0
+        if dry_run:
+            n = len(buf)
+            buf = []
+            return n
+        pipe = client.pipeline(transaction=False)
+        for k in buf:
+            pipe.delete(k)
+        results = pipe.execute()
+        n = int(sum(1 for r in results if isinstance(r, int) and r > 0))
+        buf = []
+        return n
+
+    for k in keys:
+        buf.append(k)
+        if len(buf) >= batch_size:
+            deleted += flush()
+    deleted += flush()
+    return deleted
+
+
+def build_tasks(
+    *,
+    embedding_prefix: str,
+    anchor_prefix: str,
+    keep_translation_prefix: str,
+    include_translation: bool,
+) -> List[dict]:
+    tasks = [
+        {"name": "embedding", "pattern": f"{embedding_prefix}:*", "exclude_prefix": None},
+        {"name": "embedding_legacy_embed", "pattern": "embed:*", "exclude_prefix": None},
+        {"name": "anchors", "pattern": f"{anchor_prefix}:*", "exclude_prefix": None},
+    ]
+    if include_translation:
+        tasks.append(
+            {
+                "name": "translation",
+                "pattern": "trans:*",
+                "exclude_prefix": keep_translation_prefix,
+            }
+        )
+    return tasks
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Purge Redis caches (skip trans:deepl*)")
+    parser.add_argument("--db", type=int, default=0, help="Redis database number (default: 0)")
+    parser.add_argument("--dry-run", action="store_true", help="Only count keys; do not delete")
+    parser.add_argument(
+        "--include-translation",
+        action="store_true",
+        default=True,
+        help="Also purge translation cache (default: true)",
+    )
+    parser.add_argument(
+        "--no-translation",
+        dest="include_translation",
+        action="store_false",
+        help="Do not purge translation cache",
+    )
+    parser.add_argument(
+        "--keep-translation-prefix",
+        type=str,
+        default="trans:deepl",
+        help='Do not delete translation keys starting with this prefix (default: "trans:deepl")',
+    )
+    parser.add_argument(
+        "--embedding-prefix",
+        type=str,
+        default=str(REDIS_CONFIG.get("embedding_cache_prefix", "embedding")),
+        help='Embedding cache prefix (default from REDIS_CONFIG["embedding_cache_prefix"])',
+    )
+    parser.add_argument(
+        "--anchor-prefix",
+        type=str,
+        default=str(REDIS_CONFIG.get("anchor_cache_prefix", "product_anchors")),
+        help='Anchors cache prefix (default from REDIS_CONFIG["anchor_cache_prefix"])',
+    )
+    parser.add_argument("--batch-size", type=int, default=1000, help="DEL pipeline batch size")
+    args = parser.parse_args()
+
+    client = get_redis_client(db=args.db)
+    client.ping()
+
+    tasks = build_tasks(
+        embedding_prefix=args.embedding_prefix,
+        anchor_prefix=args.anchor_prefix,
+        keep_translation_prefix=args.keep_translation_prefix,
+        include_translation=args.include_translation,
+    )
+
+    total_matched = 0
+    total_deleted = 0
+
+    for t in tasks:
+        pattern: str = t["pattern"]
+        exclude_prefix: Optional[str] = t["exclude_prefix"]
+
+        def filtered_keys() -> Iterable[str]:
+            for k in iter_scan_keys(client, pattern=pattern):
+                if exclude_prefix and k.startswith(exclude_prefix):
+                    continue
+                yield k
+
+        n = delete_keys(
+            client=client,
+            keys=filtered_keys(),
+            dry_run=args.dry_run,
+            batch_size=args.batch_size,
+        )
+        total_matched += n
+        if not args.dry_run:
+            total_deleted += n
+
+        action = "would delete" if args.dry_run else "deleted"
+        print(f"[{t['name']}] pattern={pattern} exclude_prefix={exclude_prefix!r} -> {action} {n:,} keys")
+
+    if args.dry_run:
+        print(f"\nDry run complete. Total keys that would be deleted: {total_matched:,}")
+    else:
+        print(f"\nPurge complete. Total keys deleted: {total_deleted:,}")
+
+
+if __name__ == "__main__":
+    main()
+
@@ -0,0 +1,89 @@
+from types import SimpleNamespace
+
+from translation.backends.llm import LLMTranslationBackend
+
+
+class _FakeCompletions:
+    def __init__(self, responses):
+        self.responses = list(responses)
+        self.calls = []
+
+    def create(self, *, model, messages, timeout):
+        self.calls.append(
+            {
+                "model": model,
+                "messages": messages,
+                "timeout": timeout,
+            }
+        )
+        content = self.responses.pop(0)
+        return SimpleNamespace(
+            choices=[
+                SimpleNamespace(
+                    message=SimpleNamespace(content=content),
+                )
+            ]
+        )
+
+
+def _build_backend(monkeypatch, responses):
+    fake_completions = _FakeCompletions(responses)
+    fake_client = SimpleNamespace(chat=SimpleNamespace(completions=fake_completions))
+    monkeypatch.setattr(LLMTranslationBackend, "_create_client", lambda self: fake_client)
+    backend = LLMTranslationBackend(
+        capability_name="llm",
+        model="test-model",
+        timeout_sec=5.0,
+        base_url="https://example.com",
+        api_key="test-key",
+    )
+    return backend, fake_completions
+
+
+def test_llm_translate_batch_uses_single_request(monkeypatch):
+    backend, fake_completions = _build_backend(monkeypatch, ["1. Dress\n2. Shirt"])
+
+    results = backend.translate(
+        ["连衣裙", "衬衫"],
+        target_lang="en",
+        source_lang="zh",
+        scene="sku_name",
+    )
+
+    assert results == ["Dress", "Shirt"]
+    assert len(fake_completions.calls) == 1
+    prompt = fake_completions.calls[0]["messages"][0]["content"]
+    assert "Output exactly one line for each input item, in the same order, using this exact format:" in prompt
+    assert "1. translation\n2. translation" in prompt
+    assert "1. 连衣裙\n2. 衬衫" in prompt
+
+
+def test_llm_translate_batch_falls_back_to_single_on_invalid_output(monkeypatch):
+    backend, fake_completions = _build_backend(monkeypatch, ["Dress\n2. Shirt", "Dress", "Shirt"])
+
+    results = backend.translate(
+        ["连衣裙", "衬衫"],
+        target_lang="en",
+        source_lang="zh",
+        scene="sku_name",
+    )
+
+    assert results == ["Dress", "Shirt"]
+    assert len(fake_completions.calls) == 3
+
+
+def test_llm_translate_batch_preserves_empty_items(monkeypatch):
+    backend, fake_completions = _build_backend(monkeypatch, ["1. Product"])
+
+    results = backend.translate(
+        [None, " ", "商品"],
+        target_lang="en",
+        source_lang="zh",
+        scene="general",
+    )
+
+    assert results == [None, " ", "Product"]
+    assert len(fake_completions.calls) == 1
+    prompt = fake_completions.calls[0]["messages"][0]["content"]
+    assert "1. 商品" in prompt
+    assert "Input:\n1. 商品" in prompt
@@ -120,6 +120,88 @@ def test_service_caches_all_capabilities(monkeypatch):
     ]
+def test_service_batch_only_sends_cache_misses_to_backend(monkeypatch):
+    monkeypatch.setattr(TranslationCache, "_init_redis_client", staticmethod(lambda: None))
+    backend_calls = []
+
+    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):
+                backend_calls.append(
+                    {
+                        "text": text,
+                        "target_lang": target_lang,
+                        "source_lang": source_lang,
+                        "scene": scene,
+                    }
+                )
+                if isinstance(text, list):
+                    return [f"{name}:{item}" for item in text]
+                return f"{name}:{text}"
+
+        return _Backend()
+
+    monkeypatch.setattr(TranslationService, "_create_backend", _fake_create_backend)
+    service = TranslationService(
+        {
+            "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,
+                }
+            },
+            "cache": {
+                "ttl_seconds": 60,
+                "sliding_expiration": True,
+            },
+        }
+    )
+
+    fake_cache = _FakeCache()
+    fake_cache.storage[("llm", "en", "商品1")] = "cached:商品1"
+    fake_cache.storage[("llm", "en", "商品3")] = "cached:商品3"
+    service._translation_cache = fake_cache
+
+    results = service.translate(
+        ["商品1", "商品2", "商品3", "商品4"],
+        target_lang="en",
+        source_lang="zh",
+        model="llm",
+        scene="sku_name",
+    )
+
+    assert results == [
+        "cached:商品1",
+        "llm:商品2",
+        "cached:商品3",
+        "llm:商品4",
+    ]
+    assert backend_calls == [
+        {
+            "text": ["商品2", "商品4"],
+            "target_lang": "en",
+            "source_lang": "zh",
+            "scene": "sku_name",
+        }
+    ]
+
+
 def test_translation_request_filter_injects_reqid():
     reqid, token = bind_translation_request_id("req-test-1234567890")
     try:
@@ -3,33 +3,48 @@
 from __future__ import annotations
 import logging
+import re
 import time
 from typing import List, Optional, Sequence, Union
 from openai import OpenAI
 from translation.languages import LANGUAGE_LABELS
-from translation.prompts import TRANSLATION_PROMPTS
+from translation.prompts import BATCH_TRANSLATION_PROMPTS, TRANSLATION_PROMPTS
 from translation.scenes import normalize_scene_name
 logger = logging.getLogger(__name__)
+_NUMBERED_LINE_RE = re.compile(r"^\s*(\d+)[\.\uFF0E]\s*(.*)\s*$")
-def _build_prompt(
-    text: str,
+def _resolve_prompt_template(
+    prompt_groups: dict[str, dict[str, str]],
     *,
-    source_lang: Optional[str],
     target_lang: str,
     scene: Optional[str],
-) -> str:
+) -> tuple[str, str, str]:
     tgt = str(target_lang or "").strip().lower()
-    src = str(source_lang or "auto").strip().lower() or "auto"
     normalized_scene = normalize_scene_name(scene)
-    group = TRANSLATION_PROMPTS[normalized_scene]
+    group = prompt_groups[normalized_scene]
     template = group.get(tgt) or group.get("en")
     if template is None:
         raise ValueError(f"Missing llm translation prompt for scene='{normalized_scene}' target_lang='{tgt}'")
+    return tgt, normalized_scene, template
+
+def _build_prompt(
+    text: str,
+    *,
+    source_lang: Optional[str],
+    target_lang: str,
+    scene: Optional[str],
+) -> str:
+    src = str(source_lang or "auto").strip().lower() or "auto"
+    tgt, _normalized_scene, template = _resolve_prompt_template(
+        TRANSLATION_PROMPTS,
+        target_lang=target_lang,
+        scene=scene,
+    )
     source_lang_label = LANGUAGE_LABELS.get(src, src)
     target_lang_label = LANGUAGE_LABELS.get(tgt, tgt)
@@ -42,6 +57,63 @@ def _build_prompt(
     )
+def _build_batch_prompt(
+    texts: Sequence[str],
+    *,
+    source_lang: Optional[str],
+    target_lang: str,
+    scene: Optional[str],
+) -> str:
+    src = str(source_lang or "auto").strip().lower() or "auto"
+    tgt, _normalized_scene, template = _resolve_prompt_template(
+        BATCH_TRANSLATION_PROMPTS,
+        target_lang=target_lang,
+        scene=scene,
+    )
+    source_lang_label = LANGUAGE_LABELS.get(src, src)
+    target_lang_label = LANGUAGE_LABELS.get(tgt, tgt)
+    numbered_input = "\n".join(f"{idx}. {item}" for idx, item in enumerate(texts, start=1))
+    format_example = "\n".join(f"{idx}. translation" for idx in range(1, len(texts) + 1))
+
+    return template.format(
+        source_lang=source_lang_label,
+        src_lang_code=src,
+        target_lang=target_lang_label,
+        tgt_lang_code=tgt,
+        item_count=len(texts),
+        format_example=format_example,
+        text=numbered_input,
+    )
+
+
+def _parse_batch_translation_output(content: str, *, expected_count: int) -> Optional[List[str]]:
+    numbered_lines: dict[int, str] = {}
+    for raw_line in content.splitlines():
+        stripped = raw_line.strip()
+        if not stripped or stripped.startswith("```"):
+            continue
+        match = _NUMBERED_LINE_RE.match(stripped)
+        if match is None:
+            logger.warning("[llm] Invalid batch line format | line=%s", raw_line)
+            return None
+        index = int(match.group(1))
+        if index in numbered_lines:
+            logger.warning("[llm] Duplicate batch line index | index=%s", index)
+            return None
+        numbered_lines[index] = match.group(2).strip()
+
+    expected_indices = set(range(1, expected_count + 1))
+    actual_indices = set(numbered_lines.keys())
+    if actual_indices != expected_indices:
+        logger.warning(
+            "[llm] Batch line indices mismatch | expected=%s actual=%s",
+            sorted(expected_indices),
+            sorted(actual_indices),
+        )
+        return None
+    return [numbered_lines[idx] for idx in range(1, expected_count + 1)]
+
+
 class LLMTranslationBackend:
     def __init__(
         self,
@@ -136,6 +208,150 @@ class LLMTranslationBackend:
             )
             return None
+    def _translate_batch_serial_fallback(
+        self,
+        texts: Sequence[Optional[str]],
+        target_lang: str,
+        source_lang: Optional[str] = None,
+        scene: Optional[str] = None,
+    ) -> List[Optional[str]]:
+        results: List[Optional[str]] = []
+        for item in texts:
+            if item is None:
+                results.append(None)
+                continue
+            normalized = str(item)
+            if not normalized.strip():
+                results.append(normalized)
+                continue
+            results.append(
+                self._translate_single(
+                    text=normalized,
+                    target_lang=target_lang,
+                    source_lang=source_lang,
+                    scene=scene,
+                )
+            )
+        return results
+
+    def _translate_batch(
+        self,
+        texts: Sequence[Optional[str]],
+        target_lang: str,
+        source_lang: Optional[str] = None,
+        scene: Optional[str] = None,
+    ) -> List[Optional[str]]:
+        results: List[Optional[str]] = [None] * len(texts)
+        prompt_texts: List[str] = []
+        prompt_positions: List[int] = []
+
+        for idx, item in enumerate(texts):
+            if item is None:
+                continue
+            normalized = str(item)
+            if not normalized.strip():
+                results[idx] = normalized
+                continue
+            if "\n" in normalized or "\r" in normalized:
+                logger.info("[llm] Batch fallback to serial | reason=multiline_input item_index=%s", idx)
+                return self._translate_batch_serial_fallback(
+                    texts=texts,
+                    target_lang=target_lang,
+                    source_lang=source_lang,
+                    scene=scene,
+                )
+            prompt_texts.append(normalized)
+            prompt_positions.append(idx)
+
+        if not prompt_texts:
+            return results
+        if not self.client:
+            return results
+
+        tgt = str(target_lang or "").strip().lower()
+        src = str(source_lang or "auto").strip().lower() or "auto"
+        if scene is None:
+            raise ValueError("llm translation scene is required")
+        normalized_scene = normalize_scene_name(scene)
+        user_prompt = _build_batch_prompt(
+            texts=prompt_texts,
+            source_lang=src,
+            target_lang=tgt,
+            scene=normalized_scene,
+        )
+
+        start = time.time()
+        try:
+            logger.info(
+                "[llm] Batch request | src=%s tgt=%s model=%s item_count=%s prompt=%s",
+                src,
+                tgt,
+                self.model,
+                len(prompt_texts),
+                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 batch result | src=%s tgt=%s item_count=%s latency=%.1fms",
+                    src,
+                    tgt,
+                    len(prompt_texts),
+                    latency_ms,
+                )
+                return self._translate_batch_serial_fallback(
+                    texts=texts,
+                    target_lang=target_lang,
+                    source_lang=source_lang,
+                    scene=scene,
+                )
+
+            parsed = _parse_batch_translation_output(content, expected_count=len(prompt_texts))
+            if parsed is None:
+                logger.warning(
+                    "[llm] Batch parse failed, fallback to serial | src=%s tgt=%s item_count=%s response=%s",
+                    src,
+                    tgt,
+                    len(prompt_texts),
+                    content,
+                )
+                return self._translate_batch_serial_fallback(
+                    texts=texts,
+                    target_lang=target_lang,
+                    source_lang=source_lang,
+                    scene=scene,
+                )
+
+            for position, translated in zip(prompt_positions, parsed):
+                results[position] = translated
+            logger.info(
+                "[llm] Batch success | src=%s tgt=%s item_count=%s response=%s latency=%.1fms",
+                src,
+                tgt,
+                len(prompt_texts),
+                content,
+                latency_ms,
+            )
+            return results
+        except Exception as exc:
+            latency_ms = (time.time() - start) * 1000
+            logger.warning(
+                "[llm] Batch failed | src=%s tgt=%s item_count=%s latency=%.1fms error=%s",
+                src,
+                tgt,
+                len(prompt_texts),
+                latency_ms,
+                exc,
+                exc_info=True,
+            )
+            return results
+
     def translate(
         self,
         text: Union[str, Sequence[str]],
@@ -144,20 +360,12 @@ class LLMTranslationBackend:
         scene: Optional[str] = None,
     ) -> Union[Optional[str], List[Optional[str]]]:
         if isinstance(text, (list, tuple)):
-            results: List[Optional[str]] = []
-            for item in text:
-                if item is None:
-                    results.append(None)
-                    continue
-                results.append(
-                    self._translate_single(
-                        text=str(item),
-                        target_lang=target_lang,
-                        source_lang=source_lang,
-                        scene=scene,
-                    )
-                )
-            return results
+            return self._translate_batch(
+                text,
+                target_lang=target_lang,
+                source_lang=source_lang,
+                scene=scene,
+            )
         return self._translate_single(
             text=str(text),
@@ -43,3 +43,110 @@ TRANSLATION_PROMPTS: Dict[str, Dict[str, str]] = {
         "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}",
     },
 }
+
+
+BATCH_TRANSLATION_PROMPTS: Dict[str, Dict[str, str]] = {
+    "general": {
+        "en": (
+            "Translate each item from {source_lang} ({src_lang_code}) to {target_lang} ({tgt_lang_code}). "
+            "Accurately and completely preserve the meaning of the original text, including key features and attributes.\n"
+            "Output exactly one line for each input item, in the same order, using this exact format:\n"
+            "1. translation\n"
+            "2. translation\n"
+            "...\n"
+            "Do not explain or output anything else.\n"
+            "Input:\n{text}"
+        ),
+        "zh": (
+            "将每一项从 {source_lang} ({src_lang_code}) 翻译为 {target_lang} ({tgt_lang_code})。"
+            "请准确完整地保留原文含义，包括关键特征和属性。\n"
+            "请按输入顺序逐行输出，每个输入对应一行，格式必须如下：\n"
+            "1. 翻译结果\n"
+            "2. 翻译结果\n"
+            "...\n"
+            "不要解释或输出其他任何内容。\n"
+            "输入：\n{text}"
+        ),
+        "ru": (
+            "Переведите каждый элемент с {source_lang} ({src_lang_code}) на {target_lang} ({tgt_lang_code}). "
+            "Точно и полностью сохраняйте смысл исходного текста, включая ключевые характеристики и свойства.\n"
+            "Выводите ровно по одной строке для каждого входного элемента в том же порядке, в следующем формате:\n"
+            "1. перевод\n"
+            "2. перевод\n"
+            "...\n"
+            "Не добавляйте объяснений и ничего лишнего.\n"
+            "Входные данные:\n{text}"
+        ),
+    },
+    "sku_name": {
+        "en": (
+            "Translate each item from {source_lang} ({src_lang_code}) to a {target_lang} ({tgt_lang_code}) product title.\n"
+            "Accurately and completely preserve the meaning of the original text, including key features and attributes.\n"
+            "As a product title, keep the style concise, clear, professional, and natural in {target_lang}.\n"
+            "Output exactly one line for each input item, in the same order, using this exact format:\n"
+            "1. translation\n"
+            "2. translation\n"
+            "...\n"
+            "Do not explain or output anything else.\n"
+            "Input:\n{text}"
+        ),
+        "zh": (
+            "将每一项从 {source_lang} ({src_lang_code}) 翻译为 {target_lang} ({tgt_lang_code}) 的商品标题。\n"
+            "请准确完整地保留原文含义，包括关键特征和属性。\n"
+            "作为商品标题，请保持简洁、清晰、专业，并符合 {target_lang} 的自然表达。\n"
+            "请按输入顺序逐行输出，每个输入对应一行，格式必须如下：\n"
+            "1. 翻译结果\n"
+            "2. 翻译结果\n"
+            "...\n"
+            "不要解释或输出其他任何内容。\n"
+            "输入：\n{text}"
+        ),
+        "ru": (
+            "Переведите каждый элемент с {source_lang} ({src_lang_code}) на {target_lang} ({tgt_lang_code}) в виде названия товара.\n"
+            "Точно и полностью сохраняйте смысл исходного текста, включая ключевые характеристики и свойства.\n"
+            "Как название товара, текст должен быть кратким, ясным, профессиональным и естественным на {target_lang}.\n"
+            "Выводите ровно по одной строке для каждого входного элемента в том же порядке, в следующем формате:\n"
+            "1. перевод\n"
+            "2. перевод\n"
+            "...\n"
+            "Не добавляйте объяснений и ничего лишнего.\n"
+            "Входные данные:\n{text}"
+        ),
+    },
+    "ecommerce_search_query": {
+        "en": (
+            "Translate each item from {source_lang} ({src_lang_code}) to a natural {target_lang} ({tgt_lang_code}) "
+            "ecommerce search query.\n"
+            "Accurately and completely preserve the meaning of the original text, including key features and attributes.\n"
+            "Use concise search-style wording.\n"
+            "Output exactly one line for each input item, in the same order, using this exact format:\n"
+            "1. translation\n"
+            "2. translation\n"
+            "...\n"
+            "Do not explain or output anything else.\n"
+            "Input:\n{text}"
+        ),
+        "zh": (
+            "将每一项从 {source_lang} ({src_lang_code}) 翻译为自然的 {target_lang} ({tgt_lang_code}) 电商搜索词。\n"
+            "请准确完整地保留原文含义，包括关键特征和属性。\n"
+            "使用简洁的搜索关键词风格表达。\n"
+            "请按输入顺序逐行输出，每个输入对应一行，格式必须如下：\n"
+            "1. 翻译结果\n"
+            "2. 翻译结果\n"
+            "...\n"
+            "不要解释或输出其他任何内容。\n"
+            "输入：\n{text}"
+        ),
+        "ru": (
+            "Переведите каждый элемент с {source_lang} ({src_lang_code}) на естественный поисковый запрос для электронной коммерции на {target_lang} ({tgt_lang_code}).\n"
+            "Точно и полностью сохраняйте смысл исходного текста, включая ключевые характеристики и свойства.\n"
+            "Используйте краткую форму, характерную для поисковых запросов.\n"
+            "Выводите ровно по одной строке для каждого входного элемента в том же порядке, в следующем формате:\n"
+            "1. перевод\n"
+            "2. перевод\n"
+            "...\n"
+            "Не добавляйте объяснений и ничего лишнего.\n"
+            "Входные данные:\n{text}"
+        ),
+    },
+}
 \ No newline at end of file