Продакшн: надёжность, стоимость, латентность

Большая часть модуля знакома любому Go-бэкендеру: backoff, context, middleware, circuit breaker, token bucketJava-бэкендеру: backoff, таймауты, декораторы и interceptors, circuit breaker (Resilience4j), token bucket, OpenTelemetry. У LLM есть три особенности, из-за которых привычные приёмы приходится донастраивать.

• Длинные и сильно разбросанные по длительности запросы. Ответ идёт от 300 мс до нескольких минут, и время зависит от числа токенов на выходе. Один общий таймаут на 30 секунд либо оборвёт длинные ответы, либо будет ждать полминуты вместо того, чтобы вовремя переключиться на fallback.
• Лимиты считаются в токенах. Помимо RPM провайдер ограничивает input и output tokens в минуту. Один запрос с документом на 200K токенов расходует квоту сотни обычных.
• Стоимость каждого вызова заметна. Retry storm здесь бьёт не только по доступности, но и по счёту: каждая повторная попытка оплачивается заново.

Ретраи и таймауты

Начинать стоит с классификации ошибок. Ретрай имеет смысл, если следующая попытка может закончиться иначе. На 400 «messages: roles must alternate» повтор всегда даст тот же ответ, и вы просто потратите время пользователя.

Экспоненциальный backoff с jitter

Задержка перед попыткой n равна min(cap, base·2ⁿ). Без jitter все клиенты, получившие ошибку в одну и ту же секунду, придут повторно тоже одновременно. Так возникает thundering herd: пики нагрузки повторяются через 200 мс, 400 мс, 800 мс, и каждый пик снова упирается в лимит. Full jitter (sleep = rand(0, min(cap, base·2ⁿ))) из статьи AWS размазывает повторы по интервалу, и пики исчезают.

Что видно в симуляторе:

• Без jitter повторные попытки приходят волнами, и каждая волна снова больше ёмкости. Success rate упирается в число ретраев, а p99 растёт экспоненциально.
• С jitter та же нагрузка растягивается по времени и помещается в ёмкость. Доля успешных запросов растёт, хотя попыток становится не больше.
• Retry budget ограничивает долю повторов, например 20% сверху к обычному трафику. Когда провайдер лежит целиком (100% ошибок), без бюджета вы увеличиваете нагрузку в 1 + max_retries раз, а с бюджетом только в ~1.2 раза, и ошибка или fallback срабатывают быстро. У бюджета есть цена: при коротком всплеске на здоровом сервисе он отсекает ретраи, которые могли бы пройти. Поэтому бюджет считают по скользящему окну на реальном трафике, а не по одной вспышке.

Идемпотентность

Сам вызов POST /v1/messages побочных эффектов не имеет, поэтому повторять его безопасно, хотя и платно. Опасно повторять шаг выше:

• Агент с инструментами. Если упал вызов модели после того, как create_refund уже выполнился, при повторе всего шага деньги вернутся дважды. Инструментам с побочными эффектами нужны idempotency key (например, hash(conversation_id, tool_use_id)) и дедупликация на стороне исполнителя, как в платёжных API.
• Стриминг. Если пользователь уже видел половину ответа, повтор вернёт другой текст: генерация недетерминирована. UI должен явно перезапустить сообщение, склеивать куски нельзя.
• Batch и фоновые пайплайны. Нужны custom_id и upsert результата, чтобы повторная обработка не создавала дубликаты.

Таймауты и дедлайны

Задача таймаута в том, чтобы вовремя признать попытку неудачной, пока остаётся время на ретрай или fallback. Одного общего значения на LLM-вызов не хватает, нужно несколько:

Главная идея: при стриминге TTFT и полное время ограничиваются раздельно. Без стриминга есть только total. Если его выставить в 60 с, то при зависшем провайдере вы узнаете об этом через минуту. Со стримингом через 8 секунд без первого токена можно отменить попытку и уйти на fallback, а длинная генерация по-прежнему может идти хоть 5 минут.

// Стриминг с раздельными дедлайнами: TTFT, простой между чанками, общий.
var (
	ErrTTFT = errors.New("llm: no first token before deadline")
	ErrIdle = errors.New("llm: stream stalled")
)

func StreamWithDeadlines(ctx context.Context, c anthropic.Client, p anthropic.MessageNewParams,
	ttft, idle, total time.Duration, onText func(string)) error {

	ctx, cancelTotal := context.WithTimeout(ctx, total)
	defer cancelTotal()
	ctx, cancel := context.WithCancelCause(ctx)
	defer cancel(nil)

	var started atomic.Bool
	watchdog := time.AfterFunc(ttft, func() {
		if started.Load() {
			cancel(ErrIdle)
		} else {
			cancel(ErrTTFT)
		}
	})
	defer watchdog.Stop()

	stream := c.Messages.NewStreaming(ctx, p)
	defer stream.Close()
	for stream.Next() {
		watchdog.Reset(idle) // любое событие, включая ping, продлевает жизнь
		ev := stream.Current()
		d, ok := ev.AsAny().(anthropic.ContentBlockDeltaEvent)
		if !ok {
			continue
		}
		if t, ok := d.Delta.AsAny().(anthropic.TextDelta); ok {
			started.Store(true)
			onText(t.Text)
		}
	}
	if err := stream.Err(); err != nil {
		if cause := context.Cause(ctx); cause != nil {
			return cause // ErrTTFT / ErrIdle / DeadlineExceeded: понятно, что именно сработало
		}
		return err
	}
	return nil
}

// Стриминг с раздельными дедлайнами: TTFT, простой между чанками, общий.
package course.m09prod.llmmw;

import com.anthropic.client.AnthropicClient;
import com.anthropic.core.http.StreamResponse;
import com.anthropic.models.messages.MessageCreateParams;
import com.anthropic.models.messages.RawMessageStreamEvent;
import java.time.Duration;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

public final class Streams {
    public static final class TtftException extends TimeoutException {
        public TtftException() { super("llm: no first token before deadline"); }
    }

    public static final class IdleException extends TimeoutException {
        public IdleException() { super("llm: stream stalled"); }
    }

    public static void streamWithDeadlines(AnthropicClient c, MessageCreateParams p,
            Duration ttft, Duration idle, Duration total, Consumer<String> onText) throws TimeoutException {
        var started = new AtomicBoolean();
        var cause = new AtomicReference<TimeoutException>();
        ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(Thread.ofVirtual().factory());
        // createStreaming возвращается после заголовков ответа; до них стрим ограничивает
        // таймаут клиента (AnthropicOkHttpClient.builder().timeout(...)).
        try (StreamResponse<RawMessageStreamEvent> stream = c.messages().createStreaming(p)) {
            // Срабатывание любого таймера закрывает стрим: итерация ниже падает с исключением.
            Consumer<TimeoutException> kill = e -> {
                cause.compareAndSet(null, e);
                stream.close();
            };
            timer.schedule(() -> kill.accept(new TimeoutException("llm: total deadline")),
                    total.toMillis(), TimeUnit.MILLISECONDS);
            var watchdog = new AtomicReference<ScheduledFuture<?>>(
                    timer.schedule(() -> kill.accept(new TtftException()), ttft.toMillis(), TimeUnit.MILLISECONDS));

            stream.stream().forEach(ev -> {
                // Любое событие, включая ping, продлевает жизнь.
                watchdog.getAndSet(timer.schedule(
                        () -> kill.accept(started.get() ? new IdleException() : new TtftException()),
                        idle.toMillis(), TimeUnit.MILLISECONDS)).cancel(false);
                ev.contentBlockDelta()
                        .flatMap(d -> d.delta().text())
                        .ifPresent(t -> {
                            started.set(true);
                            onText.accept(t.text());
                        });
            });
        } catch (RuntimeException e) {
            if (cause.get() != null) {
                throw cause.get(); // TTFT / idle / total: понятно, что именно сработало
            }
            throw e;
        } finally {
            timer.shutdownNow();
        }
        if (cause.get() != null) {
            throw cause.get(); // стрим закрыли таймером между событиями
        }
    }
}

Перед каждым ретраем стоит проверить остаток родительского дедлайна: if dl, ok := ctx.Deadline(); ok && time.Until(dl) < backoff+minAttempt { return err }if (req.timeLeft().map(left -> left.compareTo(backoff.plus(minAttempt)) < 0).orElse(false)) throw e;. Пятисекундный backoff при оставшихся двух секундах бюджета бессмысленен. Лучше сразу вернуть ошибку или быстрый fallback.

Fallback и circuit breaker

Ретраи спасают от кратких сбоев, но не от минутных инцидентов у провайдера. На этот случай нужна цепочка fallback, например: основная модель, модель поменьше у того же провайдера, модель другого провайдера или того же через другое облако (Bedrock, Vertex), self-hosted модель и, наконец, деградированный ответ вроде «сервис перегружен, вот ссылки на FAQ».

Circuit breaker перед каждым звеном нужен, чтобы не тратить на заведомо лежащего провайдера TTFT-таймаут в каждом запросе. После N ошибок подряд breaker переходит в open, и запросы сразу идут к следующему звену. Через cooldown он становится half-open и пропускает один пробный запрос: при успехе возвращается в closed, при ошибке снова в open.

Fallback-модель — не бесплатная копия основной

• Промпты под модель. Промпт, отточенный под Opus 5, на Haiku или open-source модели часто работает хуже: другие привычки форматирования, другое следование инструкциям, другой формат tool calling. Храните промпт на каждое звено, prompts/answer@v12/{opus,haiku,qwen}.md, и адаптируйте запрос в роутере.
• Различия API. У одной модели temperature запрещён (новые Claude), у другой он обязателен для детерминизма. Structured outputs, strict tools, размер контекста (1M против 200K у Haiku 4.5), vision: запрос на 400K токенов fallback на Haiku физически не примет. Роутер должен уметь сказать «это звено не подходит» до отправки.
• Evals на каждое звено. Прогоняйте тот же golden set из модуля 8 по всей цепочке и знайте разницу в качестве, например «B на 4 п.п. хуже по faithfulness». Это продуктовое решение: иногда честный отказ лучше плохого ответа.
• Fallback, которым не пользуются, со временем ломается. Если путь срабатывает раз в квартал, к моменту инцидента там окажется протухший ключ или удалённая модель. Пропускайте через него 1% трафика постоянно или устраивайте регулярные game day.
• Стоимость. Fallback на более дорогую модель (Sonnet → Opus при 529) во время инцидента может утроить счёт. Лимиты расходов действуют и для fallback.

Rate limits и бюджеты токенов

У провайдеров лимиты заданы на организацию или workspace и на модель, в нескольких измерениях сразу:

• RPM — запросов в минуту;
• ITPM — input tokens в минуту. У Anthropic для большинства актуальных моделей cache_read_input_tokens в ITPM не входят, так что prompt caching фактически увеличивает пропускную способность (проверьте для своей модели в доке);
• OTPM — output tokens в минуту. Оценивается по max_tokens в начале запроса и корректируется по факту, поэтому завышенный max_tokens заранее «занимает» квоту.

Anthropic описывает свои лимиты как token bucket, который пополняется непрерывно, а не сбрасывается раз в минуту. Остаток видно в заголовках ответа anthropic-ratelimit-requests-remaining, anthropic-ratelimit-input-tokens-remaining, …-output-tokens-remaining и в retry-after на 429. У OpenAI это RPM/TPM с заголовками x-ratelimit-remaining-*.

Клиентский лимитер, очередь, приоритеты

Упираться в лимит провайдера и потом ретраить 429 — плохая стратегия: вы тратите попытки, увеличиваете латентность и конкурируете со своими же сервисами. Лучше держать свой token bucket чуть ниже провайдерского (90%) в трёх измерениях и ждать на нём:

• golang.org/x/time/rate: rate.NewLimiter(rate.Limit(itpm/60.0), burst) и WaitN(ctx, estTokens). Если дедлайн контекста не дождётся токенов, WaitN сразу вернёт ошибку, и вы получаете load shedding без дополнительного кода.
• Token bucket с резервированием: свой на 30 строк (Limiter.TokenBucket ниже) или Bucket4j. Вызов acquire(estTokens, deadline): если до дедлайна запроса токенов не дождаться, он сразу бросает исключение, и вы получаете load shedding без дополнительного кода. Resilience4j RateLimiter тоже ждёт с таймаутом (acquirePermission(n)), но пополняет разрешения циклами фиксированной длины, а не непрерывно.
• Оценку input tokens даёт /v1/messages/count_tokens или эвристика «символы / 3.5» с запасом. OTPM резервируется по max_tokens.
• Несколько инстансов сервиса делят один лимит организации. Локальный лимитер делит квоту поровну (limit / replicas), а распределённый хранит состояние в Redis (GCRA, скрипт на Lua). Второй честнее, но добавляет сетевой hop и точку отказа.
• Приоритеты. Интерактивный чат важнее фоновой суммаризации. Держите две очереди с раздельными лимитерами или взвешенную очередь. Фоновые задачи лучше отправлять в Batch API: −50% к цене и отдельная ёмкость.
• Fairness между тенантами. Один клиент, запустивший импорт на 10 000 документов, не должен съесть квоту остальных. Решение — per-tenant лимитер перед общим.

Бюджеты токенов

Латентность и кэши

Латентность LLM-вызова раскладывается так: T ≈ очередь + TTFT(prefill ~ input tokens) + output_tokens / скорость генерации. На практике почти всегда доминирует второе слагаемое. 1000 токенов вывода при 70 tok/s занимают 14 с, а prefill 20K токенов — порядка секунды. Отсюда порядок рычагов по силе:

1. Сократить вывод. Структурированный короткий ответ, «без преамбулы», JSON вместо прозы, ограничение max_tokens. Эффект линейный и бесплатный.
2. Стриминг. Полное время не уменьшается, но воспринимаемая латентность сокращается до TTFT. Для чата это главный UX-рычаг.
3. Модель поменьше или ниже effort. Haiku-класс генерирует в разы быстрее и дешевле. Для новых Claude output_config.effort: "low" сокращает thinking. Маршрутизируйте по сложности: простое на маленькую модель, сложное на большую (код и схема — в уроке про шлюз).
4. Параллелизм. Retrieval из нескольких источников, независимые подзадачи, input guardrail параллельно с основным вызовом — это errgroup.параллельно с основным вызовом — это CompletableFuture на virtual threads или StructuredTaskScope.
5. Кэши. Точный и семантический кэш убирают вызов целиком. Prompt caching сокращает TTFT на длинном префиксе (по заявлению Anthropic до ~85% для длинных промптов) и снижает стоимость input до 0.1×.

Бюджет латентности голосового агента — модуль 16 (факультатив).

Кэши: точный, семантический, prompt caching

Это три разных механизма, их часто путают.

Семантический кэш: где он ломается

Идея: если новый вопрос по embedding похож на уже отвеченный (cosine ≥ порога), вернуть сохранённый ответ. Проблема в том, что embedding-модели плохо различают отрицание, числа, даты и сущности. «Как отменить заказ?» и «Как оформить заказ?», «тариф Pro» и «тариф Team», «доставка в Москву» и «в Казань» оказываются рядом в пространстве, а правильные ответы у них разные.

• Порог калибруется, а не угадывается. Соберите 300–500 размеченных пар «один и тот же вопрос / разные вопросы» из реального трафика, постройте precision от порога и выберите порог под допустимую долю ложных попаданий, например ≤ 0.5%. Типичные значения 0.90–0.97, но они сильно зависят от embedding-модели.
• Scope ключа. Тенант, язык, модель, версия промпта и хэш system и tools. Для RAG добавьте права пользователя: иначе ответ, собранный из документов HR, достанется человеку без доступа. Это уже утечка данных, а не просто ошибка кэша.
• Инвалидация. TTL (часы или дни); смена версии промпта или модели меняет scope, и кэш «сбрасывается» сам. Для RAG храните id документов, на которых основан ответ, и удаляйте записи при их изменении.
• Что не кэшировать. Персонализированные ответы, ответы с stop_reason != "end_turn", refusal, диалоги с историей (кэшировать стоит только первый вопрос), всё, что зависит от времени.
• Дополнительная проверка. Для чувствительных доменов попадание можно перепроверить дешёвой моделью: «отвечает ли этот ответ на этот вопрос? да/нет». Вызов Haiku всё равно дешевле и быстрее генерации.

