diff --git a/BACKLOG.md b/BACKLOG.md index ebacd4e..3a17b73 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -1728,7 +1728,10 @@ templated HTTP call with an expression for "block". what it says to, and a service that does not answer is treated as `fail` says. -### L4. Routing by latency and by cost, and failover by priority (M) +### L4. Routing by latency and by cost, and failover by priority (M) — latency done 2026-10-25 +`endpoint.Registry.ObserveTTFT`, `scheduler.Latency()` in `chat`. Failover in tiers is +what `fallbacks` already does (local, then an outside model when no room). Left: +routing by cost between several outside providers. The original item follows. The scorers (`internal/router/scheduler/scorers.go`) know sessions, queues and KV cache. Kong also balances by measured latency, by tokens or cost used, and in priority tiers. diff --git a/CHANGELOG.md b/CHANGELOG.md index 7287bdc..4cf3dab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,11 @@ APIs change between minors without a deprecation period. ### Added +- **Chat goes to the replica answering soonest.** The router keeps each + replica's time to first token, smoothed over its last answers, and the + `chat` profile scores by it beside the queue and the KV cache: of two + replicas of one model, the one on the slower card or the busier host gets + less, and a replica not yet measured is tried. - **llmgw and Kong, written down** ([docs/kong.md](docs/kong.md)): Kong 3.9 in front of llmgw by a plain route or its `ai-proxy`, each team's llmgw key passed on; Kong as a provider behind llmgw; and that Kong 3.9 counts diff --git a/README.md b/README.md index 872f02b..338139d 100644 --- a/README.md +++ b/README.md @@ -313,8 +313,9 @@ within 1.24x of idle while 500 concurrent batch requests run, against 58x with admission off. Routing profiles select how endpoints are scored: `chat` favours session -affinity, `batch` favours the shortest queue, `long-context` favours free KV -cache. Pick one per model in the config, or per request with the +affinity and the replica that has been answering soonest (its time to first +token, smoothed), `batch` favours the shortest queue, `long-context` favours +free KV cache. Pick one per model in the config, or per request with the `x-llmgw-profile` header. ## sim-engine diff --git a/internal/router/endpoint/endpoint.go b/internal/router/endpoint/endpoint.go index 000d4f2..4057033 100644 --- a/internal/router/endpoint/endpoint.go +++ b/internal/router/endpoint/endpoint.go @@ -117,6 +117,10 @@ type Snapshot struct { // that a replica is slow. DrainRate float64 + // TTFT is the time to first token of the streamed answers this endpoint + // gave lately, exponentially smoothed; zero until it has given one. + TTFT time.Duration + // Inflight counts requests the router has dispatched here that have not // finished. It compensates for scrape lag (§7.4); without it, herding is // unavoidable within a 2s scrape interval. @@ -165,6 +169,7 @@ type live struct { lastSucc float64 drainRate float64 maxRunning int + ttft float64 // seconds, smoothed inflight int brk *breaker.Breaker @@ -247,6 +252,25 @@ func (r *Registry) Infos() []Info { return out } +// ObserveTTFT records how long an answer streamed here took to its first token. +func (r *Registry) ObserveTTFT(id string, d time.Duration) { + if d <= 0 { + return + } + r.mu.Lock() + defer r.mu.Unlock() + l, ok := r.m[id] + if !ok { + return + } + const alpha = 0.2 // a few answers move it; one slow one does not + if l.ttft == 0 { + l.ttft = d.Seconds() + } else { + l.ttft = alpha*d.Seconds() + (1-alpha)*l.ttft + } +} + // Observe records a successful scrape and updates the drain rate. func (r *Registry) Observe(id string, s Stats) { r.mu.Lock() @@ -296,6 +320,7 @@ func (r *Registry) Snapshots(model string) []Snapshot { Info: l.info, Stats: l.stats, DrainRate: l.drainRate, + TTFT: time.Duration(l.ttft * float64(time.Second)), Inflight: l.inflight, RunningCapacity: l.maxRunning, Stale: stale, diff --git a/internal/router/endpoint/endpoint_test.go b/internal/router/endpoint/endpoint_test.go index 39542a6..999ecbe 100644 --- a/internal/router/endpoint/endpoint_test.go +++ b/internal/router/endpoint/endpoint_test.go @@ -307,3 +307,15 @@ func TestLastUsedSurvivesAnEndpointListRefresh(t *testing.T) { t.Errorf("last used = %v (present=%v) after a republish, want %v", got, ok, before) } } + +func TestTTFTIsSmoothedPerEndpoint(t *testing.T) { + r := NewRegistry(Config{}) + r.Replace([]Info{{ID: "a", Models: []string{"m"}}}) + r.ObserveTTFT("a", time.Second) + r.ObserveTTFT("a", 2*time.Second) + r.ObserveTTFT("nobody", time.Second) + s := r.Snapshots("m") + if len(s) != 1 || s[0].TTFT != 1200*time.Millisecond { + t.Errorf("%v", s) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 92ea837..1a8ad4b 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -1261,6 +1261,9 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa r.keyRefused(model, dec.Endpoint.ID) } releaseAdapter() + if perr == nil && out.Status == http.StatusOK { + r.reg.ObserveTTFT(dec.Endpoint.ID, out.TTFT) + } uspan.SetAttributes(attribute.Int("http.response.status_code", out.Status)) uspan.SetAttributes(genAIUsage(req.Path, model, out.Usage)...) endSpan(uspan, perr) diff --git a/internal/router/scheduler/latency_test.go b/internal/router/scheduler/latency_test.go new file mode 100644 index 0000000..1123825 --- /dev/null +++ b/internal/router/scheduler/latency_test.go @@ -0,0 +1,20 @@ +package scheduler + +import ( + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint" +) + +func TestLatencyPrefersTheEndpointThatAnswersSooner(t *testing.T) { + eps := []endpoint.Snapshot{ + {TTFT: 200 * time.Millisecond}, + {TTFT: 400 * time.Millisecond}, + {}, // not yet measured + } + got := Latency().Score(nil, eps) + if got[0] != 1 || got[1] != 0.5 || got[2] != 1 { + t.Errorf("scores %v, want [1 0.5 1]", got) + } +} diff --git a/internal/router/scheduler/profiles.go b/internal/router/scheduler/profiles.go index 6a6c342..f98f664 100644 --- a/internal/router/scheduler/profiles.go +++ b/internal/router/scheduler/profiles.go @@ -56,6 +56,7 @@ func Chat(c ProfileConfig) *Profile { {Queue(c.TargetWaiting), 1.0}, {KV(c.KVCeiling), 1.0}, {Inflight(c.TargetInflight), 1.0}, + {Latency(), 1.0}, {Weight(), 0.5}, }, 2.0), Picker: P2C(), diff --git a/internal/router/scheduler/scorers.go b/internal/router/scheduler/scorers.go index ab4471d..dd41c84 100644 --- a/internal/router/scheduler/scorers.go +++ b/internal/router/scheduler/scorers.go @@ -2,6 +2,7 @@ package scheduler import ( "math" + "time" "gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint" ) @@ -178,3 +179,35 @@ func (s sessionScorer) Score(c *Ctx, eps []endpoint.Snapshot) []float64 { } return out } + +// Latency scores by time to first token as this router has seen it: the +// fastest endpoint 1, one twice as slow 0.5. Kong and others balance by it; +// for chat it is what a person waits for. An endpoint not yet measured +// scores 1, so a new replica is tried rather than avoided. +// +// Beside Load, not instead: Load sees a replica draining slowly from its +// own counters, before any answer of this router's has come back from it; +// this sees what answers actually took, a slow card or a busy host included. +func Latency() Scorer { return latencyScorer{} } + +type latencyScorer struct{} + +func (latencyScorer) Name() string { return "latency" } + +func (latencyScorer) Score(_ *Ctx, eps []endpoint.Snapshot) []float64 { + best := time.Duration(0) + for _, e := range eps { + if e.TTFT > 0 && (best == 0 || e.TTFT < best) { + best = e.TTFT + } + } + out := make([]float64, len(eps)) + for i, e := range eps { + if e.TTFT <= 0 || best == 0 { + out[i] = 1 + continue + } + out[i] = float64(best) / float64(e.TTFT) + } + return out +}