-
Notifications
You must be signed in to change notification settings - Fork 5
Expand file tree
/
Copy pathllm_config_service.py
More file actions
97 lines (82 loc) · 2.91 KB
/
llm_config_service.py
File metadata and controls
97 lines (82 loc) · 2.91 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
"""
API-level service for configuring global LLM behaviour.
Example::
from memiris.api.llm_config_service import LlmConfigService
from memiris.llm.retry_config import RetryConfig
LlmConfigService.configure_retry(
RetryConfig(max_attempts=5, initial_delay=2.0, backoff_factor=2.0)
)
"""
from typing import Optional, Tuple, Type
from memiris.llm.retry_config import (
RetryConfig,
get_retry_config,
set_retry_config,
)
class LlmConfigService:
"""Service for reading and updating the global LLM configuration.
All methods are class-methods so no instantiation is required.
"""
@classmethod
def configure_retry(
cls,
config: RetryConfig,
) -> None:
"""Replace the process-wide LLM retry configuration.
Parameters
----------
config:
The new :class:`~memiris.llm.retry_config.RetryConfig` to apply
to every subsequent LLM call in the process.
"""
set_retry_config(config)
@classmethod
def configure_retry_params(
cls,
*,
max_attempts: Optional[int] = None,
initial_delay: Optional[float] = None,
backoff_factor: Optional[float] = None,
exceptions: Optional[Tuple[Type[BaseException], ...]] = None,
) -> None:
"""Update individual retry parameters while keeping the rest unchanged.
Only the keyword arguments that are explicitly provided will be
changed; all others are taken from the currently active config.
Parameters
----------
max_attempts:
Total number of attempts (1 = no retry).
initial_delay:
Seconds to wait before the first retry.
backoff_factor:
Multiplier applied to the delay after each failed attempt.
exceptions:
Tuple of exception types that should trigger a retry.
"""
current = get_retry_config()
set_retry_config(
RetryConfig(
max_attempts=(
max_attempts if max_attempts is not None else current.max_attempts
),
initial_delay=(
initial_delay
if initial_delay is not None
else current.initial_delay
),
backoff_factor=(
backoff_factor
if backoff_factor is not None
else current.backoff_factor
),
exceptions=exceptions if exceptions is not None else current.exceptions,
)
)
@classmethod
def get_retry_config(cls) -> RetryConfig:
"""Return the currently active process-wide retry configuration."""
return get_retry_config()
@classmethod
def reset_retry_config(cls) -> None:
"""Reset the retry configuration to the built-in defaults."""
set_retry_config(RetryConfig())