1Panel/agent/utils/ai_tools/vllm/history.go

package vllm

import (
	"encoding/json"
	"math"

	"github.com/1Panel-dev/1Panel/agent/app/model"
)

type HistoryAccumulator struct {
	histograms    map[string]histogram
	prefixHits    float64
	prefixQueries float64
	finishReasons map[string]float64
	finishSeconds map[string]float64
}

func (a *HistoryAccumulator) Add(histograms, requests, aggregation, group string) error {
	if histograms != "" {
		var sample map[string]histogram
		if aggregation == "max" || group == "requests" {
			var distributions struct {
				Prompt     *histogram `json:"request_prompt_tokens"`
				Generation *histogram `json:"request_generation_tokens"`
			}
			if err := json.Unmarshal([]byte(histograms), &distributions); err != nil {
				return err
			}
			sample = map[string]histogram{}
			if distributions.Prompt != nil {
				sample["request_prompt_tokens"] = *distributions.Prompt
			}
			if distributions.Generation != nil {
				sample["request_generation_tokens"] = *distributions.Generation
			}
		} else if group == "latency" {
			var latency struct {
				FirstToken *histogram `json:"time_to_first_token_seconds"`
				InterToken *histogram `json:"inter_token_latency_seconds"`
				Request    *histogram `json:"e2e_request_latency_seconds"`
				Prefill    *histogram `json:"request_prefill_time_seconds"`
				Decode     *histogram `json:"request_decode_time_seconds"`
				Queue      *histogram `json:"request_queue_time_seconds"`
			}
			if err := json.Unmarshal([]byte(histograms), &latency); err != nil {
				return err
			}
			sample = map[string]histogram{}
			for name, item := range map[string]*histogram{
				"time_to_first_token_seconds":  latency.FirstToken,
				"inter_token_latency_seconds":  latency.InterToken,
				"e2e_request_latency_seconds":  latency.Request,
				"request_prefill_time_seconds": latency.Prefill,
				"request_decode_time_seconds":  latency.Decode,
				"request_queue_time_seconds":   latency.Queue,
			} {
				if item != nil {
					sample[name] = *item
				}
			}
		} else if err := json.Unmarshal([]byte(histograms), &sample); err != nil {
			return err
		}
		if a.histograms == nil {
			a.histograms = map[string]histogram{}
		}
		for name, item := range sample {
			switch name {
			case "request_prefill_time_seconds", "request_decode_time_seconds", "request_queue_time_seconds":
				item.Buckets = nil
			}
			total, exists := a.histograms[name]
			if !exists {
				a.histograms[name] = item
				continue
			}
			total.Seconds += item.Seconds
			total.Sum += item.Sum
			total.Count += item.Count
			if len(total.Buckets) != len(item.Buckets) {
				total.Buckets = nil
			}
			for bound, count := range total.Buckets {
				addition, ok := item.Buckets[bound]
				if !ok {
					total.Buckets = nil
					break
				}
				total.Buckets[bound] = count + addition
			}
			a.histograms[name] = total
		}
	}
	if requests == "" {
		return nil
	}
	if group == "overview" {
		var prefix struct {
			Seconds float64  `json:"seconds"`
			Hits    *float64 `json:"prefixHits"`
			Queries *float64 `json:"prefixQueries"`
		}
		if err := json.Unmarshal([]byte(requests), &prefix); err != nil {
			return err
		}
		if prefix.Seconds > 0 && prefix.Hits != nil && prefix.Queries != nil {
			a.prefixHits += *prefix.Hits
			a.prefixQueries += *prefix.Queries
		}
		return nil
	}
	var sample requestMetrics
	if err := json.Unmarshal([]byte(requests), &sample); err != nil {
		return err
	}
	if sample.Seconds <= 0 {
		return nil
	}
	if group == "" && aggregation != "max" && sample.PrefixHits != nil && sample.PrefixQueries != nil {
		a.prefixHits += *sample.PrefixHits
		a.prefixQueries += *sample.PrefixQueries
	}
	for reason, count := range sample.FinishReasons {
		if a.finishReasons == nil {
			a.finishReasons = map[string]float64{}
		}
		if aggregation == "max" {
			a.finishReasons[reason] = math.Max(a.finishReasons[reason], count/sample.Seconds)
		} else {
			if a.finishSeconds == nil {
				a.finishSeconds = map[string]float64{}
			}
			a.finishReasons[reason] += count
			a.finishSeconds[reason] += sample.Seconds
		}
	}
	return nil
}

func (a *HistoryAccumulator) Apply(result *model.MonitorVLLM, aggregation string) {
	applyHistograms(result, a.histograms, aggregation != "max")
	for reason, count := range a.finishReasons {
		if result.FinishReasons == nil {
			result.FinishReasons = map[string]float64{}
		}
		if aggregation == "max" {
			result.FinishReasons[reason] = count
		} else {
			result.FinishReasons[reason] = count / a.finishSeconds[reason]
		}
	}
	if aggregation != "max" && a.prefixQueries > 0 {
		rate := a.prefixHits / a.prefixQueries * 100
		result.PrefixCacheHitRate = &rate
	}
}