Spaces:
Running
Running
| """ | |
| lumen_typing_pace.py — самокалибрующийся темп "печати" при стриминге в Telegram. | |
| Скорость — свойство бэкенда под слагом, а не модели, поэтому замер по факту + EMA. Единица — символы/с. EMA только в памяти. | |
| """ | |
| from __future__ import annotations | |
| # ── границы скорости печати (символов/сек) ── | |
| # Подобраны под ощущение живого набора, владелец может менять прямо здесь. | |
| DEFAULT_CHARS_PER_SEC = 90.0 | |
| MIN_CHARS_PER_SEC = 40.0 | |
| MAX_CHARS_PER_SEC = 260.0 | |
| # Меньше альфа — стабильнее оценка, но медленнее подстройка под смену бэкенда под тем же слагом. | |
| _EMA_ALPHA = 0.3 | |
| # Выше, чем у EMA всего ответа: темп прихода меняется внутри одного стрима. | |
| _ARRIVAL_ALPHA = 0.5 | |
| _speed_ema: dict[str, float] = {} | |
| def speed_key(provider: str, model_id: str) -> str: | |
| """Единый ключ пары (provider, model_id), как в GLOBAL_QUOTA.""" | |
| return f"{provider}:{model_id}" | |
| def get_typing_speed(key: str) -> float: | |
| """Оценка для модели или дефолт без замеров. Результат всегда в границах скорости.""" | |
| return max(MIN_CHARS_PER_SEC, min(MAX_CHARS_PER_SEC, _speed_ema.get(key, DEFAULT_CHARS_PER_SEC))) | |
| def record_observed_speed(key: str, elapsed_sec: float, chars_len: int) -> None: | |
| """Учесть один завершённый стрим, вызывать раз в конце. elapsed_sec — без фазы довывода, иначе EMA поверит своим же задержкам. Наблюдение зажимается до усреднения, иначе один выброс утащит EMA.""" | |
| if elapsed_sec <= 0 or chars_len <= 0: | |
| return | |
| observed = max(MIN_CHARS_PER_SEC, min(MAX_CHARS_PER_SEC, chars_len / elapsed_sec)) | |
| prev = _speed_ema.get(key) | |
| _speed_ema[key] = observed if prev is None else (_EMA_ALPHA * observed + (1 - _EMA_ALPHA) * prev) | |
| def catchup_reveal_steps(remaining_len: int, chars_per_sec: float, tick_interval_sec: float, max_ticks: int) -> list[int]: | |
| """Довывод остатка шагами; хвост добирается форсированно, задержка не больше max_ticks × tick. | |
| Без обращения к часам: в тестах ожидание подменено no-op, цикл на monotonic завис бы.""" | |
| if remaining_len <= 0: | |
| return [] | |
| chars_per_tick = max(1, int(chars_per_sec * tick_interval_sec)) | |
| steps: list[int] = [] | |
| revealed = 0 | |
| while revealed < remaining_len and len(steps) < max_ticks: | |
| revealed = min(remaining_len, revealed + chars_per_tick) | |
| steps.append(revealed) | |
| if steps and steps[-1] < remaining_len: | |
| steps[-1] = remaining_len | |
| return steps | |
| def blend_arrival_speed(prev_ewma: float | None, piece_len: int, dt_sec: float) -> float | None: | |
| """Подмешивает кусок в темп прихода. Нулевая дельта не наблюдение, возвращаем прежнее.""" | |
| if piece_len <= 0 or dt_sec <= 0: | |
| return prev_ewma | |
| inst = max(MIN_CHARS_PER_SEC, min(MAX_CHARS_PER_SEC, piece_len / dt_sec)) | |
| return inst if prev_ewma is None else _ARRIVAL_ALPHA * inst + (1 - _ARRIVAL_ALPHA) * prev_ewma | |
| def display_speed_for(arrival_ewma: float | None, pace_key: str) -> float: | |
| """Скорость показа: измеренный темп прихода, пока он есть; иначе глобальная EMA модели (затравка на первые куски).""" | |
| if arrival_ewma is not None: | |
| return max(MIN_CHARS_PER_SEC, min(MAX_CHARS_PER_SEC, arrival_ewma)) | |
| return get_typing_speed(pace_key) | |