Свои модели: инференс и адаптация

# 1) Запуск OpenAI-совместимого сервера (GPU с 24+ ГБ для 8B в bf16):
#    pip install vllm
#    vllm serve Qwen/Qwen3-8B \
#      --max-model-len 32768 \          # верхняя граница контекста: определяет бюджет KV-cache
#      --gpu-memory-utilization 0.90 \  # доля VRAM под веса + KV-cache
#      --max-num-seqs 128 \             # максимум одновременных последовательностей в батче
#      --enable-prefix-caching \        # в свежих версиях включено по умолчанию
#      --api-key "$VLLM_API_KEY"
#
# 2) Клиент — обычный OpenAI SDK. В vLLM temperature/top_p работают (в отличие от новых Claude).
import asyncio, os, time
from openai import AsyncOpenAI

client = AsyncOpenAI(base_url="http://localhost:8000/v1", api_key=os.environ["VLLM_API_KEY"])

async def one(i: int) -> tuple[float, float, int]:
    t0 = time.perf_counter(); ttft = None; n = 0
    stream = await client.chat.completions.create(
        model="Qwen/Qwen3-8B",
        messages=[{"role": "user", "content": f"Кратко: что такое continuous batching? #{i}"}],
        max_tokens=300, temperature=0.2, stream=True,
    )
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            ttft = ttft or time.perf_counter() - t0
            n += 1  # чанк ≈ токен; для точности берите usage
    return ttft or 0.0, time.perf_counter() - t0, n

async def main(concurrency: int = 64) -> None:
    t0 = time.perf_counter()
    res = await asyncio.gather(*(one(i) for i in range(concurrency)))
    wall = time.perf_counter() - t0
    ttfts = sorted(r[0] for r in res)
    print(f"p50 TTFT {ttfts[len(ttfts)//2]:.2f}s  p99 {ttfts[int(len(ttfts)*.99)]:.2f}s")
    print(f"суммарно {sum(r[2] for r in res)/wall:.0f} tok/s при {concurrency} параллельных")

asyncio.run(main())
# Локально без GPU: `ollama pull qwen3:8b` и base_url="http://localhost:11434/v1" — тот же код.

# Встроенный бенчмарк: случайные промпты 2048 → 512 токенов, 64 запроса одновременно.
vllm bench serve --model Qwen/Qwen3-8B \
  --dataset-name random --random-input-len 2048 --random-output-len 512 \
  --num-prompts 640 --max-concurrency 64 \
  --percentile-metrics ttft,tpot,itl --metric-percentiles 50,95,99 \
  --goodput ttft:1000 tpot:50
# Повторите для --max-concurrency 8, 16, 32, 64, 128. Ёмкость реплики — максимум,
# при котором p95 TTFT и TPOT ещё в SLO. Это и есть goodput для калькулятора выше.

// Шлюз перед OpenAI-совместимым сервером vLLM (подойдёт и Ollama на :11434).
// Делает пять вещей: публичное имя модели → база или LoRA-адаптер в vLLM, лимит
// одновременных запросов под --max-num-seqs, ретраи на 429/5xx, пока клиенту
// ещё ничего не отправлено, таймаут на первый байт стрима и метрики TTFT/TPOT.
package main

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"math/rand/v2"
	"net/http"
	"strconv"
	"time"
)

type Config struct {
	Upstream    string            // http://vllm:8000
	APIKey      string            // --api-key vLLM; пусто — без авторизации
	Models      map[string]string // публичное имя → имя в vLLM: база или адаптер из --lora-modules
	MaxInFlight int               // ≈ max-num-seqs × число реплик
	QueueWait   time.Duration     // сколько запрос ждёт свободный слот, потом 429
	FirstByte   time.Duration     // стрим: нет первого байта за это время — отменяем попытку и повторяем
	Timeout     time.Duration     // общий потолок на запрос
	MaxRetries  int
	Backoff     time.Duration // база экспоненциального backoff с full jitter
	Observe     func(Stats)   // куда отдать метрики: лог, Prometheus, OTel (модуль 9)
}

type Stats struct {
	Model     string
	Status    int
	Attempts  int
	TTFT      time.Duration // от прихода запроса до первого непустого токена
	TPOT      time.Duration // (последний токен − первый) / (токенов − 1)
	OutTokens int
}

type Gateway struct {
	cfg    Config
	client *http.Client
	slots  chan struct{}
}

func New(cfg Config) *Gateway {
	t := http.DefaultTransport.(*http.Transport).Clone()
	t.MaxIdleConnsPerHost = cfg.MaxInFlight // по умолчанию 2: при 128 параллельных стримах соединения пересоздавались бы
	return &Gateway{cfg: cfg, client: &http.Client{Transport: t}, slots: make(chan struct{}, cfg.MaxInFlight)}
}

func (g *Gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	start := time.Now()
	if r.Method != http.MethodPost || r.URL.Path != "/v1/chat/completions" {
		apiError(w, http.StatusNotFound, "only POST /v1/chat/completions")
		return
	}
	var req map[string]json.RawMessage // остальные поля запроса пробрасываем как есть
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&req); err != nil {
		apiError(w, http.StatusBadRequest, "bad JSON: "+err.Error())
		return
	}
	var public string
	var stream bool
	_ = json.Unmarshal(req["model"], &public)
	_ = json.Unmarshal(req["stream"], &stream)
	target, ok := g.cfg.Models[public]
	if !ok {
		apiError(w, http.StatusNotFound, fmt.Sprintf("model %q not found", public))
		return
	}
	req["model"], _ = json.Marshal(target)
	if _, set := req["stream_options"]; stream && !set {
		req["stream_options"] = json.RawMessage(`{"include_usage":true}`) // точное число токенов в последнем чанке
	}
	body, _ := json.Marshal(req)

	st := Stats{Model: public}
	defer func() {
		if g.cfg.Observe != nil {
			g.cfg.Observe(st)
		}
	}()
	ctx, cancel := context.WithTimeout(r.Context(), g.cfg.Timeout)
	defer cancel()
	wait := time.NewTimer(g.cfg.QueueWait)
	defer wait.Stop()
	select { // слот: в vLLM не больше MaxInFlight запросов, остальные ждут здесь, а не в его очереди
	case g.slots <- struct{}{}:
		defer func() { <-g.slots }()
	case <-wait.C:
		st.Status = http.StatusTooManyRequests
		w.Header().Set("Retry-After", "1")
		apiError(w, st.Status, "all slots busy")
		return
	case <-ctx.Done():
		return
	}

	up, attempts, err := g.send(ctx, body, stream)
	st.Attempts = attempts
	if err != nil {
		st.Status = http.StatusBadGateway
		if ctx.Err() == context.DeadlineExceeded {
			st.Status = http.StatusGatewayTimeout
		}
		apiError(w, st.Status, "upstream: "+err.Error())
		return
	}
	defer up.Close()
	st.Status = up.resp.StatusCode
	if stream && st.Status == http.StatusOK {
		pipeSSE(w, up.body, start, &st)
		return
	}
	raw, err := io.ReadAll(up.body)
	if err != nil {
		st.Status = http.StatusBadGateway
		apiError(w, st.Status, "upstream: "+err.Error())
		return
	}
	var resp struct {
		Usage struct {
			CompletionTokens int `json:"completion_tokens"`
		} `json:"usage"`
	}
	if json.Unmarshal(raw, &resp) == nil {
		st.OutTokens = resp.Usage.CompletionTokens
	}
	st.TTFT = time.Since(start) // без стрима пользователь видит первый токен вместе с последним
	w.Header().Set("Content-Type", up.resp.Header.Get("Content-Type"))
	w.WriteHeader(st.Status)
	_, _ = w.Write(raw)
}

type upstream struct {
	resp   *http.Response
	body   *bufio.Reader
	cancel context.CancelFunc
}

func (u *upstream) Close() { u.resp.Body.Close(); u.cancel() }

