« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""
2Health monitoring for Lexigram Intelligence components.
4This module provides health check capabilities for:
5- LLM endpoints (connectivity, latency)
6- Vector stores (connectivity, query performance)
7- Cache services (connectivity, hit rates)
8- Embedding services
10Integrates with lexigram-monitor's health check system.
11"""
13from __future__ import annotations
15from typing import Any
17from lexigram.contracts import (
18 HealthCheckResult,
19 HealthStatus,
20)
21from lexigram.logging import (
22 get_logger,
23)
25logger = get_logger(__name__)
28class AIHealthMonitor:
29 """Health monitoring for intelligence components.
31 Performs health checks on:
32 - LLM endpoints
33 - Vector stores
34 - Cache services
35 - Embedding services
37 Example:
38 >>> from lexigram.logging import get_logger
39 >>> logger = get_logger(__name__)
40 >>> monitor = AIHealthMonitor()
41 >>> # Add health checks
42 >>> monitor.add_llm_check("openai", check_openai_health)
43 >>> monitor.add_vector_check("pgvector", check_pgvector_health)
44 >>> # Run all checks
45 >>> results = await monitor.check_all()
46 >>> if all(r.is_healthy() for r in results.values()):
47 ... logger.info("health_check", status="all_systems_healthy")
48 """
50 def __init__(self) -> None:
51 """Initialize health monitor."""
52 self._llm_checks: dict[str, Any] = {}
53 self._vector_checks: dict[str, Any] = {}
54 self._cache_checks: dict[str, Any] = {}
55 self._embedding_checks: dict[str, Any] = {}
57 def register_check(self, name: str, check: Any) -> None:
58 """Register a named health-check callable.
60 Args:
61 name: Unique check name.
62 check: An async callable returning a health status.
63 """
64 self._llm_checks[name] = check
66 async def check(self) -> Any:
67 """Run health checks and return a ``HealthCheckResult``-like object.
69 Returns:
70 An object with at least a ``status`` attribute.
71 """
72 return await self.check_all()
74 def add_llm_check(self, provider: str, check_func: Any) -> None:
75 """Add LLM health check.
77 Args:
78 provider: LLM provider name
79 check_func: Async function that returns HealthCheckResult
80 """
81 self._llm_checks[provider] = check_func
83 def add_vector_check(self, provider: str, check_func: Any) -> None:
84 """Add vector store health check.
86 Args:
87 provider: Vector store provider name
88 check_func: Async function that returns HealthCheckResult
89 """
90 self._vector_checks[provider] = check_func
92 def add_cache_check(self, service: str, check_func: Any) -> None:
93 """Add cache service health check.
95 Args:
96 service: Cache service name
97 check_func: Async function that returns HealthCheckResult
98 """
99 self._cache_checks[service] = check_func
101 def add_embedding_check(self, model: str, check_func: Any) -> None:
102 """Add embedding service health check.
104 Args:
105 model: Embedding model name
106 check_func: Async function that returns HealthCheckResult
107 """
108 self._embedding_checks[model] = check_func
110 async def check_llm(self, provider: str) -> HealthCheckResult:
111 """Check LLM endpoint health.
113 Args:
114 provider: LLM provider name
116 Returns:
117 Health check result
118 """
119 if provider not in self._llm_checks:
120 return HealthCheckResult(
121 status=HealthStatus.UNKNOWN,
122 component=f"llm.{provider}",
123 message="No health check configured",
124 )
126 try:
127 result = await self._llm_checks[provider]()
128 except (
129 Exception
130 ) as e: # health check must catch all failures and return UNHEALTHY
131 logger.exception("Health check for %s failed", provider)
132 return HealthCheckResult(
133 status=HealthStatus.UNHEALTHY,
134 component=f"llm.{provider}",
135 message=f"Health check failed: {e!s}",
136 error=str(e),
137 details={"error": str(e)},
138 )
139 else:
140 return result
142 async def check_vector(self, provider: str) -> HealthCheckResult:
143 """Check vector store health.
145 Args:
146 provider: Vector store provider name
148 Returns:
149 Health check result
150 """
151 if provider not in self._vector_checks:
152 return HealthCheckResult(
153 status=HealthStatus.UNKNOWN,
154 component=f"vector.{provider}",
155 message="No health check configured",
156 )
158 try:
159 result = await self._vector_checks[provider]()
160 except (
161 Exception
162 ) as e: # health check must catch all failures and return UNHEALTHY
163 logger.exception("Health check for vector store %s failed", provider)
164 return HealthCheckResult(
165 status=HealthStatus.UNHEALTHY,
166 component=f"vector.{provider}",
167 message=f"Health check failed: {e!s}",
168 error=str(e),
169 details={"error": str(e)},
170 )
171 else:
172 return result
174 async def check_cache(self, service: str) -> HealthCheckResult:
175 """Check cache service health.
177 Args:
178 service: Cache service name
180 Returns:
181 Health check result
182 """
183 if service not in self._cache_checks:
184 return HealthCheckResult(
185 status=HealthStatus.UNKNOWN,
186 component=f"cache.{service}",
187 message="No health check configured",
188 )
190 try:
191 result = await self._cache_checks[service]()
192 except (
193 Exception
194 ) as e: # health check must catch all failures and return UNHEALTHY
195 logger.exception("Health check for cache service %s failed", service)
196 return HealthCheckResult(
197 status=HealthStatus.UNHEALTHY,
198 component=f"cache.{service}",
199 message=f"Health check failed: {e!s}",
200 error=str(e),
201 details={"error": str(e)},
202 )
203 else:
204 return result
206 async def check_all(self) -> dict[str, HealthCheckResult]:
207 """Run all health checks.
209 Returns:
210 Dictionary mapping component names to health check results
211 """
212 results = {}
214 # Check all LLM endpoints
215 for provider in self._llm_checks:
216 results[f"llm.{provider}"] = await self.check_llm(provider)
218 # Check all vector stores
219 for provider in self._vector_checks:
220 results[f"vector.{provider}"] = await self.check_vector(provider)
222 # Check all cache services
223 for service in self._cache_checks:
224 results[f"cache.{service}"] = await self.check_cache(service)
226 # Check all embedding services
227 for model in self._embedding_checks:
228 if model in self._embedding_checks:
229 try:
230 results[f"embedding.{model}"] = await self._embedding_checks[
231 model
232 ]()
233 except (
234 Exception
235 ) as e: # health check must catch all failures and return UNHEALTHY
236 logger.exception(
237 "Health check for embedding model %s failed",
238 model,
239 )
240 results[f"embedding.{model}"] = HealthCheckResult(
241 status=HealthStatus.UNHEALTHY,
242 component=f"embedding.{model}",
243 message=f"Health check failed: {e!s}",
244 error=str(e),
245 details={"error": str(e)},
246 )
248 return results
250 async def is_ready(self) -> bool:
251 """Check if all components are ready (healthy or degraded).
253 Returns:
254 True if all components are ready, False otherwise
255 """
256 results = await self.check_all()
257 return all(
258 r.status in (HealthStatus.HEALTHY, HealthStatus.DEGRADED)
259 for r in results.values()
260 )
262 async def is_live(self) -> bool:
263 """Check if service is alive (at least one component healthy).
265 Returns:
266 True if at least one component is healthy, False otherwise
267 """
268 results = await self.check_all()
269 if not results:
270 return True # No checks configured = assume live
272 return any(r.status == HealthStatus.HEALTHY for r in results.values())