Files
LocalAI/core/http/middleware/route_model.go
Richard Palethorpe 49ef40a187 feat(classifier/VAD): support voice control on low power devices (#10804)
* feat(llama-cpp): route Score through the slot loop

Score previously bypassed the slot loop with a direct llama_decode: a
conflict guard aborted the whole process if scoring raced generation, the
config validator had to reject score alongside chat/completion/embeddings,
and every candidate re-decoded the full shared prompt.

Add SERVER_TASK_TYPE_SCORE to the (patched) upstream server so score tasks
are scheduled like any other slot work: generation and scoring serialize
naturally, the shared prompt is decoded once per call, and the slot's
prompt cache carries the conversation prefix across calls. Context
checkpoints at the score boundary and at the cache-divergence point keep
SWA/hybrid/recurrent models (e.g. LFM2.5) from re-prefilling the whole
prompt per candidate: warm-turn scoring on a 6-option set drops from ~8s
to ~0.5s on a desktop CPU.

The conflict guard and the validation split are removed; declaring score
with generation usecases on one config is now supported and shares the
slot cache.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* feat(realtime): classifier wire types and pipeline config

Wire types and YAML config for realtime classifier mode: sessions carry a
localai_classifier extension (options with canned replies/tool calls,
softmax threshold, normalization, history trimming, fallback modes, and a
deterministic wake-word address gate), mirrored by pipeline.classifier in
the model YAML and surfaced in the config-meta registry. The
localai.classifier.result server event reports the full score distribution
per turn.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* feat(realtime): classifier response flow

Classifier-mode responses: instead of autoregressive generation, each user
turn is prefill-scored against the option list (router.ScoreClassifier
prompt/candidate shapes over the Score primitive) and the winning option's
canned reply and tool call are emitted through the existing response
machinery. Below-threshold turns take the configured fallback (none /
canned reply / generate); empty transcripts and unaddressed turns (wake
word not mentioned) skip scoring entirely. The scoring probe defaults to
the latest user message only — small scorers echo canned replies from
prior turns back as the top option otherwise.

Built for hardware that can afford prompt processing but not decode: with
slot-based Score the option list stays KV-cached across turns, so a turn
costs roughly one forward pass over the new words.

session_update_error events now carry the validation cause instead of a
generic message.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* fix(realtime): bound the VAD tick's scan window and buffer retention

The VAD tick loop re-scanned the entire input buffer every 300ms and only
trimmed it on zero-segment ticks or commits. Audio that keeps producing
segments without a committing pause (steady noise a mic pipeline lets
through, music, continuous speech) grew the buffer toward the 100MB cap
with each tick rescanning all of it — O(n^2), measured at ~3.3ms of silero
per buffered second: past ~90s retained, ticks run back to back and pin
~4 cores until the stream stops.

Silero's recurrent state only carries a few hundred ms of context, so
rescanning old audio buys nothing. Clip the slice handed to the VAD to the
largest silence the commit test can need to measure (server_vad silence
window or the semantic eagerness fallback) plus a warm-up margin, and
rebase the returned segment times so every downstream consumer keeps
whole-buffer coordinates. An open turn whose clipped window is all silence
now commits (the silence outran the window) instead of being discarded as
no-speech. Independently, retain at most 90s of raw buffer, rebasing the
live-feed and EOU cursors on trim — this also bounds the previously
unbounded VAD-error path. Turn boundaries are otherwise unchanged: no
forced commits, no new coordinator states.

pipeline.turn_detection.vad_window_sec can widen the scan window; values
below the automatic floor are ignored. The tick body is extracted into
vadTick so specs can drive turn detection synchronously (same shape as
classifySoundWindow); the babble reproduction that pinned 4 cores now
plateaus under 10% of one core.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* fix(backend): let per-model threads override the global default

ModelOptions overrode a set per-model threads value with the app-level
--threads whenever the latter was non-zero — and WithThreads defaults it
to the physical core count, so it always was. The YAML threads: knob has
been dead config: a tiny VAD model could never opt down from the global
pool size.

SetDefaults already fills an unset per-model value from the app config,
which is the intended precedence; resolve threads through a helper that
honors it (explicit threads: 0 still means unset).

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* chore(gallery): single-thread the silero VAD

Silero is a ~2MB recurrent model with no exploitable graph parallelism:
measured per-call latency is identical at 1 and 10 ORT threads, while
every extra pool thread just spin-waits between the realtime loop's
frequent tiny inferences.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* docs(realtime): classifier mode, VAD scan window, threads precedence

Document the realtime classifier mode (options, threshold guidance,
wake-word address gate, empty-transcript handling), the VAD scan window
and 90s buffer retention (pipeline.turn_detection.vad_window_sec), the
per-model threads precedence, and the M3 classifier note in the realtime
state-machine design doc.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* perf(llama-cpp): score all candidates in one batched decode

One scoring call is now a single SERVER_TASK_TYPE_SCORE task: the slot
decodes the shared prefix (prompt + longest common candidate token
prefix) once, then forks one sequence per candidate off it
(metadata-only for the unified KV cache, copy-on-write for recurrent
state) and decodes every candidate's unique tail in one llama_decode.
Previously each candidate was its own task that restored the boundary
checkpoint and re-decoded its full tail sequentially, paying
per-candidate task and decode overhead.

The context reserves SERVER_SCORE_FORK_SEQS extra sequence ids (and
recurrent-state cells) beyond the parallel slots via the new
common_params::n_seq_score_forks. Forking requires the unified KV cache
(already this backend's default) since per-sequence streams would shrink
n_ctx_seq; an explicit kv_unified:false disables forking and Score calls
that need it fail cleanly. Candidates beyond the fork/output budget
decode in successive chunks.

Wire contract and scores are unchanged: per-token logprobs are stitched
from the shared region and the forked tails. Verified bitwise
deterministic call-to-call and independent of candidate order (no
cross-fork leakage via equal-length candidate swap); ranking matches the
per-candidate implementation on the drone battery (winner softmax
0.99996 vs 0.99997), and >16-candidate chunking, prefix-of-another and
empty candidates all pass.

Measured on a desktop CPU: warm /api/score calls 0.52s -> 0.23s; warm
realtime classifier turns 196-303ms. The 9-candidate drone turn decodes
~17 unique tail tokens in one batch instead of nine sequential ~220ms
checkpoint-restore tasks.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* fix(realtime): gate scoring capacity by model usecase

Reserve llama.cpp scoring slots only for models that explicitly declare the score usecase, while allowing score to coexist with chat and completion. Reject incompatible unified-KV settings and classifier activation on models without scoring capacity.

Propagate application defaults when resolving realtime and preload pipeline stages so unset thread counts are resolved consistently without overriding explicit model settings.

Assisted-by: Codex:gpt-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* fix(ci): honor APT mirrors in the prebuilt llama-cpp compile step

The builder-prebuilt path installs gcc-14 with apt directly and ignored
the APT_MIRROR/APT_PORTS_MIRROR build args the from-source path already
honors, so an ubuntu mirror outage broke every arm64 backend build. Pass
the args into the stage and run apt-mirror.sh (already in the build
context via COPY . /LocalAI) before the apt step.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* feat(realtime): classifier argument slots via constrained completion

Hybrid classify-then-complete: a classifier option's canned tool call can
declare typed argument slots (number | enum | string, with defaults and
prompt hints) referenced as "{{name}}" in the arguments template. When
the option wins, the slots are filled by a short grammar-constrained
completion that continues the exact scoring prompt — rendered by the same
cached ScoreClassifier, so the llama.cpp prompt cache is already warm —
with the chosen route JSON re-opened at the first slot field. A GBNF
grammar pins the field skeleton and frees only the values; temperature 0,
a couple dozen tokens at most (~300ms on a desktop CPU for two slots).

Slot declarations and hints ride the option descriptions in the shared
system prompt, informing scoring and the fill alike at no per-turn token
cost. The localai.classifier.result event carries the final arguments and
a fill_latency_ms. On inference failure the slots' defaults apply; a slot
without a default fails the response (or falls through with
fallback.mode: generate). Slot filling requires completion alongside
score in the scoring model's known_usecases.

Verified end-to-end on the Pi drone demo: "fly forward three meters" in
distance mode classifies forward and infers {"distance": 3, "units":
"meters"} in ~310ms, and the drone flies exactly 3 units.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* feat(realtime): splice filled slot values into classifier replies

A classifier option's spoken reply can now reference its tool's argument
slots ("Going forward {{distance}} {{units}}."): the values inferred by
the slot-fill completion — or the recovery defaults — are spliced into
the reply as plain text before it is emitted, so what the assistant says
confirms what it actually inferred. Placeholders without a value stay
literal, and options without slots are untouched.

FillToolArguments now returns the raw slot values alongside the spliced
arguments JSON to make the reply templating possible.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* fix(realtime): harden classifier slot completion

Reserve context for constrained slot filling, size completions from their encoded output, and encode enum grammar literals as valid JSON. Reject empty enum values and cover the failure modes with regression tests.

Assisted-by: Codex:gpt-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* feat(realtime): prewarm the classifier scoring prompt on registration

Swapping a session's classifier option list (a voice-switched command
mode, for instance) made the next turns pay a full re-prefill of the new
option-list prompt — measured 2.4s vs 0.3s warm on a desktop CPU, and
worse: on hybrid-memory models like LFM2.5, whose state cannot be
partially rewound (llama.cpp can only restore checkpoints), *every*
probe change re-prefilled from scratch whenever the last checkpoint
missed the probe boundary, so even same-list turns intermittently cost
full prefills.

Registering an option list (pipeline seed or session.update) now fires a
best-effort background prewarm: two throwaway scores with distinct
probes. The first prefills the new option-list prompt; the second,
diverging exactly where per-turn probe text starts, plants the backend's
rewind point (KV checkpoint) at the stable-prefix boundary that every
real turn reuses. The prewarm hides behind the canned mode-switch reply
— by the time it finishes speaking, the cache is warm. Idempotent per
option set, detached from the registering request's lifetime.

Measured on the drone demo (LFM2.5-1.2B, desktop CPU): first turn after
a mode switch 2374ms -> 340ms; intermittent same-list full prefills
(1.3-2.1s) all -> under 0.5s. For clients that swap lists frequently,
options: [parallel:2] on the scoring model additionally keeps one slot
per list via prefix-similarity routing (+26MB RSS, unified KV).

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* perf(llama-cpp): checkpoint scoring at the caller-declared stable prefix

Hybrid-memory models (LFM2.5 shortconv, Qwen3.5 deltanet — where new
small models are headed) cannot rewind their state, so any prompt-cache
reuse that needs a rewind falls back to a full re-prefill. For classifier
scoring that meant every probe change re-processed the whole option-list
prompt: the server's checkpoints were placed reactively (at wherever the
previous task happened to diverge), so a checkpoint past the next
divergence was erased rather than restored — measured as intermittent
2-10s turns on prompts with a 95%+ common prefix.