Prompt caching на практике

Коротко (подробно в модуле 2). Порядок: tools → system → messages. Ставьте cache_control на конец стабильной части, всего до 4 breakpoint'ов. Любой байт, изменившийся раньше breakpoint'а, инвалидирует всё после него. Классические ошибки: time.Now() в system prompt, недетерминированный порядок tools (ключи map в Go!)Instant.now() в system prompt, недетерминированный порядок tools (HashMap вместо LinkedHashMap!), id запроса в начале сообщения. Проверяйте кэш метрикой cache_read_input_tokens / (input + cache_read + cache_write) в дашборде. Если она внезапно упала до нуля, значит, кто-то поменял префикс.

Механика у провайдеров разная, и это влияет на дизайн промпта и на то, сломает ли шлюз кэш (по состоянию на октябрь 2026, сверено с документацией):

Практический вывод для шлюза с fallback: префикс, стабильный для одного провайдера, у другого может не дотянуть до минимума или кэшироваться иначе. Метрику cache hit rate считайте по провайдеру и модели, а при переключении трафика (fallback, эксперимент на флаге) закладывайте временное падение попаданий и рост стоимости.

LLM-шлюз: роутинг, провайдеры, build vs buy

Ретраи, fallback, лимиты, бюджеты и кэши из уроков выше вместе и составляют LLM-шлюз (LLM gateway) — единую точку, через которую приложение ходит к моделям. Вопрос не в том, нужен ли он, а в том, где он живёт: библиотекой внутри сервиса (цепочка middleware из последнего урока) или отдельным сервисом-прокси. Полный system design шлюза для компании из 40 команд, с расчётами и режимами отказа, — в модуле 12. Здесь — решения, которые принимает инженер одного продукта.

Правило: один-два сервиса на одном стеке — библиотека. Три и больше команд, требования комплаенса к ключам и аудиту, несколько языков — сервис. На практике часто и то и другое: тонкая клиентская библиотека (таймауты, трейсинг, идемпотентность) поверх шлюза (ключи, квоты, роутинг, аудит).

Роутинг по стоимости и сложности

Haiku 4.5 стоит $1 / $5 за 1M токенов, Opus 5 — $5 / $25: разница в 5 раз при той же длине запроса. Если 60% трафика простые вопросы, которые маленькая модель решает так же хорошо, перевод их на Haiku снижает счёт почти вдвое: 0.4 + 0.6 × 0.2 = 0.52 от исходного. Вопрос в том, как отличить простое от сложного до ответа и что делать, когда классификатор ошибся.

• Эвристика — длина, ключевые слова («сравни», «почему», код), число вопросов, глубина диалога. Бесплатно и за микросекунды, но шумно. Хороша как первый фильтр.
• Дешёвая модель как классификатор — Haiku отвечает «simple / complex». +200–400 мс и доли цента. Разумно звать её только в серой зоне эвристики.
• Обученный классификатор — эмбеддинг запроса и логистическая регрессия на размеченных логах (разметка: «ответ маленькой модели прошёл judge или нет»). Миллисекунды и лучшая точность, но нужен датасет и переобучение при дрейфе.
• Каскад с откатом — сначала дешёвая модель, проверка ответа (схема JSON, бизнес-правила, stop_reason, при необходимости быстрый judge), при провале — эскалация на сильную. Ловит ошибки классификатора, но платите дважды и дольше ждёте именно на трудных запросах.

Что видно на схеме: с обученным классификатором или с эскалацией можно отправить на Haiku больше половины трафика, сэкономить около 40% и потерять меньше пункта качества; шумная эвристика при том же пороге теряет больше. Правее порога кривая резко падает — это сложные запросы, ошибочно принятые за простые. Лучший классификатор сдвигает всю кривую к идеальной. Эскалация поднимает качество на дешёвом конце, но съедает часть экономии.

package llmmw

import (
	"context"
	"expvar"
	"strings"
	"unicode/utf8"

	"github.com/anthropics/anthropic-sdk-go"
)

// Classifier оценивает сложность запроса в [0, 1].
type Classifier func(ctx context.Context, req Request) (float64, error)

// Heuristic — бесплатный классификатор на признаках текста. Веса и слова подбираются
// на размеченной выборке из логов, а не на глаз.
// ponytail: словарь признаков; когда его станет мало — ModelClassifier или свой маленький классификатор.
func Heuristic(_ context.Context, req Request) (float64, error) {
	q := strings.ToLower(lastUserText(req))
	score := 0.0
	switch n := utf8.RuneCountInString(q); {
	case n > 1500:
		score += 0.4
	case n > 400:
		score += 0.2
	}
	for _, w := range []string{"почему", "сравни", "проанализируй", "спроектируй", "оптимизируй", "```"} {
		if strings.Contains(q, w) {
			score += 0.3
		}
	}
	if strings.Count(q, "?") > 1 {
		score += 0.15 // несколько вопросов в одном
	}
	if len(req.Messages) > 6 {
		score += 0.15 // длинный диалог: контекст запутаннее
	}
	return min(score, 1), nil
}

// ModelClassifier спрашивает дешёвую модель «simple или complex». Это +200–400 мс и доли цента,
// поэтому зовите её только в «серой зоне» эвристики, а не на каждый запрос.
func ModelClassifier(cheap LLM) Classifier {
	return func(ctx context.Context, req Request) (float64, error) {
		resp, err := cheap.Complete(ctx, Request{
			Tenant: req.Tenant, Model: "claude-haiku-4-5", MaxTokens: 5,
			System:   "Оцени, нужна ли для ответа сильная модель. Ответь одним словом: simple или complex.",
			Messages: []anthropic.MessageParam{anthropic.NewUserMessage(anthropic.NewTextBlock(lastUserText(req)))},
		})
		if err != nil {
			return 0, err
		}
		if strings.Contains(strings.ToLower(resp.Text), "complex") {
			return 1, nil
		}
		return 0, nil
	}
}

// routed — счётчики решений роутера; expvar отдаёт их на /debug/vars, в проде — та же метрика в OTel.
var routed = expvar.NewMap("llm_router")

// Router отправляет простое на дешёвую цепочку, сложное — на сильную.
// Cheap и Strong — обычные LLM: как правило, каждая — свой Fallback[...] со своими моделями.
type Router struct {
	Cheap, Strong LLM
	Classify      Classifier
	Threshold     float64              // score ≥ Threshold → Strong; значение приходит из фича-флага
	Accept        func(*Response) bool // проверка ответа дешёвой модели; false → эскалация
}

func (r *Router) Complete(ctx context.Context, req Request) (*Response, error) {
	score, err := r.Classify(ctx, req)
	if err != nil {
		routed.Add("classifier_error", 1)
		score = 1 // не знаем — платим за качество, а не экономим вслепую
	}
	if score >= r.Threshold {
		routed.Add("strong", 1)
		return r.Strong.Complete(ctx, req)
	}
	routed.Add("cheap", 1)
	resp, err := r.Cheap.Complete(ctx, req)
	if err != nil {
		return nil, err // доступность — забота Fallback внутри цепочки; роутер отвечает за качество
	}
	if r.Accept != nil && !r.Accept(resp) {
		// Откат на сильную модель: платим за оба вызова. Доля эскалаций — главная метрика роутера:
		// если она растёт, порог пора поднимать (или дешёвая модель деградировала).
		routed.Add("escalated", 1)
		return r.Strong.Complete(ctx, req)
	}
	return resp, nil
}

package course.m09prod.llmmw;

import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Predicate;

/**
 * Отправляет простое на дешёвую цепочку, сложное — на сильную. cheap и strong — обычные LLM:
 * как правило, каждая — свой Fallback со своими моделями. threshold приходит из фича-флага:
 * score ≥ threshold → strong. accept — проверка ответа дешёвой модели; false → эскалация (null — без проверки).
 */
public record Router(LLM cheap, LLM strong, Classifier classify, double threshold,
                     Predicate<Response> accept) implements LLM {

    /** Оценивает сложность запроса в [0, 1]. */
    @FunctionalInterface
    public interface Classifier {
        double score(Request req) throws Exception;
    }

    private static final List<String> HARD = List.of("почему", "сравни", "проанализируй", "спроектируй", "оптимизируй", "```");

    /**
     * Бесплатный классификатор на признаках текста. Веса и слова подбираются
     * на размеченной выборке из логов, а не на глаз.
     */
    // ponytail: словарь признаков; когда его станет мало — modelClassifier или свой маленький классификатор.
    public static double heuristic(Request req) {
        String q = req.lastUserText().toLowerCase(Locale.ROOT);
        double score = 0;
        int n = q.codePointCount(0, q.length());
        if (n > 1500) {
            score += 0.4;
        } else if (n > 400) {
            score += 0.2;
        }
        for (String w : HARD) {
            if (q.contains(w)) {
                score += 0.3;
            }
        }
        if (q.chars().filter(c -> c == '?').count() > 1) {
            score += 0.15; // несколько вопросов в одном
        }
        if (req.messages().size() > 6) {
            score += 0.15; // длинный диалог: контекст запутаннее
        }
        return Math.min(score, 1);
    }

    /**
     * Спрашивает дешёвую модель «simple или complex». Это +200–400 мс и доли цента,
     * поэтому зовите её только в «серой зоне» эвристики, а не на каждый запрос.
     */
    public static Classifier modelClassifier(LLM cheap) {
        return req -> {
            Response resp = cheap.complete(new Request(req.tenant(), "claude-haiku-4-5",
                    "Оцени, нужна ли для ответа сильная модель. Ответь одним словом: simple или complex.",
                    List.of(Msg.user(req.lastUserText())), 5, 0, req.deadline()));
            return resp.text().toLowerCase(Locale.ROOT).contains("complex") ? 1 : 0;
        };
    }

    /** Счётчики решений роутера; в проде — Micrometer Counter или та же метрика в OTel. */
    private static final Map<String, LongAdder> ROUTED = new ConcurrentHashMap<>();

    public static long routed(String decision) {
        LongAdder a = ROUTED.get(decision);
        return a == null ? 0 : a.sum();
    }

    private static void count(String decision) {
        ROUTED.computeIfAbsent(decision, k -> new LongAdder()).increment();
    }

    @Override
    public Response complete(Request req) throws Exception {
        double score;
        try {
            score = classify.score(req);
        } catch (Exception e) {
            count("classifier_error");
            score = 1; // не знаем — платим за качество, а не экономим вслепую
        }
        if (score >= threshold) {
            count("strong");
            return strong.complete(req);
        }
        count("cheap");
        // Доступность — забота Fallback внутри цепочки; роутер отвечает за качество.
        Response resp = cheap.complete(req);
        if (accept != null && !accept.test(resp)) {
            // Откат на сильную модель: платим за оба вызова. Доля эскалаций — главная метрика роутера:
            // если она растёт, порог пора поднимать (или дешёвая модель деградировала).
            count("escalated");
            return strong.complete(req);
        }
        return resp;
    }
}

• Метрики роутера. Доля трафика на дешёвую модель, доля эскалаций, качество по маршрутам (онлайн-judge на выборке, см. урок про дрейф), стоимость на запрос. Рост эскалаций — сигнал, что поменялся трафик или деградировала маленькая модель.
• Откат. Threshold приходит из фича-флага (урок ниже): поднять порог до 1.0 — значит мгновенно вернуть весь трафик на сильную модель без деплоя.
• Где в цепочке. Traced → SemCache → Metered → Router → Fallback[…]: у каждой из двух ветвей роутера своя цепочка fallback. Кэш стоит до роутера, потому что попадание не требует ни классификации, ни модели.
• Качество проверяется на evals, а не на вере. Перед включением прогоните golden set из модуля 8 через обе модели: роутинг оправдан, только если на «простых» запросах разница в пределах доверительного интервала.

Fallback на другого провайдера: адаптер

Цепочка fallback из урока выше умеет переключаться между звеньями, но второму провайдеру нужен адаптер под интерфейс LLM. Главное в нём не HTTP-вызов, а нормализация. Ошибки надо привести к той же классификации, иначе Retryable не узнаетиначе Retry.retryable не узнает 503 чужого провайдера: он не будет ретраиться и не откроет breaker. stop_reason тоже переводится в общий словарь, иначе семантический кэш закэширует обрезанный ответ. Тот же адаптер подходит для self-hosted vLLM или Ollama: у них OpenAI-совместимый API.

package llmmw

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"time"
)

// HTTPError — ошибка адаптера в той же классификации, что и ошибки Anthropic SDK:
// по ней работают Retryable, breaker и метрика error.type.
type HTTPError struct {
	StatusCode int
	RetryAfter time.Duration
	Body       string
}

func (e *HTTPError) Error() string { return fmt.Sprintf("http %d: %s", e.StatusCode, e.Body) }

// OpenAICompat — звено под второго провайдера или self-hosted (vLLM, Ollama)
// через OpenAI-совместимый Chat Completions. Только текст, без tools: для fallback этого хватает.
func OpenAICompat(hc *http.Client, baseURL, apiKey string) LLM {
	type msg struct {
		Role    string `json:"role"`
		Content string `json:"content"`
	}
	return Func(func(ctx context.Context, req Request) (*Response, error) {
		msgs := []msg{{Role: "system", Content: req.System}}
		for _, m := range req.Messages {
			msgs = append(msgs, msg{Role: string(m.Role), Content: textOf(m)})
		}
		body, err := json.Marshal(map[string]any{"model": req.Model, "messages": msgs, "max_tokens": req.MaxTokens})
		if err != nil {
			return nil, err
		}
		hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/chat/completions", bytes.NewReader(body))
		if err != nil {
			return nil, err
		}
		hreq.Header.Set("Content-Type", "application/json")
		hreq.Header.Set("Authorization", "Bearer "+apiKey)
		resp, err := hc.Do(hreq)
		if err != nil {
			return nil, err // net.Error: Retryable разберётся
		}
		defer resp.Body.Close()
		if resp.StatusCode != http.StatusOK {
			b, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
			sec, _ := strconv.Atoi(resp.Header.Get("retry-after"))
			return nil, &HTTPError{StatusCode: resp.StatusCode, RetryAfter: time.Duration(sec) * time.Second, Body: string(b)}
		}
		var out struct {
			Model   string `json:"model"`
			Choices []struct {
				Message      msg    `json:"message"`
				FinishReason string `json:"finish_reason"`
			} `json:"choices"`
			Usage struct {
				PromptTokens     int64 `json:"prompt_tokens"`
				CompletionTokens int64 `json:"completion_tokens"`
			} `json:"usage"`
		}
		if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
			return nil, err
		}
		if len(out.Choices) == 0 {
			return nil, &HTTPError{StatusCode: http.StatusBadGateway, Body: "no choices"}
		}
		c := out.Choices[0]
		// stop_reason приводим к словарю Anthropic: остальная цепочка (кэш, метрики) знает только его.
		stop := map[string]string{"stop": "end_turn", "length": "max_tokens", "content_filter": "refusal"}[c.FinishReason]
		return &Response{Text: c.Message.Content, Model: out.Model, StopReason: stop,
			InputTokens: out.Usage.PromptTokens, OutputTokens: out.Usage.CompletionTokens}, nil
	})
}

package course.m09prod.llmmw;

