services_config.py
2.95 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
"""
Unified service configuration accessors.
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 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 get_app_config().services.translation.endpoint
def get_translation_cache_config() -> Dict[str, Any]:
return dict(get_app_config().services.translation.cache)
def get_embedding_text_base_url() -> str:
provider_cfg = get_app_config().services.embedding.get_provider_config()
base = provider_cfg.get("text_base_url")
if not base:
raise ValueError("Embedding text base_url is not configured")
return str(base).rstrip("/")
def get_embedding_image_base_url() -> str:
provider_cfg = get_app_config().services.embedding.get_provider_config()
base = provider_cfg.get("image_base_url")
if not base:
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
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)
return cfg.backend, cfg.get_backend_config()
def get_rerank_base_url(profile: str | None = None) -> str:
provider_cfg = get_app_config().services.rerank.get_provider_config()
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")
if not base:
raise ValueError("Rerank service URL is not configured")
return str(base).rstrip("/")
def get_rerank_service_url(profile: str | None = None) -> str:
return get_rerank_base_url(profile=profile)