Чем AI engineering отличается от бэкенда

Прикладной AI engineer строит продукты поверх готовых LLM (large language model — большая языковая модель, которую вызывают по API). Модель вы не обучаете. Вы проектируете систему вокруг неё: как собрать контекст, как вызвать модель, как проверить ответ, как это масштабировать и во сколько это обойдётся. Инфраструктурная часть знакома. Новое — четыре свойства «зависимости», которой в обычном бэкенде не бывает.

Отсюда главное отличие в мышлении. В обычном сервисе ошибка — исключение, и вы её ловите. В LLM-системе «ошибка» — это часть распределения: при 95% качества каждый двадцатый ответ плохой по определению. Архитектура должна это переживать: валидировать выход, давать пользователю поправить, откатываться на человека, мерить долю плохих ответов в проде.

Карта переноса навыков

Ниже — сопоставление того, что вы уже делаете в Go-сервисах, с практиками AI engineering. Кликните на навык слева, чтобы увидеть, во что он превращается, или на практику справа, чтобы увидеть, из чего она растёт. Связи «многие ко многим»: например, context участвует и в параллельных вызовах, и в стриминге.

Ниже — сопоставление того, что вы уже делаете в Java-сервисах, с практиками AI engineering. Кликните на навык слева, чтобы увидеть, во что он превращается, или на практику справа, чтобы увидеть, из чего она растёт. Связи «многие ко многим»: например, отмена и таймауты участвуют и в параллельных вызовах, и в стриминге.

Конкурентность → параллельные вызовы

Самый прямой перенос. LLM-вызов — это I/O на секунды, CPU почти не занят. Горутины идеальны: 200 одновременных вызовов стоят несколько мегабайт памяти. Типичные задачи:

Самый прямой перенос. LLM-вызов — это I/O на секунды, CPU почти не занят. Virtual threads из Java 21 идеальны: 200 одновременных блокирующих вызовов стоят несколько мегабайт памяти, без реактивщины и без пула на 200 платформенных потоков. Типичные задачи:

• Fan-out/fan-in. Суммаризировать 500 документов, классифицировать 10 000 тикетов, прогнать eval-набор. Каждый элемент независим, результат собирается в срезмассив по индексу.
• Parallelization-паттерн из статьи Anthropic «Building effective agents». Sectioning: разные аспекты одной задачи (проверка на безопасность и основной ответ) параллельно. Voting: один и тот же вопрос N раз, берём большинство — дешёвый способ поднять надёжность классификации.
• Параллельные tool calls. Модель может вернуть несколько блоков tool_use в одном ответе. Их выполняют конкурентно и возвращают все tool_result одним user-сообщением.
• Hedging. Если запрос не ответил за p95, запустить второй и взять тот, что придёт первым. Работает против тяжёлых хвостов латентности, но удваивает стоимость хвоста.

Что меняется по сравнению с обычным fan-out на HTTP-сервисы:

• Лимит конкурентности задаёт не ваш CPU, а rate limit провайдера. У Anthropic три лимита на организацию и модель: RPM (запросы в минуту), ITPM (входные токены в минуту) и OTPM (выходные токены в минуту). Выбить ITPM проще, чем RPM: 50 параллельных запросов по 30K токенов контекста — это 1.5M входных токенов за несколько секунд.
• Длинные хвосты латентности. Время ответа пропорционально длине вывода, а длину вывода вы контролируете лишь частично. При fan-out из N вызовов общее время определяет самый медленный — классический «tail at scale».
• Частичный успех часто лучше полного провала. errgroup по умолчанию отменяет всё при первой ошибке.StructuredTaskScope.ShutdownOnFailure и самописный fail-fast поверх CompletableFuture отменяют всё при первой ошибке. Для батча из 500 документов это неправильно: лучше собрать 497 результатов и 3 ошибки, чем потерять уже оплаченные 497 ответов.

Что видно на модели. При K ≥ N общее время равно максимальной длительности, а не сумме: 12 вызовов по 4 с укладываются примерно в 6–8 с вместо ~48. Увеличение разброса бьёт именно по параллельному варианту: он ждёт самый медленный вызов. При K < N время примерно ⌈N/K⌉ × средняя латентность, и выгода от K растёт почти линейно до упора в rate limit провайдера. В проде K подбирают так, чтобы K × (токены на запрос) × (запросов в минуту на один слот) укладывалось в ITPM с запасом.