import com.fasterxml.jackson.annotation.JsonProperty;
import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeoutException;

/**
 * Звено под второго провайдера или self-hosted (vLLM, Ollama) через OpenAI-совместимый
 * Chat Completions. Только текст, без tools: для fallback этого хватает.
 */
public record OpenAiCompat(HttpClient http, String baseUrl, String apiKey) implements LLM {

    /**
     * Ошибка адаптера в той же классификации, что и ошибки Anthropic SDK:
     * по ней работают retryable, breaker и метрика error.type.
     */
    public static final class HttpError extends IOException {
        private final int statusCode;
        private final Duration retryAfter;

        public HttpError(int statusCode, Duration retryAfter, String body) {
            super("http " + statusCode + ": " + body);
            this.statusCode = statusCode;
            this.retryAfter = retryAfter;
        }

        public int statusCode() {
            return statusCode;
        }

        public Duration retryAfter() {
            return retryAfter;
        }
    }

    record ChatMsg(String role, String content) {}

    record ChatResp(String model, List<Choice> choices, Usage usage) {
        record Choice(ChatMsg message, @JsonProperty("finish_reason") String finishReason) {}

        record Usage(@JsonProperty("prompt_tokens") long promptTokens,
                     @JsonProperty("completion_tokens") long completionTokens) {}
    }

    @Override
    public Response complete(Request req) throws Exception {
        List<ChatMsg> msgs = new ArrayList<>();
        msgs.add(new ChatMsg("system", req.system()));
        req.messages().forEach(m -> msgs.add(new ChatMsg(m.role(), m.text())));
        byte[] body = Flags.JSON.writeValueAsBytes(
                Map.of("model", req.model(), "messages", msgs, "max_tokens", req.maxTokens()));
        var hreq = HttpRequest.newBuilder(URI.create(baseUrl + "/chat/completions"))
                .header("Content-Type", "application/json")
                .header("Authorization", "Bearer " + apiKey)
                .POST(HttpRequest.BodyPublishers.ofByteArray(body));
        if (req.expired()) {
            throw new TimeoutException("deadline exceeded");
        }
        req.timeLeft().ifPresent(hreq::timeout);
        // IOException (сеть, HttpTimeoutException): retryable разберётся.
        HttpResponse<byte[]> resp = http.send(hreq.build(), HttpResponse.BodyHandlers.ofByteArray());
        if (resp.statusCode() != 200) {
            byte[] b = resp.body();
            throw new HttpError(resp.statusCode(),
                    Retry.retryAfter(resp.headers().firstValue("retry-after").orElse(null)),
                    new String(b, 0, Math.min(b.length, 1024), StandardCharsets.UTF_8));
        }
        ChatResp out = Flags.JSON.readValue(resp.body(), ChatResp.class);
        if (out.choices() == null || out.choices().isEmpty()) {
            throw new HttpError(502, Duration.ZERO, "no choices");
        }
        var c = out.choices().getFirst();
        // stop_reason приводим к словарю Anthropic: остальная цепочка (кэш, метрики) знает только его.
        String stop = switch (Objects.requireNonNullElse(c.finishReason(), "")) {
            case "stop" -> "end_turn";
            case "length" -> "max_tokens";
            case "content_filter" -> "refusal";
            default -> "";
        };
        var usage = out.usage() != null ? out.usage() : new ChatResp.Usage(0, 0);
        return new Response(c.message().content(), out.model(), stop, "",
                usage.promptTokens(), usage.completionTokens(), 0, 0);
    }
}

Чтобы Retryable, breaker и метрика error.type понимали обе ошибки, классификация статуса вынесена в одну функцию httpStatus (вкладка 1 в сборке). Ретраи с backoff и jitter — из первого урока: в шлюзе они живут ровно на одном слое, а клиенты свои выключают. Бюджеты на тенанта — Metered из той же сборки.

Build vs buy

Готовые шлюзы закрывают большую часть функций за дни. Ландшафт по состоянию на октябрь 2026:

• Что заставляет писать своё: специфичные политики данных (классы данных, PII-маскирование с обратной подстановкой), требования к аудиту и резидентности, жёсткий бюджет на overhead, нативные функции провайдеров. Унифицированный OpenAI-формат теряет prompt caching с явными breakpoint'ами, thinking, citations — проверьте, умеет ли шлюз проксировать нативные API без переупаковки.
• Что заставляет покупать: скорость, покрытие провайдеров, готовый UI для ключей и бюджетов, отсутствие команды на поддержку.
• Риск вендора реален. Только за 2026 год (по состоянию на октябрь) репозиторий TensorZero заархивирован, Helicone после покупки Mintlify в марте перешёл в режим поддержки (исправления безопасности и новые модели, без новых функций), Portkey сменил владельца. Держите шлюз за своим интерфейсом LLM в коде, чтобы миграция была конфигом, а не переписыванием.

Observability

Классических RED-метрик (rate, errors, duration) для LLM мало. Вам нужны токены, стоимость, TTFT, stop reasons, cache hit rate, а для отладки ещё и содержимое промптов и ответов, что сразу упирается в PII.

OpenTelemetry GenAI semantic conventions

В OpenTelemetry есть семантические конвенции для GenAI. Их поддерживают Langfuse, Arize Phoenix, Datadog и другие бэкенды, поэтому, если придерживаться конвенций, трейсы читаются где угодно. Статус на октябрь 2026 — Development: имена ещё меняются. С semconv 1.42 GenAI-конвенции вынесены в отдельный репозиторий semantic-conventions-genai, у которого пока нет релизов; в основном репозитории они с 1.42 помечены устаревшими, последний релиз без этой пометки — v1.41.1 (май 2026), и имена ниже даны по нему. Инструментации, которые пишут имена до v1.36, переключаются на новые переменной OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental. В основной ветке нового репозитория уже лежат несовместимые переименования (например, cache_creation → cache_write, метрики инференса gen_ai.client.inference.*), поэтому имена держите в одном пакете, а переходные маппинги делайте в OTel Collector.

Метрики по конвенциям — гистограммы с рекомендованными границами бакетов: gen_ai.client.operation.duration (секунды, от 0.01 до 81.92 с удвоением), gen_ai.client.token.usage (с gen_ai.token.type = input|output, бакеты — степени 4) и, с v1.41, клиентский TTFT gen_ai.client.operation.time_to_first_chunk вместе с time_per_output_chunk. Для серверов инференса — gen_ai.server.request.duration, gen_ai.server.time_to_first_token и gen_ai.server.time_per_output_token. Содержимое промптов и ответов (gen_ai.input.messages, gen_ai.output.messages, gen_ai.system_instructions) по конвенциям пишется только при явном opt-in, и это правильный дефолт.

Структура трейса

POST /api/ask                                   1.84s   tenant=acme  user=u_9f2… (hash)
├─ guardrail.input  chat claude-haiku-4-5        0.21s   (параллельно с retrieval)
├─ retrieve                                      0.19s
│  ├─ embeddings voyage-3                        0.07s
│  ├─ db.query pgvector top_k=40                 0.05s
│  └─ search bm25 top_k=40                       0.04s   (параллельно)
├─ rerank                                        0.15s
├─ chat claude-opus-5                            1.21s   in=11 820 (cache_read=10 400) out=310  $0.0158
│  └─ execute_tool lookup_order                  0.09s
└─ guardrail.output                              0.06s

Для агентов спаны вкладываются так: invoke_agent, внутри него каждая итерация с chat и execute_tool. Так видно, на какой итерации агент «заблудился», и сколько стоила каждая.

Спаны агента и TTFT в коде

Агент, работающий в вашем процессе, получает спан invoke_agent {имя} с kind INTERNAL; вызов удалённого агента по сети — CLIENT. Вызов инструмента — execute_tool {имя}, тоже INTERNAL. Вложенность получается сама: Traced и ExecuteTool берут родителя из ctx.Traced и executeTool делают свой спан текущим (makeCurrent()), а дочерние берут родителя из Context.current(). При переходе в другой поток контекст переносят явно: Context.current().wrap(executor). TTFT пишется колбэком из стриминга: первый чанк записывает гистограмму, остальные ничего не делают.

package llmmw

import (
	"context"
	"sync"
	"time"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/metric"
	"go.opentelemetry.io/otel/trace"
)

// TTFT на клиенте — метрика конвенции (semconv v1.41, статус Development).
var ttftHist, _ = meter.Float64Histogram("gen_ai.client.operation.time_to_first_chunk", metric.WithUnit("s"),
	metric.WithExplicitBucketBoundaries(0.01, 0.02, 0.04, 0.08, 0.16, 0.32, 0.64, 1.28, 2.56, 5.12, 10.24, 20.48, 40.96, 81.92))

// FirstChunk возвращает колбэк для стрима: первый вызов пишет TTFT, остальные — no-op.
// Вызывайте его из onText в StreamWithDeadlines.
func FirstChunk(ctx context.Context, start time.Time, attrs ...attribute.KeyValue) func() {
	var once sync.Once
	return func() {
		once.Do(func() { ttftHist.Record(ctx, time.Since(start).Seconds(), metric.WithAttributes(attrs...)) })
	}
}

// InvokeAgent — спан всей задачи агента. Спаны chat и execute_tool внутри run
// становятся его детьми через ctx: в трейсе видно, на какой итерации агент «заблудился».
func InvokeAgent(ctx context.Context, agent string, run func(ctx context.Context) error) error {
	ctx, span := tracer.Start(ctx, "invoke_agent "+agent,
		trace.WithSpanKind(trace.SpanKindInternal), // агент в нашем процессе; удалённый агент — CLIENT
		trace.WithAttributes(
			attribute.String("gen_ai.operation.name", "invoke_agent"),
			attribute.String("gen_ai.agent.name", agent),
		))
	defer span.End()
	err := run(ctx)
	if err != nil {
		span.RecordError(err)
		span.SetStatus(codes.Error, "agent failed")
		span.SetAttributes(attribute.String("error.type", errorType(err)))
	}
	return err
}

// ExecuteTool — спан одного вызова инструмента. Аргументы и результат в атрибуты
// не пишем: по конвенции это opt-in (gen_ai.tool.call.arguments/result), там бывают PII.
func ExecuteTool(ctx context.Context, tool, callID string, fn func(ctx context.Context) (string, error)) (string, error) {
	ctx, span := tracer.Start(ctx, "execute_tool "+tool,
		trace.WithSpanKind(trace.SpanKindInternal),
		trace.WithAttributes(
			attribute.String("gen_ai.operation.name", "execute_tool"),
			attribute.String("gen_ai.tool.name", tool),
			attribute.String("gen_ai.tool.call.id", callID),
			attribute.String("gen_ai.tool.type", "function"),
		))
	defer span.End()
	out, err := fn(ctx)
	if err != nil {
		span.RecordError(err)
		span.SetStatus(codes.Error, "tool failed")
		span.SetAttributes(attribute.String("error.type", errorType(err)))
	}
	return out, err
}

package course.m09prod.llmmw;

import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.metrics.DoubleHistogram;
import io.opentelemetry.api.metrics.LongHistogram;
import io.opentelemetry.api.metrics.Meter;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.SpanKind;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.Scope;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicBoolean;

/** Инструменты OpenTelemetry по конвенции GenAI: трейсер, гистограммы, спаны агента и инструментов. */
public final class GenAi {
    final Tracer tracer;
    final LongHistogram tokenUsage;
    final DoubleHistogram opDuration, ttft;

    // Границы бакетов — рекомендованные конвенцией: токены растут степенями 4, секунды — удвоением.
    private static final List<Long> TOKENS = List.of(1L, 4L, 16L, 64L, 256L, 1024L, 4096L, 16384L, 65536L,
            262144L, 1048576L, 4194304L, 16777216L, 67108864L);
    private static final List<Double> SECONDS = List.of(0.01, 0.02, 0.04, 0.08, 0.16, 0.32, 0.64, 1.28,
            2.56, 5.12, 10.24, 20.48, 40.96, 81.92);

    public GenAi(OpenTelemetry otel) {
        tracer = otel.getTracer("llmmw");
        Meter meter = otel.getMeter("llmmw");
        tokenUsage = meter.histogramBuilder("gen_ai.client.token.usage").ofLongs().setUnit("{token}")
                .setExplicitBucketBoundariesAdvice(TOKENS).build();
        opDuration = meter.histogramBuilder("gen_ai.client.operation.duration").setUnit("s")
                .setExplicitBucketBoundariesAdvice(SECONDS).build();
        // TTFT на клиенте — метрика конвенции (semconv v1.41, статус Development).
        ttft = meter.histogramBuilder("gen_ai.client.operation.time_to_first_chunk").setUnit("s")
                .setExplicitBucketBoundariesAdvice(SECONDS).build();
    }

    /**
     * Колбэк для стрима: первый вызов пишет TTFT, остальные — no-op.
     * Вызывайте его из onText в streamWithDeadlines.
     */
    public Runnable firstChunk(long startNanos, Attributes attrs) {
        var once = new AtomicBoolean();
        return () -> {
            if (once.compareAndSet(false, true)) {
                ttft.record((System.nanoTime() - startNanos) / 1e9, attrs);
            }
        };
    }

    /**
     * Спан всей задачи агента. Спаны chat и execute_tool внутри run становятся его детьми
     * через Context.current(): в трейсе видно, на какой итерации агент «заблудился».
     */
    public <T> T invokeAgent(String agent, Callable<T> run) throws Exception {
        Span span = tracer.spanBuilder("invoke_agent " + agent)
                .setSpanKind(SpanKind.INTERNAL) // агент в нашем процессе; удалённый агент — CLIENT
                .setAttribute("gen_ai.operation.name", "invoke_agent")
                .setAttribute("gen_ai.agent.name", agent)
                .startSpan();
        try (Scope ignored = span.makeCurrent()) {
            return run.call();
        } catch (Exception e) {
            fail(span, e, "agent failed");
            throw e;
        } finally {
            span.end();
        }
    }

    /**
     * Спан одного вызова инструмента. Аргументы и результат в атрибуты не пишем: по конвенции
     * это opt-in (gen_ai.tool.call.arguments/result), там бывают PII.
     */
    public String executeTool(String tool, String callId, Callable<String> fn) throws Exception {
        Span span = tracer.spanBuilder("execute_tool " + tool)
                .setSpanKind(SpanKind.INTERNAL)
                .setAttribute("gen_ai.operation.name", "execute_tool")
                .setAttribute("gen_ai.tool.name", tool)
                .setAttribute("gen_ai.tool.call.id", callId)
                .setAttribute("gen_ai.tool.type", "function")
                .startSpan();
        try (Scope ignored = span.makeCurrent()) {
            return fn.call();
        } catch (Exception e) {
            fail(span, e, "tool failed");
            throw e;
        } finally {
            span.end();
        }
    }

    static void fail(Span span, Exception e, String msg) {
        span.recordException(e);
        span.setStatus(StatusCode.ERROR, msg);
        span.setAttribute("error.type", errorType(e));
    }

    static String errorType(Throwable e) {
        Retry.Status s = Retry.httpStatus(e);
        return s != null ? String.valueOf(s.code()) : "_OTHER"; // значение по конвенции OTel для неклассифицированных ошибок
    }
}

Использование: onText в StreamWithDeadlines оборачивается так: first := FirstChunk(ctx, time.Now(), attribute.String("gen_ai.request.model", model)), а внутри колбэка первой строкой вызывается first(). Тест в examples/m09-prod проверяет вложенность спанов через tracetest.SpanRecorder и наличие метрик через metric.NewManualReader, без коллектора и сети.

