latencypredictorclient

package
v0.9.2 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jun 23, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

README

latencypredictorclient

Go client for the llm-d-latency-predictor Python service, which predicts per-request latency (TTFT / TPOT) for LLM inference and is trained online from observed request metrics.

This client is used by the predictedlatency data producer plugin to fetch predictions during scheduling and to stream training samples back to the Python server.

What it does

  • Prediction. Issues bulk HTTP prediction calls to one or more prediction server URLs (PredictionURLs), with a coalescing layer that merges concurrent callers of PredictBulkStrict into a single batched HTTP request (CoalesceWindow, MaxBulkSize).
  • Training. Buffers TrainingEntry samples locally and periodically flushes them to the training server (TrainingURL) on FlushInterval, downsampling to MaxSampleSize when the buffer grows large.
  • Model snapshot caching. Periodically pulls the trained model itself from the Python server — coefficients (Bayesian Ridge), serialized trees (XGBoost), and bucket counts — on MetricsRefreshInterval, and caches it locally. (The server exposes this via a /metrics-style endpoint; the artifacts are not sent along with prediction requests.)
  • Local vs. remote inference. With the cached snapshot, Bayesian Ridge predictions run in-process from the cached coefficients, no HTTP hop per request. XGBoost and LightGBM predictions go over HTTP via bulk prediction calls.

Configuration can be supplied programmatically via Config or loaded from environment variables via ConfigFromEnv(). See types.go for the full set of knobs.

Entry points

  • New(cfg, logger) *Predictor — construct a client. Spawns two background goroutines (flush loop + coalescing dispatcher).
  • Start(ctx) error — prime the predictor with initial server status and model info.
  • PredictBulkStrict, AddTrainingEntry, and related methods — see prediction.go, training.go.

Tests

tests/ contains a standalone load-test harness (main.go) that drives the client against a running predictor server. It is a package main binary, not part of the scheduler build.

Documentation

Index

Constants

View Source
const (

	// ObjectiveType constants for prediction objective.
	ObjectiveQuantile = "quantile"
	ObjectiveMean     = "mean"
)

Variables

This section is empty.

Functions

This section is empty.

Types

type BucketCounts

type BucketCounts struct {
	TTFTBuckets map[int]int `json:"ttft_buckets"`
	TPOTBuckets map[int]int `json:"tpot_buckets"`
}

type BulkPredictionError

type BulkPredictionError struct {
	Index   int               `json:"index"`
	Error   string            `json:"error"`
	Request PredictionRequest `json:"request"`
}

type BulkPredictionRequest

type BulkPredictionRequest struct {
	Requests []PredictionRequest `json:"requests"`
}

New data models for bulk predictions

type BulkPredictionResponse

type BulkPredictionResponse struct {
	Predictions           []PredictionResponse `json:"predictions"`
	TotalRequests         int                  `json:"total_requests"`
	SuccessfulPredictions int                  `json:"successful_predictions"`
	FailedPredictions     int                  `json:"failed_predictions"`
	ProcessingTimeMs      float64              `json:"processing_time_ms"`
}

type BulkPredictionResponseWithErrors

type BulkPredictionResponseWithErrors struct {
	Predictions           []*PredictionResponse `json:"predictions"`
	Errors                []BulkPredictionError `json:"errors"`
	TotalRequests         int                   `json:"total_requests"`
	SuccessfulPredictions int                   `json:"successful_predictions"`
	FailedPredictions     int                   `json:"failed_predictions"`
	ProcessingTimeMs      float64               `json:"processing_time_ms"`
}

type BulkTrainingRequest

type BulkTrainingRequest struct {
	Entries []TrainingEntry `json:"entries"`
}

type Config

