From 73e1bb67330a1129d473e734e200bc3d22be7330 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sun, 25 Oct 2026 17:01:30 +0100 Subject: [PATCH] Chat goes to the replica that has been answering soonest The router keeps each endpoint's time to first token, smoothed over its last streamed answers, and the chat profile scores by it beside session, queue, KV and inflight: the fastest endpoint 1, one twice as slow 0.5, one not yet measured 1 so a new replica is tried. Load still sees a slow drain from the engine's own counters; this sees what answers actually took. BACKLOG L4. Co-Authored-By: Claude Opus 5.5 --- BACKLOG.md | 5 +++- CHANGELOG.md | 5 ++++ README.md | 5 ++-- internal/router/endpoint/endpoint.go | 25 +++++++++++++++++ internal/router/endpoint/endpoint_test.go | 12 +++++++++ internal/router/router.go | 3 +++ internal/router/scheduler/latency_test.go | 20 ++++++++++++++ internal/router/scheduler/profiles.go | 1 + internal/router/scheduler/scorers.go | 33 +++++++++++++++++++++++ 9 files changed, 106 insertions(+), 3 deletions(-) create mode 100644 internal/router/scheduler/latency_test.go 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 +}