Использование: onText в streamWithDeadlines оборачивается так: Runnable first = genai.firstChunk(System.nanoTime(), Attributes.of(stringKey("gen_ai.request.model"), model)), а внутри колбэка первой строкой вызывается first.run(). Тест в examples/java/m09-prod проверяет вложенность спанов через InMemorySpanExporter и наличие метрик через InMemoryMetricReader (артефакт opentelemetry-sdk-testing), без коллектора и сети.

Инструменты

• Langfuse — open source, можно поднять у себя (важно для приватности), принимает OTLP, умеет трейсы, промпт-менеджмент, датасеты и оценки.
• LangSmith — SaaS от LangChain, удобен, если вы уже на LangChain/LangGraph; поддерживает OTel.
• Arize Phoenix, Datadog LLM Observability — альтернативы. Выбор определяется тем, где у вас уже OTel-коллектор и что разрешает security.

Практичная схема: инструментируете код OTel-ом, отправляете в свой OTel Collector, а оттуда и в Tempo/Jaeger/Datadog для SRE, и в Langfuse для AI-команды. Вы не привязаны к вендору, а PII можно вычищать процессором на коллекторе.

Метрики стоимости

Стоимость считается на клиенте из usage и таблицы цен, счёта провайдера раз в месяц для этого недостаточно. Метка {tenant, feature, model, route} позволяет отвечать на вопросы «сколько стоит фича X на одного активного пользователя» и «почему вчера расходы выросли на 40%» (обычно из-за промаха кэша или зациклившегося агента). Не забудьте учесть cache_creation_input_tokens (1.25× или 2×) и cache_read_input_tokens (0.1× у большинства моделей): в input_tokens у Anthropic они не входят.

PII в логах

• Маскирование до записи. Регэкспы для email, телефонов, карт и паспортов плюс NER (Presidio) для имён и адресов. Маскируйте в приложении или процессором в OTel Collector, но до того, как данные покинут периметр.
• Сэмплирование содержимого. Метаданные (токены, латентность, модель) пишите для 100% запросов, полный текст для 1–5% и для всех ошибок. Это и дешевле, и безопаснее.
• Retention и доступ. Сырые промпты 14–30 дней в отдельном хранилище с отдельными правами; агрегаты храните дольше. Нужны процедура удаления по запросу пользователя (GDPR) и аудит доступа.
• Идентификаторы. user_id пишите через HMAC, а не открытым текстом. Ключ HMAC ротируется.

// Грубое маскирование PII перед логированием. Регэкспы ловят «структурные» PII;
// имена и адреса ловит NER (Presidio). Маскировать надо до того, как текст уйдёт в логи.
var piiPatterns = []struct {
	re   *regexp.Regexp
	mask string
}{
	{regexp.MustCompile(`[\w.+-]+@[\w-]+\.[\w.-]+`), "<EMAIL>"},
	{regexp.MustCompile(`\b(?:\d[ -]?){13,19}\b`), "<CARD>"},
	{regexp.MustCompile(`\+?\d[\d\s()-]{9,}\d`), "<PHONE>"},
}

func MaskPII(s string) string {
	for _, p := range piiPatterns {
		s = p.re.ReplaceAllString(s, p.mask)
	}
	return s
}

// Грубое маскирование PII перед логированием. Регэкспы ловят «структурные» PII;
// имена и адреса ловит NER (Presidio). Маскировать надо до того, как текст уйдёт в логи.
record PiiPattern(Pattern re, String mask) {}

static final List<PiiPattern> PII_PATTERNS = List.of(
        new PiiPattern(Pattern.compile("[\\w.+-]+@[\\w-]+\\.[\\w.-]+"), "<EMAIL>"),
        new PiiPattern(Pattern.compile("\\b(?:\\d[ -]?){13,19}\\b"), "<CARD>"),
        new PiiPattern(Pattern.compile("\\+?\\d[\\d\\s()-]{9,}\\d"), "<PHONE>"));

static String maskPii(String s) {
    for (PiiPattern p : PII_PATTERNS) {
        s = p.re().matcher(s).replaceAll(p.mask());
    }
    return s;
}

SLO для недетерминированных систем

«Доступность 99.9%» для LLM-фичи почти ничего не говорит. Провайдер может вернуть 200 с обрезанным стримом, невалидным JSON или уверенной ерундой, а может тихо деградировать после обновления модели. Классический подход SRE при этом работает, если правильно выбрать, что считать «хорошим событием». SLI (service level indicator) — доля хороших событий среди всех; SLO — цель для этой доли за окно (обычно 30 дней); error budget — разрешённая доля плохих событий, 1 − SLO.

Латентность записывайте как долю запросов быстрее порога, а не как «p95 ≤ 2 с». Доля складывается в error budget так же, как ошибки, а перцентиль — нет.

Burn rate и multiwindow-алерты

Burn rate — во сколько раз быстрее допустимого тратится бюджет: доля плохих событий в окне, делённая на 1 − SLO. При burn rate 1 бюджет кончится ровно к концу 30-дневного окна, при 14.4 — за 30 / 14.4 ≈ 2 дня. Алерт «доля ошибок больше 1% за 5 минут» либо будит дежурного на каждый всплеск, либо, при длинном окне, срабатывает через час после начала аварии и ещё час горит после её конца. Google SRE Workbook рекомендует несколько правил с разными burn rate, каждое на двух окнах. Длинное окно отвечает за значимость, короткое (1/12 длинного) — за то, чтобы алерт погас, как только инцидент закончился.

Параметры из SRE Workbook для SLO 99.9% на 30 дней; burn rate и окна не зависят от цели SLO, от неё зависит только порог по доле ошибок (burn × (1 − SLO)). Recording rules и такие алерты для Prometheus руками обычно не пишут: генераторы вроде Sloth и Pyrra строят их из короткого описания SLO.

Особенность LLM: SLI качества по выборке

• Шум. Judge оценивает 2% ответов: при 100 запросах в минуту это 10 оценок за 5 минут и 120 за час. Долю плохих в доли процента на таком объёме не измерить: одна плохая оценка — уже «10% плохих» в коротком окне, и быстрое правило срабатывает от случайного скопления нескольких плохих оценок. Для выборочных SLI берите длинные пары окон (6 ч / 30 мин и 3 дня / 6 ч) или минимум событий в коротком окне, а во время инцидента временно поднимайте долю выборки.
• Ошибки самого судьи. Измеренная доля плохих = истинная × TPR + (1 − истинная) × (1 − TNR). Судья с TNR 97% даже на идеальной системе показывает около 3% «плохих», а это шесть бюджетов SLO 99.5%. Калибруйте судью (модуль 8) и задавайте SLO, соизмеримый с его точностью (например, 97% «хороших» по мнению судьи), а не 99.5%.
• Задержка. Асинхронный judge отстаёт на минуты. Для быстрых алертов годятся только SLI, которые считаются на месте: доступность, TTFT, полнота стрима, валидность.
• Тихая деградация. Обновление модели у провайдера выглядит как падение качества без единой ошибки. Это ловит только SLI качества и мониторинг дрейфа (следующий урок).

package slo

import "time"

// Counts — события SLI за окно: всего и «плохих».
type Counts struct{ Total, Bad float64 }

// BurnRate — во сколько раз быстрее допустимого тратится error budget:
// 1 — бюджет кончится ровно к концу окна SLO, 14.4 — за 30 дней / 14.4 ≈ 2 дня.
func BurnRate(c Counts, objective float64) float64 {
	if c.Total == 0 {
		return 0
	}
	return (c.Bad / c.Total) / (1 - objective)
}

// BudgetLeft — какая доля error budget осталась за окно SLO (уходит в минус при перерасходе).
func BudgetLeft(c Counts, objective float64) float64 {
	if c.Total == 0 {
		return 1
	}
	return 1 - c.Bad/(c.Total*(1-objective))
}

type Rule struct {
	Long, Short time.Duration
	Burn        float64
	Page        bool // false — тикет, а не звонок дежурному
}

// Workbook — multiwindow, multi-burn-rate правила из Google SRE Workbook для SLO на 30 дней.
// Короткое окно = 1/12 длинного: алерт гаснет, как только инцидент закончился.
var Workbook = []Rule{
	{Long: time.Hour, Short: 5 * time.Minute, Burn: 14.4, Page: true},   // 2% бюджета за час
	{Long: 6 * time.Hour, Short: 30 * time.Minute, Burn: 6, Page: true}, // 5% за 6 часов
	{Long: 72 * time.Hour, Short: 6 * time.Hour, Burn: 1, Page: false},  // 10% за 3 дня
}

// Firing возвращает сработавшие правила. window(d) отдаёт счётчики за последние d.
// minEvents отсекает решения по горстке событий: на малом трафике и на выборочном SLI
// (онлайн-judge на 2% ответов) короткое окно — это шум, а не сигнал.
func Firing(rules []Rule, objective, minEvents float64, window func(time.Duration) Counts) []Rule {
	var out []Rule
	for _, r := range rules {
		long, short := window(r.Long), window(r.Short)
		if short.Total < minEvents {
			continue
		}
		if BurnRate(long, objective) >= r.Burn && BurnRate(short, objective) >= r.Burn {
			out = append(out, r)
		}
	}
	return out
}

package course.m09prod.slo;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;

public final class BurnRate {
    private BurnRate() {}

    /** События SLI за окно: всего и «плохих». */
    public record Counts(double total, double bad) {}

    /**
     * Во сколько раз быстрее допустимого тратится error budget:
     * 1 — бюджет кончится ровно к концу окна SLO, 14.4 — за 30 дней / 14.4 ≈ 2 дня.
     */
    public static double burnRate(Counts c, double objective) {
        return c.total() == 0 ? 0 : (c.bad() / c.total()) / (1 - objective);
    }

    /** Какая доля error budget осталась за окно SLO (уходит в минус при перерасходе). */
    public static double budgetLeft(Counts c, double objective) {
        return c.total() == 0 ? 1 : 1 - c.bad() / (c.total() * (1 - objective));
    }

    /** page = false — тикет, а не звонок дежурному. */
    public record Rule(Duration longWindow, Duration shortWindow, double burn, boolean page) {}

    /**
     * Multiwindow, multi-burn-rate правила из Google SRE Workbook для SLO на 30 дней.
     * Короткое окно = 1/12 длинного: алерт гаснет, как только инцидент закончился.
     */
    public static final List<Rule> WORKBOOK = List.of(
            new Rule(Duration.ofHours(1), Duration.ofMinutes(5), 14.4, true),   // 2% бюджета за час
            new Rule(Duration.ofHours(6), Duration.ofMinutes(30), 6, true),     // 5% за 6 часов
            new Rule(Duration.ofHours(72), Duration.ofHours(6), 1, false));     // 10% за 3 дня

    /**
     * Возвращает сработавшие правила. window(d) отдаёт счётчики за последние d.
     * minEvents отсекает решения по горстке событий: на малом трафике и на выборочном SLI
     * (онлайн-judge на 2% ответов) короткое окно — это шум, а не сигнал.
     */
    public static List<Rule> firing(List<Rule> rules, double objective, double minEvents,
                                    Function<Duration, Counts> window) {
        List<Rule> out = new ArrayList<>();
        for (Rule r : rules) {
            Counts lng = window.apply(r.longWindow()), shrt = window.apply(r.shortWindow());
            if (shrt.total() < minEvents) {
                continue;
            }
            if (burnRate(lng, objective) >= r.burn() && burnRate(shrt, objective) >= r.burn()) {
                out.add(r);
            }
        }
        return out;
    }
}

Тест в examples/m09-prod/sloТест BurnRateTest в examples/java/m09-prod прогоняет четыре сценария: норма (тишина), 20-минутная авария на 30% (только быстрый page), четыре дня тлеющей деградации на 0.6% при SLO 99.5% (только тикет) и авария, закончившаяся два часа назад (тишина благодаря короткому окну). Отдельно проверяется выборочный SLI: 9 плохих оценок judge за 15 минут без порога событий будят дежурного, а с порогом 50 событий в коротком окне — нет. Это осознанный компромисс: по выборочному SLI быстрый page не строят.

Организационная сторона — что делать, когда бюджет кончился (заморозка выкаток промптов и моделей, приоритет надёжности над фичами), кто дежурит и как писать постмортем AI-инцидента, — в модуле 13 (политика error budget и on-call, постмортем).

Мониторинг дрейфа и качества в проде

LLM-фича деградирует без ошибок. Пользователи начинают спрашивать о новом (запустили продукт, сменился сезон), меняется доля языков, в чат вставляют длинные документы, провайдер обновляет модель за алиасом, кто-то переиндексировал базу RAG. SLO доступности при этом зелёный. Дрейф (drift) — сдвиг распределения входов или выходов относительно эталонного периода; сам по себе это не инцидент, а сигнал проверить, работает ли система на новом трафике так же хорошо.

PSI (population stability index) сравнивает доли по корзинам: PSI = Σ (qᵢ − pᵢ) · ln(qᵢ / pᵢ), где p — эталон, q — текущее окно. Распространённое эмпирическое правило из кредитного скоринга: меньше 0.1 — стабильно, 0.1–0.25 — заметный сдвиг, больше 0.25 — сильный. Это ориентир, а не закон. Посчитайте PSI между двумя спокойными соседними неделями: это ваш шумовой фон. JS-дивергенция (Йенсена–Шеннона) симметрична, ограничена отрезком [0, 1] при логарифме по основанию 2 и не ломается на пустых корзинах, поэтому её удобнее показывать на дашборде.

package drift

import "math"

// PSI — population stability index между эталонным распределением (например, прошлый месяц)
// и текущим (последние сутки) по категориям: темы, языки, бакеты длины.
// Эмпирика из кредитного скоринга: < 0.1 — стабильно, 0.1–0.25 — заметный сдвиг, > 0.25 — сильный.
func PSI(ref, cur map[string]float64) float64 {
	const eps = 1e-4 // пустая корзина дала бы ln(0) = −∞
	p, q := shares(ref, cur)
	var psi float64
	for k := range p {
		a, b := max(p[k], eps), max(q[k], eps)
		psi += (b - a) * math.Log(b/a)
	}
	return psi
}

// JS — дивергенция Йенсена–Шеннона по основанию 2: симметрична, ограничена [0, 1],
// пустые корзины не ломают формулу.
func JS(ref, cur map[string]float64) float64 {
	p, q := shares(ref, cur)
	kl := func(a, m float64) float64 {
		if a == 0 {
			return 0
		}
		return a * math.Log2(a/m)
	}
	var js float64
	for k := range p {
		m := (p[k] + q[k]) / 2
		js += kl(p[k], m)/2 + kl(q[k], m)/2
	}
	return js
}

// shares переводит счётчики в доли по объединению ключей: новая тема в текущем окне —
// тоже сигнал дрейфа, её нельзя потерять.
func shares(a, b map[string]float64) (map[string]float64, map[string]float64) {
	p, q := map[string]float64{}, map[string]float64{}
	var sa, sb float64
	for _, v := range a {
		sa += v
	}
	for _, v := range b {
		sb += v
	}
	for k := range a {
		p[k], q[k] = a[k]/sa, b[k]/sb
	}
	for k := range b {
		p[k], q[k] = a[k]/sa, b[k]/sb
	}
	return p, q
}

// LengthBucket — корзина длины запроса в токенах: длина — самый дешёвый признак дрейфа.
func LengthBucket(tokens int) string {
	switch {
	case tokens < 50:
		return "<50"
	case tokens < 200:
		return "50-199"
	case tokens < 1000:
		return "200-999"
	default:
		return "1000+"
	}
}

package course.m09prod.drift;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

public final class Drift {
    private Drift() {}

