config optimize

tangwang
1 parent 77bfa7e3
Showing 37 changed files with 2087 additions and 1123 deletions Show diff stats
api/app.py
api/indexer_app.py
api/routes/admin.py
config/__init__.py
config/config.yaml
config/config_loader.py
config/dictionaries/query_rewrite.dict
config/env_config.py
config/loader.py
config/query_rewrite.dict
config/schema.py
config/services_config.py
config/tenant_config_loader.py
config/utils.py
docs/config-system-review-and-redesign.md
embeddings/config.py
embeddings/image_encoder.py
embeddings/redis_embedding_cache.py
embeddings/server.py
embeddings/text_encoder.py
@@ -86,8 +86,7 @@ limiter = Limiter(key_func=get_remote_address)
 # Add parent directory to path
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from config.env_config import ES_CONFIG, DB_CONFIG
-from config import ConfigLoader
+from config import get_app_config
 from utils import ESClient
 from search import Searcher
 from query import QueryParser
@@ -99,7 +98,7 @@ _es_client: Optional[ESClient] = None
 _searcher: Optional[Searcher] = None
 _query_parser: Optional[QueryParser] = None
 _suggestion_service: Optional[SuggestionService] = None
-_config = None
+_app_config = None
 def init_service(es_host: str = "http://localhost:9200"):
