-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathconfig.py
More file actions
311 lines (256 loc) · 10.1 KB
/
config.py
File metadata and controls
311 lines (256 loc) · 10.1 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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
"""Configuration management for api_optimizer."""
import os
import json
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
from dataclasses import dataclass, field, asdict
from enum import Enum
try:
import yaml
YAML_AVAILABLE = True
except ImportError:
YAML_AVAILABLE = False
@dataclass
class CacheConfig:
"""Cache configuration."""
enabled: bool = True
db_path: str = "cache.db"
default_ttl: int = 3600
similarity_threshold: float = 0.92
max_entries: int = 10000
semantic_search: bool = True
embedding_provider: str = "simple" # simple, openai
embedding_model: str = "text-embedding-3-small"
@dataclass
class RouterConfig:
"""Router configuration."""
enabled: bool = True
default_model: str = "gpt-4o-mini"
cost_optimization: bool = True
max_cost_per_request: Optional[float] = None
allowed_providers: List[str] = field(default_factory=lambda: ["openai", "anthropic"])
@dataclass
class MonitorConfig:
"""Monitor configuration."""
enabled: bool = True
db_path: str = "usage.db"
daily_limit: Optional[float] = None
weekly_limit: Optional[float] = None
monthly_limit: Optional[float] = None
per_request_limit: Optional[float] = None
alert_threshold: float = 0.8
hard_cap: bool = False
@dataclass
class OptimizerConfig:
"""Optimizer configuration."""
enabled: bool = True
aggressive: bool = False
preserve_formatting: bool = True
auto_optimize: bool = False
min_savings_threshold: float = 0.05
@dataclass
class ProviderConfig:
"""API provider configuration."""
name: str
api_key: str = ""
api_key_env: str = "" # Environment variable name
base_url: Optional[str] = None
default_model: str = ""
timeout: int = 60
max_retries: int = 3
def get_api_key(self) -> str:
"""Get API key from config or environment."""
if self.api_key:
return self.api_key
if self.api_key_env:
return os.environ.get(self.api_key_env, "")
# Default environment variable names
env_map = {
"openai": "OPENAI_API_KEY",
"anthropic": "ANTHROPIC_API_KEY",
}
return os.environ.get(env_map.get(self.name.lower(), ""), "")
@dataclass
class OptimizedClientConfig:
"""Main configuration for OptimizedClient."""
cache: CacheConfig = field(default_factory=CacheConfig)
router: RouterConfig = field(default_factory=RouterConfig)
monitor: MonitorConfig = field(default_factory=MonitorConfig)
optimizer: OptimizerConfig = field(default_factory=OptimizerConfig)
providers: Dict[str, ProviderConfig] = field(default_factory=dict)
# General settings
default_provider: str = "openai"
log_level: str = "INFO"
data_dir: str = ".api_optimizer"
def __post_init__(self):
# Ensure data directory exists
Path(self.data_dir).mkdir(parents=True, exist_ok=True)
# Update paths to use data directory
if not Path(self.cache.db_path).is_absolute():
self.cache.db_path = str(Path(self.data_dir) / self.cache.db_path)
if not Path(self.monitor.db_path).is_absolute():
self.monitor.db_path = str(Path(self.data_dir) / self.monitor.db_path)
# Set default providers if not configured
if not self.providers:
self.providers = {
"openai": ProviderConfig(
name="openai",
api_key_env="OPENAI_API_KEY",
default_model="gpt-4o-mini"
),
"anthropic": ProviderConfig(
name="anthropic",
api_key_env="ANTHROPIC_API_KEY",
default_model="claude-3-5-sonnet-20241022"
)
}
def load_config(config_path: Optional[str] = None) -> OptimizedClientConfig:
"""
Load configuration from file or environment.
Args:
config_path: Path to config file (YAML or JSON)
Returns:
OptimizedClientConfig instance
"""
config_data = {}
# Try to find config file
if config_path:
config_file = Path(config_path)
else:
# Look for config in common locations
search_paths = [
Path("api_optimizer.yaml"),
Path("api_optimizer.yml"),
Path("api_optimizer.json"),
Path(".api_optimizer.yaml"),
Path(".api_optimizer.yml"),
Path(".api_optimizer.json"),
Path.home() / ".config" / "api_optimizer" / "config.yaml",
Path.home() / ".config" / "api_optimizer" / "config.json",
]
config_file = None
for path in search_paths:
if path.exists():
config_file = path
break
# Load from file if found
if config_file and config_file.exists():
config_data = _load_file(config_file)
# Override with environment variables
config_data = _apply_env_overrides(config_data)
# Build config object
return _build_config(config_data)
def _load_file(path: Path) -> Dict[str, Any]:
"""Load configuration from file."""
content = path.read_text()
if path.suffix in (".yaml", ".yml"):
if not YAML_AVAILABLE:
raise ImportError("PyYAML required for YAML config files: pip install pyyaml")
return yaml.safe_load(content) or {}
elif path.suffix == ".json":
return json.loads(content)
else:
# Try YAML first, then JSON
if YAML_AVAILABLE:
try:
return yaml.safe_load(content) or {}
except Exception:
pass
return json.loads(content)
def _apply_env_overrides(config: Dict[str, Any]) -> Dict[str, Any]:
"""Apply environment variable overrides."""
env_mappings = {
"API_OPTIMIZER_CACHE_ENABLED": ("cache", "enabled", _parse_bool),
"API_OPTIMIZER_CACHE_TTL": ("cache", "default_ttl", int),
"API_OPTIMIZER_CACHE_THRESHOLD": ("cache", "similarity_threshold", float),
"API_OPTIMIZER_ROUTER_ENABLED": ("router", "enabled", _parse_bool),
"API_OPTIMIZER_ROUTER_DEFAULT_MODEL": ("router", "default_model", str),
"API_OPTIMIZER_ROUTER_COST_OPT": ("router", "cost_optimization", _parse_bool),
"API_OPTIMIZER_MONITOR_ENABLED": ("monitor", "enabled", _parse_bool),
"API_OPTIMIZER_MONITOR_DAILY_LIMIT": ("monitor", "daily_limit", float),
"API_OPTIMIZER_MONITOR_WEEKLY_LIMIT": ("monitor", "weekly_limit", float),
"API_OPTIMIZER_MONITOR_MONTHLY_LIMIT": ("monitor", "monthly_limit", float),
"API_OPTIMIZER_OPTIMIZER_ENABLED": ("optimizer", "enabled", _parse_bool),
"API_OPTIMIZER_OPTIMIZER_AGGRESSIVE": ("optimizer", "aggressive", _parse_bool),
"API_OPTIMIZER_DEFAULT_PROVIDER": (None, "default_provider", str),
"API_OPTIMIZER_DATA_DIR": (None, "data_dir", str),
"API_OPTIMIZER_LOG_LEVEL": (None, "log_level", str),
}
for env_var, (section, key, converter) in env_mappings.items():
value = os.environ.get(env_var)
if value is not None:
try:
converted = converter(value)
if section:
if section not in config:
config[section] = {}
config[section][key] = converted
else:
config[key] = converted
except (ValueError, TypeError):
pass
return config
def _parse_bool(value: str) -> bool:
"""Parse boolean from string."""
return value.lower() in ("true", "1", "yes", "on")
def _build_config(data: Dict[str, Any]) -> OptimizedClientConfig:
"""Build config object from dictionary."""
# Build sub-configs
cache_data = data.get("cache", {})
cache_config = CacheConfig(**{k: v for k, v in cache_data.items() if k in CacheConfig.__dataclass_fields__})
router_data = data.get("router", {})
router_config = RouterConfig(**{k: v for k, v in router_data.items() if k in RouterConfig.__dataclass_fields__})
monitor_data = data.get("monitor", {})
monitor_config = MonitorConfig(**{k: v for k, v in monitor_data.items() if k in MonitorConfig.__dataclass_fields__})
optimizer_data = data.get("optimizer", {})
optimizer_config = OptimizerConfig(**{k: v for k, v in optimizer_data.items() if k in OptimizerConfig.__dataclass_fields__})
# Build provider configs
providers = {}
providers_data = data.get("providers", {})
for name, pdata in providers_data.items():
if isinstance(pdata, dict):
providers[name] = ProviderConfig(name=name, **{k: v for k, v in pdata.items() if k in ProviderConfig.__dataclass_fields__ and k != "name"})
# Build main config
main_fields = {"default_provider", "log_level", "data_dir"}
main_data = {k: v for k, v in data.items() if k in main_fields}
return OptimizedClientConfig(
cache=cache_config,
router=router_config,
monitor=monitor_config,
optimizer=optimizer_config,
providers=providers,
**main_data
)
def save_config(config: OptimizedClientConfig, path: str):
"""
Save configuration to file.
Args:
config: Configuration to save
path: Output file path
"""
# Convert to dictionary
data = {
"cache": asdict(config.cache),
"router": asdict(config.router),
"monitor": asdict(config.monitor),
"optimizer": asdict(config.optimizer),
"providers": {name: asdict(p) for name, p in config.providers.items()},
"default_provider": config.default_provider,
"log_level": config.log_level,
"data_dir": config.data_dir,
}
path = Path(path)
if path.suffix in (".yaml", ".yml"):
if not YAML_AVAILABLE:
raise ImportError("PyYAML required for YAML config files")
content = yaml.dump(data, default_flow_style=False, sort_keys=False)
else:
content = json.dumps(data, indent=2)
path.write_text(content)
def create_default_config(path: str = "api_optimizer.yaml"):
"""Create a default configuration file."""
config = OptimizedClientConfig()
save_config(config, path)
return path
# Alias for convenience
Config = OptimizedClientConfig