Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/pricing/sources.py: 39%

1"""Pricing data sources for LLM models.

3This module provides abstraction for different pricing data sources with a clear

4hierarchy: JSON files → API endpoints → Static fallback.

6Example:

7 >>> from lexigram.ai.llm.pricing_sources import JSONFilePricingSource

8 >>>

9 >>> source = JSONFilePricingSource(Path("pricing.json"))

10 >>> pricing = await source.get_pricing("gpt-4-turbo")

12"""

14from __future__ import annotations

16from abc import ABC, abstractmethod

17import asyncio

18from pathlib import Path

19from typing import Any

21from aiohttp import ClientError as AiohttpClientError

23from lexigram.ai.llm.http.client import ResilientHTTPClient

24from lexigram.ai.llm.pricing.types import ModelPricing

25from lexigram.contracts.exceptions.base import LexigramError

26from lexigram.logging import (

27 get_logger,

28)

29from lexigram.serialization import loads

31logger = get_logger(__name__)

33# Every pricing source is best-effort: an unreachable or failing endpoint must

34# never propagate (a pricing outage must not abort application startup), so

35# sources degrade to empty pricing. ``HttpStatusError`` from

36# ``ResilientHTTPClient.raise_for_status`` and the rich aiohttp connection

37# errors are NOT ``OSError`` subclasses and must be caught explicitly.

38_PRICING_FETCH_ERRORS: tuple[type[BaseException], ...] = (

39 OSError,

40 ValueError,

41 TypeError,

42 LexigramError,

43 AiohttpClientError,

44)

47class AbstractPricingSource(ABC):

48 """Abstract base class for pricing data sources.

50 All pricing sources must implement get_pricing() to return ModelPricing

51 for a given model name, or None if not found.

53 """

55 @abstractmethod

56 async def get_pricing(self, model: str) -> ModelPricing | None:

57 """Get pricing for a specific model.

59 Args:

60 model: Model identifier (e.g., "gpt-4-turbo").

62 Returns:

63 ModelPricing if found, None otherwise.

65 """

67 @abstractmethod

68 async def get_all_pricing(self) -> dict[str, ModelPricing]:

69 """Get all available pricing data.

71 Returns:

72 Dictionary mapping model names to pricing.

74 """

76 @property

77 @abstractmethod

78 def source_name(self) -> str:

79 """Get the name of this pricing source.

81 Returns:

82 Human-readable source name.

84 """

87class JSONFilePricingSource(AbstractPricingSource):

88 """Pricing source from local JSON file.

90 This is the fastest and most reliable source as it doesn't require

91 network calls and works offline.

93 Attributes:

94 file_path: Path to JSON pricing file.

95 cache: In-memory cache of loaded pricing.

97 Example:

98 >>> source = JSONFilePricingSource(Path("custom_pricing.json"))

99 >>> pricing = await source.get_pricing("gpt-4-turbo")

100

101 """

102

103 def __init__(self, file_path: Path):

104 """Initialize JSON file pricing source.

105

106 Args:

107 file_path: Path to JSON file containing pricing data.

108

109 """

110 self.file_path = file_path

111 self._cache: dict[str, ModelPricing] | None = None

112

113 async def _load_cache(self) -> dict[str, ModelPricing]:

114 """Load pricing data from JSON file.

115

116 Returns:

117 Dictionary of model name to pricing.

118

119 """

120 if self._cache is not None:

121 return self._cache

122

123 exists = await asyncio.to_thread(self.file_path.exists)

124 if not exists:

125 logger.warning("JSON pricing file not found: %s", self.file_path)

126 return {}

127

128 try:

129

130 def _read_json() -> Any:

131 with open(self.file_path, "rb") as f:

132 return f.read()

133

134 content = await asyncio.to_thread(_read_json)

135 data = loads(content)

136

137 pricing = {}

138

139 # Load regular models

140 for model_name, model_data in data.get("models", {}).items():

