services_config.py
4.56 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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
"""
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,
RerankServiceInstanceConfig,
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_rerank_instance_config(profile: str | None = None) -> RerankServiceInstanceConfig:
cfg = get_app_config().services.rerank
instance_name = str(
profile
or os.getenv("RERANK_INSTANCE")
or cfg.default_instance
).strip() or cfg.default_instance
return cfg.get_instance(instance_name)
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(profile: str | None = None) -> Tuple[str, Dict[str, Any]]:
cfg = get_app_config().services.rerank
instance = get_rerank_instance_config(profile)
backend = str(os.getenv("RERANK_BACKEND") or instance.backend).strip()
if backend != instance.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 instance.backend, cfg.get_backend_config(profile)
def get_rerank_base_url(profile: str | None = None) -> str:
cfg = get_app_config().services.rerank
instance = get_rerank_instance_config(profile)
provider_cfg = cfg.get_provider_config()
profile_name = str(profile or os.getenv("RERANK_INSTANCE") or cfg.default_instance).strip() or cfg.default_instance
base = None
provider_instances = provider_cfg.get("instances")
if isinstance(provider_instances, dict):
instance_provider_cfg = provider_instances.get(profile_name)
if isinstance(instance_provider_cfg, dict):
base = instance_provider_cfg.get("base_url") or instance_provider_cfg.get("service_url")
if not base:
base = instance.base_url or instance.service_url
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")
base = str(base).rstrip("/")
if base.endswith("/rerank"):
base = base[: -len("/rerank")]
return base
def get_rerank_service_url(profile: str | None = None) -> str:
cfg = get_app_config().services.rerank
instance = get_rerank_instance_config(profile)
provider_cfg = cfg.get_provider_config()
profile_name = str(profile or os.getenv("RERANK_INSTANCE") or cfg.default_instance).strip() or cfg.default_instance
service_url = None
provider_instances = provider_cfg.get("instances")
if isinstance(provider_instances, dict):
instance_provider_cfg = provider_instances.get(profile_name)
if isinstance(instance_provider_cfg, dict):
service_url = instance_provider_cfg.get("service_url")
if not service_url:
service_url = instance.service_url
if not service_url:
service_url = f"{get_rerank_base_url(profile=profile)}/rerank"
return str(service_url).rstrip("/")