    // PSI — population stability index между эталонным распределением (например, прошлый месяц)
    // и текущим (последние сутки) по категориям: темы, языки, бакеты длины.
    // Эмпирика из кредитного скоринга: < 0.1 — стабильно, 0.1–0.25 — заметный сдвиг, > 0.25 — сильный.
    public static double psi(Map<String, Double> ref, Map<String, Double> cur) {
        final double eps = 1e-4; // пустая корзина дала бы ln(0) = −∞
        Shares s = shares(ref, cur);
        double psi = 0;
        for (String k : s.p().keySet()) {
            double a = Math.max(s.p().get(k), eps), b = Math.max(s.q().get(k), eps);
            psi += (b - a) * Math.log(b / a);
        }
        return psi;
    }

    /** JS — дивергенция Йенсена–Шеннона по основанию 2: симметрична, ограничена [0, 1], пустые корзины не ломают формулу. */
    public static double js(Map<String, Double> ref, Map<String, Double> cur) {
        Shares s = shares(ref, cur);
        double js = 0;
        for (String k : s.p().keySet()) {
            double p = s.p().get(k), q = s.q().get(k), m = (p + q) / 2;
            js += kl(p, m) / 2 + kl(q, m) / 2;
        }
        return js;
    }

    private static double kl(double a, double m) {
        return a == 0 ? 0 : a * Math.log(a / m) / Math.log(2);
    }

    private record Shares(Map<String, Double> p, Map<String, Double> q) {}

    // shares переводит счётчики в доли по объединению ключей: новая тема в текущем окне —
    // тоже сигнал дрейфа, её нельзя потерять.
    private static Shares shares(Map<String, Double> a, Map<String, Double> b) {
        double sa = a.values().stream().mapToDouble(Double::doubleValue).sum();
        double sb = b.values().stream().mapToDouble(Double::doubleValue).sum();
        Set<String> keys = new HashSet<>(a.keySet());
        keys.addAll(b.keySet());
        Map<String, Double> p = new HashMap<>(), q = new HashMap<>();
        for (String k : keys) {
            p.put(k, a.getOrDefault(k, 0.0) / sa);
            q.put(k, b.getOrDefault(k, 0.0) / sb);
        }
        return new Shares(p, q);
    }

    /** Корзина длины запроса в токенах: длина — самый дешёвый признак дрейфа. */
    public static String lengthBucket(int tokens) {
        if (tokens < 50) return "<50";
        if (tokens < 200) return "50-199";
        if (tokens < 1000) return "200-999";
        return "1000+";
    }
}

• Сравнивайте сопоставимое. Понедельник с понедельником, рабочие часы с рабочими часами; окно не меньше нескольких сотен событий, иначе PSI шумит. Эталон обновляйте осознанно (после того как убедились, что новое распределение обслуживается хорошо), а не скользящим окном, которое «привыкает» к деградации.
• Онлайн-judge на выборке. Асинхронно, стратифицированно по кластерам (иначе редкие новые темы почти не попадут в выборку), тем же откалиброванным судьёй, что в CI. Порядок цены: 2% от 1M запросов в месяц — 20 000 оценок на Haiku 4.5 по ~3K токенов на вход и ~100 на выход, примерно $70 в месяц.
• Смена модели провайдером. Используйте датированные идентификаторы модели, где они есть, а не алиасы; пишите gen_ai.response.model в каждый спан; раз в сутки прогоняйте golden set против продакшн-конфигурации (тот же eval, что гейт в CI, модуль 8) и алертите на падение за пределы доверительного интервала.
• Алерты. Дрейф — это тикет, а не page: он означает «проверьте», а не «горит». Будит дежурного только burn rate SLI качества. Новые кластеры из корзины «новое» — кандидаты в golden set и разметку (data flywheel, модуль 11).

Фича-флаги для промптов и моделей

Смена промпта или модели меняет поведение продукта сильнее многих релизов кода, но часто выкатывается правкой строки сразу на 100% трафика. Фича-флаг разделяет деплой (новый вариант лежит в проде) и релиз (кто его получает). Флаг для LLM переключает не булево значение, а вариант: промпт с версией, модель и параметры вместе. Промпт, отлаженный под Opus 5, на Sonnet 5 ведёт себя иначе, поэтому раздельные флаги «промпт» и «модель» порождают непроверенные комбинации.

• Процент трафика с детерминированным бакетингом: hash(ключ флага + unit) mod 10000. Один пользователь всегда получает один вариант, а при росте процента 1 → 5 → 25 → 100 уже попавшие в эксперимент из него не выпадают. Ключ флага в хэше нужен, чтобы разные флаги резали трафик независимо.
• Unit бакетинга. Пользователь или диалог — чтобы вариант не сменился посреди разговора. Тенант — в B2B, где клиент должен видеть одно поведение. Отдельный запрос — только для stateless-задач вроде классификации: так больше статистической мощности, но пользователь видит разное поведение.
• Таргетинг. Сначала внутренние пользователи, затем бета-тенанты; регулируемых клиентов можно исключить до конца эксперимента.
• Kill switch. Вернуть всех на проверенный вариант за секунды и без деплоя. Учения с kill switch проводят заранее, а не во время инцидента.
• Exposure в трейсе. Имя варианта пишется в спан (app.flag.variant) и в события онлайн-evals: без этого метрики не разложить по вариантам.
• Кэши. Вариант входит в scope семантического кэша, а префикс prompt caching у варианта свой, поэтому во время эксперимента hit rate падает, а стоимость растёт. Закладывайте это в бюджет.

Жизненный цикл изменения промпта: офлайн-eval на golden set как гейт в CI (модуль 8, quality gate) → canary на внутренних пользователях → 5% → 25% → 50% с guardrail-метриками на каждой ступени (burn rate SLI, качество по judge, доля отказов и эскалаций, стоимость на запрос) → 100% → удаление флага и старого промпта. Флаг, переживший эксперимент на полгода, — это технический долг: два пути в коде, которые оба надо тестировать.

package llmmw

import (
	"encoding/json"
	"fmt"
	"hash/fnv"
	"sync/atomic"
)

// Variant — что переключает флаг: промпт, модель и параметры вместе. Промпт, отлаженный
// под одну модель, на другой ведёт себя иначе, поэтому раздельные флаги дают непроверенные комбинации.
type Variant struct {
	Name      string `json:"name"`   // пишется в трейс (app.flag.variant) и в онлайн-evals
	Prompt    string `json:"prompt"` // "answer@v13"
	Model     string `json:"model"`
	MaxTokens int64  `json:"max_tokens"`
}

type Flag struct {
	Key       string          `json:"key"`
	Killed    bool            `json:"killed"` // kill switch: всем Control, без деплоя
	Control   Variant         `json:"control"`
	Treatment Variant         `json:"treatment"`
	Percent   float64         `json:"percent"` // доля трафика на Treatment, 0–100
	Tenants   map[string]bool `json:"tenants"` // таргетинг: true — всегда Treatment, false — никогда
}

// Evaluate детерминирован: один unit (пользователь или тенант) всегда получает один вариант,
// а при росте Percent 1 → 5 → 25 → 100 попавшие в Treatment из него не выпадают.
func (f *Flag) Evaluate(tenant, unit string) Variant {
	if f.Killed {
		return f.Control
	}
	if on, ok := f.Tenants[tenant]; ok {
		if on {
			return f.Treatment
		}
		return f.Control
	}
	h := fnv.New32a()
	h.Write([]byte(f.Key + "/" + unit)) // ключ флага в хэше: разные флаги режут трафик независимо
	if float64(h.Sum32()%10000) < f.Percent*100 {
		return f.Treatment
	}
	return f.Control
}

// Registry хранит флаги и перечитывает их на лету (файл, etcd, сервис флагов).
type Registry struct {
	flags atomic.Pointer[map[string]Flag]
}

// Load заменяет конфиг целиком. Битый конфиг отклоняется, и работает предыдущий:
// опечатка в JSON не должна переключить весь прод на непроверенный промпт.
func (r *Registry) Load(data []byte) error {
	var list []Flag
	if err := json.Unmarshal(data, &list); err != nil {
		return fmt.Errorf("flags: %w", err)
	}
	m := make(map[string]Flag, len(list))
	for _, f := range list {
		if f.Percent < 0 || f.Percent > 100 || f.Control.Model == "" || f.Treatment.Model == "" {
			return fmt.Errorf("flags: %q: percent 0–100 and both models required", f.Key)
		}
		m[f.Key] = f
	}
	r.flags.Store(&m)
	return nil
}

// Variant — вариант для запроса; если флага нет, возвращается def (поведение «как до флагов»).
func (r *Registry) Variant(key, tenant, unit string, def Variant) Variant {
	m := r.flags.Load()
	if m == nil {
		return def
	}
	f, ok := (*m)[key]
	if !ok {
		return def
	}
	return f.Evaluate(tenant, unit)
}

package course.m09prod.llmmw;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;

public final class Flags {
    private Flags() {}

    static final ObjectMapper JSON = new ObjectMapper()
            .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

    /**
     * Что переключает флаг: промпт, модель и параметры вместе. Промпт, отлаженный под одну модель,
     * на другой ведёт себя иначе, поэтому раздельные флаги дают непроверенные комбинации.
     * name пишется в трейс (app.flag.variant) и в онлайн-evals; prompt — "answer@v13".
     */
    public record Variant(String name, String prompt, String model, @JsonProperty("max_tokens") long maxTokens) {
        public Variant(String name, String model) {
            this(name, null, model, 0);
        }
    }

    /**
     * killed — kill switch: всем control, без деплоя. percent — доля трафика на treatment, 0–100.
     * tenants — таргетинг: true — всегда treatment, false — никогда.
     */
    public record Flag(String key, boolean killed, Variant control, Variant treatment,
                       double percent, Map<String, Boolean> tenants) {

        /**
         * Детерминирован: один unit (пользователь или тенант) всегда получает один вариант,
         * а при росте percent 1 → 5 → 25 → 100 попавшие в treatment из него не выпадают.
         */
        public Variant evaluate(String tenant, String unit) {
            if (killed) {
                return control;
            }
            Boolean on = tenants == null ? null : tenants.get(tenant);
            if (on != null) {
                return on ? treatment : control;
            }
            // Ключ флага в хэше: разные флаги режут трафик независимо.
            long bucket = Integer.toUnsignedLong(fnv32a(key + "/" + unit)) % 10000;
            return bucket < percent * 100 ? treatment : control;
        }
    }

    static int fnv32a(String s) {
        int h = 0x811c9dc5;
        for (byte b : s.getBytes(StandardCharsets.UTF_8)) {
            h ^= b & 0xff;
            h *= 0x01000193;
        }
        return h;
    }

    /** Хранит флаги и перечитывает их на лету (файл, etcd, сервис флагов). */
    public static final class Registry {
        private final AtomicReference<Map<String, Flag>> flags = new AtomicReference<>();

        /**
         * Заменяет конфиг целиком. Битый конфиг отклоняется, и работает предыдущий:
         * опечатка в JSON не должна переключить весь прод на непроверенный промпт.
         */
        public void load(byte[] data) throws IOException {
            List<Flag> list = JSON.readValue(data, new TypeReference<>() {});
            Map<String, Flag> m = new HashMap<>();
            for (Flag f : list) {
                if (f.percent() < 0 || f.percent() > 100 || !hasModel(f.control()) || !hasModel(f.treatment())) {
                    throw new IllegalArgumentException("flags: \"" + f.key() + "\": percent 0–100 and both models required");
                }
                m.put(f.key(), f);
            }
            flags.set(Map.copyOf(m));
        }

        private static boolean hasModel(Variant v) {
            return v != null && v.model() != null && !v.model().isEmpty();
        }

        /** Вариант для запроса; если флага нет, возвращается def (поведение «как до флагов»). */
        public Variant variant(String key, String tenant, String unit, Variant def) {
            Map<String, Flag> m = flags.get();
            Flag f = m == null ? null : m.get(key);
            return f == null ? def : f.evaluate(tenant, unit);
        }
    }
}

Тест проверяет то, что ломается на практике: доля Treatment на 20 000 пользователях около 10%, пользователи из 10% остаются в Treatment при росте до 30%, таргетинг и kill switch возвращают всех на Control, битый конфиг (процент 250) отклоняется, и работает предыдущий.

В проде флаги обычно живут в сервисе флагов (flagd, Unleash, LaunchDarkly, GO Feature Flag и др.), а код читает их через вендор-нейтральный OpenFeature SDK: процент, таргетинг и kill switch настраиваются в сервисе, а вызов в коде не зависит от провайдера. Сверено с github.com/open-feature/go-sdk v1.19:Сверено с dev.openfeature:sdk 1.20:

package llmmw

import (
	"context"
	"encoding/json"

	"github.com/open-feature/go-sdk/openfeature"
)

// VariantOF — тот же выбор варианта через OpenFeature: процент, таргетинг и kill switch
// живут в сервисе флагов (flagd, Unleash, LaunchDarkly и др.), код от вендора не зависит.
func VariantOF(ctx context.Context, c *openfeature.Client, key, tenant, user string, def Variant) Variant {
	v, err := c.ObjectValue(ctx, key, def, openfeature.NewEvaluationContext(user, map[string]any{"tenant": tenant}))
	if err != nil {
		return def // сервис флагов недоступен — работаем на проверенном варианте
	}
	var out Variant // провайдеры отдают объект как map[string]any из JSON
	if b, err := json.Marshal(v); err != nil || json.Unmarshal(b, &out) != nil || out.Model == "" {
		return def
	}
	return out
}

package course.m09prod.llmmw;

import course.m09prod.llmmw.Flags.Variant;
import dev.openfeature.sdk.Client;
import dev.openfeature.sdk.MutableContext;
import dev.openfeature.sdk.Value;

public final class OpenFeatureFlags {
    private OpenFeatureFlags() {}

    /**
     * Тот же выбор варианта через OpenFeature: процент, таргетинг и kill switch живут в сервисе
     * флагов (flagd, Unleash, LaunchDarkly и др.), код от вендора не зависит.
     */
    public static Variant variant(Client c, String key, String tenant, String user, Variant def) {
        try {
            var d = c.getObjectDetails(key, new Value(), new MutableContext(user).add("tenant", tenant));
            if (d.getErrorCode() != null || !d.getValue().isStructure()) {
                return def; // флага нет или сервис флагов недоступен — работаем на проверенном варианте
            }
            // Провайдеры отдают объект как Structure из JSON.
            Variant out = Flags.JSON.convertValue(d.getValue().asStructure().asObjectMap(), Variant.class);
            return out.model() == null || out.model().isEmpty() ? def : out;
        } catch (RuntimeException e) {
            return def;
        }
    }
}

Guardrails и модерация

Guardrails — проверки вокруг вызова модели, которые не зависят от её доброй воли. Системный промпт «не обсуждай конкурентов» считается пожеланием, а не гарантией. Здесь — как guardrails встают в продакшн-цепочку: латентность, fail-open или fail-closed, метрики. Модель угроз, OWASP Top 10 для LLM-приложений и red-teaming — в модуле 7 (OWASP, red-teaming).

• Чем проверять. Регэкспы и правила быстрые, дешёвые и предсказуемые. Специализированные классификаторы — Llama Guard, модерационные эндпоинты провайдеров, свой fine-tuned маленький классификатор (модуль 10). LLM-as-judge на Haiku-классе гибкий, но добавляет 200–500 мс и стоит денег. На практике их комбинируют: сначала дешёвое, дорогое только при сомнении.
• Метрики guardrails. У каждого есть precision и recall. False positive — это легитимный пользователь, которому отказали, и такие ошибки убивают продукт так же, как пропуски. Держите размеченный набор и считайте обе ошибки.
• Fail-open или fail-closed. Если классификатор недоступен, пропускаем или блокируем? Для модерации детского контента ответ «блокируем», для тематического фильтра внутреннего инструмента — «пропускаем и логируем». Это решение надо принять явно, а не получить случайно из-за if err != nil { return true }catch (Exception e) { return true; }.
• Стриминг и выходные проверки конфликтуют. Нельзя проверить ответ, пока он не сгенерирован. Варианты: проверять по предложениям с буфером в 1–2 предложения, стримить и «отзывать» сообщение (плохой UX), или не стримить там, где риск высокий.
• Политика отказов. Единый тон и формат отказа, объяснение («не могу помочь с X, могу с Y»), метрика доли отказов по причинам. Отказ должен быть продуктовым сценарием, а не исключением.

