Go и ИИ

type CompletionRequest struct {
    Model       string    [INLINE]json:"model"[/INLINE]
    Messages    []Message `json:"messages"`
    Temperature float64   [INLINE]json:"temperature"[/INLINE]
    MaxTokens   int       [INLINE]json:"max_tokens"[/INLINE]
}
 
type Message struct {
    Role    string `json:"role"`
    Content string `json:"content"`
}
 
// Отправка запроса с retry-логикой
func (c *OpenAIClient) Complete(ctx context.Context, req CompletionRequest) (*CompletionResponse, error) {
    var resp *CompletionResponse
    
    operation := func() error {
        httpReq, err := c.buildRequest(ctx, req)
        if err != nil {
            return backoff.Permanent(err) // Не ретраим ошибки построения запроса
        }
        
        httpResp, err := c.httpClient.Do(httpReq)
        if err != nil {
            return err // Сетевые ошибки ретраим
        }
        defer httpResp.Body.Close()
        
        if httpResp.StatusCode == 429 {
            return fmt.Errorf("rate limit exceeded") // Ретраим
        }
        
        if httpResp.StatusCode != 200 {
            return backoff.Permanent(fmt.Errorf("unexpected status: %d", httpResp.StatusCode))
        }
        
        return json.NewDecoder(httpResp.Body).Decode(&resp)
    }
    
    // Экспоненциальный backoff с jitter
    backoffStrategy := backoff.NewExponentialBackOff()
    backoffStrategy.MaxElapsedTime = 30 * time.Second
    
    if err := backoff.Retry(operation, backoff.WithContext(backoffStrategy, ctx)); err != nil {
        return nil, err
    }
    
    return resp, nil
}

package inference
 
/*
#cgo CFLAGS: -I/usr/local/libtorch/include
#cgo LDFLAGS: -L/usr/local/libtorch/lib -ltorch -lc10 -ltorch_cpu
 
#include <torch/script.h>
#include <stdlib.h>
 
// Обёртка для загрузки модели
void* load_model(const char* path) {
    try {
        torch::jit::script::Module* module = new torch::jit::script::Module;
        *module = torch::jit::load(path);
        module->eval();
        return module;
    } catch (const std::exception& e) {
        return NULL;
    }
}
 
// Обёртка для инференса
float* run_inference(void* model_ptr, float* input_data, int size) {
    torch::jit::script::Module* module = (torch::jit::script::Module*)model_ptr;
    
    std::vector<float> vec(input_data, input_data + size);
    torch::Tensor input_tensor = torch::from_blob(vec.data(), {1, size});
    
    std::vector<torch::jit::IValue> inputs;
    inputs.push_back(input_tensor);
    
    torch::Tensor output = module->forward(inputs).toTensor();
    
    float* result = (float*)malloc(output.numel() * sizeof(float));
    memcpy(result, output.data_ptr<float>(), output.numel() * sizeof(float));
    
    return result;
}
*/
import "C"
import (
    "fmt"
    "unsafe"
)
 
type TorchModel struct {
    ptr unsafe.Pointer
}
 
func LoadModel(path string) (*TorchModel, error) {
    cPath := C.CString(path)
    defer C.free(unsafe.Pointer(cPath))
    
    ptr := C.load_model(cPath)
    if ptr == nil {
        return nil, fmt.Errorf("failed to load model from %s", path)
    }
    
    return &TorchModel{ptr: ptr}, nil
}
 
func (m *TorchModel) Predict(input []float32) ([]float32, error) {
    // Конвертация Go-слайса в C-массив
    cInput := (*C.float)(unsafe.Pointer(&input[0]))
    
    result := C.run_inference(m.ptr, cInput, C.int(len(input)))
    defer C.free(unsafe.Pointer(result))
    
    // Копируем результат обратно в Go
    // В реальном коде нужно знать размер выхода
    output := make([]float32, 10)
    outputPtr := (*[1 << 30]C.float)(unsafe.Pointer(result))
    for i := range output {
        output[i] = float32(outputPtr[i])
    }
    
    return output, nil
}

package main
 
import (
    "fmt"
    ort "github.com/yalue/onnxruntime_go"
)
 
type ImageClassifier struct {
    session *ort.Session
    input   *ort.Tensor
    output  *ort.Tensor
}
 
func NewClassifier(modelPath string) (*ImageClassifier, error) {
    // Инициализация библиотеки (один раз за lifetime приложения)
    if err := ort.InitializeEnvironment(); err != nil {
        return nil, fmt.Errorf("init environment: %w", err)
    }
    
    // Создание сессии с дефолтными опциями
    session, err := ort.NewSession(modelPath, ort.NewSessionOptions())
    if err != nil {
        return nil, fmt.Errorf("create session: %w", err)
    }
    
    // Получаем информацию о входах/выходах модели
    inputInfo, err := session.GetInputTypeInfo(0)
    if err != nil {
        return nil, fmt.Errorf("get input info: %w", err)
    }
    
    // Создаем тензор нужной формы
    inputShape := inputInfo.GetTensorTypeAndShape().GetShape()
    input, err := ort.NewTensor(inputShape, []float32{})
    if err != nil {
        return nil, fmt.Errorf("create input tensor: %w", err)
    }
    
    return &ImageClassifier{
        session: session,
        input:   input,
    }, nil
}
 
