Blame view

config/services_config.py 2.95 KB
42e3aea6   tangwang   tidy
1
  """
86d8358b   tangwang   config optimize
2
  Unified service configuration accessors.
42e3aea6   tangwang   tidy
3
  
86d8358b   tangwang   config optimize
4
5
  This module is now a thin adapter over ``config.loader.get_app_config`` and
  contains no independent parsing or precedence logic.
42e3aea6   tangwang   tidy
6
7
8
9
  """
  
  from __future__ import annotations
  
5c21a485   tangwang   qwen3-reranker-0....
10
  import os
86d8358b   tangwang   config optimize
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
  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
42e3aea6   tangwang   tidy
27
28
29
  
  
  def get_translation_base_url() -> str:
86d8358b   tangwang   config optimize
30
      return get_app_config().services.translation.endpoint
42e3aea6   tangwang   tidy
31
32
  
  
d4cadc13   tangwang   翻译重构
33
  def get_translation_cache_config() -> Dict[str, Any]:
86d8358b   tangwang   config optimize
34
      return dict(get_app_config().services.translation.cache)
d4cadc13   tangwang   翻译重构
35
36
  
  
7214c2e7   tangwang   mplemented**
37
  def get_embedding_text_base_url() -> str:
86d8358b   tangwang   config optimize
38
39
      provider_cfg = get_app_config().services.embedding.get_provider_config()
      base = provider_cfg.get("text_base_url")
7214c2e7   tangwang   mplemented**
40
      if not base:
86d8358b   tangwang   config optimize
41
          raise ValueError("Embedding text base_url is not configured")
7214c2e7   tangwang   mplemented**
42
43
44
45
      return str(base).rstrip("/")
  
  
  def get_embedding_image_base_url() -> str:
86d8358b   tangwang   config optimize
46
47
      provider_cfg = get_app_config().services.embedding.get_provider_config()
      base = provider_cfg.get("image_base_url")
7214c2e7   tangwang   mplemented**
48
      if not base:
86d8358b   tangwang   config optimize
49
          raise ValueError("Embedding image base_url is not configured")
7214c2e7   tangwang   mplemented**
50
51
52
      return str(base).rstrip("/")
  
  
86d8358b   tangwang   config optimize
53
54
55
56
57
58
59
60
61
62
63
64
  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
5c21a485   tangwang   qwen3-reranker-0....
65
66
67
68
69
70
      backend = str(os.getenv("RERANK_BACKEND") or cfg.backend).strip()
      if backend != cfg.backend:
          backend_cfg = cfg.backends.get(backend)
          if backend_cfg is None:
              raise ValueError(f"Unknown rerank backend override from RERANK_BACKEND: {backend!r}")
          return backend, dict(backend_cfg)
86d8358b   tangwang   config optimize
71
72
73
      return cfg.backend, cfg.get_backend_config()
  
  
8c8b9d84   tangwang   ES 拉取 coarse_rank...
74
  def get_rerank_base_url(profile: str | None = None) -> str:
86d8358b   tangwang   config optimize
75
      provider_cfg = get_app_config().services.rerank.get_provider_config()
8c8b9d84   tangwang   ES 拉取 coarse_rank...
76
77
78
79
80
81
82
83
      base = None
      profile_name = str(profile).strip() if profile else ""
      if profile_name:
          service_urls = provider_cfg.get("service_urls")
          if isinstance(service_urls, dict):
              base = service_urls.get(profile_name)
      if not base:
          base = provider_cfg.get("service_url") or provider_cfg.get("base_url")
26b910bd   tangwang   refactor service ...
84
      if not base:
86d8358b   tangwang   config optimize
85
          raise ValueError("Rerank service URL is not configured")
5e4dc8e4   tangwang   翻译架构按“一个翻译服务 +
86
      return str(base).rstrip("/")
42e3aea6   tangwang   tidy
87
88
  
  
8c8b9d84   tangwang   ES 拉取 coarse_rank...
89
90
  def get_rerank_service_url(profile: str | None = None) -> str:
      return get_rerank_base_url(profile=profile)