// Input guardrail параллельно с основным вызовом: латентность не растёт,
// при блокировке основной вызов отменяется через общий ctx.
var ErrBlocked = errors.New("guardrail: input blocked")

func GuardedComplete(ctx context.Context, llm LLM, req Request,
	guard func(context.Context, string) (allowed bool, err error)) (*Response, error) {

	g, gctx := errgroup.WithContext(ctx)
	var resp *Response
	g.Go(func() error {
		ok, err := guard(gctx, lastUserText(req))
		switch {
		case err != nil:
			return fmt.Errorf("guardrail unavailable: %w", err) // fail-closed: осознанный выбор
		case !ok:
			return ErrBlocked
		}
		return nil
	})
	g.Go(func() (err error) {
		resp, err = llm.Complete(gctx, req)
		return err
	})
	if err := g.Wait(); err != nil {
		return nil, err
	}
	return resp, nil
}

// Input guardrail параллельно с основным вызовом: латентность не растёт,
// при блокировке основной вызов отменяется (interrupt его виртуального потока).
public static final class BlockedException extends RuntimeException {
    public BlockedException() {
        super("guardrail: input blocked");
    }
}

@FunctionalInterface
public interface Guard {
    boolean allowed(String text) throws Exception;
}

public static Response guardedComplete(LLM llm, Request req, Guard guard) throws Exception {
    try (var exec = Executors.newVirtualThreadPerTaskExecutor()) {
        Future<Response> main = exec.submit(() -> llm.complete(req));
        boolean ok;
        try {
            ok = guard.allowed(req.lastUserText());
        } catch (Exception e) {
            main.cancel(true);
            throw new IllegalStateException("guardrail unavailable", e); // fail-closed: осознанный выбор
        }
        if (!ok) {
            main.cancel(true);
            throw new BlockedException();
        }
        try {
            return main.get();
        } catch (ExecutionException e) {
            throw e.getCause() instanceof Exception c ? c : e;
        }
    }
}

Полная цепочка guards с fail-closed по умолчанию, allowlist инструментов со scopes и red-team как gate — модуль 7.

Self-hosted инференс: коротко

Свой инференс поднимают по четырём причинам: приватность (данные не покидают периметр), стоимость на большом и ровном объёме, контроль (своя fine-tuned модель, фиксированная версия, предсказуемая латентность) и офлайн или edge. Для этого модуля важно одно: self-hosted модель — ещё одно звено цепочки. Её подключают через тот же OpenAICompat-адаптерчерез тот же OpenAiCompat-адаптер (vLLM, SGLang и Ollama отдают OpenAI-совместимый API), закрывают breaker'ом и прогоняют по тем же evals.

• Сравнивайте $ за 1M токенов при реальной утилизации, а не цену GPU-часа: $/1M = GPU $/ч ÷ (tok/s × утилизация × 3600) × 10⁶. GPU, простаивающий ночью, стоит как днём; API ночью не стоит ничего.
• Ограничивает конкурентность KV-cache, а не веса. 8B в bf16 занимает 16 ГБ, а одна последовательность на 32K токенов у Llama-3-8B — ещё около 4 ГБ KV-cache.
• Скрытые расходы: пиковая нагрузка вдвое-втрое выше средней, минимум две реплики ради HA, дежурства и обновления. Один инженер на полставки легко стоит больше $8000 в месяц.
• Типичный рабочий вариант — гибрид: self-hosted модель 8–70B на массовую простую нагрузку, API на сложное и как fallback.

Подробно — движки (vLLM, SGLang, TGI, Ollama), PagedAttention и continuous batching, расчёт VRAM, калькулятор «self-hosted против API» и нагрузочный тест — в модуле 10.

Сборка: middleware LLM-клиента на Go

Сборка: middleware LLM-клиента на Java

Собираем всё в одну обёртку. Интерфейс один, реализаций-декораторов несколько, и порядок оборачивания задаёт семантику:

Traced → SemCache → Metered(квоты, $) → [Router] → Fallback[ для каждого звена: Breaker → Retry → Limiter → клиент провайдера ]

Трейс снаружи видит всё, включая попадания в кэш. Кэш стоит до квот, потому что попадание квоту не тратит. Роутер по сложности (урок про шлюз) — необязательный слой между квотами и fallback, у каждой его ветви своя цепочка. Ретраи и лимитер живут внутри звена: у каждого провайдера свои лимиты, и ретраить A, когда есть B, долго не нужно. Всё ниже, вместе с роутером, адаптером, флагами, SLO и PSI, собирается и проходит go vet и go test -race в examples/m09-prod репозитория курса.собирается и проходит mvn verify в examples/java/m09-prod репозитория курса (Java 21, OpenTelemetry API, Jackson, OpenFeature SDK).

package llmmw

import (
	"context"
	"errors"
	"math/rand/v2"
	"net"
	"net/http"
	"strconv"
	"sync"
	"time"

	"github.com/anthropics/anthropic-sdk-go"
)

type Request struct {
	Tenant, Model, System string
	Messages              []anthropic.MessageParam
	MaxTokens             int64
	EstInputTokens        int // для лимитера ITPM
}

type Response struct {
	Text, Model, StopReason, Route    string
	InputTokens, OutputTokens         int64
	CacheReadTokens, CacheWriteTokens int64
}

type LLM interface {
	Complete(ctx context.Context, req Request) (*Response, error)
}

// Retryable: повтор может закончиться иначе. Второе значение — retry-after от сервера.
func Retryable(err error) (bool, time.Duration) {
	if s, ra, ok := httpStatus(err); ok {
		switch {
		case s == 408, s == 409, s == 429, s >= 500: // 529 overloaded входит сюда
			return true, ra
		default: // 400/401/403/404/413: повтор даст то же самое
			return false, 0
		}
	}
	var ne net.Error
	return errors.As(err, &ne) || errors.Is(err, context.DeadlineExceeded), 0 // таймаут попытки
}

// httpStatus достаёт HTTP-статус из ошибки любого провайдера: Anthropic SDK или своего адаптера.
func httpStatus(err error) (status int, ra time.Duration, ok bool) {
	var apierr *anthropic.Error
	if errors.As(err, &apierr) {
		return apierr.StatusCode, retryAfter(apierr.Response), true
	}
	var herr *HTTPError
	if errors.As(err, &herr) {
		return herr.StatusCode, herr.RetryAfter, true
	}
	return 0, 0, false
}

func retryAfter(r *http.Response) time.Duration {
	if r == nil {
		return 0
	}
	sec, _ := strconv.Atoi(r.Header.Get("retry-after"))
	return time.Duration(sec) * time.Second
}

// Budget: ретраев не больше Ratio от числа запросов (как retry budget в Finagle/gRPC).
type Budget struct {
	mu            sync.Mutex
	tokens, Ratio float64
	Max           float64
}

func (b *Budget) onRequest() { b.mu.Lock(); b.tokens = min(b.Max, b.tokens+b.Ratio); b.mu.Unlock() }
func (b *Budget) tryRetry() bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.tokens < 1 {
		return false
	}
	b.tokens--
	return true
}

type Retry struct {
	Next       LLM
	MaxRetries int
	Base, Cap  time.Duration
	Budget     *Budget
}

func (r *Retry) Complete(ctx context.Context, req Request) (*Response, error) {
	r.Budget.onRequest()
	for attempt := 0; ; attempt++ {
		resp, err := r.Next.Complete(ctx, req)
		if err == nil || ctx.Err() != nil {
			return resp, err // родительский дедлайн истёк: не ретраим
		}
		ok, ra := Retryable(err)
		if !ok || attempt == r.MaxRetries || !r.Budget.tryRetry() {
			return nil, err
		}
		d := min(r.Cap, r.Base<<attempt)
		d = max(time.Duration(rand.Int64N(int64(d)+1)), ra) // full jitter, но не раньше retry-after
		if dl, ok := ctx.Deadline(); ok && time.Until(dl) < d {
			return nil, err // не успеем: пусть решает fallback выше
		}
		select {
		case <-ctx.Done():
			return nil, errors.Join(err, ctx.Err())
		case <-time.After(d):
		}
	}
}

// ===== LLM.java =====
package course.m09prod.llmmw;

import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.Optional;

/** Звено цепочки. Каждый слой (retry, breaker, limiter, trace) — декоратор над LLM. */
@FunctionalInterface
public interface LLM {
    Response complete(Request req) throws Exception;

    /** Текстовое сообщение диалога: role — "user" или "assistant". */
    record Msg(String role, String text) {
        public static Msg user(String text) {
            return new Msg("user", text);
        }
    }

    /**
     * deadline — бюджет всего пользовательского запроса: ретраи, лимитер и fallback
     * укладываются внутрь него. null — без дедлайна.
     */
    record Request(String tenant, String model, String system, List<Msg> messages,
                   long maxTokens, int estInputTokens, Instant deadline) {

        public Request withModel(String m) {
            return new Request(tenant, m, system, messages, maxTokens, estInputTokens, deadline);
        }

        public Request withSystem(String s) {
            return new Request(tenant, model, s, messages, maxTokens, estInputTokens, deadline);
        }

        /** Сколько осталось до дедлайна; пусто — дедлайна нет. */
        public Optional<Duration> timeLeft() {
            return Optional.ofNullable(deadline).map(d -> Duration.between(Instant.now(), d));
        }

        public boolean expired() {
            return timeLeft().map(d -> !d.isPositive()).orElse(false);
        }

        /** Текст последнего user-сообщения. */
        public String lastUserText() {
            for (int i = messages.size() - 1; i >= 0; i--) {
                if (messages.get(i).role().equals("user")) {
                    return messages.get(i).text();
                }
            }
            return "";
        }
    }

    record Response(String text, String model, String stopReason, String route,
                    long inputTokens, long outputTokens, long cacheReadTokens, long cacheWriteTokens) {

        public static Response of(String text, String model, String stopReason) {
            return new Response(text, model, stopReason, "", 0, 0, 0, 0);
        }

        public Response withRoute(String r) {
            return new Response(text, model, stopReason, r, inputTokens, outputTokens, cacheReadTokens, cacheWriteTokens);
        }
    }
}

// ===== Retry.java =====
package course.m09prod.llmmw;

import com.anthropic.errors.AnthropicIoException;
import com.anthropic.errors.AnthropicServiceException;
import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeoutException;

/** Ретраи с full jitter, уважением retry-after, retry budget и проверкой дедлайна. */
public record Retry(LLM next, int maxRetries, Duration base, Duration cap, Budget budget) implements LLM {

    /** retry — повтор может закончиться иначе; retryAfter — подсказка сервера. */
    public record Verdict(boolean retry, Duration retryAfter) {}

    record Status(int code, Duration retryAfter) {}

    public static Verdict retryable(Throwable e) {
        Status s = httpStatus(e);
        if (s != null) {
            int c = s.code();
            boolean ok = c == 408 || c == 409 || c == 429 || c >= 500; // 529 overloaded входит сюда
            return new Verdict(ok, ok ? s.retryAfter() : Duration.ZERO); // 400/401/403/404/413: повтор даст то же самое
        }
        boolean network = e instanceof IOException || e instanceof AnthropicIoException
                || e instanceof TimeoutException; // сеть или таймаут попытки
        return new Verdict(network, Duration.ZERO);
    }

    /** HTTP-статус из ошибки любого провайдера: Anthropic SDK или своего адаптера. null — не HTTP-ошибка. */
    static Status httpStatus(Throwable e) {
        for (Throwable t = e; t != null; t = t.getCause()) {
            if (t instanceof AnthropicServiceException a) {
                var ra = a.headers().values("retry-after");
                return new Status(a.statusCode(), retryAfter(ra.isEmpty() ? null : ra.getFirst()));
            }
            if (t instanceof OpenAiCompat.HttpError h) {
                return new Status(h.statusCode(), h.retryAfter());
            }
        }
        return null;
    }

    static boolean isRateLimit(Throwable e) {
        Status s = httpStatus(e);
        return s != null && s.code() == 429;
    }

    static Duration retryAfter(String header) {
        try {
            return header == null ? Duration.ZERO : Duration.ofSeconds(Long.parseLong(header.trim()));
        } catch (NumberFormatException e) {
            return Duration.ZERO; // HTTP-date и мусор не разбираем
        }
    }

    /** Ретраев не больше ratio от числа запросов (как retry budget в Finagle/gRPC). */
    public static final class Budget {
        private final double ratio, max;
        private double tokens;

        public Budget(double ratio, double max) {
            this.ratio = ratio;
            this.max = max;
        }

        synchronized void onRequest() {
            tokens = Math.min(max, tokens + ratio);
        }

        synchronized boolean tryRetry() {
            if (tokens < 1) {
                return false;
            }
            tokens--;
            return true;
        }
    }

    @Override
    public Response complete(Request req) throws Exception {
        budget.onRequest();
        for (int attempt = 0; ; attempt++) {
            try {
                return next.complete(req);
            } catch (Exception e) {
                if (req.expired()) {
                    throw e; // дедлайн запроса истёк: не ретраим
                }
                Verdict v = retryable(e);
                if (!v.retry() || attempt == maxRetries || !budget.tryRetry()) {
                    throw e;
                }
                long ceil = Math.min(cap.toNanos(), base.toNanos() << attempt);
                Duration jitter = Duration.ofNanos(ThreadLocalRandom.current().nextLong(ceil + 1));
                Duration d = jitter.compareTo(v.retryAfter()) < 0 ? v.retryAfter() : jitter; // full jitter, но не раньше retry-after
                if (req.timeLeft().map(left -> left.compareTo(d) < 0).orElse(false)) {
                    throw e; // не успеем: пусть решает fallback выше
                }
                Thread.sleep(d); // interrupt потока = отмена запроса
            }
        }
    }
}

package llmmw

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
)

// Breaker: closed → (Threshold ошибок подряд) → open → (Cooldown) → half-open (1 проба).
// В проде можно взять sony/gobreaker; логика та же.
type Breaker struct {
	mu        sync.Mutex
	fails     int
	openUntil time.Time
	probing   bool
	Threshold int
	Cooldown  time.Duration
}

func (b *Breaker) Allow() bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	switch {
	case b.fails < b.Threshold:
		return true // closed
	case time.Now().Before(b.openUntil), b.probing:
		return false // open, или проба уже в полёте
	default:
		b.probing = true // half-open: пропускаем ровно один запрос
		return true
	}
}

func (b *Breaker) Report(failed bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.probing = false
	if !failed {
		b.fails = 0
		return
	}
	if b.fails++; b.fails >= b.Threshold {
		b.openUntil = time.Now().Add(b.Cooldown)
	}
}

type Route struct {
	Name, Model string
	LLM         LLM // уже обёрнут в Retry и Limiter
	Breaker     *Breaker
	Fits        func(Request) bool    // влезет ли контекст, есть ли нужные фичи
	Adapt       func(Request) Request // свой промпт и параметры под модель
}

type Fallback struct{ Routes []Route }

