File size: 1,967 Bytes
3f7b296 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 | """Environment-driven settings for the observability layer.
One small frozen dataclass reads the ``MAL_*`` env vars once at
:func:`src.observability.configure` time. Keeping the parsing here (not scattered
across modules) means every layer sees the same, already-validated knobs.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
_DEFAULT_CAPACITY = 4000
_DEFAULT_TEXT_LIMIT = 4000
def _int_env(name: str, default: int) -> int:
raw = os.getenv(name)
if not raw:
return default
try:
return int(raw)
except ValueError:
return default
@dataclass(frozen=True)
class ObservabilitySettings:
"""Resolved observability configuration.
* ``level`` — root log level (``DEBUG`` surfaces full prompts + memory).
* ``fmt`` — terminal log format: ``text`` (human) or ``json`` (structured).
* ``tracing`` — span sink: ``off`` | ``console`` | ``memory`` | ``both``.
``memory`` (default) feeds the in-app Gradio Telemetry panel with zero
terminal noise; ``console`` also prints spans; ``both`` does each.
* ``store_capacity`` — ring-buffer size for logs/spans kept for the UI.
* ``store_text_limit`` — prompt/memory truncation length in stored snapshots
(the full text still reaches the terminal at ``DEBUG``).
"""
level: str = "INFO"
fmt: str = "text"
tracing: str = "memory"
store_capacity: int = _DEFAULT_CAPACITY
store_text_limit: int = _DEFAULT_TEXT_LIMIT
@classmethod
def from_env(cls) -> "ObservabilitySettings":
return cls(
level=(os.getenv("MAL_LOG_LEVEL") or "INFO").upper(),
fmt=(os.getenv("MAL_LOG_FORMAT") or "text").lower(),
tracing=(os.getenv("MAL_TRACING") or "memory").lower(),
store_capacity=_int_env("MAL_TELEMETRY_BUFFER", _DEFAULT_CAPACITY),
store_text_limit=_int_env("MAL_TELEMETRY_TEXT_LIMIT", _DEFAULT_TEXT_LIMIT),
)
|