func (c *ImageClassifier) Classify(imageData []float32) ([]float32, error) {
    // Копируем данные в тензор
    if err := c.input.SetData(imageData); err != nil {
        return nil, fmt.Errorf("set input data: %w", err)
    }
    
    // Запускаем инференс
    outputs, err := c.session.Run([]ort.Value{c.input})
    if err != nil {
        return nil, fmt.Errorf("run inference: %w", err)
    }
    defer outputs[0].Destroy()
    
    // Извлекаем результат
    outputTensor := outputs[0].GetTensor()
    result := outputTensor.GetData().([]float32)
    
    return result, nil
}
 
func (c *ImageClassifier) Close() {
    if c.input != nil {
        c.input.Destroy()
    }
    if c.session != nil {
        c.session.Destroy()
    }
}

func ProcessImage(ctx context.Context, imgData []byte) (*ProcessedImage, error) {
    // Канал для сбора результатов параллельных операций
    type result struct {
        thumbnails map[string][]byte
        colors     []Color
        objects    []DetectedObject
        err        error
    }
    
    results := make(chan result, 3)
    
    // Параллельная генерация превьюшек
    go func() {
        thumbs, err := generateThumbnails(imgData, []int{150, 300, 600})
        results <- result{thumbnails: thumbs, err: err}
    }()
    
    // Параллельный анализ цветов
    go func() {
        colors, err := extractDominantColors(imgData, 5)
        results <- result{colors: colors, err: err}
    }()
    
    // Параллельная детекция объектов
    go func() {
        objects, err := detectObjects(imgData)
        results <- result{objects: objects, err: err}
    }()
    
    // Собираем результаты
    processed := &ProcessedImage{}
    for i := 0; i < 3; i++ {
        select {
        case r := <-results:
            if r.err != nil {
                return nil, r.err
            }
            if r.thumbnails != nil {
                processed.Thumbnails = r.thumbnails
            }
            if r.colors != nil {
                processed.Colors = r.colors
            }
            if r.objects != nil {
                processed.Objects = r.objects
            }
        case <-ctx.Done():
            return nil, ctx.Err()
        }
    }
    
    return processed, nil
}

func ProcessMetricsBatch(metrics []Metric, model *AnomalyDetector) []Result {
    resultsCh := make(chan Result, len(metrics))
    semaphore := make(chan struct{}, runtime.NumCPU()*2) // Ограничиваем параллелизм
    
    for _, m := range metrics {
        m := m // Захват переменной для горутины
        semaphore <- struct{}{} // Берём слот
        
        go func() {
            defer func() { <-semaphore }() // Освобождаем слот
            
            // Препроцессинг в отдельной горутине
            normalized := normalizeMetric(m)
            features := extractFeatures(normalized)
            
            // Модель вызываем синхронно (она thread-safe)
            score := model.PredictAnomaly(features)
            
            resultsCh <- Result{
                MetricID: m.ID,
                Score:    score,
                IsAnomaly: score > 0.95,
            }
        }()
    }
    
    // Собираем результаты
    results := make([]Result, 0, len(metrics))
    for i := 0; i < len(metrics); i++ {
        results = append(results, <-resultsCh)
    }
    
    return results
}

func PrepareTrainingData(files []string, workers int) ([]FeatureVector, error) {
    jobs := make(chan string, len(files))
    results := make(chan []FeatureVector, workers)
    
    // Пул воркеров для обработки файлов
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for filePath := range jobs {
                // Скачиваем и распаковываем
                data, err := downloadAndDecompress(filePath)
                if err != nil {
                    continue
                }
                
                // Параллельная обработка строк внутри файла
                vectors := processLogLines(data)
                results <- vectors
            }
        }()
    }
    
    // Распределяем файлы по воркерам
    go func() {
        for _, f := range files {
            jobs <- f
        }
        close(jobs)
    }()
    
    // Ждём завершения и собираем результаты
    go func() {
        wg.Wait()
        close(results)
    }()
    
    var allVectors []FeatureVector
    for vectors := range results {
        allVectors = append(allVectors, vectors...)
    }
    
    return allVectors, nil
}

type FrameDistributor struct {
    frames   chan Frame
    results  chan AnalysisResult
    workers  map[string]chan Frame
    mu       sync.RWMutex
}
 
