e_lib.assistants

20class Assistant(ABC):
21    """Abstract base class for all assistants.
22
23    An assistant implements a single cycle of checks and/or device control.
24    Periodic scheduling is handled by
25    :class:`e_lib.assistants.ScheduledAssistant`.
26
27    Attributes:
28        name: Human-readable assistant name.
29    """
30
31    def __init__(
32        self, name: str = "Ассистент без имени", logger: Optional[Any] = None
33    ) -> None:
34        """Initialize the assistant.
35
36        Args:
37            name: Assistant name (defaults to "Ассистент без имени").
38            logger: Logger for messages; defaults to a standard-library logger
39                for this module. Projects may pass their own logger (e.g. a
40                loguru logger) to keep messages in the project's log sinks.
41        """
42        self.name: str = name
43        self._logger = logger if logger is not None else logging.getLogger(__name__)
44
45    @abstractmethod
46    def run(self) -> None:
47        """Perform one processing cycle.
48
49        Must be overridden in a subclass.
50        """

53class AsyncAssistant(Assistant):
54    """Base class for asynchronous assistants.
55
56    Same contract as :class:`Assistant` (``name``, optional ``logger``), but
57    :meth:`run` is a coroutine: schedule it with ``asyncio.create_task`` or
58    ``await`` it from another coroutine. Not compatible with
59    :class:`~e_lib.assistants.ScheduledAssistant` (threading.Timer) — that
60    wrapper only works with synchronous assistants.
61    """
62
63    @abstractmethod
64    async def run(self) -> None:
65        """Perform one asynchronous processing cycle.
66
67        Must be overridden in a subclass.
68        """

71class Notifier(ABC):
72    """Abstract contract for delivering notifications.
73
74    Messengers, webhooks and other delivery components implement
75    :meth:`send_message` to push text notifications.
76    """
77
78    @abstractmethod
79    def send_message(self, text: str) -> None:
80        """Deliver a notification.
81
82        Args:
83            text: Message text.
84        """

 20class ThresholdMonitorAssistant(Assistant):
 21    """Monitor a single numeric metric against low/high thresholds.
 22
 23    The template reads the metric, rounds it and notifies through a
 24    ``Notifier`` whenever the value leaves the allowed range.  Comparison
 25    rules: ``value >= threshold_high`` triggers the high handler and
 26    ``value < threshold_low`` triggers the low handler.
 27
 28    Attributes:
 29        threshold_low: Lower bound; values below it are reported as "low".
 30        threshold_high: Upper bound; values at/above it are reported as "high".
 31        rounding: Number of decimal places used when rounding the metric.
 32        notifier: Optional notifier used to deliver messages.
 33    """
 34
 35    def __init__(
 36        self,
 37        name: str,
 38        threshold_low: Optional[float] = None,
 39        threshold_high: Optional[float] = None,
 40        rounding: int = 0,
 41        notifier: Optional[Any] = None,
 42        logger: Optional[Any] = None,
 43    ) -> None:
 44        """Initialize the threshold monitor.
 45
 46        Args:
 47            name: Assistant name.
 48            threshold_low: Lower threshold.
 49            threshold_high: Upper threshold.
 50            rounding: Number of decimal places for rounding.
 51            notifier: Notifier used to deliver messages (e.g. a Bot).
 52            logger: Logger for messages (see ``Assistant``).
 53        """
 54        super().__init__(name, logger=logger)
 55        self.threshold_low = threshold_low
 56        self.threshold_high = threshold_high
 57        self.rounding = rounding
 58        self.notifier = notifier
 59
 60    @abstractmethod
 61    def read_metric(self) -> Optional[float]:
 62        """Read the current metric value.
 63
 64        Returns:
 65            The current metric value, or ``None`` when it cannot be read.
 66        """
 67
 68    def _round(self, value: float) -> float:
 69        """Round the metric value according to ``self.rounding``.
 70
 71        Args:
 72            value: Raw metric value.
 73
 74        Returns:
 75            Rounded value.
 76        """
 77        if self.rounding == 0:
 78            return int(value)
 79        return round(value, self.rounding)
 80
 81    def check_value(
 82        self,
 83        value: float,
 84        threshold_low: Optional[float] = None,
 85        threshold_high: Optional[float] = None,
 86    ) -> None:
 87        """Compare a value against thresholds and notify on violations.
 88
 89        Args:
 90            value: Metric value to check.
 91            threshold_low: Optional override of the lower threshold.
 92            threshold_high: Optional override of the upper threshold.
 93        """
 94        rounded = self._round(value)
 95        high = self.threshold_high if threshold_high is None else threshold_high
 96        low = self.threshold_low if threshold_low is None else threshold_low
 97        if high is not None and rounded >= high:
 98            self.handle_high(rounded)
 99        elif low is not None and rounded < low:
100            self.handle_low(rounded)
101
102    def run(self) -> None:
103        """Perform one monitoring cycle.
104
105        Reads the metric and compares it against the thresholds.  If the
106        metric cannot be read the cycle is skipped.
107        """
108        self._logger.debug(f'Запуск цикла контроля для "{self.name}"...')
109        value = self.read_metric()
110        if value is None:
111            self._logger.error(
112                f'Не удалось получить значение метрики, пропускаем цикл "{self.name}"'
113            )
114            return
115        self.check_value(value)
116
117    def handle_high(self, value: float) -> None:
118        """Handle a value at/above the upper threshold.
119
120        Args:
121            value: The rounded metric value.
122        """
123        message = self._format_high(value)
124        self._logger.warning(message)
125        if self.notifier is not None:
126            self.notifier.send_message(message)
127
128    def handle_low(self, value: float) -> None:
129        """Handle a value below the lower threshold.
130
131        Args:
132            value: The rounded metric value.
133        """
134        message = self._format_low(value)
135        self._logger.warning(message)
136        if self.notifier is not None:
137            self.notifier.send_message(message)
138
139    def _format_high(self, value: float) -> str:
140        """Build the "high" notification message.
141
142        Args:
143            value: The rounded metric value.
144
145        Returns:
146            Notification text.
147        """
148        return (
149            f"{self.name}: значение {value} достигло верхнего порога "
150            f"{self.threshold_high}"
151        )
152
153    def _format_low(self, value: float) -> str:
154        """Build the "low" notification message.
155
156        Args:
157            value: The rounded metric value.
158
159        Returns:
160            Notification text.
161        """
162        return (
163            f"{self.name}: значение {value} опустилось ниже порога {self.threshold_low}"
164        )

