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
|
from typing import Any, Dict, Tuple
from config.loader import get_app_config
|
daa2690b
tangwang
漏斗参数调优&呈现优化
|
14
15
16
17
18
19
|
from config.schema import (
EmbeddingServiceConfig,
RerankServiceConfig,
RerankServiceInstanceConfig,
TranslationServiceConfig,
)
|
86d8358b
tangwang
config optimize
|
20
21
22
23
24
25
26
27
28
29
30
31
|
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
|
32
33
|
|
daa2690b
tangwang
漏斗参数调优&呈现优化
|
34
35
36
37
38
39
40
41
42
43
|
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)
|
42e3aea6
tangwang
tidy
|
44
|
def get_translation_base_url() -> str:
|
86d8358b
tangwang
config optimize
|
45
|
return get_app_config().services.translation.endpoint
|
42e3aea6
tangwang
tidy
|
46
47
|
|
d4cadc13
tangwang
翻译重构
|
48
|
def get_translation_cache_config() -> Dict[str, Any]:
|
86d8358b
tangwang
config optimize
|
49
|
return dict(get_app_config().services.translation.cache)
|
d4cadc13
tangwang
翻译重构
|
50
51
|
|
7214c2e7
tangwang
mplemented**
|
52
|
def get_embedding_text_base_url() -> str:
|
86d8358b
tangwang
config optimize
|
53
54
|
provider_cfg = get_app_config().services.embedding.get_provider_config()
base = provider_cfg.get("text_base_url")
|
7214c2e7
tangwang
mplemented**
|
55
|
if not base:
|
86d8358b
tangwang
config optimize
|
56
|
raise ValueError("Embedding text base_url is not configured")
|
7214c2e7
tangwang
mplemented**
|
57
58
59
60
|
return str(base).rstrip("/")
def get_embedding_image_base_url() -> str:
|
86d8358b
tangwang
config optimize
|
61
62
|
provider_cfg = get_app_config().services.embedding.get_provider_config()
base = provider_cfg.get("image_base_url")
|
7214c2e7
tangwang
mplemented**
|
63
|
if not base:
|
86d8358b
tangwang
config optimize
|
64
|
raise ValueError("Embedding image base_url is not configured")
|
7214c2e7
tangwang
mplemented**
|
65
66
67
|
return str(base).rstrip("/")
|
86d8358b
tangwang
config optimize
|
68
69
70
71
72
73
74
75
76
77
|
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()
|
daa2690b
tangwang
漏斗参数调优&呈现优化
|
78
|
def get_rerank_backend_config(profile: str | None = None) -> Tuple[str, Dict[str, Any]]:
|
86d8358b
tangwang
config optimize
|
79
|
cfg = get_app_config().services.rerank
|
daa2690b
tangwang
漏斗参数调优&呈现优化
|
80
81
82
|
instance = get_rerank_instance_config(profile)
backend = str(os.getenv("RERANK_BACKEND") or instance.backend).strip()
if backend != instance.backend:
|
5c21a485
tangwang
qwen3-reranker-0....
|
83
84
85
86
|
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)
|
daa2690b
tangwang
漏斗参数调优&呈现优化
|
87
|
return instance.backend, cfg.get_backend_config(profile)
|
86d8358b
tangwang
config optimize
|
88
89
|
|
8c8b9d84
tangwang
ES 拉取 coarse_rank...
|
90
|
def get_rerank_base_url(profile: str | None = None) -> str:
|
daa2690b
tangwang
漏斗参数调优&呈现优化
|
91
92
93
94
95
|
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
|
8c8b9d84
tangwang
ES 拉取 coarse_rank...
|
96
|
base = None
|
daa2690b
tangwang
漏斗参数调优&呈现优化
|
97
98
99
100
101
102
103
|
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
|
8c8b9d84
tangwang
ES 拉取 coarse_rank...
|
104
105
|
if not base:
base = provider_cfg.get("service_url") or provider_cfg.get("base_url")
|
26b910bd
tangwang
refactor service ...
|
106
|
if not base:
|
86d8358b
tangwang
config optimize
|
107
|
raise ValueError("Rerank service URL is not configured")
|
daa2690b
tangwang
漏斗参数调优&呈现优化
|
108
109
110
111
|
base = str(base).rstrip("/")
if base.endswith("/rerank"):
base = base[: -len("/rerank")]
return base
|
42e3aea6
tangwang
tidy
|
112
113
|
|
8c8b9d84
tangwang
ES 拉取 coarse_rank...
|
114
|
def get_rerank_service_url(profile: str | None = None) -> str:
|
daa2690b
tangwang
漏斗参数调优&呈现优化
|
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
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("/")
|