func (d *FrameDistributor) RegisterWorker(workerID string) <-chan Frame {
    d.mu.Lock()
    defer d.mu.Unlock()
    
    workerChan := make(chan Frame, 10)
    d.workers[workerID] = workerChan
    
    // Горутина для распределения задач этому воркеру
    go func() {
        for frame := range d.frames {
            select {
            case workerChan <- frame:
            case <-time.After(5 * time.Second):
                // Воркер не отвечает, возвращаем в очередь
                d.frames <- frame
            }
        }
    }()
    
    return workerChan
}
 
func (d *FrameDistributor) SubmitResult(result AnalysisResult) {
    d.results <- result
}
 
func (d *FrameDistributor) Aggregate(videoID string) *VideoReport {
    frameResults := make(map[int]AnalysisResult)
    timeout := time.After(30 * time.Second)
    
    for {
        select {
        case result := <-d.results:
            if result.VideoID == videoID {
                frameResults[result.FrameNumber] = result
                
                // Все кадры обработаны?
                if len(frameResults) == expectedFrameCount(videoID) {
                    return buildReport(frameResults)
                }
            }
        case <-timeout:
            return buildPartialReport(frameResults)
        }
    }
}

type WorkerPool struct {
    tasks      chan Task
    results    chan Result
    maxWorkers int
    wg         sync.WaitGroup
}
 
func NewWorkerPool(maxWorkers, queueSize int) *WorkerPool {
    return &WorkerPool{
        tasks:      make(chan Task, queueSize),
        results:    make(chan Result, queueSize),
        maxWorkers: maxWorkers,
    }
}
 
func (p *WorkerPool) Start(processor func(Task) Result) {
    for i := 0; i < p.maxWorkers; i++ {
        p.wg.Add(1)
        go func(workerID int) {
            defer p.wg.Done()
            for task := range p.tasks {
                // Собственно обработка - вызов модели, трансформация данных
                result := processor(task)
                p.results <- result
            }
        }(i)
    }
}
 
func (p *WorkerPool) Submit(task Task) {
    p.tasks <- task
}
 
func (p *WorkerPool) Shutdown() {
    close(p.tasks)
    p.wg.Wait()
    close(p.results)
}

import "github.com/jdkato/prose/v2"
 
func extractEntities(text string) ([]string, error) {
    doc, err := prose.NewDocument(text)
    if err != nil {
        return nil, err
    }
    
    var entities []string
    for _, ent := range doc.Entities() {
        // Извлекаем имена людей и организации
        if ent.Label == "PERSON" || ent.Label == "ORG" {
            entities = append(entities, ent.Text)
        }
    }
    
    return entities, nil
}

type SentimentAnalyzer struct {
    session   *ort.Session
    tokenizer *Tokenizer
    mu        sync.RWMutex
}
 
func (sa *SentimentAnalyzer) Analyze(text string) (float64, error) {
    // Токенизация - самая тяжёлая часть препроцессинга
    tokens := sa.tokenizer.Encode(text, 128) // Максимум 128 токенов
    
    // Создаём тензор входных данных
    inputIDs := make([]int64, len(tokens))
    attentionMask := make([]int64, len(tokens))
    for i, token := range tokens {
        inputIDs[i] = int64(token)
        attentionMask[i] = 1
    }
    
    // Инференс модели
    sa.mu.RLock()
    defer sa.mu.RUnlock()
    
    outputs, err := sa.session.Run([]ort.Value{
        createTensor(inputIDs, []int64{1, int64(len(tokens))}),
        createTensor(attentionMask, []int64{1, int64(len(tokens))}),
    })
    if err != nil {
        return 0, err
    }
    defer destroyOutputs(outputs)
    
    // Извлекаем logits и применяем softmax
    logits := outputs[0].GetTensor().GetData().([]float32)
    score := softmax(logits)[1] // Индекс 1 - положительный класс
    
    return float64(score), nil
}

import (
    "image"
    _ "image/jpeg"
    _ "image/png"
    "golang.org/x/image/draw"
)
 
func preprocessImage(img image.Image, targetWidth, targetHeight int) []float32 {
    // Ресайз с сохранением пропорций
    bounds := img.Bounds()
    resized := image.NewRGBA(image.Rect(0, 0, targetWidth, targetHeight))
    draw.CatmullRom.Scale(resized, resized.Bounds(), img, bounds, draw.Over, nil)
    
    // Нормализация и конвертация в тензор [1, 3, H, W]
    tensor := make([]float32, 3*targetHeight*targetWidth)
    idx := 0
    
    for y := 0; y < targetHeight; y++ {
        for x := 0; x < targetWidth; x++ {
            r, g, b, _ := resized.At(x, y).RGBA()
            
            // ImageNet нормализация
            tensor[idx] = (float32(r>>8) - 123.675) / 58.395
            tensor[idx+targetHeight*targetWidth] = (float32(g>>8) - 116.28) / 57.12
            tensor[idx+2*targetHeight*targetWidth] = (float32(b>>8) - 103.53) / 57.375
            idx++
        }
    }
    
    return tensor
}

import "github.com/davidbyttow/govips/v2/vips"
 