19class ScheduledAssistant(Assistant):
20    """Run an assistant periodically with a given interval.
21
22    Uses ``threading.Timer`` to schedule repeated runs and supports a clean
23    stop through :meth:`stop`.
24
25    Attributes:
26        assistant: The wrapped assistant.
27        interval: Interval between runs in seconds.
28        timer: Timer for the next run.
29        is_running: Scheduler state flag.
30    """
31
32    def __init__(
33        self, assistant: Assistant, interval: int, logger: Optional[Any] = None
34    ) -> None:
35        """Initialize the scheduler for an assistant.
36
37        Args:
38            assistant: Assistant to run periodically.
39            interval: Interval between runs in seconds.
40            logger: Logger for messages (see ``Assistant``).
41        """
42        super().__init__(assistant.name, logger=logger)
43        self.assistant = assistant
44        self.interval = interval
45        self.timer: Optional[threading.Timer] = None
46        self.is_running: bool = False
47        self._lock: threading.Lock = threading.Lock()
48
49    def run(self) -> None:
50        """Run one assistant cycle and schedule the next one.
51
52        Errors are logged but do not stop the scheduler while it is active.
53        An awaitable result (an ``AsyncAssistant`` passed by mistake) is closed
54        and logged instead of silently producing an unawaited coroutine.
55        """
56        try:
57            result = self.assistant.run()
58            if inspect.isawaitable(result):
59                result.close()
60                self._logger.error(
61                    f'Ассистент "{self.assistant.name}" асинхронный — '
62                    "используйте asyncio, а не ScheduledAssistant"
63                )
64        except Exception as e:
65            self._logger.error(f"Ошибка в контроллере: {e}")
66        finally:
67            if self.is_running:
68                self.timer = threading.Timer(self.interval, self.run)
69                self.timer.start()
70
71    def start(self) -> None:
72        """Start the scheduler.
73
74        A repeated start is ignored.
75        """
76        with self._lock:
77            if self.is_running:
78                self._logger.warning(
79                    f'Планировщик для "{self.assistant.name}" уже запущен'
80                )
81                return
82            self.is_running = True
83        self._logger.info(
84            f'Запуск планировщика для "{self.assistant.name}" '
85            f"с интервалом {format_time(self.interval)}"
86        )
87        self.run()
88
89    def stop(self) -> None:
90        """Stop the scheduler.
91
92        Cancels a pending timer and clears the running flag.
93        """
94        with self._lock:
95            self.is_running = False
96            if self.timer:
97                self.timer.cancel()
98                self.timer = None
99        self._logger.info(f'Планировщик остановлен для "{self.assistant.name}"')

 44class LangfuseCleanerAssistant(Assistant):
 45    """Deletes old Langfuse traces through the public HTTP API.
 46
 47    Uses HTTP Basic Auth and pages through the traces endpoint, collecting the
 48    traces older than ``period`` days and removing them in one bulk request.
 49
 50    Attributes:
 51        description: Short description of the cleaner.
 52        project_slug: Langfuse project identifier.
 53        period: Retention period in days; older traces are deleted.
 54        limit: Page size for each API request.
 55    """
 56
 57    description = "Cleans and optimizes Langfuse files for better performance."
 58
 59    def __init__(
 60        self,
 61        project_slug: str,
 62        period: int = 1,
 63        limit: int = 100,
 64        base_url: Optional[str] = None,
 65        logger: Optional[Any] = None,
 66    ) -> None:
 67        """Initialize the cleaner.
 68
 69        Args:
 70            project_slug: Langfuse project identifier.
 71            period: Retention period in days.
 72            limit: Page size for each API request.
 73            base_url: Langfuse base URL; defaults to the ``LANGFUSE_HOST``
 74                environment variable.
 75            logger: Logger for messages; defaults to a standard-library logger
 76                for this module. Projects may pass their own logger (e.g. a
 77                loguru logger) to keep messages in the project's log sinks.
 78        """
 79        super().__init__(name="Langfuse Cleaner", logger=logger)
 80        self.project_slug = project_slug
 81        self.period = period
 82        self.limit = limit
 83        self._base_url = base_url or os.getenv(
 84            "LANGFUSE_HOST", "http://192.168.0.250:13000"
 85        )
 86
 87    def should_delete_trace(self, trace: Dict[str, Any]) -> bool:
 88        """Return True if the trace is older than the retention period.
 89
 90        Args:
 91            trace: A trace object from the Langfuse API.
 92
 93        Returns:
 94            True when the trace's timestamp is older than ``period`` days.
 95        """
 96        trace_timestamp = trace.get("timestamp")  # '2025-11-18T16:14:28.624Z'
 97
 98        if not trace_timestamp:
 99            return False