// send повторяет запрос, пока vLLM отвечает 429/502/503/504 или стрим молчит дольше FirstByte.
// Повторять можно только до первого байта клиенту: после него ответ уже частично отдан.
func (g *Gateway) send(ctx context.Context, body []byte, stream bool) (*upstream, int, error) {
	for attempt := 1; ; attempt++ {
		up, err := g.attempt(ctx, body, stream)
		if err == nil && !retryable(up.resp.StatusCode) {
			return up, attempt, nil
		}
		if attempt > g.cfg.MaxRetries || ctx.Err() != nil {
			return up, attempt, err // последний 429/503 отдаём клиенту как есть
		}
		delay := time.Duration(0)
		if d := g.cfg.Backoff << (attempt - 1); d > 0 {
			delay = rand.N(d)
		}
		if err == nil {
			if s, perr := strconv.Atoi(up.resp.Header.Get("Retry-After")); perr == nil {
				delay = max(delay, time.Duration(s)*time.Second)
			}
			up.Close()
		}
		select {
		case <-time.After(delay):
		case <-ctx.Done():
			return nil, attempt, ctx.Err()
		}
	}
}

func (g *Gateway) attempt(ctx context.Context, body []byte, stream bool) (*upstream, error) {
	actx, cancel := context.WithCancel(ctx) // отмена закрывает соединение, и vLLM освобождает слот и KV-cache
	req, err := http.NewRequestWithContext(actx, http.MethodPost, g.cfg.Upstream+"/v1/chat/completions", bytes.NewReader(body))
	if err != nil {
		cancel()
		return nil, err
	}
	req.Header.Set("Content-Type", "application/json")
	if g.cfg.APIKey != "" {
		req.Header.Set("Authorization", "Bearer "+g.cfg.APIKey)
	}
	if stream && g.cfg.FirstByte > 0 {
		t := time.AfterFunc(g.cfg.FirstByte, cancel)
		defer t.Stop() // остановится, как только Peek дождётся первого байта
	}
	resp, err := g.client.Do(req)
	if err != nil {
		cancel()
		return nil, err
	}
	up := &upstream{resp: resp, body: bufio.NewReader(resp.Body), cancel: cancel}
	if _, err := up.body.Peek(1); err != nil && err != io.EOF {
		up.Close()
		return nil, err
	}
	return up, nil
}

func retryable(code int) bool {
	return code == http.StatusTooManyRequests || code == http.StatusBadGateway ||
		code == http.StatusServiceUnavailable || code == http.StatusGatewayTimeout
}

// chunk — то, что нужно шлюзу из SSE-чанка chat.completion.chunk.
type chunk struct {
	Choices []struct {
		Delta struct {
			Content   string `json:"content"`
			Reasoning string `json:"reasoning"` // reasoning-модели; в старых vLLM поле звалось reasoning_content
		} `json:"delta"`
	} `json:"choices"`
	Usage *struct {
		CompletionTokens int `json:"completion_tokens"`
	} `json:"usage"`
}

// pipeSSE пробрасывает стрим построчно и сбрасывает буфер после каждой строки.
func pipeSSE(w http.ResponseWriter, src *bufio.Reader, start time.Time, st *Stats) {
	h := w.Header()
	h.Set("Content-Type", "text/event-stream")
	h.Set("Cache-Control", "no-cache")
	h.Set("X-Accel-Buffering", "no") // nginx перед шлюзом не должен копить стрим
	w.WriteHeader(http.StatusOK)
	rc := http.NewResponseController(w)
	var first, last time.Time
	chunks := 0
	for {
		line, err := src.ReadBytes('\n')
		if len(line) > 0 {
			if _, werr := w.Write(line); werr != nil {
				return // клиент ушёл: defer отменит запрос к vLLM, генерация остановится
			}
			_ = rc.Flush()
			if data, ok := bytes.CutPrefix(line, []byte("data:")); ok {
				var c chunk
				if json.Unmarshal(data, &c) == nil { // "[DONE]" не JSON — пропускаем
					if len(c.Choices) > 0 && (c.Choices[0].Delta.Content != "" || c.Choices[0].Delta.Reasoning != "") {
						if first.IsZero() {
							first = time.Now()
						}
						last = time.Now()
						chunks++
					}
					if c.Usage != nil {
						st.OutTokens = c.Usage.CompletionTokens
					}
				}
			}
		}
		if err != nil {
			break
		}
	}
	if st.OutTokens == 0 {
		st.OutTokens = chunks // без usage: чанк ≈ токен, если stream_interval = 1
	}
	if !first.IsZero() {
		st.TTFT = first.Sub(start)
		if st.OutTokens > 1 {
			st.TPOT = last.Sub(first) / time.Duration(st.OutTokens-1)
		}
	}
}

func apiError(w http.ResponseWriter, code int, msg string) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	_ = json.NewEncoder(w).Encode(map[string]any{"error": map[string]any{"message": msg, "code": code}})
}

// Шлюз перед OpenAI-совместимым сервером vLLM (подойдёт и Ollama на :11434).
// Делает пять вещей: публичное имя модели → база или LoRA-адаптер в vLLM, лимит
// одновременных запросов под --max-num-seqs, ретраи на 429/5xx, пока клиенту
// ещё ничего не отправлено, таймаут на первый байт стрима и метрики TTFT/TPOT.
package course.m10models;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpTimeoutException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

public final class Gateway implements HttpHandler {

    /**
     * upstream — http://vllm:8000; apiKey — --api-key vLLM, пусто — без авторизации;
     * models — публичное имя → имя в vLLM: база или адаптер из --lora-modules;
     * maxInFlight ≈ max-num-seqs × число реплик; queueWait — сколько запрос ждёт свободный слот, потом 429;
     * firstByte — стрим: нет первого байта за это время — отменяем попытку и повторяем;
     * timeout — общий потолок на запрос; backoff — база экспоненциального backoff с full jitter;
     * observe — куда отдать метрики: лог, Micrometer, OTel (модуль 9).
     */
    public record Config(String upstream, String apiKey, Map<String, String> models, int maxInFlight,
                         Duration queueWait, Duration firstByte, Duration timeout, int maxRetries,
                         Duration backoff, Consumer<Stats> observe) {}

    /** ttft — от прихода запроса до первого непустого токена; tpot — (последний токен − первый) / (токенов − 1). */
    public static final class Stats {
        public final String model;
        public int status, attempts, outTokens;
        public Duration ttft = Duration.ZERO, tpot = Duration.ZERO;

        Stats(String model) {
            this.model = model;
        }

        @Override
        public String toString() {
            return "Stats[model=%s, status=%d, attempts=%d, ttft=%s, tpot=%s, outTokens=%d]"
                    .formatted(model, status, attempts, ttft, tpot, outTokens);
        }
    }

    private static final ObjectMapper JSON = new ObjectMapper();
    private static final int MAX_BODY = 4 << 20;
    private static final ExecutorService VT = Executors.newVirtualThreadPerTaskExecutor();

    private final Config cfg;
    // Пул соединений HttpClient для HTTP/1.1 по умолчанию не ограничен, так что 128 параллельных
    // стримов не пересоздают соединения. HTTP/1.1 явно: vLLM не говорит h2c.
    private final HttpClient client = HttpClient.newBuilder()
            .version(HttpClient.Version.HTTP_1_1)
            .connectTimeout(Duration.ofSeconds(5))
            .build();
    final Semaphore slots;

    public Gateway(Config cfg) {
        this.cfg = cfg;
        this.slots = new Semaphore(cfg.maxInFlight());
    }

    @Override
    public void handle(HttpExchange ex) throws IOException {
        long start = System.nanoTime();
        try (ex) {
            serve(ex, start);
        }
    }

    private void serve(HttpExchange ex, long start) throws IOException {
        if (!ex.getRequestMethod().equals("POST") || !ex.getRequestURI().getPath().equals("/v1/chat/completions")) {
            apiError(ex, 404, "only POST /v1/chat/completions");
            return;
        }
        ObjectNode req; // остальные поля запроса пробрасываем как есть
        try {
            byte[] raw = ex.getRequestBody().readNBytes(MAX_BODY + 1);
            if (raw.length > MAX_BODY) {
                throw new IOException("request body too large");
            }
            req = (ObjectNode) JSON.readTree(raw);
        } catch (IOException | ClassCastException e) {
            apiError(ex, 400, "bad JSON: " + e.getMessage());
            return;
        }
        String pub = req.path("model").asText();
        boolean stream = req.path("stream").asBoolean(false);
        String target = cfg.models().get(pub);
        if (target == null) {
            apiError(ex, 404, "model \"%s\" not found".formatted(pub));
            return;
        }
        req.put("model", target);
        if (stream && !req.has("stream_options")) {
            req.putObject("stream_options").put("include_usage", true); // точное число токенов в последнем чанке
        }
        byte[] body = JSON.writeValueAsBytes(req);

        var st = new Stats(pub);
        try {
            proxy(ex, body, stream, start, st);
        } finally {
            if (cfg.observe() != null) {
                cfg.observe().accept(st);
            }
        }
    }