type Config struct {
	// TrainingURL is the base URL of the Python training server.
	TrainingURL string
	// PredictionURLs is a list of prediction server URLs for load balancing.
	PredictionURLs []string
	// MaxSampleSize is the maximum number of training entries to send in each flush.
	// If the buffer contains more entries, they will be randomly sampled.
	MaxSampleSize int
	// FlushInterval determines how often to flush training & refresh metrics.
	FlushInterval time.Duration
	// UseNativeXGBoost when true, attempts to use local XGBoost models for prediction.
	// When false, falls back to HTTP calls to the Python server for XGBoost predictions.
	UseNativeXGBoost bool
	// HTTPTimeout is the timeout for HTTP requests to the Python server.
	HTTPTimeout time.Duration
	// MetricsRefreshInterval determines how often to refresh cached metrics.
	MetricsRefreshInterval time.Duration
	// MaxBulkSize is the maximum number of predictions to send in a single bulk request.
	MaxBulkSize int
	// CoalesceWindow is how long the coalescer waits to accumulate concurrent
	// PredictBulkStrict callers before firing one mega-batch HTTP call.
	// Set to 0 to disable coalescing (each caller gets its own HTTP call).
	CoalesceWindow time.Duration
	// MaxCoalescedRows caps the total number of rows in one coalesced mega-batch,
	// causing an early dispatch before the window expires.
	// This is separate from MaxBulkSize (the per-caller row limit).
	// Default 0 means no row cap (window-only dispatch).
	MaxCoalescedRows int
	// MaxConcurrentDispatches limits how many coalesced HTTP calls can be
	// in-flight to the prediction server simultaneously. Defaults to 36.
	MaxConcurrentDispatches int
}

func ConfigFromEnv

func ConfigFromEnv() *Config

func DefaultConfig

func DefaultConfig() *Config

type MetricsResponse

type MetricsResponse struct {
	ModelType    string             `json:"model_type"`
	Coefficients *ModelCoefficients `json:"coefficients"`
	XGBoostTrees *XGBoostTrees      `json:"xgboost_trees"`
	BucketCounts *BucketCounts      `json:"bucket_counts"`
	RawMetrics   string             `json:"raw_metrics"`
}

type ModelCoefficients

type ModelCoefficients struct {
	TTFTIntercept float64            `json:"ttft_intercept"`
	TTFTCoeffs    map[string]float64 `json:"ttft_coefficients"`
	TPOTIntercept float64            `json:"tpot_intercept"`
	TPOTCoeffs    map[string]float64 `json:"tpot_coefficients"`
}

type ModelInfo

type ModelInfo struct {
	ModelType     string          `json:"model_type"`
	ObjectiveType string          `json:"objective_type,omitempty"`
	ModelStatus   map[string]bool `json:"model_status"`
	Quantile      float64         `json:"quantile"`
}

type PredictionRequest

type PredictionRequest struct {
	KVCachePercentage     float64 `json:"kv_cache_percentage"`
	InputTokenLength      int     `json:"input_token_length"`
	NumRequestWaiting     int     `json:"num_request_waiting"`
	NumRequestRunning     int     `json:"num_request_running"`
	NumTokensGenerated    int     `json:"num_tokens_generated"`
	PrefixCacheScore      float64 `json:"prefix_cache_score"`
	PodType               string  `json:"pod_type,omitempty"` // "prefill", "decode", or "" for monolithic
	PrefillTokensInFlight int64   `json:"prefill_tokens_in_flight"`
	DecodeTokensInFlight  int64   `json:"decode_tokens_in_flight"`
}

func NewPredictionRequest

func NewPredictionRequest(
	kvCachePercentage float64,
	inputTokenLength int,
	numRequestWaiting int,
	numRequestRunning int,
	numTokensGenerated int,
	prefixCacheScore float64,
) (PredictionRequest, error)

NewPredictionRequest is a helper function to create a new PredictionRequest with proper validation.

type PredictionResponse