func fastPreprocess(inputPath string, width, height int) ([]float32, error) {
    // Инициализация vips (один раз при старте)
    vips.Startup(nil)
    defer vips.Shutdown()
    
    // Загрузка и ресайз в одну операцию
    img, err := vips.NewImageFromFile(inputPath)
    if err != nil {
        return nil, err
    }
    defer img.Close()
    
    // Ресайз с сохранением aspect ratio
    if err := img.Thumbnail(width, height, vips.InterestingCentre); err != nil {
        return nil, err
    }
    
    // Нормализация пикселей
    if err := img.Linear([]float64{1.0/255}, []float64{0}); err != nil {
        return nil, err
    }
    
    // Конвертация в float32 массив
    bytes := img.Export(nil)
    tensor := bytesToFloat32(bytes, img.Bands())
    
    return tensor, nil
}

type ChatBot struct {
    llmClient    *OpenAIClient
    vectorDB     *PineconeClient
    conversationStore *ConversationStore
    rateLimiter  *rate.Limiter
}
 
func (bot *ChatBot) HandleMessage(ctx context.Context, userID, message string) (string, error) {
    // Rate limiting на пользователя
    if !bot.rateLimiter.Allow() {
        return "Слишком много запросов, попробуйте через минуту", nil
    }
    
    // Получаем историю диалога
    history := bot.conversationStore.GetHistory(userID, 5)
    
    // Ищем релевантные документы через embedding поиск
    docs, err := bot.vectorDB.Search(ctx, message, 3)
    if err != nil {
        return "", fmt.Errorf("vector search failed: %w", err)
    }
    
    // Формируем промпт с контекстом
    prompt := buildPrompt(message, history, docs)
    
    // Вызываем LLM с retry-логикой и таймаутом
    response, err := bot.llmClient.Complete(ctx, CompletionRequest{
        Model: "gpt-4-turbo",
        Messages: []Message{
            {Role: "system", Content: systemPrompt},
            {Role: "user", Content: prompt},
        },
        Temperature: 0.7,
        MaxTokens: 500,
    })
    
    if err != nil {
        // Fallback на более дешёвую модель при ошибке
        return bot.tryFallbackModel(ctx, prompt)
    }
    
    // Сохраняем в историю для следующих запросов
    bot.conversationStore.Append(userID, message, response.Content)
    
    return response.Content, nil
}

type RecommendationEngine struct {
    userEmbeddings  map[int64][]float32
    itemEmbeddings  map[int64][]float32
    userHistory     *HistoryCache
    mu              sync.RWMutex
}
 
func (re *RecommendationEngine) GetRecommendations(userID int64, count int) ([]int64, error) {
    re.mu.RLock()
    userVec, exists := re.userEmbeddings[userID]
    re.mu.RUnlock()
    
    if !exists {
        return re.getColdStartRecommendations(count), nil
    }
    
    // Параллельный подсчёт скоров для всех айтемов
    scores := make(chan scoredItem, len(re.itemEmbeddings))
    semaphore := make(chan struct{}, runtime.NumCPU())
    
    var wg sync.WaitGroup
    for itemID, itemVec := range re.itemEmbeddings {
        wg.Add(1)
        semaphore <- struct{}{}
        
        go func(id int64, vec []float32) {
            defer wg.Done()
            defer func() { <-semaphore }()
            
            score := dotProduct(userVec, vec)
            scores <- scoredItem{id: id, score: score}
        }(itemID, itemVec)
    }
    
    go func() {
        wg.Wait()
        close(scores)
    }()
    
    // Собираем топ-N через min-heap
    topItems := selectTopN(scores, count*3) // Берём с запасом для фильтрации
    
    // Фильтруем уже просмотренное
    history := re.userHistory.Get(userID)
    filtered := filterSeen(topItems, history)
    
    return extractIDs(filtered[:min(count, len(filtered))]), nil
}

type TimeSeriesPredictor struct {
    model      *ort.Session
    window     int
    scaler     *MinMaxScaler
    buffer     map[string]*RollingWindow
    mu         sync.RWMutex
}
 
func (tsp *TimeSeriesPredictor) Predict(serverID string, value float64) (float64, error) {
    tsp.mu.Lock()
    window, exists := tsp.buffer[serverID]
    if !exists {
        window = NewRollingWindow(tsp.window)
        tsp.buffer[serverID] = window
    }
    window.Add(value)
    tsp.mu.Unlock()
    
    if window.Size() < tsp.window {
        return value, nil // Недостаточно данных для прогноза
    }
    
    // Нормализация окна
    normalized := tsp.scaler.Transform(window.Values())
    
    // Формируем тензор [1, window, 1]
    input := reshapeForLSTM(normalized)
    
    // Инференс
    outputs, err := tsp.model.Run([]ort.Value{input})
    if err != nil {
        return 0, err
    }
    defer outputs[0].Destroy()
    
    prediction := outputs[0].GetTensor().GetData().([]float32)[0]
    return float64(tsp.scaler.InverseTransform(prediction)), nil
}