    private void proxy(HttpExchange ex, byte[] body, boolean stream, long start, Stats st) throws IOException {
        long deadline = start + cfg.timeout().toNanos();
        // Слот: в vLLM не больше maxInFlight запросов, остальные ждут здесь, а не в его очереди.
        try {
            if (!slots.tryAcquire(cfg.queueWait().toNanos(), TimeUnit.NANOSECONDS)) {
                st.status = 429;
                ex.getResponseHeaders().set("Retry-After", "1");
                apiError(ex, st.status, "all slots busy");
                return;
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return;
        }
        try {
            Upstream up;
            try {
                up = send(body, stream, deadline, st);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            } catch (Exception e) {
                st.status = System.nanoTime() >= deadline ? 504 : 502;
                apiError(ex, st.status, "upstream: " + e);
                return;
            }
            try (up) {
                st.status = up.resp().statusCode();
                if (stream && st.status == 200) {
                    pipeSSE(ex, up.body(), start, st);
                    return;
                }
                byte[] raw;
                try {
                    raw = up.body().readAllBytes();
                } catch (IOException e) {
                    st.status = 502;
                    apiError(ex, st.status, "upstream: " + e);
                    return;
                }
                try {
                    st.outTokens = JSON.readTree(raw).path("usage").path("completion_tokens").asInt(0);
                } catch (IOException ignored) {
                    // тело ошибки бывает не JSON — отдаём как есть
                }
                st.ttft = Duration.ofNanos(System.nanoTime() - start); // без стрима пользователь видит первый токен вместе с последним
                ex.getResponseHeaders().set("Content-Type",
                        up.resp().headers().firstValue("Content-Type").orElse("application/json"));
                ex.sendResponseHeaders(st.status, raw.length == 0 ? -1 : raw.length);
                ex.getResponseBody().write(raw);
            }
        } finally {
            slots.release();
        }
    }

    /** close() закрывает тело: соединение рвётся, и vLLM освобождает слот и KV-cache. */
    private record Upstream(HttpResponse<InputStream> resp, BufferedInputStream body) implements AutoCloseable {
        @Override
        public void close() {
            try {
                body.close();
            } catch (IOException ignored) {
                // соединение и так закрывается
            }
        }
    }

    /**
     * Повторяет запрос, пока vLLM отвечает 429/502/503/504 или стрим молчит дольше firstByte.
     * Повторять можно только до первого байта клиенту: после него ответ уже частично отдан.
     */
    private Upstream send(byte[] body, boolean stream, long deadline, Stats st) throws Exception {
        for (int attempt = 1; ; attempt++) {
            st.attempts = attempt;
            Upstream up = null;
            Exception err = null;
            try {
                up = attempt(body, stream, deadline);
            } catch (InterruptedException e) {
                throw e;
            } catch (Exception e) {
                err = e;
            }
            if (up != null && !retryable(up.resp().statusCode())) {
                return up;
            }
            if (attempt > cfg.maxRetries() || System.nanoTime() >= deadline) {
                if (up != null) {
                    return up; // последний 429/503 отдаём клиенту как есть
                }
                throw err;
            }
            long ceil = cfg.backoff().toNanos() << (attempt - 1);
            long delay = ceil > 0 ? ThreadLocalRandom.current().nextLong(ceil) : 0;
            if (up != null) {
                delay = Math.max(delay, retryAfterNanos(up.resp()));
                up.close();
            }
            Thread.sleep(Duration.ofNanos(Math.max(0, Math.min(delay, deadline - System.nanoTime()))));
        }
    }

    private Upstream attempt(byte[] body, boolean stream, long deadline) throws Exception {
        long left = deadline - System.nanoTime();
        if (left <= 0) {
            throw new HttpTimeoutException("gateway deadline exceeded");
        }
        var b = HttpRequest.newBuilder(URI.create(cfg.upstream() + "/v1/chat/completions"))
                .timeout(Duration.ofNanos(left))
                .header("Content-Type", "application/json")
                .POST(HttpRequest.BodyPublishers.ofByteArray(body));
        if (!cfg.apiKey().isEmpty()) {
            b.header("Authorization", "Bearer " + cfg.apiKey());
        }
        long wait = stream && cfg.firstByte().isPositive() ? Math.min(cfg.firstByte().toNanos(), left) : left;
        var sent = client.sendAsync(b.build(), HttpResponse.BodyHandlers.ofInputStream());
        var ready = sent.thenApplyAsync(Gateway::peekFirstByte, VT);
        try {
            return ready.get(wait, TimeUnit.NANOSECONDS);
        } catch (java.util.concurrent.TimeoutException e) {
            sent.cancel(true);                                    // заголовков ещё нет — обрываем запрос,
            sent.thenAccept(r -> new Upstream(r, new BufferedInputStream(r.body())).close()); // а если есть — закрываем тело
            throw new HttpTimeoutException("no first byte in " + Duration.ofNanos(wait));
        } catch (ExecutionException e) {
            throw e.getCause() instanceof Exception c ? c : e;
        }
    }

    /** Ждёт первый байт тела: заголовки могли уйти, а токенов нет — реплика зависла на prefill. */
    private static Upstream peekFirstByte(HttpResponse<InputStream> resp) {
        var up = new Upstream(resp, new BufferedInputStream(resp.body()));
        try {
            up.body().mark(1);
            up.body().read(); // -1 (пустое тело) тоже годится
            up.body().reset();
            return up;
        } catch (IOException e) {
            up.close();
            throw new UncheckedIOException(e);
        }
    }

    private static long retryAfterNanos(HttpResponse<?> resp) {
        try {
            return resp.headers().firstValue("Retry-After")
                    .map(s -> TimeUnit.SECONDS.toNanos(Long.parseLong(s.trim()))).orElse(0L);
        } catch (NumberFormatException e) {
            return 0;
        }
    }

    static boolean retryable(int code) {
        return code == 429 || code == 502 || code == 503 || code == 504;
    }

    /** Пробрасывает стрим построчно и сбрасывает буфер после каждой строки. */
    private static void pipeSSE(HttpExchange ex, InputStream src, long start, Stats st) throws IOException {
        var h = ex.getResponseHeaders();
        h.set("Content-Type", "text/event-stream");
        h.set("Cache-Control", "no-cache");
        h.set("X-Accel-Buffering", "no"); // nginx перед шлюзом не должен копить стрим
        ex.sendResponseHeaders(200, 0);   // 0 — chunked: длина заранее неизвестна
        OutputStream out = ex.getResponseBody();
        var in = new BufferedReader(new InputStreamReader(src, StandardCharsets.UTF_8));
        long first = 0, last = 0;
        int chunks = 0;
        while (true) {
            String line;
            try {
                line = in.readLine();
            } catch (IOException e) {
                break; // upstream оборвал стрим
            }
            if (line == null) {
                break;
            }
            try {
                out.write((line + "\n").getBytes(StandardCharsets.UTF_8));
                out.flush();
            } catch (IOException e) {
                return; // клиент ушёл: Upstream.close() оборвёт запрос к vLLM, генерация остановится
            }
            if (!line.startsWith("data:")) {
                continue;
            }
            JsonNode c;
            try {
                c = JSON.readTree(line.substring(5));
            } catch (JsonProcessingException e) {
                continue; // "[DONE]" не JSON — пропускаем
            }
            // reasoning — у reasoning-моделей; в старых vLLM поле звалось reasoning_content.
            JsonNode delta = c.path("choices").path(0).path("delta");
            if (!delta.path("content").asText("").isEmpty() || !delta.path("reasoning").asText("").isEmpty()) {
                long now = System.nanoTime();
                if (first == 0) {
                    first = now;
                }
                last = now;
                chunks++;
            }
            if (c.hasNonNull("usage")) {
                st.outTokens = c.path("usage").path("completion_tokens").asInt();
            }
        }
        if (st.outTokens == 0) {
            st.outTokens = chunks; // без usage: чанк ≈ токен, если stream_interval = 1
        }
        if (first != 0) {
            st.ttft = Duration.ofNanos(first - start);
            if (st.outTokens > 1) {
                st.tpot = Duration.ofNanos((last - first) / (st.outTokens - 1));
            }
        }
    }

    static void apiError(HttpExchange ex, int code, String msg) throws IOException {
        byte[] b = JSON.writeValueAsBytes(Map.of("error", Map.of("message", msg, "code", code)));
        ex.getResponseHeaders().set("Content-Type", "application/json");
        ex.sendResponseHeaders(code, b.length);
        ex.getResponseBody().write(b);
    }
}

// go run . -upstream http://localhost:8000 -models "clf=ticket-clf,chat=Qwen/Qwen3-8B" -max-inflight 128
// Локально без GPU: -upstream http://localhost:11434 -models "chat=qwen3:8b" (Ollama).
package main

import (
	"flag"
	"log/slog"
	"net/http"
	"os"
	"strings"
	"time"
)

func main() {
	listen := flag.String("listen", ":8080", "адрес шлюза")
	upstream := flag.String("upstream", "http://localhost:8000", "vLLM или Ollama")
	models := flag.String("models", "chat=Qwen/Qwen3-8B", "публичное=имя_в_vLLM через запятую")
	inflight := flag.Int("max-inflight", 128, "≈ --max-num-seqs × реплики")
	flag.Parse()

	m := map[string]string{}
	for _, kv := range strings.Split(*models, ",") {
		if pub, target, ok := strings.Cut(kv, "="); ok {
			m[pub] = target
		}
	}
	gw := New(Config{
		Upstream: *upstream, APIKey: os.Getenv("VLLM_API_KEY"), Models: m,
		MaxInFlight: *inflight, QueueWait: 2 * time.Second,
		FirstByte: 10 * time.Second, Timeout: 5 * time.Minute,
		MaxRetries: 2, Backoff: 200 * time.Millisecond,
		Observe: func(s Stats) {
			slog.Info("llm", "model", s.Model, "status", s.Status, "attempts", s.Attempts,
				"ttft_ms", s.TTFT.Milliseconds(), "tpot_ms", s.TPOT.Milliseconds(), "out_tokens", s.OutTokens)
		},
	})
	srv := &http.Server{Addr: *listen, Handler: gw, ReadHeaderTimeout: 5 * time.Second}
	slog.Info("gateway", "listen", *listen, "upstream", *upstream, "models", m)
	if err := srv.ListenAndServe(); err != nil {
		slog.Error("serve", "err", err)
		os.Exit(1)
	}
}

// mvn -q -pl m10-models compile exec:java -Dexec.mainClass=course.m10models.Main \
//   -Dexec.args='--upstream http://localhost:8000 --models clf=ticket-clf,chat=Qwen/Qwen3-8B --max-inflight 128'
// Локально без GPU: --upstream http://localhost:11434 --models chat=qwen3:8b (Ollama).
package course.m10models;

import com.sun.net.httpserver.HttpServer;
import java.lang.System.Logger.Level;
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executors;

public final class Main {
    public static void main(String[] args) throws Exception {
        Map<String, String> flags = new HashMap<>(Map.of(
                "--listen", ":8080",                       // адрес шлюза
                "--upstream", "http://localhost:8000",     // vLLM или Ollama
                "--models", "chat=Qwen/Qwen3-8B",          // публичное=имя_в_vLLM через запятую
                "--max-inflight", "128"));                 // ≈ --max-num-seqs × реплики
        for (int i = 0; i + 1 < args.length; i += 2) {
            if (!flags.containsKey(args[i])) {
                throw new IllegalArgumentException("unknown flag " + args[i]);
            }
            flags.put(args[i], args[i + 1]);
        }

        Map<String, String> models = new HashMap<>();
        for (String kv : flags.get("--models").split(",")) {
            String[] p = kv.split("=", 2);
            if (p.length == 2) {
                models.put(p[0], p[1]);
            }
        }
        var log = System.getLogger("gateway");
        var gw = new Gateway(new Gateway.Config(
                flags.get("--upstream"), Objects.requireNonNullElse(System.getenv("VLLM_API_KEY"), ""), models,
                Integer.parseInt(flags.get("--max-inflight")), Duration.ofSeconds(2),
                Duration.ofSeconds(10), Duration.ofMinutes(5), 2, Duration.ofMillis(200),
                s -> log.log(Level.INFO, "llm model={0} status={1} attempts={2} ttft_ms={3} tpot_ms={4} out_tokens={5}",
                        s.model, s.status, s.attempts, s.ttft.toMillis(), s.tpot.toMillis(), s.outTokens)));

        String listen = flags.get("--listen");
        int colon = listen.lastIndexOf(':');
        String host = listen.substring(0, colon);
        int port = Integer.parseInt(listen.substring(colon + 1));
        var srv = HttpServer.create(host.isEmpty() ? new InetSocketAddress(port) : new InetSocketAddress(host, port), 0);
        srv.createContext("/", gw);
        srv.setExecutor(Executors.newVirtualThreadPerTaskExecutor()); // поток на запрос: стримы держат соединение минутами
        srv.start();
        log.log(Level.INFO, "gateway listen={0} upstream={1} models={2}", listen, flags.get("--upstream"), models);
    }
}

// fakeVLLM — httptest-сервер, который отвечает как vLLM; handle решает, что вернуть на n-й вызов.
func fakeVLLM(t *testing.T, handle func(n int, w http.ResponseWriter, body map[string]any)) (*httptest.Server, *atomic.Int32) {
	t.Helper()
	var calls atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/v1/chat/completions" || r.Header.Get("Authorization") != "Bearer k" {
			http.Error(w, "bad request to upstream", http.StatusBadRequest)
			return
		}
		var body map[string]any
		_ = json.NewDecoder(r.Body).Decode(&body)
		handle(int(calls.Add(1)), w, body)
	}))
	t.Cleanup(srv.Close)
	return srv, &calls
}