The classifier now computes the probe-invariant prompt prefix once (the
byte-wise common prefix of two synthetic probe renders) and declares its
length with every Score request; the server maps it to a token boundary
and forces a KV checkpoint exactly there on each score prefill. That
checkpoint sits at or before every future divergence under the same
option list, so it always survives and always restores — repeat scoring
costs probe+candidates regardless of how the probe changes.

Also:
- prewarm reruns on every option-list registration instead of memoizing
  per list: with boundary checkpoints a redundant rewarm costs two
  probe-sized decodes, while skipping one after a slot eviction (three
  lists sharing fewer slots evict in LRU cascades) silently moves a full
  re-prefill onto the user's next turn
- new llama.cpp backend option rs_seq:N exposes bounded recurrent-state
  rollback outside speculative decoding; measured impractical for
  deltanet-scale states (65GB for 64 snapshots on Qwen3.5-4B) but cheap
  insurance for small-state models
- docs: the multi-list recipe (parallel:N + sps:0.5 — the default slot
  similarity threshold funnels distinct lists onto one slot)

Measured on the drone demo (LFM2.5-1.2B scorer, desktop CPU), steady
state: every turn 285-421ms including mode switches, vs 2.4s post-switch
and intermittent 1.3-2.9s re-prefills before.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* fix(realtime): align classifier cache guidance