@@ -109,20 +108,20 @@ def init_service(es_host: str = &quot;http://localhost:9200&quot;):
     Args:
         es_host: Elasticsearch host URL
     """
-    global _es_client, _searcher, _query_parser, _suggestion_service, _config
+    global _es_client, _searcher, _query_parser, _suggestion_service, _app_config
     start_time = time.time()
     logger.info("Initializing search service (multi-tenant)")
     # Load configuration
     logger.info("Loading configuration...")
-    config_loader = ConfigLoader("config/config.yaml")
-    _config = config_loader.load_config()
+    _app_config = get_app_config()
+    search_config = _app_config.search
     logger.info("Configuration loaded")
     # Get ES credentials
-    es_username = os.getenv('ES_USERNAME') or ES_CONFIG.get('username')
-    es_password = os.getenv('ES_PASSWORD') or ES_CONFIG.get('password')
+    es_username = _app_config.infrastructure.elasticsearch.username
+    es_password = _app_config.infrastructure.elasticsearch.password
     # Connect to Elasticsearch
     logger.info(f"Connecting to Elasticsearch at {es_host}...")
@@ -139,15 +138,15 @@ def init_service(es_host: str = &quot;http://localhost:9200&quot;):
     # Initialize components
     logger.info("Initializing query parser...")
-    _query_parser = QueryParser(_config)
+    _query_parser = QueryParser(search_config)
     logger.info("Initializing searcher...")
-    _searcher = Searcher(_es_client, _config, _query_parser)
+    _searcher = Searcher(_es_client, search_config, _query_parser)
     logger.info("Initializing suggestion service...")
     _suggestion_service = SuggestionService(_es_client)
     elapsed = time.time() - start_time
-    logger.info(f"Search service ready! (took {elapsed:.2f}s) | Index: {_config.es_index_name}")
+    logger.info(f"Search service ready! (took {elapsed:.2f}s) | Index: {search_config.es_index_name}")
@@ -182,9 +181,9 @@ def get_suggestion_service() -&gt; SuggestionService:
 def get_config():
     """Get global config instance."""
-    if _config is None:
+    if _app_config is None:
         raise RuntimeError("Service not initialized")
-    return _config
+    return _app_config
 # Create FastAPI app with enhanced configuration
@@ -240,7 +239,7 @@ async def startup_event():
     except Exception as e:
         logger.warning(f"Failed to set thread pool size: {e}, using default")
-    es_host = os.getenv("ES_HOST", "http://localhost:9200")
+    es_host = get_app_config().infrastructure.elasticsearch.host
     logger.info("Starting E-Commerce Search API (Multi-Tenant)")
     logger.info(f"Elasticsearch Host: {es_host}")
@@ -38,8 +38,7 @@ logger = logging.getLogger(__name__)
 # Add parent directory to path
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from config.env_config import ES_CONFIG  # noqa: E402
-from config import ConfigLoader  # noqa: E402
+from config import get_app_config  # noqa: E402
 from utils import ESClient  # noqa: E402
 from utils.db_connector import create_db_connection  # noqa: E402
 from indexer.incremental_service import IncrementalIndexerService  # noqa: E402
@@ -55,7 +54,7 @@ from .service_registry import (
 _es_client: Optional[ESClient] = None
-_config = None
+_app_config = None
 _incremental_service: Optional[IncrementalIndexerService] = None
 _bulk_indexing_service: Optional[BulkIndexingService] = None
 _suggestion_builder: Optional[SuggestionIndexBuilder] = None
@@ -68,20 +67,19 @@ def init_indexer_service(es_host: str = &quot;http://localhost:9200&quot;):
     This mirrors the indexing-related initialization logic in api.app.init_service
     but without search-related components.
     """
-    global _es_client, _config, _incremental_service, _bulk_indexing_service, _suggestion_builder
+    global _es_client, _app_config, _incremental_service, _bulk_indexing_service, _suggestion_builder
     start_time = time.time()
     logger.info("Initializing Indexer service")
     # Load configuration (kept for parity/logging; indexer routes don't depend on it)
     logger.info("Loading configuration...")
-    config_loader = ConfigLoader("config/config.yaml")
-    _config = config_loader.load_config()
+    _app_config = get_app_config()
     logger.info("Configuration loaded")
     # Get ES credentials
-    es_username = os.getenv("ES_USERNAME") or ES_CONFIG.get("username")
-    es_password = os.getenv("ES_PASSWORD") or ES_CONFIG.get("password")
+    es_username = _app_config.infrastructure.elasticsearch.username
+    es_password = _app_config.infrastructure.elasticsearch.password
     # Connect to Elasticsearch
     logger.info(f"Connecting to Elasticsearch at {es_host} for indexer...")
@@ -97,11 +95,12 @@ def init_indexer_service(es_host: str = &quot;http://localhost:9200&quot;):
     set_es_client(_es_client)
     # Initialize indexing services (DB is required here)
-    db_host = os.getenv("DB_HOST")
-    db_port = int(os.getenv("DB_PORT", 3306))
-    db_database = os.getenv("DB_DATABASE")
-    db_username = os.getenv("DB_USERNAME")
-    db_password = os.getenv("DB_PASSWORD")
+    db_config = _app_config.infrastructure.database
+    db_host = db_config.host
+    db_port = db_config.port
+    db_database = db_config.database
+    db_username = db_config.username
+    db_password = db_config.password
     if all([db_host, db_database, db_username, db_password]):
         logger.info("Initializing database connection for indexing services...")
@@ -166,7 +165,7 @@ async def startup_event():
     except Exception as e:
         logger.warning(f"Failed to set thread pool size: {e}, using default")
-    es_host = os.getenv("ES_HOST", "http://localhost:9200")
+    es_host = get_app_config().infrastructure.elasticsearch.host
     logger.info("Starting Indexer API service")
     logger.info(f"Elasticsearch Host: {es_host}")
     try:
@@ -176,14 +175,11 @@ async def startup_event():
         # Eager warmup: build per-tenant transformer bundles at startup to avoid
         # first-request latency (config/provider/encoder + transformer wiring).
         try:
-            if _incremental_service is not None and _config is not None:
+            if _incremental_service is not None and _app_config is not None:
                 tenants = []
-                # config.tenant_config shape: {"default": {...}, "tenants": {"1": {...}, ...}}
-                tc = getattr(_config, "tenant_config", None) or {}
-                if isinstance(tc, dict):
-                    tmap = tc.get("tenants")
-                    if isinstance(tmap, dict):
-                        tenants = [str(k) for k in tmap.keys()]
+                tmap = _app_config.tenants.tenants
+                if isinstance(tmap, dict):
+                    tenants = [str(k) for k in tmap.keys()]
                 # If no explicit tenants configured, skip warmup.
                 if tenants:
                     warm = _incremental_service.warmup_transformers(tenants)
@@ -42,23 +42,32 @@ async def health_check():
 @router.get("/config")
 async def get_configuration():
     """
-    Get current search configuration (sanitized).
+    Get the effective application configuration (sanitized).
     """
     try:
         from ..app import get_config
-        config = get_config()
+        return get_config().sanitized_dict()
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/config/meta")
+async def get_configuration_meta():
+    """Get configuration metadata for observability."""
+    try:
+        from ..app import get_config
+
+        config = get_config()
         return {
-            "es_index_name": config.es_index_name,
-            "num_field_boosts": len(config.field_boosts),
-            "multilingual_fields": config.query_config.multilingual_fields,
-            "shared_fields": config.query_config.shared_fields,
-            "core_multilingual_fields": config.query_config.core_multilingual_fields,
-            "supported_languages": config.query_config.supported_languages,
-            "spu_enabled": config.spu_config.enabled
+            "environment": config.runtime.environment,
+            "config_hash": config.metadata.config_hash,
+            "loaded_files": list(config.metadata.loaded_files),
+            "deprecated_keys": list(config.metadata.deprecated_keys),
         }
-
     except HTTPException:
         raise
     except Exception as e:
-"""
-Configuration package for search engine.
+"""Unified configuration package exports."""
-Provides configuration loading, validation, and utility functions.
-"""
-
-from .config_loader import (
-    SearchConfig,
-    QueryConfig,
-    IndexConfig,
-    SPUConfig,
+from config.config_loader import ConfigLoader, ConfigurationError
+from config.loader import AppConfigLoader, get_app_config, reload_app_config
+from config.schema import (
+    AppConfig,
     FunctionScoreConfig,
+    IndexConfig,
+    QueryConfig,
     RerankConfig,
-    ConfigLoader,
-    ConfigurationError
-)
-
-from .utils import (
-    get_match_fields_for_index,
-    get_domain_fields
+    SPUConfig,
+    SearchConfig,
+    ServicesConfig,
 )
-from .services_config import (
-    get_translation_config,
-    get_embedding_config,
-    get_rerank_config,
+from config.services_config import (
     get_embedding_backend_config,
-    get_rerank_backend_config,
-    get_translation_base_url,
-    get_embedding_text_base_url,
+    get_embedding_config,
+    get_embedding_image_backend_config,
     get_embedding_image_base_url,
+    get_embedding_text_base_url,
+    get_rerank_backend_config,
+    get_rerank_config,
     get_rerank_service_url,
+    get_translation_base_url,
     get_translation_cache_config,
-    ServiceConfig,
+    get_translation_config,
 )
+from config.utils import get_domain_fields, get_match_fields_for_index
 __all__ = [
-    # Main config classes
-    'SearchConfig',
-    'QueryConfig',
-    'IndexConfig',
-    'SPUConfig',
-    'FunctionScoreConfig',
-    'RerankConfig',
-    
-    # Loader and utilities
-    'ConfigLoader',
-    'ConfigurationError',
-    'get_match_fields_for_index',
-    'get_domain_fields',
-    'get_translation_config',
-    'get_embedding_config',
-    'get_rerank_config',
-    'get_embedding_backend_config',
-    'get_rerank_backend_config',
-    'get_translation_base_url',
-    'get_embedding_text_base_url',
-    'get_embedding_image_base_url',
-    'get_rerank_service_url',
-    'get_translation_cache_config',
-    'ServiceConfig',
+    "AppConfig",
+    "AppConfigLoader",
+    "ConfigLoader",
+    "ConfigurationError",
+    "FunctionScoreConfig",
+    "IndexConfig",
+    "QueryConfig",
+    "RerankConfig",
+    "SPUConfig",
+    "SearchConfig",
+    "ServicesConfig",
+    "get_app_config",
+    "reload_app_config",
+    "get_domain_fields",
+    "get_match_fields_for_index",
+    "get_translation_config",
+    "get_embedding_config",
+    "get_rerank_config",
+    "get_embedding_backend_config",
+    "get_embedding_image_backend_config",
+    "get_rerank_backend_config",
+    "get_translation_base_url",
+    "get_embedding_text_base_url",
+    "get_embedding_image_base_url",
+    "get_rerank_service_url",
+    "get_translation_cache_config",
 ]
@@ -5,6 +5,10 @@
 # Elasticsearch Index
 es_index_name: "search_products"
+# Config assets
+assets:
+  query_rewrite_dictionary_path: "config/dictionaries/query_rewrite.dict"
+
 # ES Index Settings (基础设置)
 es_settings:
   number_of_shards: 1
@@ -211,6 +215,19 @@ services:
         device: "cuda"
         batch_size: 32
         normalize_embeddings: true
+    # 服务内图片后端（embedding 进程启动时读取）
+    image_backend: "clip_as_service"  # clip_as_service | local_cnclip
+    image_backends:
+      clip_as_service:
+        server: "grpc://127.0.0.1:51000"
+        model_name: "CN-CLIP/ViT-L-14"
+        batch_size: 8
+        normalize_embeddings: true
+      local_cnclip:
+        model_name: "ViT-L-14"
+        device: null
+        batch_size: 8
+        normalize_embeddings: true
   rerank:
     provider: "http"
     base_url: "http://127.0.0.1:6007"
@@ -218,6 +235,9 @@ services:
       http:
         base_url: "http://127.0.0.1:6007"
         service_url: "http://127.0.0.1:6007/rerank"
+    request:
+      max_docs: 1000
+      normalize: true
     # 服务内后端（reranker 进程启动时读取）
     backend: "qwen3_vllm"  # bge | qwen3_vllm | qwen3_transformers | dashscope_rerank
     backends:
 """
-Configuration loader and validator for search engine configurations.
+Compatibility wrapper for search-behavior config access.
-This module handles loading, parsing, and validating YAML configuration files
-that define how search should be executed (NOT how data should be indexed).
-
-索引结构由 mappings/search_products.json 定义。
-此配置只定义搜索行为：字段权重、搜索域、查询策略等。
+The unified loader lives in :mod:`config.loader`. This module now exposes the
+search subtree only, so existing search/indexer code can consume a single
+source-of-truth search config without reparsing YAML separately.
 """
-import yaml
-from typing import Dict, Any, List, Optional
-from dataclasses import dataclass, field
-from pathlib import Path
-
-
-@dataclass
-class IndexConfig:
-    """Configuration for an index domain (e.g., default, title, brand)."""
-    name: str
-    label: str
-    fields: List[str]  # List of field names to include in this search domain
-    boost: float = 1.0
-    example: Optional[str] = None
-
-
-@dataclass
-class QueryConfig:
-    """Configuration for query processing."""
-    supported_languages: List[str] = field(default_factory=lambda: ["zh", "en"])
-    default_language: str = "en"
-    
-    # Feature flags
-    enable_text_embedding: bool = True
-    enable_query_rewrite: bool = True
-
-    # Query rewrite dictionary (loaded from external file)
-    rewrite_dictionary: Dict[str, str] = field(default_factory=dict)
-    
-    # Embedding field names
-    text_embedding_field: Optional[str] = "title_embedding"
-    image_embedding_field: Optional[str] = None
-        
-    # Source fields configuration
-    source_fields: Optional[List[str]] = None
-    
-    # KNN boost configuration
-    knn_boost: float = 0.25  # Boost value for KNN (embedding recall)
-    
-    # Dynamic text fields for multi-language retrieval
-    multilingual_fields: List[str] = field(
-        default_factory=lambda: ["title", "brief", "description", "vendor", "category_path", "category_name_text"]
-    )
-    shared_fields: List[str] = field(
-        default_factory=lambda: ["tags", "option1_values", "option2_values", "option3_values"]
-    )
-    core_multilingual_fields: List[str] = field(
-        default_factory=lambda: ["title", "brief", "vendor", "category_name_text"]
-    )
-    
-    # Unified text strategy tuning
-    base_minimum_should_match: str = "75%"
-    translation_minimum_should_match: str = "75%"
-    translation_boost: float = 0.4
-    translation_boost_when_source_missing: float = 1.0
-    source_boost_when_missing: float = 0.6
-    original_query_fallback_boost_when_translation_missing: float = 0.2
-    tie_breaker_base_query: float = 0.9
-
-    # Query-time translation model selection (configurable)
-    # - zh_to_en_model: model for zh -> en
-    # - en_to_zh_model: model for en -> zh
-    # - default_translation_model: fallback model for all other language pairs
-    zh_to_en_model: str = "opus-mt-zh-en"
-    en_to_zh_model: str = "opus-mt-en-zh"
-    default_translation_model: str = "nllb-200-distilled-600m"
-
-
-@dataclass
-class SPUConfig:
-    """Configuration for SPU aggregation."""
-    enabled: bool = False
-    spu_field: Optional[str] = None
-    inner_hits_size: int = 3
-    # 配置哪些option维度参与检索（进索引、以及在线搜索）
-    searchable_option_dimensions: List[str] = field(default_factory=lambda: ['option1', 'option2', 'option3'])
-
-
-@dataclass
-class FunctionScoreConfig:
-    """Function Score配置（ES层打分规则）"""
-    score_mode: str = "sum"
-    boost_mode: str = "multiply"
-    functions: List[Dict[str, Any]] = field(default_factory=list)
-
-
-@dataclass
-class RerankConfig:
-    """重排配置（provider/URL 在 services.rerank）"""
-    enabled: bool = True
-    rerank_window: int = 384
-    timeout_sec: float = 15.0
-    weight_es: float = 0.4
-    weight_ai: float = 0.6
-    rerank_query_template: str = "{query}"
-    rerank_doc_template: str = "{title}"
-
-
-@dataclass
-class SearchConfig:
-    """Complete configuration for search engine (multi-tenant)."""
-    
-    # 字段权重配置（用于搜索）
-    field_boosts: Dict[str, float]
-    
-    # Legacy index domains (deprecated; kept for compatibility)
-    indexes: List[IndexConfig]
-    
-    # Query processing
-    query_config: QueryConfig
-    
-    # Function Score configuration (ES层打分)
-    function_score: FunctionScoreConfig
-    
-    # Rerank configuration (本地重排)
-    rerank: RerankConfig
-    
-    # SPU configuration
-    spu_config: SPUConfig
-    
-    # ES index settings
-    es_index_name: str
-    
-    # Tenant configuration
-    tenant_config: Dict[str, Any] = field(default_factory=dict)
-    
-    # ES settings
-    es_settings: Dict[str, Any] = field(default_factory=dict)
-    # Extensible service/provider registry (translation/embedding/rerank/...)
-    services: Dict[str, Any] = field(default_factory=dict)
+from __future__ import annotations
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any, Dict, List, Optional
-class ConfigurationError(Exception):
-    """Raised when configuration validation fails."""
-    pass
+from config.loader import AppConfigLoader, ConfigurationError
+from config.schema import (
+    FunctionScoreConfig,
+    IndexConfig,
+    QueryConfig,
+    RerankConfig,
+    SPUConfig,
+    SearchConfig,
+)
 class ConfigLoader:
-    """Loads and validates unified search engine configuration from YAML file."""
-    
-    def __init__(self, config_file: Optional[Path] = None):
-        """
-        Initialize config loader.
-        
-        Args:
-            config_file: Path to config YAML file (defaults to config/config.yaml)
-        """
-        if config_file is None:
-            config_file = Path(__file__).parent / "config.yaml"
-        self.config_file = Path(config_file)
-    
-    def _load_rewrite_dictionary(self) -> Dict[str, str]:
-        """Load query rewrite dictionary from external file."""
-        rewrite_file = Path(__file__).parent / "rewrite_dictionary.txt"
-        rewrite_dict = {}
-        
-        if not rewrite_file.exists():
-            return rewrite_dict
-        
-        try:
-            with open(rewrite_file, 'r', encoding='utf-8') as f:
-                for line in f:
-                    line = line.strip()
-                    if not line or line.startswith('#'):
-                        continue
-                    
-                    parts = line.split('\t')
-                    if len(parts) >= 2:
-                        original = parts[0].strip()
-                        replacement = parts[1].strip()
-                        if original and replacement:
-                            rewrite_dict[original] = replacement
-        except Exception as e:
-            print(f"Warning: Failed to load rewrite dictionary: {e}")
-        
-        return rewrite_dict
-    
-    def load_config(self, validate: bool = True) -> SearchConfig:
-        """
-        Load unified configuration from YAML file.
-        
-        Args:
-            validate: Whether to validate configuration after loading
-        
-        Returns:
-            SearchConfig object
-        
-        Raises:
-            ConfigurationError: If config file not found, invalid, or validation fails
-        """
-        if not self.config_file.exists():
-            raise ConfigurationError(f"Configuration file not found: {self.config_file}")
-        
-        try:
-            with open(self.config_file, 'r', encoding='utf-8') as f:
-                config_data = yaml.safe_load(f)
-        except yaml.YAMLError as e:
-            raise ConfigurationError(f"Invalid YAML in {self.config_file}: {e}")
-        
-        config = self._parse_config(config_data)
-        
-        # Auto-validate configuration
-        if validate:
-            errors = self.validate_config(config)
-            if errors:
-                error_msg = "Configuration validation failed:\n" + "\n".join(f"  - {err}" for err in errors)
-                raise ConfigurationError(error_msg)
-        
-        return config
-    
-    def _parse_config(self, config_data: Dict[str, Any]) -> SearchConfig:
-        """Parse configuration dictionary into SearchConfig object."""
-        
-        # Parse field_boosts
-        field_boosts = config_data.get("field_boosts", {})
-        if not isinstance(field_boosts, dict):
-            raise ConfigurationError("field_boosts must be a dictionary")
-        
-        # Parse indexes (deprecated; compatibility only)
-        indexes = []
-        for index_data in config_data.get("indexes", []):
-            indexes.append(self._parse_index_config(index_data))
-        
-        # Parse query config
-        query_config_data = config_data.get("query_config", {})
-        rewrite_dictionary = self._load_rewrite_dictionary()
-        search_fields_cfg = query_config_data.get("search_fields", {})
-        text_strategy_cfg = query_config_data.get("text_query_strategy", {})
+    """Load the unified app config and return the search subtree."""
-        query_config = QueryConfig(
-            supported_languages=query_config_data.get("supported_languages") or ["zh", "en"],
-            default_language=query_config_data.get("default_language") or "en",
-            enable_text_embedding=query_config_data.get("enable_text_embedding", True),
-            enable_query_rewrite=query_config_data.get("enable_query_rewrite", True),
-            rewrite_dictionary=rewrite_dictionary,
-            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"),
-            knn_boost=query_config_data.get("knn_boost", 0.25),
-            multilingual_fields=search_fields_cfg.get(
-                "multilingual_fields",
-                ["title", "brief", "description", "vendor", "category_path", "category_name_text"],
-            ),
-            shared_fields=search_fields_cfg.get(
-                "shared_fields",
-                ["tags", "option1_values", "option2_values", "option3_values"],
-            ),
-            core_multilingual_fields=search_fields_cfg.get(
-                "core_multilingual_fields",
-                ["title", "brief", "vendor", "category_name_text"],
-            ),
-            base_minimum_should_match=str(text_strategy_cfg.get("base_minimum_should_match", "75%")),
-            translation_minimum_should_match=str(text_strategy_cfg.get("translation_minimum_should_match", "75%")),
-            translation_boost=float(text_strategy_cfg.get("translation_boost", 0.4)),
-            translation_boost_when_source_missing=float(
-                text_strategy_cfg.get("translation_boost_when_source_missing", 1.0)
-            ),
-            source_boost_when_missing=float(text_strategy_cfg.get("source_boost_when_missing", 0.6)),
-            original_query_fallback_boost_when_translation_missing=float(
-                text_strategy_cfg.get("original_query_fallback_boost_when_translation_missing", 0.2)
-            ),
-            tie_breaker_base_query=float(text_strategy_cfg.get("tie_breaker_base_query", 0.9)),
-            zh_to_en_model=str(query_config_data.get("zh_to_en_model") or "opus-mt-zh-en"),
-            en_to_zh_model=str(query_config_data.get("en_to_zh_model") or "opus-mt-en-zh"),
-            default_translation_model=str(
-                query_config_data.get("default_translation_model") or "nllb-200-distilled-600m"
-            ),
-        )
-        
-        # Parse Function Score configuration
-        fs_data = config_data.get("function_score", {})
-        function_score = FunctionScoreConfig(
-            score_mode=fs_data.get("score_mode") or "sum",
-            boost_mode=fs_data.get("boost_mode") or "multiply",
-            functions=fs_data.get("functions") or []
-        )
-        
-        # Parse Rerank (provider/URL in services.rerank)
-        rerank_data = config_data.get("rerank", {})
-        rerank = RerankConfig(
-            enabled=bool(rerank_data.get("enabled", True)),
-            rerank_window=int(rerank_data.get("rerank_window", 384)),
-            timeout_sec=float(rerank_data.get("timeout_sec", 15.0)),
-            weight_es=float(rerank_data.get("weight_es", 0.4)),
-            weight_ai=float(rerank_data.get("weight_ai", 0.6)),
-            rerank_query_template=str(rerank_data.get("rerank_query_template") or "{query}"),
-            rerank_doc_template=str(rerank_data.get("rerank_doc_template") or "{title}"),
-        )
-        
-        # Parse SPU config
-        spu_data = config_data.get("spu_config", {})
-        spu_config = SPUConfig(
-            enabled=spu_data.get("enabled", False),
-            spu_field=spu_data.get("spu_field"),
-            inner_hits_size=spu_data.get("inner_hits_size", 3),
-            searchable_option_dimensions=spu_data.get("searchable_option_dimensions", ['option1', 'option2', 'option3'])
-        )
-        
-        # Parse tenant config
-        tenant_config_data = config_data.get("tenant_config", {})
+    def __init__(self, config_file: Optional[Path] = None) -> None:
+        self._loader = AppConfigLoader(config_file=Path(config_file) if config_file is not None else None)
-        # Parse extensible services/provider registry
-        services_data = config_data.get("services", {}) or {}
-        if not isinstance(services_data, dict):
-            raise ConfigurationError("services must be a dictionary if provided")
+    def load_config(self, validate: bool = True) -> SearchConfig:
+        return self._loader.load(validate=validate).search
-        return SearchConfig(
-            field_boosts=field_boosts,
-            indexes=indexes,
-            query_config=query_config,
-            function_score=function_score,
-            rerank=rerank,
-            spu_config=spu_config,
-            tenant_config=tenant_config_data,
-            es_index_name=config_data.get("es_index_name", "search_products"),
-            es_settings=config_data.get("es_settings", {}),
-            services=services_data
-        )
-    
-    def _parse_index_config(self, index_data: Dict[str, Any]) -> IndexConfig:
-        """Parse index configuration from dictionary."""
-        return IndexConfig(
-            name=index_data["name"],
-            label=index_data.get("label", index_data["name"]),
-            fields=index_data.get("fields", []),
-            boost=index_data.get("boost", 1.0),
-            example=index_data.get("example")
-        )
-    
     def validate_config(self, config: SearchConfig) -> List[str]:
-        """
-        Validate configuration for common errors.
-        
-        Args:
-            config: SearchConfig to validate
-        
-        Returns:
-            List of error messages (empty if valid)
-        """
-        errors = []
-        
-        # Validate es_index_name
+        errors: List[str] = []
         if not config.es_index_name:
             errors.append("es_index_name is required")
-        
-        # Validate field_boosts
         if not config.field_boosts:
             errors.append("field_boosts is empty")
-        
-        for field_name, boost in config.field_boosts.items():
-            if not isinstance(boost, (int, float)):
-                errors.append(f"field_boosts['{field_name}']: boost must be a number, got {type(boost).__name__}")
-            elif boost < 0:
-                errors.append(f"field_boosts['{field_name}']: boost must be non-negative")
-        
-        # Validate indexes (deprecated, optional)
-        index_names = set()
-        for index in config.indexes:
-            # Check for duplicate index names
-            if index.name in index_names:
-                errors.append(f"Duplicate index name: {index.name}")
-            index_names.add(index.name)
-            
-            # Validate fields in index
-            if not index.fields:
-                errors.append(f"Index '{index.name}': fields list is empty")
-        
-        # Validate SPU config
-        if config.spu_config.enabled:
-            if not config.spu_config.spu_field:
-                errors.append("SPU aggregation enabled but no spu_field specified")
-        
-        # Validate query config
-        if not config.query_config.supported_languages:
-            errors.append("At least one supported language must be specified")
-        
         if config.query_config.default_language not in config.query_config.supported_languages:
-            errors.append(
-                f"Default language '{config.query_config.default_language}' "
-                f"not in supported languages: {config.query_config.supported_languages}"
-            )
-        
-        # Validate dynamic search fields
-        def _validate_str_list(name: str, values: List[str]) -> None:
-            if not isinstance(values, list) or not values:
-                errors.append(f"query_config.{name} must be a non-empty list[str]")
-                return
-            for i, val in enumerate(values):
-                if not isinstance(val, str) or not val.strip():
-                    errors.append(f"query_config.{name}[{i}] must be a non-empty string")
-        
-        _validate_str_list("multilingual_fields", config.query_config.multilingual_fields)
-        _validate_str_list("shared_fields", config.query_config.shared_fields)
-        _validate_str_list("core_multilingual_fields", config.query_config.core_multilingual_fields)
-        
-        core_set = set(config.query_config.core_multilingual_fields)
-        multi_set = set(config.query_config.multilingual_fields)
-        if not core_set.issubset(multi_set):
-            errors.append("query_config.core_multilingual_fields must be subset of multilingual_fields")
-        
-        # Validate text query strategy numbers
-        for name in (
-            "translation_boost",
-            "translation_boost_when_source_missing",
-            "source_boost_when_missing",
-            "original_query_fallback_boost_when_translation_missing",
-            "tie_breaker_base_query",
-        ):
-            value = getattr(config.query_config, name, None)
-            if not isinstance(value, (int, float)):
-                errors.append(f"query_config.{name} must be a number")
-            elif value < 0:
-                errors.append(f"query_config.{name} must be non-negative")
-
-        # Validate source_fields tri-state semantics
-        source_fields = config.query_config.source_fields
-        if source_fields is not None:
-            if not isinstance(source_fields, list):
-                errors.append("query_config.source_fields must be null or list[str]")
-            else:
-                for idx, field_name in enumerate(source_fields):
-                    if not isinstance(field_name, str) or not field_name.strip():
-                        errors.append(
-                            f"query_config.source_fields[{idx}] must be a non-empty string"
-                        )
-
-        # Validate tenant config shape (default must exist in config)
-        tenant_cfg = config.tenant_config
-        if not isinstance(tenant_cfg, dict):
-            errors.append("tenant_config must be an object")
-        else:
-            default_cfg = tenant_cfg.get("default")
-            if not isinstance(default_cfg, dict):
-                errors.append("tenant_config.default must be configured")
-            else:
-                index_languages = default_cfg.get("index_languages")
-                if not isinstance(index_languages, list) or len(index_languages) == 0:
-                    errors.append("tenant_config.default.index_languages must be a non-empty list")
-
+            errors.append("default_language must be included in supported_languages")
+        if config.spu_config.enabled and not config.spu_config.spu_field:
+            errors.append("spu_field is required when SPU is enabled")
         return errors
-    
+
     def to_dict(self, config: SearchConfig) -> Dict[str, Any]:
-        """Convert SearchConfig to dictionary representation."""
-        
-        # Build query_config dict
-        query_config_dict = {
-            "supported_languages": config.query_config.supported_languages,
-            "default_language": config.query_config.default_language,
-            "enable_text_embedding": config.query_config.enable_text_embedding,
-            "enable_query_rewrite": config.query_config.enable_query_rewrite,
-            "text_embedding_field": config.query_config.text_embedding_field,
-            "image_embedding_field": config.query_config.image_embedding_field,
-            "source_fields": config.query_config.source_fields,
-            "search_fields": {
-                "multilingual_fields": config.query_config.multilingual_fields,
-                "shared_fields": config.query_config.shared_fields,
-                "core_multilingual_fields": config.query_config.core_multilingual_fields,
-            },
-            "text_query_strategy": {
-                "base_minimum_should_match": config.query_config.base_minimum_should_match,
-                "translation_minimum_should_match": config.query_config.translation_minimum_should_match,
-                "translation_boost": config.query_config.translation_boost,
-                "translation_boost_when_source_missing": config.query_config.translation_boost_when_source_missing,
-                "source_boost_when_missing": config.query_config.source_boost_when_missing,
-                "original_query_fallback_boost_when_translation_missing": (
-                    config.query_config.original_query_fallback_boost_when_translation_missing
-                ),
-                "tie_breaker_base_query": config.query_config.tie_breaker_base_query,
-            }
-        }
-        
-        return {
-            "es_index_name": config.es_index_name,
-            "es_settings": config.es_settings,
-            "field_boosts": config.field_boosts,
-            "indexes": [self._index_to_dict(index) for index in config.indexes],
-            "query_config": query_config_dict,
-            "function_score": {
-                "score_mode": config.function_score.score_mode,
-                "boost_mode": config.function_score.boost_mode,
-                "functions": config.function_score.functions
-            },
-            "rerank": {
-                "enabled": config.rerank.enabled,
-                "rerank_window": config.rerank.rerank_window,
-                "timeout_sec": config.rerank.timeout_sec,
-                "weight_es": config.rerank.weight_es,
-                "weight_ai": config.rerank.weight_ai,
-                "rerank_query_template": config.rerank.rerank_query_template,
-                "rerank_doc_template": config.rerank.rerank_doc_template,
-            },
-            "spu_config": {
-                "enabled": config.spu_config.enabled,
-                "spu_field": config.spu_config.spu_field,
-                "inner_hits_size": config.spu_config.inner_hits_size,
-                "searchable_option_dimensions": config.spu_config.searchable_option_dimensions
-            },
-            "services": config.services,
-        }
-    
-    def _index_to_dict(self, index: IndexConfig) -> Dict[str, Any]:
-        """Convert IndexConfig to dictionary."""
-        result = {
-            "name": index.name,
-            "label": index.label,
-            "fields": index.fields,
-            "boost": index.boost
-        }
-        
-        if index.example:
-            result["example"] = index.example
-        
-        return result
+        return asdict(config)
@@ -0,0 +1,2 @@
+玩具	category.keyword:玩具 OR default:玩具
+消防	category.keyword:消防 OR default:消防
 """
-Centralized configuration management for saas-search.
+Compatibility accessors for infrastructure/runtime environment settings.
-Loads configuration from environment variables and .env file.
-This module provides a single point for loading .env and setting defaults.
-All configuration variables are exported directly - no need for getter functions.
+All values are derived from the unified application config. This module no
+longer owns any independent loading or precedence rules.
 """
-import os
-from pathlib import Path
-from dotenv import load_dotenv
-
-# Load .env file from project root
-PROJECT_ROOT = Path(__file__).parent.parent
-load_dotenv(PROJECT_ROOT / '.env')
-
-
-# Elasticsearch Configuration
-ES_CONFIG = {
-    'host': os.getenv('ES_HOST', 'http://localhost:9200'),
-    'username': os.getenv('ES_USERNAME'),
-    'password': os.getenv('ES_PASSWORD'),
-}
-
-# Runtime environment & index namespace
-# RUNTIME_ENV: 当前运行环境，建议使用 prod / uat / test / dev 等枚举值
-RUNTIME_ENV = os.getenv('RUNTIME_ENV', 'prod')
-# ES_INDEX_NAMESPACE: 用于按环境隔离索引的命名空间前缀，例如 "uat_" / "test_"
-# 为空字符串时表示不加前缀（通常是 prod 环境）
-ES_INDEX_NAMESPACE = os.getenv('ES_INDEX_NAMESPACE')
-if ES_INDEX_NAMESPACE is None:
-    # 未显式配置时，非 prod 环境默认加 "<env>_" 前缀，prod 环境默认不加前缀
-    ES_INDEX_NAMESPACE = '' if RUNTIME_ENV == 'prod' else f'{RUNTIME_ENV}_'
-
-# Redis Configuration
-REDIS_CONFIG = {
-    'host': os.getenv('REDIS_HOST', 'localhost'),
-    'port': int(os.getenv('REDIS_PORT', 6479)),
-    'snapshot_db': int(os.getenv('REDIS_SNAPSHOT_DB', 0)),
-    'password': os.getenv('REDIS_PASSWORD'),
-    'socket_timeout': int(os.getenv('REDIS_SOCKET_TIMEOUT', 1)),
-    '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
-    # Embedding 缓存 key 前缀，例如 "embedding"
-    'embedding_cache_prefix': os.getenv('REDIS_EMBEDDING_CACHE_PREFIX', 'embedding'),
-}
-
-# DeepL API Key
-DEEPL_AUTH_KEY = os.getenv('DEEPL_AUTH_KEY')
-
-# DashScope API Key (for Qwen models)
-DASHSCOPE_API_KEY = os.getenv('DASHSCOPE_API_KEY')
-
-# API Service Configuration
-API_HOST = os.getenv('API_HOST', '0.0.0.0')
-API_PORT = int(os.getenv('API_PORT', 6002))
-# Indexer service
-INDEXER_HOST = os.getenv('INDEXER_HOST', '0.0.0.0')
-INDEXER_PORT = int(os.getenv('INDEXER_PORT', 6004))
-# Optional dependent services
-# EMBEDDING_HOST / EMBEDDING_PORT are only used by the optional combined embedding mode.
-EMBEDDING_HOST = os.getenv('EMBEDDING_HOST', '127.0.0.1')
-EMBEDDING_PORT = int(os.getenv('EMBEDDING_PORT', 6005))
-EMBEDDING_TEXT_HOST = os.getenv('EMBEDDING_TEXT_HOST', '127.0.0.1')
-EMBEDDING_TEXT_PORT = int(os.getenv('EMBEDDING_TEXT_PORT', 6005))
-EMBEDDING_IMAGE_HOST = os.getenv('EMBEDDING_IMAGE_HOST', '127.0.0.1')
-EMBEDDING_IMAGE_PORT = int(os.getenv('EMBEDDING_IMAGE_PORT', 6008))
-TRANSLATION_HOST = os.getenv('TRANSLATION_HOST', '127.0.0.1')
-TRANSLATION_PORT = int(os.getenv('TRANSLATION_PORT', 6006))
-RERANKER_HOST = os.getenv('RERANKER_HOST', '127.0.0.1')
-RERANKER_PORT = int(os.getenv('RERANKER_PORT', 6007))
-RERANK_PROVIDER = os.getenv('RERANK_PROVIDER', 'http')
-# API_BASE_URL: 如果未设置，根据API_HOST构建（0.0.0.0使用localhost）
-API_BASE_URL = os.getenv('API_BASE_URL')
-if not API_BASE_URL:
-    API_BASE_URL = f'http://localhost:{API_PORT}' if API_HOST == '0.0.0.0' else f'http://{API_HOST}:{API_PORT}'
-INDEXER_BASE_URL = os.getenv('INDEXER_BASE_URL') or (
-    f'http://localhost:{INDEXER_PORT}' if INDEXER_HOST == '0.0.0.0' else f'http://{INDEXER_HOST}:{INDEXER_PORT}'
-)
-EMBEDDING_TEXT_SERVICE_URL = os.getenv('EMBEDDING_TEXT_SERVICE_URL') or (
-    f'http://{EMBEDDING_TEXT_HOST}:{EMBEDDING_TEXT_PORT}'
-)
-EMBEDDING_IMAGE_SERVICE_URL = os.getenv('EMBEDDING_IMAGE_SERVICE_URL') or (
-    f'http://{EMBEDDING_IMAGE_HOST}:{EMBEDDING_IMAGE_PORT}'
-)
-RERANKER_SERVICE_URL = os.getenv('RERANKER_SERVICE_URL') or f'http://{RERANKER_HOST}:{RERANKER_PORT}/rerank'
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from config.loader import get_app_config
+
+
+def _app():
+    return get_app_config()
+
+
+def _runtime():
+    return _app().runtime
-# Model IDs / paths
-TEXT_MODEL_DIR = os.getenv('TEXT_MODEL_DIR', os.getenv('TEXT_MODEL_ID', 'Qwen/Qwen3-Embedding-0.6B'))
-IMAGE_MODEL_DIR = os.getenv('IMAGE_MODEL_DIR', '/data/tw/models/cn-clip')
-# Cache Directory
-CACHE_DIR = os.getenv('CACHE_DIR', '.cache')
+def _infra():
+    return _app().infrastructure
-# MySQL Database Configuration (Shoplazza)
-DB_CONFIG = {
-    'host': os.getenv('DB_HOST'),
-    'port': int(os.getenv('DB_PORT', 3306)) if os.getenv('DB_PORT') else 3306,
-    'database': os.getenv('DB_DATABASE'),
-    'username': os.getenv('DB_USERNAME'),
-    'password': os.getenv('DB_PASSWORD'),
-}
+def _elasticsearch_dict() -> Dict[str, Any]:
+    cfg = _infra().elasticsearch
+    return {
+        "host": cfg.host,
+        "username": cfg.username,
+        "password": cfg.password,
+    }
-def print_config():
-    """Print current configuration (with sensitive data masked)."""
-    print("=" * 60)
-    print("saas-search Configuration")
-    print("=" * 60)
-    print("\nElasticsearch:")
-    print(f"  Host: {ES_CONFIG['host']}")
-    print(f"  Username: {ES_CONFIG['username']}")
-    print(f"  Password: {'*' * 10 if ES_CONFIG['password'] else 'None'}")
+def _redis_dict() -> Dict[str, Any]:
+    cfg = _infra().redis
+    return {
+        "host": cfg.host,
+        "port": cfg.port,
+        "snapshot_db": cfg.snapshot_db,
+        "password": cfg.password,
+        "socket_timeout": cfg.socket_timeout,
+        "socket_connect_timeout": cfg.socket_connect_timeout,
+        "retry_on_timeout": cfg.retry_on_timeout,
+        "cache_expire_days": cfg.cache_expire_days,
+        "embedding_cache_prefix": cfg.embedding_cache_prefix,
+        "anchor_cache_prefix": cfg.anchor_cache_prefix,
+        "anchor_cache_expire_days": cfg.anchor_cache_expire_days,
+    }
-    print("\nRedis:")
-    print(f"  Host: {REDIS_CONFIG['host']}")
-    print(f"  Port: {REDIS_CONFIG['port']}")
-    print(f"  Password: {'*' * 10 if REDIS_CONFIG['password'] else 'None'}")
-    print("\nDeepL:")
-    print(f"  API Key: {'*' * 10 if DEEPL_AUTH_KEY else 'None (translation disabled)'}")
+def _db_dict() -> Dict[str, Any]:
+    cfg = _infra().database
+    return {
+        "host": cfg.host,
+        "port": cfg.port,
+        "database": cfg.database,
+        "username": cfg.username,
+        "password": cfg.password,
+    }
+
+
+ES_CONFIG = _elasticsearch_dict()
+REDIS_CONFIG = _redis_dict()
+DB_CONFIG = _db_dict()
+
+RUNTIME_ENV = _runtime().environment
+ES_INDEX_NAMESPACE = _runtime().index_namespace
+
+DEEPL_AUTH_KEY = _infra().secrets.deepl_auth_key
+DASHSCOPE_API_KEY = _infra().secrets.dashscope_api_key
+
+API_HOST = _runtime().api_host
+API_PORT = _runtime().api_port
+INDEXER_HOST = _runtime().indexer_host
+INDEXER_PORT = _runtime().indexer_port
+EMBEDDING_HOST = _runtime().embedding_host
+EMBEDDING_PORT = _runtime().embedding_port
+EMBEDDING_TEXT_HOST = _runtime().embedding_host
+EMBEDDING_TEXT_PORT = _runtime().embedding_text_port
+EMBEDDING_IMAGE_HOST = _runtime().embedding_host
+EMBEDDING_IMAGE_PORT = _runtime().embedding_image_port
+TRANSLATION_HOST = _runtime().translator_host
+TRANSLATION_PORT = _runtime().translator_port
+RERANKER_HOST = _runtime().reranker_host
+RERANKER_PORT = _runtime().reranker_port
+
+API_BASE_URL = f"http://localhost:{API_PORT}" if API_HOST == "0.0.0.0" else f"http://{API_HOST}:{API_PORT}"
+INDEXER_BASE_URL = (
+    f"http://localhost:{INDEXER_PORT}" if INDEXER_HOST == "0.0.0.0" else f"http://{INDEXER_HOST}:{INDEXER_PORT}"
+)
+EMBEDDING_TEXT_SERVICE_URL = _app().services.embedding.get_provider_config().get("text_base_url")
+EMBEDDING_IMAGE_SERVICE_URL = _app().services.embedding.get_provider_config().get("image_base_url")
+RERANKER_SERVICE_URL = (
+    _app().services.rerank.get_provider_config().get("service_url")
+    or _app().services.rerank.get_provider_config().get("base_url")
+)
+
-    print("\nAPI Service:")
-    print(f"  Host: {API_HOST}")
-    print(f"  Port: {API_PORT}")
+def get_es_config() -> Dict[str, Any]:
+    return dict(ES_CONFIG)
-    print("\nModels:")
-    print(f"  Text Model: {TEXT_MODEL_DIR}")
-    print(f"  Image Model: {IMAGE_MODEL_DIR}")
-    print("\nCache:")
-    print(f"  Cache Directory: {CACHE_DIR}")
+def get_db_config() -> Dict[str, Any]:
+    return dict(DB_CONFIG)
-    print("\nMySQL Database:")
-    print(f"  Host: {DB_CONFIG['host']}")
-    print(f"  Port: {DB_CONFIG['port']}")
-    print(f"  Database: {DB_CONFIG['database']}")
-    print(f"  Username: {DB_CONFIG['username']}")
-    print(f"  Password: {'*' * 10 if DB_CONFIG['password'] else 'None'}")
-    print("=" * 60)
+def get_redis_config() -> Dict[str, Any]:
+    return dict(REDIS_CONFIG)
-if __name__ == "__main__":
-    print_config()
+def print_config() -> None:
+    config = _app().sanitized_dict()
+    print(config)
@@ -0,0 +1,592 @@
+"""
+Unified application configuration loader.
+
+This module is the single source of truth for loading, merging, normalizing,
+and validating application configuration.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from copy import deepcopy
+from dataclasses import asdict
+from functools import lru_cache
+from pathlib import Path
+from typing import Any, Dict, Iterable, List, Optional, Tuple
+
+import yaml
+
+try:
+    from dotenv import load_dotenv as _load_dotenv  # type: ignore
+except Exception:  # pragma: no cover
+    _load_dotenv = None
+
+from config.schema import (
+    AppConfig,
+    AssetsConfig,
+    ConfigMetadata,
+    DatabaseSettings,
+    ElasticsearchSettings,
+    EmbeddingServiceConfig,
+    FunctionScoreConfig,
+    IndexConfig,
+    InfrastructureConfig,
+    QueryConfig,
+    RedisSettings,
+    RerankConfig,
+    RerankServiceConfig,
+    RuntimeConfig,
+    SearchConfig,
+    SecretsConfig,
+    ServicesConfig,
+    SPUConfig,
+    TenantCatalogConfig,
+    TranslationServiceConfig,
+)
+from translation.settings import build_translation_config
+
+
+class ConfigurationError(Exception):
+    """Raised when configuration validation fails."""
+
+
+def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
+    result = deepcopy(base)
+    for key, value in (override or {}).items():
+        if (
+            key in result
+            and isinstance(result[key], dict)
+            and isinstance(value, dict)
+        ):
+            result[key] = _deep_merge(result[key], value)
+        else:
+            result[key] = deepcopy(value)
+    return result
+
+
+def _load_yaml(path: Path) -> Dict[str, Any]:
+    with open(path, "r", encoding="utf-8") as handle:
+        data = yaml.safe_load(handle) or {}
+    if not isinstance(data, dict):
+        raise ConfigurationError(f"Configuration file root must be a mapping: {path}")
+    return data
+
+
+def _read_rewrite_dictionary(path: Path) -> Dict[str, str]:
+    rewrite_dict: Dict[str, str] = {}
+    if not path.exists():
+        return rewrite_dict
+
+    with open(path, "r", encoding="utf-8") as handle:
+        for raw_line in handle:
+            line = raw_line.strip()
+            if not line or line.startswith("#"):
+                continue
+            parts = line.split("\t")
+            if len(parts) < 2:
+                continue
+            original = parts[0].strip()
+            replacement = parts[1].strip()
+            if original and replacement:
+                rewrite_dict[original] = replacement
+    return rewrite_dict
+
+
+class AppConfigLoader:
+    """Load the unified application configuration."""
+
+    def __init__(
+        self,
+        *,
+        config_dir: Optional[Path] = None,
+        config_file: Optional[Path] = None,
+        env_file: Optional[Path] = None,
+    ) -> None:
+        self.config_dir = Path(config_dir or Path(__file__).parent)
+        self.config_file = Path(config_file) if config_file is not None else None
+        self.project_root = self.config_dir.parent
+        self.env_file = Path(env_file) if env_file is not None else self.project_root / ".env"
+
+    def load(self, validate: bool = True) -> AppConfig:
+        self._load_env()
+        raw_config, loaded_files = self._load_raw_config()
+        app_config = self._build_app_config(raw_config, loaded_files)
+        if validate:
+            self._validate(app_config)
+        return app_config
+
+    def _load_env(self) -> None:
+        if _load_dotenv is not None:
+            _load_dotenv(self.env_file, override=False)
+            return
+        _load_env_file_fallback(self.env_file)
+
+    def _load_raw_config(self) -> Tuple[Dict[str, Any], List[str]]:
+        env_name = (os.getenv("APP_ENV") or os.getenv("RUNTIME_ENV") or "prod").strip().lower() or "prod"
+        loaded_files: List[str] = []
+        raw: Dict[str, Any] = {}
+
+        if self.config_file is not None:
+            config_path = self.config_file
+            if not config_path.exists():
+                raise ConfigurationError(f"Configuration file not found: {config_path}")
+            raw = _deep_merge(raw, _load_yaml(config_path))
+            loaded_files.append(str(config_path))
+        else:
+            base_path = self.config_dir / "base.yaml"
+            legacy_path = self.config_dir / "config.yaml"
+            primary_path = base_path if base_path.exists() else legacy_path
+            if not primary_path.exists():
+                raise ConfigurationError(f"Configuration file not found: {primary_path}")
+            raw = _deep_merge(raw, _load_yaml(primary_path))
+            loaded_files.append(str(primary_path))
+
+            env_path = self.config_dir / "environments" / f"{env_name}.yaml"
+            if env_path.exists():
+                raw = _deep_merge(raw, _load_yaml(env_path))
+                loaded_files.append(str(env_path))
+
+        tenant_dir = self.config_dir / "tenants"
+        if tenant_dir.is_dir():
+            tenant_files = sorted(tenant_dir.glob("*.yaml"))
+            if tenant_files:
+                tenant_config = {"default": {}, "tenants": {}}
+                default_path = tenant_dir / "_default.yaml"
+                if default_path.exists():
+                    tenant_config["default"] = _load_yaml(default_path)
+                    loaded_files.append(str(default_path))
+                for tenant_path in tenant_files:
+                    if tenant_path.name == "_default.yaml":
+                        continue
+                    tenant_config["tenants"][tenant_path.stem] = _load_yaml(tenant_path)
+                    loaded_files.append(str(tenant_path))
+                raw["tenant_config"] = tenant_config
+
+        return raw, loaded_files
+
+    def _build_app_config(self, raw: Dict[str, Any], loaded_files: List[str]) -> AppConfig:
+        assets_cfg = raw.get("assets") if isinstance(raw.get("assets"), dict) else {}
+        rewrite_path = (
+            assets_cfg.get("query_rewrite_dictionary_path")
+            or assets_cfg.get("rewrite_dictionary_path")
+            or self.config_dir / "dictionaries" / "query_rewrite.dict"
+        )
+        rewrite_path = Path(rewrite_path)
+        if not rewrite_path.is_absolute():
+            rewrite_path = (self.project_root / rewrite_path).resolve()
+        if not rewrite_path.exists():
+            legacy_rewrite_path = (self.config_dir / "query_rewrite.dict").resolve()
+            if legacy_rewrite_path.exists():
+                rewrite_path = legacy_rewrite_path
+
+        rewrite_dictionary = _read_rewrite_dictionary(rewrite_path)
+        search_config = self._build_search_config(raw, rewrite_dictionary)
+        services_config = self._build_services_config(raw.get("services") or {})
+        tenants_config = self._build_tenants_config(raw.get("tenant_config") or {})
+        runtime_config = self._build_runtime_config()
+        infrastructure_config = self._build_infrastructure_config(runtime_config.environment)
+
+        metadata = ConfigMetadata(
+            loaded_files=tuple(loaded_files),
+            config_hash="",
+            deprecated_keys=tuple(self._detect_deprecated_keys(raw)),
+        )
+
+        app_config = AppConfig(
+            runtime=runtime_config,
+            infrastructure=infrastructure_config,
+            search=search_config,
+            services=services_config,
+            tenants=tenants_config,
+            assets=AssetsConfig(query_rewrite_dictionary_path=rewrite_path),
+            metadata=metadata,
+        )
+
+        config_hash = self._compute_hash(app_config)
+        return AppConfig(
+            runtime=app_config.runtime,
+            infrastructure=app_config.infrastructure,
+            search=app_config.search,
+            services=app_config.services,
+            tenants=app_config.tenants,
+            assets=app_config.assets,
+            metadata=ConfigMetadata(
+                loaded_files=app_config.metadata.loaded_files,
+                config_hash=config_hash,
+                deprecated_keys=app_config.metadata.deprecated_keys,
+            ),
+        )
+
+    def _build_search_config(self, raw: Dict[str, Any], rewrite_dictionary: Dict[str, str]) -> SearchConfig:
+        field_boosts = raw.get("field_boosts") or {}
+        if not isinstance(field_boosts, dict):
+            raise ConfigurationError("field_boosts must be a mapping")
+
+        indexes: List[IndexConfig] = []
+        for item in raw.get("indexes") or []:
+            if not isinstance(item, dict):
+                raise ConfigurationError("indexes items must be mappings")
+            indexes.append(
+                IndexConfig(
+                    name=str(item["name"]),
+                    label=str(item.get("label") or item["name"]),
+                    fields=list(item.get("fields") or []),
+                    boost=float(item.get("boost", 1.0)),
+                    example=item.get("example"),
+                )
+            )
+
+        query_cfg = raw.get("query_config") if isinstance(raw.get("query_config"), dict) else {}
+        search_fields = query_cfg.get("search_fields") if isinstance(query_cfg.get("search_fields"), dict) else {}
+        text_strategy = (
+            query_cfg.get("text_query_strategy")
+            if isinstance(query_cfg.get("text_query_strategy"), dict)
+            else {}
+        )
+        query_config = QueryConfig(
+            supported_languages=list(query_cfg.get("supported_languages") or ["zh", "en"]),
+            default_language=str(query_cfg.get("default_language") or "en"),
+            enable_text_embedding=bool(query_cfg.get("enable_text_embedding", True)),
+            enable_query_rewrite=bool(query_cfg.get("enable_query_rewrite", True)),
+            rewrite_dictionary=rewrite_dictionary,
+            text_embedding_field=query_cfg.get("text_embedding_field"),
+            image_embedding_field=query_cfg.get("image_embedding_field"),
+            source_fields=query_cfg.get("source_fields"),
+            knn_boost=float(query_cfg.get("knn_boost", 0.25)),
+            multilingual_fields=list(
+                search_fields.get(
+                    "multilingual_fields",
+                    ["title", "brief", "description", "vendor", "category_path", "category_name_text"],
+                )
+            ),
+            shared_fields=list(
+                search_fields.get(
+                    "shared_fields",
+                    ["tags", "option1_values", "option2_values", "option3_values"],
+                )
+            ),
+            core_multilingual_fields=list(
+                search_fields.get(
+                    "core_multilingual_fields",
+                    ["title", "brief", "vendor", "category_name_text"],
+                )
+            ),
+            base_minimum_should_match=str(text_strategy.get("base_minimum_should_match", "75%")),
+            translation_minimum_should_match=str(text_strategy.get("translation_minimum_should_match", "75%")),
+            translation_boost=float(text_strategy.get("translation_boost", 0.4)),
+            translation_boost_when_source_missing=float(
+                text_strategy.get("translation_boost_when_source_missing", 1.0)
+            ),
+            source_boost_when_missing=float(text_strategy.get("source_boost_when_missing", 0.6)),
+            original_query_fallback_boost_when_translation_missing=float(
+                text_strategy.get("original_query_fallback_boost_when_translation_missing", 0.2)
+            ),
+            tie_breaker_base_query=float(text_strategy.get("tie_breaker_base_query", 0.9)),
+            zh_to_en_model=str(query_cfg.get("zh_to_en_model") or "opus-mt-zh-en"),
+            en_to_zh_model=str(query_cfg.get("en_to_zh_model") or "opus-mt-en-zh"),
+            default_translation_model=str(
+                query_cfg.get("default_translation_model") or "nllb-200-distilled-600m"
+            ),
+        )
+
+        function_score_cfg = raw.get("function_score") if isinstance(raw.get("function_score"), dict) else {}
+        rerank_cfg = raw.get("rerank") if isinstance(raw.get("rerank"), dict) else {}
+        spu_cfg = raw.get("spu_config") if isinstance(raw.get("spu_config"), dict) else {}
+
+        return SearchConfig(
+            field_boosts={str(key): float(value) for key, value in field_boosts.items()},
+            indexes=indexes,
+            query_config=query_config,
+            function_score=FunctionScoreConfig(
+                score_mode=str(function_score_cfg.get("score_mode") or "sum"),
+                boost_mode=str(function_score_cfg.get("boost_mode") or "multiply"),
+                functions=list(function_score_cfg.get("functions") or []),
+            ),
+            rerank=RerankConfig(
+                enabled=bool(rerank_cfg.get("enabled", True)),
+                rerank_window=int(rerank_cfg.get("rerank_window", 384)),
+                timeout_sec=float(rerank_cfg.get("timeout_sec", 15.0)),
+                weight_es=float(rerank_cfg.get("weight_es", 0.4)),
+                weight_ai=float(rerank_cfg.get("weight_ai", 0.6)),
+                rerank_query_template=str(rerank_cfg.get("rerank_query_template") or "{query}"),
+                rerank_doc_template=str(rerank_cfg.get("rerank_doc_template") or "{title}"),
+            ),
+            spu_config=SPUConfig(
+                enabled=bool(spu_cfg.get("enabled", False)),
+                spu_field=spu_cfg.get("spu_field"),
+                inner_hits_size=int(spu_cfg.get("inner_hits_size", 3)),
+                searchable_option_dimensions=list(
+                    spu_cfg.get("searchable_option_dimensions") or ["option1", "option2", "option3"]
+                ),
+            ),
+            es_index_name=str(raw.get("es_index_name") or "search_products"),
+            es_settings=dict(raw.get("es_settings") or {}),
+        )
+
+    def _build_services_config(self, raw: Dict[str, Any]) -> ServicesConfig:
+        if not isinstance(raw, dict):
+            raise ConfigurationError("services must be a mapping")
+
+        translation_raw = raw.get("translation") if isinstance(raw.get("translation"), dict) else {}
+        normalized_translation = build_translation_config(translation_raw)
+        translation_config = TranslationServiceConfig(
+            endpoint=str(normalized_translation["service_url"]).rstrip("/"),
+            timeout_sec=float(normalized_translation["timeout_sec"]),
+            default_model=str(normalized_translation["default_model"]),
+            default_scene=str(normalized_translation["default_scene"]),
+            cache=dict(normalized_translation["cache"]),
+            capabilities={str(key): dict(value) for key, value in normalized_translation["capabilities"].items()},
+        )
+
+        embedding_raw = raw.get("embedding") if isinstance(raw.get("embedding"), dict) else {}
+        embedding_provider = str(embedding_raw.get("provider") or "http").strip().lower()
+        embedding_providers = dict(embedding_raw.get("providers") or {})
+        if embedding_provider not in embedding_providers:
+            raise ConfigurationError(f"services.embedding.providers.{embedding_provider} must be configured")
+        embedding_backend = str(embedding_raw.get("backend") or "").strip().lower()
+        embedding_backends = {
+            str(key).strip().lower(): dict(value)
+            for key, value in dict(embedding_raw.get("backends") or {}).items()
+        }
+        if embedding_backend not in embedding_backends:
+            raise ConfigurationError(f"services.embedding.backends.{embedding_backend} must be configured")
+        image_backend = str(embedding_raw.get("image_backend") or "clip_as_service").strip().lower()
+        image_backends = {
+            str(key).strip().lower(): dict(value)
+            for key, value in dict(embedding_raw.get("image_backends") or {}).items()
+        }
+        if not image_backends:
+            image_backends = {
+                "clip_as_service": {
+                    "server": "grpc://127.0.0.1:51000",
+                    "model_name": "CN-CLIP/ViT-L-14",
+                    "batch_size": 8,
+                    "normalize_embeddings": True,
+                },
+                "local_cnclip": {
+                    "model_name": "ViT-L-14",
+                    "device": None,
+                    "batch_size": 8,
+                    "normalize_embeddings": True,
+                },
+            }
+        if image_backend not in image_backends:
+            raise ConfigurationError(f"services.embedding.image_backends.{image_backend} must be configured")
+
+        embedding_config = EmbeddingServiceConfig(
+            provider=embedding_provider,
+            providers=embedding_providers,
+            backend=embedding_backend,
+            backends=embedding_backends,
+            image_backend=image_backend,
+            image_backends=image_backends,
+        )
+
+        rerank_raw = raw.get("rerank") if isinstance(raw.get("rerank"), dict) else {}
+        rerank_provider = str(rerank_raw.get("provider") or "http").strip().lower()
+        rerank_providers = dict(rerank_raw.get("providers") or {})
+        if rerank_provider not in rerank_providers:
+            raise ConfigurationError(f"services.rerank.providers.{rerank_provider} must be configured")
+        rerank_backend = str(rerank_raw.get("backend") or "").strip().lower()
+        rerank_backends = {
+            str(key).strip().lower(): dict(value)
+            for key, value in dict(rerank_raw.get("backends") or {}).items()
+        }
+        if rerank_backend not in rerank_backends:
+            raise ConfigurationError(f"services.rerank.backends.{rerank_backend} must be configured")
+        rerank_request = dict(rerank_raw.get("request") or {})
+        rerank_request.setdefault("max_docs", 1000)
+        rerank_request.setdefault("normalize", True)
+
+        rerank_config = RerankServiceConfig(
+            provider=rerank_provider,
+            providers=rerank_providers,
+            backend=rerank_backend,
+            backends=rerank_backends,
+            request=rerank_request,
+        )
+
+        return ServicesConfig(
+            translation=translation_config,
+            embedding=embedding_config,
+            rerank=rerank_config,
+        )
+
+    def _build_tenants_config(self, raw: Dict[str, Any]) -> TenantCatalogConfig:
+        if not isinstance(raw, dict):
+            raise ConfigurationError("tenant_config must be a mapping")
+        default_cfg = raw.get("default") if isinstance(raw.get("default"), dict) else {}
+        tenants_cfg = raw.get("tenants") if isinstance(raw.get("tenants"), dict) else {}
+        return TenantCatalogConfig(
+            default=dict(default_cfg),
+            tenants={str(key): dict(value) for key, value in tenants_cfg.items()},
+        )
+
+    def _build_runtime_config(self) -> RuntimeConfig:
+        environment = (os.getenv("APP_ENV") or os.getenv("RUNTIME_ENV") or "prod").strip().lower() or "prod"
+        namespace = os.getenv("ES_INDEX_NAMESPACE")
+        if namespace is None:
+            namespace = "" if environment == "prod" else f"{environment}_"
+
+        return RuntimeConfig(
+            environment=environment,
+            index_namespace=namespace,
+            api_host=os.getenv("API_HOST", "0.0.0.0"),
+            api_port=int(os.getenv("API_PORT", 6002)),
+            indexer_host=os.getenv("INDEXER_HOST", "0.0.0.0"),
+            indexer_port=int(os.getenv("INDEXER_PORT", 6004)),
+            embedding_host=os.getenv("EMBEDDING_HOST", "127.0.0.1"),
+            embedding_port=int(os.getenv("EMBEDDING_PORT", 6005)),
+            embedding_text_port=int(os.getenv("EMBEDDING_TEXT_PORT", 6005)),
+            embedding_image_port=int(os.getenv("EMBEDDING_IMAGE_PORT", 6008)),
+            translator_host=os.getenv("TRANSLATION_HOST", "127.0.0.1"),
+            translator_port=int(os.getenv("TRANSLATION_PORT", 6006)),
+            reranker_host=os.getenv("RERANKER_HOST", "127.0.0.1"),
+            reranker_port=int(os.getenv("RERANKER_PORT", 6007)),
+        )
+
+    def _build_infrastructure_config(self, environment: str) -> InfrastructureConfig:
+        del environment
+        return InfrastructureConfig(
+            elasticsearch=ElasticsearchSettings(
+                host=os.getenv("ES_HOST", "http://localhost:9200"),
+                username=os.getenv("ES_USERNAME"),
+                password=os.getenv("ES_PASSWORD"),
+            ),
+            redis=RedisSettings(
+                host=os.getenv("REDIS_HOST", "localhost"),
+                port=int(os.getenv("REDIS_PORT", 6479)),
+                snapshot_db=int(os.getenv("REDIS_SNAPSHOT_DB", 0)),
+                password=os.getenv("REDIS_PASSWORD"),
+                socket_timeout=int(os.getenv("REDIS_SOCKET_TIMEOUT", 1)),
+                socket_connect_timeout=int(os.getenv("REDIS_SOCKET_CONNECT_TIMEOUT", 1)),
+                retry_on_timeout=os.getenv("REDIS_RETRY_ON_TIMEOUT", "false").strip().lower() == "true",
+                cache_expire_days=int(os.getenv("REDIS_CACHE_EXPIRE_DAYS", 360 * 2)),
+                embedding_cache_prefix=os.getenv("REDIS_EMBEDDING_CACHE_PREFIX", "embedding"),
+                anchor_cache_prefix=os.getenv("REDIS_ANCHOR_CACHE_PREFIX", "product_anchors"),
+                anchor_cache_expire_days=int(os.getenv("REDIS_ANCHOR_CACHE_EXPIRE_DAYS", 30)),
+            ),
+            database=DatabaseSettings(
+                host=os.getenv("DB_HOST"),
+                port=int(os.getenv("DB_PORT", 3306)) if os.getenv("DB_PORT") else 3306,
+                database=os.getenv("DB_DATABASE"),
+                username=os.getenv("DB_USERNAME"),
+                password=os.getenv("DB_PASSWORD"),
+            ),
+            secrets=SecretsConfig(
+                dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
+                deepl_auth_key=os.getenv("DEEPL_AUTH_KEY"),
+            ),
+        )
+
+    def _validate(self, app_config: AppConfig) -> None:
+        errors: List[str] = []
+
+        if not app_config.search.es_index_name:
+            errors.append("search.es_index_name is required")
+
+        if not app_config.search.field_boosts:
+            errors.append("search.field_boosts cannot be empty")
+        else:
+            for field_name, boost in app_config.search.field_boosts.items():
+                if boost < 0:
+                    errors.append(f"field_boosts.{field_name} must be non-negative")
+
+        query_config = app_config.search.query_config
+        if not query_config.supported_languages:
+            errors.append("query_config.supported_languages must not be empty")
+        if query_config.default_language not in query_config.supported_languages:
+            errors.append("query_config.default_language must be included in supported_languages")
+        for name, values in (
+            ("multilingual_fields", query_config.multilingual_fields),
+            ("shared_fields", query_config.shared_fields),
+            ("core_multilingual_fields", query_config.core_multilingual_fields),
+        ):
+            if not values:
+                errors.append(f"query_config.{name} must not be empty")
+
+        if not set(query_config.core_multilingual_fields).issubset(set(query_config.multilingual_fields)):
+            errors.append("query_config.core_multilingual_fields must be a subset of multilingual_fields")
+
+        if app_config.search.spu_config.enabled and not app_config.search.spu_config.spu_field:
+            errors.append("spu_config.spu_field is required when spu_config.enabled is true")
+
+        if not app_config.tenants.default or not app_config.tenants.default.get("index_languages"):
+            errors.append("tenant_config.default.index_languages must be configured")
+
+        if app_config.metadata.deprecated_keys:
+            errors.append(
+                "Deprecated tenant config keys are not supported: "
+                + ", ".join(app_config.metadata.deprecated_keys)
+            )
+
+        embedding_provider_cfg = app_config.services.embedding.get_provider_config()
+        if not embedding_provider_cfg.get("text_base_url"):
+            errors.append("services.embedding.providers.<provider>.text_base_url is required")
+        if not embedding_provider_cfg.get("image_base_url"):
+            errors.append("services.embedding.providers.<provider>.image_base_url is required")
+
+        rerank_provider_cfg = app_config.services.rerank.get_provider_config()
+        if not rerank_provider_cfg.get("service_url") and not rerank_provider_cfg.get("base_url"):
+            errors.append("services.rerank.providers.<provider>.service_url or base_url is required")
+
+        if errors:
+            raise ConfigurationError("Configuration validation failed:\n" + "\n".join(f"  - {err}" for err in errors))
+
+    def _compute_hash(self, app_config: AppConfig) -> str:
+        payload = asdict(app_config)
+        payload["metadata"]["config_hash"] = ""
+        payload["infrastructure"]["elasticsearch"]["password"] = "***" if payload["infrastructure"]["elasticsearch"].get("password") else None
+        payload["infrastructure"]["database"]["password"] = "***" if payload["infrastructure"]["database"].get("password") else None
+        payload["infrastructure"]["redis"]["password"] = "***" if payload["infrastructure"]["redis"].get("password") else None
+        payload["infrastructure"]["secrets"]["dashscope_api_key"] = "***" if payload["infrastructure"]["secrets"].get("dashscope_api_key") else None
+        payload["infrastructure"]["secrets"]["deepl_auth_key"] = "***" if payload["infrastructure"]["secrets"].get("deepl_auth_key") else None
+        blob = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str)
+        return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]
+
+    def _detect_deprecated_keys(self, raw: Dict[str, Any]) -> Iterable[str]:
+        tenant_raw = raw.get("tenant_config") if isinstance(raw.get("tenant_config"), dict) else {}
+        for key in ("default",):
+            item = tenant_raw.get(key)
+            if isinstance(item, dict):
+                for deprecated in ("translate_to_en", "translate_to_zh"):
+                    if deprecated in item:
+                        yield f"tenant_config.{key}.{deprecated}"
+        tenants = tenant_raw.get("tenants") if isinstance(tenant_raw.get("tenants"), dict) else {}
+        for tenant_id, cfg in tenants.items():
+            if not isinstance(cfg, dict):
+                continue
+            for deprecated in ("translate_to_en", "translate_to_zh"):
+                if deprecated in cfg:
+                    yield f"tenant_config.tenants.{tenant_id}.{deprecated}"
+
+
+@lru_cache(maxsize=1)
+def get_app_config() -> AppConfig:
+    """Return the process-global application configuration."""
+
+    return AppConfigLoader().load()
+
+
+def reload_app_config() -> AppConfig:
+    """Clear the cached configuration and reload it."""
+
+    get_app_config.cache_clear()
+    return get_app_config()
+
+
+def _load_env_file_fallback(path: Path) -> None:
+    if not path.exists():
+        return
+    with open(path, "r", encoding="utf-8") as handle:
+        for raw_line in handle:
+            line = raw_line.strip()
+            if not line or line.startswith("#") or "=" not in line:
+                continue
+            key, value = line.split("=", 1)
+            key = key.strip()
+            value = value.strip().strip('"').strip("'")
+            if key and key not in os.environ:
+                os.environ[key] = value
@@ -1,3 +0,0 @@
-玩具	category.keyword:玩具 OR default:玩具
-消防	category.keyword:消防 OR default:消防
-
@@ -0,0 +1,307 @@
+"""
+Typed configuration schema for the unified application configuration.
+
+This module defines the normalized in-memory structure used by all services.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+
+@dataclass(frozen=True)
+class IndexConfig:
+    """Deprecated compatibility shape for legacy diagnostics/tests."""
+
+    name: str
+    label: str
+    fields: List[str]
+    boost: float = 1.0
+    example: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class QueryConfig:
+    """Configuration for query processing."""
+
+    supported_languages: List[str] = field(default_factory=lambda: ["zh", "en"])
+    default_language: str = "en"
+    enable_text_embedding: bool = True
+    enable_query_rewrite: bool = True
+    rewrite_dictionary: Dict[str, str] = field(default_factory=dict)
+    text_embedding_field: Optional[str] = "title_embedding"
+    image_embedding_field: Optional[str] = None
+    source_fields: Optional[List[str]] = None
+    knn_boost: float = 0.25
+    multilingual_fields: List[str] = field(
+        default_factory=lambda: [
+            "title",
+            "brief",
+            "description",
+            "vendor",
+            "category_path",
+            "category_name_text",
+        ]
+    )
+    shared_fields: List[str] = field(
+        default_factory=lambda: ["tags", "option1_values", "option2_values", "option3_values"]
+    )
+    core_multilingual_fields: List[str] = field(
+        default_factory=lambda: ["title", "brief", "vendor", "category_name_text"]
+    )
+    base_minimum_should_match: str = "75%"
+    translation_minimum_should_match: str = "75%"
+    translation_boost: float = 0.4
+    translation_boost_when_source_missing: float = 1.0
+    source_boost_when_missing: float = 0.6
+    original_query_fallback_boost_when_translation_missing: float = 0.2
+    tie_breaker_base_query: float = 0.9
+    zh_to_en_model: str = "opus-mt-zh-en"
+    en_to_zh_model: str = "opus-mt-en-zh"
+    default_translation_model: str = "nllb-200-distilled-600m"
+
+
+@dataclass(frozen=True)
+class SPUConfig:
+    """SPU aggregation/search configuration."""
+
+    enabled: bool = False
+    spu_field: Optional[str] = None
+    inner_hits_size: int = 3
+    searchable_option_dimensions: List[str] = field(
+        default_factory=lambda: ["option1", "option2", "option3"]
+    )
+
+
+@dataclass(frozen=True)
+class FunctionScoreConfig:
+    """Function score configuration."""
+
+    score_mode: str = "sum"
+    boost_mode: str = "multiply"
+    functions: List[Dict[str, Any]] = field(default_factory=list)
+
+
+@dataclass(frozen=True)
+class RerankConfig:
+    """Search-time rerank configuration."""
+
+    enabled: bool = True
+    rerank_window: int = 384
+    timeout_sec: float = 15.0
+    weight_es: float = 0.4
+    weight_ai: float = 0.6
+    rerank_query_template: str = "{query}"
+    rerank_doc_template: str = "{title}"
+
+
+@dataclass(frozen=True)
+class SearchConfig:
+    """Search behavior configuration shared by backend and indexer."""
+
+    field_boosts: Dict[str, float]
+    indexes: List[IndexConfig] = field(default_factory=list)
+    query_config: QueryConfig = field(default_factory=QueryConfig)
+    function_score: FunctionScoreConfig = field(default_factory=FunctionScoreConfig)
+    rerank: RerankConfig = field(default_factory=RerankConfig)
+    spu_config: SPUConfig = field(default_factory=SPUConfig)
+    es_index_name: str = "search_products"
+    es_settings: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class TranslationServiceConfig:
+    """Translator service configuration."""
+
+    endpoint: str
+    timeout_sec: float
+    default_model: str
+    default_scene: str
+    cache: Dict[str, Any]
+    capabilities: Dict[str, Dict[str, Any]]
+
+    def as_dict(self) -> Dict[str, Any]:
+        return {
+            "service_url": self.endpoint,
+            "timeout_sec": self.timeout_sec,
+            "default_model": self.default_model,
+            "default_scene": self.default_scene,
+            "cache": self.cache,
+            "capabilities": self.capabilities,
+        }
+
+
+@dataclass(frozen=True)
+class EmbeddingServiceConfig:
+    """Embedding service configuration."""
+
+    provider: str
+    providers: Dict[str, Any]
+    backend: str
+    backends: Dict[str, Dict[str, Any]]
+    image_backend: str
+    image_backends: Dict[str, Dict[str, Any]]
+
+    def get_provider_config(self) -> Dict[str, Any]:
+        return dict(self.providers.get(self.provider, {}) or {})
+
+    def get_backend_config(self) -> Dict[str, Any]:
+        return dict(self.backends.get(self.backend, {}) or {})
+
+    def get_image_backend_config(self) -> Dict[str, Any]:
+        return dict(self.image_backends.get(self.image_backend, {}) or {})
+
+
+@dataclass(frozen=True)
+class RerankServiceConfig:
+    """Reranker service configuration."""
+
+    provider: str
+    providers: Dict[str, Any]
+    backend: str
+    backends: Dict[str, Dict[str, Any]]
+    request: Dict[str, Any]
+
+    def get_provider_config(self) -> Dict[str, Any]:
+        return dict(self.providers.get(self.provider, {}) or {})
+
+    def get_backend_config(self) -> Dict[str, Any]:
+        return dict(self.backends.get(self.backend, {}) or {})
+
+
+@dataclass(frozen=True)
+class ServicesConfig:
+    """All service-level configuration."""
+
+    translation: TranslationServiceConfig
+    embedding: EmbeddingServiceConfig
+    rerank: RerankServiceConfig
+
+
+@dataclass(frozen=True)
+class TenantCatalogConfig:
+    """Tenant catalog configuration."""
+
+    default: Dict[str, Any]
+    tenants: Dict[str, Dict[str, Any]]
+
+    def get_raw(self) -> Dict[str, Any]:
+        return {
+            "default": dict(self.default),
+            "tenants": {str(key): dict(value) for key, value in self.tenants.items()},
+        }
+
+
+@dataclass(frozen=True)
+class ElasticsearchSettings:
+    host: str = "http://localhost:9200"
+    username: Optional[str] = None
+    password: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class RedisSettings:
+    host: str = "localhost"
+    port: int = 6479
+    snapshot_db: int = 0
+    password: Optional[str] = None
+    socket_timeout: int = 1
+    socket_connect_timeout: int = 1
+    retry_on_timeout: bool = False
+    cache_expire_days: int = 720
+    embedding_cache_prefix: str = "embedding"
+    anchor_cache_prefix: str = "product_anchors"
+    anchor_cache_expire_days: int = 30
+
+
+@dataclass(frozen=True)
+class DatabaseSettings:
+    host: Optional[str] = None
+    port: int = 3306
+    database: Optional[str] = None
+    username: Optional[str] = None
+    password: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class SecretsConfig:
+    dashscope_api_key: Optional[str] = None
+    deepl_auth_key: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class InfrastructureConfig:
+    elasticsearch: ElasticsearchSettings
+    redis: RedisSettings
+    database: DatabaseSettings
+    secrets: SecretsConfig
+
+
+@dataclass(frozen=True)
+class RuntimeConfig:
+    environment: str = "prod"
+    index_namespace: str = ""
+    api_host: str = "0.0.0.0"
+    api_port: int = 6002
+    indexer_host: str = "0.0.0.0"
+    indexer_port: int = 6004
+    embedding_host: str = "127.0.0.1"
+    embedding_port: int = 6005
+    embedding_text_port: int = 6005
+    embedding_image_port: int = 6008
+    translator_host: str = "127.0.0.1"
+    translator_port: int = 6006
+    reranker_host: str = "127.0.0.1"
+    reranker_port: int = 6007
+
+
+@dataclass(frozen=True)
+class AssetsConfig:
+    query_rewrite_dictionary_path: Path
+
+
+@dataclass(frozen=True)
+class ConfigMetadata:
+    loaded_files: Tuple[str, ...]
+    config_hash: str
+    deprecated_keys: Tuple[str, ...] = field(default_factory=tuple)
+
+
+@dataclass(frozen=True)
+class AppConfig:
+    """Root application configuration."""
+
+    runtime: RuntimeConfig
+    infrastructure: InfrastructureConfig
+    search: SearchConfig
+    services: ServicesConfig
+    tenants: TenantCatalogConfig
+    assets: AssetsConfig
+    metadata: ConfigMetadata
+
+    def sanitized_dict(self) -> Dict[str, Any]:
+        data = asdict(self)
+        data["infrastructure"]["elasticsearch"]["password"] = _mask_secret(
+            data["infrastructure"]["elasticsearch"].get("password")
+        )
+        data["infrastructure"]["database"]["password"] = _mask_secret(
+            data["infrastructure"]["database"].get("password")
+        )
+        data["infrastructure"]["redis"]["password"] = _mask_secret(
+            data["infrastructure"]["redis"].get("password")
+        )
+        data["infrastructure"]["secrets"]["dashscope_api_key"] = _mask_secret(
+            data["infrastructure"]["secrets"].get("dashscope_api_key")
+        )
+        data["infrastructure"]["secrets"]["deepl_auth_key"] = _mask_secret(
+            data["infrastructure"]["secrets"].get("deepl_auth_key")
+        )
+        return data
+
+
+def _mask_secret(value: Optional[str]) -> Optional[str]:
+    if not value:
+        return value
+    return "***"
 """
-Services configuration - single source for translation, embedding, rerank.
+Unified service configuration accessors.
-Translation is modeled as:
-- one translator service endpoint used by business callers
-- multiple translation capabilities loaded inside the translator service
+This module is now a thin adapter over ``config.loader.get_app_config`` and
+contains no independent parsing or precedence logic.
 """
 from __future__ import annotations
-import os
-from dataclasses import dataclass, field
-from functools import lru_cache
-from pathlib import Path
-from typing import Any, Dict, List, Optional
-
-import yaml
-from translation.settings import TranslationConfig, build_translation_config, get_translation_cache
-
-
-@dataclass
-class ServiceConfig:
-    """Config for one capability (embedding/rerank)."""
-
-    provider: str
-    providers: Dict[str, Any] = field(default_factory=dict)
-
-    def get_provider_cfg(self) -> Dict[str, Any]:
-        p = (self.provider or "").strip().lower()
-        return self.providers.get(p, {}) if isinstance(self.providers, dict) else {}
-
-
-def _load_services_raw(config_path: Optional[Path] = None) -> Dict[str, Any]:
-    if config_path is None:
-        config_path = Path(__file__).parent / "config.yaml"
-    path = Path(config_path)
-    if not path.exists():
-        raise FileNotFoundError(f"services config file not found: {path}")
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            data = yaml.safe_load(f)
-    except Exception as exc:
-        raise RuntimeError(f"failed to parse services config from {path}: {exc}") from exc
-    if not isinstance(data, dict):
-        raise RuntimeError(f"invalid config format in {path}: expected mapping root")
-    services = data.get("services")
-    if not isinstance(services, dict):
-        raise RuntimeError("config.yaml must contain a valid 'services' mapping")
-    return services
-
-
-def _resolve_provider_name(env_name: str, config_provider: Any, capability: str) -> str:
-    provider = os.getenv(env_name) or config_provider
-    if not provider:
-        raise ValueError(
-            f"services.{capability}.provider is required "
-            f"(or set env override {env_name})"
-        )
-    return str(provider).strip().lower()
-
-
-def _resolve_translation() -> TranslationConfig:
-    raw = _load_services_raw()
-    cfg = raw.get("translation", {}) if isinstance(raw.get("translation"), dict) else {}
-    return build_translation_config(cfg)
-
-
-def _resolve_embedding() -> ServiceConfig:
-    raw = _load_services_raw()
-    cfg = raw.get("embedding", {}) if isinstance(raw.get("embedding"), dict) else {}
-    providers = cfg.get("providers", {}) if isinstance(cfg.get("providers"), dict) else {}
-
-    provider = _resolve_provider_name(
-        env_name="EMBEDDING_PROVIDER",
-        config_provider=cfg.get("provider"),
-        capability="embedding",
-    )
-    if provider != "http":
-        raise ValueError(f"Unsupported embedding provider: {provider}")
-
-    env_text_url = os.getenv("EMBEDDING_TEXT_SERVICE_URL")
-    env_image_url = os.getenv("EMBEDDING_IMAGE_SERVICE_URL")
-    if provider == "http":
-        providers = dict(providers)
-        http_cfg = dict(providers.get("http", {}))
-        if env_text_url:
-            http_cfg["text_base_url"] = env_text_url.rstrip("/")
-        if env_image_url:
-            http_cfg["image_base_url"] = env_image_url.rstrip("/")
-        if not http_cfg.get("text_base_url"):
-            raise ValueError("services.embedding.providers.http.text_base_url is required")
-        if not http_cfg.get("image_base_url"):
-            raise ValueError("services.embedding.providers.http.image_base_url is required")
-        providers["http"] = http_cfg
-
-    return ServiceConfig(provider=provider, providers=providers)
-
-
-def _resolve_rerank() -> ServiceConfig:
-    raw = _load_services_raw()
-    cfg = raw.get("rerank", {}) if isinstance(raw.get("rerank"), dict) else {}
-    providers = cfg.get("providers", {}) if isinstance(cfg.get("providers"), dict) else {}
-
-    provider = _resolve_provider_name(
-        env_name="RERANK_PROVIDER",
-        config_provider=cfg.get("provider"),
-        capability="rerank",
-    )
-    if provider != "http":
-        raise ValueError(f"Unsupported rerank provider: {provider}")
-
-    env_url = os.getenv("RERANKER_SERVICE_URL")
-    if env_url:
-        url = env_url.rstrip("/")
-        if not url.endswith("/rerank"):
-            url = f"{url}/rerank" if "/rerank" not in url else url
-        providers = dict(providers)
-        providers["http"] = dict(providers.get("http", {}))
-        providers["http"]["base_url"] = url.replace("/rerank", "")
-        providers["http"]["service_url"] = url
-
-    return ServiceConfig(provider=provider, providers=providers)
-
-
-def get_rerank_backend_config() -> tuple[str, dict]:
-    raw = _load_services_raw()
-    cfg = raw.get("rerank", {}) if isinstance(raw.get("rerank"), dict) else {}
-    backends = cfg.get("backends", {}) if isinstance(cfg.get("backends"), dict) else {}
-    name = os.getenv("RERANK_BACKEND") or cfg.get("backend")
-    if not name:
-        raise ValueError("services.rerank.backend is required (or env RERANK_BACKEND)")
-    name = str(name).strip().lower()
-    backend_cfg = backends.get(name, {}) if isinstance(backends.get(name), dict) else {}
-    if not backend_cfg:
-        raise ValueError(f"services.rerank.backends.{name} is required")
-    return name, backend_cfg
-
-
-def get_embedding_backend_config() -> tuple[str, dict]:
-    raw = _load_services_raw()
-    cfg = raw.get("embedding", {}) if isinstance(raw.get("embedding"), dict) else {}
-    backends = cfg.get("backends", {}) if isinstance(cfg.get("backends"), dict) else {}
-    name = os.getenv("EMBEDDING_BACKEND") or cfg.get("backend")
-    if not name:
-        raise ValueError("services.embedding.backend is required (or env EMBEDDING_BACKEND)")
-    name = str(name).strip().lower()
-    backend_cfg = backends.get(name, {}) if isinstance(backends.get(name), dict) else {}
-    if not backend_cfg:
-        raise ValueError(f"services.embedding.backends.{name} is required")
-    return name, backend_cfg
-
-
-@lru_cache(maxsize=1)
-def get_translation_config() -> TranslationConfig:
-    return _resolve_translation()
-
-
-@lru_cache(maxsize=1)
-def get_embedding_config() -> ServiceConfig:
-    return _resolve_embedding()
-
-
-@lru_cache(maxsize=1)
-def get_rerank_config() -> ServiceConfig:
-    return _resolve_rerank()
+from typing import Any, Dict, Tuple
+
+from config.loader import get_app_config
+from config.schema import EmbeddingServiceConfig, RerankServiceConfig, TranslationServiceConfig
+
+
+def get_translation_config() -> Dict[str, Any]:
+    return get_app_config().services.translation.as_dict()
+
+
+def get_embedding_config() -> EmbeddingServiceConfig:
+    return get_app_config().services.embedding
+
+
+def get_rerank_config() -> RerankServiceConfig:
+    return get_app_config().services.rerank
 def get_translation_base_url() -> str:
-    return str(get_translation_config()["service_url"])
+    return get_app_config().services.translation.endpoint
 def get_translation_cache_config() -> Dict[str, Any]:
-    return get_translation_cache(get_translation_config())
+    return dict(get_app_config().services.translation.cache)
 def get_embedding_text_base_url() -> str:
-    provider_cfg = get_embedding_config().providers.get("http", {})
-    base = os.getenv("EMBEDDING_TEXT_SERVICE_URL") or provider_cfg.get("text_base_url")
+    provider_cfg = get_app_config().services.embedding.get_provider_config()
+    base = provider_cfg.get("text_base_url")
     if not base:
-        raise ValueError("Embedding text HTTP base_url is not configured")
+        raise ValueError("Embedding text base_url is not configured")
     return str(base).rstrip("/")
 def get_embedding_image_base_url() -> str:
-    provider_cfg = get_embedding_config().providers.get("http", {})
-    base = os.getenv("EMBEDDING_IMAGE_SERVICE_URL") or provider_cfg.get("image_base_url")
+    provider_cfg = get_app_config().services.embedding.get_provider_config()
+    base = provider_cfg.get("image_base_url")
     if not base:
-        raise ValueError("Embedding image HTTP base_url is not configured")
+        raise ValueError("Embedding image base_url is not configured")
     return str(base).rstrip("/")
+def get_embedding_backend_config() -> Tuple[str, Dict[str, Any]]:
+    cfg = get_app_config().services.embedding
+    return cfg.backend, cfg.get_backend_config()
+
+
+def get_embedding_image_backend_config() -> Tuple[str, Dict[str, Any]]:
+    cfg = get_app_config().services.embedding
+    return cfg.image_backend, cfg.get_image_backend_config()
+
+
+def get_rerank_backend_config() -> Tuple[str, Dict[str, Any]]:
+    cfg = get_app_config().services.rerank
+    return cfg.backend, cfg.get_backend_config()
+
+
 def get_rerank_base_url() -> str:
-    base = (
-        os.getenv("RERANKER_SERVICE_URL")
-        or get_rerank_config().providers.get("http", {}).get("service_url")
-        or get_rerank_config().providers.get("http", {}).get("base_url")
-    )
+    provider_cfg = get_app_config().services.rerank.get_provider_config()
+    base = provider_cfg.get("service_url") or provider_cfg.get("base_url")
     if not base:
-        raise ValueError("Rerank HTTP base_url is not configured")
+        raise ValueError("Rerank service URL is not configured")
     return str(base).rstrip("/")
 def get_rerank_service_url() -> str:
-    """Backward-compatible alias."""
     return get_rerank_base_url()
@@ -2,12 +2,13 @@
 租户配置加载器。
 从统一配置文件（config.yaml）加载租户配置，包括主语言和索引语言（index_languages）。
-支持旧配置 translate_to_en / translate_to_zh 的兼容解析。
 """
 import logging
 from typing import Dict, Any, Optional, List
+from config.loader import get_app_config
+
 logger = logging.getLogger(__name__)
 # 支持的索引语言：code -> display name（供商家勾选主市场语言等场景使用）
@@ -83,25 +84,13 @@ def resolve_index_languages(
 ) -> List[str]:
     """
     从租户配置解析 index_languages。
-    若存在 index_languages 则用之；否则按旧配置 translate_to_en / translate_to_zh 推导。
+    若配置缺失或非法，则回退到默认配置。
     """
-    if "index_languages" in tenant_config:
-        normalized = normalize_index_languages(
-            tenant_config["index_languages"],
-            tenant_config.get("primary_language") or "en",
-        )
-        return normalized if normalized else list(default_index_languages)
-    primary = (tenant_config.get("primary_language") or "en").strip().lower()
-    to_en = bool(tenant_config.get("translate_to_en"))
-    to_zh = bool(tenant_config.get("translate_to_zh"))
-    langs: List[str] = []
-    if primary and primary in SOURCE_LANG_CODE_MAP:
-        langs.append(primary)
-    for code in ("en", "zh"):
-        if code not in langs and ((code == "en" and to_en) or (code == "zh" and to_zh)):
-            if code in SOURCE_LANG_CODE_MAP:
-                langs.append(code)
-    return langs if langs else list(default_index_languages)
+    normalized = normalize_index_languages(
+        tenant_config.get("index_languages"),
+        tenant_config.get("primary_language") or "en",
+    )
+    return normalized if normalized else list(default_index_languages)
 class TenantConfigLoader:
@@ -122,15 +111,8 @@ class TenantConfigLoader:
             return self._config
         try:
-            from config import ConfigLoader
-
-            config_loader = ConfigLoader()
-            search_config = config_loader.load_config()
-            tenant_cfg = search_config.tenant_config
-            if not isinstance(tenant_cfg, dict):
-                raise RuntimeError("tenant_config must be an object")
-
-            default_cfg = tenant_cfg.get("default")
+            tenant_cfg = get_app_config().tenants
+            default_cfg = tenant_cfg.default
             if not isinstance(default_cfg, dict):
                 raise RuntimeError("tenant_config.default must be configured in config.yaml")
             default_primary = (default_cfg.get("primary_language") or "en").strip().lower()
@@ -143,7 +125,7 @@ class TenantConfigLoader:
                     "tenant_config.default.index_languages must include at least one supported language"
                 )
-            tenants_cfg = tenant_cfg.get("tenants", {})
+            tenants_cfg = tenant_cfg.tenants
             if not isinstance(tenants_cfg, dict):
                 raise RuntimeError("tenant_config.tenants must be an object")
 """Configuration helper functions for dynamic multi-language search fields."""
 from typing import Dict, List
-from .config_loader import SearchConfig
+
+from config.schema import SearchConfig
 def _format_field_with_boost(field_name: str, boost: float) -> str:
@@ -0,0 +1,738 @@
+# Configuration System Review And Redesign
+
+## 1. Goal
+
+This document reviews the current configuration system and proposes a practical redesign for long-term maintainability.
+
+The target is a configuration system that is:
+
+- unified in loading and ownership
+- clear in boundaries and precedence
+- visible in effective behavior
+- easy to evolve across development, deployment, and operations
+
+This review is based on the current implementation, not only on the intended architecture in docs.
+
+## 2. Project Context
+
+The repo already defines the right architectural direction:
+
+- `config/config.yaml` should be the main configuration source for search behavior and service wiring
+- `.env` should mainly carry deployment-specific values and secrets
+- provider/backend expansion should stay centralized instead of spreading through business code
+
+That direction is described in:
+
+- [`README.md`](/data/saas-search/README.md)
+- [`docs/DEVELOPER_GUIDE.md`](/data/saas-search/docs/DEVELOPER_GUIDE.md)
+- [`docs/QUICKSTART.md`](/data/saas-search/docs/QUICKSTART.md)
+- [`translation/README.md`](/data/saas-search/translation/README.md)
+
+The problem is not the architectural intent. The problem is that the current implementation only partially follows it.
+
+## 3. Current-State Review
+
+### 3.1 What exists today
+
+The current system effectively has several configuration channels:
+
+- `config/config.yaml`
+  - search behavior
+  - rerank behavior
+  - services registry
+  - tenant config
+- `config/config_loader.py`
+  - parses search behavior and tenant config into `SearchConfig`
+  - also injects some defaults from code
+- `config/services_config.py`
+  - reparses `config/config.yaml` again, independently
+  - resolves translation, embedding, rerank service config
+  - also applies env overrides
+- `config/env_config.py`
+  - loads `.env`
+  - defines ES, Redis, DB, host/port, service URLs, namespace, model path defaults
+- service-local config modules
+  - [`embeddings/config.py`](/data/saas-search/embeddings/config.py)
+  - [`reranker/config.py`](/data/saas-search/reranker/config.py)
+- startup scripts
+  - derive defaults from shell env, Python config, and YAML in different combinations
+- inline fallbacks in business logic
+  - query parsing
+  - indexing
+  - service startup
+
+### 3.2 Main findings
+
+#### Finding A: there is no single loader for the full effective configuration
+
+`ConfigLoader` and `services_config` both parse `config/config.yaml`, but they do so separately and with different responsibilities.
+
+- [`config/config_loader.py`](/data/saas-search/config/config_loader.py#L148)
+- [`config/services_config.py`](/data/saas-search/config/services_config.py#L33)
+
+Impact:
+
+- the same file is loaded twice through different code paths
+- search config and services config can drift in interpretation
+- alternative config paths are hard to support cleanly
+- tests and tools cannot ask one place for the full effective config tree
+
+#### Finding B: precedence is not explicit, stable, or globally enforced
+
+Current precedence differs by subsystem:
+
+- search behavior mostly comes from YAML plus code defaults
+- embedding and rerank allow env overrides for provider/backend/url
+- translation intentionally blocks some env overrides
+- startup scripts still choose host/port and mode via env
+- some values are reconstructed from other env vars
+
+Examples:
+
+- env override for embedding provider/url/backend:
+  - [`config/services_config.py`](/data/saas-search/config/services_config.py#L52)
+  - [`config/services_config.py`](/data/saas-search/config/services_config.py#L68)
+  - [`config/services_config.py`](/data/saas-search/config/services_config.py#L139)
+- host/port and service URL reconstruction:
+  - [`config/env_config.py`](/data/saas-search/config/env_config.py#L55)
+  - [`config/env_config.py`](/data/saas-search/config/env_config.py#L75)
+- translator host/port still driven by startup env:
+  - [`scripts/start_translator.sh`](/data/saas-search/scripts/start_translator.sh#L28)
+
+Impact:
+
+- operators cannot reliably predict the effective configuration by reading one file
+- the same setting category behaves differently across services
+- incidents become harder to debug because source-of-truth depends on the code path
+
+#### Finding C: defaults are duplicated across YAML and code
+
+There are several layers of default values:
+
+- dataclass defaults in `QueryConfig`
+- fallback defaults in `ConfigLoader._parse_config`
+- defaults in `config.yaml`
+- defaults in `env_config.py`
+- defaults in `embeddings/config.py`
+- defaults in `reranker/config.py`
+- defaults in startup scripts
+
+Examples:
+
+- query defaults duplicated in dataclass and parser:
+  - [`config/config_loader.py`](/data/saas-search/config/config_loader.py#L24)
+  - [`config/config_loader.py`](/data/saas-search/config/config_loader.py#L240)
+- embedding defaults duplicated in YAML, `services_config`, `embeddings/config.py`, and startup script:
+  - [`config/config.yaml`](/data/saas-search/config/config.yaml#L196)
+  - [`embeddings/config.py`](/data/saas-search/embeddings/config.py#L14)
+  - [`scripts/start_embedding_service.sh`](/data/saas-search/scripts/start_embedding_service.sh#L29)
+- reranker defaults duplicated in YAML and `reranker/config.py`:
+  - [`config/config.yaml`](/data/saas-search/config/config.yaml#L214)
+  - [`reranker/config.py`](/data/saas-search/reranker/config.py#L6)
+
+Impact:
+
+- changing a default is risky because there may be multiple hidden copies
+- code review cannot easily tell whether a value is authoritative or dead legacy
+- “same config” may behave differently across processes
+
+#### Finding D: config is still embedded in runtime logic
+
+Some important behavior remains encoded as inline fallback logic rather than declared config.
+
+Examples:
+
+- query-time translation target languages fallback to `["en", "zh"]`:
+  - [`query/query_parser.py`](/data/saas-search/query/query_parser.py#L339)
+- indexer text handling and LLM enrichment also fallback to `["en", "zh"]`:
+  - [`indexer/document_transformer.py`](/data/saas-search/indexer/document_transformer.py#L216)
+  - [`indexer/document_transformer.py`](/data/saas-search/indexer/document_transformer.py#L310)
+  - [`indexer/document_transformer.py`](/data/saas-search/indexer/document_transformer.py#L649)
+
+Impact:
+
+- configuration is not fully visible in config files
+- behavior can silently change when tenant config is missing or malformed
+- “default behavior” is spread across business modules
+
+#### Finding E: some configuration assets are not managed as first-class config
+
+Query rewrite is configured through an external file, but the file path is hardcoded and currently inconsistent with the repository content.
+
+- loader expects:
+  - [`config/config_loader.py`](/data/saas-search/config/config_loader.py#L162)
+- repo currently contains:
+  - [`config/query_rewrite.dict`](/data/saas-search/config/query_rewrite.dict)
+
+There is also an admin API that mutates rewrite rules in memory only:
+
+- [`api/routes/admin.py`](/data/saas-search/api/routes/admin.py#L68)
+- [`query/query_parser.py`](/data/saas-search/query/query_parser.py#L622)
+
+Impact:
+
+- rewrite rules are neither cleanly file-backed nor fully runtime-managed
+- restart behavior is unclear
+- configuration visibility and persistence are weak
+
+#### Finding F: visibility is limited
+
+The system exposes only a small sanitized subset at `/admin/config`.
+
+- [`api/routes/admin.py`](/data/saas-search/api/routes/admin.py#L42)
+
+At the same time, the true effective config includes:
+
+- tenant overlays
+- env overrides
+- service backend selections
+- script-selected modes
+- hidden defaults in code
+
+Impact:
+
+- there is no authoritative “effective config” view
+- debugging configuration mismatches requires source reading
+- operators cannot easily verify what each process actually started with
+
+#### Finding G: the indexer does not really consume the unified config as a first-class dependency
+
+Indexer startup explicitly says config is loaded only for parity/logging and routes do not depend on it.
+
+- [`api/indexer_app.py`](/data/saas-search/api/indexer_app.py#L76)
+
+Impact:
+
+- configuration is not truly system-wide
+- search-side and indexer-side behavior can drift
+- the current “unified config” is only partially unified
+
+#### Finding H: docs still carry legacy and mixed mental models
+
+Most high-level docs describe the desired centralized model, but some implementation/docs still expose legacy concepts such as `translate_to_en` and `translate_to_zh`.
+
+- desired model:
+  - [`README.md`](/data/saas-search/README.md#L78)
+  - [`docs/DEVELOPER_GUIDE.md`](/data/saas-search/docs/DEVELOPER_GUIDE.md#L207)
+  - [`translation/README.md`](/data/saas-search/translation/README.md#L161)
+- legacy tenant translation flags still documented:
+  - [`indexer/README.md`](/data/saas-search/indexer/README.md#L39)
+
+Impact:
+
+- new developers may follow old mental models
+- cleanup work keeps getting deferred because old and new systems appear both “supported”
+
+## 4. Design Principles For The Redesign
+
+The redesign should follow these rules.
+
+### 4.1 One logical configuration system
+
+It is acceptable to have multiple files, but not multiple loaders with overlapping ownership.
+
+There must be one loader pipeline that produces one typed `AppConfig`.
+
+### 4.2 Configuration files declare, parser code interprets, env provides runtime injection
+
+Responsibilities should be:
+
+- configuration files
+  - declare non-secret desired behavior and non-secret deployable settings
+- parsing logic
+  - load, merge, validate, normalize, and expose typed config
+  - never invent hidden business behavior
+- environment variables
+  - carry secrets and a small set of runtime/process values
+  - do not redefine business behavior casually
+
+### 4.3 One precedence rule for the whole system
+
+Every config category should follow the same merge model unless explicitly exempted.
+
+### 4.4 No silent implicit fallback for business behavior
+
+Fail fast at startup when required config is missing or invalid.
+
+Do not silently fall back to legacy behavior such as hardcoded language lists.
+
+### 4.5 Effective configuration must be observable
+
+Every service should be able to show:
+
+- config version or hash
+- source files loaded
+- environment name
+- sanitized effective configuration
+
+## 5. Recommended Target Design
+
+## 5.1 Boundary model
+
+Use three clear layers.
+
+### Layer 1: repository-managed static config
+
+Purpose:
+
+- search behavior
+- tenant behavior
+- provider/backend registry
+- non-secret service topology defaults
+- feature switches
+
+Examples:
+
+- field boosts
+- query strategy
+- rerank fusion parameters
+- tenant language plans
+- translation capability registry
+- embedding backend selection default
+
+### Layer 2: environment-specific overlays
+
+Purpose:
+
+- per-environment non-secret differences
+- service endpoints by environment
+- resource sizing defaults by environment
+- dev/test/prod operational differences
+
+Examples:
+
+- local embedding URL vs production URL
+- dev rerank backend vs prod rerank backend
+- lower concurrency in local development
+
+### Layer 3: environment variables
+
+Purpose:
+
+- secrets
+- bind host/port
+- external infrastructure credentials
+- container-orchestrator last-mile injection
+
+Examples:
+
+- `ES_HOST`, `ES_USERNAME`, `ES_PASSWORD`
+- `DB_HOST`, `DB_USERNAME`, `DB_PASSWORD`
+- `REDIS_HOST`, `REDIS_PASSWORD`
+- `DASHSCOPE_API_KEY`, `DEEPL_AUTH_KEY`
+- `API_HOST`, `API_PORT`, `INDEXER_PORT`, `TRANSLATION_PORT`
+
+Rule:
+
+- environment variables should not be the normal path for choosing business behavior such as translation model, embedding backend, or tenant language policy
+- if an env override is allowed for a non-secret field, it must be explicitly listed and documented as an operational override, not a hidden convention
+
+## 5.2 Unified precedence
+
+Recommended precedence:
+
+1. schema defaults in code
+2. `config/base.yaml`
+3. `config/environments/<env>.yaml`
+4. tenant overlay from `config/tenants/`
+5. environment variables for the explicitly allowed runtime keys
+6. CLI flags for the current process only
+
+Important rule:
+
+- only one module may implement this merge logic
+- no business module may call `os.getenv()` directly for configuration
+
+## 5.3 Recommended directory structure
+
+```text
+config/
+  schema.py
+  loader.py
+  sources.py
+  base.yaml
+  environments/
+    dev.yaml
+    test.yaml
+    prod.yaml
+  tenants/
+    _default.yaml
+    1.yaml
+    162.yaml
+    170.yaml
+  dictionaries/
+    query_rewrite.dict
+  README.md
+.env.example
+```
+
+Notes:
+
+- `base.yaml` contains shared defaults and feature behavior
+- `environments/*.yaml` contains environment-specific non-secret overrides
+- `tenants/*.yaml` contains tenant-specific overrides only
+- `dictionaries/` stores first-class config assets such as rewrite dictionaries
+- `schema.py` defines the typed config model
+- `loader.py` is the only entry point that loads and merges config
+
+If the team prefers fewer files, `tenants.yaml` is also acceptable. The key requirement is not “one file”, but “one loading model with clear ownership”.
+
+## 5.4 Typed configuration model
+
+Introduce one root object, for example:
+
+```python
+class AppConfig(BaseModel):
+    runtime: RuntimeConfig
+    infrastructure: InfrastructureConfig
+    search: SearchConfig
+    services: ServicesConfig
+    tenants: TenantCatalogConfig
+    assets: ConfigAssets
+```
+
+Suggested subtrees:
+
+- `runtime`
+  - environment name
+  - config revision/hash
+  - bind addresses/ports
+- `infrastructure`
+  - ES
+  - DB
+  - Redis
+  - index namespace
+- `search`
+  - field boosts
+  - query config
+  - function score
+  - rerank behavior
+  - spu config
+- `services`
+  - translation
+  - embedding
+  - rerank
+- `tenants`
+  - default tenant config
+  - tenant overrides
+- `assets`
+  - rewrite dictionary path
+
+Benefits:
+
+- one validated object shared by backend, indexer, translator, embedding, reranker
+- one place for defaults
+- one place for schema evolution
+
+## 5.5 Loading flow
+
+Recommended loading flow:
+
+1. determine `APP_ENV` or `RUNTIME_ENV`
+2. load schema defaults
+3. load `config/base.yaml`
+4. load `config/environments/<env>.yaml` if present
+5. load tenant files
+6. inject first-class assets such as rewrite dictionary
+7. apply allowed env overrides
+8. validate the final `AppConfig`
+9. freeze and cache the config object
+10. expose a sanitized effective-config view
+
+Important:
+
+- every process should call the same loader
+- services should receive a resolved `AppConfig`, not re-open YAML independently
+
+## 5.6 Clear responsibility split
+
+### Configuration files are responsible for
+
+- what the system should do
+- what providers/backends are available
+- which features are enabled
+- tenant language/index policies
+- non-secret service topology
+
+### Parser/loader code is responsible for
+
+- locating sources
+- merge precedence
+- type validation
+- normalization
+- deprecation warnings
+- producing the final immutable config object
+
+### Environment variables are responsible for
+
+- secrets
+- bind addresses/ports
+- infrastructure endpoints when the deployment platform injects them
+- a very small set of documented operational overrides
+
+### Business code is not responsible for
+
+- inventing defaults for missing config
+- loading YAML directly
+- calling `os.getenv()` for normal application behavior
+
+## 5.7 How to handle service config
+
+Unify all service-facing config under one structure:
+
+```yaml
+services:
+  translation:
+    endpoint: "http://translator:6006"
+    timeout_sec: 10
+    default_model: "llm"
+    default_scene: "general"
+    capabilities: ...
+  embedding:
+    endpoint:
+      text: "http://embedding:6005"
+      image: "http://embedding-image:6008"
+    backend: "tei"
+    backends: ...
+  rerank:
+    endpoint: "http://reranker:6007/rerank"
+    backend: "qwen3_vllm"
+    backends: ...
+```
+
+Rules:
+
+- `endpoint` is how callers reach the service
+- `backend` is how the service itself is implemented
+- only the service process cares about `backend`
+- only callers care about `endpoint`
+- both still belong to the same config tree, because they are part of one system
+
+## 5.8 How to handle tenant config
+
+Tenant config should become explicit policy, not translation-era leftovers.
+
+Recommended tenant fields:
+
+- `primary_language`
+- `index_languages`
+- `search_languages`
+- `translation_policy`
+- `facet_policy`
+- optional tenant-specific ranking overrides
+
+Avoid keeping `translate_to_en` and `translate_to_zh` as active concepts in the long-term model.
+
+If compatibility is needed, support them only in the loader as deprecated aliases and emit warnings.
+
+## 5.9 How to handle rewrite rules and similar assets
+
+Treat them as declared config assets.
+
+Recommended rules:
+
+- file path declared in config
+- one canonical location under `config/dictionaries/`
+- loader validates presence and format
+- admin runtime updates either:
+  - are removed, or
+  - write back through a controlled persistence path
+
+Do not keep a hybrid model where startup loads one file and admin mutates only in memory.
+
+## 5.10 Observability improvements
+
+Add the following:
+
+- `config dump` CLI that prints sanitized effective config
+- startup log with config hash, environment, and config file list
+- `/admin/config/effective` endpoint returning sanitized effective config
+- `/admin/config/meta` endpoint returning:
+  - environment
+  - config hash
+  - loaded source files
+  - deprecated keys in use
+
+This is important for operations and for multi-service debugging.
+
+## 6. Practical Refactor Plan
+
+The refactor should be incremental.
+
+### Phase 1: establish the new config core without changing behavior
+
+- create `config/schema.py`
+- create `config/loader.py`
+- move all current defaults into schema models
+- make loader read current `config/config.yaml`
+- make loader read `.env` only for approved keys
+- expose one `get_app_config()`
+
+Result:
+
+- same behavior, but one typed root config becomes available
+
+### Phase 2: remove duplicate readers
+
+- make `services_config.py` a thin adapter over `get_app_config()`
+- make `tenant_config_loader.py` read from `get_app_config()`
+- stop reparsing YAML in `services_config.py`
+- stop service modules from depending on legacy local config modules for behavior
+
+Result:
+
+- one parsing path
+- fewer divergence risks
+
+### Phase 3: move hidden defaults out of business logic
+
+- remove hardcoded fallback language lists from query/indexer modules
+- require tenant defaults to come from config schema only
+- remove duplicate behavior defaults from service code
+
+Result:
+
+- behavior becomes visible and reviewable
+
+### Phase 4: clean service startup configuration
+
+- make startup scripts ask the unified loader for resolved values
+- keep only bind host/port and secret injection in shell env
+- retire or reduce `embeddings/config.py` and `reranker/config.py`
+
+Result:
+
+- startup behavior matches runtime config model
+
+### Phase 5: split config files by responsibility
+
+- keep a single root loader
+- split current giant `config.yaml` into:
+  - `base.yaml`
+  - `environments/<env>.yaml`
+  - `tenants/*.yaml`
+  - `dictionaries/query_rewrite.dict`
+
+Result:
+
+- config remains unified logically, but is easier to read and maintain physically
+
+### Phase 6: deprecate legacy compatibility
+
+- deprecate `translate_to_en` and `translate_to_zh`
+- deprecate env-based backend/provider selection except for explicitly approved keys
+- remove old code paths after one or two release cycles
+
+Result:
+
+- the system becomes simpler instead of carrying two generations forever
+
+## 7. Concrete Rules To Adopt
+
+These rules should be documented and enforced in code review.
+
+### Rule 1
+
+Only `config/loader.py` may load config files or `.env`.
+
+### Rule 2
+
+Only `config/loader.py` may read `os.getenv()` for application config.
+
+### Rule 3
+
+Business modules receive typed config objects and do not read files or env directly.
+
+### Rule 4
+
+Each config key has one owner.
+
+Examples:
+
+- `search.query.knn_boost` belongs to search behavior config
+- `services.embedding.backend` belongs to service implementation config
+- `infrastructure.redis.password` belongs to env/secrets
+
+### Rule 5
+
+Every fallback must be either:
+
+- declared in schema defaults, or
+- rejected at startup
+
+No hidden fallback in runtime logic.
+
+### Rule 6
+
+Every configuration asset must be visible in one of these places only:
+
+- config file
+- env var
+- generated runtime metadata
+
+Not inside parser code as an implicit constant.
+
+## 8. Recommended Naming Conventions
+
+Suggested conventions:
+
+- config keys use noun-based hierarchical names
+- avoid mixing transport and implementation concepts in one field
+- use `endpoint` for caller-facing addresses
+- use `backend` for service-internal implementation choice
+- use `enabled` only for true feature toggles
+- use `default_*` only when a real selection happens at runtime
+
+Examples:
+
+- good: `services.rerank.endpoint`
+- good: `services.rerank.backend`
+- good: `tenants.default.index_languages`
+- avoid: `service_url`, `base_url`, `provider`, `backend`, and script env all meaning slightly different things without a common model
+
+## 9. Highest-Priority Cleanup Items
+
+If the team wants the shortest path to improvement, start here:
+
+1. build one root `AppConfig`
+2. make `services_config.py` stop reparsing YAML
+3. declare rewrite dictionary path explicitly and fix the current mismatch
+4. remove hardcoded `["en", "zh"]` fallbacks from query/indexer logic
+5. replace `/admin/config` with an effective-config endpoint
+6. retire `embeddings/config.py` and `reranker/config.py` as behavior sources
+7. deprecate legacy tenant translation flags
+
+## 10. Expected Outcome
+
+After the redesign:
+
+- developers can answer “where does this setting come from?” in one step
+- operators can see effective config without reading source code
+- backend, indexer, translator, embedding, and reranker all share one model
+- tenant behavior is explicit instead of partially implicit
+- migration becomes safer because defaults and precedence are centralized
+- adding a new provider/backend becomes configuration extension, not configuration archaeology
+
+## 11. Summary
+
+The current system has the right intent but not yet the right implementation shape.
+
+Today the main problems are:
+
+- duplicate config loaders
+- inconsistent precedence
+- duplicated defaults
+- config hidden in runtime logic
+- weak effective-config visibility
+- leftover legacy concepts
+
+The recommended direction is:
+
+- one root typed config
+- one loader pipeline
+- explicit layered sources
+- narrow env responsibility
+- no hidden business fallbacks
+- observable effective config
+
+That design is practical to implement incrementally in this repository and aligns well with the project's multi-tenant, multi-service, provider/backend-based architecture.
-"""
-Embedding module configuration.
+"""Embedding service compatibility config derived from unified app config."""
-This module is intentionally a plain Python file (no env var parsing, no extra deps).
-Edit values here to configure:
-- server host/port
-- local model settings (paths/devices/batch sizes)
-"""
+from __future__ import annotations
 from typing import Optional
-import os
+
+from config.loader import get_app_config
 class EmbeddingConfig(object):
-    # Server
-    HOST = os.getenv("EMBEDDING_HOST", "0.0.0.0")
-    PORT = int(os.getenv("EMBEDDING_PORT", 6005))
-
-    # Text backend defaults
-    TEXT_MODEL_ID = os.getenv("TEXT_MODEL_ID", "Qwen/Qwen3-Embedding-0.6B")
-    # Keep TEXT_MODEL_DIR as an alias so code can refer to one canonical text model value.
-    TEXT_MODEL_DIR = TEXT_MODEL_ID
-    TEXT_DEVICE = os.getenv("TEXT_DEVICE", "cuda")  # "cuda" or "cpu"
-    TEXT_BATCH_SIZE = int(os.getenv("TEXT_BATCH_SIZE", "32"))
-    TEXT_NORMALIZE_EMBEDDINGS = os.getenv("TEXT_NORMALIZE_EMBEDDINGS", "true").lower() in ("1", "true", "yes")
-    TEI_BASE_URL = os.getenv("TEI_BASE_URL", "http://127.0.0.1:8080")
-    TEI_TIMEOUT_SEC = int(os.getenv("TEI_TIMEOUT_SEC", "60"))
-
-    # Image embeddings
-    # Option A: clip-as-service (Jina CLIP server, recommended)
-    USE_CLIP_AS_SERVICE = os.getenv("USE_CLIP_AS_SERVICE", "true").lower() in ("1", "true", "yes")
-    CLIP_AS_SERVICE_SERVER = os.getenv("CLIP_AS_SERVICE_SERVER", "grpc://127.0.0.1:51000")
-    CLIP_AS_SERVICE_MODEL_NAME = os.getenv("CLIP_AS_SERVICE_MODEL_NAME", "CN-CLIP/ViT-L-14")
-
-    # Option B: local CN-CLIP (when USE_CLIP_AS_SERVICE=false)
-    IMAGE_MODEL_NAME = os.getenv("IMAGE_MODEL_NAME", "ViT-L-14")
-    IMAGE_DEVICE = None  # type: Optional[str]  # "cuda" / "cpu" / None(auto)
-
-    # Service behavior
-    IMAGE_BATCH_SIZE = 8
-    IMAGE_NORMALIZE_EMBEDDINGS = os.getenv("IMAGE_NORMALIZE_EMBEDDINGS", "true").lower() in ("1", "true", "yes")
+    def __init__(self) -> None:
+        app_config = get_app_config()
+        runtime = app_config.runtime
+        services = app_config.services.embedding
+        text_backend = services.get_backend_config()
+        image_backend = services.get_image_backend_config()
+
+        self.HOST = runtime.embedding_host
+        self.PORT = runtime.embedding_port
+
+        self.TEXT_MODEL_ID = str(text_backend.get("model_id") or "Qwen/Qwen3-Embedding-0.6B")
+        self.TEXT_MODEL_DIR = self.TEXT_MODEL_ID
+        self.TEXT_DEVICE = str(text_backend.get("device") or "cuda")
+        self.TEXT_BATCH_SIZE = int(text_backend.get("batch_size", 32))
+        self.TEXT_NORMALIZE_EMBEDDINGS = bool(text_backend.get("normalize_embeddings", True))
+        self.TEI_BASE_URL = str(text_backend.get("base_url") or "http://127.0.0.1:8080")
+        self.TEI_TIMEOUT_SEC = int(text_backend.get("timeout_sec", 60))
+
+        self.USE_CLIP_AS_SERVICE = services.image_backend == "clip_as_service"
+        self.CLIP_AS_SERVICE_SERVER = str(image_backend.get("server") or "grpc://127.0.0.1:51000")
+        self.CLIP_AS_SERVICE_MODEL_NAME = str(image_backend.get("model_name") or "CN-CLIP/ViT-L-14")
+
+        self.IMAGE_MODEL_NAME = str(image_backend.get("model_name") or "ViT-L-14")
+        self.IMAGE_DEVICE = image_backend.get("device")  # type: Optional[str]
+        self.IMAGE_BATCH_SIZE = int(image_backend.get("batch_size", 8))
+        self.IMAGE_NORMALIZE_EMBEDDINGS = bool(image_backend.get("normalize_embeddings", True))
 CONFIG = EmbeddingConfig()
@@ -9,8 +9,8 @@ from PIL import Image
 logger = logging.getLogger(__name__)
+from config.loader import get_app_config
 from config.services_config import get_embedding_image_base_url
-from config.env_config import REDIS_CONFIG
 from embeddings.cache_keys import build_image_cache_key
 from embeddings.redis_embedding_cache import RedisEmbeddingCache
@@ -24,10 +24,11 @@ class CLIPImageEncoder:
     def __init__(self, service_url: Optional[str] = None):
         resolved_url = service_url or get_embedding_image_base_url()
+        redis_config = get_app_config().infrastructure.redis
         self.service_url = str(resolved_url).rstrip("/")
         self.endpoint = f"{self.service_url}/embed/image"
         # Reuse embedding cache prefix, but separate namespace for images to avoid collisions.
-        self.cache_prefix = str(REDIS_CONFIG.get("embedding_cache_prefix", "embedding")).strip() or "embedding"
+        self.cache_prefix = str(redis_config.embedding_cache_prefix).strip() or "embedding"
         logger.info("Creating CLIPImageEncoder instance with service URL: %s", self.service_url)
         self.cache = RedisEmbeddingCache(
             key_prefix=self.cache_prefix,
@@ -20,7 +20,7 @@ try:
 except ImportError:  # pragma: no cover - runtime fallback for minimal envs
     redis = None  # type: ignore[assignment]
-from config.env_config import REDIS_CONFIG
+from config.loader import get_app_config
 from embeddings.bf16 import decode_embedding_from_redis, encode_embedding_for_redis
 logger = logging.getLogger(__name__)
@@ -37,7 +37,8 @@ class RedisEmbeddingCache:
     ):
         self.key_prefix = (key_prefix or "").strip() or "embedding"
         self.namespace = (namespace or "").strip()
-        self.expire_time = expire_time or timedelta(days=REDIS_CONFIG.get("cache_expire_days", 180))
+        redis_config = get_app_config().infrastructure.redis
+        self.expire_time = expire_time or timedelta(days=redis_config.cache_expire_days)
         if redis_client is not None:
             self.redis_client = redis_client
@@ -50,13 +51,13 @@ class RedisEmbeddingCache:
         try:
             client = redis.Redis(
-                host=REDIS_CONFIG.get("host", "localhost"),
-                port=REDIS_CONFIG.get("port", 6479),
-                password=REDIS_CONFIG.get("password"),
+                host=redis_config.host,
+                port=redis_config.port,
+                password=redis_config.password,
                 decode_responses=False,
-                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),
+                socket_timeout=redis_config.socket_timeout,
+                socket_connect_timeout=redis_config.socket_connect_timeout,
+                retry_on_timeout=redis_config.retry_on_timeout,
                 health_check_interval=10,
             )
             client.ping()
@@ -470,16 +470,8 @@ def load_models():
             if backend_name == "tei":
                 from embeddings.text_embedding_tei import TEITextModel
-                base_url = (
-                    os.getenv("TEI_BASE_URL")
-                    or backend_cfg.get("base_url")
-                    or CONFIG.TEI_BASE_URL
-                )
-                timeout_sec = int(
-                    os.getenv("TEI_TIMEOUT_SEC")
-                    or backend_cfg.get("timeout_sec")
-                    or CONFIG.TEI_TIMEOUT_SEC
-                )
+                base_url = backend_cfg.get("base_url") or CONFIG.TEI_BASE_URL
+                timeout_sec = int(backend_cfg.get("timeout_sec") or CONFIG.TEI_TIMEOUT_SEC)
                 logger.info("Loading text backend: tei (base_url=%s)", base_url)
                 _text_model = TEITextModel(
                     base_url=str(base_url),
@@ -488,11 +480,7 @@ def load_models():
             elif backend_name == "local_st":
                 from embeddings.text_embedding_sentence_transformers import Qwen3TextModel
-                model_id = (
-                    os.getenv("TEXT_MODEL_ID")
-                    or backend_cfg.get("model_id")
-                    or CONFIG.TEXT_MODEL_ID
-                )
+                model_id = backend_cfg.get("model_id") or CONFIG.TEXT_MODEL_ID
                 logger.info("Loading text backend: local_st (model=%s)", model_id)
                 _text_model = Qwen3TextModel(model_id=str(model_id))
                 _start_text_batch_worker()
@@ -9,13 +9,11 @@ import requests
 logger = logging.getLogger(__name__)
+from config.loader import get_app_config
 from config.services_config import get_embedding_text_base_url
 from embeddings.cache_keys import build_text_cache_key
 from embeddings.redis_embedding_cache import RedisEmbeddingCache
-# Try to import REDIS_CONFIG, but allow import to fail
-from config.env_config import REDIS_CONFIG
-
 class TextEmbeddingEncoder:
     """
@@ -24,10 +22,11 @@ class TextEmbeddingEncoder:
     def __init__(self, service_url: Optional[str] = None):
         resolved_url = service_url or get_embedding_text_base_url()
+        redis_config = get_app_config().infrastructure.redis
         self.service_url = str(resolved_url).rstrip("/")
         self.endpoint = f"{self.service_url}/embed/text"
-        self.expire_time = timedelta(days=REDIS_CONFIG.get("cache_expire_days", 180))
-        self.cache_prefix = str(REDIS_CONFIG.get("embedding_cache_prefix", "embedding")).strip() or "embedding"
+        self.expire_time = timedelta(days=redis_config.cache_expire_days)
+        self.cache_prefix = str(redis_config.embedding_cache_prefix).strip() or "embedding"
         logger.info("Creating TextEmbeddingEncoder instance with service URL: %s", self.service_url)
         self.cache = RedisEmbeddingCache(
@@ -13,7 +13,6 @@ import numpy as np
 import logging
 import re
 from typing import Dict, Any, Optional, List
-from config import ConfigLoader
 from indexer.product_enrich import analyze_products
 logger = logging.getLogger(__name__)
@@ -13,7 +13,7 @@ from indexer.mapping_generator import get_tenant_index_name
 from indexer.indexer_logger import (
     get_indexer_logger, log_index_request, log_index_result, log_spu_processing
 )
-from config import ConfigLoader
+from config import get_app_config
 from translation import create_translation_client
 # Configure logger
@@ -51,7 +51,7 @@ 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._config = get_app_config().search
         self._searchable_option_dimensions = (
             getattr(self._config.spu_config, "searchable_option_dimensions", None)
             or ["option1", "option2", "option3"]
@@ -7,7 +7,7 @@
 import logging
 from typing import Dict, Any, Optional
 from sqlalchemy import Engine, text
-from config import ConfigLoader
+from config import get_app_config
 from config.tenant_config_loader import get_tenant_config_loader
 from indexer.document_transformer import SPUDocumentTransformer
 from translation import create_translation_client
@@ -92,8 +92,7 @@ def create_document_transformer(
         or config is None
     ):
         if config is None:
-            config_loader = ConfigLoader()
-            config = config_loader.load_config()
+            config = get_app_config().search
         if searchable_option_dimensions is None:
             searchable_option_dimensions = config.spu_config.searchable_option_dimensions
@@ -9,7 +9,7 @@ import json
 import logging
 from pathlib import Path
-from config.env_config import ES_INDEX_NAMESPACE
+from config.loader import get_app_config
 logger = logging.getLogger(__name__)
@@ -30,7 +30,7 @@ def get_tenant_index_name(tenant_id: str) -&gt; str:
     其中 ES_INDEX_NAMESPACE 由 config.env_config.ES_INDEX_NAMESPACE 控制，
     用于区分 prod/uat/test 等不同运行环境。
     """
-    prefix = ES_INDEX_NAMESPACE or ""
+    prefix = get_app_config().runtime.index_namespace or ""
     return f"{prefix}search_products_tenant_{tenant_id}"
@@ -20,7 +20,7 @@ import redis
 import requests
 from pathlib import Path
-from config.env_config import REDIS_CONFIG
+from config.loader import get_app_config
 from config.tenant_config_loader import SOURCE_LANG_CODE_MAP
 from indexer.product_enrich_prompts import (
     SYSTEM_MESSAGE,
@@ -91,19 +91,20 @@ logger.info(&quot;Verbose LLM logs are written to: %s&quot;, verbose_log_file)
 # Redis 缓存（用于 anchors / 语义属性）
-ANCHOR_CACHE_PREFIX = REDIS_CONFIG.get("anchor_cache_prefix", "product_anchors")
-ANCHOR_CACHE_EXPIRE_DAYS = int(REDIS_CONFIG.get("anchor_cache_expire_days", 30))
+_REDIS_CONFIG = get_app_config().infrastructure.redis
+ANCHOR_CACHE_PREFIX = _REDIS_CONFIG.anchor_cache_prefix
+ANCHOR_CACHE_EXPIRE_DAYS = int(_REDIS_CONFIG.anchor_cache_expire_days)
 _anchor_redis: Optional[redis.Redis] = None
 try:
     _anchor_redis = redis.Redis(
-        host=REDIS_CONFIG.get("host", "localhost"),
-        port=REDIS_CONFIG.get("port", 6479),
-        password=REDIS_CONFIG.get("password"),
+        host=_REDIS_CONFIG.host,
+        port=_REDIS_CONFIG.port,
+        password=_REDIS_CONFIG.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),
+        socket_timeout=_REDIS_CONFIG.socket_timeout,
+        socket_connect_timeout=_REDIS_CONFIG.socket_connect_timeout,
+        retry_on_timeout=_REDIS_CONFIG.retry_on_timeout,
         health_check_interval=10,
     )
     _anchor_redis.ping()
@@ -16,8 +16,7 @@ import json
 # Add parent directory to path
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
-from config import ConfigLoader
-from config.env_config import ES_CONFIG
+from config import get_app_config
 from utils import ESClient
 from search import Searcher
 from suggestion import SuggestionIndexBuilder
@@ -61,8 +60,7 @@ def cmd_serve_indexer(args):
 def cmd_search(args):
     """Test search from command line."""
     # Load config
-    config_loader = ConfigLoader("config/config.yaml")
-    config = config_loader.load_config()
+    config = get_app_config().search
     # Initialize ES and searcher
     es_client = ESClient(hosts=[args.es_host])
@@ -106,8 +104,9 @@ def cmd_search(args):
 def cmd_build_suggestions(args):
     """Build/update suggestion index for a tenant."""
     # Initialize ES client with optional authentication
-    es_username = os.getenv("ES_USERNAME") or ES_CONFIG.get("username")
-    es_password = os.getenv("ES_PASSWORD") or ES_CONFIG.get("password")
+    es_cfg = get_app_config().infrastructure.elasticsearch
+    es_username = es_cfg.username
+    es_password = es_cfg.password
     if es_username and es_password:
         es_client = ESClient(hosts=[args.es_host], username=es_username, password=es_password)
     else:
@@ -117,11 +116,12 @@ def cmd_build_suggestions(args):
         return 1
     # Build DB config directly from environment to avoid dotenv dependency
-    db_host = os.getenv("DB_HOST")
-    db_port = int(os.getenv("DB_PORT", "3306"))
-    db_name = os.getenv("DB_DATABASE")
-    db_user = os.getenv("DB_USERNAME")
-    db_pass = os.getenv("DB_PASSWORD")
+    db_cfg = get_app_config().infrastructure.database
+    db_host = db_cfg.host
+    db_port = db_cfg.port
+    db_name = db_cfg.database
+    db_user = db_cfg.username
+    db_pass = db_cfg.password
     if not all([db_host, db_name, db_user, db_pass]):
         print("ERROR: DB_HOST/DB_PORT/DB_DATABASE/DB_USERNAME/DB_PASSWORD must be set in environment")
         return 1
@@ -170,7 +170,7 @@ def main():
     serve_parser = subparsers.add_parser('serve', help='Start API service (multi-tenant)')
     serve_parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')
     serve_parser.add_argument('--port', type=int, default=6002, help='Port to bind to')
-    serve_parser.add_argument('--es-host', default=ES_CONFIG.get('host', 'http://localhost:9200'), help='Elasticsearch host')
+    serve_parser.add_argument('--es-host', default=get_app_config().infrastructure.elasticsearch.host, help='Elasticsearch host')
     serve_parser.add_argument('--reload', action='store_true', help='Enable auto-reload')
     # Serve-indexer command
@@ -180,14 +180,14 @@ def main():
     )
     serve_indexer_parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')
     serve_indexer_parser.add_argument('--port', type=int, default=6004, help='Port to bind to')
-    serve_indexer_parser.add_argument('--es-host', default=ES_CONFIG.get('host', 'http://localhost:9200'), help='Elasticsearch host')
+    serve_indexer_parser.add_argument('--es-host', default=get_app_config().infrastructure.elasticsearch.host, help='Elasticsearch host')
     serve_indexer_parser.add_argument('--reload', action='store_true', help='Enable auto-reload')
     # Search command
     search_parser = subparsers.add_parser('search', help='Test search from command line')
     search_parser.add_argument('query', help='Search query')
     search_parser.add_argument('--tenant-id', required=True, help='Tenant ID (required)')
-    search_parser.add_argument('--es-host', default=ES_CONFIG.get('host', 'http://localhost:9200'), help='Elasticsearch host')
+    search_parser.add_argument('--es-host', default=get_app_config().infrastructure.elasticsearch.host, help='Elasticsearch host')
     search_parser.add_argument('--size', type=int, default=10, help='Number of results')
     search_parser.add_argument('--no-translation', action='store_true', help='Disable translation')
     search_parser.add_argument('--no-embedding', action='store_true', help='Disable embeddings')
@@ -199,7 +199,7 @@ def main():
         help='Build tenant suggestion index (full/incremental)'
     )
     suggest_build_parser.add_argument('--tenant-id', required=True, help='Tenant ID')
-    suggest_build_parser.add_argument('--es-host', default=ES_CONFIG.get('host', 'http://localhost:9200'), help='Elasticsearch host')
+    suggest_build_parser.add_argument('--es-host', default=get_app_config().infrastructure.elasticsearch.host, help='Elasticsearch host')
     suggest_build_parser.add_argument(
         '--mode',
         choices=['full', 'incremental'],
@@ -336,13 +336,13 @@ class QueryParser:
         translations = {}
         translation_futures = {}
         translation_executor = None
-        index_langs = ["en", "zh"]
+        index_langs: List[str] = []
         try:
             # 根据租户配置的 index_languages 决定翻译目标语言
             from config.tenant_config_loader import get_tenant_config_loader
             tenant_loader = get_tenant_config_loader()
             tenant_cfg = tenant_loader.get_tenant_config(tenant_id or "default")
-            raw_index_langs = tenant_cfg.get("index_languages") or ["en", "zh"]
+            raw_index_langs = tenant_cfg.get("index_languages") or []
             index_langs = []
             seen_langs = set()
             for lang in raw_index_langs:
@@ -63,43 +63,19 @@ class DashScopeRerankBackend:
       - max_retries: int, default 1
       - retry_backoff_sec: float, default 0.2
-    Env overrides:
-      - RERANK_DASHSCOPE_ENDPOINT
-      - RERANK_DASHSCOPE_MODEL
-      - RERANK_DASHSCOPE_TIMEOUT_SEC
-      - RERANK_DASHSCOPE_TOP_N_CAP
-      - RERANK_DASHSCOPE_BATCHSIZE
     """
     def __init__(self, config: Dict[str, Any]) -> None:
         self._config = config or {}
-        self._model_name = str(
-            os.getenv("RERANK_DASHSCOPE_MODEL")
-            or self._config.get("model_name")
-            or "qwen3-rerank"
-        )
+        self._model_name = str(self._config.get("model_name") or "qwen3-rerank")
         self._endpoint = str(
-            os.getenv("RERANK_DASHSCOPE_ENDPOINT")
-            or self._config.get("endpoint")
-            or "https://dashscope.aliyuncs.com/compatible-api/v1/reranks"
+            self._config.get("endpoint") or "https://dashscope.aliyuncs.com/compatible-api/v1/reranks"
         ).strip()
         self._api_key_env = str(self._config.get("api_key_env") or "").strip()
         self._api_key = str(os.getenv(self._api_key_env) or "").strip().strip('"').strip("'")
-        self._timeout_sec = float(
-            os.getenv("RERANK_DASHSCOPE_TIMEOUT_SEC")
-            or self._config.get("timeout_sec")
-            or 15.0
-        )
-        self._top_n_cap = int(
-            os.getenv("RERANK_DASHSCOPE_TOP_N_CAP")
-            or self._config.get("top_n_cap")
-            or 0
-        )
-        self._batchsize = int(
-            os.getenv("RERANK_DASHSCOPE_BATCHSIZE")
-            or self._config.get("batchsize")
-            or 0
-        )
+        self._timeout_sec = float(self._config.get("timeout_sec") or 15.0)
+        self._top_n_cap = int(self._config.get("top_n_cap") or 0)
+        self._batchsize = int(self._config.get("batchsize") or 0)
         self._instruct = str(self._config.get("instruct") or "").strip()
         self._max_retries = int(self._config.get("max_retries", 1))
         self._retry_backoff_sec = float(self._config.get("retry_backoff_sec", 0.2))
-"""Reranker service configuration (simple Python config)."""
+"""Reranker service compatibility config derived from unified app config."""
-import os
+from __future__ import annotations
+
+from config.loader import get_app_config
 class RerankerConfig(object):
-    # Server
-    HOST = os.getenv("RERANKER_HOST", "0.0.0.0")
-    PORT = int(os.getenv("RERANKER_PORT", 6007))
-
-    # Model
-    MODEL_NAME = "Qwen/Qwen3-Reranker-0.6B"
-    DEVICE = None  # None -> auto (cuda if available)
-    USE_FP16 = True
-    BATCH_SIZE = 64
-    MAX_LENGTH = 512
-    CACHE_DIR = "./model_cache"
-    ENABLE_WARMUP = True
-
-    # Request limits
-    MAX_DOCS = 1000
-
-    # Output
-    NORMALIZE = True
+    def __init__(self) -> None:
+        app_config = get_app_config()
+        runtime = app_config.runtime
+        service = app_config.services.rerank
+        backend = service.get_backend_config()
+        request = service.request
+
+        self.HOST = runtime.reranker_host
+        self.PORT = runtime.reranker_port
+
+        self.MODEL_NAME = str(backend.get("model_name") or "Qwen/Qwen3-Reranker-0.6B")
+        self.DEVICE = backend.get("device")
+        self.USE_FP16 = bool(backend.get("use_fp16", True))
+        self.BATCH_SIZE = int(backend.get("batch_size", backend.get("infer_batch_size", 64)))
+        self.MAX_LENGTH = int(backend.get("max_length", 512))
+        self.CACHE_DIR = str(backend.get("cache_dir") or "./model_cache")
+        self.ENABLE_WARMUP = bool(backend.get("enable_warmup", True))
+
+        self.MAX_DOCS = int(request.get("max_docs", 1000))
+        self.NORMALIZE = bool(request.get("normalize", True))
 CONFIG = RerankerConfig()
@@ -18,7 +18,7 @@ from typing import Any, Dict, Iterator, List, Optional, Tuple
 from sqlalchemy import text
-from config.env_config import ES_INDEX_NAMESPACE
+from config.loader import get_app_config
 from config.tenant_config_loader import get_tenant_config_loader
 from suggestion.mapping import build_suggestion_mapping
 from utils.es_client import ESClient
@@ -27,7 +27,7 @@ logger = logging.getLogger(__name__)
 def _index_prefix() -> str:
-    return ES_INDEX_NAMESPACE or ""
+    return get_app_config().runtime.index_namespace or ""
 def get_suggestion_alias_name(tenant_id: str) -> str:
@@ -3,7 +3,6 @@
 from __future__ import annotations
 import logging
-import os
 import re
 from typing import List, Optional, Sequence, Tuple, Union
@@ -24,7 +23,7 @@ class DeepLTranslationBackend:
         timeout: float,
         glossary_id: Optional[str] = None,
     ) -> None:
-        self.api_key = api_key or os.getenv("DEEPL_AUTH_KEY")
+        self.api_key = api_key
         self.api_url = api_url
         self.timeout = float(timeout)
         self.glossary_id = glossary_id
@@ -3,13 +3,11 @@
 from __future__ import annotations
 import logging
-import os
 import time
 from typing import List, Optional, Sequence, Union
 from openai import OpenAI
-from config.env_config import DASHSCOPE_API_KEY
 from translation.languages import LANGUAGE_LABELS
 from translation.prompts import TRANSLATION_PROMPTS
 from translation.scenes import normalize_scene_name
@@ -52,11 +50,13 @@ class LLMTranslationBackend:
         model: str,
         timeout_sec: float,
         base_url: str,
+        api_key: Optional[str],
     ) -> None:
         self.capability_name = capability_name
         self.model = model
         self.timeout_sec = float(timeout_sec)
         self.base_url = base_url
+        self.api_key = api_key
         self.client = self._create_client()
     @property
@@ -64,12 +64,11 @@ class LLMTranslationBackend:
         return True
     def _create_client(self) -> Optional[OpenAI]:
-        api_key = DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY")
-        if not api_key:
+        if not self.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)
+            return OpenAI(api_key=self.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
@@ -3,14 +3,12 @@
 from __future__ import annotations
 import logging
-import os
 import re
 import time
 from typing import List, Optional, Sequence, Union
 from openai import OpenAI
-from config.env_config import DASHSCOPE_API_KEY
 from translation.languages import QWEN_LANGUAGE_CODES
 logger = logging.getLogger(__name__)
@@ -64,7 +62,7 @@ class QwenMTTranslationBackend:
     @staticmethod
     def _default_api_key(model: str) -> Optional[str]:
         del model
-        return DASHSCOPE_API_KEY or os.getenv("DASHSCOPE_API_KEY")
+        return None
     def translate(
         self,
@@ -6,9 +6,12 @@ import hashlib
 import logging
 from typing import Mapping, Optional
-import redis
+try:
+    import redis
+except ImportError:  # pragma: no cover - runtime fallback for minimal envs
+    redis = None  # type: ignore[assignment]
-from config.env_config import REDIS_CONFIG
+from config.loader import get_app_config
 logger = logging.getLogger(__name__)
@@ -70,15 +73,19 @@ class TranslationCache:
     @staticmethod
     def _init_redis_client() -> Optional[redis.Redis]:
+        if redis is None:
+            logger.warning("redis package is not installed; translation cache disabled")
+            return None
+        redis_config = get_app_config().infrastructure.redis
         try:
             client = redis.Redis(
-                host=REDIS_CONFIG.get("host", "localhost"),
-                port=REDIS_CONFIG.get("port", 6479),
-                password=REDIS_CONFIG.get("password"),
+                host=redis_config.host,
+                port=redis_config.port,
+                password=redis_config.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),
+                socket_timeout=redis_config.socket_timeout,
+                socket_connect_timeout=redis_config.socket_connect_timeout,
+                retry_on_timeout=redis_config.retry_on_timeout,
                 health_check_interval=10,
             )
             client.ping()
@@ -7,7 +7,7 @@ from typing import List, Optional, Sequence, Union
 import requests
-from config.services_config import get_translation_config
+from config.loader import get_app_config
 from translation.settings import normalize_translation_model, normalize_translation_scene
 logger = logging.getLogger(__name__)
@@ -24,7 +24,7 @@ class TranslationServiceClient:
         default_scene: Optional[str] = None,
         timeout_sec: Optional[float] = None,
     ) -> None:
-        cfg = get_translation_config()
+        cfg = get_app_config().services.translation.as_dict()
         self.base_url = str(base_url or cfg["service_url"]).rstrip("/")
         self.default_model = normalize_translation_model(cfg, default_model or cfg["default_model"])
         self.default_scene = normalize_translation_scene(cfg, default_scene or cfg["default_scene"])
@@ -5,7 +5,8 @@ from __future__ import annotations
 import logging
 from typing import Dict, List, Optional
-from config.services_config import get_translation_config
+from config.loader import get_app_config
+from config.schema import AppConfig
 from translation.cache import TranslationCache
 from translation.protocols import TranslateInput, TranslateOutput, TranslationBackendProtocol
 from translation.settings import (
@@ -22,8 +23,9 @@ logger = logging.getLogger(__name__)
 class TranslationService:
     """Owns translation backends and routes calls by model and scene."""
-    def __init__(self, config: Optional[TranslationConfig] = None) -> None:
-        self.config = config or get_translation_config()
+    def __init__(self, config: Optional[TranslationConfig] = None, app_config: Optional[AppConfig] = None) -> None:
+        self._app_config = app_config or get_app_config()
+        self.config = config or self._app_config.services.translation.as_dict()
         self._enabled_capabilities = self._collect_enabled_capabilities()
         if not self._enabled_capabilities:
             raise ValueError("No enabled translation backends found in services.translation.capabilities")
@@ -85,7 +87,7 @@ class TranslationService:
             capability_name=name,
             model=str(cfg["model"]).strip(),
             base_url=str(cfg["base_url"]).strip(),
-            api_key=cfg.get("api_key"),
+            api_key=self._app_config.infrastructure.secrets.dashscope_api_key,
             timeout=int(cfg["timeout_sec"]),
             glossary_id=cfg.get("glossary_id"),
         )
@@ -94,7 +96,7 @@ class TranslationService:
         from translation.backends.deepl import DeepLTranslationBackend
         return DeepLTranslationBackend(
-            api_key=cfg.get("api_key"),
+            api_key=self._app_config.infrastructure.secrets.deepl_auth_key,
             api_url=str(cfg["api_url"]).strip(),
             timeout=float(cfg["timeout_sec"]),
             glossary_id=cfg.get("glossary_id"),
@@ -108,6 +110,7 @@ class TranslationService:
             model=str(cfg["model"]).strip(),
             timeout_sec=float(cfg["timeout_sec"]),
             base_url=str(cfg["base_url"]).strip(),
+            api_key=self._app_config.infrastructure.secrets.dashscope_api_key,
         )
     def _create_local_nllb_backend(self, *, name: str, cfg: Dict[str, object]) -> TranslationBackendProtocol:
@@ -5,10 +5,9 @@ Elasticsearch client wrapper.
 from elasticsearch import Elasticsearch
 from elasticsearch.helpers import bulk
 from typing import Dict, Any, List, Optional
-import os
 import logging
-from config.env_config import ES_CONFIG
+from config.loader import get_app_config
 logger = logging.getLogger(__name__)
@@ -33,7 +32,7 @@ class ESClient:
             **kwargs: Additional ES client parameters
         """
         if hosts is None:
-            hosts = [os.getenv('ES_HOST', 'http://localhost:9200')]
+            hosts = [get_app_config().infrastructure.elasticsearch.host]
         # Build client config
         client_config = {
@@ -325,16 +324,9 @@ def get_es_client_from_env() -&gt; ESClient:
     Returns:
         ESClient instance
     """
-    if ES_CONFIG:
-        return ESClient(
-            hosts=[ES_CONFIG['host']],
-            username=ES_CONFIG.get('username'),
-            password=ES_CONFIG.get('password')
-        )
-    else:
-        # Fallback to env variables
-        return ESClient(
-            hosts=[os.getenv('ES_HOST', 'http://localhost:9200')],
-            username=os.getenv('ES_USERNAME'),
-            password=os.getenv('ES_PASSWORD')
-        )
+    cfg = get_app_config().infrastructure.elasticsearch
+    return ESClient(
+        hosts=[cfg.host],
+        username=cfg.username,
+        password=cfg.password,
+    )