type PredictionResponse struct {
	TTFT                 float64    `json:"ttft_ms"`
	TPOT                 float64    `json:"tpot_ms"`
	TTFTUncertainty      float64    `json:"ttft_uncertainty,omitempty"`
	TPOTUncertainty      float64    `json:"tpot_uncertainty,omitempty"`
	TTFTPredictionBounds [2]float64 `json:"ttft_prediction_bounds,omitempty"`
	TPOTPredictionBounds [2]float64 `json:"tpot_prediction_bounds,omitempty"`
	PredictedAt          time.Time  `json:"predicted_at"`
	ModelType            string     `json:"model_type"`
	ObjectiveType        string     `json:"objective_type,omitempty"`
	Quantile             float64    `json:"quantile"`
	LastModelLoad        *time.Time `json:"last_model_load"`
}

type Predictor

type Predictor struct {
	// contains filtered or unexported fields
}

func New

func New(config *Config, logger logr.Logger) *Predictor

func (*Predictor) AddTrainingDataBulk

func (p *Predictor) AddTrainingDataBulk(entries []TrainingEntry) error

AddTrainingDataBulk buffers entries for periodic flush.

func (*Predictor) GetBucketCounts

func (p *Predictor) GetBucketCounts(ctx context.Context) (*BucketCounts, error)

GetBucketCounts fetches the latest metrics and returns the parsed bucket counts.

func (*Predictor) GetCachedMetrics

func (p *Predictor) GetCachedMetrics() (*MetricsResponse, bool)

GetCachedMetrics returns the last metrics fetched. The bool indicates if a value is cached.

func (*Predictor) GetCurrentModelType

func (p *Predictor) GetCurrentModelType() string

GetCurrentModelType returns the current model type from cached server status or model info.

func (*Predictor) GetCurrentObjectiveType

func (p *Predictor) GetCurrentObjectiveType() string

GetCurrentObjectiveType returns the current objective type from server status or defaults to "quantile". Returns ObjectiveQuantile ("quantile") for backward compatibility with servers that don't report objective_type.

func (*Predictor) GetCurrentQuantile

func (p *Predictor) GetCurrentQuantile() float64

GetCurrentQuantile returns the current quantile from server status or defaults to 0.9

func (*Predictor) GetMetrics

func (p *Predictor) GetMetrics(ctx context.Context) (*MetricsResponse, error)

GetMetrics fetches & parses metrics from the training server (for Bayesian Ridge).

func (*Predictor) GetModelCoefficients

func (p *Predictor) GetModelCoefficients(ctx context.Context) (*ModelCoefficients, error)

GetModelCoefficients fetches the latest metrics and returns the parsed coefficients.

func (*Predictor) GetModelInfo

func (p *Predictor) GetModelInfo(ctx context.Context) (*ModelInfo, error)

GetModelInfo fetches the latest model info from the training server.

func (*Predictor) GetPredictionURLs

func (p *Predictor) GetPredictionURLs() []string

GetPredictionURLs returns the list of configured prediction URLs for debugging/monitoring.

func (*Predictor) GetTrainingURL

func (p *Predictor) GetTrainingURL() string

GetTrainingURL returns the configured training URL for debugging/monitoring.

func (*Predictor) GetXGBoostTrees

func (p *Predictor) GetXGBoostTrees(ctx context.Context) (*XGBoostTrees, error)

GetXGBoostTrees returns the cached XGBoost tree data. It does not fetch new data.

func (*Predictor) IsBayesianRidgeReady

func (p *Predictor) IsBayesianRidgeReady() bool

IsBayesianRidgeReady returns true if Bayesian Ridge coefficients are cached.

func (*Predictor) IsLightGBMReady

func (p *Predictor) IsLightGBMReady() bool

IsLightGBMReady returns true if LightGBM models are available via HTTP.

func (*Predictor) IsReady

func (p *Predictor) IsReady() bool

IsReady returns true if a prediction method is ready based on the current model type.

func (*Predictor) IsXGBoostReady

func (p *Predictor) IsXGBoostReady() bool

IsXGBoostReady returns true if native XGBoost models are loaded and ready.

func (*Predictor) Predict