func newGW(upstream string, stats *[]Stats) *Gateway {
	return New(Config{
		Upstream: upstream, APIKey: "k",
		Models:      map[string]string{"clf": "ticket-clf", "chat": "Qwen/Qwen3-8B"},
		MaxInFlight: 4, QueueWait: 50 * time.Millisecond, FirstByte: 200 * time.Millisecond,
		Timeout: 5 * time.Second, MaxRetries: 2, Backoff: time.Millisecond,
		Observe: func(s Stats) { *stats = append(*stats, s) },
	})
}

func post(t *testing.T, gw http.Handler, body string) *httptest.ResponseRecorder {
	t.Helper()
	rec := httptest.NewRecorder()
	gw.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(body)))
	return rec
}

func TestStreamPassthroughAndMetrics(t *testing.T) {
	up, _ := fakeVLLM(t, func(_ int, w http.ResponseWriter, body map[string]any) {
		if so, _ := body["stream_options"].(map[string]any); so["include_usage"] != true {
			t.Errorf("gateway must request usage in stream, got %v", body["stream_options"])
		}
		w.Header().Set("Content-Type", "text/event-stream")
		fl := w.(http.Flusher)
		send := func(s string) { fmt.Fprintf(w, "data: %s\n\n", s); fl.Flush() }
		send(`{"choices":[{"delta":{"role":"assistant","content":""}}]}`)
		time.Sleep(30 * time.Millisecond) // prefill
		for _, tok := range []string{"При", "вет", "!"} {
			send(fmt.Sprintf(`{"choices":[{"delta":{"content":%q}}]}`, tok))
			time.Sleep(10 * time.Millisecond) // decode
		}
		send(`{"choices":[],"usage":{"prompt_tokens":12,"completion_tokens":3}}`)
		send("[DONE]")
	})
	var stats []Stats
	rec := post(t, newGW(up.URL, &stats), `{"model":"chat","stream":true,"messages":[{"role":"user","content":"привет"}]}`)
	out := rec.Body.String()
	if rec.Code != 200 || rec.Header().Get("Content-Type") != "text/event-stream" {
		t.Fatalf("got %d %q", rec.Code, rec.Header().Get("Content-Type"))
	}
	if strings.Index(out, `"При"`) > strings.Index(out, `"вет"`) || !strings.HasSuffix(out, "data: [DONE]\n\n") {
		t.Fatalf("stream is not passed through in order:\n%s", out)
	}
	s := stats[0]
	if s.OutTokens != 3 || s.TTFT < 30*time.Millisecond || s.TPOT < 5*time.Millisecond || s.TPOT > s.TTFT {
		t.Fatalf("stats = %+v", s)
	}
}