141 pricing[model_name.lower()] = ModelPricing(

142 model=model_name,

143 prompt_per_1m=model_data.get("prompt_per_1m", 1.0),

144 completion_per_1m=model_data.get("completion_per_1m", 2.0),

145 provider=model_data.get("provider", "unknown"),

146 source=f"json:{self.file_path.name}",

147 )

148 except _PRICING_FETCH_ERRORS as e:

149 logger.warning("Failed to load pricing from JSON %s: %s", self.file_path, e)

150 return {}

151 else:

152 self._cache = pricing

153 logger.info(

154 "Loaded pricing for %d models from %s",

155 len(pricing),

156 self.file_path.name,

157 )

158 return pricing

159

160 async def get_pricing(self, model: str) -> ModelPricing | None:

161 """Get pricing for a specific model.

162

163 Args:

164 model: Model identifier.

165

166 Returns:

167 ModelPricing if found, None otherwise.

168

169 """

170 cache = await self._load_cache()

171 return cache.get(model.lower())

172

173 async def get_all_pricing(self) -> dict[str, ModelPricing]:

174 """Get all pricing data.

175

176 Returns:

177 All pricing data from JSON file.

178

179 """

180 return await self._load_cache()

181

182 @property

183 def source_name(self) -> str:

184 """Get source name."""

185 return f"JSON File ({self.file_path.name})"

186

187 def invalidate_cache(self) -> None:

188 """Clear cached pricing data to force reload."""

189 self._cache = None

190

191

192class APIPricingSource(AbstractPricingSource):

193 """Pricing source from HTTP API endpoint.

194

195 Fetches pricing data from a remote API. Useful for getting the latest

196 pricing updates, but requires network connectivity.

197

198 Attributes:

199 endpoint: API endpoint URL.

200 timeout: Request timeout in seconds.

201

202 Example:

203 >>> source = APIPricingSource(

204 ... "https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json"

205 ... )

206 >>> pricing = await source.get_pricing("gpt-4")

207

208 """

209

210 def __init__(self, endpoint: str, timeout: float = 10.0):

211 """Initialize API pricing source.

212

213 Args:

214 endpoint: URL to fetch pricing from.

215 timeout: Request timeout in seconds (default: 10).

216

217 """

218 self.endpoint = endpoint

219 self.timeout = timeout

220 self._cache: dict[str, ModelPricing] | None = None

221

222 async def _fetch_pricing(self) -> dict[str, ModelPricing]:

223 """Fetch pricing from API endpoint.

224

225 Returns:

226 Dictionary of model pricing.

227

228 """

229 if self._cache is not None:

230 return self._cache

231

232 try:

233 async with ResilientHTTPClient(

234 timeout=self.timeout,

235 name="pricing-api",

236 ) as client:

237 response = await client.get(self.endpoint)

238 response.raise_for_status()

239

240 data: Any = response.json

241 if asyncio.iscoroutine(data):

242 data = await data

243 pricing = {}

244

245 for model_name, model_data in (data or {}).items():

246 # Extract pricing (LiteLLM format stores per-token, convert to per-1M)

247 input_cost = model_data.get("input_cost_per_token", 0) * 1_000_000

248 output_cost = model_data.get("output_cost_per_token", 0) * 1_000_000

249

250 if input_cost > 0:

251 # Infer provider from model name

252 provider = self._infer_provider(model_name)

253

254 pricing[model_name.lower()] = ModelPricing(

255 model=model_name,

256 prompt_per_1m=input_cost,

257 completion_per_1m=output_cost,

258 provider=provider,

259 source=f"api:{self.endpoint}",

260 )

261

262 self._cache = pricing

263 logger.info(

264 "Fetched pricing for %d models from %s",

265 len(pricing),

266 self.endpoint,

267 )

268 return pricing

269

270 except _PRICING_FETCH_ERRORS as e:

271 logger.warning("Failed to fetch pricing from %s: %s", self.endpoint, e)

272 return {}

273

