ml

package
v1.18.2 Latest Latest
Warning

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

 Go to latest Published: Oct 13, 2026 License: MIT Imports: 41 Imported by: 0

Documentation

Overview

Package ml is Hermod's model registry and inference path: the models a vhost can call, and the one Predict every caller uses — the Predict node, the REST endpoints and the gRPC InferenceService.

Index

Constants

View Source 
const (
	// MaxFeedbackEvents bounds one feedback call.
	MaxFeedbackEvents = 1000
	// DefaultFeedbackDays is the summary's window when the call names none.
	DefaultFeedbackDays = 30
	// MaxFeedbackDays bounds the summary's window.
	MaxFeedbackDays = 365
)
View Source 
const (
	CallbackTimestampHeader = "X-Hermod-Timestamp"
	CallbackSignatureHeader = "X-Hermod-Signature-256"
	CallbackJobHeader       = "X-Hermod-Job-Id"
)

The headers a callback is signed with.

View Source 
const (
	// JobRetention is how long a job, and its result, is kept after it was
	// made.
	JobRetention = 7 * 24 * time.Hour
	// MaxRowsPerJob bounds one prediction job. The runner sends the rows in
	// calls of at most MaxRowsPerCall, so a job is how a caller with more
	// rows than one call takes sends them.
	MaxRowsPerJob = 50_000

	// MaxJobResultBytes bounds a result Hermod keeps. MongoDB holds one in a
	// document, at most 16 MB, so a larger answer fails the job with what to
	// do instead.
	MaxJobResultBytes = 12 << 20

	// DefaultSyncTimeout bounds a prediction or recommendation call that
	// waits for its answer, unless HERMOD_ML_SYNC_TIMEOUT says otherwise.
	DefaultSyncTimeout = 10 * time.Second
	// CallbackSecretName is the vhost secret a job's callback is signed with.
	CallbackSecretName = "ML_CALLBACK_SECRET" //nolint:gosec // G101: the name of a vhost secret, not a credential.
	// QuotaQueuedJobs names the bound on a vhost's queued jobs.
	QuotaQueuedJobs = "max_queued_jobs"
)
View Source 
const (
	DeviceCPU = "cpu"
	DeviceGPU = "gpu"
)

Devices a training can ask for.

View Source 
const (
	QuotaDatasets             = "max_datasets"
	QuotaDatasetRows          = "max_dataset_rows"
	QuotaDatasetBytes         = "max_dataset_bytes"
	QuotaModels               = "max_models"
	QuotaConcurrentTrainings  = "max_concurrent_trainings"
	QuotaPredictionsPerSecond = "max_predictions_per_second"
)

Quota names, as the API, the errors and the refusals metric spell them.

View Source 
const (
	TriggerSchedule = "schedule"
	TriggerNewRows  = "new_rows"
)

What started a retraining.

View Source 
const (
	// ScheduleInterval is how often a Scheduler looks for due schedules.
	ScheduleInterval = time.Minute

	// MaxBuildRows bounds one build's rows: subjects x k (x plan days).
	MaxBuildRows = 1_000_000
)
View Source 
const (
	BuildAllSalesmen  = "all_salesmen"
	BuildAllCustomers = "all_customers"
	BuildAllProducts  = "all_products"
)

The subjects a build answers for.

View Source 
const (
	BuildReplace = "replace"
	BuildAppend  = "append"
)

The ways a build writes its target table.

View Source 
const (
	// GoLiveAlways puts every new version live.
	GoLiveAlways = "always"
	// GoLiveNever keeps every new version off; an Editor promotes one.
	GoLiveNever = "never"
	// GoLiveIf puts a version live when one of its metrics is within bounds.
	GoLiveIf = "if"
)

How a newly trained version is put live.

View Source 
const CustomPrefix = "custom:"

CustomPrefix marks an algorithm that is a custom training script: "custom:<script name>".

View Source 
const DefaultMaxQueryRows = 1_000_000

DefaultMaxQueryRows caps a dataset filled from a query when the caller names no cap.

View Source 
const MaxRowsPerCall = 1000

MaxRowsPerCall bounds one prediction call. A caller with more rows splits them; this keeps one request from holding a model server, or Hermod's memory, hostage.

View Source 
const (
	// RetrainInterval is how often a Retrainer looks for models that are due.
	// A schedule is therefore kept to within a minute.
	RetrainInterval = time.Minute
)

Variables