func TestNoFirstByteRetries(t *testing.T) {
	up, calls := fakeVLLM(t, func(n int, w http.ResponseWriter, _ map[string]any) {
		w.Header().Set("Content-Type", "text/event-stream")
		w.(http.Flusher).Flush() // заголовки ушли, а токенов нет: реплика зависла на prefill
		if n == 1 {
			time.Sleep(time.Second)
			return
		}
		fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\ndata: [DONE]\n\n")
	})
	var stats []Stats
	rec := post(t, newGW(up.URL, &stats), `{"model":"chat","stream":true,"messages":[]}`)
	if rec.Code != 200 || calls.Load() != 2 || !strings.Contains(rec.Body.String(), `"ok"`) {
		t.Fatalf("code %d, calls %d, body %s", rec.Code, calls.Load(), rec.Body)
	}
}

// Фрагмент GatewayTest (JUnit 5): фейковый vLLM на com.sun.net.httpserver, шлюз на случайном порту.
class GatewayTest {
    static final ObjectMapper JSON = new ObjectMapper();
    static final HttpClient HTTP = HttpClient.newHttpClient();
    final List<HttpServer> servers = new ArrayList<>();
    final List<Gateway.Stats> stats = new CopyOnWriteArrayList<>();

    @FunctionalInterface
    interface Handle {
        void handle(int n, HttpExchange ex, JsonNode body) throws Exception;
    }

    HttpServer serve(com.sun.net.httpserver.HttpHandler h) throws IOException {
        var srv = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
        srv.createContext("/", h);
        srv.setExecutor(Executors.newVirtualThreadPerTaskExecutor());
        srv.start();
        servers.add(srv);
        return srv;
    }

    static String url(HttpServer s) {
        return "http://127.0.0.1:" + s.getAddress().getPort();
    }

    /** Фейковый vLLM: handle решает, что вернуть на n-й вызов. */
    String fakeVLLM(AtomicInteger calls, Handle handle) throws IOException {
        return url(serve(ex -> {
            try (ex) {
                if (!ex.getRequestURI().getPath().equals("/v1/chat/completions")
                        || !"Bearer k".equals(ex.getRequestHeaders().getFirst("Authorization"))) {
                    reply(ex, 400, "text/plain", "bad request to upstream");
                    return;
                }
                handle.handle(calls.incrementAndGet(), ex, JSON.readTree(ex.getRequestBody()));
            } catch (Exception e) {
                throw new IOException(e);
            }
        }));
    }

    static void reply(HttpExchange ex, int code, String contentType, String body) throws IOException {
        byte[] b = body.getBytes(StandardCharsets.UTF_8);
        ex.getResponseHeaders().set("Content-Type", contentType);
        ex.sendResponseHeaders(code, b.length == 0 ? -1 : b.length);
        ex.getResponseBody().write(b);
    }

    static Gateway.Config config(String upstream, int maxInFlight, Consumer<Gateway.Stats> observe) {
        return new Gateway.Config(upstream, "k", Map.of("clf", "ticket-clf", "chat", "Qwen/Qwen3-8B"),
                maxInFlight, Duration.ofMillis(50), Duration.ofMillis(200), Duration.ofSeconds(5), 2,
                Duration.ofMillis(1), observe);
    }

    String newGW(String upstream) throws IOException {
        return url(serve(new Gateway(config(upstream, 4, stats::add))));
    }

    static HttpResponse<String> post(String gw, String body) throws Exception {
        return HTTP.send(HttpRequest.newBuilder(URI.create(gw + "/v1/chat/completions"))
                .POST(HttpRequest.BodyPublishers.ofString(body)).build(), HttpResponse.BodyHandlers.ofString());
    }

    /** observe вызывается после записи ответа: ждём его, а не полагаемся на порядок. */
    Gateway.Stats awaitStats() throws InterruptedException {
        for (int i = 0; i < 200 && stats.isEmpty(); i++) {
            Thread.sleep(5);
        }
        assertEquals(1, stats.size(), "stats = " + stats);
        return stats.getFirst();
    }

    @Test
    void streamPassthroughAndMetrics() throws Exception {
        String up = fakeVLLM(new AtomicInteger(), (n, ex, body) -> {
            assertTrue(body.path("stream_options").path("include_usage").asBoolean(),
                    "gateway must request usage in stream");
            ex.getResponseHeaders().set("Content-Type", "text/event-stream");
            ex.sendResponseHeaders(200, 0);
            var out = ex.getResponseBody();
            Consumer<String> send = s -> {
                try {
                    out.write(("data: " + s + "\n\n").getBytes(StandardCharsets.UTF_8));
                    out.flush();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            };
            send.accept("{\"choices\":[{\"delta\":{\"role\":\"assistant\",\"content\":\"\"}}]}");
            Thread.sleep(30); // prefill
            for (String tok : List.of("При", "вет", "!")) {
                send.accept("{\"choices\":[{\"delta\":{\"content\":\"" + tok + "\"}}]}");
                Thread.sleep(10); // decode
            }
            send.accept("{\"choices\":[],\"usage\":{\"prompt_tokens\":12,\"completion_tokens\":3}}");
            send.accept("[DONE]");
        });
        var rec = post(newGW(up), "{\"model\":\"chat\",\"stream\":true,\"messages\":[{\"role\":\"user\",\"content\":\"привет\"}]}");
        String out = rec.body();
        assertEquals(200, rec.statusCode());
        assertEquals("text/event-stream", rec.headers().firstValue("Content-Type").orElse(""));
        assertTrue(out.indexOf("\"При\"") < out.indexOf("\"вет\"") && out.endsWith("data: [DONE]\n\n"),
                "stream is not passed through in order:\n" + out);
        var s = awaitStats();
        assertEquals(3, s.outTokens, s.toString());
        assertTrue(s.ttft.compareTo(Duration.ofMillis(30)) >= 0, s.toString());
        assertTrue(s.tpot.compareTo(Duration.ofMillis(5)) >= 0 && s.tpot.compareTo(s.ttft) <= 0, s.toString());
    }

    @Test
    void noFirstByteRetries() throws Exception {
        var calls = new AtomicInteger();
        String up = fakeVLLM(calls, (n, ex, body) -> {
            ex.getResponseHeaders().set("Content-Type", "text/event-stream");
            ex.sendResponseHeaders(200, 0); // заголовки ушли, а токенов нет: реплика зависла на prefill
            if (n == 1) {
                Thread.sleep(1000);
                return;
            }
            ex.getResponseBody().write("data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\ndata: [DONE]\n\n"
                    .getBytes(StandardCharsets.UTF_8));
        });
        var rec = post(newGW(up), "{\"model\":\"chat\",\"stream\":true,\"messages\":[]}");
        assertEquals(200, rec.statusCode());
        assertEquals(2, calls.get());
        assertTrue(rec.body().contains("\"ok\""), rec.body());
    }
}

# Полный файл с офлайн-учителем (StubTeacher) и main — examples/m10-models/synth.py
SYSTEM = "Классифицируй тикет поддержки. Ответ: одна метка из списка: billing, delivery, account, bug, other."

LABELS = {"billing": "оплата, двойные списания, возвраты, счета", "delivery": "сроки и место доставки, курьер",
          "account": "вход, пароль, профиль, коды подтверждения", "bug": "ошибки и падения сайта или приложения",
          "other": "всё остальное: вопросы о магазине, отзывы, сотрудничество"}

def norm(t: str) -> str:
    return re.sub(r"\s+", " ", t.lower()).strip()

def shingles(t: str, k: int = 5) -> set[str]:
    t = norm(t)
    return {t[i:i + k] for i in range(max(1, len(t) - k + 1))}

def jaccard(a: set[str], b: set[str]) -> float:
    return len(a & b) / len(a | b) if a | b else 1.0

class ClaudeTeacher:
    def __init__(self) -> None:
        import anthropic
        self.client = anthropic.Anthropic()  # ANTHROPIC_API_KEY из окружения

    def _text(self, **kw) -> str:
        resp = self.client.messages.create(**kw)
        return "".join(b.text for b in resp.content if b.type == "text")

    def generate(self, label: str, persona: str, n: int) -> list[str]:
        text = self._text(model="claude-opus-5", max_tokens=8000, messages=[{"role": "user", "content":
            f"Сгенерируй {n} разных обращений в поддержку интернет-магазина на тему «{label}» ({LABELS[label]}). "
            f"Автор: {persona}. Варьируй длину (5–60 слов), детали и первые слова. "
            "Не упоминай название темы. Верни только JSON-массив строк."}])
        return json.loads(text[text.find("["):text.rfind("]") + 1])

