core.util¶
Utility modules: performance telemetry, lazy resource loading, system resource management, and benchmarking.
Telemetry¶
core.util.telemetry
¶
Local-first performance telemetry and benchmarking.
Provides a bounded, in-memory metrics collector that records timing
samples, gauges, and system resource snapshots. All data stays local —
no network calls, no external dependencies beyond psutil and GPUtil
(both already used in agent_classes.py).
Key classes¶
MetricSample : Frozen dataclass for a single timing/counter measurement. GaugeValue : Frozen dataclass for a point-in-time gauge (queue depth, %CPU). MetricTimer : Context manager that records elapsed milliseconds. MetricsCollector : Bounded ring buffer with percentile summaries.
Usage¶
from core.telemetry import get_metrics, capture_system_snapshot
m = get_metrics()
m.record("agent.first_token_ms", 120.5, labels={"agent": "axiom"})
m.gauge("queue.depth", 3)
async with m.timer("agent.generation_ms") as t:
await agent.generate(...)
snapshot = capture_system_snapshot()
GaugeValue
dataclass
¶
A point-in-time measurement (queue depth, CPU %, etc.).
Parameters¶
name: Gauge identifier. value: Current numeric value. timestamp: UTC ISO-8601 timestamp. labels: Optional dimension key-value pairs.
MetricSample
dataclass
¶
A single metric observation (timer or counter).
Parameters¶
name:
Metric identifier, e.g. "agent.first_token_ms".
value:
Numeric measurement.
unit:
Human-readable unit ("ms", "bytes", "count").
timestamp:
UTC ISO-8601 timestamp of measurement.
labels:
Optional dimension key-value pairs (e.g. {"agent": "axiom"}).
correlation_id:
Optional trace ID linking related events.
MetricTimer
¶
Context manager that records elapsed time to a MetricsCollector.
Usage¶
async with collector.timer("agent.generation_ms") as t:
await generate(...)
# t.elapsed_ms is available after exit
MetricsCollector
¶
Bounded, in-memory metrics collector with percentile summaries.
Parameters¶
max_samples: Maximum number of samples to retain in the ring buffer.
gauge_count: int
property
¶
Total gauges stored.
sample_count: int
property
¶
Total samples stored.
export_json(path: str | Path) -> None
async
¶
Export all samples and gauges to a JSON file.
gauge(name: str, value: float, labels: dict[str, str] | None = None) -> GaugeValue
¶
Record or update a gauge value (queue depth, CPU %, etc.).
get_gauges(name: str | None = None) -> dict[str, GaugeValue]
¶
Retrieve current gauge values, optionally filtered by name prefix.
get_samples(name: str | None = None, since: datetime | None = None, limit: int = 1000) -> list[MetricSample]
¶
Retrieve samples, optionally filtered by name and time window.
get_summary(window_seconds: float = 60.0) -> dict[str, Any]
¶
Compute per-metric min / p50 / p95 / max / count over a time window.
Returns¶
dict
Mapping of metric name to summary dict with keys:
min, p50, p95, max, count.
record(name: str, value: float, unit: str = 'ms', labels: dict[str, str] | None = None, correlation_id: str = '') -> MetricSample
¶
Record a metric sample (timing, counter, etc.).
reset() -> None
¶
Clear all stored samples and gauges.
capture_system_snapshot() -> dict[str, Any]
¶
Capture a snapshot of system resource usage.
Returns a dict with CPU %, RAM usage (MB), and GPU usage (if CUDA
GPUs are available). Graceful fallbacks: missing GPU returns
None for GPU fields.
Returns¶
dict
Keys: cpu_percent, ram_used_mb, ram_total_mb,
ram_percent, gpu_count, gpu_used_percent,
gpu_memory_used_mb, gpu_memory_total_mb.
get_metrics(max_samples: int = 10000) -> MetricsCollector
¶
Return (or create) the global MetricsCollector singleton.
Follows the lazy-initialization pattern of get_shared_state().
Pass max_samples on first call to configure the ring buffer size.
Subsequent calls ignore the argument.
record_system_snapshot(collector: MetricsCollector) -> dict[str, Any]
¶
Capture a system snapshot and record each field as a gauge.
Returns the snapshot dict.
reset_metrics() -> None
¶
Reset the global metrics singleton (for testing).
Lazy Loader¶
core.util.lazy_loader
¶
Lazy-loading manager for heavy resources.
Defers loading of Qwen ASR, Chatterbox TTS, reranker, and CUDA setup until first use. Each resource tracks a readiness state so callers can check availability without blocking the main loop at startup.
Key concepts¶
ReadyState— enum of lifecycle states (NOT_LOADED, LOADING, LOADED, FAILED).LazyResource— frozen dataclass describing a single resource's state.LazyLoader— orchestrates async loading with timeouts, retry support, and readiness queries.
Examples¶
import asyncio from core.lazy_loader import LazyLoader, ReadyState
async def main(): ... loader = LazyLoader() ... loader.register( ... "tts", ... load_fn=lambda: asyncio.sleep(0.1), # simulate loading ... timeout=30.0, ... ) ... await loader.ensure_loaded("tts") ... assert loader.get_state("tts") == ReadyState.LOADED asyncio.run(main())
LazyLoader
¶
Manages lazy-loading of heavy resources.
Parameters¶
default_timeout: Default timeout in seconds for each resource load (default 120). max_retries: Maximum number of retry attempts for failed loads (default 1).
Examples¶
loader = LazyLoader(default_timeout=60.0) loader.register("tts", load_fn=lambda: some_async_load_tts()) await loader.ensure_loaded("tts") if loader.is_ready("tts"): ... tts_model = loader.get_result("tts")
registered_names: list[str]
property
¶
Return list of registered resource names.
ensure_loaded(name: str) -> Any
async
¶
Ensure the named resource is loaded.
If the resource is already loaded, returns the cached result immediately. If it is loading, waits for completion. If it has not been attempted (or previously failed), initiates the load with timeout and retry support.
Parameters¶
name: The resource identifier.
Returns¶
Any The result returned by the load function.
Raises¶
KeyError If the resource is not registered. TimeoutError If the load exceeds the timeout and all retries are exhausted. RuntimeError If the load function raises an exception after all retries.
get_all_resources() -> list[LazyResource]
¶
Return snapshots for all registered resources.
get_resource(name: str) -> LazyResource
¶
Return a snapshot of the resource's state.
get_result(name: str) -> Any
¶
Return the loaded result (may raise KeyError if not loaded).
get_state(name: str) -> ReadyState
¶
Return the current state of a resource.
get_status_summary() -> dict[str, Any]
¶
Return a summary dict of all resource states.
is_ready(name: str) -> bool
¶
Return True if the resource is loaded and available.
register(name: str, *, load_fn: Callable[..., Awaitable[Any]], timeout: float | None = None) -> None
¶
Register a resource for lazy loading.
Parameters¶
name:
Unique identifier (e.g. "tts", "asr").
load_fn:
Async callable that loads the resource and returns it.
timeout:
Override the default timeout in seconds.
retry(name: str) -> Any
async
¶
Retry loading a failed resource.
This clears the failed state and calls ensure_loaded again.
unload(name: str) -> None
async
¶
Unload a resource (resets to NOT_LOADED state).
Note: This does not free the underlying memory; it merely resets the loader's bookkeeping. The actual resource should be released by the caller if needed.
LazyResource
dataclass
¶
Immutable snapshot of a lazy-loaded resource's state.
Parameters¶
name:
Unique identifier (e.g. "tts", "reranker").
state:
Current readiness state.
loaded_at:
UTC timestamp when the resource finished loading (or None).
error:
Error message if state is FAILED (or "").
load_duration_ms:
Time taken to load the resource in milliseconds (or None).
ReadyState
¶
Bases: Enum
Lifecycle states for a lazy-loaded resource.
FAILED = auto()
class-attribute
instance-attribute
¶
Resource loading failed; can be retried.
LOADED = auto()
class-attribute
instance-attribute
¶
Resource is loaded and ready for use.
LOADING = auto()
class-attribute
instance-attribute
¶
Resource is currently being loaded (in-progress).
NOT_LOADED = auto()
class-attribute
instance-attribute
¶
Resource has not been attempted yet.
Resource Manager¶
core.util.resource_manager
¶
Resource management — GPU/CPU/RAM tracking and degraded-mode heuristics.
Centralizes system resource budgeting so that heavy models (Parakeet-TDT STT,
Qwen3-ASR fallback, Chatterbox TTS, CrossEncoder reranker) load lazily and
unload gracefully when memory pressure mounts. The ResourceManager provides
readiness states for each capability and makes load/unload decisions based on
available GPU memory, CPU load, and RAM utilization.
Components¶
ResourceState— enum of component readiness statesDeviceInfo— frozen snapshot of detected hardware capabilitiesSystemResources— frozen snapshot of current resource usageResourceManager— tracks budgets, makes degraded-mode decisionscreate_resource_manager()— factory with auto-detected hardware
Usage¶
from core.resource_manager import create_resource_manager
mgr = create_resource_manager()
print(mgr.device_info) # cuda_available, gpu_count, ...
print(mgr.can_load_model("tts")) # True/False based on current budget
mgr.mark_loading("tts")
mgr.mark_ready("tts", gpu_memory_mb=2048)
DeviceInfo
dataclass
¶
Detected hardware capabilities.
Parameters¶
cuda_available : Whether CUDA is available (torch.cuda.is_available()). gpu_count : Number of CUDA devices detected. gpu_name : Primary GPU name (empty string if no GPU). cpu_cores : Number of logical CPU cores. total_ram_mb : Total system RAM in megabytes. max_gpu_memory_mb : Total GPU memory across all GPUs (0 if no GPU).
ResourceManager
¶
Tracks system resources and makes load/unload decisions.
Monitors GPU memory, CPU load, RAM utilization, and active model
count. Provides can_load_model() heuristics that account for
available headroom, current load, and degraded-mode policies.
Parameters¶
device_info : Hardware capabilities. Auto-detected if not provided. min_ram_free_mb : Minimum free RAM (MB) required before loading a model. min_gpu_free_mb : Minimum free GPU memory (MB) required before loading a model. max_cpu_percent : CPU utilization threshold above which new loads are deferred. max_active_models : Maximum concurrent loaded models (soft limit, can be exceeded in degraded mode).
can_load_model(name: str) -> tuple[bool, str]
¶
Check if a model can be safely loaded given current resources.
Returns¶
(bool, str)
Tuple of (can_load, reason). If can_load is False,
reason explains the blocking condition.
get_readiness_summary() -> dict[str, Any]
¶
Return a UI-friendly readiness summary.
Returns a dict mapping component names to their state strings, plus system-level resource utilization.
get_resources() -> SystemResources
¶
Capture a snapshot of current system resource usage.
get_state(name: str) -> ResourceState
¶
Return the current state of a named component.
mark_error(name: str) -> bool
¶
Mark a component as errored.
Returns True if the transition is valid.
mark_loading(name: str) -> bool
¶
Transition a component to LOADING.
Returns True if the transition is valid, False otherwise.
mark_ready(name: str, gpu_memory_mb: int = 0, degraded: bool = False) -> bool
¶
Mark a component as fully loaded.
Parameters¶
name :
Component identifier.
gpu_memory_mb :
Estimated GPU memory consumed by this component.
degraded :
If True, mark as DEGRADED instead of READY.
Returns True if the transition is valid.
mark_unavailable(name: str) -> bool
¶
Mark a component as fully unloaded.
Cleans up tracked GPU memory and decrements active model count.
Returns True if the transition is valid.
mark_unloading(name: str) -> bool
¶
Begin unload sequence for a component.
Returns True if the transition is valid.
ResourceState
¶
Bases: Enum
Lifecycle states for resource-heavy components.
Transitions
UNAVAILABLE -> LOADING -> READY -> DEGRADED * -> ERROR (on failure) ERROR -> LOADING (on retry) READY/DEGRADED -> UNLOADING -> UNAVAILABLE (on unload)
SystemResources
dataclass
¶
Current system resource utilization snapshot.
Parameters¶
cpu_percent : Overall CPU utilization (0-100). ram_used_mb : RAM currently in use (MB). ram_total_mb : Total system RAM (MB). ram_percent : RAM utilization percentage (0-100). gpu_memory_used_mb : GPU memory currently allocated (MB, 0 if no GPU). gpu_memory_total_mb : Total GPU memory (MB, 0 if no GPU). active_models : Number of models currently loaded in memory.
create_resource_manager(min_ram_free_mb: int = 2048, min_gpu_free_mb: int = 1024) -> ResourceManager
¶
Create a ResourceManager with auto-detected hardware.
Parameters¶
min_ram_free_mb : Minimum free RAM (MB) before loading new models. min_gpu_free_mb : Minimum free GPU memory (MB) before loading new models.
Benchmarks¶
core.util.benchmark
¶
Benchmark runner — deterministic performance regression detection.
Measures startup time, message latency, SSE dispatch, TTS queue delay,
memory retrieval, tool-call duration, and database write time. Uses
fake components from tests/testkit/ for fully deterministic CI
runs, and optional real backends for local benchmarking.
Components¶
BenchmarkResult— frozen result with timing percentilesBenchmarkConfig— run configuration (iterations, warmup, timeout)BenchmarkRunner— executes registered benchmarks, produces reportscreate_benchmark_runner()— factory with default fake benchmarksrun_benchmarks_cli()— CLI entry point with markdown output
Usage¶
from core.benchmark import create_benchmark_runner
runner = create_benchmark_runner(iterations=10)
results = await runner.run_all()
print(results.to_markdown())
Or from command line:
python -c "from core.benchmark import run_benchmarks_cli; run_benchmarks_cli()"
BenchmarkConfig
dataclass
¶
Configuration for benchmark execution.
Parameters¶
iterations : Number of measurements per benchmark. warmup_iterations : Discarded runs before timing (warms caches, JIT, file handles). timeout_seconds : Per-benchmark timeout (0 = no timeout). label : Optional run label for reports (e.g., ""ci-pr-123"").
BenchmarkResult
dataclass
¶
Result of running a single benchmark.
Parameters¶
name : Benchmark identifier (e.g., ""startup_init"", ""fake_message_latency""). unit : Measurement unit (always ""ms""). iterations : Number of successful timing samples collected. min_ms : Fastest observation. p50_ms : Median latency. p95_ms : 95th percentile latency. max_ms : Slowest observation. metadata : Optional extra context (component name, device info, etc.).
to_dict() -> dict[str, Any]
¶
Serialize to JSON-safe dict.
BenchmarkRunner
¶
Executes a suite of benchmarks and produces structured results.
Benchmarks are registered with register() and executed sequentially.
Each benchmark is an async function that returns a single timing in ms.
The runner handles warmup, iterations, timeout, and percentile computation.
Parameters¶
config : Run configuration. Defaults to 10 iterations, 2 warmup.
results: list[BenchmarkResult]
property
¶
Results from the last run_all() call.
export(path: str | Path, format: str = 'json') -> None
async
¶
register(name: str, fn: BenchmarkFn, metadata: dict[str, Any] | None = None) -> None
¶
Register a benchmark function.
Parameters¶
name : Unique identifier for this benchmark. fn : Async function returning milliseconds. Accepts no required args. metadata : Optional extra context included in the result.
run_all() -> list[BenchmarkResult]
async
¶
Execute all registered benchmarks and return results.
to_json(indent: int = 2) -> str
¶
Serialize all results to JSON string.
to_markdown() -> str
¶
Generate a markdown performance report.
Produces a table with min / p50 / p95 / max for each benchmark, suitable for PR descriptions and release notes.
create_benchmark_runner(iterations: int = 10, warmup_iterations: int = 2, label: str = '') -> BenchmarkRunner
¶
Create a runner with all default fake benchmarks registered.
Parameters¶
iterations : Number of timed measurements per benchmark. warmup_iterations : Discarded runs before timing begins. label : Optional run label included in reports.
run_benchmarks_cli() -> None
¶
Run benchmarks from command line with markdown + JSON output.