Код: errgroup + semaphore + context

Код: virtual threads + Semaphore + дедлайн

Классическая схема: общий дедлайн на операцию, отдельный таймаут на вызов, ограничение одновременных вызовов, запись результата по индексу (без мьютекса, каждая горутина пишет в свой элемент).

Классическая схема: общий дедлайн на операцию, отдельный таймаут на вызов, ограничение одновременных вызовов, запись результата по индексу (без блокировок, каждый поток пишет в свой элемент массива). ExecutorCompletionService отдаёт задачи в порядке завершения, поэтому первая же ошибка отменяет остальные, а не ждёт своей очереди.

package fanout;

import com.anthropic.client.AnthropicClient;
import com.anthropic.core.RequestOptions;
import com.anthropic.models.messages.Message;
import com.anthropic.models.messages.MessageCreateParams;
import com.anthropic.models.messages.TextBlock;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public final class FanOut {
    public record Result(String text, long inputTokens, long outputTokens) {}

    private static final RequestOptions PER_CALL = RequestOptions.builder()
            .timeout(Duration.ofSeconds(45)) // таймаут одного вызова
            .build();

    /**
     * Делает docs.size() вызовов, не более maxInFlight одновременно.
     * Fail-fast: первая ошибка прерывает остальные вызовы.
     */
    public static List<Result> summarize(AnthropicClient client, List<String> docs, int maxInFlight)
            throws InterruptedException, ExecutionException, TimeoutException {
        long deadline = System.nanoTime() + Duration.ofMinutes(2).toNanos(); // бюджет на всю операцию
        Semaphore sem = new Semaphore(maxInFlight);
        Result[] results = new Result[docs.size()]; // каждый поток пишет только в свой индекс

        try (ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor()) {
            var done = new ExecutorCompletionService<Void>(pool);
            List<Future<Void>> tasks = new ArrayList<>();
            for (int i = 0; i < docs.size(); i++) {
                int idx = i;
                tasks.add(done.submit(() -> {
                    sem.acquire(); // ждущий виртуальный поток почти ничего не стоит
                    try {
                        results[idx] = summarizeOne(client, docs.get(idx));
                    } catch (RuntimeException e) {
                        throw new RuntimeException("doc " + idx, e);
                    } finally {
                        sem.release();
                    }
                    return null;
                }));
            }
            try {
                for (int n = 0; n < tasks.size(); n++) {
                    Future<Void> f = done.poll(deadline - System.nanoTime(), TimeUnit.NANOSECONDS);
                    if (f == null) {
                        throw new TimeoutException("fan-out aborted: deadline exceeded");
                    }
                    f.get(); // ExecutionException первой упавшей задачи
                }
            } catch (ExecutionException | TimeoutException | InterruptedException e) {
                tasks.forEach(t -> t.cancel(true)); // прерываем ждущие и летящие вызовы
                throw e;
            }
        }
        return List.of(results);
    }

    static Result summarizeOne(AnthropicClient client, String doc) {
        Message resp = client.messages().create(MessageCreateParams.builder()
                .model("claude-opus-5")
                .maxTokens(1024)
                .addUserMessage("Суммаризируй в 3 пунктах:\n\n" + doc)
                .build(), PER_CALL);
        String text = resp.content().stream()
                .flatMap(b -> b.text().stream())
                .map(TextBlock::text)
                .collect(Collectors.joining());
        return new Result(text, resp.usage().inputTokens(), resp.usage().outputTokens());
    }
}

package fanout

import (
	"context"
	"fmt"
	"time"

	"github.com/anthropics/anthropic-sdk-go"
	"golang.org/x/sync/errgroup"
	"golang.org/x/sync/semaphore"
)

type Result struct {
	Text          string
	InputTokens   int64
	OutputTokens  int64
}

// Summarize делает len(docs) вызовов, не более maxInFlight одновременно.
// Fail-fast: первая ошибка отменяет остальные вызовы через ctx.
func Summarize(ctx context.Context, client *anthropic.Client, docs []string, maxInFlight int64) ([]Result, error) {
	ctx, cancel := context.WithTimeout(ctx, 2*time.Minute) // бюджет на всю операцию
	defer cancel()

	sem := semaphore.NewWeighted(maxInFlight)
	g, ctx := errgroup.WithContext(ctx)
	results := make([]Result, len(docs)) // каждая горутина пишет только в свой индекс
	var acquireErr error

	for i, doc := range docs {
		// Acquire блокирует цикл, а не плодит тысячи ждущих горутин.
		if err := sem.Acquire(ctx, 1); err != nil {
			acquireErr = err // ctx отменён: таймаут или чья-то ошибка
			break
		}
		g.Go(func() error {
			defer sem.Release(1)
			callCtx, cancel := context.WithTimeout(ctx, 45*time.Second) // таймаут одного вызова
			defer cancel()

			resp, err := client.Messages.New(callCtx, anthropic.MessageNewParams{
				Model:     "claude-opus-5",
				MaxTokens: 1024,
				Messages: []anthropic.MessageParam{
					anthropic.NewUserMessage(anthropic.NewTextBlock("Суммаризируй в 3 пунктах:\n\n" + doc)),
				},
			})
			if err != nil {
				return fmt.Errorf("doc %d: %w", i, err)
			}
			var text string
			for _, b := range resp.Content {
				if t, ok := b.AsAny().(anthropic.TextBlock); ok {
					text += t.Text
				}
			}
			results[i] = Result{Text: text, InputTokens: resp.Usage.InputTokens, OutputTokens: resp.Usage.OutputTokens}
			return nil
		})
	}
	if err := g.Wait(); err != nil {
		return nil, err
	}
	if acquireErr != nil {
		return nil, fmt.Errorf("fan-out aborted: %w", acquireErr)
	}
	return results, nil
}

import asyncio
import anthropic

client = anthropic.AsyncAnthropic()  # ключ из ANTHROPIC_API_KEY

async def summarize(docs: list[str], max_in_flight: int = 8) -> list[str]:
    sem = asyncio.Semaphore(max_in_flight)
    results: list[str] = [""] * len(docs)

    async def one(i: int, doc: str) -> None:
        async with sem:  # ограничение параллелизма, как семафор
            async with asyncio.timeout(45):  # таймаут одного вызова
                resp = await client.messages.create(
                    model="claude-opus-5",
                    max_tokens=1024,
                    messages=[{"role": "user", "content": f"Суммаризируй в 3 пунктах:\n\n{doc}"}],
                )
        results[i] = "".join(b.text for b in resp.content if b.type == "text")

    # TaskGroup: первая ошибка отменяет остальные задачи
    async with asyncio.timeout(120):
        async with asyncio.TaskGroup() as tg:
            for i, d in enumerate(docs):
                tg.create_task(one(i, d))
    return results

Стриминг и отмена через context

Стриминг и отмена запроса

Ответ в 800 токенов генерируется 10–20 секунд. Без стриминга пользователь смотрит на спиннер всё это время, со стримингом видит первые слова через TTFT. Провайдеры отдают поток через SSE (Server-Sent Events: HTTP-ответ text/event-stream, события вида event: …\ndata: {json}\n\n). У Anthropic последовательность такая: message_start → для каждого блока content_block_start, серия content_block_delta (text_delta, input_json_delta для аргументов инструментов, thinking_delta), content_block_stop → message_delta (stop_reason и итоговый usage) → message_stop. Между ними бывают ping, а посреди стрима может прийти error (например, overloaded_error) — уже после 200 OK.

Для Go-разработчика тут всё знакомо: bufio.Scanner по телу ответа, канал событий наружу, http.Flusher на своей стороне, чтобы проксировать поток в браузер. Ключевая вещь — отмена. Пользователь закрыл вкладку → r.Context() вашего хендлера отменён → вы должны закрыть исходящий запрос к провайдеру, иначе модель продолжит генерировать токены, за которые вы заплатите. Если передавать r.Context() в SDK, это происходит само.

Для Java-разработчика тут всё знакомо: SDK отдаёт поток событий как StreamResponse (это AutoCloseable с java.util.stream.Stream внутри), а на своей стороне вы проксируете его в браузер через SseEmitter в Spring MVC или Flux<ServerSentEvent> в WebFlux. Ключевая вещь — отмена. Пользователь закрыл вкладку → следующий emitter.send() бросает IOException → вы должны закрыть исходящий запрос к провайдеру, иначе модель продолжит генерировать токены, за которые вы заплатите. Если держать StreamResponse в try-with-resources, закрытие происходит само: исключение выходит из блока, и соединение с провайдером рвётся. В WebFlux то же делает отмена подписки.