    def label(self, text: str) -> str:  # разметчик не знает, под какую метку генерировали текст
        words = self._text(model="claude-haiku-4-5", max_tokens=10, system=SYSTEM,
                           messages=[{"role": "user", "content": text}]).split()
        return words[0].strip(".,").lower() if words else ""

def filter_examples(cands: list[tuple[str, str]], eval_texts: list[str], labeler, sim: float = 0.8):
    """Дешёвые проверки первыми, вызов LLM-разметчика — последним. Порог sim подбирают по выборке глазами."""
    kept, dropped, hashes, kept_sh = [], Counter(), set(), []
    eval_sh = [shingles(t) for t in eval_texts]
    for text, label in cands:
        if label not in LABELS or not 15 <= len(text) <= 2000:
            dropped["схема или длина"] += 1
            continue
        h = hashlib.sha256(norm(text).encode()).hexdigest()
        if h in hashes:
            dropped["точный дубль"] += 1
            continue
        sh = shingles(text)
        if any(jaccard(sh, e) >= sim for e in eval_sh):
            dropped["похож на eval"] += 1
            continue
        if any(jaccard(sh, k) >= sim for k in kept_sh):  # ponytail: O(n²); от ~10⁴ примеров — MinHash/LSH
            dropped["почти дубль"] += 1
            continue
        if labeler(text) != label:
            dropped["разметчик не согласен"] += 1
            continue
        hashes.add(h)
        kept_sh.append(sh)
        kept.append((text, label))
    return kept, dropped

# pip install scikit-learn pandas
import hashlib, json, re
from collections import Counter

import pandas as pd
from sklearn.model_selection import train_test_split

SYSTEM = "Классифицируй тикет поддержки. Ответ: одна метка из списка: billing, delivery, account, bug, other."

def norm(text: str) -> str:
    text = re.sub(r"\b(order|заказ)\s*#?\d+", "<ID>", text.lower())   # id не делают тексты разными
    return re.sub(r"\s+", " ", text).strip()

def to_chat(row) -> dict:
    return {"messages": [
        {"role": "system", "content": SYSTEM},          # тот же промпт, что будет в проде
        {"role": "user", "content": row.text},
        {"role": "assistant", "content": row.label},
    ]}

df = pd.read_csv("tickets_labeled.csv")                  # колонки: text, label
df = df.dropna(subset=["text", "label"])
df["h"] = df.text.map(lambda t: hashlib.sha256(norm(t).encode()).hexdigest())

# 1) Дедупликация ДО разбиения. Конфликтующие метки у одного текста — сигнал о шуме разметки.
conflicts = df.groupby("h").label.nunique()
print("текстов с конфликтующими метками:", int((conflicts > 1).sum()))
df = df[df.h.isin(conflicts[conflicts == 1].index)].drop_duplicates("h")

# 2) Стратифицированное разбиение 80/10/10: редкие классы есть в каждой части.
train, rest = train_test_split(df, test_size=0.2, stratify=df.label, random_state=42)
val, test = train_test_split(rest, test_size=0.5, stratify=rest.label, random_state=42)

# 3) Проверка утечки: ни один нормализованный текст не встречается в двух сплитах.
assert not (set(train.h) & set(test.h)) and not (set(train.h) & set(val.h)), "утечка!"

for name, part in [("train", train), ("val", val), ("test", test)]:
    with open(f"{name}.jsonl", "w", encoding="utf-8") as f:
        for row in part.itertuples():
            f.write(json.dumps(to_chat(row), ensure_ascii=False) + "\n")
    print(name, len(part), dict(Counter(part.label)))

# Почти-дубликаты (MinHash/embeddings) и групповое разбиение по клиенту — следующий шаг,
# если в данных есть шаблонные тексты или несколько сообщений одного треда.

"""LoRA-дообучение малой модели под классификацию тикетов (TRL + PEFT) с оценкой до и после.

GPU с bf16 (Colab L4/A100, RTX 30xx+, H100), данные из скрипта подготовки датасета модуля 10:
  uv run --python 3.12 --with trl==1.14.2 --with peft==0.21.2 --with transformers==5.19.0 \
      --with datasets==5.1.0 --with accelerate==1.15.0 --with torch \
      python lora_sft.py --train train.jsonl --val val.jsonl --test test.jsonl --model Qwen/Qwen3-4B
CPU dry-run (проверка, что пайплайн собирается; метрики на демо-данных ничего не значат):
  ... python lora_sft.py --demo --model trl-internal-testing/tiny-Qwen3ForCausalLM --cpu --max-steps 2
"""
import argparse
import random

import torch
from datasets import Dataset, load_dataset
from peft import LoraConfig
from transformers import EarlyStoppingCallback
from trl import SFTConfig, SFTTrainer

SYSTEM = "Классифицируй тикет поддержки. Ответ: одна метка из списка: billing, delivery, account, bug, other."
LABELS = ["billing", "delivery", "account", "bug", "other"]


def demo_split(n: int, seed: int) -> Dataset:
    texts = {
        "billing": ["Списали деньги дважды за заказ", "Не пришёл возврат", "Счёт больше, чем сумма в корзине"],
        "delivery": ["Курьер не приехал", "Где мой заказ, прошла неделя", "Привезли не в тот пункт выдачи"],
        "account": ["Не могу войти в аккаунт", "Как сменить email в профиле?", "Не приходит код подтверждения"],
        "bug": ["Приложение падает при оплате", "Кнопка «Купить» не нажимается", "Ошибка 500 на странице заказа"],
        "other": ["Есть ли подарочные карты?", "Хочу похвалить курьера", "Работаете ли вы в праздники?"],
    }
    rnd = random.Random(seed)
    rows = []
    for _ in range(n):
        label = rnd.choice(LABELS)
        rows.append({"messages": [{"role": "system", "content": SYSTEM},
                                  {"role": "user", "content": rnd.choice(texts[label])},
                                  {"role": "assistant", "content": label}]})
    return Dataset.from_list(rows)


@torch.no_grad()
def correct(model, tok, rows: Dataset) -> list[int]:
    """1/0 на каждый пример: совпала ли первая сгенерированная метка с эталоном (greedy)."""
    model.eval()
    out = []
    for ex in rows:
        prompt = tok.apply_chat_template(ex["messages"][:-1], add_generation_prompt=True, tokenize=False,
                                         enable_thinking=False)  # Qwen3: без <think>; другие шаблоны игнорируют
        ids = tok(prompt, return_tensors="pt", add_special_tokens=False).to(model.device)
        gen = model.generate(**ids, max_new_tokens=6, do_sample=False)
        words = tok.decode(gen[0, ids["input_ids"].shape[1]:], skip_special_tokens=True).split()
        out.append(int(bool(words) and words[0].strip(".,") == ex["messages"][-1]["content"]))
    return out


def paired_bootstrap(before: list[int], after: list[int], n: int = 10_000, seed: int = 0) -> tuple[float, float]:
    """95% ДИ разницы accuracy (after − before) на ОДНИХ И ТЕХ ЖЕ примерах: ресэмплим пары, а не системы по отдельности."""
    rnd, k = random.Random(seed), len(before)
    diffs = sorted(sum(after[i] - before[i] for i in (rnd.randrange(k) for _ in range(k))) / k for _ in range(n))
    return diffs[int(0.025 * n)], diffs[int(0.975 * n)]


def main() -> None:
    p = argparse.ArgumentParser()
    p.add_argument("--model", default="Qwen/Qwen3-4B")
    for f in ("--train", "--val", "--test"):
        p.add_argument(f)
    p.add_argument("--demo", action="store_true", help="встроенные игрушечные данные вместо JSONL")
    p.add_argument("--cpu", action="store_true")
    p.add_argument("--max-steps", type=int, default=-1, help="-1 — обучать по эпохам")
    p.add_argument("--out", default="out/ticket-clf-lora")
    p.add_argument("--merge", action="store_true", help="дополнительно сохранить модель со слитым адаптером")
    a = p.parse_args()

    if a.demo:
        train, val, test = demo_split(48, 1), demo_split(8, 2), demo_split(8, 3)
    else:
        files = {"train": a.train, "validation": a.val, "test": a.test or a.val}
        data = load_dataset("json", data_files=files)
        train, val, test = data["train"], data["validation"], data["test"]