274 def _infer_provider(self, model_name: str) -> str:

275 """Infer provider from model name.

276

277 Args:

278 model_name: Model identifier.

279

280 Returns:

281 Provider name.

282

283 """

284 model_lower = model_name.lower()

285 if "gpt" in model_lower or "openai" in model_lower:

286 return "openai"

287 if "claude" in model_lower or "anthropic" in model_lower:

288 return "anthropic"

289 if "gemini" in model_lower:

290 return "google"

291 if "mistral" in model_lower or "mixtral" in model_lower:

292 return "mistral"

293 if "command" in model_lower or "cohere" in model_lower:

294 return "cohere"

295 if "llama" in model_lower and "groq" not in model_lower:

296 return "meta"

297 return "unknown"

298

299 async def get_pricing(self, model: str) -> ModelPricing | None:

300 """Get pricing for a specific model.

301

302 Args:

303 model: Model identifier.

304

305 Returns:

306 ModelPricing if found, None otherwise.

307

308 """

309 cache = await self._fetch_pricing()

310 return cache.get(model.lower())

311

312 async def get_all_pricing(self) -> dict[str, ModelPricing]:

313 """Get all pricing data.

314

315 Returns:

316 All pricing data from API.

317

318 """

319 return await self._fetch_pricing()

320

321 @property

322 def source_name(self) -> str:

323 """Get source name."""

324 return f"API ({self.endpoint})"

325

326 def invalidate_cache(self) -> None:

327 """Clear cached pricing data to force refresh."""

328 self._cache = None

329

330

331class OpenRouterPricingSource(AbstractPricingSource):

332 """Pricing source from the OpenRouter models API.

333

334 Fetches ``https://openrouter.ai/api/v1/models`` (no API key required)

335 and maps each model's per-token USD prices to :class:`ModelPricing`.

336 OpenRouter prices reflect what you actually pay for OpenAI, Anthropic,

337 Google, Meta, Cohere, DeepSeek, xAI, and Qwen models and are updated

338 continuously by OpenRouter.

339

340 Each model is indexed under its full slug (e.g. ``openai/gpt-4o``) and,

341 when unambiguous, under its bare name (e.g. ``gpt-4o``) so callers can

342 look up prices by either form.

343

344 Attributes:

345 endpoint: Models API endpoint URL.

346 timeout: Request timeout in seconds.

347 cache: In-memory cache of fetched pricing.

348

349 Example:

350 >>> source = OpenRouterPricingSource()

351 >>> pricing = await source.get_pricing("gpt-4o")

352

353 """

354

355 def __init__(

356 self,

357 endpoint: str | None = None,

358 timeout: float = 10.0,

359 ):

360 """Initialize OpenRouter pricing source.

361

362 Args:

363 endpoint: Models API URL. Defaults to

364 ``https://openrouter.ai/api/v1/models``.

365 timeout: Request timeout in seconds (default: 10).

366 """

367 self.endpoint = endpoint or "https://openrouter.ai/api/v1/models"

368 self.timeout = timeout

369 self._cache: dict[str, ModelPricing] | None = None

370

371 async def _fetch_pricing(self) -> dict[str, ModelPricing]:

372 """Fetch pricing from the OpenRouter models API.

373

374 Returns:

375 Dictionary of model name to pricing.

376

377 """

378 if self._cache is not None:

379 return self._cache

380

381 try:

382 async with ResilientHTTPClient(

383 timeout=self.timeout,

384 name="openrouter-pricing",

385 ) as client:

386 response = await client.get(self.endpoint)

387 response.raise_for_status()

388

389 data: Any = response.json

390 if asyncio.iscoroutine(data):

391 data = await data

392

393 pricing: dict[str, ModelPricing] = {}

394 for model in (data or {}).get("data", []):

395 model_id = str(model.get("id") or "").strip().lower()

396 if not model_id:

397 continue

398

399 prices = model.get("pricing") or {}

400 prompt_per_token = self._to_float(prices.get("prompt"))