100
101        try:
102            trace_date = datetime.fromisoformat(trace_timestamp.replace("Z", "+00:00"))
103        except ValueError as e:
104            self._logger.error(f"Ошибка преобразования даты {trace_timestamp}: {e}")
105            return False
106
107        cutoff_date = datetime.now(timezone.utc) - timedelta(days=self.period)
108        return trace_date < cutoff_date
109
110    def delete_old_traces(self) -> int:
111        """Delete traces older than ``period`` days.
112
113        Pages through the traces endpoint in ascending timestamp order, collects
114        matching trace ids and removes them in a single bulk ``DELETE`` request.
115
116        Returns:
117            The number of traces deleted.
118        """
119        self._logger.debug(f'Запуск цикла удаления старых трейсов "{self.name}"...')
120
121        secret_key = os.getenv("LANGFUSE_SECRET_KEY")
122        public_key = os.getenv("LANGFUSE_PUBLIC_KEY")
123
124        credentials = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
125        headers = {
126            "Authorization": f"Basic {credentials}",
127            "Content-Type": "application/json",
128        }
129
130        cutoff = (
131            (datetime.now(timezone.utc) - timedelta(days=self.period))
132            .isoformat()
133            .replace("+00:00", "Z")
134        )
135
136        deleted_count = 0
137        failed_count = 0
138        page = 1
139        trace_ids = []
140
141        while True:
142            # Get traces (core fields only, oldest first).
143            url = f"{self._base_url}/api/public/traces"
144            params = {
145                "projectSlug": self.project_slug,
146                "limit": self.limit,
147                "page": page,
148                "toTimestamp": cutoff,
149                "orderBy": "timestamp.asc",
150                "fields": "core",
151            }
152
153            response = requests.get(url, headers=headers, params=params)
154
155            if response.status_code != 200:
156                self._logger.error(f"Ошибка получения трейсов: {response.text}")
157                break
158
159            traces = response.json().get("data", [])
160
161            if not traces:
162                break
163
164            for trace in traces:
165                if self.should_delete_trace(trace):
166                    trace_ids.append(trace["id"])
167
168            page += 1
169
170        # Delete the collected traces in one bulk request.
171        if trace_ids:
172            delete_response = requests.delete(
173                url, headers=headers, json={"traceIds": trace_ids}
174            )
175
176            if delete_response.status_code in [200, 204]:
177                try:
178                    deleted_count = delete_response.json().get("count", len(trace_ids))
179                except ValueError:
180                    deleted_count = len(trace_ids)
181                self._logger.debug(f"Удалено {deleted_count} трейсов одним запросом.")
182            else:
183                failed_count = len(trace_ids)
184                self._logger.error(
185                    f"Ошибка массового удаления: {delete_response.status_code} "
186                    f"{delete_response.text}"
187                )
188
189        if failed_count > 0:
190            self._logger.warning(f"Не удалось удалить {failed_count} трейсов.")
191        if deleted_count > 0:
192            self._logger.debug(f"Всего удалено: {deleted_count} трейсов")
193        return deleted_count
194
195    @_observe(name=f"Цикл очистки {__qualname__}", capture_input=False)
196    def run(self) -> Optional[int]:
197        """Run one cleanup cycle.
198
199        The number of deleted traces is returned so that the ``@observe``
200        decorator records it as the Langfuse trace output, while the requested
201        period and project are written into the trace input by
202        :meth:`_record_trace_input`.
203
204        Returns:
205            The number of deleted traces, or ``None`` on error.
206        """
207        return self._run_cycle()
208
209    def _run_cycle(self) -> Optional[int]:
210        """Execute one cleanup cycle without the ``@observe`` wrapper.
211
212        Kept separate from :meth:`run` so the cycle can be tested without
213        Langfuse tracing active.
214        """
215        cutoff = (
216            (datetime.now(timezone.utc) - timedelta(days=self.period))
217            .isoformat()
218            .replace("+00:00", "Z")
219        )
220        try:
221            self._record_trace_input(cutoff)
222            deleted = self.delete_old_traces()
223            self._logger.info(
224                f"Цикл {self.name} завершен: удалено {deleted} старых трейсов."
225            )
226            return deleted
227        except Exception as e:
228            self._logger.error(f'Ошибка в LangfuseCleanerAssistant "{self.name}": {e}')
229            return None
230
231    def _record_trace_input(self, cutoff: str) -> None:
232        """Write the cleanup parameters into the current Langfuse trace input.
233
234        ``@observe`` strips the ``self`` argument, so the requested period,
235        project and cutoff are attached manually.  Failures are logged at
236        DEBUG and never break the cleanup itself.
237
238        Args:
239            cutoff: ISO-8601 cutoff timestamp of the cleanup request.
240        """
241        client = _get_langfuse_client()
242        if client is None:
243            return
244        try:
245            client.update_current_span(
246                input={
247                    "project": self.project_slug,
248                    "retention_days": self.period,
249                    "cutoff": cutoff,
250                }
251            )
252        except Exception as e:
253            self._logger.debug(f"Langfuse: не удалось записать параметры в трейс: {e}")

 25class LLMWarmupAssistant(AsyncAssistant):
 26    """Preload an LLM model by sending a lightweight request.
 27
 28    Sends up to ``retries`` short completion requests (``asyncio.to_thread``
 29    around a blocking OpenAI call) with a ``delay`` between attempts. A second
 30    concurrent :meth:`run` is skipped while one is already in progress.
 31
 32    Attributes:
 33        description: Short description of the assistant.
 34        prompt: Text of the lightweight request.
 35        model: Model identifier used for the request.
 36        retries: Number of attempts before giving up.
 37        delay: Delay in seconds between attempts.
 38        max_tokens: Token limit for the warm-up request.
 39    """
 40
 41    description = (
 42        "Warms up the LLM so the model is preloaded before the first user request."
 43    )
 44
 45    def __init__(
 46        self,
 47        prompt: str = "Привет",
 48        model: str = "openai/gpt-oss-20b",
 49        retries: int = 3,
 50        delay: int = 5,
 51        max_tokens: int = 8,
 52        base_url: Optional[str] = None,
 53        api_key: Optional[str] = None,
 54        logger: Optional[Any] = None,
 55    ) -> None:
 56        """Initialize the warm-up assistant.
 57
 58        Args:
 59            prompt: Text of the lightweight request.
 60            model: Model identifier used for the request.
 61            retries: Number of attempts before giving up.
 62            delay: Delay in seconds between attempts.
 63            max_tokens: Token limit for the warm-up request.
 64            base_url: OpenAI-compatible API base URL; defaults to the
 65                ``LLM_BASE_URL`` environment variable.
 66            api_key: API key; defaults to the ``LLM_API_KEY`` environment
 67                variable.
 68            logger: Logger for messages; defaults to a standard-library logger
 69                for this module. Projects may pass their own logger (e.g. a
 70                loguru logger) to keep messages in the project's log sinks.
 71        """
 72        super().__init__(name="LLM Warmup", logger=logger)
 73        self.prompt = prompt
 74        self.model = model
 75        self.retries = retries
 76        self.delay = delay
 77        self.max_tokens = max_tokens
 78        self._base_url = base_url or os.getenv(
 79            "LLM_BASE_URL", "http://192.168.0.251:1234/v1"
 80        )
 81        self._api_key = api_key or os.getenv("LLM_API_KEY", "")
 82        self._running = False
 83
 84    def _warmup_once(self) -> bool:
 85        """Send one lightweight request; return True when the model answered.
 86
 87        Errors are logged and never raised — the warm-up is best-effort.
 88        """
 89        if OpenAI is None:
 90            self._logger.warning("LLM прогрев пропущен: openai не установлен")
 91            return False
 92        try:
 93            client = OpenAI(base_url=self._base_url, api_key=self._api_key)
 94            client.chat.completions.create(
 95                model=self.model,
 96                messages=[{"role": "user", "content": self.prompt}],
 97                temperature=0.1,
 98                max_tokens=self.max_tokens,
 99            )
100            return True
101        except Exception as e:
102            self._logger.warning(f"LLM прогрев не удался: {e}")
103            return False
104
105    async def run(self) -> bool:
106        """Run one warm-up cycle with retries and deduplication.
107
108        A concurrent :meth:`run` while one is already in progress is skipped
109        and returns ``False`` immediately.
110
111        Returns:
112            True when the model answered, False otherwise (best-effort).
113        """
114        if self._running:
115            self._logger.warning("LLM прогрев уже выполняется — пропускаю")
116            return False
117        self._running = True
118        try:
119            for attempt in range(1, self.retries + 1):
120                if await asyncio.to_thread(self._warmup_once):
121                    self._logger.info("LLM прогрев выполнен: модель загружена")
122                    return True
123                self._logger.warning(
124                    f"LLM прогрев не удался (попытка {attempt}/{self.retries})"
125                )
126                if attempt < self.retries:
127                    await asyncio.sleep(self.delay)
128            return False
129        finally:
130            self._running = False