Document the single-score prewarm behavior and clean the vendored score patch formatting.

Assisted-by: Codex:gpt-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* fix(llama-cpp): guard score task for fork backends

TurboQuant and Bonsai reuse the primary gRPC server against llama.cpp forks that do not carry LocalAI's slot-based Score patches. Compile the Score integration only for the patched primary backend and return UNIMPLEMENTED from fork builds instead of referencing absent task types and common_params fields.

Assisted-by: Codex:gpt-5 [gh]
Signed-off-by: Richard Palethorpe <io@richiejp.com>

* fix(dev): generate gRPC code before commit lint

The coverage phase regenerates ignored protobuf bindings, but lint runs first and can fail against missing or stale output. Generate the pinned bindings before lint so the gate always type-checks the current schema.

Assisted-by: Codex:gpt-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>

---------

Signed-off-by: Richard Palethorpe <io@richiejp.com>
2026-07-29 12:50:22 +02:00

682 lines
27 KiB
Go

package middleware
import (
"context"
"crypto/rand"
"encoding/hex"
"fmt"
"hash/fnv"
"strconv"
"strings"
"time"
"github.com/labstack/echo/v4"
"github.com/mudler/LocalAI/core/backend"
"github.com/mudler/LocalAI/core/config"
"github.com/mudler/LocalAI/core/http/auth"
"github.com/mudler/LocalAI/core/schema"
"github.com/mudler/LocalAI/core/services/routing/router"
"github.com/mudler/LocalAI/core/templates"
"github.com/mudler/xlog"
"gopkg.in/yaml.v3"
)
// ScorerFactory returns a backend.Scorer bound to a named classifier
// model. The score classifier uses it to compute joint log-prob of
// every policy label against the routing prompt.
type ScorerFactory func(modelName string) backend.Scorer
// EmbedderFactory returns a backend.Embedder bound to a named model.
// Used by the L2 embedding cache. Returning nil signals "model not
// loadable" — the middleware then falls back to the uncached
// classifier so routing still happens.
type EmbedderFactory func(modelName string) backend.Embedder
// VectorStoreFactory returns a backend.VectorStore bound to a named
// collection. Each router model's cache lives in its own collection
// so two routers can't poison each other's hits.
type VectorStoreFactory func(storeName string) backend.VectorStore
// RerankerFactory returns a backend.Reranker bound to a named model.
// Used by the colbert classifier to score policy descriptions against
// the prompt via LocalAI's rerankers backend. Returning nil signals
// "model not loadable" — buildClassifier reports a config error.
type RerankerFactory func(modelName string) backend.Reranker
// ModelConfigLookup resolves a model name to its config, or nil when
// unknown. Used by buildClassifier to confirm the classifier_model
// declared the score usecase — the actual usecase-conflict check
// lives in ModelConfig.Validate() and runs at config load/save time.
type ModelConfigLookup func(modelName string) *config.ModelConfig
// ClassifierDeps bundles the backend factories the router middleware
// needs to build a classifier and its optional L2 cache. Bundled into
// one struct because RouteModel already takes many positional
// arguments — additions to the dependency surface go here instead of
// growing the signature.
//
// Embedder and VectorStore are optional: when both are non-nil and the
// router config declares an embedding_cache block, the score
// classifier is wrapped in EmbeddingCacheClassifier. Otherwise the
// score classifier runs unwrapped and the embedding-cache YAML is
// ignored with a warning.
type ClassifierDeps struct {
Scorer ScorerFactory
Embedder EmbedderFactory
VectorStore VectorStoreFactory
Reranker RerankerFactory
// ModelLookup resolves the classifier_model name to its config so
// buildClassifier can reject misconfigurations that would
// otherwise crash the llama-cpp backend at request time. Optional
// — when nil, the check is skipped (tests, embedded callers that
// haven't wired the loader).
ModelLookup ModelConfigLookup
// Registry is the shared classifier cache. Both the OpenAI and
// Anthropic routes pass the same registry so the admin stats
// endpoint sees every live classifier. Nil falls back to a local
// registry — tests that don't need cross-route stats use this.
Registry *router.Registry
// Evaluator renders the classifier model's chat template around
// the routing system + user prompt. Optional — when nil, the
// score classifier falls back to a built-in ChatML envelope,
// which is correct for Arch-Router/Qwen but wrong for non-ChatML
// routing models. Production wiring passes the app-wide
// templates.Evaluator so any model the operator points at gets
// its own chat template applied.
Evaluator *templates.Evaluator
// TokenCounter binds the classifier model's tokenizer for the score
// classifier's token-trim path. Optional; nil falls back to the
// backend's n_ctx guard. Plain func type so core/application supplies
// it as a method value without importing this package.
TokenCounter func(modelName string) func(text string) (int, error)
}
// ProbeExtractor pulls the prompt content out of a parsed request so
// the classifier can inspect it without taking a dependency on the
// schema package. One extractor per request shape — wired by the
// route registration site (mirrors the piiadapter pattern).
//
// Returns ok=false when the parsed value isn't the expected type — the
// middleware then passes through without engaging the router.
type ProbeExtractor func(parsed any) (router.Probe, bool)
// RouteModel runs after SetModelAndConfig and the schema-specific
// SetXRequest, looks at the resolved model's Router config, and (when
// present) reclassifies the request to one of the candidates.
//
// The middleware:
//
// 1. Loads MODEL_CONFIG from the echo context. If nil or HasRouter()
// is false, passes through.
// 2. Extracts the probe via the supplied ProbeExtractor.
// 3. Invokes the classifier matching cfg.Router.Classifier
// ("score" or "colbert"). If the classifier can't be built —
// missing classifier_model, misconfigured policies, etc. — the
// request fails with 503. cfg.Router.Fallback only catches
// Classify-time errors and label-coverage misses, not config
// bugs that would otherwise be silent.
// 4. Resolves the chosen candidate to its model name. Reloads the
// ModelConfig for that model and asserts depth-1 (the candidate
// must NOT itself have a Router). Violation returns 500 — config
// bug, not a request bug.
// 5. Updates input.Model in place, replaces MODEL_CONFIG with the
// candidate's config, and stamps RequestedModel/ServedModel on the
// context so UsageMiddleware records the routing.
// 6. Writes a DecisionRecord to the store for the admin page.
//
// store may be nil when --disable-stats turns off the routing log;
// classification still runs.
//
// Composition with SmartRouter (distributed mode): this middleware
// only does *model* selection. Node selection still happens in
// SmartRouter.Route() downstream of this middleware.
// RouteModel wires the router middleware. source is the value written to
// DecisionRecord.Source (router.SourceChat / SourceAnthropic / ...) so
// the admin page can split decisions by entry point. Pass
// router.SourceChat for the OpenAI chat endpoint, router.SourceAnthropic
// for the Anthropic messages endpoint.
func RouteModel(loader *config.ModelConfigLoader, appConfig *config.ApplicationConfig, store router.DecisionStore, fallbackUser *auth.User, extractor ProbeExtractor, source string, deps ClassifierDeps) echo.MiddlewareFunc {
registry := deps.Registry
if registry == nil {
registry = router.NewRegistry()
}
candidateLoader := func(name string) (*config.ModelConfig, error) {
return loader.LoadModelConfigFileByNameDefaultOptions(name, appConfig)
}
return func(next echo.HandlerFunc) echo.HandlerFunc {
return func(c echo.Context) error {
cfg, ok := c.Get(CONTEXT_LOCALS_KEY_MODEL_CONFIG).(*config.ModelConfig)
if !ok || cfg == nil || !cfg.HasRouter() {
return next(c)
}
parsed := c.Get(CONTEXT_LOCALS_KEY_LOCALAI_REQUEST)
if parsed == nil {
return next(c)
}
probe, probeOK := extractor(parsed)
if !probeOK {
return next(c)
}
classifier, err := GetOrBuildClassifier(registry, cfg, deps)
if err != nil {
// Build-time failures are config bugs (missing
// classifier_model, undeclared usecase, policy
// validation, ...). Silently falling back would hide
// them and make the router look "working" while the
// classifier model is never invoked — surface as 503
// with the underlying reason so operators see it.
xlog.Warn("router: classifier build failed",
"router_model", cfg.Name, "classifier", cfg.Router.Classifier, "error", err)
return echo.NewHTTPError(503, "router classifier unavailable: "+err.Error())
}
result, err := router.Resolve(c.Request().Context(), cfg, classifier, candidateLoader, probe)
if err != nil {
xlog.Warn("router: resolve failed", "router_model", cfg.Name, "error", err)
return echo.NewHTTPError(500, err.Error())
}
if req, ok := parsed.(schema.LocalAIRequest); ok {
chosen := result.ChosenModel
req.ModelName(&chosen)
}
c.Set(CONTEXT_LOCALS_KEY_MODEL_CONFIG, result.ChosenConfig)
// Preserve an upstream requested model (e.g. an alias that points
// at this router model) so accounting keeps the name the client
// actually sent. Served always reflects the final candidate.
if c.Get(ContextKeyRequestedModel) == nil {
c.Set(ContextKeyRequestedModel, result.RouterModel)
}
c.Set(ContextKeyServedModel, result.ChosenModel)
if store != nil {
recordHTTPDecision(c, store, result, fallbackUser, source)
}
return next(c)
}
}
}
// recordHTTPDecision writes the resolved decision to the store with
// HTTP-shaped audit metadata (correlation id from header, user from
// auth middleware, fallback to the synthetic local user). Realtime
// has its own recorder that supplies session-derived metadata
// instead.
func recordHTTPDecision(c echo.Context, store router.DecisionStore, result *router.ResolveResult, fallbackUser *auth.User, source string) {
correlationID, _ := c.Get(ContextKeyCorrelationID).(string)
if correlationID == "" {
correlationID = c.Response().Header().Get("X-Correlation-ID")
}
userID := ""
if u := auth.GetUser(c); u != nil {
userID = u.ID
} else if fallbackUser != nil {
userID = fallbackUser.ID
}
_ = store.Record(context.Background(), result.ToDecisionRecord(newDecisionID(), correlationID, userID, source))
}
// GetOrBuildClassifier looks up a built Classifier for the named router
// model in the registry and builds it on miss. Exported so the
// /api/router/decide decision-oracle endpoint can share the same
// build-once cache that the in-band RouteModel middleware uses.
func GetOrBuildClassifier(registry *router.Registry, cfg *config.ModelConfig, deps ClassifierDeps) (router.Classifier, error) {
// Fingerprint folds the classifier model's renderer-affecting
// fields (chat templates + stopwords) in alongside the router
// config. Without this, hot-reloading the classifier model's
// YAML (via ReloadModelsEndpoint, /import-model, or the MCP
// reload_models tool) wouldn't rebuild the cached classifier —
// the candidates slice and renderer closure are baked at build
// time from those fields and would silently keep the stale
// stop token / template until process restart.
var classifierCfg *config.ModelConfig
if deps.ModelLookup != nil {
classifierCfg = deps.ModelLookup(cfg.Router.ClassifierModel)
}
fp := routerConfigFingerprint(cfg.Router, classifierCfg)
if cached, ok := registry.Get(cfg.Name, fp); ok {
return cached, nil
}
c, err := buildClassifier(cfg, deps)
if err != nil {
return nil, err
}
registry.Put(cfg.Name, fp, c)
return c, nil
}
// routerConfigFingerprint is a stable cache key for the (router cfg,
// classifier model cfg) tuple. FNV-64 over the YAML form of the
// router block plus the renderer-affecting fields of the classifier
// model — equality-only, not cryptographic. YAML-marshal picks up
// any future RouterConfig field without this function needing to be
// touched; for the classifier model we hash a narrow projection so
// unrelated changes (parameters, files, ...) don't burst the cache.
// Pass classifierCfg=nil when no lookup is wired — the fingerprint
// degenerates to the router-only form, matching pre-refactor behaviour.
func routerConfigFingerprint(rc config.RouterConfig, classifierCfg *config.ModelConfig) uint64 {
bytes, err := yaml.Marshal(rc)
if err != nil {
// Marshalling a value type can't fail in practice; fall
// back to a hash that varies per call so we don't quietly
// share a cache entry across distinct configs.
return uint64(time.Now().UnixNano())
}
h := fnv.New64a()
h.Write(bytes)
if classifierCfg != nil {
// Narrow projection: only the fields buildClassifier reads (renderer,
// stop tokens, context_size → MaxContextTokens). Hashing the whole
// ModelConfig would invalidate the cache on irrelevant changes;
// omitting context_size would let a reload leave a stale token budget.
h.Write([]byte{0}) // separator so empty fields don't collide
h.Write([]byte(classifierCfg.TemplateConfig.Chat))
h.Write([]byte{0})
h.Write([]byte(classifierCfg.TemplateConfig.ChatMessage))
h.Write([]byte{0})
for _, sw := range classifierCfg.StopWords {
h.Write([]byte(sw))
h.Write([]byte{0})
}
h.Write([]byte{0})
if classifierCfg.ContextSize != nil {
h.Write([]byte(strconv.Itoa(*classifierCfg.ContextSize)))
}
}
return h.Sum64()
}
func buildClassifier(cfg *config.ModelConfig, deps ClassifierDeps) (router.Classifier, error) {
rc := cfg.Router
name := rc.Classifier
if name == "" {
name = router.ClassifierScore
}
policies, err := validateRouterPolicies(name, rc)
if err != nil {
return nil, err
}
cacheCap := rc.ClassifierCacheSize
if cacheCap == 0 {
cacheCap = 1024
}
var inner router.Classifier
switch name {
case router.ClassifierScore:
if deps.Scorer == nil {
return nil, fmt.Errorf("router classifier score unavailable: no scorer factory wired")
}
if err := assertClassifierDeclaresScore(rc.ClassifierModel, deps.ModelLookup); err != nil {
return nil, err
}
scorer := deps.Scorer(rc.ClassifierModel)
if scorer == nil {
return nil, fmt.Errorf("router classifier score: classifier_model %q not loadable", rc.ClassifierModel)
}
opts := router.ScoreClassifierOptions{
CacheCap: cacheCap,
ActivationThreshold: rc.ActivationThreshold,
Normalization: rc.ScoreNormalization,
SystemPromptTemplate: rc.ClassifierSystemTemplate,
}
// Build the prompt renderer + stop token from the classifier
// model's own config when available. Without ModelLookup
// (tests, embedded callers) the score classifier's built-in
// ChatML defaults kick in, which is correct for Arch-Router.
if deps.ModelLookup != nil {
if classifierCfg := deps.ModelLookup(rc.ClassifierModel); classifierCfg != nil {
if deps.Evaluator != nil {
// The router renders the scoring prompt client-side, so the
// classifier model MUST carry a chat template — refusing
// here beats silently falling back to a generic ChatML
// envelope the model may not have been trained on.
renderer := NewTemplateRenderer(deps.Evaluator, classifierCfg)
if renderer == nil {
return nil, fmt.Errorf(
"router classifier score: classifier_model %q has no chat template "+
"(set template.chat and template.chat_message in its config). The router "+
"renders the scoring prompt with the classifier model's own template; "+
"without it the prompt format would not match the model",
rc.ClassifierModel)
}
opts.PromptRenderer = renderer
}
if st := PickAssistantTurnEnd(classifierCfg.StopWords, classifierCfg.TemplateConfig.ChatMessage); st != "" {
opts.StopToken = st
}
// Token-exact conversation trim — score classifier drops the
// oldest turns using the model's own tokenizer.
if count, ctxTokens := modelTokenTrim(rc.ClassifierModel, deps); count != nil {
opts.TokenCounter = count
opts.MaxContextTokens = ctxTokens
}
}
}
inner = router.NewScoreClassifier(policies, scorer, opts)
case router.ClassifierColbert:
if deps.Reranker == nil {
return nil, fmt.Errorf("router classifier colbert unavailable: no reranker factory wired")
}
reranker := deps.Reranker(rc.ClassifierModel)
if reranker == nil {
return nil, fmt.Errorf("router classifier colbert: classifier_model %q not loadable", rc.ClassifierModel)
}
rerankClassifier := router.NewRerankClassifier(policies, reranker, cacheCap, rc.ActivationThreshold)
if count, ctxTokens := modelTokenTrim(rc.ClassifierModel, deps); count != nil {
rerankClassifier = rerankClassifier.WithTokenTrim(count, ctxTokens)
}
inner = rerankClassifier
default:
return nil, fmt.Errorf("router: unknown classifier %q (supported: %s)", name, strings.Join([]string{router.ClassifierScore, router.ClassifierColbert}, ", "))
}
if rc.EmbeddingCache == nil {
return inner, nil
}
wrapped, err := wrapWithEmbeddingCache(cfg, inner, deps)
if err != nil {
// Caching plumbing problems must not break routing — log,
// drop the cache layer, and return the uncached classifier.
// The admin UI surfaces the warning via the classifier-build
// error path used elsewhere.
xlog.Warn("router: embedding cache disabled",
"router_model", cfg.Name, "error", err)
return inner, nil
}
return wrapped, nil
}
// assertClassifierDeclaresScore refuses to build the score classifier
// unless classifier_model's config declares FLAG_SCORE. This check only
// refuses to bind a model that never declared itself for Score in the
// first place; that model could be a misconfigured chat model the
// operator pointed at by accident.
//
// When lookup is nil (test wiring) the check is skipped.
func assertClassifierDeclaresScore(classifierModel string, lookup ModelConfigLookup) error {
if lookup == nil {
return nil
}
cfg := lookup(classifierModel)
if cfg == nil {
// Unknown model — Scorer() will produce a clearer "not
// loadable" error a few lines down.
return nil
}
if !cfg.HasUsecases(config.FLAG_SCORE) {
return fmt.Errorf(
"router classifier score: classifier_model %q does not declare the "+
"score usecase. Add `known_usecases: [score]` (alongside any other "+
"usecases the model serves) to its config",
classifierModel)
}
return nil
}
// validateRouterPolicies checks the shared invariants both classifiers
// rely on (non-empty policies, every candidate label declared as a
// policy, every candidate has a model + at least one label) and
// returns the parsed []ScorePolicy. Both Score and Rerank classifiers
// take the same policy shape.
func validateRouterPolicies(classifierName string, rc config.RouterConfig) ([]router.ScorePolicy, error) {
if rc.ClassifierModel == "" {
return nil, fmt.Errorf("router classifier %s requires classifier_model", classifierName)
}
if len(rc.Policies) == 0 {
return nil, fmt.Errorf("router classifier %s requires at least one policy", classifierName)
}
policies := make([]router.ScorePolicy, 0, len(rc.Policies))
for _, p := range rc.Policies {
if p.Label == "" {
return nil, fmt.Errorf("router classifier %s: policy with empty label", classifierName)
}
if p.Description == "" {
return nil, fmt.Errorf("router classifier %s: policy %q has no description", classifierName, p.Label)
}
policies = append(policies, router.ScorePolicy{Label: p.Label, Description: p.Description})
}
policyLabels := make(map[string]struct{}, len(policies))
for _, p := range policies {
policyLabels[p.Label] = struct{}{}
}
for _, c := range rc.Candidates {
if c.Model == "" {
return nil, fmt.Errorf("router classifier %s: candidate has empty model field", classifierName)
}
if len(c.Labels) == 0 {
return nil, fmt.Errorf("router classifier %s: candidate %q has no labels", classifierName, c.Model)
}
for _, l := range c.Labels {
if _, ok := policyLabels[l]; !ok {
return nil, fmt.Errorf("router classifier %s: candidate %q references unknown label %q (not in policies)", classifierName, c.Model, l)
}
}
}
return policies, nil
}
// NewTemplateRenderer adapts the templates.Evaluator + the classifier
// model's config into the router.PromptRenderer callback. The
// resulting renderer pushes the routing system + user prompt through
// the classifier model's full chat-template pipeline — per-role
// formatting via TemplateConfig.ChatMessage, then the outer
// TemplateConfig.Chat — so non-ChatML routing models render
// correctly without router-package awareness of the template format.
//
// We must go through TemplateMessages, not EvaluateTemplateForPrompt
// directly: the gallery's outer Chat templates are uniformly
// `{{.Input -}}<|im_start|>assistant` (or the Llama-3 equivalent)
// and reference {{.Input}} only — never {{.SystemPrompt}}. Passing
// our routing system prompt through .SystemPrompt would silently
// drop it because Go text/template ignores unreferenced fields.
// TemplateMessages instead renders each role through ChatMessage and
// joins them into the .Input the outer template DOES read.
//
// Returns nil (forcing the score classifier's chatMLRenderer
// fallback) when either template piece is missing — partial
// templating would still drop content.
func NewTemplateRenderer(eval *templates.Evaluator, classifierCfg *config.ModelConfig) router.PromptRenderer {
if classifierCfg.TemplateConfig.Chat == "" || classifierCfg.TemplateConfig.ChatMessage == "" {
return nil
}
cfgCopy := *classifierCfg
return func(system, user string) (string, error) {
messages := []schema.Message{
{Role: "system", StringContent: system},
{Role: "user", StringContent: user},
}
rendered := eval.TemplateMessages(schema.OpenAIRequest{}, messages, &cfgCopy, nil, false)
if rendered == "" {
return "", fmt.Errorf("router: classifier %q chat template produced empty output", cfgCopy.Name)
}
return rendered, nil
}
}
// PickAssistantTurnEnd returns the classifier model's assistant
// turn-end token — the one to suffix candidates with so the model's
// "I'm done" signal folds into the per-candidate joint log-prob.
//
// Strategy: prefer the stopword that *literally appears* in the
// chat_message template, because that token is the assistant
// turn-end by construction. ChatML's chat_message ends with
// "<|im_end|>", Llama-3's ends with "<|eot_id|>", etc. — the
// template is the source of truth.
//
// Fallback: the first non-empty stopword. That's right for
// well-ordered configs (ChatML conventionally lists <|im_end|>
// first) but wrong for some gallery Llama-3 templates that defensively
// list <|im_end|> first even though the actual turn-end is <|eot_id|>.
// The template-scan above catches those.
//
// When no stopwords are configured at all, return "" — caller falls
// back to defaultStopToken (<|im_end|>) inside the score classifier.
func PickAssistantTurnEnd(words []string, chatMessageTemplate string) string {
if chatMessageTemplate != "" {
for _, w := range words {
if w != "" && strings.Contains(chatMessageTemplate, w) {
return w
}
}
}
for _, w := range words {
if w != "" {
return w
}
}
return ""
}
func wrapWithEmbeddingCache(cfg *config.ModelConfig, inner router.Classifier, deps ClassifierDeps) (router.Classifier, error) {
ec := cfg.Router.EmbeddingCache
if ec.EmbeddingModel == "" {
return nil, fmt.Errorf("embedding_cache requires embedding_model")
}
if deps.Embedder == nil || deps.VectorStore == nil {
return nil, fmt.Errorf("embedding cache factories not wired")
}
embedder := deps.Embedder(ec.EmbeddingModel)
if embedder == nil {
return nil, fmt.Errorf("embedding_model %q not loadable", ec.EmbeddingModel)
}
storeName := ec.StoreName
if storeName == "" {
storeName = "router-cache-" + cfg.Name
}
vstore := deps.VectorStore(storeName)
if vstore == nil {
return nil, fmt.Errorf("vector store %q not loadable", storeName)
}
cache := router.NewEmbeddingCacheClassifier(inner, embedder, vstore, ec.SimilarityThreshold, ec.ConfidenceThreshold)
// Trim the probe to the embedder model's own context (e.g. nomic-embed at
// 8k) rather than a fixed guess — otherwise the cache key is an embedding
// of a silently-truncated conversation.
if count, ctxTokens := modelTokenTrim(ec.EmbeddingModel, deps); count != nil {
cache = cache.WithTokenTrim(count, ctxTokens)
}
return cache, nil
}
// modelTokenTrim returns a model's own tokenizer and the token ceiling its
// probe must fit, or (nil, 0) when no tokenizer is available (only then can we
// not trim exactly). The ceiling is min(effective context, effective batch):
// score/embed/rerank all decode the whole prompt in one pass, so it must fit
// both the context window and a single batch. Using the backend's *effective*
// values — not the raw config fields — means trimming still works when
// context_size and batch are unset; otherwise a non-trivial prompt overflows
// the default window and every classification fails.
func modelTokenTrim(modelName string, deps ClassifierDeps) (func(string) (int, error), int) {
if deps.TokenCounter == nil || deps.ModelLookup == nil {
return nil, 0
}
cfg := deps.ModelLookup(modelName)
if cfg == nil {
return nil, 0
}
count := deps.TokenCounter(modelName)
if count == nil {
return nil, 0
}
ceiling := backend.EffectiveContextSize(*cfg)
if b := backend.EffectiveBatchSize(*cfg); b < ceiling {
ceiling = b
}
return count, ceiling
}
func newDecisionID() string {
var b [12]byte
_, _ = rand.Read(b[:])
return "rd_" + hex.EncodeToString(b[:])
}
// OpenAIProbe extracts a router.Probe from a parsed *schema.OpenAIRequest.
// Concatenates message contents (string-form or text blocks of the
// structured `[]any` content) so the classifier sees a single corpus
// for length and content-shape rules. Image blocks are skipped — a
// future multimodal classifier can take a different route.
func OpenAIProbe(parsed any) (router.Probe, bool) {
req, ok := parsed.(*schema.OpenAIRequest)
if !ok || req == nil {
return router.Probe{}, false
}
return OpenAIProbeFromRequest(req), true
}
// messageText flattens a chat message's Content to plain text: string content
// verbatim; []any structured content contributes only its "text" blocks.
func messageText(content any) string {
switch ct := content.(type) {
case string:
return ct
case []any:
var b strings.Builder
for _, block := range ct {
if bm, ok := block.(map[string]any); ok && bm["type"] == "text" {
if t, ok := bm["text"].(string); ok {
if b.Len() > 0 {
b.WriteByte('\n')
}
b.WriteString(t)
}
}
}
return b.String()
}
return ""
}
// messageProbeParts drops empty (e.g. image-only) messages so they don't
// consume budget or emit blank lines.
func messageProbeParts(texts []string) []string {
parts := make([]string, 0, len(texts))
for _, t := range texts {
if t != "" {
parts = append(parts, t)
}
}
return parts
}
// OpenAIProbeFromRequest is the typed counterpart of OpenAIProbe — same
// extraction logic, but takes the request struct directly. Realtime and
// other non-HTTP callers use it to feed a probe to router.Resolve
// without going through an echo.Context first.
func OpenAIProbeFromRequest(req *schema.OpenAIRequest) router.Probe {
if req == nil {
return router.Probe{}
}
texts := make([]string, len(req.Messages))
for i := range req.Messages {
texts[i] = messageText(req.Messages[i].Content)
}
parts := messageProbeParts(texts)
// Prompt carries the full conversation; each classifier trims it to its own
// model's context (see modelTokenTrim). Messages preserves the per-turn
// split the trimmer drops oldest-first.
return router.Probe{Prompt: router.JoinTurns(parts), Messages: parts}
}
// AnthropicProbe is the AnthropicRequest analogue of OpenAIProbe.
func AnthropicProbe(parsed any) (router.Probe, bool) {
req, ok := parsed.(*schema.AnthropicRequest)
if !ok || req == nil {
return router.Probe{}, false
}
texts := make([]string, len(req.Messages))
for i := range req.Messages {
texts[i] = messageText(req.Messages[i].Content)
}
parts := messageProbeParts(texts)
return router.Probe{Prompt: router.JoinTurns(parts), Messages: parts}, true
}