Coverage for src/lexigram/ai/config.py: 73%
81 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Configuration schemas for Lexigram Intelligence.
3This module defines Pydantic models for configuring LLM providers,
4vector stores, and ML components.
6Example:
7 from lexigram.ai.config import AIConfig
9 # From YAML
10 config = AIConfig.from_yaml("application.yaml")
12 # From environment
13 config = AIConfig() # reads LEX_AI__* env vars
14"""
16from __future__ import annotations
18from typing import Any, ClassVar
20from lexigram.ai import constants as const
21from lexigram.config import BaseConfig
22from lexigram.contracts.core.provider import ProviderProtocol
23from lexigram.validation import ConfigDict, Field, model_validator
25# -- Environment Variable Prefixes -------------------------------------------
27ENV_PREFIX: str = const.ENV_PREFIX
28ENV_NESTED_DELIMITER: str = const.ENV_NESTED_DELIMITER
31class _DisabledSubsystem:
32 """Sentinel config returned when an optional AI sub-package is not installed."""
34 enabled = False
36 def __bool__(self) -> bool:
37 return False
39 def __getattr__(self, name: str) -> None:
40 return None
43try:
44 from lexigram.ai.llm.config import ClientConfig
45except ImportError:
46 ClientConfig = None # type: ignore[assignment, misc]
48try:
49 from lexigram.vector.config import VectorConfig
50except ImportError:
51 VectorConfig = None # type: ignore[assignment, misc]
53try:
54 from lexigram.ai.rag.config import RAGConfig
55except ImportError:
56 RAGConfig = None # type: ignore[assignment, misc]
58try:
59 from lexigram.ai.governance.config import GovernanceConfig
60except ImportError:
61 GovernanceConfig = None # type: ignore[assignment, misc]
63try:
64 from lexigram.ai.observability.config import ObservabilityConfig
65except ImportError:
66 ObservabilityConfig = None # type: ignore[assignment, misc]
69class AIConfig(BaseConfig):
70 """Complete configuration for Lexigram Intelligence.
72 Attributes:
73 name: Configuration name (default: "ai")
74 enabled: Whether AI features are enabled
75 llm: LLM configuration
76 vector: Vector store configuration
77 rag: RAG pipeline configuration
78 governance: AI governance configuration
79 observability: Observability configuration
80 subsystems: Dynamic configuration for third-party AI subsystems
81 """
83 config_section: ClassVar[str] = "ai"
85 model_config: ClassVar[ConfigDict] = ConfigDict( # type: ignore[typeddict-unknown-key]
86 env_prefix=ENV_PREFIX,
87 env_nested_delimiter=ENV_NESTED_DELIMITER,
88 extra="forbid",
89 arbitrary_types_allowed=True,
90 )
92 name: str = Field(default="ai", description="Configuration name")
93 enabled: bool = Field(default=True, description="Enable AI features")
94 llm: Any | None = Field(
95 default=None,
96 description="LLM configuration (optional)",
97 )
98 # Vector configuration is optional - avoid requiring a default provider
99 # that pulls in external dependencies (e.g., Chroma) during simple test runs.
100 vector: Any | None = Field(
101 default=None,
102 description="Vector store configuration",
103 )
104 rag: Any | None = Field(
105 default=None,
106 description="RAG pipeline configuration (optional)",
107 )
108 governance: Any = Field(
109 default_factory=lambda: (
110 GovernanceConfig() if GovernanceConfig is not None else _DisabledSubsystem()
111 ),
112 description="AI governance configuration",
113 )
114 observability: Any = Field(
115 default_factory=lambda: (
116 ObservabilityConfig()
117 if ObservabilityConfig is not None
118 else _DisabledSubsystem()
119 ),
120 description="AI observability configuration (tracing and metrics)",
121 )
122 subsystems: dict[str, dict[str, Any]] = Field(
123 default_factory=dict,
124 description=(
125 "Dynamic configuration for third-party AI subsystems discovered "
126 "via entry points. Keys are subsystem names; values are their "
127 "configuration dictionaries."
128 ),
129 )
131 @classmethod
132 def get_provider_class(cls) -> type[ProviderProtocol]:
133 """Return the provider class for this config."""
134 from lexigram.ai import AIProvider
136 return AIProvider
138 @model_validator(mode="after")
139 def validate_production_security(self) -> AIConfig:
140 """Block insecure AI configurations in production."""
141 import os
143 env = os.getenv("LEX_ENV", "development").lower()
144 if env == "production":
145 if self.llm and self.llm.api_key:
146 key_value = self.llm.api_key.get_secret_value()
147 insecure_defaults = ("sk-...", "sk-test", "change-me")
148 if any(
149 key_value.startswith(d) or d in key_value.lower()
150 for d in insecure_defaults
151 ):
152 raise ValueError(
153 "CRITICAL SECURITY ERROR: Insecure LLM API key detected in PRODUCTION.\n"
154 "You MUST set a valid API key via LEX_AI_LLM__API_KEY.",
155 )
156 return self
159def get_subsystem_config(
160 ai_config: AIConfig,
161 subsystem_name: str,
162 default: Any = None,
163) -> Any:
164 """Get configuration for a dynamically discovered AI subsystem.
166 First checks the known top-level fields (``llm``, ``vector``, ``rag``,
167 ``governance``, ``observability``). Falls back to the ``subsystems``
168 dict for third-party subsystems registered via entry points.
170 Args:
171 ai_config: The root AI configuration.
172 subsystem_name: Name of the subsystem (e.g. ``"llm"``, ``"fine_tuning"``).
173 default: Value returned when the subsystem has no configuration.
175 Returns:
176 The subsystem's configuration object or dict, or *default*.
177 """
178 # Built-in subsystems are explicit fields
179 if hasattr(ai_config, subsystem_name):
180 value = getattr(ai_config, subsystem_name)
181 if value is not None:
182 return value
183 # Dynamic subsystems live in the subsystems dict
184 return ai_config.subsystems.get(subsystem_name, default)
187try:
188 # RAG pipeline detail configs — available when lexigram-ai-rag is installed
189 from lexigram.ai.rag.config import (
190 ContextOptimizationConfig,
191 DocumentFormat,
192 IngestionConfig,
193 PipelineConfig,
194 PipelineStageType,
195 PostProcessingConfig,
196 QualityAssuranceConfig,
197 QueryProcessingConfig,
198 RetrievalConfig,
199 RoutingStrategyType,
200 SynthesisConfig,
201 )
202except ImportError:
203 ContextOptimizationConfig = None # type: ignore[assignment, misc]
204 DocumentFormat = None # type: ignore[assignment, misc]
205 IngestionConfig = None # type: ignore[assignment, misc]
206 PipelineConfig = None # type: ignore[assignment, misc]
207 PipelineStageType = None # type: ignore[assignment, misc]
208 PostProcessingConfig = None # type: ignore[assignment, misc]
209 QualityAssuranceConfig = None # type: ignore[assignment, misc]
210 QueryProcessingConfig = None # type: ignore[assignment, misc]
211 RetrievalConfig = None # type: ignore[assignment, misc]
212 RoutingStrategyType = None # type: ignore[assignment, misc]
213 SynthesisConfig = None # type: ignore[assignment, misc]
215__all__ = [
216 "ENV_NESTED_DELIMITER",
217 "ENV_PREFIX",
218 "AIConfig",
219 "get_subsystem_config",
220]