type Pipeline struct {
    stages []Stage
}
 
type Stage interface {
    Process(context.Context, *Data) error
}
 
func (p *Pipeline) Execute(ctx context.Context, data *Data) error {
    for _, stage := range p.stages {
        if err := stage.Process(ctx, data); err != nil {
            return fmt.Errorf("stage failed: %w", err)
        }
    }
    return nil
}

type DocumentPipeline struct {
    ocrClient         *OCRServiceClient
    classifierClient  *ClassifierClient
    extractorClient   *EntityExtractorClient
    validatorClient   *ValidatorClient
    resultStore       *ResultStore
}
 
func (dp *DocumentPipeline) Process(ctx context.Context, doc *Document) (*ProcessedDocument, error) {
    // Параллельные вызовы независимых сервисов
    ocrCh := make(chan *OCRResult, 1)
    metaCh := make(chan *MetaData, 1)
    
    go func() {
        text, err := dp.ocrClient.ExtractText(ctx, doc.ImageData)
        ocrCh <- &OCRResult{Text: text, Error: err}
    }()
    
    go func() {
        meta, err := extractMetadata(doc)
        metaCh <- &MetaData{Data: meta, Error: err}
    }()
    
    // Собираем результаты
    ocrRes := <-ocrCh
    metaRes := <-metaCh
    
    if ocrRes.Error != nil {
        return nil, fmt.Errorf("ocr failed: %w", ocrRes.Error)
    }
    
    // Последовательные зависимые вызовы
    docType, err := dp.classifierClient.Classify(ctx, ocrRes.Text, metaRes.Data)
    if err != nil {
        return nil, fmt.Errorf("classification failed: %w", err)
    }
    
    entities, err := dp.extractorClient.Extract(ctx, ocrRes.Text, docType)
    if err != nil {
        return nil, fmt.Errorf("extraction failed: %w", err)
    }
    
    validated, err := dp.validatorClient.Validate(ctx, entities, docType)
    if err != nil {
        return nil, fmt.Errorf("validation failed: %w", err)
    }
    
    return &ProcessedDocument{
        Type:     docType,
        Entities: validated,
        RawText:  ocrRes.Text,
    }, nil
}

type Processor interface {
    Process(ctx context.Context, data interface{}) (interface{}, error)
    Name() string
}
 
type MLPipeline struct {
    stages []Processor
    logger *Logger
}
 
func (p *MLPipeline) Run(ctx context.Context, input interface{}) (interface{}, error) {
    current := input
    
    for _, stage := range p.stages {
        start := time.Now()
        
        result, err := stage.Process(ctx, current)
        if err != nil {
            return nil, fmt.Errorf("%s failed: %w", stage.Name(), err)
        }
        
        // Логируем метрики каждого этапа
        p.logger.LogMetrics(stage.Name(), time.Since(start), getSizeEstimate(result))
        current = result
    }
    
    return current, nil
}
 
// Конкретная реализация этапа
type TextNormalizer struct {
    lowercae bool
    removeStopwords bool
}
 
func (tn *TextNormalizer) Process(ctx context.Context, data interface{}) (interface{}, error) {
    text, ok := data.(string)
    if !ok {
        return nil, fmt.Errorf("expected string, got %T", data)
    }
    
    if tn.lowercae {
        text = strings.ToLower(text)
    }
    
    if tn.removeStopwords {
        text = removeStopWords(text)
    }
    
    return text, nil
}
 
func (tn *TextNormalizer) Name() string {
    return "text_normalizer"
}

type AsyncProcessor struct {
    queue   chan Task
    workers int
    wg      sync.WaitGroup
}
 
func NewAsyncProcessor(queueSize, workers int) *AsyncProcessor {
    ap := &AsyncProcessor{
        queue:   make(chan Task, queueSize),
        workers: workers,
    }
    ap.start()
    return ap
}
 
func (ap *AsyncProcessor) start() {
    for i := 0; i < ap.workers; i++ {
        ap.wg.Add(1)
        go func(workerID int) {
            defer ap.wg.Done()
            for task := range ap.queue {
                ap.processTask(task)
            }
        }(i)
    }
}
 
func (ap *AsyncProcessor) Submit(task Task) error {
    select {
    case ap.queue <- task:
        return nil
    default:
        return errors.New("queue full")
    }
}

type InferenceMetrics struct {
    requestCount    prometheus.Counter
    latencyHistogram prometheus.Histogram
    errorCount      prometheus.Counter
    modelVersion    prometheus.Gauge
}
 
func (m *InferenceMetrics) RecordRequest(duration time.Duration, err error) {
    m.requestCount.Inc()
    m.latencyHistogram.Observe(duration.Seconds())
    if err != nil {
        m.errorCount.Inc()
    }
}
 
func InferenceMiddleware(model *Model, metrics *InferenceMetrics) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        start := time.Now()
        
        // Извлекаем trace ID из контекста для корреляции
        traceID := r.Header.Get("X-Trace-ID")
        if traceID == "" {
            traceID = generateTraceID()
        }
        
        ctx := context.WithValue(r.Context(), "trace_id", traceID)
        
        result, err := model.Predict(ctx, extractInput(r))
        duration := time.Since(start)
        
        metrics.RecordRequest(duration, err)
        
        // Структурированный лог для последующего анализа
        log.Info().
            Str("trace_id", traceID).
            Dur("duration", duration).
            Int("input_size", len(extractInput(r))).
            Bool("error", err != nil).
            Msg("inference completed")
        
        if err != nil {
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
        }
        
        json.NewEncoder(w).Encode(result)
    }
}

text-classifier/
├── cmd/
│   └── server/
│       └── main.go
├── internal/
│   ├── classifier/
│   │   ├── model.go
│   │   └── tokenizer.go
│   ├── handler/
│   │   └── http.go
│   └── metrics/
│       └── prometheus.go
├── models/
│   └── classifier.onnx
└── go.mod

// internal/classifier/model.go
package classifier
 
import (
    "context"
    "fmt"
    "sync"
 
    ort "github.com/yalue/onnxruntime_go"
)
 
type Model struct {
    session    *ort.Session
    inputName  string
    outputName string
    mu         sync.RWMutex
    categories []string
}
 
func NewModel(modelPath string) (*Model, error) {
    if err := ort.InitializeEnvironment(); err != nil {
        return nil, fmt.Errorf("init onnx runtime: %w", err)
    }
 
    opts := ort.NewSessionOptions()
    defer opts.Destroy()
 
    session, err := ort.NewSession(modelPath, opts)
    if err != nil {
        return nil, fmt.Errorf("load model: %w", err)
    }
 
    // Получаем метаданные модели
    inputCount, err := session.GetInputCount()
    if err != nil {
        return nil, fmt.Errorf("get input count: %w", err)
    }
 
    if inputCount != 1 {
        return nil, fmt.Errorf("expected 1 input, got %d", inputCount)
    }
 
    inputName, err := session.GetInputName(0)
    if err != nil {
        return nil, fmt.Errorf("get input name: %w", err)
    }
 
    outputName, err := session.GetOutputName(0)
    if err != nil {
        return nil, fmt.Errorf("get output name: %w", err)
    }
 
    return &Model{
        session:    session,
        inputName:  inputName,
        outputName: outputName,
        categories: []string{
            "технологии",
            "спорт",
            "политика",
            "бизнес",
            "развлечения",
            "наука",
        },
    }, nil
}
 
func (m *Model) Predict(ctx context.Context, tokens []int64) (*Result, error) {
    m.mu.RLock()
    defer m.mu.RUnlock()
 
    // Проверяем контекст перед тяжёлой операцией
    select {
    case <-ctx.Done():
        return nil, ctx.Err()
    default:
    }
 
    // Создаём входной тензор [1, seq_len]
    shape := []int64{1, int64(len(tokens))}
    inputTensor, err := ort.NewTensor(shape, tokens)
    if err != nil {
        return nil, fmt.Errorf("create tensor: %w", err)
    }
    defer inputTensor.Destroy()
 
    // Запускаем инференс
    outputs, err := m.session.Run(
        []ort.Value{inputTensor},
        []string{m.inputName},
        []string{m.outputName},
    )
    if err != nil {
        return nil, fmt.Errorf("run inference: %w", err)
    }
    defer outputs[0].Destroy()
 
    // Извлекаем logits и применяем softmax
    logits := outputs[0].GetTensor().GetData().([]float32)
    probs := softmax(logits)
 
    // Находим класс с максимальной вероятностью
    maxIdx, maxProb := 0, probs[0]
    for i, p := range probs[1:] {
        if p > maxProb {
            maxIdx = i + 1
            maxProb = p
        }
    }
 
    return &Result{
        Category:   m.categories[maxIdx],
        Confidence: float64(maxProb),
        Probabilities: convertToMap(m.categories, probs),
    }, nil
}
 
func (m *Model) Close() error {
    if m.session != nil {
        m.session.Destroy()
    }
    return nil
}
 
type Result struct {
    Category      string             [INLINE]json:"category"[/INLINE]
    Confidence    float64            [INLINE]json:"confidence"[/INLINE]
    Probabilities map[string]float64 `json:"probabilities"`
}
 
func softmax(logits []float32) []float32 {
    // Находим максимум для численной стабильности
    maxLogit := logits[0]
    for _, l := range logits[1:] {
        if l > maxLogit {
            maxLogit = l
        }
    }
 
    // Вычисляем exp(x - max) и сумму
    exp := make([]float32, len(logits))
    sum := float32(0)
    for i, l := range logits {
        exp[i] = float32(expApprox(float64(l - maxLogit)))
        sum += exp[i]
    }
 
    // Нормализуем
    probs := make([]float32, len(logits))
    for i := range exp {
        probs[i] = exp[i] / sum
    }
 
    return probs
}
 