Predict uses cached coefficients (Bayesian Ridge) or HTTP calls (XGBoost/LightGBM) for prediction.

func (*Predictor) PredictBulk

func (p *Predictor) PredictBulk(ctx context.Context, requests []PredictionRequest) (*BulkPredictionResponse, error)

PredictBulk makes bulk predictions with error handling (allows partial failures)

func (*Predictor) PredictBulkStrict

func (p *Predictor) PredictBulkStrict(ctx context.Context, requests []PredictionRequest) (*BulkPredictionResponse, error)

PredictBulkStrict makes bulk predictions that fail if any single prediction fails. When CoalesceWindow > 0, concurrent callers are coalesced into a single HTTP call.

func (*Predictor) Start

func (p *Predictor) Start(ctx context.Context) error

Start initializes the predictor by fetching server status and model info.

func (*Predictor) Stop

func (p *Predictor) Stop(ctx context.Context)

Stop stops background work, then does a final flush/refresh.

func (*Predictor) ValidatePredictionRequest

func (p *Predictor) ValidatePredictionRequest(req PredictionRequest) error

ValidatePredictionRequest validates that a prediction request has all required fields with valid values, including the new prefix_cache_score field.

func (*Predictor) ValidateTrainingEntry

func (p *Predictor) ValidateTrainingEntry(entry TrainingEntry) error

ValidateTrainingEntry validates that a training entry has all required fields with valid values, including the new prefix_cache_score field.

type PredictorInterface

type PredictorInterface interface {
	Predict(ctx context.Context, req PredictionRequest) (*PredictionResponse, error)
	PredictBulk(ctx context.Context, requests []PredictionRequest) (*BulkPredictionResponse, error)
	PredictBulkStrict(ctx context.Context, requests []PredictionRequest) (*BulkPredictionResponse, error)
	AddTrainingDataBulk(entry []TrainingEntry) error
}

Predictor defines the interface for latency prediction and training.

type ServerStatusResponse

type ServerStatusResponse struct {
	IsReady           bool            `json:"is_ready"`
	ModelType         string          `json:"model_type"`
	ObjectiveType     string          `json:"objective_type,omitempty"`
	Quantile          float64         `json:"quantile"`
	LastModelLoad     *time.Time      `json:"last_model_load"`
	TrainingServerURL string          `json:"training_server_url"`
	ModelsExist       map[string]bool `json:"models_exist"`
}

Server status response

type TrainingEntry

type TrainingEntry struct {
	KVCachePercentage     float64   `json:"kv_cache_percentage"`
	InputTokenLength      int       `json:"input_token_length"`
	NumRequestWaiting     int       `json:"num_request_waiting"`
	NumRequestRunning     int       `json:"num_request_running"`
	NumTokensGenerated    int       `json:"num_tokens_generated"`
	ActualTTFT            float64   `json:"actual_ttft_ms"`
	ActualTPOT            float64   `json:"actual_tpot_ms"`
	PrefixCacheScore      float64   `json:"prefix_cache_score"`
	PodType               string    `json:"pod_type,omitempty"` // "prefill", "decode", or "" for monolithic
	PrefillTokensInFlight int64     `json:"prefill_tokens_in_flight"`
	DecodeTokensInFlight  int64     `json:"decode_tokens_in_flight"`
	Timestamp             time.Time `json:"timestamp"`
}

func NewTrainingEntry

func NewTrainingEntry(
	kvCachePercentage float64,
	inputTokenLength int,
	numRequestWaiting int,
	numRequestRunning int,
	numTokensGenerated int,
	actualTTFT float64,
	actualTPOT float64,
	prefixCacheScore float64,
) (TrainingEntry, error)

NewTrainingEntry is a helper function to create a new TrainingEntry with proper validation.

type XGBoostTrees

type XGBoostTrees struct {
	TTFTTrees []any `json:"ttft_trees"`
	TPOTTrees []any `json:"tpot_trees"`
}

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL