Coverage for src/lexigram/ai/di/provider.py: 97%
167 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Intelligence Provider for Lexigram Framework dependency injection.
3This module defines the main provider class that integrates Lexigram AI
4with the Lexigram Framework's dependency injection system.
5"""
7from __future__ import annotations
9from typing import TYPE_CHECKING, Any
11from lexigram.ai.config import AIConfig
12from lexigram.contracts import (
13 CacheBackendProtocol,
14 HealthCheckResult,
15 HealthStatus,
16 ProviderPriority,
17)
18from lexigram.contracts.ai import AIProviderProtocol
19from lexigram.contracts.core.di import (
20 ContainerRegistrarProtocol,
21 ContainerResolverProtocol,
22)
23from lexigram.contracts.data import DatabaseProviderProtocol
24from lexigram.di.provider import Provider
25from lexigram.logging import get_logger
27if TYPE_CHECKING:
28 from lexigram.ai.llm.config import ClientConfig
29 from lexigram.ai.observability.metrics import AIMetrics
30 from lexigram.vector.config import VectorConfig
32logger = get_logger(__name__)
35class AIProvider(Provider, AIProviderProtocol):
36 """Provider for registering Intelligence services with Lexigram DI container.
38 This provider orchestrates sub-providers (LLMProvider, VectorProvider,
39 RAGProvider) and is solely responsible for monitoring,
40 governance, and AIProvider-specific services (RAGCache).
42 Example:
43 >>> from lexigram.app import Application
44 >>> from lexigram.ai import AIModule
45 >>>
46 >>> app = Application()
47 >>> app.add_module(AIModule.configure(...))
48 >>>
49 >>> # LLMClientProtocol is now available for injection
50 >>> @Controller("/chat")
51 >>> class ChatController:
52 ... def __init__(self, llm: LLMClientProtocol):
53 ... self.llm = llm
54 """
56 name = "ai"
57 priority = ProviderPriority.DOMAIN
58 config_key: str | None = "ai"
59 config_model: type | None = AIConfig
60 optional_dependencies: tuple[str, ...] = ("db", "cache")
62 def __init__(
63 self,
64 config: AIConfig | None = None,
65 llm_config: ClientConfig | None = None,
66 vector_config: VectorConfig | None = None,
67 name: str = "ai",
68 ) -> None:
69 """Initialize the Intelligence Provider.
71 Args:
72 config: Initial AI configuration (optional; can be set by orchestrator)
73 llm_config: LLM-specific configuration (overrides config.llm)
74 vector_config: Vector-specific configuration (overrides config.vector)
75 name: Provider name
76 """
77 super().__init__(name=name)
78 self._config_override = config
80 # Configuration overrides
81 self._llm_config_override = llm_config
82 self._vector_config_override = vector_config
84 # Sub-provider references — populated in register()
85 self._llm_sub: Any | None = None # LLMProvider
86 self._vector_sub: Any | None = None # VectorProvider
87 self._rag_sub: Any | None = None # RAGProvider
89 # AIProvider-specific instances — populated in boot()
90 self._rag_cache: Any | None = None
91 self._metrics: AIMetrics | None = None
92 self._governance: Any | None = None
94 # Resolved optional dependencies
95 self._database_provider: DatabaseProviderProtocol | None = None
96 self._cache_backend: CacheBackendProtocol | None = None
98 @property
99 def intelligence_config(self) -> AIConfig:
100 """Get the current AI configuration (from override or container-provided config)."""
101 cfg = self._config_override or self.config or AIConfig()
102 # Apply config overrides
103 if self._llm_config_override:
104 cfg.llm = self._llm_config_override
105 if self._vector_config_override:
106 cfg.vector = self._vector_config_override
107 return cfg
109 @property
110 def database_provider(self) -> DatabaseProviderProtocol | None:
111 """Get the resolved database provider (set during boot)."""
112 return self._database_provider
114 @property
115 def cache_backend(self) -> CacheBackendProtocol | None:
116 """Get the resolved cache backend (set during boot)."""
117 return self._cache_backend
119 async def register(self, container: ContainerRegistrarProtocol) -> None:
120 """Register services with the DI container.
122 Registers monitoring and config singletons directly; governance is
123 registered by the GovernanceProvider discovered via the
124 "lexigram.ai.subsystems" entry point. Delegates LLM, Vector, and
125 RAG service registration to the respective sub-providers.
127 Args:
128 container: The Lexigram DI container
129 """
130 logger.info("Registering Lexigram AI services with container")
132 # Resolve AIConfig from container (set by orchestrator) or use default
133 intelligence_config = self.config or AIConfig()
135 # Apply config overrides if provided
136 if self._llm_config_override:
137 intelligence_config.llm = self._llm_config_override
138 if self._vector_config_override:
139 intelligence_config.vector = self._vector_config_override
141 # Register config first so sub-services can inject it
142 container.singleton(AIConfig, lambda: intelligence_config)
144 # Monitoring — always registered; AIProvider is the observability orchestrator
145 from lexigram.ai.observability.callbacks.manager import CallbackManagerImpl
146 from lexigram.ai.observability.health import AIHealthMonitor
147 from lexigram.ai.observability.metrics import AIMetrics
148 from lexigram.ai.observability.tracing import AITracer
149 from lexigram.contracts.ai.callbacks import CallbackManagerProtocol
151 container.singleton(AIHealthMonitor)
152 container.singleton(AIMetrics)
153 container.singleton("ai_metrics", AIMetrics)
154 container.singleton(AITracer)
155 container.singleton("ai_tracer", AITracer)
156 container.singleton(CallbackManagerProtocol, CallbackManagerImpl)
157 container.singleton("callback_manager", CallbackManagerImpl)
159 # LLM — delegate to LLMProvider
160 if intelligence_config.llm:
161 from lexigram.ai.llm.di.provider import LLMProvider
163 self._llm_sub = LLMProvider(
164 intelligence_config.llm,
165 cache_backend=self._cache_backend,
166 )
167 await self._llm_sub.register(container)
169 # Vector — delegate to VectorProvider
170 if intelligence_config.vector:
171 from lexigram.vector.di.provider import VectorProvider
173 self._vector_sub = VectorProvider(intelligence_config.vector)
174 await self._vector_sub.register(container)
176 # RAG — delegate to RAGProvider
177 if intelligence_config.rag:
178 from lexigram.ai.rag.di.provider import RAGProvider
180 self._rag_sub = RAGProvider(intelligence_config.rag)
181 await self._rag_sub.register(container)
183 # RAG cache — AIProvider-specific (wraps injected CacheBackendProtocol)
184 if self._cache_backend:
185 from lexigram.ai.rag.cache.manager import RAGCache
187 container.singleton(RAGCache, lambda: self._rag_cache)
188 container.singleton("rag_cache", lambda: self._rag_cache)
190 # ------------------------------------------------------------------
191 # Entry-point discovery for additional AI sub-packages.
192 # Packages that declare the "lexigram.ai.subsystems" entry-point
193 # group are loaded here without any hardcoded imports.
194 # The three core subsystems (llm, vector, rag) are intentionally
195 # skipped as they are wired above with explicit config injection.
196 # ------------------------------------------------------------------
197 _SUBSYSTEM_CONFIGS: dict[str, Any] = {
198 "llm": intelligence_config.llm,
199 "vector": intelligence_config.vector,
200 "rag": intelligence_config.rag,
201 "governance": intelligence_config.governance,
202 "observability": intelligence_config.observability,
203 }
204 try:
205 from importlib.metadata import entry_points as _entry_points
207 for _ep in _entry_points(group="lexigram.ai.subsystems"):
208 config_arg = _SUBSYSTEM_CONFIGS.get(_ep.name)
209 _provider_cls = _ep.load()
210 _sub_provider = (
211 _provider_cls(config=config_arg)
212 if config_arg is not None
213 else _provider_cls()
214 )
215 await _sub_provider.register(container)
216 logger.info(
217 "Registered AI subsystem via entry-point",
218 subsystem=_ep.name,
219 provider=_ep.value,
220 )
221 except ImportError:
222 logger.debug(
223 "importlib.metadata unavailable; skipping AI subsystem discovery"
224 )
226 logger.info("Lexigram AI services registered successfully")
228 async def chat(
229 self,
230 messages: list[Any],
231 tools: list[Any] | None = None,
232 **kwargs: Any,
233 ) -> Any:
234 """Chat with optional tool calling. Delegates to LLM sub-provider's client."""
235 if not self._llm_sub or not self._llm_sub._llm_client:
236 raise RuntimeError("LLM client not configured. Cannot perform chat.")
237 return await self._llm_sub._llm_client.complete(messages, tools=tools, **kwargs)
239 async def boot(self, container: ContainerResolverProtocol) -> None:
240 """Start the intelligence provider.
242 Performs async I/O only for AIProvider-specific services: RAGCache.
243 Sub-providers handle their own async initialization during register().
245 Args:
246 container: The DI container
247 """
248 logger.info("Starting Lexigram AI provider")
250 # Resolve optional dependencies from container
251 try:
252 self._database_provider = await container.resolve(DatabaseProviderProtocol)
253 except (ValueError, KeyError, TypeError):
254 logger.debug(
255 "DatabaseProviderProtocol not available; governance persistence disabled"
256 )
258 try:
259 self._cache_backend = await container.resolve(CacheBackendProtocol)
260 except (ValueError, KeyError, TypeError):
261 logger.debug(
262 "CacheBackendProtocol not available; RAG cache and governance caching disabled"
263 )
265 # RAG cache (wraps platform CacheBackendProtocol)
266 if self._cache_backend:
267 from lexigram.ai.rag.cache.manager import RAGCache
269 self._rag_cache = RAGCache(backend=self._cache_backend)
270 logger.info("Initialized RAGCache")
272 logger.info("Lexigram AI provider started successfully")
274 async def shutdown(self) -> None:
275 """Clean up resources on application shutdown."""
276 logger.info("Shutting down Lexigram AI provider")
278 # Shutdown sub-providers
279 for sub in (self._llm_sub, self._vector_sub, self._rag_sub):
280 if sub is not None:
281 try:
282 await sub.shutdown()
283 except (ConnectionError, TimeoutError, OSError, RuntimeError) as exc:
284 logger.warning("Error shutting down sub-provider: %s", exc)
286 # Clear all references
287 self._llm_sub = None
288 self._vector_sub = None
289 self._rag_sub = None
290 self._rag_cache = None
292 logger.info("Lexigram AI provider shutdown complete")
294 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
295 """Check provider health by aggregating sub-provider and local service checks.
297 Returns:
298 Structured HealthCheckResult with component health information
299 """
300 import time
302 start_time = time.perf_counter()
303 details: dict[str, Any] = {"components": {}}
304 overall_status = HealthStatus.HEALTHY
305 errors = []
307 # LLM health — delegate to sub-provider
308 if self._llm_sub is not None:
309 cl_start = time.perf_counter()
310 try:
311 cl_health = await self._llm_sub.health_check(timeout=timeout)
312 if isinstance(cl_health, dict):
313 details["components"]["llm"] = cl_health
314 if cl_health.get("status") not in ("healthy", None):
315 overall_status = HealthStatus.DEGRADED
316 elif hasattr(cl_health, "model_dump"):
317 details["components"]["llm"] = cl_health.model_dump()
318 if (
319 hasattr(cl_health, "status")
320 and cl_health.status != HealthStatus.HEALTHY
321 ):
322 overall_status = HealthStatus.DEGRADED
323 else:
324 details["components"]["llm"] = {"status": "healthy"}
325 except (ConnectionError, TimeoutError, RuntimeError) as e:
326 logger.exception("LLM health check failed")
327 details["components"]["llm"] = {"status": "unhealthy", "error": str(e)}
328 overall_status = HealthStatus.DEGRADED
329 errors.append(f"LLM: {e}")
330 details["components"]["llm"]["latency_ms"] = (
331 time.perf_counter() - cl_start
332 ) * 1000
334 # Vector health — delegate to sub-provider
335 if self._vector_sub is not None:
336 v_start = time.perf_counter()
337 try:
338 v_health = await self._vector_sub.health_check(timeout=timeout)
339 if isinstance(v_health, HealthCheckResult):
340 details["components"]["vector"] = (
341 v_health.model_dump()
342 if hasattr(v_health, "model_dump")
343 else vars(v_health).copy()
344 )
345 if v_health.status != HealthStatus.HEALTHY:
346 overall_status = HealthStatus.DEGRADED
347 elif isinstance(v_health, dict):
348 details["components"]["vector"] = v_health
349 else:
350 details["components"]["vector"] = {"status": "healthy"}
351 except (ConnectionError, TimeoutError, RuntimeError) as e:
352 logger.exception("Vector store health check failed")
353 details["components"]["vector"] = {
354 "status": "unhealthy",
355 "error": str(e),
356 }
357 overall_status = HealthStatus.DEGRADED
358 errors.append(f"Vector: {e}")
359 details["components"]["vector"]["latency_ms"] = (
360 time.perf_counter() - v_start
361 ) * 1000
363 return HealthCheckResult(
364 component="ai",
365 status=overall_status,
366 details=details,
367 error=" | ".join(errors) if errors else None,
368 duration_ms=(time.perf_counter() - start_time) * 1000,
369 )
372__all__ = ["AIProvider"]