View Source 
var (
	// ErrJobNotFound is returned for a job the vhost does not hold.
	ErrJobNotFound = errors.New("no job by that id")
	// ErrBadJob marks a job that is wrong in itself.
	ErrBadJob = errors.New("invalid job")
	// ErrSyncTimeout is returned for a call that waited for its answer
	// longer than the sync time limit.
	ErrSyncTimeout = errors.New("the call took longer than the limit for an answer while you wait")
	// ErrJobResultTooLarge is returned for a result over MaxJobResultBytes.
	ErrJobResultTooLarge = errors.New("the result is too large to keep")
	// ErrNoCallbackSecret is returned, with ErrBadJob, for a callback URL on
	// a vhost that holds no CallbackSecretName to sign it with.
	ErrNoCallbackSecret = errors.New("no callback signing secret")
)
View Source 
var (
	// ErrCustomScriptsOff is returned for anything to do with custom training
	// scripts while the server has them off, which is the default.
	ErrCustomScriptsOff = errors.New("custom training scripts are off on this server: an operator turns them on with HERMOD_ML_CUSTOM_SCRIPTS=true")
	// ErrNoCustomPool is returned for a custom-script training when no worker
	// pool is set aside for custom scripts. They never run on the main worker.
	ErrNoCustomPool = errors.New("custom training scripts run only on their own worker pool, and none is configured: set HERMOD_ML_CUSTOM_WORKER_URL")
	// ErrNoGPUPool is returned for a training on the gpu when no GPU worker
	// pool is configured.
	ErrNoGPUPool = errors.New(`no GPU worker pool is configured: set HERMOD_ML_GPU_WORKER_URL, or train with device "cpu"`)
	// ErrScriptNotFound is returned for a script the vhost does not hold.
	ErrScriptNotFound = errors.New("training script not found")
	// ErrScriptsUnsupported is returned when the storage backend cannot hold
	// training scripts.
	ErrScriptsUnsupported = errors.New("this storage backend cannot hold training scripts")
	// ErrBadTraining marks a training request that is wrong in itself.
	ErrBadTraining = errors.New("invalid training request")
)
View Source 
var (
	// ErrRecommender is returned for rows sent to a recommendation model.
	ErrRecommender = errors.New("this is a recommendation model: call it with a user or an item, not rows")
	// ErrNotRecommender is returned for a user or an item sent to a model
	// that predicts a column.
	ErrNotRecommender = errors.New("this model predicts from rows: a user or an item applies only to a recommendation model")
	// ErrBadRecommendRequest marks a recommendation call that is wrong in
	// itself.
	ErrBadRecommendRequest = errors.New("invalid recommendation request")
)
View Source 
var (
	// ErrModelNotFound is returned for a model the vhost does not hold.
	ErrModelNotFound = errors.New("model not found")
	// ErrTooManyRows is returned for a call over MaxRowsPerCall.
	ErrTooManyRows = fmt.Errorf("at most %d rows per call", MaxRowsPerCall)
	// ErrServingOff is returned when a model has no serving key.
	ErrServingOff = errors.New("serving is not enabled for this model")
	// ErrBadServingKey is returned when the key presented is not the model's.
	ErrBadServingKey = errors.New("invalid serving key")
	// ErrNoWorker is returned for training, datasets, or a trained model's
	// prediction when Hermod has no ML worker configured.
	ErrNoWorker = errors.New("no ML worker is connected: connect one on the Machine Learning page, " +
		"or set HERMOD_ML_WORKER_URL")
	// ErrNoLiveVersion is returned for a trained model none of whose versions
	// has been put live.
	ErrNoLiveVersion = errors.New("no version of this model is live yet")
	// ErrVersionNotFound is returned for a version the model does not have.
	ErrVersionNotFound = errors.New("model version not found")
)
View Source 
var (
	// ErrWorkerFromEnvironment is returned for a change to the worker
	// connection while HERMOD_ML_WORKER_URL names one, which wins.
	ErrWorkerFromEnvironment = errors.New("the ML worker is set by the server's environment " +
		"(HERMOD_ML_WORKER_URL): change it there")
	// ErrBadWorkerURL is returned for a worker address Hermod cannot call.
	ErrBadWorkerURL = errors.New("invalid ML worker address")
	// ErrSettingsUnsupported is returned when the store keeps no settings.
	ErrSettingsUnsupported = errors.New("this storage backend cannot keep settings")
)
View Source 
var BuildColumns = []string{"model", "version", "recommendation_id", "for_id", "item_id", "rank", "score", "plan_date",
	"reasons_text", "built_at"}

BuildColumns are the columns a build writes, in order.

View Source 
var ErrBadFeedback = errors.New("invalid feedback")

ErrBadFeedback marks a feedback call that is wrong in itself.

View Source 
var ErrBadSchedule = errors.New("invalid schedule")

ErrBadSchedule marks a schedule that is wrong in itself.

View Source 
var ErrNoBlobStore = errors.New("the metadata storage cannot keep the built-in ML engine's datasets and models " +
	"(it is not set up yet, or its backend keeps none)")

ErrNoBlobStore is the built-in engine's answer while Hermod's storage cannot keep its datasets and versions: it is not set up yet, or its backend keeps no ML blobs and no directory was given instead.

View Source 
var ErrNotTrainedHere = errors.New("only a model trained here can retrain")

ErrNotTrainedHere is returned for a retrain policy on a model served elsewhere: Hermod cannot train it.

View Source 
var ErrQuotaExceeded = errors.New("ML quota exceeded")

ErrQuotaExceeded matches every QuotaError with errors.Is.

View Source 
var ErrScoringNotTrained = errors.New("only a model trained here can be scored in-process")

ErrScoringNotTrained is returned for in-process scoring of a model served elsewhere: Hermod does not hold its graph.

View Source 
var ErrTrainingRunning = errors.New("this model is already being trained")

ErrTrainingRunning is returned for a model another training holds, here or on another Hermod sharing the store.

Functions

func BuiltinBlobs added in v1.18.2 

func BuiltinBlobs(store func() any, fallback builtin.Blobs) builtin.Blobs

BuiltinBlobs keeps the built-in engine's datasets and model versions in Hermod's metadata store, the one store returns at each call (setup and a database switch replace it while Hermod runs). So every replica sharing the database serves every model, and none depends on a server's disk. fallback, which may be nil, is used only by a store whose backend keeps no ML blobs.

func CheckCallbackURL 

func CheckCallbackURL(raw string) error

CheckCallbackURL reports what is wrong with a callback URL. Hermod posts to it on its own, so it is held to the rules of a URL fetched from somewhere the operator does not control (handlers.IsSafeWasmURL): http or https, a host that is neither localhost nor a loopback, link-local, private or unspecified address. A name that resolves to such an address is refused again when the callback is sent, by httpclient.SafeDialer. Credentials in the URL are refused too, so they cannot end up in the job listing.

func JobStatusPath 

func JobStatusPath(j storage.MLJob) string

JobStatusPath is where a job is read: the serving path for a job submitted with a serving key, the vhost's API otherwise.

func QuotaDefaultsFromEnv 