401 completion_per_token = self._to_float(prices.get("completion"))

402 if prompt_per_token is None and completion_per_token is None:

403 continue

404

405 provider = model_id.split("/", 1)[0] or "unknown"

406 entry = ModelPricing(

407 model=model_id,

408 prompt_per_1m=(prompt_per_token or 0.0) * 1_000_000,

409 completion_per_1m=(completion_per_token or 0.0) * 1_000_000,

410 provider=provider,

411 source="api:openrouter",

412 )

413 pricing[model_id] = entry

414

415 # Bare-name alias when unambiguous (e.g. "gpt-4o").

416 parts = model_id.split("/", 1)

417 if len(parts) == 2 and parts[1] not in pricing:

418 pricing[parts[1]] = entry

419

420 self._cache = pricing

421 logger.info(

422 "Fetched pricing for %d models from %s",

423 len(pricing),

424 self.endpoint,

425 )

426 return pricing

427

428 except _PRICING_FETCH_ERRORS as e:

429 logger.warning(

430 "Failed to fetch pricing from %s: %s",

431 self.endpoint,

432 e,

433 )

434 return {}

435

436 @staticmethod

437 def _to_float(value: Any) -> float | None:

438 """Convert a pricing value to float, tolerating None/empty strings.

439

440 Args:

441 value: Raw pricing value (string, number, or None).

442

443 Returns:

444 Float value, or ``None`` when absent or unparseable.

445 """

446 if value is None:

447 return None

448 try:

449 return float(str(value).strip())

450 except (TypeError, ValueError):

451 return None

452

453 async def get_pricing(self, model: str) -> ModelPricing | None:

454 """Get pricing for a specific model.

455

456 Args:

457 model: Model identifier (full slug or bare name).

458

459 Returns:

460 ModelPricing if found, None otherwise.

461 """

462 cache = await self._fetch_pricing()

463 return cache.get(model.lower())

464

465 async def get_all_pricing(self) -> dict[str, ModelPricing]:

466 """Get all pricing data.

467

468 Returns:

469 All pricing data from the OpenRouter API.

470 """

471 return await self._fetch_pricing()

472

473 @property

474 def source_name(self) -> str:

475 """Get source name."""

476 return "OpenRouter API"

477

478 def invalidate_cache(self) -> None:

479 """Clear cached pricing data to force refresh."""

480 self._cache = None

481

482

483class StaticPricingSource(AbstractPricingSource):

484 """Pricing source from static dictionary.

485

486 Hardcoded pricing data as a fallback when other sources are unavailable.

487 Useful for custom internal models or as ultimate fallback.

488

489 Attributes:

490 pricing_map: Dictionary of model name to pricing.

491

492 Example:

493 >>> source = StaticPricingSource({

494 ... "my-model": ModelPricing(

495 ... model="my-model",

496 ... prompt_per_1m=5.0,

497 ... completion_per_1m=10.0,

498 ... provider="custom"

499 ... )

500 ... })

501

502 """

503

504 def __init__(self, pricing_map: dict[str, ModelPricing]):

505 """Initialize static pricing source.

506

507 Args:

508 pricing_map: Dictionary mapping model names to pricing.

509

510 """

511 # Normalize keys to lowercase

512 self.pricing_map = {k.lower(): v for k, v in pricing_map.items()}

513

514 async def get_pricing(self, model: str) -> ModelPricing | None:

515 """Get pricing for a specific model.

516

517 Args:

518 model: Model identifier.

519

520 Returns:

521 ModelPricing if found, None otherwise.

522

523 """

524 return self.pricing_map.get(model.lower())

525

526 async def get_all_pricing(self) -> dict[str, ModelPricing]:

527 """Get all pricing data.

528

529 Returns:

530 All static pricing data.

531

532 """

533 return self.pricing_map.copy()

534

535 @property

536 def source_name(self) -> str:

537 """Get source name."""

538 return f"Static ({len(self.pricing_map)} models)"