    # bf16 есть на Ampere и новее (L4, A100, RTX 30xx+, H100). На T4 и CPU — fp32: берите модель ≤ 1.7B или QLoRA.
    bf16 = torch.cuda.is_available() and torch.cuda.is_bf16_supported() and not a.cpu
    peft_config = LoraConfig(
        r=16,                         # ёмкость адаптера
        lora_alpha=32,                # масштаб ΔW = (alpha / r) · B·A
        lora_dropout=0.05,
        target_modules="all-linear",  # q,k,v,o + MLP: ближе к full FT, чем только attention
        task_type="CAUSAL_LM",
    )
    args = SFTConfig(
        output_dir=a.out,
        model_init_kwargs={"dtype": torch.bfloat16 if bf16 else torch.float32},
        bf16=bf16,                    # в SFTConfig по умолчанию True: без bf16 в железе обучение упадёт
        use_cpu=a.cpu,
        assistant_only_loss=True,     # loss только на ответе; TRL сам подставит шаблон с {% generation %} для Qwen, Llama, Gemma
        max_length=1024,              # длинные примеры обрезаются: проверьте, что метки не теряются
        num_train_epochs=3,
        max_steps=a.max_steps,
        learning_rate=1e-4,           # для LoRA выше, чем для full FT (дефолт SFTConfig 2e-5)
        per_device_train_batch_size=8,
        gradient_accumulation_steps=2,
        eval_strategy="steps", eval_steps=2 if a.demo else 50,
        save_strategy="steps", save_steps=2 if a.demo else 50, save_total_limit=2,
        load_best_model_at_end=True, metric_for_best_model="eval_loss", greater_is_better=False,
        logging_steps=1 if a.demo else 10, report_to="none", seed=42,
    )
    trainer = SFTTrainer(
        model=a.model,                # строка: тренер сам загрузит модель и токенизатор
        args=args,
        train_dataset=train,
        eval_dataset=val,
        peft_config=peft_config,
        callbacks=[EarlyStoppingCallback(early_stopping_patience=3)],
        # QLoRA: quantization_config=BitsAndBytesConfig(load_in_4bit=True, bnb_4bit_quant_type="nf4",
        #          bnb_4bit_compute_dtype=torch.bfloat16, bnb_4bit_use_double_quant=True)
    )
    tok = trainer.processing_class
    trainer.model.print_trainable_parameters()

    before = correct(trainer.model, tok, test)  # B = 0 → адаптер пока ничего не меняет, это базовая модель
    trainer.train()
    after = correct(trainer.model, tok, test)
    lo, hi = paired_bootstrap(before, after)
    print(f"test accuracy: до {sum(before) / len(before):.3f} → после {sum(after) / len(after):.3f}, "
          f"разница 95% ДИ [{lo:+.3f}; {hi:+.3f}] на {len(test)} примерах")

    trainer.save_model(f"{a.out}/best")  # только адаптер: десятки МБ
    if a.merge:  # одна модель без адаптера: проще сервить, но теряется multi-LoRA
        trainer.model.merge_and_unload().save_pretrained(f"{a.out}/merged")
        tok.save_pretrained(f"{a.out}/merged")
    print(f"vllm serve {a.model} --enable-lora --lora-modules ticket-clf={a.out}/best --max-lora-rank 16")


if __name__ == "__main__":
    main()

# pip install scikit-learn numpy
import numpy as np
from sklearn.metrics import (average_precision_score, classification_report,
                             confusion_matrix, precision_recall_curve,
                             precision_recall_fscore_support, roc_auc_score)

def pick_threshold(y_val, s_val, min_precision=0.95):
    """Порог на VALIDATION: максимальный recall при precision ≥ min_precision."""
    p, r, t = precision_recall_curve(y_val, s_val)       # len(t) == len(p) - 1
    ok = np.where(p[:-1] >= min_precision)[0]
    if len(ok) == 0:
        raise ValueError(f"precision {min_precision} недостижима: модель или данные слабые")
    best = ok[np.argmax(r[:-1][ok])]
    return float(t[best])

def report(y_true, scores, threshold):
    y_pred = (scores >= threshold).astype(int)
    p, r, f1, _ = precision_recall_fscore_support(y_true, y_pred, average="binary", zero_division=0)
    tn, fp, fn, tp = confusion_matrix(y_true, y_pred).ravel()
    print(f"threshold={threshold:.3f}  P={p:.3f} R={r:.3f} F1={f1:.3f}")
    print(f"TP={tp} FP={fp} FN={fn} TN={tn}  accuracy={(tp + tn) / len(y_true):.3f}"
          f"  (у «всегда нет»: {1 - y_true.mean():.3f})")
    print(f"ROC-AUC={roc_auc_score(y_true, scores):.3f}  PR-AUC={average_precision_score(y_true, scores):.3f}")

# y_* — 0/1 (например, «нужна эскалация»), s_* — скор модели (logprob метки, вероятность головы).
y_val, s_val = np.load("val_y.npy"), np.load("val_scores.npy")
y_test, s_test = np.load("test_y.npy"), np.load("test_scores.npy")

th = pick_threshold(y_val, s_val, min_precision=0.95)   # выбираем на val…
report(y_test, s_test, th)                               # …а честную оценку даёт test

# Мультикласс: macro-F1 показывает редкие классы, weighted — «средний» опыт пользователя.
labels_true = np.load("test_labels.npy", allow_pickle=True)
labels_pred = np.load("test_pred_labels.npy", allow_pickle=True)
print(classification_report(labels_true, labels_pred, digits=3, zero_division=0))

// Валидатор и дедупликатор чат-датасета: go run . -in train.jsonl -out clean.jsonl
// Падает (exit 1) при ошибках схемы — удобно в CI перед запуском обучения.
package main

import (
	"bufio"
	"crypto/sha256"
	"encoding/json"
	"flag"
	"fmt"
	"os"
	"strings"
)

type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}
type Example struct {
	Messages []Message `json:"messages"`
}

func validate(ex Example, maxChars int) error {
	if len(ex.Messages) < 2 {
		return fmt.Errorf("меньше двух сообщений")
	}
	prev, total := "", 0
	for i, m := range ex.Messages {
		switch {
		case m.Role != "system" && m.Role != "user" && m.Role != "assistant":
			return fmt.Errorf("msg %d: неизвестная роль %q", i, m.Role)
		case m.Role == "system" && i != 0:
			return fmt.Errorf("msg %d: system не первым", i)
		case strings.TrimSpace(m.Content) == "":
			return fmt.Errorf("msg %d: пустой content", i)
		case m.Role == prev && m.Role != "system":
			return fmt.Errorf("msg %d: роли не чередуются", i)
		}
		prev, total = m.Role, total+len([]rune(m.Content))
	}
	if last := ex.Messages[len(ex.Messages)-1]; last.Role != "assistant" {
		return fmt.Errorf("последнее сообщение не assistant: учить нечему")
	}
	if total > maxChars { // грубо ~3–4 символа на токен; точнее — токенизатором модели
		return fmt.Errorf("слишком длинный пример: %d символов", total)
	}
	return nil
}

// key: нормализованный текст всех не-assistant сообщений. Одинаковый вход с разными
// ответами — это конфликт разметки, его стоит показать человеку, а не молча выбросить.
func key(ex Example) [32]byte {
	var b strings.Builder
	for _, m := range ex.Messages {
		if m.Role != "assistant" {
			b.WriteString(strings.Join(strings.Fields(strings.ToLower(m.Content)), " "))
			b.WriteByte(0)
		}
	}
	return sha256.Sum256([]byte(b.String()))
}

