Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/config.py: 65%
75 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Configuration schema for the LLM package.
3Defines LLMConfig, the typed configuration object accepted by LLMProvider
4and all LLM client implementations.
5"""
7from __future__ import annotations
9from dataclasses import dataclass
10from pathlib import Path
11from typing import TYPE_CHECKING, Any, ClassVar
13from lexigram.ai.llm.pinning import ModelPinPolicy
14from lexigram.config.base import BaseConfig
15from lexigram.contracts.ai.thinking import ThinkingConfig
16from lexigram.contracts.ai.types import ModelProvider
17from lexigram.domain import DomainModel
18from lexigram.validation import ConfigDict, Field, SecretStr
20if TYPE_CHECKING:
21 from lexigram.ai.llm.pricing.sources import AbstractPricingSource
24@dataclass(init=False)
25class PricingSourceConfig(DomainModel):
26 """A single pricing data source, configurable from YAML.
28 Attributes:
29 type: Source kind. One of ``"litellm"``, ``"openrouter"``, ``"json"``,
30 or ``"static"``.
31 endpoint: API endpoint URL. Used by ``litellm`` (defaults to the
32 LiteLLM model cost map on GitHub) and ``openrouter`` (defaults to
33 ``https://openrouter.ai/api/v1/models``).
34 file_path: Path to a local pricing JSON file (``json`` type only).
35 timeout: HTTP timeout in seconds for API sources.
36 models: Inline static prices (``static`` type only). Maps a model
37 name to ``{prompt_per_1m, completion_per_1m, provider?}``.
39 Example:
40 .. code-block:: yaml
42 ai_llm:
43 pricing:
44 enabled: true
45 sources:
46 - type: openrouter
47 - type: litellm
48 - type: json
49 file_path: pricing/custom.json
50 - type: static
51 models:
52 internal-model:
53 prompt_per_1m: 0.5
54 completion_per_1m: 1.5
55 provider: custom
56 """
58 type: str = Field(
59 ...,
60 description='Source kind: "litellm", "openrouter", "json", or "static".',
61 )
62 endpoint: str | None = Field(
63 default=None,
64 description="API endpoint URL (litellm/openrouter sources).",
65 )
66 file_path: str | None = Field(
67 default=None,
68 description="Path to a pricing JSON file (json source).",
69 )
70 timeout: float = Field(
71 default=10.0,
72 ge=1.0,
73 description="HTTP timeout in seconds for API sources.",
74 )
75 models: dict[str, dict[str, float]] = Field(
76 default_factory=dict,
77 description="Inline prices per model (static source).",
78 )
81@dataclass(init=False)
82class PricingConfig(DomainModel):
83 """Pricing and cost-estimation configuration for the LLM subsystem.
85 When attached to :class:`ClientConfig` (``ai_llm.pricing`` section in
86 YAML), the LLM provider registers a ``CostEstimatorProtocol`` backed by
87 a ``PricingManager`` over the configured sources. Agents wired to the
88 container then get real USD cost estimates per execution.
90 Sources are queried in the configured order; the first source that knows
91 a model wins. When no sources are listed, defaults are used:
92 OpenRouter (freshest prices for OpenAI/Anthropic/Google/Meta/Cohere/
93 DeepSeek/xAI/Qwen) then the LiteLLM model cost map (covers the long
94 tail including Groq and Mistral).
96 Example:
97 .. code-block:: yaml
99 ai_llm:
100 pricing:
101 enabled: true
102 cache_ttl: 43200
103 enable_fuzzy_match: true
104 sources:
105 - type: openrouter
106 - type: litellm
107 - type: json
108 file_path: pricing/private.json
109 - type: static
110 models:
111 my-internal-model:
112 prompt_per_1m: 0.25
113 completion_per_1m: 0.75
114 """
116 enabled: bool = Field(
117 default=True,
118 description="Register pricing manager and cost estimator.",
119 )
120 cache_ttl: int = Field(
121 default=86400,
122 ge=60,
123 description="Pricing cache TTL in seconds (default: 24 hours).",
124 )
125 enable_fuzzy_match: bool = Field(
126 default=True,
127 description="Allow substring matching of model names to prices.",
128 )
129 sources: list[PricingSourceConfig] = Field(
130 default_factory=list,
131 description="Pricing sources in priority order.",
132 )
134 def build_sources(self) -> list[AbstractPricingSource]:
135 """Build pricing source instances from this config.
137 Returns:
138 Configured ``AbstractPricingSource`` instances in priority
139 order. Empty sources list yields the defaults (OpenRouter
140 then LiteLLM).
142 Raises:
143 ValueError: On unknown source type or a ``json`` source
144 without ``file_path``.
145 """
146 from lexigram.ai.llm.pricing.sources import (
147 APIPricingSource,
148 JSONFilePricingSource,
149 OpenRouterPricingSource,
150 StaticPricingSource,
151 )
152 from lexigram.ai.llm.pricing.types import ModelPricing
154 litellm_url = (
155 "https://raw.githubusercontent.com/BerriAI/litellm/main/"
156 "model_prices_and_context_window.json"
157 )
159 if not self.sources:
160 return [
161 OpenRouterPricingSource(),
162 APIPricingSource(litellm_url),
163 ]
165 sources: list[AbstractPricingSource] = []
166 for cfg in self.sources:
167 source_type = cfg.type.strip().lower()
168 if source_type == "litellm":
169 sources.append(
170 APIPricingSource(cfg.endpoint or litellm_url, cfg.timeout)
171 )
172 elif source_type == "openrouter":
173 sources.append(OpenRouterPricingSource(cfg.endpoint, cfg.timeout))
174 elif source_type == "json":
175 if not cfg.file_path:
176 msg = "pricing source of type 'json' requires 'file_path'"
177 raise ValueError(msg)
178 sources.append(JSONFilePricingSource(Path(cfg.file_path)))
179 elif source_type == "static":
180 static: dict[str, ModelPricing] = {}
181 for model_name, prices in cfg.models.items():
182 static[model_name] = ModelPricing(
183 model=model_name,
184 prompt_per_1m=float(prices.get("prompt_per_1m", 0.0)),
185 completion_per_1m=float(prices.get("completion_per_1m", 0.0)),
186 provider=str(prices.get("provider", "custom")),
187 source="static:config",
188 )
189 sources.append(StaticPricingSource(static))
190 else:
191 msg = (
192 f"Unknown pricing source type {cfg.type!r}. "
193 "Supported types: litellm, openrouter, json, static"
194 )
195 raise ValueError(msg)
196 return sources
199@dataclass(init=False)
200class ClientConfig(BaseConfig):
201 """Configuration for LLM clients.
203 Example:
204 >>> config = ClientConfig(
205 ... provider="openai",
206 ... model="gpt-4-turbo",
207 ... api_key="sk-...",
208 ... temperature=0.7,
209 ... max_tokens=2000,
210 ... )
211 """
213 config_section: ClassVar[str] = "ai_llm"
215 model_config: ClassVar[ConfigDict] = ConfigDict(
216 extra="ignore",
217 arbitrary_types_allowed=True,
218 )
220 enabled: bool = Field(default=True, description="Enable the LLM subsystem")
222 provider: ModelProvider = Field(
223 default=ModelProvider.OPENAI,
224 description="LLM provider name.",
225 )
226 model: str = Field(default="gpt-4-turbo", description="Model name or identifier.")
227 model_revision: str | None = Field(
228 default=None,
229 description="Pinned model revision (provider-specific, e.g. date or version string).",
230 )
231 pin_policy: ModelPinPolicy = Field(
232 default=ModelPinPolicy.LATEST,
233 description="Policy for enforcing the model revision pin.",
234 )
235 api_key: SecretStr | None = Field(
236 default=None,
237 description="API key for the chosen provider.",
238 )
239 api_base: str | None = Field(
240 default=None,
241 description="Custom API base URL (for Azure, local, or proxied endpoints).",
242 )
243 temperature: float = Field(
244 default=0.7,
245 ge=0.0,
246 le=2.0,
247 description="Sampling temperature.",
248 )
249 max_tokens: int | None = Field(
250 default=None,
251 ge=1,
252 description="Maximum tokens in response.",
253 )
254 timeout: float = Field(
255 default=60.0,
256 ge=1.0,
257 description="Request timeout in seconds.",
258 )
259 enable_cache: bool = Field(
260 default=False,
261 description="Enable response caching (requires CacheBackendProtocol in container).",
262 )
263 cache_ttl: int = Field(
264 default=3600,
265 description="Cache TTL in seconds.",
266 )
267 thinking: ThinkingConfig | None = Field(
268 default=None,
269 description=(
270 "Thinking/reasoning configuration. ``None`` disables thinking. "
271 "Set to a ``ThinkingConfig`` instance to enable provider-appropriate "
272 "reasoning output (Anthropic extended thinking, Gemini thinking, "
273 "OpenAI reasoning effort, Bedrock Claude reasoning, OpenRouter reasoning)."
274 ),
275 )
276 extra: dict[str, Any] = Field(
277 default_factory=dict,
278 description="Provider-specific extra parameters passed verbatim.",
279 )
280 pricing: PricingConfig | None = Field(
281 default=None,
282 description=(
283 "Pricing source configuration. When set, registers a pricing "
284 "manager and a CostEstimatorProtocol in the container so agents "
285 "get USD cost estimates. See PricingConfig for the YAML schema."
286 ),
287 )
289 def __post_init__(self) -> None:
290 """Coerce provider string to ModelProvider enum and api_key to SecretStr."""
291 if isinstance(self.provider, str):
292 self.provider = ModelProvider(self.provider)
293 api_key: Any = self.api_key
294 if api_key is not None and not isinstance(api_key, SecretStr):
295 self.api_key = SecretStr(api_key)
298__all__ = ["ClientConfig", "PricingConfig", "PricingSourceConfig"]