func convertToMap(categories []string, probs []float32) map[string]float64 {
    result := make(map[string]float64, len(categories))
    for i, cat := range categories {
        result[cat] = float64(probs[i])
    }
    return result
}
 
// Быстрое приближение экспоненты для небольших значений
func expApprox(x float64) float64 {
    if x < -10 {
        return 0
    }
    if x > 10 {
        return 22026.46579 // e^10
    }
    // Taylor series для небольших x
    result := 1.0 + x
    term := x
    for i := 2; i < 10; i++ {
        term *= x / float64(i)
        result += term
        if term < 1e-10 {
            break
        }
    }
    return result
}

// internal/classifier/tokenizer.go
package classifier
 
import (
    "strings"
    "unicode"
)
 
type Tokenizer struct {
    vocab   map[string]int64
    maxLen  int
    padID   int64
    unkID   int64
}
 
func NewTokenizer(vocabPath string, maxLen int) (*Tokenizer, error) {
    vocab := make(map[string]int64)
    
    // В реальности загружали бы из файла
    // Для демо захардкодим базовый vocab
    vocab["<PAD>"] = 0
    vocab["<UNK>"] = 1
    
    // Добавляем частые русские слова
    commonWords := []string{
        "в", "и", "на", "с", "по", "для", "от", "к", "о",
        "новый", "год", "день", "время", "работа", "человек",
        "компания", "сообщил", "сказал", "стал", "может",
    }
    
    for i, word := range commonWords {
        vocab[word] = int64(i + 2)
    }
 
    return &Tokenizer{
        vocab:  vocab,
        maxLen: maxLen,
        padID:  0,
        unkID:  1,
    }, nil
}
 
func (t *Tokenizer) Encode(text string) []int64 {
    // Нормализация: lowercase + удаление пунктуации
    text = strings.ToLower(text)
    text = removePunctuation(text)
    
    words := strings.Fields(text)
    tokens := make([]int64, 0, t.maxLen)
    
    for _, word := range words {
        if len(tokens) >= t.maxLen {
            break
        }
        
        if id, exists := t.vocab[word]; exists {
            tokens = append(tokens, id)
        } else {
            tokens = append(tokens, t.unkID)
        }
    }
    
    // Padding до maxLen
    for len(tokens) < t.maxLen {
        tokens = append(tokens, t.padID)
    }
    
    return tokens
}
 
func removePunctuation(text string) string {
    var builder strings.Builder
    builder.Grow(len(text))
    
    for _, r := range text {
        if unicode.IsLetter(r) || unicode.IsSpace(r) {
            builder.WriteRune(r)
        } else {
            builder.WriteRune(' ')
        }
    }
    
    return builder.String()
}

// internal/handler/http.go
package handler
 
import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"
    "time"
 
    "text-classifier/internal/classifier"
    "text-classifier/internal/metrics"
)
 
type Handler struct {
    model     *classifier.Model
    tokenizer *classifier.Tokenizer
    metrics   *metrics.Collector
    maxTextLen int
}
 
func NewHandler(model *classifier.Model, tokenizer *classifier.Tokenizer, m *metrics.Collector) *Handler {
    return &Handler{
        model:      model,
        tokenizer:  tokenizer,
        metrics:    m,
        maxTextLen: 5000,
    }
}
 
type ClassifyRequest struct {
    Text string `json:"text"`
}
 
type ClassifyResponse struct {
    Category      string             [INLINE]json:"category"[/INLINE]
    Confidence    float64            [INLINE]json:"confidence"[/INLINE]
    Probabilities map[string]float64 `json:"probabilities,omitempty"`
    ProcessedIn   string             [INLINE]json:"processed_in"[/INLINE]
}
 
type ErrorResponse struct {
    Error   string `json:"error"`
    Message string `json:"message"`
}
 
func (h *Handler) Classify(w http.ResponseWriter, r *http.Request) {
    start := time.Now()
    
    // Только POST запросы
    if r.Method != http.MethodPost {
        h.writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only POST requests allowed")
        return
    }
    
    // Парсим входные данные
    var req ClassifyRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        h.metrics.RecordRequest(time.Since(start), false)
        h.writeError(w, http.StatusBadRequest, "invalid_json", "Failed to parse JSON body")
        return
    }
    
    // Валидация
    if req.Text == "" {
        h.metrics.RecordRequest(time.Since(start), false)
        h.writeError(w, http.StatusBadRequest, "empty_text", "Text field cannot be empty")
        return
    }
    
    if len(req.Text) > h.maxTextLen {
        h.metrics.RecordRequest(time.Since(start), false)
        h.writeError(w, http.StatusBadRequest, "text_too_long", 
            fmt.Sprintf("Text exceeds maximum length of %d characters", h.maxTextLen))
        return
    }
    
    // Токенизация
    tokens := h.tokenizer.Encode(req.Text)
    
    // Инференс с таймаутом
    ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
    defer cancel()
    
    result, err := h.model.Predict(ctx, tokens)
    if err != nil {
        h.metrics.RecordRequest(time.Since(start), false)
        h.writeError(w, http.StatusInternalServerError, "prediction_failed", 
            "Failed to classify text")
        return
    }
    
    duration := time.Since(start)
    h.metrics.RecordRequest(duration, true)
    h.metrics.RecordConfidence(result.Confidence)
    
    // Формируем ответ
    response := ClassifyResponse{
        Category:      result.Category,
        Confidence:    result.Confidence,
        Probabilities: result.Probabilities,
        ProcessedIn:   duration.String(),
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(response)
}
 
