Залипание стриминга токенов

Для стриминговых интерфейсов важно не «дошёл ли ответ», а «как он капал». Два симптома портят UX, оставаясь невидимыми для обычного мониторинга:

• TTFT спайкнул — время до первого токена вместо 300 мс стало 5 секунд, пользователь смотрит на пустой курсор;
• стрим залип посередине — токены шли, потом провайдер завис на секунды, ответ рвётся кусками.

Итоговый статус при этом 200 OK. Ловим именно тайминги внутри стрима.

Оборачиваем async-стрим и меряем тайминги

Замеряем TTFT и максимальную паузу между токенами; при превышении любого порога — push. COOLDOWN не даёт зафлудить телефон при массовой деградации.

import os, time, asyncio, requests
from openai import AsyncOpenAI   # тот же принцип для anthropic.AsyncAnthropic

NOTIFLY_URL   = os.environ["NOTIFLY_URL"]
NOTIFLY_TOKEN = os.environ["NOTIFLY_TOKEN"]

TTFT_MS   = 2000     # порог времени до первого токена
GAP_MS    = 1500     # порог паузы между соседними токенами
COOLDOWN  = 120      # не чаще одного алёрта в 2 минуты

client = AsyncOpenAI()
_last_alert = 0.0

def notify(title, msg, prio):
    requests.post(f"{NOTIFLY_URL}/message",
                  params={"token": NOTIFLY_TOKEN},
                  json={"title": title, "message": msg, "priority": prio},
                  timeout=5)

def _maybe_alert(title, msg):
    global _last_alert
    now = time.time()
    if now - _last_alert > COOLDOWN:
        _last_alert = now
        notify(title, msg, priority=6)

async def stream_watched(**kwargs):
    """Стримит ответ и на лету следит за TTFT и паузами между токенами."""
    t0 = time.time()
    last = t0
    ttft = None
    max_gap = 0.0

    stream = await client.chat.completions.create(stream=True, **kwargs)
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if not delta:
            continue
        now = time.time()
        if ttft is None:
            ttft = (now - t0) * 1000
            if ttft > TTFT_MS:
                _maybe_alert(
                    f"🐌 TTFT {int(ttft)} мс",
                    f"Время до первого токена > {TTFT_MS} мс — стрим тупит на старте.")
        else:
            max_gap = max(max_gap, (now - last) * 1000)
        last = now
        yield delta

    if max_gap > GAP_MS:
        _maybe_alert(
            f"⏸️ Стрим залипал: пауза {int(max_gap)} мс",
            f"Между токенами была пауза > {GAP_MS} мс — рвётся вывод у пользователя.")

# использование
async def main():
    async for token in stream_watched(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Напиши хайку про латентность"}]):
        print(token, end="", flush=True)

asyncio.run(main())

TTFT проверяется на самом первом токене — реагируем, не дожидаясь конца ответа. Пауза между токенами (max_gap) ловит именно «залипание посередине», которое суммарным временем размазывается и теряется. Оба алёрта — priority=6: это про UX, а не про полный отказ.

Порог должен зависеть от модели

У reasoning-моделей (o-серия, extended thinking) TTFT штатно выше — они «думают» до первого видимого токена. Не ставьте один порог на всех: держите TTFT_MS per-model, иначе завалите себя ложными алёртами на медленных, но исправных моделях.

Серверная сторона: агрегируйте, а не алёртите на каждый

В проде считайте p95 TTFT по окну запросов и алёртите на деградацию медианы — как в рецепте Деградация latency LLM. Пер-запросный алёрт из примера выше хорош для дев-окружения и отладки, а в проде превратится в шум.

Что положить в текст алёрта

• TTFT и максимальную паузу между токенами в мс;
• модель и провайдера (пороги у них разные);
• endpoint/регион, если их несколько;
• p95 TTFT за последнее окно — чтобы отличить разовый спайк от тренда.