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 <noreply@anthropic.com>
This commit is contained in:
1 parent
0b37796067
commit
73e1bb6733
9 files changed
+106
-3
No files matched your search
+4
-1
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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(),
|
||||
|
||||
@@ -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
|
||||
}
|
||||
Reference in new issue
Block a user