func (h *Handler) Health(w http.ResponseWriter, r *http.Request) {
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string]string{
        "status": "healthy",
    })
}
 
func (h *Handler) writeError(w http.ResponseWriter, status int, errorType, message string) {
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(status)
    json.NewEncoder(w).Encode(ErrorResponse{
        Error:   errorType,
        Message: message,
    })
}

// internal/metrics/prometheus.go
package metrics
 
import (
    "time"
 
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)
 
type Collector struct {
    requestsTotal   prometheus.Counter
    requestDuration prometheus.Histogram
    requestErrors   prometheus.Counter
    confidenceGauge prometheus.Histogram
    activeRequests  prometheus.Gauge
}
 
func NewCollector() *Collector {
    return &Collector{
        requestsTotal: promauto.NewCounter(prometheus.CounterOpts{
            Name: "classifier_requests_total",
            Help: "Total number of classification requests",
        }),
        requestDuration: promauto.NewHistogram(prometheus.HistogramOpts{
            Name:    "classifier_request_duration_seconds",
            Help:    "Request processing duration in seconds",
            Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0},
        }),
        requestErrors: promauto.NewCounter(prometheus.CounterOpts{
            Name: "classifier_errors_total",
            Help: "Total number of failed requests",
        }),
        confidenceGauge: promauto.NewHistogram(prometheus.HistogramOpts{
            Name:    "classifier_confidence_score",
            Help:    "Distribution of model confidence scores",
            Buckets: []float64{0.5, 0.6, 0.7, 0.8, 0.85, 0.9, 0.95, 0.99, 1.0},
        }),
        activeRequests: promauto.NewGauge(prometheus.GaugeOpts{
            Name: "classifier_active_requests",
            Help: "Number of requests currently being processed",
        }),
    }
}
 
func (c *Collector) RecordRequest(duration time.Duration, success bool) {
    c.requestsTotal.Inc()
    c.requestDuration.Observe(duration.Seconds())
    if !success {
        c.requestErrors.Inc()
    }
}
 
func (c *Collector) RecordConfidence(score float64) {
    c.confidenceGauge.Observe(score)
}
 
func (c *Collector) IncActiveRequests() {
    c.activeRequests.Inc()
}
 
func (c *Collector) DecActiveRequests() {
    c.activeRequests.Dec()
}

// cmd/server/main.go
package main
 
import (
    "context"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"
 
    "github.com/prometheus/client_golang/prometheus/promhttp"
    
    "text-classifier/internal/classifier"
    "text-classifier/internal/handler"
    "text-classifier/internal/metrics"
)
 
func main() {
    // Инициализация компонентов
    model, err := classifier.NewModel("models/classifier.onnx")
    if err != nil {
        log.Fatalf("Failed to load model: %v", err)
    }
    defer model.Close()
 
    tokenizer, err := classifier.NewTokenizer("", 128)
    if err != nil {
        log.Fatalf("Failed to init tokenizer: %v", err)
    }
 
    metricsCollector := metrics.NewCollector()
    h := handler.NewHandler(model, tokenizer, metricsCollector)
 
    // Настройка роутов
    mux := http.NewServeMux()
    mux.HandleFunc("/classify", h.Classify)
    mux.HandleFunc("/health", h.Health)
    mux.Handle("/metrics", promhttp.Handler())
 
    // HTTP сервер с таймаутами
    srv := &http.Server{
        Addr:         ":8080",
        Handler:      mux,
        ReadTimeout:  10 * time.Second,
        WriteTimeout: 10 * time.Second,
        IdleTimeout:  60 * time.Second,
    }
 
    // Graceful shutdown
    done := make(chan os.Signal, 1)
    signal.Notify(done, os.Interrupt, syscall.SIGTERM)
 
    go func() {
        log.Printf("Server starting on %s", srv.Addr)
        if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Fatalf("Server failed: %v", err)
        }
    }()
 
    <-done
    log.Println("Server stopping...")
 
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
 
    if err := srv.Shutdown(ctx); err != nil {
        log.Fatalf("Server forced shutdown: %v", err)
    }
 
    log.Println("Server stopped gracefully")
}