func QuotaDefaultsFromEnv() storage.MLQuotas

QuotaDefaultsFromEnv reads the server-wide defaults. A variable that is unset, unreadable or negative sets nothing, so that quota is unlimited for a vhost without its own.

func RefreshWorker added in v1.18.1 

func RefreshWorker(ctx context.Context, store any, p *worker.Provider)

RefreshWorker re-reads the saved worker into p when it is due (see worker.Provider.Refresh). A store without settings, or a failed read, leaves p as it is; the failure is logged.

func ServingJob 

func ServingJob(j storage.MLJob, model string) bool

ServingJob reports whether a serving key may read the job: one submitted with a serving key, for the model the key is for.

func SignCallback 

func SignCallback(secret string, timestamp int64, body []byte) string

SignCallback is the signature of a callback body sent at timestamp (Unix seconds): "sha256=" and the hex HMAC-SHA256, keyed with the vhost's ML_CALLBACK_SECRET, of the timestamp, a full stop and the body. Signing the timestamp lets a receiver refuse a replayed callback by its age.

func ValidateRetrainPolicy 

func ValidateRetrainPolicy(p storage.MLRetrainPolicy) error

ValidateRetrainPolicy reports what is wrong with a retrain policy.

func WithCaller 

func WithCaller(ctx context.Context, kind, id string) context.Context

WithCaller records who is calling Predict, for the prediction log: one of the storage.MLCaller kinds, and the workflow for a Predict node.

func WriteBuild 

func WriteBuild(ctx context.Context, db *sql.DB, driver string, t BuildTarget, model string, rows []BuildRow) error

WriteBuild writes rows to a build's target table in one transaction: with mode replace, the model's earlier rows are deleted first. driver is the source's type. The table's columns are BuildColumns; plan_date is NULL outside a visit plan.

Types

type BuildRow 

type BuildRow struct {
	Model            string    `json:"model"`
	Version          string    `json:"version"`
	RecommendationID string    `json:"recommendation_id"`
	ForID            string    `json:"for_id"`
	ItemID           string    `json:"item_id"`
	Rank             int       `json:"rank"`
	Score            float64   `json:"score"`
	PlanDate         string    `json:"plan_date,omitempty"`
	ReasonsText      string    `json:"reasons_text"`
	BuiltAt          time.Time `json:"built_at"`
}

BuildRow is one written row: one item of one subject's answer. PlanDate is the day a visit plan puts it on, empty otherwise.

type BuildTarget 

type BuildTarget struct {
	SourceID string `json:"source_id"`
	Table    string `json:"table"`
	Mode     string `json:"mode,omitempty"`
}

BuildTarget is the table a build writes: Table of the database source SourceID, whose rows of the model are replaced or appended to (Mode, replace when empty). The table must exist with the columns BuildColumns.

type CallbackPayload 

type CallbackPayload struct {
	JobID      string          `json:"job_id"`
	VHost      string          `json:"vhost"`
	Model      string          `json:"model"`
	Kind       string          `json:"kind"`
	Status     string          `json:"status"`
	Error      string          `json:"error,omitempty"`
	FinishedAt *time.Time      `json:"finished_at,omitempty"`
	StatusURL  string          `json:"status_url"`
	ResultURL  string          `json:"result_url,omitempty"`
	Result     json.RawMessage `json:"result,omitempty"`
}

CallbackPayload is the body a job's callback posts. Result is the answer when it is small enough to be kept on the job; otherwise ResultURL is where it is read, with the same credentials as StatusURL.

type Capabilities 

type Capabilities struct {
	// CustomScripts: custom scripts are on and their pool is configured.
	CustomScripts bool `json:"custom_scripts"`
	// ScriptsEnabled: custom scripts are on, pool or not; scripts can be
	// managed but not trained with until the pool is configured.
	ScriptsEnabled bool `json:"scripts_enabled"`
	GPU            bool `json:"gpu"`
}

Capabilities says which training options this server offers, for the UI.

type DatasetRefresh 

type DatasetRefresh struct {
	Schedule string `json:"schedule"`
	SourceID string `json:"source_id"`
	Query    string `json:"query"`
	MaxRows  int    `json:"max_rows,omitempty"`
}

