Skip to main content

Core Module

The core module is the foundation of Chengeta AI — the central orchestrator, key generation, policies, invalidation, metrics, serialization, compression, stampede protection, cache warming, and multi-tenant namespacing.


Components

ComponentModuleDescription
CacheManagerchengeta_ai.core.cache_managerCentral orchestrator — get, set, invalidate, for_tenant()
CacheKeyBuilderchengeta_ai.core.key_buildernamespace:type:sha256[:16] canonical keys
CacheMetricschengeta_ai.core.metricsHit/miss/eviction counters + provider cache savings
Serializerchengeta_ai.core.serializerPluggable encode/decode — PickleSerializer, JsonSerializer
Compressorchengeta_ai.core.compressorOptional compression — GzipCompressor, NoopCompressor
StampedeShieldchengeta_ai.core.stampedePer-key lock — prevents concurrent duplicate LLM calls
RequestConfigchengeta_ai.core.request_configPer-request TTL / threshold / skip_cache overrides
CacheWarmerchengeta_ai.core.warmerBulk warm from query lists or CSV
TTLPolicychengeta_ai.core.policiesGlobal + per-layer TTL configuration
EvictionPolicychengeta_ai.core.policiesLRU / TTL-only strategy
InvalidationEnginechengeta_ai.core.invalidationTag-based bulk eviction
Observabilitychengeta_ai.core.exportersPrometheus + OpenTelemetry exporters
ChengetaSettingschengeta_ai.config.settingsUnified config dataclass + from_env()

Architecture


Quick Example

from chengeta_ai import CacheManager, ChengetaSettings, CacheMetrics

manager = CacheManager.from_settings(ChengetaSettings.from_env())

key = manager.key_builder.build("response", "my prompt")
manager.set(key, b"cached-result", tags=["model:gpt-4o"])
value = manager.get(key)

# Per-tenant scope
tenant = manager.for_tenant("customer-42")

# Metrics
snap = manager.metrics.snapshot()
print(f"Hit rate: {snap['hit_rate']:.0%}")

# Tag invalidation
manager.invalidate("model:gpt-4o")

Next Steps