func main() {
	in := flag.String("in", "train.jsonl", "входной JSONL")
	out := flag.String("out", "clean.jsonl", "выход без дубликатов")
	maxChars := flag.Int("max-chars", 16000, "лимит длины примера")
	flag.Parse()

	f, err := os.Open(*in)
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	defer f.Close()
	w, err := os.Create(*out)
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	defer w.Close()
	bw := bufio.NewWriter(w)
	defer bw.Flush()

	seen := map[[32]byte]string{} // ключ → ответ ассистента
	sc := bufio.NewScanner(f)
	sc.Buffer(make([]byte, 0, 1<<20), 16<<20) // строки JSONL бывают длиннее 64 КБ
	var line, bad, dups, conflicts int
	for sc.Scan() {
		line++
		var ex Example
		if err := json.Unmarshal(sc.Bytes(), &ex); err != nil {
			bad++
			fmt.Fprintf(os.Stderr, "line %d: невалидный JSON: %v\n", line, err)
			continue
		}
		if err := validate(ex, *maxChars); err != nil {
			bad++
			fmt.Fprintf(os.Stderr, "line %d: %v\n", line, err)
			continue
		}
		k, ans := key(ex), ex.Messages[len(ex.Messages)-1].Content
		if prev, ok := seen[k]; ok {
			if prev != ans {
				conflicts++
				fmt.Fprintf(os.Stderr, "line %d: тот же вход, другой ответ: %q vs %q\n", line, prev, ans)
			}
			dups++
			continue
		}
		seen[k] = ans
		bw.Write(sc.Bytes())
		bw.WriteByte('\n')
	}
	if err := sc.Err(); err != nil {
		fmt.Fprintln(os.Stderr, "read:", err)
		os.Exit(1)
	}
	fmt.Printf("строк %d, ок %d, ошибок %d, дубликатов %d (конфликтов %d)\n",
		line, len(seen), bad, dups, conflicts)
	if bad > 0 {
		os.Exit(1)
	}
}

// Валидатор и дедупликатор чат-датасета: java -cp ... dataset.JsonlValidator --in train.jsonl --out clean.jsonl
// Падает (exit 1) при ошибках схемы — удобно в CI перед запуском обучения.
package dataset;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.HashMap;
import java.util.HexFormat;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

public final class JsonlValidator {
    record Message(String role, String content) {}

    record Example(List<Message> messages) {}

    static final ObjectMapper JSON = new ObjectMapper()
            .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    static final Set<String> ROLES = Set.of("system", "user", "assistant");

    /** Бросает IllegalArgumentException с причиной, если пример не годится для обучения. */
    static void validate(Example ex, int maxChars) {
        List<Message> ms = Objects.requireNonNullElse(ex.messages(), List.of());
        if (ms.size() < 2) {
            throw new IllegalArgumentException("меньше двух сообщений");
        }
        String prev = "";
        int total = 0;
        for (int i = 0; i < ms.size(); i++) {
            String role = Objects.requireNonNullElse(ms.get(i).role(), "");
            String content = Objects.requireNonNullElse(ms.get(i).content(), "");
            if (!ROLES.contains(role)) {
                throw new IllegalArgumentException("msg %d: неизвестная роль \"%s\"".formatted(i, role));
            } else if (role.equals("system") && i != 0) {
                throw new IllegalArgumentException("msg %d: system не первым".formatted(i));
            } else if (content.isBlank()) {
                throw new IllegalArgumentException("msg %d: пустой content".formatted(i));
            } else if (role.equals(prev) && !role.equals("system")) {
                throw new IllegalArgumentException("msg %d: роли не чередуются".formatted(i));
            }
            prev = role;
            total += content.codePointCount(0, content.length());
        }
        if (!ms.getLast().role().equals("assistant")) {
            throw new IllegalArgumentException("последнее сообщение не assistant: учить нечему");
        }
        if (total > maxChars) { // грубо ~3–4 символа на токен; точнее — токенизатором модели
            throw new IllegalArgumentException("слишком длинный пример: %d символов".formatted(total));
        }
    }

    // key: нормализованный текст всех не-assistant сообщений. Одинаковый вход с разными
    // ответами — это конфликт разметки, его стоит показать человеку, а не молча выбросить.
    static String key(Example ex) throws NoSuchAlgorithmException {
        var b = new StringBuilder();
        for (Message m : ex.messages()) {
            if (!m.role().equals("assistant")) {
                b.append(String.join(" ", m.content().toLowerCase(Locale.ROOT).strip().split("\\s+"))).append('\0');
            }
        }
        byte[] h = MessageDigest.getInstance("SHA-256").digest(b.toString().getBytes(StandardCharsets.UTF_8));
        return HexFormat.of().formatHex(h);
    }

    public static void main(String[] args) throws Exception {
        Map<String, String> flags = new HashMap<>(Map.of(
                "--in", "train.jsonl",     // входной JSONL
                "--out", "clean.jsonl",    // выход без дубликатов
                "--max-chars", "16000"));  // лимит длины примера
        for (int i = 0; i + 1 < args.length; i += 2) {
            flags.put(args[i], args[i + 1]);
        }
        int maxChars = Integer.parseInt(flags.get("--max-chars"));

        Map<String, String> seen = new HashMap<>(); // ключ → ответ ассистента
        int line = 0, bad = 0, dups = 0, conflicts = 0;
        // BufferedReader не ограничивает длину строки: длинные примеры JSONL читаются целиком.
        try (BufferedReader in = Files.newBufferedReader(Path.of(flags.get("--in")));
             BufferedWriter out = Files.newBufferedWriter(Path.of(flags.get("--out")))) {
            for (String raw; (raw = in.readLine()) != null; ) {
                line++;
                Example ex;
                try {
                    ex = JSON.readValue(raw, Example.class);
                    validate(ex, maxChars);
                } catch (JsonProcessingException e) {
                    bad++;
                    System.err.printf("line %d: невалидный JSON: %s%n", line, e.getOriginalMessage());
                    continue;
                } catch (IllegalArgumentException e) {
                    bad++;
                    System.err.printf("line %d: %s%n", line, e.getMessage());
                    continue;
                }
                String k = key(ex), ans = ex.messages().getLast().content();
                String prev = seen.putIfAbsent(k, ans);
                if (prev != null) {
                    if (!prev.equals(ans)) {
                        conflicts++;
                        System.err.printf("line %d: тот же вход, другой ответ: \"%s\" vs \"%s\"%n", line, prev, ans);
                    }
                    dups++;
                    continue;
                }
                out.write(raw);
                out.newLine();
            }
        } catch (IOException e) {
            System.err.println("read: " + e);
            System.exit(1);
        }
        System.out.printf("строк %d, ок %d, ошибок %d, дубликатов %d (конфликтов %d)%n",
                line, seen.size(), bad, dups, conflicts);
        if (bad > 0) {
            System.exit(1);
        }
    }
}

# Полный файл с демо-парами и запуском — examples/m10-models/dpo.py
def main() -> None:
    p = argparse.ArgumentParser()
    p.add_argument("--model", default="Qwen/Qwen3-4B", help="лучше — уже прошедшая SFT модель")
    for f in ("--train", "--val"):
        p.add_argument(f)
    p.add_argument("--demo", action="store_true")
    p.add_argument("--cpu", action="store_true")
    p.add_argument("--max-steps", type=int, default=-1)
    p.add_argument("--out", default="out/support-dpo")
    a = p.parse_args()

    if a.demo:
        train, val = demo_rows(), demo_rows().select(range(4))
    else:
        data = load_dataset("json", data_files={"train": a.train, "validation": a.val})
        train, val = data["train"], data["validation"]

    bf16 = torch.cuda.is_available() and torch.cuda.is_bf16_supported() and not a.cpu
    args = DPOConfig(
        output_dir=a.out,
        model_init_kwargs={"dtype": torch.bfloat16 if bf16 else torch.float32},
        bf16=bf16,
        use_cpu=a.cpu,
        beta=0.1,               # сила «поводка» к референсной модели: больше β — меньше уход от неё
        loss_type=["sigmoid"],  # классический DPO; "ipo" устойчивее к шуму в парах
        learning_rate=5e-6,     # DPO чувствителен к LR: на порядок ниже, чем для SFT
        max_length=1024,        # prompt + ответ
        num_train_epochs=1,
        max_steps=a.max_steps,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        eval_strategy="steps", eval_steps=2 if a.demo else 50,
        logging_steps=1 if a.demo else 10, report_to="none", seed=42,
    )
    trainer = DPOTrainer(
        model=a.model,
        ref_model=None,  # с peft_config референсом служит та же модель с выключенным адаптером: вторая копия не нужна
        args=args,
        train_dataset=train,
        eval_dataset=val,
        peft_config=LoraConfig(r=16, lora_alpha=32, lora_dropout=0.05, target_modules="all-linear", task_type="CAUSAL_LM"),
    )
    trainer.train()
    m = trainer.evaluate()
    # rewards/accuracies — доля пар, где chosen получил больший неявный reward, чем rejected;
    # margins — средний разрыв. Растут — модель учится различать; ушли в потолок за пару шагов — проверьте утечку или β.
    print({k: round(v, 4) for k, v in m.items() if "rewards" in k or k == "eval_loss"})
    trainer.save_model(f"{a.out}/best")