func (f *Fallback) Complete(ctx context.Context, req Request) (*Response, error) {
	var errs []error
	for _, rt := range f.Routes {
		if rt.Fits != nil && !rt.Fits(req) {
			continue
		}
		if !rt.Breaker.Allow() {
			errs = append(errs, fmt.Errorf("%s: circuit open", rt.Name))
			continue
		}
		r := req
		r.Model = rt.Model
		if rt.Adapt != nil {
			r = rt.Adapt(r)
		}
		resp, err := rt.LLM.Complete(ctx, r)
		retryable, _ := Retryable(err)
		rt.Breaker.Report(err != nil && retryable && !isRateLimit(err)) // 429 — наша квота, не авария
		if err == nil {
			resp.Route = rt.Name
			return resp, nil
		}
		if !retryable {
			return nil, err // 400 на A будет 400 и на B: не маскируем баг перебором
		}
		errs = append(errs, fmt.Errorf("%s: %w", rt.Name, err))
		if ctx.Err() != nil {
			break
		}
	}
	return nil, errors.Join(errs...)
}

package course.m09prod.llmmw;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Predicate;
import java.util.function.UnaryOperator;

/** Цепочка звеньев: первое здоровое и подходящее отвечает. */
public record Fallback(List<Route> routes) implements LLM {

    /**
     * closed → (threshold ошибок подряд) → open → (cooldown) → half-open (1 проба).
     * В проде можно взять Resilience4j CircuitBreaker; логика та же.
     */
    public static final class Breaker {
        private final int threshold;
        private final Duration cooldown;
        private int fails;
        private Instant openUntil = Instant.MIN;
        private boolean probing;

        public Breaker(int threshold, Duration cooldown) {
            this.threshold = threshold;
            this.cooldown = cooldown;
        }

        public synchronized boolean allow() {
            if (fails < threshold) {
                return true; // closed
            }
            if (Instant.now().isBefore(openUntil) || probing) {
                return false; // open, или проба уже в полёте
            }
            probing = true; // half-open: пропускаем ровно один запрос
            return true;
        }

        public synchronized void report(boolean failed) {
            probing = false;
            if (!failed) {
                fails = 0;
                return;
            }
            if (++fails >= threshold) {
                openUntil = Instant.now().plus(cooldown);
            }
        }
    }

    /**
     * llm уже обёрнут в Retry и Limiter; fits — влезет ли контекст, есть ли нужные фичи;
     * adapt — свой промпт и параметры под модель.
     */
    public record Route(String name, String model, LLM llm, Breaker breaker,
                        Predicate<Request> fits, UnaryOperator<Request> adapt) {
        public Route(String name, String model, LLM llm, Breaker breaker) {
            this(name, model, llm, breaker, r -> true, UnaryOperator.identity());
        }
    }

    @Override
    public Response complete(Request req) throws Exception {
        List<Exception> errs = new ArrayList<>();
        for (Route rt : routes) {
            if (!rt.fits().test(req)) {
                continue;
            }
            if (!rt.breaker().allow()) {
                errs.add(new Exception(rt.name() + ": circuit open"));
                continue;
            }
            Request r = rt.adapt().apply(req.withModel(rt.model()));
            try {
                Response resp = rt.llm().complete(r);
                rt.breaker().report(false);
                return resp.withRoute(rt.name());
            } catch (Exception e) {
                boolean retryable = Retry.retryable(e).retry();
                rt.breaker().report(retryable && !Retry.isRateLimit(e)); // 429 — наша квота, не авария
                if (!retryable) {
                    throw e; // 400 на A будет 400 и на B: не маскируем баг перебором
                }
                errs.add(new Exception(rt.name() + ": " + e.getMessage(), e));
                if (req.expired()) {
                    break;
                }
            }
        }
        // cause — последняя ошибка: по ней retryable и error.type видят HTTP-статус.
        var all = new Exception("fallback: all routes failed", errs.isEmpty() ? null : errs.getLast());
        errs.forEach(all::addSuppressed);
        throw all;
    }
}

package llmmw

import (
	"context"
	"errors"

	"golang.org/x/time/rate"
)

// Limiter: три token bucket'а под RPM/ITPM/OTPM провайдера, выставленные на ~90% лимита.
type Limiter struct {
	Next            LLM
	RPM, ITPM, OTPM *rate.Limiter
}

func NewLimiter(next LLM, rpm, itpm, otpm float64) *Limiter {
	per := func(v float64) *rate.Limiter { return rate.NewLimiter(rate.Limit(v*0.9/60), int(v*0.9/6)) }
	return &Limiter{Next: next, RPM: per(rpm), ITPM: per(itpm), OTPM: per(otpm)}
}

func (l *Limiter) Complete(ctx context.Context, req Request) (*Response, error) {
	// WaitN сразу вернёт ошибку, если дедлайн ctx не дождётся токенов: бесплатный load shedding.
	if err := l.RPM.Wait(ctx); err != nil {
		return nil, err
	}
	if err := l.ITPM.WaitN(ctx, min(req.EstInputTokens, l.ITPM.Burst())); err != nil {
		return nil, err
	}
	if err := l.OTPM.WaitN(ctx, min(int(req.MaxTokens), l.OTPM.Burst())); err != nil {
		return nil, err
	}
	return l.Next.Complete(ctx, req)
}

// Цены $ за 1M токенов на момент написания (сентябрь 2026).
var prices = map[string]struct{ In, Out float64 }{
	"claude-opus-5":    {5, 25},
	"claude-sonnet-5":  {2, 10},
	"claude-haiku-4-5": {1, 5},
}

func CostUSD(r *Response) float64 {
	p, ok := prices[r.Model]
	if !ok {
		return 0 // неизвестная модель: алертим отдельно, а не падаем
	}
	in := float64(r.InputTokens) + 0.1*float64(r.CacheReadTokens) + 1.25*float64(r.CacheWriteTokens)
	return (in*p.In + float64(r.OutputTokens)*p.Out) / 1e6
}

var ErrQuota = errors.New("llm: tenant budget exceeded")

// Metered: проверка квоты тенанта до вызова и учёт фактической стоимости после.
type Metered struct {
	Next  LLM
	Spent func(ctx context.Context, tenant string) (usd float64, err error) // напр. Redis GET
	Add   func(ctx context.Context, tenant string, usd float64) error       // INCRBYFLOAT + EXPIRE
	Limit func(tenant string) float64
}

func (m *Metered) Complete(ctx context.Context, req Request) (*Response, error) {
	spent, err := m.Spent(ctx, req.Tenant)
	if err == nil && spent >= m.Limit(req.Tenant) {
		return nil, ErrQuota // при ошибке Redis — fail-open: квота мягкая, логируем
	}
	resp, err := m.Next.Complete(ctx, req)
	if err != nil {
		return nil, err
	}
	_ = m.Add(ctx, req.Tenant, CostUSD(resp))
	return resp, nil
}

// ===== Limiter.java =====
package course.m09prod.llmmw;

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.RejectedExecutionException;

/** Три token bucket'а под RPM/ITPM/OTPM провайдера, выставленные на ~90% лимита. */
public record Limiter(LLM next, TokenBucket rpm, TokenBucket itpm, TokenBucket otpm) implements LLM {

    public Limiter(LLM next, double rpm, double itpm, double otpm) {
        this(next, per(rpm), per(itpm), per(otpm));
    }

    private static TokenBucket per(double perMinute) {
        return new TokenBucket(perMinute * 0.9 / 60, perMinute * 0.9 / 6);
    }

    @Override
    public Response complete(Request req) throws Exception {
        // acquire сразу бросит исключение, если дедлайн запроса не дождётся токенов: бесплатный load shedding.
        rpm.acquire(1, req.deadline());
        itpm.acquire(Math.min(req.estInputTokens(), itpm.burst()), req.deadline());
        otpm.acquire(Math.min(req.maxTokens(), otpm.burst()), req.deadline());
        return next.complete(req);
    }

    /** Token bucket с резервированием: токены можно «занять» вперёд и подождать их. */
    public static final class TokenBucket {
        private final double perSecond, burst;
        private double tokens;
        private long last = System.nanoTime();

        public TokenBucket(double perSecond, double burst) {
            this.perSecond = perSecond;
            this.burst = burst;
            this.tokens = burst;
        }

        public double burst() {
            return burst;
        }

        public void acquire(double n, Instant deadline) throws InterruptedException {
            Duration wait;
            synchronized (this) {
                long now = System.nanoTime();
                tokens = Math.min(burst, tokens + (now - last) / 1e9 * perSecond);
                last = now;
                wait = tokens >= n ? Duration.ZERO : Duration.ofNanos((long) ((n - tokens) / perSecond * 1e9));
                if (deadline != null && Instant.now().plus(wait).isAfter(deadline)) {
                    throw new RejectedExecutionException("limiter: wait would exceed deadline");
                }
                tokens -= n;
            }
            if (wait.isPositive()) {
                Thread.sleep(wait);
            }
        }
    }
}

// ===== Metered.java =====
package course.m09prod.llmmw;

import java.util.Map;
import java.util.function.ToDoubleFunction;

/** Проверка квоты тенанта до вызова и учёт фактической стоимости после. */
public record Metered(LLM next, Spent spent, Add add, ToDoubleFunction<String> limit) implements LLM {

    @FunctionalInterface
    public interface Spent {
        double get(String tenant) throws Exception; // напр. Redis GET
    }

    @FunctionalInterface
    public interface Add {
        void add(String tenant, double usd) throws Exception; // INCRBYFLOAT + EXPIRE
    }

    public static final class QuotaExceededException extends RuntimeException {
        public QuotaExceededException() {
            super("llm: tenant budget exceeded");
        }
    }

    record Price(double in, double out) {}

    // Цены $ за 1M токенов на момент написания (сентябрь 2026).
    static final Map<String, Price> PRICES = Map.of(
            "claude-opus-5", new Price(5, 25),
            "claude-sonnet-5", new Price(2, 10),
            "claude-haiku-4-5", new Price(1, 5));

    public static double costUsd(Response r) {
        Price p = PRICES.get(r.model());
        if (p == null) {
            return 0; // неизвестная модель: алертим отдельно, а не падаем
        }
        double in = r.inputTokens() + 0.1 * r.cacheReadTokens() + 1.25 * r.cacheWriteTokens();
        return (in * p.in() + r.outputTokens() * p.out()) / 1e6;
    }

    @Override
    public Response complete(Request req) throws Exception {
        double used;
        try {
            used = spent.get(req.tenant());
        } catch (Exception e) {
            used = 0; // при ошибке Redis — fail-open: квота мягкая, логируем
        }
        if (used >= limit.applyAsDouble(req.tenant())) {
            throw new QuotaExceededException();
        }
        Response resp = next.complete(req);
        try {
            add.add(req.tenant(), costUsd(resp));
        } catch (Exception ignored) {
            // учёт не должен ронять уже оплаченный ответ
        }
        return resp;
    }
}

package llmmw

import (
	"context"
	"strconv"
	"time"

	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/metric"
	"go.opentelemetry.io/otel/trace"
)

var (
	tracer = otel.Tracer("llmmw")
	meter  = otel.Meter("llmmw")
	// Границы бакетов — рекомендованные конвенцией: токены растут степенями 4, секунды — удвоением.
	tokenUsage, _ = meter.Int64Histogram("gen_ai.client.token.usage", metric.WithUnit("{token}"),
		metric.WithExplicitBucketBoundaries(1, 4, 16, 64, 256, 1024, 4096, 16384, 65536, 262144, 1048576, 4194304, 16777216, 67108864))
	opDuration, _ = meter.Float64Histogram("gen_ai.client.operation.duration", metric.WithUnit("s"),
		metric.WithExplicitBucketBoundaries(0.01, 0.02, 0.04, 0.08, 0.16, 0.32, 0.64, 1.28, 2.56, 5.12, 10.24, 20.48, 40.96, 81.92))
)

type Traced struct {
	Next     LLM
	Provider string // "anthropic"
}

func (t *Traced) Complete(ctx context.Context, req Request) (*Response, error) {
	start := time.Now()
	ctx, span := tracer.Start(ctx, "chat "+req.Model,
		trace.WithSpanKind(trace.SpanKindClient),
		trace.WithAttributes(
			attribute.String("gen_ai.operation.name", "chat"),
			attribute.String("gen_ai.provider.name", t.Provider),
			attribute.String("gen_ai.request.model", req.Model),
			attribute.Int64("gen_ai.request.max_tokens", req.MaxTokens),
			attribute.String("app.tenant", req.Tenant), // свой namespace для не-конвенционных полей
		))
	defer span.End()

	common := []attribute.KeyValue{
		attribute.String("gen_ai.operation.name", "chat"),
		attribute.String("gen_ai.provider.name", t.Provider),
		attribute.String("gen_ai.request.model", req.Model),
	}
	resp, err := t.Next.Complete(ctx, req)
	if err != nil {
		span.RecordError(err)
		span.SetStatus(codes.Error, "llm call failed")
		span.SetAttributes(attribute.String("error.type", errorType(err)))
		opDuration.Record(ctx, time.Since(start).Seconds(),
			metric.WithAttributes(append(common, attribute.String("error.type", errorType(err)))...))
		return nil, err
	}
	span.SetAttributes(
		attribute.String("gen_ai.response.model", resp.Model),
		attribute.StringSlice("gen_ai.response.finish_reasons", []string{resp.StopReason}),
		attribute.Int64("gen_ai.usage.input_tokens", resp.InputTokens),
		attribute.Int64("gen_ai.usage.output_tokens", resp.OutputTokens),
		attribute.Int64("gen_ai.usage.cache_read.input_tokens", resp.CacheReadTokens),
		attribute.Int64("gen_ai.usage.cache_creation.input_tokens", resp.CacheWriteTokens),
		attribute.String("app.llm.route", resp.Route),
		attribute.Float64("app.llm.cost_usd", CostUSD(resp)),
	)
	common = append(common, attribute.String("gen_ai.response.model", resp.Model))
	opDuration.Record(ctx, time.Since(start).Seconds(), metric.WithAttributes(common...))
	tokenUsage.Record(ctx, resp.InputTokens, metric.WithAttributes(append(common, attribute.String("gen_ai.token.type", "input"))...))
	tokenUsage.Record(ctx, resp.OutputTokens, metric.WithAttributes(append(common, attribute.String("gen_ai.token.type", "output"))...))
	// Текст промпта и ответа в атрибуты НЕ пишем: только opt-in, с маскированием и сэмплированием.
	return resp, nil
}

func errorType(err error) string {
	if s, _, ok := httpStatus(err); ok {
		return strconv.Itoa(s)
	}
	return "_OTHER" // значение по конвенции OTel для неклассифицированных ошибок
}

// GenAi (трейсер и гистограммы gen_ai.client.*) — в уроке «Спаны агента и TTFT в коде».
package course.m09prod.llmmw;

import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.SpanKind;
import io.opentelemetry.context.Scope;
import java.util.List;
import java.util.Objects;

/** Спан chat {model} и метрики gen_ai.client.* вокруг всей цепочки. provider — "anthropic". */
public record Traced(LLM next, String provider, GenAi otel) implements LLM {