DatasetRefresh is a dataset's refresh: run Query on the database source SourceID on Schedule and replace the dataset's rows with what it returns, at most MaxRows (the worker's default when 0).

func (DatasetRefresh) Validate 

func (r DatasetRefresh) Validate() error

Validate reports what is wrong with a refresh.

type DriftStatus 

type DriftStatus struct {
	Report *monitor.Report `json:"report"`
	Reason string          `json:"reason,omitempty"`
	// Window and MinRows say when a report is made: every Window, for a
	// window of at least MinRows predicted rows.
	Window  string `json:"window,omitempty"`
	MinRows int    `json:"min_rows,omitempty"`
	// Recommendation is a recommendation model's latest report, which takes
	// the place of feature drift: coverage and popular share.
	Recommendation *monitor.RecommendationReport `json:"recommendation,omitempty"`
}

DriftStatus is a model's latest drift report, or why it has none.

type FeedbackEvent 

type FeedbackEvent struct {
	ID      string     `json:"id"`
	Outcome string     `json:"outcome"`
	At      *time.Time `json:"at,omitempty"`
}

FeedbackEvent is one outcome of one recommended item: ID is the item's id (a customer or a product) and At when it happened, now when left out.

type FeedbackRequest 

type FeedbackRequest struct {
	RecommendationID string          `json:"recommendation_id"`
	Events           []FeedbackEvent `json:"events"`
}

FeedbackRequest is the body of POST .../feedback.

type FeedbackSummary 

type FeedbackSummary struct {
	Days           int     `json:"days"`
	Shown          int     `json:"shown"`
	Accepted       int     `json:"accepted"`
	Dismissed      int     `json:"dismissed"`
	Visited        int     `json:"visited"`
	Ordered        int     `json:"ordered"`
	AcceptanceRate float64 `json:"acceptance_rate"`
}

FeedbackSummary counts a model's feedback by outcome over Days days. AcceptanceRate is accepted / shown, or accepted / (accepted + dismissed) when nothing was reported shown, and 0 with neither.

type GoLive 

type GoLive storage.MLGoLive

GoLive says whether a newly trained version is put live. The zero value keeps it off, the safe choice for a caller that says nothing. It is the rule storage keeps with a retrain policy (storage.MLGoLive): Mode, and Metric with Min and Max for GoLiveIf, where an empty Metric means "score", which is accuracy for a classifier and R² for a regression.

func (GoLive) Validate 

func (g GoLive) Validate() error

Validate reports what is wrong with the rule.

type JobRunner 

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

JobRunner runs the queued ML jobs of every vhost. Every Hermod serving the API runs one; a job is claimed in storage before it runs, so each runs on one of them, and a job whose Hermod stopped is taken over by another once its claim expires.

func NewJobRunner 

func NewJobRunner(svc func() *Service, logger hermod.Logger) *JobRunner

NewJobRunner builds a JobRunner. svc is called on every round, because setup and a database switch replace the store while Hermod runs. logger may be nil.

func (*JobRunner) Run 

func (r *JobRunner) Run(ctx context.Context)

Run runs jobs, up to its number of workers at a time, and sends their callbacks, until ctx ends. Jobs still running then are left to their claims: another Hermod, or this one restarted, takes them over.

func (*JobRunner) Tick 

func (r *JobRunner) Tick(ctx context.Context)

Tick runs, one after another, every job that can be claimed now, then sends the callbacks that are due. Run does the same with workers; Tick is for a caller that wants it done before it goes on.

type JobSubmission 

type JobSubmission struct {
	VHost, Model, Kind string

	Rows      []inference.Row
	Recommend *RecommendRequest
	Scenario  *ScenarioRequest
	Train     *TrainJob

	CallbackURL string
	CallerKind  string
	CreatedBy   string
}

JobSubmission is one job to queue: the model of a vhost, the kind, and the request of that kind (Rows for a prediction, Recommend or Scenario for recommendations, Train for a training). CallbackURL, when set, is told when the job ends. CallerKind and CreatedBy say who asked.

type ModelBuild 

type ModelBuild struct {
	Schedule   string      `json:"schedule"`
	For        string      `json:"for"`
	PeriodDays int         `json:"period_days,omitempty"`
	K          int         `json:"k,omitempty"`
	Target     BuildTarget `json:"target"`
}

ModelBuild is a recommendation model's scheduled build.

func (ModelBuild) Validate 

func (b ModelBuild) Validate(scheduled bool) error

Validate reports what is wrong with a build, as far as the build alone can tell; the model's scenario is checked when it runs.

type Pools 

type Pools struct {
	// Custom runs custom training scripts, and CustomScripts allows them at
	// all. A worker pool that runs uploaded code is kept apart from the one
	// holding every vhost's data and serving its models.
	Custom        *worker.Client
	CustomScripts bool
	// GPU runs trainings that ask for device "gpu".
	GPU *worker.Client
}

Pools are the worker pools a training can run on besides the main worker. A pool trains only: it keeps nothing between trainings, and every version it makes is served by the main worker.

func PoolsFromEnv 

func PoolsFromEnv() Pools

PoolsFromEnv reads HERMOD_ML_CUSTOM_SCRIPTS, HERMOD_ML_CUSTOM_WORKER_URL and _TOKEN, and HERMOD_ML_GPU_WORKER_URL and _TOKEN. A HERMOD_ML_CUSTOM_SCRIPTS that is not a boolean leaves custom scripts off.

type QuotaError 

type QuotaError struct {
	VHost     string
	Quota     string
	Retryable bool
	// contains filtered or unexported fields
}

QuotaError is a refusal because a vhost is at one of its ML quotas. Retryable is set when waiting is enough (a training finishing, the next second's predictions); otherwise an Administrator has to raise the quota or the vhost has to free something.

func (*QuotaError) Error 

func (e *QuotaError) Error() string

func (*QuotaError) Is 

func (e *QuotaError) Is(target error) bool

Is makes errors.Is(err, ErrQuotaExceeded) true for every QuotaError.

type Quotas 

type Quotas struct {
	MaxDatasets             int64   `json:"max_datasets"`
	MaxDatasetRows          int64   `json:"max_dataset_rows"`
	MaxDatasetBytes         int64   `json:"max_dataset_bytes"`
	MaxModels               int64   `json:"max_models"`
	MaxConcurrentTrainings  int64   `json:"max_concurrent_trainings"`
	MaxPredictionsPerSecond float64 `json:"max_predictions_per_second"`
}

Quotas are the limits in force for a vhost: its own where it has them, the server's default otherwise. Zero means no limit.

type RecommendRequest 

type RecommendRequest struct {
	User        any   `json:"user,omitempty"`
	Item        any   `json:"item,omitempty"`
	K           int   `json:"k,omitempty"`
	ExcludeSeen *bool `json:"exclude_seen,omitempty"`
}

RecommendRequest is one call of a recommendation model: the items for a person (User), or the items like one item (Item). User and Item are ids as JSON gives them: a string, or a number for an id column that held numbers. K is how many items, recommend.DefaultK when zero and at most recommend.MaxK. ExcludeSeen, for a person, leaves out what they already have; it is true when not given.

type Retrainer 

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

Retrainer retrains the models whose policies say so. Every Hermod serving the API runs one; a model's training claim makes each retraining happen on one of them, once.

func NewRetrainer 

func NewRetrainer(svc func() *Service, logger hermod.Logger) *Retrainer

NewRetrainer builds a Retrainer. svc is called on every round, because setup and a database switch replace the store while Hermod runs. logger may be nil.

func (*Retrainer) Run 

func (r *Retrainer) Run(ctx context.Context)

Run looks for due models every RetrainInterval until ctx ends.

func (*Retrainer) Tick 

func (r *Retrainer) Tick(ctx context.Context)

Tick retrains, one after another, every model that is due now.

type ScenarioAnswer 

type ScenarioAnswer struct {
	RecommendationID string             `json:"recommendation_id"`
	Results          []recommend.Result `json:"results"`
}

ScenarioAnswer is a scenario call's answer: one result per subject, and the id feedback on it is sent with.

type ScenarioFor 

type ScenarioFor struct {
	Customer  any   `json:"customer,omitempty"`
	Customers []any `json:"customers,omitempty"`
	Product   any   `json:"product,omitempty"`
	Salesman  any   `json:"salesman,omitempty"`
	Territory any   `json:"territory,omitempty"`
}

ScenarioFor names who or what a scenario call is for. Ids are as JSON gives them: a string, or a number for an id column that held numbers. Territory, given here or at the top of the request, is what a rule's from_request "territory" binds to.

type ScenarioPeriod 

type ScenarioPeriod struct {
	From string `json:"from"`
	To   string `json:"to,omitempty"`
}

ScenarioPeriod is a visit plan's days, both included, as YYYY-MM-DD. To may be left out: the model's period from From.

type ScenarioRequest 

type ScenarioRequest struct {
	For         ScenarioFor      `json:"for"`
	Territory   any              `json:"territory,omitempty"`
	Period      *ScenarioPeriod  `json:"period,omitempty"`
	K           int              `json:"k,omitempty"`
	Rules       []recommend.Rule `json:"rules,omitempty"`
	ExcludeSeen *bool            `json:"exclude_seen,omitempty"`
	Respond     string           `json:"respond,omitempty"`
	CallbackURL string           `json:"callback_url,omitempty"`
}

ScenarioRequest is one call of POST .../recommend. Respond and CallbackURL ask for an asynchronous answer, which the REST API gives by queueing the call as a job; RecommendScenario itself answers at once, and refuses them.

type ScheduleRunner 

type ScheduleRunner interface {
	MLDatasetFromQuery(ctx context.Context, vhost, dataset, sourceID, query string, maxRows int) (int, error)
	OpenMLTarget(ctx context.Context, vhost, sourceID string) (*sql.DB, string, error)
}

ScheduleRunner is what a schedule's run needs of the engine: refilling a dataset from a database source, and opening a source of the vhost for a build's writes. The registry implements it.

type ScheduleStatus 

type ScheduleStatus struct {
	Schedule  string     `json:"schedule"`
	LastRun   *time.Time `json:"last_run"`
	LastRows  int        `json:"last_rows"`
	LastError string     `json:"last_error"`
	NextRun   *time.Time `json:"next_run"`
}

ScheduleStatus is what GET shows of a schedule: its cron expression, when it last ran and what that did, and when it runs next.

type Scheduler 

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

Scheduler runs the due dataset refreshes and model builds. Every Hermod serving the API runs one; a schedule's claim makes each run happen on one of them, once.

func NewScheduler 

func NewScheduler(svc func() *Service, runner ScheduleRunner, logger hermod.Logger) *Scheduler

NewScheduler builds a Scheduler. svc is called on every round, as the Retrainer's is; runner reaches the engine's database sources. logger may be nil.

func (*Scheduler) Run 

func (sc *Scheduler) Run(ctx context.Context)

Run looks for due schedules every ScheduleInterval until ctx ends.

func (*Scheduler) Tick 

func (sc *Scheduler) Tick(ctx context.Context)

Tick runs, one after another, every schedule that is due now.

type ScoringStatus 

type ScoringStatus struct {
	// Scoring is the setting: storage.MLScoringWorker or MLScoringInProcess.
	Scoring string `json:"scoring"`
	// InProcess reports whether the live version is scored in-process.
	InProcess bool `json:"in_process"`
	// Version is the live version the status is about.
	Version string `json:"version,omitempty"`
	// Ops are the operators of its graph, when it was loaded.
	Ops []string `json:"ops,omitempty"`
	// Reason says why in-process scoring is set but not in use.
	Reason string `json:"reason,omitempty"`
}

ScoringStatus says how a model's predictions are computed now.

type Service 

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

Service resolves models and calls them.

func NewService 

func NewService(store func() any, sec secrets.ScopedManager, client *inference.Client) *Service

NewService builds a Service. store is called on every use, because setup and a database switch replace the store while Hermod runs. secrets answers a model's token secret for its vhost; it may be nil when no model uses one. client may be nil for the default. The ML worker is the one in force in worker.Default when it is built.

func (*Service) AuthorizeServing 

func (s *Service) AuthorizeServing(ctx context.Context, vhost, name, key string) error

AuthorizeServing checks the key an application presented for a model. The comparison is between hashes and in constant time.

func (*Service) BuildRecommendations 

func (s *Service) BuildRecommendations(ctx context.Context, vhost, name string, b ModelBuild) ([]BuildRow, error)

BuildRecommendations answers b for every salesman, customer or product of the model's live version, as rows. The answers share one recommendation_id. A build is not counted against the per-second quota: it is a batch the vhost scheduled, not traffic.

func (*Service) CancelJob 

func (s *Service) CancelJob(ctx context.Context, vhost, id string) (storage.MLJob, error)

CancelJob cancels a queued job at once, and asks the Hermod running a running one to stop it, which it does within a few seconds. A finished job is returned as it is.

func (*Service) Capabilities 

func (s *Service) Capabilities() Capabilities

Capabilities reports the training options the configured pools allow.

func (*Service) ClearDatasetRefresh 

func (s *Service) ClearDatasetRefresh(ctx context.Context, vhost, dataset string) error

ClearDatasetRefresh stops a dataset refreshing itself; ErrNotFound for none.

func (*Service) ClearModelBuild 

func (s *Service) ClearModelBuild(ctx context.Context, vhost, name string) error

ClearModelBuild stops a model's scheduled build; ErrNotFound for none.

func (*Service) ClearRetrainPolicy 

func (s *Service) ClearRetrainPolicy(ctx context.Context, vhost, name string) error

ClearRetrainPolicy stops a model retraining by itself. What its last retraining did stays on the model.

func (*Service) DatasetFromQuery 

func (s *Service) DatasetFromQuery(ctx context.Context, vhost, dataset string, db *sql.DB, query string, maxRows int) (int, error)

DatasetFromQuery replaces a vhost's dataset with the rows a query returns, streaming them to the worker in chunks. Only a SELECT (or WITH … SELECT) is run, inside a read-only transaction where the driver offers one. A query returning more than maxRows rows, or more than the vhost's row or byte quota, is refused, and the dataset is left with the rows sent so far, which the next fill replaces.

func (*Service) DeleteModel 

func (s *Service) DeleteModel(ctx context.Context, vhost, name string) error

DeleteModel removes a model from the registry and, for a model Hermod trained, its versions from the worker. The registry entry goes first, so a worker that cannot be reached leaves files behind, never a model that points at nothing.

func (*Service) DeleteScript 

func (s *Service) DeleteScript(ctx context.Context, vhost, name string) error

DeleteScript removes every version of a script. Model versions it trained keep their record of its name and SHA-256.

func (*Service) DisableServing 

func (s *Service) DisableServing(ctx context.Context, vhost, name string) error

DisableServing removes the model's serving key.

func (*Service) Drift 

func (s *Service) Drift(ctx context.Context, vhost, name string) (DriftStatus, error)

Drift returns the model's latest drift report.

func (*Service) FeedbackSummary 

func (s *Service) FeedbackSummary(ctx context.Context, vhost, name string, days int) (FeedbackSummary, error)

FeedbackSummary counts a model's feedback over the last days days: 0 means DefaultFeedbackDays, at most MaxFeedbackDays.

func (*Service) Job 

func (s *Service) Job(ctx context.Context, vhost, id string) (storage.MLJob, error)

Job returns one job of the vhost, or ErrJobNotFound.

func (*Service) JobResult 

func (s *Service) JobResult(ctx context.Context, vhost, id string) (json.RawMessage, error)

JobResult returns a finished job's result, wherever it is kept, or nil for a job without one.

func (*Service) Jobs 

func (s *Service) Jobs() (storage.MLJobStore, error)

Jobs returns the store's job store, or ErrMLJobsUnsupported.

func (*Service) ListJobs 

func (s *Service) ListJobs(ctx context.Context, f storage.MLJobFilter) ([]storage.MLJob, int, error)

ListJobs returns a page of the vhost's jobs, newest first, and how many match.

func (*Service) ListScripts 

func (s *Service) ListScripts(ctx context.Context, vhost string) ([]storage.MLScript, error)

ListScripts lists the latest version of each of the vhost's scripts.

func (*Service) Model 

func (s *Service) Model(ctx context.Context, vhost, name string) (storage.MLModel, error)

Model returns one model of the vhost, or ErrModelNotFound.

func (*Service) Models 

func (s *Service) Models() (storage.MLModelStore, error)

Models returns the store's model registry, or ErrMLModelsUnsupported.

func (*Service) Predict 

func (s *Service) Predict(ctx context.Context, vhost, name string, rows []inference.Row) ([]inference.Row, error)

Predict sends rows to the vhost's model and returns one prediction per row. Every row counts against the vhost's predictions-per-second quota.

func (*Service) PredictionLogs 

func (s *Service) PredictionLogs(ctx context.Context, vhost, name string, limit int) ([]storage.MLPredictionLog, error)

PredictionLogs returns a model's most recently logged predictions, newest first.

func (*Service) Promote 

func (s *Service) Promote(ctx context.Context, vhost, name, version, by string) error

Promote puts one of a trained model's versions live: a later one, or an earlier one to roll back.

func (*Service) PutModel 

func (s *Service) PutModel(ctx context.Context, m storage.MLModel) error

PutModel saves a model's definition. A model the vhost does not hold yet counts against its model quota.

func (*Service) PutScript 

func (s *Service) PutScript(ctx context.Context, sc storage.MLScript) (storage.MLScript, error)

PutScript saves a script's source as its next version.

func (*Service) QuotaDefaults 

func (s *Service) QuotaDefaults() storage.MLQuotas

QuotaDefaults returns the server-wide defaults.

func (*Service) Quotas 

func (s *Service) Quotas(ctx context.Context, vhost string) (Quotas, error)

Quotas returns the limits in force for the vhost.

func (*Service) Recommend 

func (s *Service) Recommend(ctx context.Context, vhost, name string, req RecommendRequest) ([]recommend.Item, error)

Recommend calls a vhost's recommendation model, in-process from its live version's artifact. A person or item the model does not know gets the most popular items, marked recommend.ReasonPopular, so the answer is never empty. Every call counts as one row against the vhost's predictions-per-second quota, and is logged and monitored as a prediction.

func (*Service) RecommendCheck 

func (s *Service) RecommendCheck(ctx context.Context, vhost string, spec worker.TrainSpec) (worker.RecommendHealth, error)

RecommendCheck is a recommendation scenario's data health check, on the main worker, which holds the datasets: whether each role's dataset and columns are there and what vendors' minimums they meet. A body that names neither a scenario nor #268's recommend columns has nothing to check.

func (*Service) RecommendScenario 

func (s *Service) RecommendScenario(ctx context.Context, vhost, name string, req ScenarioRequest) (ScenarioAnswer, error)

RecommendScenario answers a call of a scenario model in-process from its live version's artifact. Every subject (each customer of a batch) counts as one row against the vhost's predictions-per-second quota, and the call is logged and monitored as a prediction carrying its recommendation_id.

func (*Service) RecommendWithID 

func (s *Service) RecommendWithID(ctx context.Context, vhost, name string, req RecommendRequest) ([]recommend.Item, string, error)

RecommendWithID is Recommend, and the recommendation_id of its answer, for feedback.

func (*Service) RecordFeedback 

func (s *Service) RecordFeedback(ctx context.Context, vhost, name string, req FeedbackRequest) error

RecordFeedback stores what happened to the items of one answer, every event or none, and refreshes the model's acceptance-rate gauge.

func (*Service) RotateServingKey 

func (s *Service) RotateServingKey(ctx context.Context, vhost, name string) (string, error)

RotateServingKey makes a new serving key for the model, replacing any earlier one, and returns it. Only its hash is stored: this is the one time the key can be read.

func (*Service) RunBuild 

func (s *Service) RunBuild(ctx context.Context, runner ScheduleRunner, vhost, name string, b ModelBuild) (int, error)

RunBuild builds and writes b, through runner's connection to the target source, which must be one of the vhost's. It returns the rows written.

func (*Service) Schedule 

func (s *Service) Schedule(ctx context.Context, vhost, kind, name string) (*storage.MLSchedule, *ScheduleStatus, error)

Schedule is a dataset's refresh or a model's build, and its status; nil for none, or for a store without schedules.

func (*Service) Scoring 

func (s *Service) Scoring(ctx context.Context, vhost, name string) (ScoringStatus, error)

Scoring reports how the model is scored. With in-process scoring set, it loads the live version's graph, as the next prediction would, so the answer is the one Predict acts on.

func (*Service) Script 

func (s *Service) Script(ctx context.Context, vhost, name string) (storage.MLScript, error)

Script returns the latest version of a script, with its source.

func (*Service) ScriptVersions 

func (s *Service) ScriptVersions(ctx context.Context, vhost, name string) ([]storage.MLScript, error)

ScriptVersions lists every version of a script, newest first.

func (*Service) Scripts 

func (s *Service) Scripts() (storage.MLScriptStore, error)

Scripts returns the store's training scripts, or ErrCustomScriptsOff when the server has them off.

func (*Service) SetDatasetRefresh 

func (s *Service) SetDatasetRefresh(ctx context.Context, vhost, dataset string, r DatasetRefresh, by string) (ScheduleStatus, error)

SetDatasetRefresh makes a dataset refresh itself on a schedule.

func (*Service) SetMCPExposed 

func (s *Service) SetMCPExposed(ctx context.Context, vhost, name string, exposed bool, by string) (storage.MLModel, error)

SetMCPExposed offers the model as a tool on Hermod's MCP server, or stops offering it, and returns the model as saved.

func (*Service) SetModelBuild 

func (s *Service) SetModelBuild(ctx context.Context, vhost, name string, b ModelBuild, by string) (ScheduleStatus, error)

SetModelBuild makes a recommendation model build its answers on a schedule.

func (*Service) SetMonitoring 

func (s *Service) SetMonitoring(ctx context.Context, vhost, name string, mon storage.MLMonitoring, by string) (storage.MLModel, error)

SetMonitoring replaces how a model is monitored, and nothing else of it.

func (*Service) SetQuotas 

func (s *Service) SetQuotas(ctx context.Context, q storage.MLQuotas) error

SetQuotas replaces the vhost's own quotas. Each call to a quota reads them afresh, so a change applies from the next call on every replica.

func (*Service) SetRetrainPolicy 

func (s *Service) SetRetrainPolicy(ctx context.Context, vhost, name string, p storage.MLRetrainPolicy, by string) (storage.MLModel, error)

SetRetrainPolicy makes a trained model train again by itself, on a schedule, after its dataset grows, or both. It returns the model with its policy.

func (*Service) SetScoring 

func (s *Service) SetScoring(ctx context.Context, vhost, name, scoring string) error

SetScoring sets where a trained model's predictions are computed: storage.MLScoringWorker, or storage.MLScoringInProcess to score its live version inside Hermod whenever its graph allows.

func (*Service) StoredQuotas 

func (s *Service) StoredQuotas(ctx context.Context, vhost string) (storage.MLQuotas, error)

StoredQuotas returns the quotas set on the vhost itself; a vhost with none set gets an empty set, every quota falling back to the default.

func (*Service) SubmitJob 

func (s *Service) SubmitJob(ctx context.Context, sub JobSubmission) (storage.MLJob, error)

SubmitJob queues a job and returns it. Everything a job could be refused for in itself is refused here, so a queued job fails only for what happens when it runs. A vhost may have at most HERMOD_ML_MAX_QUEUED_JOBS jobs waiting; the quotas a call is held to apply when the job runs.

func (*Service) Sync 

func (s *Service) Sync(ctx context.Context, vhost, name string, call func(context.Context) error) error

Sync runs call, a prediction or recommendation a caller waits for, within the sync time limit, and turns running over it into a SyncTimeoutError that says how to run the call as a job instead. A model whose own timeout is longer than the limit keeps its own: whoever set it expects its calls to take that long. A caller that went away is not told it timed out.

func (*Service) Train 

func (s *Service) Train(ctx context.Context, vhost, name string, spec worker.TrainSpec, rule GoLive, by string) (TrainResult, error)

Train trains a new version of a vhost's model on the worker and registers the model, if it is new. Whether the version goes live is up to rule; a version kept off stays on the worker, and Promote can put it live later.

One training of a model runs at a time, across every Hermod sharing the store: while another holds the model, Train returns ErrTrainingRunning.

The training takes one of the vhost's training slots, and a new model counts against its model quota.

func (*Service) TrainCheck 

func (s *Service) TrainCheck(ctx context.Context, vhost string, spec worker.TrainSpec) (worker.TrainCheck, error)

TrainCheck is the pre-flight check of a training: what the worker makes of the dataset (the task, the target, each column's role, warnings), with nothing trained. A spec a training would refuse in itself is refused here the same way. The check runs on the main worker whatever the spec's device or algorithm, since it holds the datasets and only measures them; it takes neither the model's claim nor a training slot. A recommender has no column to predict, so it is refused with what to do instead.

func (*Service) UploadDataset 

func (s *Service) UploadDataset(ctx context.Context, vhost, dataset, format string, body io.Reader) (worker.DatasetInfo, error)

UploadDataset replaces a vhost's dataset with a CSV or Excel file. A new dataset counts against the vhost's dataset quota, the file against its byte quota as it streams, and the rows the worker finds in it against the row quota. The worker replaces a dataset as it reads the file, so one found to hold too many rows is removed rather than left over the quota.

func (*Service) VersionStats 

func (s *Service) VersionStats(ctx context.Context, vhost, name, version string) (map[string]worker.FeatureStats, error)

VersionStats returns the training stats of one version of a trained model. It is the monitor's StatsFunc.

func (*Service) Versions 

func (s *Service) Versions(ctx context.Context, vhost, name string) ([]worker.Version, error)

Versions lists a trained model's versions, newest first.

func (*Service) WithLogStore 

func (s *Service) WithLogStore(logs func() any) *Service

WithLogStore sets the store prediction logs are read from and deleted in; without one they are the model store's.

func (*Service) WithMonitor 

func (s *Service) WithMonitor(m *monitor.Monitor) *Service

WithMonitor sets where predictions are reported; nil means nowhere.

func (*Service) WithPools 

func (s *Service) WithPools(p Pools) *Service

WithPools replaces the worker pools the environment names.

func (*Service) WithQuotaDefaults 

func (s *Service) WithQuotaDefaults(d storage.MLQuotas) *Service

WithQuotaDefaults replaces the server-wide defaults the environment sets.

func (*Service) WithSyncTimeout 

func (s *Service) WithSyncTimeout(d time.Duration) *Service

WithSyncTimeout replaces the sync time limit HERMOD_ML_SYNC_TIMEOUT sets; zero turns it off.

func (*Service) WithWorker 

func (s *Service) WithWorker(w *worker.Client) *Service

WithWorker replaces the ML worker in force; nil means none.

func (*Service) WithWorkerProvider added in v1.18.1 

func (s *Service) WithWorkerProvider(p *worker.Provider) *Service

WithWorkerProvider takes the ML worker in force in p instead of worker.Default's.

func (*Service) Worker 

func (s *Service) Worker() (*worker.Client, error)

Worker returns the configured ML worker, or ErrNoWorker.

type SyncTimeoutError 

type SyncTimeoutError struct{ Limit time.Duration }

SyncTimeoutError is a call that waited longer than Limit for its answer.

func (*SyncTimeoutError) Error 

func (e *SyncTimeoutError) Error() string

func (*SyncTimeoutError) Is 

func (e *SyncTimeoutError) Is(target error) bool

Is makes errors.Is(err, ErrSyncTimeout) true.

type TrainJob 

type TrainJob struct {
	Spec   worker.TrainSpec `json:"spec"`
	GoLive GoLive           `json:"go_live"`
}

TrainJob is what a training job runs: the spec and how its version goes live, the body of POST .../train.

type TrainResult 

type TrainResult struct {
	Model   storage.MLModel `json:"model"`
	Version worker.Version  `json:"version"`
	Live    bool            `json:"live"`
	Reason  string          `json:"reason"`
}

TrainResult is a finished training: the version made, and whether it is now the one the model serves.

type WorkerConnection added in v1.18.1 

type WorkerConnection struct {
	URL   string
	Token string
}

WorkerConnection is the ML worker connected from the console. A zero value means none is saved.

func LoadWorkerConnection added in v1.18.1 

func LoadWorkerConnection(ctx context.Context, store any) (WorkerConnection, error)

LoadWorkerConnection reads the saved worker, with its token decrypted. A token that no longer decrypts (the master key was changed without re-encrypting) reads as empty, and is logged by configsecrets.

func SaveWorkerConnection added in v1.18.1 

func SaveWorkerConnection(ctx context.Context, store any, p *worker.Provider, rawURL string, token *string) (WorkerConnection, error)

SaveWorkerConnection saves the worker at rawURL and puts it in force in p at once. token nil keeps the saved token, so the address can change without typing the token again; an empty token means the worker takes none. An empty address disconnects the saved worker. It is refused while the environment names a worker.

Source Files

View all Source files

Directories

Path Synopsis
Package monitor watches what a vhost's models are asked and answer.
 Package monitor watches what a vhost's models are asked and answer.
transport
grpc
Package grpc serves hermod.ml.v1 InferenceService: the gRPC twin of POST /api/ml/serve/{vhost}/{name}.
 Package grpc serves hermod.ml.v1 InferenceService: the gRPC twin of POST /api/ml/serve/{vhost}/{name}.
http
Package http is the REST surface of the model registry: managing a vhost's models, calling one from the UI, and the serving endpoint applications call with a model's serving key.
 Package http is the REST surface of the model registry: managing a vhost's models, calling one from the UI, and the serving endpoint applications call with a model's serving key.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.