// Проксирование стрима в браузер и замер TTFT. Клиент ушёл → send() бросает IOException →
// try-with-resources закрывает StreamResponse → SDK рвёт соединение с провайдером.
@GetMapping(path = "/answer", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter answer(@RequestParam String q) {
    SseEmitter emitter = new SseEmitter(0L); // без таймаута Spring: ответ может идти минуты
    Thread.startVirtualThread(() -> streamAnswer(emitter, q));
    return emitter;
}

void streamAnswer(SseEmitter emitter, String q) {
    long start = System.nanoTime();
    long ttftNanos = -1;
    MessageAccumulator acc = MessageAccumulator.create();
    MessageCreateParams params = MessageCreateParams.builder()
            .model("claude-opus-5")
            .maxTokens(2048)
            .addUserMessage(q)
            .build();
    try (StreamResponse<RawMessageStreamEvent> stream = client.messages().createStreaming(params)) {
        for (Iterator<RawMessageStreamEvent> it = stream.stream().iterator(); it.hasNext(); ) {
            RawMessageStreamEvent ev = it.next();
            acc.accumulate(ev);
            Optional<String> text = ev.contentBlockDelta()
                    .flatMap(d -> d.delta().text())
                    .map(TextDelta::text);
            if (text.isPresent()) {
                if (ttftNanos < 0) {
                    ttftNanos = System.nanoTime() - start;
                }
                emitter.send(SseEmitter.event().data(text.get(), MediaType.APPLICATION_JSON));
            }
        }
        Message msg = acc.message();
        log.atInfo().setMessage("llm_call")
                .addKeyValue("ttft_ms", ttftNanos / 1_000_000)
                .addKeyValue("total_ms", (System.nanoTime() - start) / 1_000_000)
                .addKeyValue("in", msg.usage().inputTokens())
                .addKeyValue("out", msg.usage().outputTokens())
                .addKeyValue("stop", msg.stopReason().orElse(null))
                .log();
        emitter.complete();
    } catch (IOException | RuntimeException e) { // включая ошибку, пришедшую посреди стрима
        emitter.completeWithError(e);
    }
}

// Проксирование стрима в браузер и замер TTFT. ctx = r.Context() хендлера:
// клиент ушёл → ctx отменён → SDK закрывает соединение с провайдером.
func streamAnswer(ctx context.Context, w http.ResponseWriter, client *anthropic.Client, q string) error {
	flusher, ok := w.(http.Flusher)
	if !ok {
		return errors.New("streaming unsupported")
	}
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")

	start := time.Now()
	var ttft time.Duration
	stream := client.Messages.NewStreaming(ctx, anthropic.MessageNewParams{
		Model:     "claude-opus-5",
		MaxTokens: 2048,
		Messages:  []anthropic.MessageParam{anthropic.NewUserMessage(anthropic.NewTextBlock(q))},
	})
	msg := anthropic.Message{}
	for stream.Next() {
		ev := stream.Current()
		if err := msg.Accumulate(ev); err != nil {
			return err
		}
		if d, ok := ev.AsAny().(anthropic.ContentBlockDeltaEvent); ok {
			if t, ok := d.Delta.AsAny().(anthropic.TextDelta); ok {
				if ttft == 0 {
					ttft = time.Since(start)
				}
				b, _ := json.Marshal(t.Text)
				fmt.Fprintf(w, "data: %s\n\n", b)
				flusher.Flush()
			}
		}
	}
	if err := stream.Err(); err != nil {
		return err // включая ошибку, пришедшую посреди стрима
	}
	slog.InfoContext(ctx, "llm_call", "ttft_ms", ttft.Milliseconds(),
		"total_ms", time.Since(start).Milliseconds(),
		"in", msg.Usage.InputTokens, "out", msg.Usage.OutputTokens, "stop", msg.StopReason)
	return nil
}

Ретраи и rate limit

Вызов LLM API падает заметно чаще, чем ваш внутренний сервис. Что ретраить, а что нет:

Ретрай в LLM-мире дороже, чем кажется. Если запрос упал на середине стрима, вы уже заплатили за выданные токены. Если ретраите запрос со 100K токенов контекста, каждая попытка — это 100K входных токенов в ITPM. Агрессивные ретраи при 429 сами поддерживают перегрузку: это retry storm. Поэтому backoff экспоненциальный и обязательно с jitter — случайной добавкой, которая «размазывает» повторы клиентов во времени.

Код: ретрай с экспоненциальным backoff + jitter

Вариант «full jitter» из классической статьи AWS: пауза равномерно случайна в интервале (0, min(cap, base·2^attempt)]. Если провайдер прислал retry-after, ждём не меньше него.

package retry;

import com.anthropic.errors.AnthropicIoException;
import com.anthropic.errors.AnthropicServiceException;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadLocalRandom;

public final class Retry {
    /**
     * Вызывает fn до attempts раз. Встроенные ретраи SDK стоит отключить:
     * AnthropicOkHttpClient.builder().fromEnv().maxRetries(0).build(), иначе попытки перемножатся.
     * Отмена — прерывание потока: Thread.sleep бросит InterruptedException.
     */
    public static <T> T call(int attempts, Duration base, Duration maxDelay, Callable<T> fn) throws Exception {
        for (int a = 0; ; a++) {
            try {
                return fn.call();
            } catch (Exception e) {
                Duration retryAfter = retryAfter(e); // null — не ретраить
                if (retryAfter == null || a == attempts - 1) {
                    throw e;
                }
                Duration delay = fullJitter(a, base, maxDelay);
                if (retryAfter.compareTo(delay) > 0) {
                    delay = retryAfter; // сервер знает лучше, когда освободится лимит
                }
                Thread.sleep(delay);
            }
        }
    }

    static Duration fullJitter(int attempt, Duration base, Duration maxDelay) {
        long cap = maxDelay.toMillis();
        long d = base.toMillis() << Math.min(attempt, 30);
        if (d <= 0 || d > cap) { // <= 0 — защита от переполнения сдвига
            d = cap;
        }
        return Duration.ofMillis(ThreadLocalRandom.current().nextLong(d) + 1);
    }

    /** null — ретраить бессмысленно; иначе минимальная пауза из retry-after (может быть ZERO). */
    static Duration retryAfter(Exception e) {
        if (e instanceof AnthropicServiceException se) {
            return switch (se.statusCode()) {
                case 408, 409, 429, 500, 502, 503, 504, 529 -> parseRetryAfter(se);
                default -> null; // 400/401/403/404/413 — повтор не поможет
            };
        }
        if (e instanceof AnthropicIoException) {
            return Duration.ZERO; // сетевая ошибка: соединение сброшено, DNS и т.п.
        }
        return null; // InterruptedException — вызывающий сдался, не спорим с ним
    }

    static Duration parseRetryAfter(AnthropicServiceException e) {
        List<String> v = e.headers().values("retry-after");
        try {
            return v.isEmpty() ? Duration.ZERO : Duration.ofSeconds(Math.max(0, Long.parseLong(v.get(0))));
        } catch (NumberFormatException ex) {
            return Duration.ZERO;
        }
    }
}

package retry

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

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

// Do вызывает fn до attempts раз. Встроенные ретраи SDK стоит отключить:
// anthropic.NewClient(option.WithMaxRetries(0)), иначе попытки перемножатся.
func Do[T any](ctx context.Context, attempts int, base, maxDelay time.Duration, fn func(context.Context) (T, error)) (T, error) {
	var zero T
	var err error
	for a := 0; a < attempts; a++ {
		var v T
		if v, err = fn(ctx); err == nil {
			return v, nil
		}
		retryable, retryAfter := classify(err)
		if !retryable || a == attempts-1 {
			break
		}
		delay := fullJitter(a, base, maxDelay)
		if retryAfter > delay {
			delay = retryAfter // сервер знает лучше, когда освободится лимит
		}
		t := time.NewTimer(delay)
		select {
		case <-ctx.Done():
			t.Stop()
			return zero, errors.Join(err, ctx.Err())
		case <-t.C:
		}
	}
	return zero, err
}

func fullJitter(attempt int, base, maxDelay time.Duration) time.Duration {
	d := base << attempt
	if d <= 0 || d > maxDelay { // <= 0 — защита от переполнения сдвига
		d = maxDelay
	}
	return time.Duration(rand.Int64N(int64(d))) + time.Millisecond
}

func classify(err error) (retryable bool, retryAfter time.Duration) {
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return false, 0 // вызывающий сдался — не спорим с ним
	}
	var apierr *anthropic.Error
	if !errors.As(err, &apierr) {
		return true, 0 // сетевая ошибка: соединение сброшено, DNS и т.п.
	}
	switch apierr.StatusCode {
	case http.StatusRequestTimeout, http.StatusConflict, http.StatusTooManyRequests,
		500, 502, 503, 504, 529:
		return true, parseRetryAfter(apierr.Response)
	}
	return false, 0 // 400/401/403/404/413 — повтор не поможет
}

func parseRetryAfter(resp *http.Response) time.Duration {
	if resp == nil {
		return 0
	}
	if s, err := strconv.Atoi(resp.Header.Get("retry-after")); err == nil && s > 0 {
		return time.Duration(s) * time.Second
	}
	return 0
}

Rate limit на своей стороне: token bucket

Ждать 429 от провайдера — реактивная стратегия. Проактивная — не превышать лимит самому. Token bucket (ведро токенов) — то, что вы знаете по golang.org/x/time/rateBucket4j, Guava RateLimiter или Resilience4j RateLimiter: ведро ёмкостью burst пополняется со скоростью r в секунду, каждый запрос забирает из него «жетоны». Новое здесь одно: жетоны — это LLM-токены, а не запросы. Лимитер на запросы защищает RPM, но не ITPM; один запрос с 150K контекста съест лимит, как сотня маленьких.

// Bucket4j. Два ведра: на запросы и на входные токены. Значения — из вашего tier'а в консоли провайдера.
static final long TOKEN_BURST = 200_000;
static final Bucket REQ_LIM = Bucket.builder()
        .addLimit(l -> l.capacity(50).refillGreedy(4_000, Duration.ofMinutes(1)))              // 4000 RPM
        .build();
static final Bucket TOK_LIM = Bucket.builder()
        .addLimit(l -> l.capacity(TOKEN_BURST).refillGreedy(2_000_000, Duration.ofMinutes(1))) // 2M ITPM, burst 200K
        .build();

static void acquire(long estInputTokens) throws InterruptedException {
    REQ_LIM.asBlocking().consume(1);
    // Больше ёмкости ведро не накопит никогда: огромный запрос надо резать или слать в Batch API.
    if (estInputTokens > TOKEN_BURST) {
        throw new IllegalArgumentException("request exceeds token burst: " + estInputTokens);
    }
    TOK_LIM.asBlocking().consume(estInputTokens);
}

// Два ведра: на запросы и на входные токены. Значения — из вашего tier'а в консоли провайдера.
reqLim := rate.NewLimiter(rate.Limit(4000.0/60), 50)           // 4000 RPM
tokLim := rate.NewLimiter(rate.Limit(2_000_000.0/60), 200_000) // 2M ITPM, burst 200K

func acquire(ctx context.Context, estInputTokens int) error {
	if err := reqLim.Wait(ctx); err != nil {
		return err
	}
	// WaitN вернёт ошибку, если n > burst: огромный запрос надо резать или слать в Batch API.
	return tokLim.WaitN(ctx, estInputTokens)
}

Circuit breaker и fallback

Circuit breaker — тот же паттерн, что вы ставите перед нестабильным сервисом (sony/gobreaker и аналогиResilience4j CircuitBreaker, в Spring Boot — через @CircuitBreaker(fallbackMethod = …) или Spring Cloud CircuitBreaker). В LLM-мире у него две задачи: не долбить перегруженного провайдера (529 сериями по несколько минут — реальность) и быстро переключиться на fallback. Fallback-цепочка типична: основная модель → более дешёвая модель того же провайдера → другой провайдер → деградированный ответ («сервис перегружен, попробуйте позже» или ответ без LLM).

Настройки, которые в LLM-контексте отличаются от привычных:

• Считать нужно правильные ошибки. 400 из-за вашего бага не должна открывать breaker. 429 — спорно: это не отказ провайдера, а ваш лимит; правильнее снизить конкурентность, а не отрубать всё. Открывать стоит на 5xx/529 и таймауты.
• Порог по доле, а не подряд. При 200 RPS «5 ошибок подряд» случается от шума. Используйте долю ошибок в скользящем окне с минимальным числом запросов.
• В Resilience4j это три настройки: failureRateThreshold, slidingWindowSize и minimumNumberOfCalls. Какие исключения считать отказом, задаёт recordException(e -> …): 5xx/529 и таймауты — да, BadRequestException и RateLimitException из SDK — нет. Медленные вызовы ловит slowCallDurationThreshold.
• Таймаут вызова — важный сигнал. Если p99 обычно 20 с, а сейчас запросы висят по 90 с, провайдер деградирует, хоть и отвечает 200.
• Отдельный breaker на модель и регион. Перегрузка одной модели не значит, что недоступна другая.
• Fallback меняет качество. Промпт, отлаженный под одну модель, на другой может дать другой формат. Fallback-модель должна проходить те же evals, иначе вы узнаете о проблеме в момент инцидента.

Observability → токены, стоимость, TTFT

Логи, метрики, трейсы остаются. Меняется набор сигналов. Классический RED (rate, errors, duration) нужен, но недостаточен: запрос может вернуть 200 за 3 секунды и быть бесполезным. Минимальный набор атрибутов на каждый LLM-вызов:

Для трейсов есть стандарт: OpenTelemetry GenAI semantic conventions — атрибуты gen_ai.request.model, gen_ai.usage.input_tokens, gen_ai.usage.output_tokens, gen_ai.response.finish_reasons и т.д. В агентах трейс особенно ценен: span на каждый шаг модели и каждый tool call показывает, где ушли 40 секунд и 200K токенов.

В Spring Boot это ложится на привычный стек: Micrometer Observation API даёт и метрики, и спаны одним вызовом, Micrometer Tracing с мостом в OpenTelemetry экспортирует их по OTLP, а /actuator/prometheus отдаёт счётчики токенов и стоимости. Spring AI и LangChain4j сами создают observation на каждый вызов модели с атрибутами gen_ai.*; если вы зовёте официальный anthropic-java напрямую, оберните вызов в свой Observation или спан OpenTelemetry Java API (пример — в проекте D, сквозные проекты).

БД, очереди, кэши → векторные БД и семантический кэш

Ваш опыт с PostgreSQL, индексами и Redis напрямую нужен в RAG (retrieval-augmented generation — поиск релевантных документов и подстановка их в контекст модели). Новые сущности:

• Embedding — вектор из сотен–тысяч чисел, который модель-эмбеддер ставит в соответствие тексту так, что близкие по смыслу тексты близки по косинусному расстоянию. Это «ключ» для поиска по смыслу.
• Векторный индекс — приближённый поиск ближайших соседей (ANN), чаще всего HNSW. Для Java-разработчика самое естественное — pgvector в том же Postgres: транзакции, JOIN с метаданными, привычные бэкапы, Flyway-миграции. Из кода — обычный JDBC или jOOQ с типом PGvector из pgvector-java, либо готовый PgVectorStore из Spring AI (или PgVectorEmbeddingStore из LangChain4j); JPA для векторного поиска неудобна — запросы с <=> всё равно пишутся нативным SQL. Интеграционные тесты — Testcontainers с образом pgvector/pgvector. Отдельные векторные БД (Qdrant, Weaviate, Milvus и др.) имеют смысл на десятках-сотнях миллионов векторов или при специфичных требованиях. Подробно — модуль 5.
• Векторный индекс — приближённый поиск ближайших соседей (ANN), чаще всего HNSW. Для Go-разработчика самое естественное — pgvector в том же Postgres: транзакции, JOIN с метаданными, привычные бэкапы. Отдельные векторные БД (Qdrant, Weaviate, Milvus и др.) имеют смысл на десятках-сотнях миллионов векторов или при специфичных требованиях. Подробно — модуль 5.
• Гибридный поиск. Векторный поиск плохо находит точные идентификаторы, коды ошибок, артикулы. Его комбинируют с полнотекстовым (BM25 / tsvector) — тоже знакомая территория.

Кэширование раздваивается:

Очереди тоже переносятся: длинные задачи (суммаризация документа на 300 страниц, агент, который работает 10 минут) не должны жить в HTTP-запросе. Паттерн «задача в очередь → воркер → статус и результат в БД → уведомление» ровно тот же. Отличие — идемпотентность дороже: повторная обработка сообщения не просто нагружает базу, а стоит денег и может дать другой результат. Храните результат по ключу задачи и проверяйте его перед вызовом модели. Для офлайн-нагрузок у провайдеров есть Batch API: у Anthropic −50% к цене, результаты до 24 часов, порядок произвольный, сопоставление по custom_id.

Тесты → evals

Самое большое смысловое изменение. Table-driven тестыПараметризованные тесты JUnit 5 (@ParameterizedTest с @CsvFileSource или @MethodSource) — отличная интуиция для evals: таблица входов и ожиданий. Но проверка меняется с «равно» на «насколько хорошо», а результат — с «зелёный/красный» на число с погрешностью.

Статистика здесь не для красоты. 50 примеров, 45 прошли — 90%, но 95% доверительный интервал (Уилсона) примерно 79–96%. Если новая версия промпта дала 47 из 50, это не доказательство улучшения, это шум. Для различения 90% и 93% нужны сотни примеров. Для сравнения двух версий на одном наборе используют парные тесты (McNemar) — смотрят только примеры, где версии разошлись.

// Запуск: mvn verify -Dgroups=evals (по умолчанию тег исключён в surefire — это стоит денег)
@Tag("evals")
class ClassifierEvalTest {
    @Test
    void classifierAccuracy() throws Exception {
        List<GoldenCase> cases = Golden.load("tickets_golden.jsonl"); // 300 размеченных тикетов
        var pass = new AtomicInteger();
        var sem = new Semaphore(8);
        try (var pool = Executors.newVirtualThreadPerTaskExecutor()) {
            var futures = cases.stream().map(c -> pool.submit(() -> {
                sem.acquire();
                try {
                    if (classifier.classify(c.text()).category().equals(c.want())) {
                        pass.incrementAndGet();
                    }
                } finally {
                    sem.release();
                }
                return null;
            })).toList();
            for (var f : futures) {
                f.get(); // инфраструктурная ошибка — это не «модель ошиблась»: тест падает
            }
        }
        double n = cases.size();
        double rate = pass.get() / n;
        double lo = wilsonLower(rate, n, 1.96);
        System.out.printf("accuracy=%.3f (95%% CI lower=%.3f, n=%.0f)%n", rate, lo, n);
        assertTrue(lo >= 0.85, // порог по нижней границе, а не по точечной оценке
                () -> "accuracy lower bound %.3f < 0.85".formatted(lo));
    }
}

//go:build evals

// Запуск: go test -tags evals -run TestClassifierEval ./... (не в каждом CI, это стоит денег)
func TestClassifierEval(t *testing.T) {
	cases := loadGolden(t, "testdata/tickets_golden.jsonl") // 300 размеченных тикетов
	var pass atomic.Int64
	g, ctx := errgroup.WithContext(context.Background())
	g.SetLimit(8)
	for _, c := range cases {
		g.Go(func() error {
			got, err := Classify(ctx, client, c.Text)
			if err != nil {
				return err // инфраструктурная ошибка — это не «модель ошиблась»
			}
			if got.Category == c.Want {
				pass.Add(1)
			}
			return nil
		})
	}
	if err := g.Wait(); err != nil {
		t.Fatal(err)
	}
	n := float64(len(cases))
	rate := float64(pass.Load()) / n
	lo := wilsonLower(rate, n, 1.96)
	t.Logf("accuracy=%.3f (95%% CI lower=%.3f, n=%.0f)", rate, lo, n)
	if lo < 0.85 { // порог по нижней границе, а не по точечной оценке
		t.Errorf("accuracy lower bound %.3f < 0.85", lo)
	}
}

В проде так делают редко на GoJava целиком: разметка, анализ и графики удобнее в Python (pandas, Jupyter, модуль 4), а evals — отдельная дисциплина (модуль 8). Но прогонщик в CI, который шлёт сотни запросов с лимитами и ретраями, — это как раз ваша сильная сторона.

Что учить с нуля

Переносится инфраструктура. Новое — всё, что про поведение модели и оценку качества:

Практика

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

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

Источники

• Anthropic — Building effective agents (паттерны, включая parallelization)
• Anthropic API — Rate limits (RPM / ITPM / OTPM, retry-after)
• Anthropic API — Errors (коды, overloaded_error 529)
• Anthropic API — Streaming Messages (типы SSE-событий)
• anthropic-sdk-go — официальный Go SDK
• errgroup, semaphore, x/time/rate
• anthropic-sdk-java — официальный Java SDK (com.anthropic:anthropic-java)
• JEP 444 — Virtual Threads, JEP 453 — Structured Concurrency (Preview), Bucket4j
• Spring AI, LangChain4j, Testcontainers for Java
• AWS Architecture Blog — Exponential Backoff And Jitter
• Martin Fowler — CircuitBreaker; sony/gobreakerResilience4j CircuitBreaker
• Google SRE Book — Handling Overload (retry budgets)
• Dean, Barroso — The Tail at Scale (hedged requests)
• OpenTelemetry — Semantic conventions for GenAI
• Hamel Husain — Your AI Product Needs Evals
• pgvector и pgvector-java