    @Override
    public Response complete(Request req) throws Exception {
        long start = System.nanoTime();
        Span span = otel.tracer.spanBuilder("chat " + req.model())
                .setSpanKind(SpanKind.CLIENT)
                .setAttribute("gen_ai.operation.name", "chat")
                .setAttribute("gen_ai.provider.name", provider)
                .setAttribute("gen_ai.request.model", req.model())
                .setAttribute("gen_ai.request.max_tokens", req.maxTokens())
                .setAttribute("app.tenant", req.tenant()) // свой namespace для не-конвенционных полей
                .startSpan();
        var common = Attributes.builder()
                .put("gen_ai.operation.name", "chat")
                .put("gen_ai.provider.name", provider)
                .put("gen_ai.request.model", req.model());
        try (Scope ignored = span.makeCurrent()) {
            Response resp = next.complete(req);
            span.setAllAttributes(Attributes.builder()
                    .put("gen_ai.response.model", resp.model())
                    .put(AttributeKey.stringArrayKey("gen_ai.response.finish_reasons"),
                            List.of(Objects.toString(resp.stopReason(), "")))
                    .put("gen_ai.usage.input_tokens", resp.inputTokens())
                    .put("gen_ai.usage.output_tokens", resp.outputTokens())
                    .put("gen_ai.usage.cache_read.input_tokens", resp.cacheReadTokens())
                    .put("gen_ai.usage.cache_creation.input_tokens", resp.cacheWriteTokens())
                    .put("app.llm.route", resp.route())
                    .put("app.llm.cost_usd", Metered.costUsd(resp))
                    .build());
            Attributes attrs = common.put("gen_ai.response.model", resp.model()).build();
            otel.opDuration.record(seconds(start), attrs);
            otel.tokenUsage.record(resp.inputTokens(), attrs.toBuilder().put("gen_ai.token.type", "input").build());
            otel.tokenUsage.record(resp.outputTokens(), attrs.toBuilder().put("gen_ai.token.type", "output").build());
            // Текст промпта и ответа в атрибуты НЕ пишем: только opt-in, с маскированием и сэмплированием.
            return resp;
        } catch (Exception e) {
            GenAi.fail(span, e, "llm call failed");
            otel.opDuration.record(seconds(start), common.put("error.type", GenAi.errorType(e)).build());
            throw e;
        } finally {
            span.end();
        }
    }

    private static double seconds(long startNanos) {
        return (System.nanoTime() - startNanos) / 1e9;
    }
}

package llmmw

// CREATE EXTENSION IF NOT EXISTS vector;
// CREATE TABLE llm_cache (
//   id bigserial PRIMARY KEY,
//   scope text NOT NULL,              -- hash(tenant, model, system prompt, версия промпта)
//   query text NOT NULL,
//   embedding vector(1024) NOT NULL,
//   response text NOT NULL,
//   created_at timestamptz NOT NULL DEFAULT now());
// CREATE INDEX ON llm_cache USING hnsw (embedding vector_cosine_ops);
// CREATE INDEX ON llm_cache (scope, created_at);
// Фильтр по scope + HNSW: в pgvector ≥ 0.8 включите SET hnsw.iterative_scan = relaxed_order,
// иначе после фильтрации может не остаться кандидатов. Пул: pgxvec.RegisterTypes в AfterConnect.

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"log/slog"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/pgvector/pgvector-go"
)

type SemCache struct {
	Next      LLM
	DB        *pgxpool.Pool
	Embed     func(ctx context.Context, text string) ([]float32, error)
	Threshold float64 // cosine similarity; калибруется на размеченных парах
	TTL       time.Duration
}

func (c *SemCache) Complete(ctx context.Context, req Request) (*Response, error) {
	q := lastUserText(req)
	if q == "" || len(req.Messages) > 1 { // кэшируем только первый вопрос без истории
		return c.Next.Complete(ctx, req)
	}
	emb, err := c.Embed(ctx, q)
	if err != nil {
		return c.Next.Complete(ctx, req) // кэш — оптимизация, а не точка отказа
	}
	vec, scope := pgvector.NewVector(emb), scopeKey(req)

	var text string
	var sim float64
	err = c.DB.QueryRow(ctx, `
		SELECT response, 1 - (embedding <=> $1) FROM llm_cache
		WHERE scope = $2 AND created_at > now() - make_interval(secs => $3)
		ORDER BY embedding <=> $1 LIMIT 1`, vec, scope, c.TTL.Seconds()).Scan(&text, &sim)
	switch {
	case err == nil && sim >= c.Threshold:
		return &Response{Text: text, Model: req.Model, StopReason: "end_turn", Route: "semcache"}, nil
	case err != nil && !errors.Is(err, pgx.ErrNoRows):
		slog.WarnContext(ctx, "semcache lookup", "err", err)
	}

	resp, err := c.Next.Complete(ctx, req)
	if err != nil {
		return nil, err
	}
	if resp.StopReason == "end_turn" { // не кэшируем обрезанные ответы и refusal
		if _, err := c.DB.Exec(ctx, `INSERT INTO llm_cache (scope, query, embedding, response)
			VALUES ($1, $2, $3, $4)`, scope, q, vec, resp.Text); err != nil {
			slog.WarnContext(ctx, "semcache store", "err", err)
		}
	}
	return resp, nil
}

func scopeKey(r Request) string {
	h := sha256.Sum256([]byte(r.Tenant + "\x00" + r.Model + "\x00" + r.System))
	return hex.EncodeToString(h[:])
}

package course.m09prod.llmmw;

// CREATE EXTENSION IF NOT EXISTS vector;
// CREATE TABLE llm_cache (
//   id bigserial PRIMARY KEY,
//   scope text NOT NULL,              -- hash(tenant, model, system prompt, версия промпта)
//   query text NOT NULL,
//   embedding vector(1024) NOT NULL,
//   response text NOT NULL,
//   created_at timestamptz NOT NULL DEFAULT now());
// CREATE INDEX ON llm_cache USING hnsw (embedding vector_cosine_ops);
// CREATE INDEX ON llm_cache (scope, created_at);
// Фильтр по scope + HNSW: в pgvector ≥ 0.8 включите SET hnsw.iterative_scan = relaxed_order,
// иначе после фильтрации может не остаться кандидатов. Вектор передаём текстом '[...]'::vector,
// чтобы не тянуть зависимость; с com.pgvector:pgvector — setObject(i, new PGvector(emb)).

import java.lang.System.Logger;
import java.lang.System.Logger.Level;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.sql.SQLException;
import java.time.Duration;
import java.util.HexFormat;
import javax.sql.DataSource;

/** threshold — cosine similarity, калибруется на размеченных парах. */
public record SemCache(LLM next, DataSource db, Embedder embed, double threshold, Duration ttl) implements LLM {

    @FunctionalInterface
    public interface Embedder {
        float[] embed(String text) throws Exception;
    }

    private static final Logger LOG = System.getLogger(SemCache.class.getName());

    @Override
    public Response complete(Request req) throws Exception {
        String q = req.lastUserText();
        if (q.isEmpty() || req.messages().size() > 1) { // кэшируем только первый вопрос без истории
            return next.complete(req);
        }
        float[] emb;
        try {
            emb = embed.embed(q);
        } catch (Exception e) {
            return next.complete(req); // кэш — оптимизация, а не точка отказа
        }
        String vec = pgVector(emb), scope = scopeKey(req);

        try (var c = db.getConnection(); var st = c.prepareStatement("""
                SELECT response, 1 - (embedding <=> ?::vector) FROM llm_cache
                WHERE scope = ? AND created_at > now() - make_interval(secs => ?)
                ORDER BY embedding <=> ?::vector LIMIT 1""")) {
            st.setString(1, vec);
            st.setString(2, scope);
            st.setDouble(3, ttl.toSeconds());
            st.setString(4, vec);
            try (var rs = st.executeQuery()) {
                if (rs.next() && rs.getDouble(2) >= threshold) {
                    return Response.of(rs.getString(1), req.model(), "end_turn").withRoute("semcache");
                }
            }
        } catch (SQLException e) {
            LOG.log(Level.WARNING, "semcache lookup", e);
        }

        Response resp = next.complete(req);
        if ("end_turn".equals(resp.stopReason())) { // не кэшируем обрезанные ответы и refusal
            try (var c = db.getConnection(); var st = c.prepareStatement(
                    "INSERT INTO llm_cache (scope, query, embedding, response) VALUES (?, ?, ?::vector, ?)")) {
                st.setString(1, scope);
                st.setString(2, q);
                st.setString(3, vec);
                st.setString(4, resp.text());
                st.executeUpdate();
            } catch (SQLException e) {
                LOG.log(Level.WARNING, "semcache store", e);
            }
        }
        return resp;
    }

    static String pgVector(float[] v) {
        var sb = new StringBuilder("[");
        for (int i = 0; i < v.length; i++) {
            sb.append(i == 0 ? "" : ",").append(v[i]);
        }
        return sb.append(']').toString();
    }

    static String scopeKey(Request r) throws Exception {
        byte[] h = MessageDigest.getInstance("SHA-256")
                .digest((r.tenant() + "\0" + r.model() + "\0" + r.system()).getBytes(StandardCharsets.UTF_8));
        return HexFormat.of().formatHex(h);
    }
}

package llmmw

import (
	"context"
	"strings"
	"time"

	"github.com/anthropics/anthropic-sdk-go"
	"github.com/anthropics/anthropic-sdk-go/option"
)

type Func func(ctx context.Context, req Request) (*Response, error)

func (f Func) Complete(ctx context.Context, req Request) (*Response, error) { return f(ctx, req) }

// Anthropic — нижний слой. Ретраи SDK выключены: они живут в нашем Retry.
func Anthropic() LLM {
	c := anthropic.NewClient(option.WithMaxRetries(0))
	return Func(func(ctx context.Context, req Request) (*Response, error) {
		msg, err := c.Messages.New(ctx, anthropic.MessageNewParams{
			Model:     anthropic.Model(req.Model),
			MaxTokens: req.MaxTokens,
			System:    []anthropic.TextBlockParam{{Text: req.System}},
			Messages:  req.Messages,
		})
		if err != nil {
			return nil, err
		}
		out := &Response{
			Model: string(msg.Model), StopReason: string(msg.StopReason),
			InputTokens: msg.Usage.InputTokens, OutputTokens: msg.Usage.OutputTokens,
			CacheReadTokens:  msg.Usage.CacheReadInputTokens,
			CacheWriteTokens: msg.Usage.CacheCreationInputTokens,
		}
		for _, b := range msg.Content {
			if t, ok := b.AsAny().(anthropic.TextBlock); ok {
				out.Text += t.Text
			}
		}
		return out, nil
	})
}

func isRateLimit(err error) bool {
	s, _, ok := httpStatus(err)
	return ok && s == 429
}

// lastUserText — текст последнего user-сообщения.
func lastUserText(r Request) string {
	for i := len(r.Messages) - 1; i >= 0; i-- {
		if r.Messages[i].Role == anthropic.MessageParamRoleUser {
			return textOf(r.Messages[i])
		}
	}
	return ""
}

// textOf склеивает текстовые блоки сообщения.
func textOf(m anthropic.MessageParam) string {
	var sb strings.Builder
	for _, b := range m.Content {
		if b.OfText != nil {
			sb.WriteString(b.OfText.Text)
		}
	}
	return sb.String()
}

func New(selfHosted LLM, cache *SemCache, meter *Metered) LLM {
	api := Anthropic()
	budget := &Budget{Ratio: 0.2, Max: 50}
	route := func(name, model string, base LLM, rpm, itpm, otpm float64) Route {
		return Route{
			Name: name, Model: model,
			Breaker: &Breaker{Threshold: 5, Cooldown: 30 * time.Second},
			LLM: &Retry{MaxRetries: 2, Base: 250 * time.Millisecond, Cap: 4 * time.Second, Budget: budget,
				Next: NewLimiter(base, rpm, itpm, otpm)},
		}
	}
	fb := &Fallback{Routes: []Route{
		route("primary", "claude-opus-5", api, 4000, 2_000_000, 400_000),
		route("smaller", "claude-sonnet-5", api, 4000, 2_000_000, 400_000),
		route("self-hosted", "qwen3-8b", selfHosted, 600, 1_000_000, 200_000),
	}}
	meter.Next = fb
	cache.Next = meter
	return &Traced{Provider: "anthropic", Next: cache}
}

package course.m09prod.llmmw;

import com.anthropic.client.AnthropicClient;
import com.anthropic.client.okhttp.AnthropicOkHttpClient;
import com.anthropic.core.RequestOptions;
import com.anthropic.models.messages.Message;
import com.anthropic.models.messages.MessageCreateParams;
import com.anthropic.models.messages.StopReason;
import java.time.Duration;
import java.util.List;
import java.util.function.UnaryOperator;

public final class Providers {
    private Providers() {}

    /** Нижний слой. Ретраи SDK выключены: они живут в нашем Retry. */
    public static LLM anthropic() {
        AnthropicClient c = AnthropicOkHttpClient.builder().fromEnv().maxRetries(0).build();
        return req -> {
            var p = MessageCreateParams.builder()
                    .model(req.model())
                    .maxTokens(req.maxTokens())
                    .system(req.system());
            for (LLM.Msg m : req.messages()) {
                if (m.role().equals("assistant")) {
                    p.addAssistantMessage(m.text());
                } else {
                    p.addUserMessage(m.text());
                }
            }
            var opts = RequestOptions.builder();
            req.timeLeft().ifPresent(opts::timeout); // дедлайн запроса — таймаут HTTP-вызова
            Message msg = c.messages().create(p.build(), opts.build());
            var text = new StringBuilder();
            msg.content().forEach(b -> b.text().ifPresent(t -> text.append(t.text())));
            var u = msg.usage();
            return new LLM.Response(text.toString(), msg.model().asString(),
                    msg.stopReason().map(StopReason::asString).orElse(""), "",
                    u.inputTokens(), u.outputTokens(),
                    u.cacheReadInputTokens().orElse(0L), u.cacheCreationInputTokens().orElse(0L));
        };
    }

    /**
     * Сборка: Traced → SemCache → Metered → Fallback[Breaker → Retry → Limiter → провайдер].
     * cache и meter получают следующее звено: next -> new SemCache(next, ds, embed, 0.92, ttl).
     */
    public static LLM chain(LLM selfHosted, UnaryOperator<LLM> cache, UnaryOperator<LLM> meter, GenAi otel) {
        LLM api = anthropic();
        var budget = new Retry.Budget(0.2, 50);
        var fb = new Fallback(List.of(
                route("primary", "claude-opus-5", api, 4000, 2_000_000, 400_000, budget),
                route("smaller", "claude-sonnet-5", api, 4000, 2_000_000, 400_000, budget),
                route("self-hosted", "qwen3-8b", selfHosted, 600, 1_000_000, 200_000, budget)));
        return new Traced(cache.apply(meter.apply(fb)), "anthropic", otel);
    }

    private static Fallback.Route route(String name, String model, LLM base,
                                        double rpm, double itpm, double otpm, Retry.Budget budget) {
        return new Fallback.Route(name, model,
                new Retry(new Limiter(base, rpm, itpm, otpm), 2, Duration.ofMillis(250), Duration.ofSeconds(4), budget),
                new Fallback.Breaker(5, Duration.ofSeconds(30)));
    }
}

Практика

Самопроверка

Вопросы с собеседований

Источники

• AWS Architecture Blog: Exponential Backoff and Jitter
• Google SRE Book: Handling Overload и Addressing Cascading Failures (retry budget, load shedding)
• Anthropic: Errors, Rate limits, Prompt caching
• anthropic-sdk-go
• anthropic-sdk-java
• OpenTelemetry GenAI semantic conventions v1.41.1: спаны, спаны агентов, метрики; новый репозиторий semantic-conventions-genai
• Google SRE Workbook: Alerting on SLOs (burn rate, multiwindow multi-burn-rate алерты); генераторы правил Sloth и Pyrra
• OpenFeature Go SDK и справочник API
• OpenFeature Java SDK
• Шлюзы: LiteLLM Proxy, Agent Router (бывший Envoy AI Gateway), Kong AI Gateway, Bifrost, APISIX ai-proxy-multi, Portkey Gateway, Cloudflare AI Gateway, OpenRouter: fallback-модели
• Prompt caching у других провайдеров: OpenAI, Gemini API
• golang.org/x/time/rate, sony/gobreaker
• Resilience4j (CircuitBreaker, Retry, RateLimiter), Bucket4j
• pgvector и pgvector-go
• pgvector и pgvector-java
• Langfuse docs, LangSmith docs
• OWASP Top 10 for LLM Applications
• Inan et al., 2023. Llama Guard, Presidio (PII)