diff --git a/deploy/examples/server.yaml b/deploy/examples/server.yaml index 5f6edf4..8f6d9f6 100644 --- a/deploy/examples/server.yaml +++ b/deploy/examples/server.yaml @@ -159,6 +159,12 @@ router: # back to a client that did not ask for it. inject_usage: true max_body_bytes: 33554432 + # Memory for the answers of projects that reuse them (llmgw keys cache), + # all together; the least recently used go first. Nothing is reused until a + # project turns it on. + # cache: + # max_bytes: 134217728 # 128 MB + # max_entry_bytes: 1048576 # 1 MB timeouts: dial: 5s diff --git a/docs/console.md b/docs/console.md index f891558..5546ce6 100644 --- a/docs/console.md +++ b/docs/console.md @@ -195,7 +195,9 @@ row: add a team or project inside it, rename, move, limits for the level and for each of its members, its capacity quota, and for a project how long conversations and agent notes are kept, and whether its requests are (**Keep requests…**: for how long, and what share), and whether they stay in -the cluster (**Requests stay in the cluster…**: never an outside provider). **Longest key lifetime…** caps how +the cluster (**Requests stay in the cluster…**: never an outside provider), +and whether the same request is answered again from memory (**Cache +answers…**: for how long, and whether only reproducible answers or any). **Longest key lifetime…** caps how long a key made there may live, and is the expiry of one made without any (`llmgw keys ttl`; `auth.max_key_ttl` in server.yaml caps every key). **Remove** takes away only something empty. diff --git a/docs/data.md b/docs/data.md index 1f97567..f6ae66b 100644 --- a/docs/data.md +++ b/docs/data.md @@ -67,10 +67,20 @@ until deleted (`DELETE /v1/files/{id}`) or, for what a batch wrote, until them can read them, and a project under a `no_store` policy cannot make them. +**Reused answers** — only for a project that asked (`llmgw keys cache`): +each answer that may be reused, as the client got it, and a hash of the +request it answers, in the gateway process's memory — never in the store, +never on disk — for the project's TTL or until pushed out by newer ones +(`router.cache.max_bytes`), and gone with a restart. Never under a `no_store` +policy. Forgetting a person does not reach into it; the answers age out by +the TTL. The store keeps only the setting, and a usage record's mark that it +was answered from the cache. + ## What is not stored Prompts. Completions. Anything the model was shown or produced, unless the project turned on conversations, notes or kept requests, or sent a batch. +(Reused answers are held in memory only, as above.) The chat page in llmgw's own web interface keeps the conversation in a JavaScript variable in the browser. Refresh it and the conversation is gone, diff --git a/docs/install.md b/docs/install.md index c5923ce..a38a816 100644 --- a/docs/install.md +++ b/docs/install.md @@ -1208,6 +1208,68 @@ llmgw keys requests --project Acme/ml/train --off requests past their hours, or of a project turned off, are removed hourly. The console has it as **Keep requests…** on a project. +## Reusing answers + +A project that asks the same thing again and again — an evaluation run over +the same prompts, embeddings of documents that do not change, an agent that +retries — can have the same request answered from memory rather than by the +model: + +``` +llmgw keys cache --project Acme/ml/train --ttl 1h # reproducible answers +llmgw keys cache --project Acme/ml/eval --ttl 24h --any # every answer +llmgw keys cache --project Acme/ml/train # how it stands +llmgw keys cache --project Acme/ml/train --off +``` + +- **What is the same request:** the same project, the same endpoint, the same + model (by its catalog name, whichever alias was used) and the same body, + with `stream`, `stream_options`, `user`, `metadata`, `store` and + `service_tier` left out and the keys in any order. Keys of another project + never see the answer, whatever they send. +- **Which answers are reused:** a chat or completion asked at `temperature` + 0 or with a `seed`, and every embedding, rerank and score, which sample + nothing. A request sampled above temperature 0 with no seed asks for a new + answer, and is asked anew — unless the project says `--any`. Only a whole + 200 from the model that was asked for is kept: not an error, not a + fallback model's or a canary's answer, not one a guardrail changed or + stopped, not one longer than `router.cache.max_entry_bytes` (1 MB). +- **Streams:** an answer given whole is streamed to a client that asks for a + stream, as one chunk per choice with the usage chunk if it asked for that; + a streamed answer is kept put back together, and served whole to a client + that does not stream. Through `/v1/messages` and `/v1/responses` too. +- **The client decides too:** `Cache-Control: no-cache` (or `no-store`), or + `x-llmgw-cache: no`, is always asked anew. Every answer of a project that + reuses answers says `x-llmgw-cache: hit`, `miss` or `bypass` (not one that + can be reused), and a hit says its `Age` in seconds. +- **Policies still apply:** the limits are checked and the guardrails judge + the request before the cache is looked in, and the output policy judges a + reused answer as it would a new one. A request continuing a stored + conversation, or under a `no_store` policy, is never answered from the + cache nor kept in it. +- **What a hit costs:** nothing. It is recorded as a request — status 200, + the model, the key, the rate limit's count — with no tokens and no cost, + marked as a cache hit (`llmgw ctl request ` says so), so it spends no + budget and no token limit. The answer itself carries the usage the model + reported when it was first given. +- **Where it lives:** in each gateway process's memory, at most + `router.cache.max_bytes` (128 MB) of answers for all projects together, + those used longest ago dropped first; nothing is written to the store, and + a restart begins it again. With two control planes behind a balancer each + keeps its own. A change of setting is seen within 30 seconds; a shorter TTL + applies to what is already kept. + +```yaml +router: + cache: + max_bytes: 268435456 # 256 MB + max_entry_bytes: 2097152 # 2 MB +``` + +`llmgw_cache_requests_total{model,result}` counts hits, misses and +bypasses, `llmgw_cache_bytes` and `llmgw_cache_entries` what is held. The +console has it as **Cache answers…** on a project. + ## Batches Work that can wait — a night's classification, embeddings for a document diff --git a/docs/reference.md b/docs/reference.md index 67325c4..da70b67 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -47,6 +47,8 @@ rather than repeated. | `max_lines` | `int` | [Batch](#batch) | | `concurrency` | `int` | [Batch](#batch) | | `retain_days` | `int` | [Batch](#batch) | +| `max_bytes` | `int64` | [Cache](#cache) | +| `max_entry_bytes` | `int64` | [Cache](#cache) | | `model` | `string` | [Canary](#canary) | | `percent` | `int` | [Canary](#canary) | | `min_requests` | `int` | [Canary](#canary) | @@ -204,6 +206,7 @@ rather than repeated. | `tag_labels` | `[]string` | [Router](#router) | | `max_body_bytes` | `int64` | [Router](#router) | | `batch` | `Batch` | [Router](#router) | +| `cache` | `Cache` | [Router](#router) | | `trusted_proxies` | `[]string` | [Router](#router) | | `timeouts` | `Timeouts` | [Router](#router) | | `scoring` | `Scoring` | [Router](#router) | @@ -570,6 +573,22 @@ every interactive request. RetainDays is how long a batch's answer and error files are kept; 30 unless set. Uploaded files are kept until deleted. +## Cache + +### `max_bytes` + +`int64` + +MaxBytes bounds every project's answers together; 128 MB unless set. +Past it, the answers used longest ago go first. + +### `max_entry_bytes` + +`int64` + +MaxEntryBytes bounds one answer; a longer one is not kept. 1 MB +unless set. + ## Canary ### `model` @@ -1692,6 +1711,16 @@ another. Needs the store. Takes a [Batch](#batch) block. +### `cache` + +`Cache` + +Cache bounds the answers kept for the projects that reuse them +(llmgw keys cache). Which projects do, and for how long, is theirs +to set; this is only how much memory it may take. + +Takes a [Cache](#cache) block. + ### `trusted_proxies` `[]string` diff --git a/internal/cli/answercache.go b/internal/cli/answercache.go new file mode 100644 index 0000000..f6ee577 --- /dev/null +++ b/internal/cli/answercache.go @@ -0,0 +1,85 @@ +package cli + +import ( + "fmt" + "time" + + "github.com/spf13/cobra" +) + +func newAnswerCacheCmd(open opener) *cobra.Command { + var ( + project string + ttl time.Duration + anyAnswer bool + off bool + ) + cmd := &cobra.Command{ + Use: "cache", + Short: "Reuse a project's answers: the same request is answered again without the model", + Long: "With this on, a request of the project that is the same as one answered within\n" + + "--ttl, to the same model, gets that answer again, from the gateway's memory,\n" + + "rather than asking the model. It is answered with x-llmgw-cache: hit, recorded\n" + + "as a request with no tokens, and costs nothing against a budget.\n\n" + + "Only answers that would come out the same are reused: a chat or completion\n" + + "asked at temperature 0 or with a seed, and embeddings, rerank and score.\n" + + "--any reuses every answer, sampled or not. A client sending\n" + + "Cache-Control: no-cache or x-llmgw-cache: no is always asked afresh, and so is\n" + + "a request continuing a stored conversation or under a guardrails no_store\n" + + "policy. The guardrails judge a reused answer as they judge any other.\n\n" + + "Off for every project until this is run. Without --ttl or --off, says how\n" + + "it stands. The answers are held in memory, router.cache bounds how much.", + Example: " llmgw keys cache --project Acme/ml/train --ttl 1h\n" + + " llmgw keys cache --project Acme/ml/eval --ttl 24h --any\n" + + " llmgw keys cache --project Acme/ml/train --off", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + switch { + case off && (ttl != 0 || anyAnswer): + return fmt.Errorf("--off, or --ttl with --any: not both") + case ttl < 0: + return fmt.Errorf("--ttl is how long an answer is reused, like 1h") + case ttl > 0 && ttl < time.Second: + return fmt.Errorf("--ttl is at least a second") + case anyAnswer && ttl == 0: + return fmt.Errorf("--any goes with --ttl: say how long answers are reused") + } + s, err := open(cmd.Context()) + if err != nil { + return err + } + defer s.Close() + p, err := resolveProject(cmd.Context(), s, project) + if err != nil { + return err + } + if off || ttl > 0 { + if err := s.SetAnswerCache(cmd.Context(), p.ID, ttl, anyAnswer); err != nil { + return err + } + } + c, err := s.AnswerCacheOf(cmd.Context(), p.ID) + if err != nil { + return err + } + switch { + case c.TTL == 0: + fmt.Fprintf(cmd.OutOrStdout(), "%s reuses no answers.\n", p.Path) + case c.Any: + fmt.Fprintf(cmd.OutOrStdout(), "%s reuses every answer for %s, sampled or not.\n", p.Path, c.TTL) + default: + fmt.Fprintf(cmd.OutOrStdout(), "%s reuses answers for %s: those asked at temperature 0 or with a seed, "+ + "and embeddings, rerank and score.\n", p.Path, c.TTL) + } + if off || ttl > 0 { + fmt.Fprintln(cmd.OutOrStdout(), "Gateways see the change within 30 seconds.") + } + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "project: id, path or unique name") + cmd.Flags().DurationVar(&ttl, "ttl", 0, "how long an answer is reused, like 1h") + cmd.Flags().BoolVar(&anyAnswer, "any", false, "reuse sampled answers too, not only reproducible ones") + cmd.Flags().BoolVar(&off, "off", false, "stop reusing answers for this project") + return cmd +} diff --git a/internal/cli/answercache_test.go b/internal/cli/answercache_test.go new file mode 100644 index 0000000..15d1e85 --- /dev/null +++ b/internal/cli/answercache_test.go @@ -0,0 +1,32 @@ +package cli + +import ( + "path/filepath" + "strings" + "testing" +) + +func TestAProjectIsToldToReuseItsAnswersAndToStop(t *testing.T) { + db := filepath.Join(t.TempDir(), "s.db") + mustRun(t, db, "org", "add", "org", "Acme") + mustRun(t, db, "org", "add", "team", "ml", "--in", "Acme") + mustRun(t, db, "org", "add", "project", "train", "--in", "Acme/ml") + + if out := mustRun(t, db, "keys", "cache", "--project", "Acme/ml/train"); !strings.Contains(out, "reuses no answers") { + t.Errorf("before: %s", out) + } + if out := mustRun(t, db, "keys", "cache", "--project", "Acme/ml/train", "--ttl", "1h"); !strings.Contains(out, "for 1h0m0s: those asked at temperature 0") { + t.Errorf("on: %s", out) + } + if out := mustRun(t, db, "keys", "cache", "--project", "train", "--ttl", "24h", "--any"); !strings.Contains(out, "every answer for 24h0m0s") { + t.Errorf("any: %s", out) + } + for _, bad := range [][]string{{"--off", "--ttl", "1h"}, {"--any"}, {"--ttl", "10ms"}} { + if _, err := run(t, db, append([]string{"keys", "cache", "--project", "train"}, bad...)...); err == nil { + t.Errorf("%v was accepted", bad) + } + } + if out := mustRun(t, db, "keys", "cache", "--project", "train", "--off"); !strings.Contains(out, "reuses no answers") { + t.Errorf("off: %s", out) + } +} diff --git a/internal/cli/ctl_request.go b/internal/cli/ctl_request.go index 74cba2e..226c104 100644 --- a/internal/cli/ctl_request.go +++ b/internal/cli/ctl_request.go @@ -56,6 +56,9 @@ func newRequestCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { fmt.Fprintf(out, "status %d\n", v.Status) fmt.Fprintf(out, "time %d ms total: %d queued, %d to the first token\n", v.LatencyMs, v.QueueMs, v.TTFTMs) fmt.Fprintf(out, "tokens %d in, %d out\n", v.PromptTokens, v.CompletionTokens) + if v.CacheHit { + fmt.Fprintf(out, "cache answered from the project's answer cache: nothing generated, nothing charged\n") + } fmt.Fprintf(out, "who key %s, project %s%s\n", v.KeyID, v.ProjectID, prefixed(", user ", v.UserID)) if v.EndUser != "" { fmt.Fprintf(out, "end user %s (the client's own name for them)\n", v.EndUser) diff --git a/internal/cli/keys.go b/internal/cli/keys.go index 0ae9e93..67a5dec 100644 --- a/internal/cli/keys.go +++ b/internal/cli/keys.go @@ -44,6 +44,7 @@ func newTenancyCmd() *cobra.Command { newThreadsCmd(open), newRequestLogCmd(open), newResidencyCmd(open), + newAnswerCacheCmd(open), newMemoryCmd(open), newMoveStoreCmd(), ) diff --git a/internal/cli/tui_people.go b/internal/cli/tui_people.go index 7dfab76..cfd9cff 100644 --- a/internal/cli/tui_people.go +++ b/internal/cli/tui_people.go @@ -72,6 +72,8 @@ type peopleData struct { notes map[string]int // project id → hours kept, agent notes // requests is how each project keeps its requests and answers. requests map[string]store.RequestLog + // cache is how each project reuses answers. + cache map[string]store.AnswerCache // guardPolicies are server.yaml's guardrail policies, guardDefault those // applied everywhere, and guardAt which are attached where. guardPolicies []string @@ -125,6 +127,7 @@ func (m consoleModel) fetchPeople() tea.Cmd { d.guardAt, _ = s.GuardAttachments(ctx) } d.threads, d.notes, d.requests = map[string]int{}, map[string]int{}, map[string]store.RequestLog{} + d.cache = map[string]store.AnswerCache{} for _, r := range d.orgs { if r.level != "project" { continue @@ -138,6 +141,9 @@ func (m consoleModel) fetchPeople() tea.Cmd { if l, err := s.RequestLogOf(ctx, r.id); err == nil { d.requests[r.id] = l } + if c, err := s.AnswerCacheOf(ctx, r.id); err == nil { + d.cache[r.id] = c + } } return consolePeopleMsg(d) } @@ -552,6 +558,39 @@ func (m consoleModel) orgMenu(r orgRow) *menuDialog { return m.runCLI("Keep requests", append(args, "--retain-hours="+v[0], "--sample="+v[1])...), nil }}) return nil + }}, + menuItem{"Cache answers…", func(m *consoleModel) tea.Cmd { + c := m.people.cache[r.id] + ttl, which := "off", "reproducible" + if c.TTL > 0 { + ttl = c.TTL.String() + } + if c.Any { + which = "any" + } + m.openForm(&formDialog{title: "Cache answers · " + r.path, + note: "Off by default. On, the same request to the same model within the time below is\n" + + "answered again from the gateway's memory, without the model, and costs nothing.\n" + + "Reproducible: only temperature 0 or a seed, and embeddings. Any: sampled answers too.", + fields: []formField{ + textF("reuse for", ttl, "like 1h; off reuses none"), + choiceF("answers", []string{"reproducible", "any"}, which), + }, + submit: func(m *consoleModel, v []string) (tea.Cmd, error) { + args := []string{"keys", "cache", "--project=" + r.id} + if v[0] == "off" || v[0] == "0" { + return m.runCLI("Cache answers", append(args, "--off")...), nil + } + if d, err := time.ParseDuration(v[0]); err != nil || d < time.Second { + return nil, fmt.Errorf("a duration like 1h, or off") + } + args = append(args, "--ttl="+v[0]) + if v[1] == "any" { + args = append(args, "--any") + } + return m.runCLI("Cache answers", args...), nil + }}) + return nil }}) } items = append(items, menuItem{"Remove…", func(m *consoleModel) tea.Cmd { diff --git a/internal/config/config.go b/internal/config/config.go index 23c0f55..0bf678b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -476,6 +476,11 @@ type Router struct { // another. Needs the store. Batch Batch `koanf:"batch"` + // Cache bounds the answers kept for the projects that reuse them + // (llmgw keys cache). Which projects do, and for how long, is theirs + // to set; this is only how much memory it may take. + Cache Cache `koanf:"cache"` + // TrustedProxies are the reverse proxies in front of the client port, // as addresses or CIDR prefixes. Their X-Forwarded-For, -Proto and -Host // are believed; anyone else's are not, since they decide the address @@ -609,6 +614,18 @@ type Batch struct { RetainDays int `koanf:"retain_days"` } +// Cache is how much of this process's memory reused answers may take. The +// answers are held here and nowhere else: each gateway process keeps its +// own, and a restart begins them again. +type Cache struct { + // MaxBytes bounds every project's answers together; 128 MB unless set. + // Past it, the answers used longest ago go first. + MaxBytes int64 `koanf:"max_bytes"` + // MaxEntryBytes bounds one answer; a longer one is not kept. 1 MB + // unless set. + MaxEntryBytes int64 `koanf:"max_entry_bytes"` +} + // Default returns the configuration used when nothing is specified. func Default() Server { return Server{ diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 920b88b..0f1d52c 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -1419,6 +1419,8 @@ type RequestView struct { // What the client said about it: OpenAI's user field, and its tags. EndUser string `json:"end_user,omitempty"` Tags map[string]string `json:"tags,omitempty"` + // CacheHit is an answer reused from the project's answer cache. + CacheHit bool `json:"cache_hit,omitempty"` } func (a *API) request(w http.ResponseWriter, r *http.Request) { @@ -1442,7 +1444,7 @@ func (a *API) request(w http.ResponseWriter, r *http.Request) { LatencyMs: u.LatencyMs, QueueMs: u.QueueMs, TTFTMs: u.TTFTMs, PromptTokens: u.PromptTokens, CompletionTokens: u.CompletionTokens, CostMicros: u.CostMicros, GuardAction: u.GuardAction, GuardPolicy: u.GuardPolicy, GuardCategory: u.GuardCategory, GuardMs: u.GuardMs, - EndUser: u.EndUser, Tags: u.Tags, + EndUser: u.EndUser, Tags: u.Tags, CacheHit: u.CacheHit, }) } diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index 4503a72..43fc840 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -44,6 +44,10 @@ var Reference = []Block{ {Key: "concurrency", Type: "int", Doc: "Concurrency is how many of one batch's requests are in flight at\nonce; 4 unless set. They queue in the lowest admission band, behind\nevery interactive request.", Nested: ""}, {Key: "retain_days", Type: "int", Doc: "RetainDays is how long a batch's answer and error files are kept;\n30 unless set. Uploaded files are kept until deleted.", Nested: ""}, }}, + {Package: "config", Name: "Cache", Title: "Cache", Doc: "", Fields: []Field{ + {Key: "max_bytes", Type: "int64", Doc: "MaxBytes bounds every project's answers together; 128 MB unless set.\nPast it, the answers used longest ago go first.", Nested: ""}, + {Key: "max_entry_bytes", Type: "int64", Doc: "MaxEntryBytes bounds one answer; a longer one is not kept. 1 MB\nunless set.", Nested: ""}, + }}, {Package: "catalog", Name: "Canary", Title: "Canary", Doc: "", Fields: []Field{ {Key: "model", Type: "string", Doc: "Model is the candidate: a catalog model, usually this one at newer\nweights, which clients cannot name themselves.", Nested: ""}, {Key: "percent", Type: "int", Doc: "Percent of requests go to it, 0 to 100.", Nested: ""}, @@ -248,6 +252,7 @@ var Reference = []Block{ {Key: "tag_labels", Type: "[]string", Doc: "TagLabels are the request tags (metadata, x-llmgw-tags) counted in\nllmgw_tagged_requests_total and llmgw_tagged_tokens_total, by value.\nOnly these: a tag is the client's to set, and every value of one is\na series. The first 100 values of each are counted apart, the rest as\n\"other\". Empty counts none; the usage records keep every tag anyway.", Nested: ""}, {Key: "max_body_bytes", Type: "int64", Doc: "MaxBodyBytes bounds the request body the router will read.", Nested: ""}, {Key: "batch", Type: "Batch", Doc: "Batch is the Files and Batch APIs (/v1/files, /v1/batches): a file of\nrequests served at the lowest priority, its answers written to\nanother. Needs the store.", Nested: "config.Batch"}, + {Key: "cache", Type: "Cache", Doc: "Cache bounds the answers kept for the projects that reuse them\n(llmgw keys cache). Which projects do, and for how long, is theirs\nto set; this is only how much memory it may take.", Nested: "config.Cache"}, {Key: "trusted_proxies", Type: "[]string", Doc: "TrustedProxies are the reverse proxies in front of the client port,\nas addresses or CIDR prefixes. Their X-Forwarded-For, -Proto and -Host\nare believed; anyone else's are not, since they decide the address\nwritten into the setup scripts and whose attempts are counted where.", Nested: ""}, {Key: "timeouts", Type: "Timeouts", Doc: "", Nested: "config.Timeouts"}, {Key: "scoring", Type: "Scoring", Doc: "", Nested: "config.Scoring"}, diff --git a/internal/router/cache.go b/internal/router/cache.go new file mode 100644 index 0000000..aab55a5 --- /dev/null +++ b/internal/router/cache.go @@ -0,0 +1,525 @@ +package router + +import ( + "bytes" + "container/list" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/guard" + "gitea.mynodes.xyz/ilker/llmgw/internal/router/proxy" + storepkg "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// Reusing answers, for a project that asks: the same request to the same +// model, from the same project and within its TTL, is answered with what +// the model said the first time rather than asking it again. +// +// Only an answer that would come out the same is reused, unless the project +// says otherwise: a chat or a completion asked at temperature 0 or with a +// seed, and an embedding, a rerank or a score, which have no sampling at +// all. A request sampled above temperature 0 is asked for a new answer, and +// handing it an old one would quietly turn variety into repetition. +// +// The answers are held in this process's memory and nowhere else. A shared +// cache in the store would write every answer of every project that turned +// this on to the database — the very thing the request log is opt-in for — +// and add a round trip to each request to save one; with two control planes +// behind a balancer each fills its own, which costs some hits and no +// correctness. + +// cacheHeader says what the cache did for a request, hit, miss or bypass, +// and is how a client asks it to stay out of the way: "no". +const cacheHeader = "x-llmgw-cache" + +// Defaults for router.cache. +const ( + defaultCacheBytes = 128 << 20 + defaultCacheEntryBytes = 1 << 20 +) + +// cacheEndpoints are those whose answers can be reused. The chat and +// completion endpoints only when the answer is reproducible; the others +// always, as they sample nothing. Audio is a form, not JSON, and is not +// among them. +var cacheEndpoints = map[string]bool{ + "/v1/chat/completions": true, + "/v1/completions": true, + "/v1/embeddings": true, + "/v1/rerank": true, + "/v1/score": true, +} + +// sampledEndpoints are those whose answer depends on sampling. +var sampledEndpoints = map[string]bool{ + "/v1/chat/completions": true, + "/v1/completions": true, +} + +// cacheState is where a request's answer is kept, for one whose answer +// may be reused and was not found. +type cacheState struct { + // key is the answer's place in the cache; empty when it is not kept. + key string + // project is whose answer it is, and ttl how long it is reused. + project string + ttl time.Duration +} + +// cachedAnswer is one reusable answer. +type cachedAnswer struct { + key, project string + // model is the model that answered, by its catalog name. + model string + path string + // body is the answer as one JSON object — a chat.completion, a + // text_completion, a list of embeddings — even when it was streamed. + body []byte + at time.Time + // expires is when the TTL in force when it was kept runs out. The TTL + // in force when it is asked for is checked too, so shortening a + // project's TTL shortens the life of what it already has. + expires time.Time + size int64 +} + +// answerCache is a least-recently-used cache of answers, bounded in bytes. +type answerCache struct { + maxBytes, maxEntry int64 + + mu sync.Mutex + lru *list.List // of *cachedAnswer, most recently used at the front + by map[string]*list.Element + bytes int64 + // swept is when expired answers were last cleared out. + swept time.Time + // size reports bytes and answers held, after each change. + size func(bytes int64, n int) +} + +func newAnswerCache(maxBytes, maxEntry int64, size func(int64, int)) *answerCache { + if maxBytes <= 0 { + maxBytes = defaultCacheBytes + } + if maxEntry <= 0 { + maxEntry = defaultCacheEntryBytes + } + maxEntry = min(maxEntry, maxBytes) + if size == nil { + size = func(int64, int) {} + } + return &answerCache{maxBytes: maxBytes, maxEntry: maxEntry, lru: list.New(), + by: map[string]*list.Element{}, size: size} +} + +// get is the answer kept under key, when it is younger than ttl and has +// not expired. +func (c *answerCache) get(key string, ttl time.Duration, now time.Time) *cachedAnswer { + c.mu.Lock() + defer c.mu.Unlock() + el, ok := c.by[key] + if !ok { + return nil + } + a := el.Value.(*cachedAnswer) + if now.After(a.expires) || now.Sub(a.at) >= ttl { + c.remove(el) + c.size(c.bytes, len(c.by)) + return nil + } + c.lru.MoveToFront(el) + return a +} + +// put keeps an answer, making room for it by dropping those used longest +// ago. One larger than an entry may be is not kept. +func (c *answerCache) put(a *cachedAnswer, now time.Time) bool { + // What the entry costs besides its body: the key twice, the names, and + // the bookkeeping, roughly. + a.size = int64(len(a.body) + 2*len(a.key) + len(a.project) + len(a.model) + len(a.path) + 160) + if a.size > c.maxEntry { + return false + } + c.mu.Lock() + defer c.mu.Unlock() + if el, ok := c.by[a.key]; ok { + c.remove(el) + } + // Now and then, those past their time, so a quiet project's answers + // do not sit in memory until pushed out by others. + if now.Sub(c.swept) > time.Minute { + c.swept = now + for el := c.lru.Back(); el != nil; { + prev := el.Prev() + if now.After(el.Value.(*cachedAnswer).expires) { + c.remove(el) + } + el = prev + } + } + for c.bytes+a.size > c.maxBytes && c.lru.Len() > 0 { + c.remove(c.lru.Back()) + } + c.by[a.key] = c.lru.PushFront(a) + c.bytes += a.size + c.size(c.bytes, len(c.by)) + return true +} + +func (c *answerCache) remove(el *list.Element) { + a := el.Value.(*cachedAnswer) + c.lru.Remove(el) + delete(c.by, a.key) + c.bytes -= a.size +} + +// answerCacheOf is a project's setting, read at most every requestLogTTL +// with its other settings. Unreadable, it reuses nothing: asking the model +// again is the safe way to be wrong. +func (r *Router) answerCacheOf(ctx context.Context, project string) storepkg.AnswerCache { + if r.store == nil || project == "" { + return storepkg.AnswerCache{} + } + r.reqLogs.mu.Lock() + c, ok := r.reqLogs.by[project] + r.reqLogs.mu.Unlock() + if ok && time.Since(c.at) < requestLogTTL { + return c.cache + } + c, err := r.readProjectPolicy(ctx, project) + if err != nil { + r.log.Warn("could not read whether a project reuses answers", "project", project, "err", err) + } + return c.cache +} + +// cacheLookup decides whether a request's answer may come from its +// project's cache, says what it decided in x-llmgw-cache, and returns the +// answer when there is one. Run after the guardrails have judged the +// request, so a cached answer reaches only a request that would have been +// sent to the model, and the request it is looked up by is the one the +// model would have been sent. +func (r *Router) cacheLookup(ctx context.Context, w http.ResponseWriter, req *parsedRequest) *cachedAnswer { + if r.answers == nil || req.internal || req.ProjectID == "" { + return nil + } + set := r.answerCacheOf(ctx, req.ProjectID) + if set.TTL <= 0 { + // Says nothing: there is no cache here to have bypassed. + return nil + } + result := "bypass" + defer func() { + w.Header().Set(cacheHeader, result) + r.m.cacheRequests.WithLabelValues(r.metricModel(req.Model), result).Inc() + }() + if !reusable(req, set.Any) { + return nil + } + model, _ := r.resolve(req.Model) + key, ok := answerKey(req.ProjectID, req.Path, model, req.UpstreamBody) + if !ok { + return nil + } + now := time.Now() + if a := r.answers.get(key, set.TTL, now); a != nil && a.project == req.ProjectID { + result = "hit" + // Age is HTTP's own way of saying how old a reused answer is. + w.Header().Set("Age", strconv.Itoa(int(now.Sub(a.at).Seconds()))) + return a + } + result = "miss" + req.cache = cacheState{key: key, project: req.ProjectID, ttl: set.TTL} + return nil +} + +// reusable says whether a request's answer may be reused at all: not one +// the client asked to be answered afresh, not one continuing a stored +// conversation (each turn is new, and is saved), not one under a policy +// that keeps nothing — a cached answer is a kept one — and, for chat and +// completions, only one asked at temperature 0 or with a seed unless the +// project reuses any answer. +func reusable(req *parsedRequest, anyAnswer bool) bool { + if req.noStore || req.ThreadID != "" || req.ContentType != "" || !cacheEndpoints[req.Path] { + return false + } + if v := strings.ToLower(strings.TrimSpace(req.Header.Get(cacheHeader))); v == "no" || v == "off" || v == "bypass" { + return false + } + for _, v := range req.Header.Values("Cache-Control") { + v = strings.ToLower(v) + if strings.Contains(v, "no-cache") || strings.Contains(v, "no-store") { + return false + } + } + if !sampledEndpoints[req.Path] || anyAnswer { + return true + } + var b struct { + Temperature *float64 `json:"temperature"` + Seed json.RawMessage `json:"seed"` + } + if json.Unmarshal(req.UpstreamBody, &b) != nil { + return false + } + seeded := len(b.Seed) > 0 && string(b.Seed) != "null" + return seeded || b.Temperature != nil && *b.Temperature == 0 +} + +// answerFields are taken out of a request before it is looked up: they +// change how an answer is delivered or recorded, never what it says. +var answerFields = []string{"model", "stream", "stream_options", "user", "metadata", "store", "service_tier"} + +// answerKey is where a request's answer is kept: the project, the endpoint, +// the model by its catalog name, and the request with what does not change +// the answer taken out and its keys sorted. The project is spelled out +// rather than only hashed, so no answer is ever looked up across projects. +func answerKey(project, path, model string, body []byte) (string, bool) { + d := json.NewDecoder(bytes.NewReader(body)) + d.UseNumber() + var m map[string]any + if err := d.Decode(&m); err != nil || m == nil { + return "", false + } + for _, f := range answerFields { + delete(m, f) + } + // encoding/json writes map keys sorted, at every depth. + norm, err := json.Marshal(canonicalNumbers(m)) + if err != nil { + return "", false + } + h := sha256.New() + for _, part := range [][]byte{[]byte(path), []byte(model), norm} { + h.Write(part) + h.Write([]byte{0}) + } + return project + "\x00" + hex.EncodeToString(h.Sum(nil)), true +} + +// canonicalNumbers writes each number one way, so 0, 0.0 and 0e0 are the +// same temperature. Whole numbers are left as written: a seed above 2^53 +// would lose its last digits as a float, and two seeds become one. +func canonicalNumbers(v any) any { + switch x := v.(type) { + case map[string]any: + for k, e := range x { + x[k] = canonicalNumbers(e) + } + case []any: + for i, e := range x { + x[i] = canonicalNumbers(e) + } + case json.Number: + s := string(x) + if _, err := strconv.ParseInt(s, 10, 64); err == nil { + return x + } + if f, err := strconv.ParseFloat(s, 64); err == nil { + if f == float64(int64(f)) && f > -1<<53 && f < 1<<53 { + return json.Number(strconv.FormatInt(int64(f), 10)) + } + return json.Number(strconv.FormatFloat(f, 'g', -1, 64)) + } + } + return v +} + +// cacheWriter keeps a copy of what is written to the client, up to a +// bound, so an answer can be kept once it is known to be a good one. +type cacheWriter struct { + http.ResponseWriter + status int + buf bytes.Buffer + limit int + over bool +} + +func (c *cacheWriter) WriteHeader(code int) { + if c.status == 0 { + c.status = code + } + c.ResponseWriter.WriteHeader(code) +} + +func (c *cacheWriter) Write(p []byte) (int, error) { + if c.status == 0 { + c.status = http.StatusOK + } + if !c.over { + if c.buf.Len()+len(p) > c.limit { + c.over = true + c.buf = bytes.Buffer{} + } else { + c.buf.Write(p) + } + } + return c.ResponseWriter.Write(p) +} + +// Flush passes through, which the proxy needs to stream. +func (c *cacheWriter) Flush() { + if f, ok := c.ResponseWriter.(http.Flusher); ok { + f.Flush() + } +} + +func (c *cacheWriter) Unwrap() http.ResponseWriter { return c.ResponseWriter } + +// cacheWriterFor wraps w when a request's answer may be kept. +func (r *Router) cacheWriterFor(w http.ResponseWriter, req *parsedRequest) (http.ResponseWriter, *cacheWriter) { + if req.cache.key == "" { + return w, nil + } + cw := &cacheWriter{ResponseWriter: w, limit: int(r.answers.maxEntry)} + return cw, cw +} + +// cacheKeep keeps an answer that may be reused: a whole 200 from the model +// asked for — not a fallback, which is another model, nor a canary, which +// is weights on trial — that no guard changed or stopped. +func (r *Router) cacheKeep(req *parsedRequest, res dispatchResult, status int, cw *cacheWriter) { + if cw == nil || cw.over || status != http.StatusOK || cw.status != http.StatusOK || res.err != nil || + res.outcome.Status != http.StatusOK || res.outcome.Stopped || res.fallback || res.onCanary { + return + } + if res.judged && res.guard.Action != guard.Allow && !res.guard.Shadow { + return + } + var body []byte + captured := bytes.TrimSpace(cw.buf.Bytes()) + if strings.Contains(cw.Header().Get("Content-Type"), "text/event-stream") || + bytes.HasPrefix(captured, []byte("data:")) { + b, ok := assembleStream(req.Path, captured, res.outcome.Usage) + if !ok { + return + } + body = b + } else { + if len(captured) == 0 || captured[0] != '{' || !json.Valid(captured) { + return + } + body = bytes.Clone(captured) + } + model := res.model + if model == "" { + model, _ = r.resolve(req.Model) + } + now := time.Now() + r.answers.put(&cachedAnswer{key: req.cache.key, project: req.cache.project, model: model, + path: req.Path, body: body, at: now, expires: now.Add(req.cache.ttl)}, now) +} + +// serveCached answers a request from the cache, as the model would have: +// as one JSON object, or as a stream of chunks to a client that asked for +// one. The guardrails' output policy judges it as it judges any answer — +// the policy in force now, not the one when it was first given — and the +// usage is recorded as a request that generated nothing: no tokens, no +// cost, marked as a cache hit. +func (r *Router) serveCached(ctx context.Context, w http.ResponseWriter, req *parsedRequest, a *cachedAnswer, start time.Time) { + body := a.body + res := dispatchResult{servedModel: servedModel{model: a.model}, cached: true} + if r.guard != nil { + if f := r.guard.Output(ctx, r.subject(req, a.model)); f != nil { + // A copy: the kept answer is shared by every request reading it, + // and a filter is free to rewrite what it is given. + body = f.Buffered(bytes.Clone(body)) + res.guard, res.judged = f.Verdict(), true + if res.guard.Stricter(req.guardIn) { + setGuardHeaders(w.Header(), res.guard) + } + } + } + var werr error + if req.Stream && sampledEndpoints[req.Path] { + werr = writeCachedStream(w, req, body) + } else { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, werr = w.Write(body) + } + status := http.StatusOK + if werr != nil { + status = 499 + } + res.outcome = proxy.Outcome{Status: http.StatusOK, Wrote: true, Usage: &proxy.Usage{}, + Reply: replyText(body)} + took := time.Since(start) + r.m.observeRequest(r.metricModel(req.Model), "cache", status, took, nil, 0) + r.account(req, res, status, took) + r.accessLog(req, res, status, took) + r.observeOutput(req, res, res.judged) + if req.logRetain > 0 { + go r.keepRequest(context.WithoutCancel(ctx), req, res, status) + } +} + +// writeCachedStream sends a kept answer as the chunks a model streams, with +// the usage chunk only to a client that asked for one. +func writeCachedStream(w http.ResponseWriter, req *parsedRequest, body []byte) error { + events, err := streamFromAnswer(req.Path, body, clientWantsUsage(req)) + if err != nil { + // Not an answer that can be streamed — a guard's replacement, say. + // Sent whole, which a client reads as a model that does not stream. + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, err = w.Write(body) + return err + } + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.WriteHeader(http.StatusOK) + for _, e := range append(events, "[DONE]") { + if _, err := io.WriteString(w, "data: "+e+"\n\n"); err != nil { + return err + } + } + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + return nil +} + +// clientWantsUsage says whether the client asked for the usage chunk +// itself, rather than the gateway asking for it on its behalf. +func clientWantsUsage(req *parsedRequest) bool { + if req.SuppressUsageChunk { + return false + } + var b struct { + StreamOptions *struct { + IncludeUsage bool `json:"include_usage"` + } `json:"stream_options"` + } + return json.Unmarshal(req.UpstreamBody, &b) == nil && b.StreamOptions != nil && b.StreamOptions.IncludeUsage +} + +// replyText is an answer's first choice as text, for the request log. +func replyText(body []byte) string { + var c struct { + Choices []struct { + Message struct { + Content *string `json:"content"` + } `json:"message"` + Text *string `json:"text"` + } `json:"choices"` + } + if json.Unmarshal(body, &c) != nil || len(c.Choices) == 0 { + return "" + } + if s := c.Choices[0].Message.Content; s != nil { + return *s + } + if s := c.Choices[0].Text; s != nil { + return *s + } + return "" +} diff --git a/internal/router/cache_test.go b/internal/router/cache_test.go new file mode 100644 index 0000000..56369dd --- /dev/null +++ b/internal/router/cache_test.go @@ -0,0 +1,615 @@ +package router + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" + + "gitea.mynodes.xyz/ilker/llmgw/internal/guard" + "gitea.mynodes.xyz/ilker/llmgw/internal/router/proxy" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// cachingRouter is pricedRouter with its project reusing answers for an +// hour, reproducible ones only unless anyAnswer. +func cachingRouter(t *testing.T, engineURL string, anyAnswer bool) (*Router, string) { + t.Helper() + rt, secret := pricedRouter(t, engineURL) + if err := rt.store.SetAnswerCache(t.Context(), mustProjectOf(t, rt, "key-1"), time.Hour, anyAnswer); err != nil { + t.Fatal(err) + } + return rt, secret +} + +func ask(temperature any, text string) map[string]any { + b := map[string]any{"model": "q14", "messages": []any{map[string]any{"role": "user", "content": text}}} + if temperature != nil { + b["temperature"] = temperature + } + return b +} + +// usageOf waits for the record of one request. +func usageOf(t *testing.T, rt *Router, id string) store.Usage { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if u, ok, err := rt.store.UsageByRequest(t.Context(), id); err == nil && ok { + return u + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("no usage was recorded for %s", id) + return store.Usage{} +} + +func TestAProjectThatReusesAnswersAnswersTheSameRequestWithoutTheModel(t *testing.T) { + e := newFakeEngine(t, "qwen2.5-14b-instruct") + rt, secret := cachingRouter(t, e.srv.URL, false) + h := rt.Handler() + auth := func(id string) map[string]string { + return map[string]string{"Authorization": "Bearer " + secret, "X-Request-Id": id} + } + + first := post(t, h, ask(0, "what is 2+2"), auth("first")) + if first.Code != 200 || first.Header().Get(cacheHeader) != "miss" { + t.Fatalf("first: %d %q %s", first.Code, first.Header().Get(cacheHeader), first.Body.String()) + } + // The same request, said differently: keys in another order, another + // end user, the temperature written as a float, by the canonical name. + again := map[string]any{"messages": []any{map[string]any{"content": "what is 2+2", "role": "user"}}, + "temperature": 0.0, "model": "qwen2.5-14b-instruct", "user": "someone-else"} + second := post(t, h, again, auth("second")) + if second.Code != 200 || second.Header().Get(cacheHeader) != "hit" { + t.Fatalf("second: %d %q %s", second.Code, second.Header().Get(cacheHeader), second.Body.String()) + } + if e.calls.Load() != 1 { + t.Errorf("the engine was asked %d times; the second answer should have been reused", e.calls.Load()) + } + if second.Body.String() != first.Body.String() { + t.Errorf("reused %s, first was %s", second.Body.String(), first.Body.String()) + } + if second.Header().Get("Age") == "" { + t.Error("a reused answer did not say its age") + } + + // Recorded, as a request that generated nothing. + miss, hit := usageOf(t, rt, "first"), usageOf(t, rt, "second") + if miss.CacheHit || miss.CostMicros == 0 || miss.PromptTokens == 0 { + t.Errorf("the answer the model gave: %+v", miss) + } + if !hit.CacheHit || hit.CostMicros != 0 || hit.PromptTokens != 0 || hit.CompletionTokens != 0 || + hit.Status != 200 || hit.ServedModel != "qwen2.5-14b-instruct" { + t.Errorf("the reused answer: %+v", hit) + } + + // A sampled answer is asked for each time, and says so. + for i := range 2 { + rec := post(t, h, ask(0.7, "write a poem"), auth(fmt.Sprintf("poem-%d", i))) + if rec.Header().Get(cacheHeader) != "bypass" { + t.Errorf("a sampled request: %q", rec.Header().Get(cacheHeader)) + } + } + if e.calls.Load() != 3 { + t.Errorf("the engine was asked %d times, expected each sampled request to reach it", e.calls.Load()) + } + // A seed makes it reproducible. + seeded := ask(0.7, "write a poem") + seeded["seed"] = 42 + post(t, h, seeded, auth("seed-1")) + if rec := post(t, h, seeded, auth("seed-2")); rec.Header().Get(cacheHeader) != "hit" { + t.Errorf("a seeded request: %q", rec.Header().Get(cacheHeader)) + } + // A client that asks for a fresh answer gets one. + hdr := auth("fresh") + hdr["Cache-Control"] = "no-cache" + if rec := post(t, h, ask(0, "what is 2+2"), hdr); rec.Header().Get(cacheHeader) != "bypass" { + t.Errorf("Cache-Control: no-cache: %q", rec.Header().Get(cacheHeader)) + } + hdr = auth("fresh-2") + hdr[cacheHeader] = "no" + if rec := post(t, h, ask(0, "what is 2+2"), hdr); rec.Header().Get(cacheHeader) != "bypass" { + t.Errorf("x-llmgw-cache: no: %q", rec.Header().Get(cacheHeader)) + } + + if n := testutil.ToFloat64(rt.m.cacheRequests.WithLabelValues("qwen2.5-14b-instruct", "hit")); n != 2 { + t.Errorf("%v hits counted", n) + } + if n := testutil.ToFloat64(rt.m.cacheRequests.WithLabelValues("qwen2.5-14b-instruct", "miss")); n != 2 { + t.Errorf("%v misses counted", n) + } + if n := testutil.ToFloat64(rt.m.cacheBytes); n <= 0 { + t.Errorf("the cache holds %v bytes", n) + } + + // Turned off, nothing is reused, and nothing is said about it. + _ = rt.store.SetAnswerCache(t.Context(), mustProjectOf(t, rt, "key-1"), 0, false) + rt.reqLogs.by = nil + calls := e.calls.Load() + if rec := post(t, h, ask(0, "what is 2+2"), auth("off")); rec.Header().Get(cacheHeader) != "" || e.calls.Load() != calls+1 { + t.Errorf("a project that reuses nothing: %q, %d calls", rec.Header().Get(cacheHeader), e.calls.Load()-calls) + } +} + +func TestAProjectThatReusesAnyAnswerReusesSampledOnes(t *testing.T) { + e := newFakeEngine(t, "qwen2.5-14b-instruct") + rt, secret := cachingRouter(t, e.srv.URL, true) + auth := map[string]string{"Authorization": "Bearer " + secret} + post(t, rt.Handler(), ask(nil, "write a poem"), auth) + if rec := post(t, rt.Handler(), ask(nil, "write a poem"), auth); rec.Header().Get(cacheHeader) != "hit" { + t.Errorf("%q", rec.Header().Get(cacheHeader)) + } +} + +func TestKeysOfOneProjectNeverSeeAnotherProjectsAnswers(t *testing.T) { + e := newFakeEngine(t, "qwen2.5-14b-instruct") + rt, secret := cachingRouter(t, e.srv.URL, false) + team, err := rt.store.CreateTeam(t.Context(), mustOrgOf(t, rt), "other") + if err != nil { + t.Fatal(err) + } + other, err := rt.store.CreateProject(t.Context(), team, "elsewhere") + if err != nil { + t.Fatal(err) + } + _ = rt.store.SetAnswerCache(t.Context(), other, time.Hour, true) + secret2 := store.KeyPrefix + strings.Repeat("b", 43) + sum := sha256.Sum256([]byte(secret2)) + if err := rt.store.CreateKey(t.Context(), store.Key{ID: "key-2", Hash: sum[:], Prefix: "bbbb", Name: "other", + ProjectID: other, CreatedAt: time.Now()}); err != nil { + t.Fatal(err) + } + + post(t, rt.Handler(), ask(0, "the secret plan"), map[string]string{"Authorization": "Bearer " + secret}) + rec := post(t, rt.Handler(), ask(0, "the secret plan"), map[string]string{"Authorization": "Bearer " + secret2}) + if rec.Header().Get(cacheHeader) != "miss" || e.calls.Load() != 2 { + t.Errorf("another project's key: %q, the engine asked %d times", rec.Header().Get(cacheHeader), e.calls.Load()) + } +} + +func mustOrgOf(t *testing.T, rt *Router) string { + t.Helper() + orgs, err := rt.store.Tree(t.Context()) + if err != nil || len(orgs) == 0 { + t.Fatalf("no organization: %v", err) + } + return orgs[0].ID +} + +func TestAFailedAnswerIsNotReused(t *testing.T) { + e := newFakeEngine(t, "qwen2.5-14b-instruct") + rt, secret := cachingRouter(t, e.srv.URL, false) + auth := map[string]string{"Authorization": "Bearer " + secret} + e.status.Store(http.StatusBadRequest) + if rec := post(t, rt.Handler(), ask(0, "hello"), auth); rec.Code == 200 { + t.Fatalf("%d", rec.Code) + } + e.status.Store(0) + if rec := post(t, rt.Handler(), ask(0, "hello"), auth); rec.Code != 200 || rec.Header().Get(cacheHeader) != "miss" { + t.Errorf("after a failure: %d %q", rec.Code, rec.Header().Get(cacheHeader)) + } +} + +// switchGuard blocks every request, or puts every one under no_store, when +// told to. +type switchGuard struct { + block, noStore atomic.Bool + // answers, when set, stops answers saying "hi". + answers atomic.Bool +} + +func (g *switchGuard) Input(context.Context, guard.Subject, []byte) (guard.Verdict, error) { + if g.block.Load() { + return guard.Verdict{Action: guard.Block, Policy: "nothing"}, nil + } + return guard.Verdict{}, nil +} + +func (g *switchGuard) Output(context.Context, guard.Subject) guard.OutputFilter { + if !g.answers.Load() { + return nil + } + return &stopHi{} +} + +func (g *switchGuard) Directives(guard.Subject) guard.Directives { + return guard.Directives{NoStore: g.noStore.Load()} +} + +func TestTheGuardrailsStillDecideAboutAnAnswerFromTheCache(t *testing.T) { + e := newFakeEngine(t, "qwen2.5-14b-instruct") + rt, secret := cachingRouter(t, e.srv.URL, false) + g := &switchGuard{} + rt.SetGuard(g) + auth := map[string]string{"Authorization": "Bearer " + secret} + if rec := post(t, rt.Handler(), ask(0, "hello"), auth); rec.Header().Get(cacheHeader) != "miss" { + t.Fatalf("%q", rec.Header().Get(cacheHeader)) + } + + // A policy that now refuses the request refuses it, answer kept or not. + g.block.Store(true) + if rec := post(t, rt.Handler(), ask(0, "hello"), auth); rec.Code != http.StatusBadRequest || rec.Header().Get(cacheHeader) != "" { + t.Errorf("a blocked request: %d %q %s", rec.Code, rec.Header().Get(cacheHeader), rec.Body.String()) + } + g.block.Store(false) + + // A policy that keeps nothing neither reads the cache nor fills it. + g.noStore.Store(true) + if rec := post(t, rt.Handler(), ask(0, "hello"), auth); rec.Code != 200 || rec.Header().Get(cacheHeader) != "bypass" { + t.Errorf("under no_store: %d %q", rec.Code, rec.Header().Get(cacheHeader)) + } + if rec := post(t, rt.Handler(), ask(0, "under no_store"), auth); rec.Header().Get(cacheHeader) != "bypass" { + t.Errorf("under no_store: %q", rec.Header().Get(cacheHeader)) + } + g.noStore.Store(false) + if rec := post(t, rt.Handler(), ask(0, "under no_store"), auth); rec.Header().Get(cacheHeader) != "miss" { + t.Errorf("an answer given under no_store was kept: %q", rec.Header().Get(cacheHeader)) + } + + // An output policy added since judges the reused answer too. + g.answers.Store(true) + rec := post(t, rt.Handler(), ask(0, "hello"), auth) + if rec.Header().Get(cacheHeader) != "hit" || !strings.Contains(rec.Body.String(), "content_filter") || + rec.Header().Get(guardHeader) != "blocked" { + t.Errorf("an answer the output policy stops: %q %q %s", rec.Header().Get(cacheHeader), + rec.Header().Get(guardHeader), rec.Body.String()) + } + // And an answer it stopped is not kept. + if rec := post(t, rt.Handler(), ask(0, "new question"), auth); rec.Header().Get(cacheHeader) != "miss" { + t.Fatalf("%q", rec.Header().Get(cacheHeader)) + } + g.answers.Store(false) + if rec := post(t, rt.Handler(), ask(0, "new question"), auth); rec.Header().Get(cacheHeader) != "miss" { + t.Errorf("an answer a guard stopped was kept: %q", rec.Header().Get(cacheHeader)) + } +} + +// greetingEngine streams "Hello world" to a request that asks for a +// stream, with usage when asked, and answers whole otherwise. +func greetingEngine(t *testing.T) (*httptest.Server, *atomic.Int64) { + t.Helper() + var calls atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/metrics" { + return + } + calls.Add(1) + var b struct { + Stream bool `json:"stream"` + StreamOptions struct { + IncludeUsage bool `json:"include_usage"` + } `json:"stream_options"` + } + body, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(body, &b) + if !b.Stream { + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"id":"c1","object":"chat.completion","created":1,"model":"qwen2.5-14b-instruct",`+ + `"choices":[{"index":0,"message":{"role":"assistant","content":"Hello world"},"finish_reason":"stop"}],`+ + `"usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}`) + return + } + w.Header().Set("Content-Type", "text/event-stream") + head := `{"id":"c1","object":"chat.completion.chunk","created":1,"model":"qwen2.5-14b-instruct","choices":` + for _, c := range []string{ + `[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}`, + `[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]}`, + `[{"index":0,"delta":{"content":" world"},"finish_reason":null}]}`, + `[{"index":0,"delta":{},"finish_reason":"stop"}]}`, + } { + fmt.Fprintf(w, "data: %s%s\n\n", head, c) + } + if b.StreamOptions.IncludeUsage { + fmt.Fprintf(w, "data: %s[],\"usage\":{\"prompt_tokens\":5,\"completion_tokens\":2,\"total_tokens\":7}}\n\n", head) + } + _, _ = io.WriteString(w, "data: [DONE]\n\n") + })) + t.Cleanup(srv.Close) + return srv, &calls +} + +// streamedText is the content of a streamed chat answer, and whether it +// carried usage and ended with [DONE]. +func streamedText(t *testing.T, sse string) (text string, usage, done bool) { + t.Helper() + for _, line := range strings.Split(sse, "\n") { + data, ok := strings.CutPrefix(line, "data: ") + if !ok { + continue + } + if data == "[DONE]" { + done = true + continue + } + var c struct { + Object string `json:"object"` + Choices []struct { + Delta struct { + Content string `json:"content"` + } `json:"delta"` + } `json:"choices"` + Usage *struct { + CompletionTokens int `json:"completion_tokens"` + } `json:"usage"` + } + if err := json.Unmarshal([]byte(data), &c); err != nil || c.Object != "chat.completion.chunk" { + t.Fatalf("not a chunk: %q", data) + } + for _, ch := range c.Choices { + text += ch.Delta.Content + } + if c.Usage != nil && c.Usage.CompletionTokens == 2 { + usage = true + } + } + return text, usage, done +} + +func TestAStreamedAnswerIsReusedForClientsThatStreamAndThatDoNot(t *testing.T) { + srv, calls := greetingEngine(t) + rt, secret := cachingRouter(t, srv.URL, false) + auth := map[string]string{"Authorization": "Bearer " + secret} + streamed := func(includeUsage bool) map[string]any { + b := ask(0, "greet me") + b["stream"] = true + if includeUsage { + b["stream_options"] = map[string]any{"include_usage": true} + } + return b + } + + first := post(t, rt.Handler(), streamed(false), auth) + if first.Header().Get(cacheHeader) != "miss" { + t.Fatalf("%q %s", first.Header().Get(cacheHeader), first.Body.String()) + } + if text, _, _ := streamedText(t, first.Body.String()); text != "Hello world" { + t.Fatalf("streamed %q", text) + } + + // Whole, to a client that does not stream. + whole := post(t, rt.Handler(), ask(0, "greet me"), auth) + var c struct { + Object string `json:"object"` + Choices []struct { + Message struct { + Role, Content string + } `json:"message"` + FinishReason string `json:"finish_reason"` + } `json:"choices"` + Usage struct { + PromptTokens, CompletionTokens int + } `json:"usage"` + } + if err := json.Unmarshal(whole.Body.Bytes(), &c); err != nil || whole.Header().Get(cacheHeader) != "hit" { + t.Fatalf("%q %s %v", whole.Header().Get(cacheHeader), whole.Body.String(), err) + } + if c.Object != "chat.completion" || len(c.Choices) != 1 || c.Choices[0].Message.Content != "Hello world" || + c.Choices[0].Message.Role != "assistant" || c.Choices[0].FinishReason != "stop" { + t.Errorf("put back together as %s", whole.Body.String()) + } + + // Streamed again, with the usage only to a client that asks for it. + for _, withUsage := range []bool{false, true} { + rec := post(t, rt.Handler(), streamed(withUsage), auth) + text, usage, done := streamedText(t, rec.Body.String()) + if rec.Header().Get(cacheHeader) != "hit" || text != "Hello world" || !done || usage != withUsage || + !strings.HasPrefix(rec.Header().Get("Content-Type"), "text/event-stream") { + t.Errorf("streamed from the cache (usage %v): %q %q usage=%v done=%v\n%s", withUsage, + rec.Header().Get(cacheHeader), text, usage, done, rec.Body.String()) + } + } + if calls.Load() != 1 { + t.Errorf("the engine was asked %d times", calls.Load()) + } +} + +func TestManyClientsReadingAndFillingTheCacheAtOnce(t *testing.T) { + e := newFakeEngine(t, "qwen2.5-14b-instruct") + rt, secret := cachingRouter(t, e.srv.URL, false) + h := rt.Handler() + auth := map[string]string{"Authorization": "Bearer " + secret} + var wg sync.WaitGroup + for i := range 40 { + wg.Go(func() { + rec := post(t, h, ask(0, fmt.Sprintf("question %d", i%5)), auth) + if rec.Code != 200 { + t.Errorf("%d %s", rec.Code, rec.Body.String()) + } + }) + } + wg.Wait() + for i := range 5 { + if rec := post(t, h, ask(0, fmt.Sprintf("question %d", i)), auth); rec.Header().Get(cacheHeader) != "hit" { + t.Errorf("question %d: %q", i, rec.Header().Get(cacheHeader)) + } + } +} + +func TestTheAnswerCacheDropsWhatWasUsedLongestAgoPastItsBudget(t *testing.T) { + var held int64 + c := newAnswerCache(1000, 400, func(b int64, _ int) { held = b }) + now := time.Now() + put := func(key string) bool { + return c.put(&cachedAnswer{key: key, project: "p", body: bytes.Repeat([]byte("x"), 100), + at: now, expires: now.Add(time.Hour)}, now) + } + for _, k := range []string{"a", "b", "c"} { + put(k) + } + c.get("a", time.Hour, now) // a is now the most recently used + put("d") + put("e") + if c.get("b", time.Hour, now) != nil { + t.Error("b, used longest ago, was kept past the budget") + } + if c.get("a", time.Hour, now) == nil { + t.Error("a, used recently, was dropped") + } + if held > 1000 { + t.Errorf("holding %d bytes of 1000", held) + } + if c.put(&cachedAnswer{key: "big", body: bytes.Repeat([]byte("x"), 500)}, now) { + t.Error("an answer larger than an entry may be was kept") + } + // Past its TTL, or past a TTL shortened since, it is gone. + if c.get("a", time.Hour, now.Add(2*time.Hour)) != nil { + t.Error("an expired answer was returned") + } + if c.get("e", time.Minute, now.Add(2*time.Minute)) != nil { + t.Error("an answer older than the project's TTL now was returned") + } +} + +func TestAnAnswerIsLookedUpByWhatChangesIt(t *testing.T) { + key := func(project, model, body string) string { + k, ok := answerKey(project, "/v1/chat/completions", model, []byte(body)) + if !ok { + t.Fatalf("no key for %s", body) + } + return k + } + base := key("p", "m", `{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0}`) + same := []string{ + `{"messages":[{"content":"hi","role":"user"}],"temperature":0.0,"model":"alias"}`, + `{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0,"stream":true,"stream_options":{"include_usage":true}}`, + `{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0e0,"user":"u","metadata":{"a":"b"}}`, + } + for _, b := range same { + if key("p", "m", b) != base { + t.Errorf("%s was looked up apart", b) + } + } + different := []string{ + `{"model":"m","messages":[{"role":"user","content":"hi!"}],"temperature":0}`, + `{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0,"max_tokens":5}`, + `{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0,"seed":9007199254740993}`, + } + for _, b := range different { + if key("p", "m", b) == base { + t.Errorf("%s was looked up as the same", b) + } + } + if key("p", "m", `{"seed":9007199254740993}`) == key("p", "m", `{"seed":9007199254740992}`) { + t.Error("two large seeds were taken for one") + } + if key("other", "m", `{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0}`) == base { + t.Error("another project's request was looked up as the same") + } + if key("p", "other", `{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0}`) == base { + t.Error("another model's request was looked up as the same") + } + if _, ok := answerKey("p", "/v1/chat/completions", "m", []byte(`not json`)); ok { + t.Error("a body that is not JSON was given a key") + } +} + +func TestAStreamedAnswerIsPutBackTogetherAndTakenApartAgain(t *testing.T) { + sse := strings.Join([]string{ + `data: {"id":"c","object":"chat.completion.chunk","created":7,"model":"m","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}`, + `data: {"id":"c","object":"chat.completion.chunk","created":7,"model":"m","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"get_weather","arguments":""}}]},"finish_reason":null}]}`, + `data: {"id":"c","object":"chat.completion.chunk","created":7,"model":"m","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"city\":"}}]},"finish_reason":null}]}`, + `data: {"id":"c","object":"chat.completion.chunk","created":7,"model":"m","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"Paris\"}"}}]},"finish_reason":null}]}`, + `data: {"id":"c","object":"chat.completion.chunk","created":7,"model":"m","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}`, + `data: [DONE]`, + }, "\n\n") + "\n\n" + body, ok := assembleStream("/v1/chat/completions", []byte(sse), &proxy.Usage{PromptTokens: 3, CompletionTokens: 4}) + if !ok { + t.Fatal("not put back together") + } + var c struct { + Choices []struct { + Message struct { + Content *string `json:"content"` + ToolCalls []toolCall `json:"tool_calls"` + } `json:"message"` + FinishReason string `json:"finish_reason"` + } `json:"choices"` + Usage struct { + TotalTokens int `json:"total_tokens"` + } `json:"usage"` + } + if err := json.Unmarshal(body, &c); err != nil { + t.Fatal(err) + } + if len(c.Choices) != 1 || c.Choices[0].Message.Content != nil || len(c.Choices[0].Message.ToolCalls) != 1 || + c.Choices[0].Message.ToolCalls[0].Function.Arguments != `{"city":"Paris"}` || + c.Choices[0].Message.ToolCalls[0].ID != "call_1" || c.Choices[0].FinishReason != "tool_calls" || + c.Usage.TotalTokens != 7 { + t.Errorf("put back together as %s", body) + } + + // Taken apart, and put back together again, it is the same answer. + events, err := streamFromAnswer("/v1/chat/completions", body, false) + if err != nil { + t.Fatal(err) + } + again, ok := assembleStream("/v1/chat/completions", + []byte("data: "+strings.Join(append(events, "[DONE]"), "\n\ndata: ")+"\n\n"), &proxy.Usage{PromptTokens: 3, CompletionTokens: 4}) + if !ok || !bytes.Equal(again, body) { + t.Errorf("round trip:\n%s\n%s", body, again) + } + + // What it does not understand, and a stream cut short, are not kept. + for _, bad := range []string{ + strings.Replace(sse, `"delta":{}`, `"delta":{"audio":{"id":"a"}}`, 1), + strings.Replace(sse, `"finish_reason":null}]}`, `"finish_reason":null,"logprobs":{"content":[]}}]}`, 1), + strings.Replace(sse, "data: [DONE]", "", 1), + } { + if _, ok := assembleStream("/v1/chat/completions", []byte(bad), nil); ok { + t.Errorf("kept: %s", bad) + } + } + + // A completion, too. + text := `data: {"id":"t","object":"text_completion","created":1,"model":"m","choices":[{"index":0,"text":"Hel","finish_reason":null}]}` + "\n\n" + + `data: {"id":"t","object":"text_completion","created":1,"model":"m","choices":[{"index":0,"text":"lo","finish_reason":"length"}]}` + "\n\n" + + "data: [DONE]\n\n" + tb, ok := assembleStream("/v1/completions", []byte(text), nil) + if !ok || !strings.Contains(string(tb), `"text":"Hello"`) || !strings.Contains(string(tb), `"object":"text_completion"`) { + t.Errorf("a completion: %s", tb) + } +} + +func TestAnAnswerFromTheCacheReachesAnthropicClientsInTheirShape(t *testing.T) { + srv, calls := greetingEngine(t) + rt, secret := cachingRouter(t, srv.URL, false) + send := func(stream bool) *httptest.ResponseRecorder { + b, _ := json.Marshal(map[string]any{"model": "q14", "max_tokens": 50, "temperature": 0, "stream": stream, + "messages": []any{map[string]any{"role": "user", "content": "greet me"}}}) + req := httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(b)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("x-api-key", secret) + rec := httptest.NewRecorder() + rt.Handler().ServeHTTP(rec, req) + return rec + } + first := send(true) + if first.Code != 200 || !strings.Contains(first.Body.String(), "Hello") { + t.Fatalf("%d %s", first.Code, first.Body.String()) + } + whole := send(false) + if whole.Code != 200 || !strings.Contains(whole.Body.String(), `"text":"Hello world"`) { + t.Errorf("whole: %d %s", whole.Code, whole.Body.String()) + } + streamed := send(true) + if !strings.Contains(streamed.Body.String(), "content_block_delta") || + !strings.Contains(streamed.Body.String(), "Hello world") || + !strings.Contains(streamed.Body.String(), "message_stop") { + t.Errorf("streamed: %s", streamed.Body.String()) + } + if calls.Load() != 1 { + t.Errorf("the engine was asked %d times", calls.Load()) + } +} diff --git a/internal/router/cachestream.go b/internal/router/cachestream.go new file mode 100644 index 0000000..0b05c63 --- /dev/null +++ b/internal/router/cachestream.go @@ -0,0 +1,364 @@ +package router + +import ( + "bufio" + "bytes" + "encoding/json" + "errors" + "strings" + + "gitea.mynodes.xyz/ilker/llmgw/internal/router/proxy" +) + +// A cached answer is kept as one JSON object whichever way it was asked +// for, so a client that streams and one that does not share it. A streamed +// answer is put back together into the object a model would have sent +// whole, and an answer is taken apart into chunks for a client that +// streams. + +// streamChunk is one streamed chunk of a chat or a completion. +type streamChunk struct { + ID string `json:"id"` + Object string `json:"object"` + Created int64 `json:"created"` + Model string `json:"model"` + SystemFingerprint string `json:"system_fingerprint"` + Choices []struct { + Index int `json:"index"` + // Delta is a chat chunk's, as fields: those not understood here + // leave the answer unkept rather than kept without them. + Delta map[string]json.RawMessage `json:"delta"` + Text *string `json:"text"` + FinishReason *string `json:"finish_reason"` + Logprobs json.RawMessage `json:"logprobs"` + } `json:"choices"` +} + +// toolCallDelta is part of one tool call, as streamed. +type toolCallDelta struct { + Index int `json:"index"` + ID string `json:"id"` + Type string `json:"type"` + Function struct { + Name string `json:"name"` + Arguments string `json:"arguments"` + } `json:"function"` +} + +type toolCall struct { + ID string `json:"id"` + Type string `json:"type"` + Function struct { + Name string `json:"name"` + Arguments string `json:"arguments"` + } `json:"function"` +} + +// assembledChoice is one choice being put back together. +type assembledChoice struct { + role string + content, reasoning, text strings.Builder + hasContent, hasReasoning bool + tools []*toolCall + finish *string +} + +// assembleStream puts a streamed chat or completion back together as the +// object the model would have answered without streaming, with usage as the +// engine reported it. ok is false for anything it does not fully +// understand — logprobs, audio, a chunk that is not JSON — which is then +// not kept: an answer kept with something missing would be served short. +func assembleStream(path string, sse []byte, usage *proxy.Usage) ([]byte, bool) { + chat := path == "/v1/chat/completions" + if !chat && path != "/v1/completions" { + return nil, false + } + var head streamChunk + choices := map[int]*assembledChoice{} + maxIndex := -1 + done := false + sc := bufio.NewScanner(bytes.NewReader(sse)) + sc.Buffer(make([]byte, 64<<10), len(sse)+1) + for sc.Scan() { + data, ok := strings.CutPrefix(sc.Text(), "data:") + if !ok { + continue + } + data = strings.TrimSpace(data) + if data == "[DONE]" { + done = true + continue + } + var c streamChunk + if json.Unmarshal([]byte(data), &c) != nil { + return nil, false + } + if head.ID == "" { + head.ID, head.Created, head.Model = c.ID, c.Created, c.Model + } + if c.SystemFingerprint != "" { + head.SystemFingerprint = c.SystemFingerprint + } + for _, ch := range c.Choices { + if len(ch.Logprobs) > 0 && string(ch.Logprobs) != "null" { + return nil, false + } + a := choices[ch.Index] + if a == nil { + a = &assembledChoice{} + choices[ch.Index] = a + maxIndex = max(maxIndex, ch.Index) + } + if ch.FinishReason != nil { + a.finish = ch.FinishReason + } + if ch.Text != nil { + a.text.WriteString(*ch.Text) + } + if !addDelta(a, ch.Delta) { + return nil, false + } + } + } + if sc.Err() != nil || !done || len(choices) == 0 || len(choices) != maxIndex+1 { + // A stream that never said it was done may have been cut short. + return nil, false + } + out := map[string]any{"id": head.ID, "created": head.Created, "model": head.Model} + if head.SystemFingerprint != "" { + out["system_fingerprint"] = head.SystemFingerprint + } + list := make([]map[string]any, 0, len(choices)) + for i := 0; i <= maxIndex; i++ { + a := choices[i] + if a.finish == nil { + return nil, false + } + c := map[string]any{"index": i, "finish_reason": *a.finish, "logprobs": nil} + if chat { + role := a.role + if role == "" { + role = "assistant" + } + msg := map[string]any{"role": role, "content": nil} + if a.hasContent || len(a.tools) == 0 { + msg["content"] = a.content.String() + } + if a.hasReasoning { + msg["reasoning_content"] = a.reasoning.String() + } + if len(a.tools) > 0 { + msg["tool_calls"] = a.tools + } + c["message"] = msg + } else { + c["text"] = a.text.String() + } + list = append(list, c) + } + out["choices"] = list + if chat { + out["object"] = "chat.completion" + } else { + out["object"] = "text_completion" + } + if usage != nil { + out["usage"] = map[string]int{"prompt_tokens": usage.PromptTokens, + "completion_tokens": usage.CompletionTokens, + "total_tokens": usage.PromptTokens + usage.CompletionTokens} + } + b, err := json.Marshal(out) + return b, err == nil +} + +// addDelta adds one chat chunk's delta to its choice, and says whether it +// understood every field of it. +func addDelta(a *assembledChoice, delta map[string]json.RawMessage) bool { + for k, raw := range delta { + if string(raw) == "null" { + continue + } + switch k { + case "role": + if json.Unmarshal(raw, &a.role) != nil { + return false + } + case "content", "reasoning_content": + var s string + if json.Unmarshal(raw, &s) != nil { + return false + } + // Empty says nothing: the first chunk carries the role with an + // empty content even when the answer is a tool call. + if s == "" { + continue + } + if k == "content" { + a.content.WriteString(s) + a.hasContent = true + } else { + a.reasoning.WriteString(s) + a.hasReasoning = true + } + case "refusal": + // An empty refusal is a field some engines always send. + var s string + if json.Unmarshal(raw, &s) != nil || s != "" { + return false + } + case "tool_calls": + var parts []toolCallDelta + if json.Unmarshal(raw, &parts) != nil { + return false + } + for _, p := range parts { + for len(a.tools) <= p.Index { + a.tools = append(a.tools, &toolCall{}) + } + t := a.tools[p.Index] + if p.ID != "" { + t.ID = p.ID + } + if p.Type != "" { + t.Type = p.Type + } + t.Function.Name += p.Function.Name + t.Function.Arguments += p.Function.Arguments + } + default: + return false + } + } + return true +} + +// streamFromAnswer is a kept chat or completion as the chunks a model would +// have streamed: per choice, its whole text in one chunk and then its +// finish reason, and the usage last for a client that asks for it. +func streamFromAnswer(path string, body []byte, withUsage bool) ([]string, error) { + var a struct { + ID string `json:"id"` + Created int64 `json:"created"` + Model string `json:"model"` + SystemFingerprint string `json:"system_fingerprint"` + Usage json.RawMessage `json:"usage"` + Choices []struct { + Index int `json:"index"` + FinishReason json.RawMessage `json:"finish_reason"` + Logprobs json.RawMessage `json:"logprobs"` + Text *string `json:"text"` + Message *struct { + Role string `json:"role"` + Content *string `json:"content"` + ReasoningContent *string `json:"reasoning_content"` + ToolCalls []json.RawMessage `json:"tool_calls"` + } `json:"message"` + } `json:"choices"` + } + if err := json.Unmarshal(body, &a); err != nil { + return nil, err + } + if len(a.Choices) == 0 { + return nil, errors.New("an answer with no choices") + } + chat := path == "/v1/chat/completions" + object := "text_completion" + if chat { + object = "chat.completion.chunk" + } + chunk := func(choices []map[string]any) (string, error) { + c := map[string]any{"id": a.ID, "object": object, "created": a.Created, "model": a.Model, + "choices": choices} + if a.SystemFingerprint != "" { + c["system_fingerprint"] = a.SystemFingerprint + } + b, err := json.Marshal(c) + return string(b), err + } + var events []string + add := func(choices []map[string]any) error { + e, err := chunk(choices) + events = append(events, e) + return err + } + for _, ch := range a.Choices { + var logprobs any + if len(ch.Logprobs) > 0 { + logprobs = ch.Logprobs + } + finish := any(json.RawMessage("null")) + if len(ch.FinishReason) > 0 { + finish = ch.FinishReason + } + if !chat { + if ch.Text == nil { + return nil, errors.New("a completion with no text") + } + if err := add([]map[string]any{{"index": ch.Index, "text": *ch.Text, "logprobs": logprobs, + "finish_reason": nil}}); err != nil { + return nil, err + } + if err := add([]map[string]any{{"index": ch.Index, "text": "", "finish_reason": finish}}); err != nil { + return nil, err + } + continue + } + m := ch.Message + if m == nil { + return nil, errors.New("a chat answer with no message") + } + role := m.Role + if role == "" { + role = "assistant" + } + if err := add([]map[string]any{{"index": ch.Index, "delta": map[string]any{"role": role, "content": ""}, + "finish_reason": nil}}); err != nil { + return nil, err + } + if m.ReasoningContent != nil && *m.ReasoningContent != "" { + if err := add([]map[string]any{{"index": ch.Index, + "delta": map[string]any{"reasoning_content": *m.ReasoningContent}, "finish_reason": nil}}); err != nil { + return nil, err + } + } + if m.Content != nil && *m.Content != "" || logprobs != nil { + content := "" + if m.Content != nil { + content = *m.Content + } + if err := add([]map[string]any{{"index": ch.Index, "delta": map[string]any{"content": content}, + "logprobs": logprobs, "finish_reason": nil}}); err != nil { + return nil, err + } + } + if len(m.ToolCalls) > 0 { + calls := make([]map[string]any, 0, len(m.ToolCalls)) + for i, raw := range m.ToolCalls { + var tc map[string]any + if err := json.Unmarshal(raw, &tc); err != nil { + return nil, err + } + tc["index"] = i + calls = append(calls, tc) + } + if err := add([]map[string]any{{"index": ch.Index, "delta": map[string]any{"tool_calls": calls}, + "finish_reason": nil}}); err != nil { + return nil, err + } + } + if err := add([]map[string]any{{"index": ch.Index, "delta": map[string]any{}, + "finish_reason": finish}}); err != nil { + return nil, err + } + } + if withUsage && len(a.Usage) > 0 && string(a.Usage) != "null" { + c := map[string]any{"id": a.ID, "object": object, "created": a.Created, "model": a.Model, + "choices": []any{}, "usage": a.Usage} + b, err := json.Marshal(c) + if err != nil { + return nil, err + } + events = append(events, string(b)) + } + return events, nil +} diff --git a/internal/router/handlers.go b/internal/router/handlers.go index 5f97865..2e6fa37 100644 --- a/internal/router/handlers.go +++ b/internal/router/handlers.go @@ -366,6 +366,9 @@ type parsedRequest struct { // internal is a request the gateway makes of itself — a judge's — which // no guard filters. internal bool + // cache is where its answer is kept, when its project reuses answers + // and this one may be. + cache cacheState } // inferenceBody covers the fields the router reads across all inference @@ -531,8 +534,17 @@ func (r *Router) handleInference(path string) http.HandlerFunc { } r.decideRequestLog(req.Context(), parsed) - res := r.dispatch(req.Context(), w, parsed) + // After the guardrails, so a policy still decides whether a request is + // answered, and the request looked up is the one the model would be + // sent. + if a := r.cacheLookup(req.Context(), w, parsed); a != nil { + r.serveCached(req.Context(), w, parsed, a, start) + return + } + dw, cw := r.cacheWriterFor(w, parsed) + res := r.dispatch(req.Context(), dw, parsed) status := r.finish(w, parsed, res, start) + r.cacheKeep(parsed, res, status, cw) r.observeCanary(parsed, res, status, time.Since(start)) if parsed.logRetain > 0 { go r.keepRequest(context.WithoutCancel(req.Context()), parsed, res, status) @@ -645,6 +657,9 @@ func (r *Router) account(req *parsedRequest, res dispatchResult, status int, too EndUser: req.EndUser, Tags: req.Tags, } guardFields(&u, req.guardIn, res.guard) + // An answer reused generates nothing: its usage is the zero the hit + // carries, and it costs nothing against a budget or a token limit. + u.CacheHit = res.cached if u.ServedModel == "" { u.ServedModel = req.Model } @@ -1169,6 +1184,9 @@ func (r *Router) accessLog(req *parsedRequest, res dispatchResult, status int, t if res.outcome.Stopped { attrs = append(attrs, "stopped_by_guard", true) } + if res.cached { + attrs = append(attrs, "cache", "hit") + } r.log.Info("request", attrs...) } diff --git a/internal/router/metrics.go b/internal/router/metrics.go index 0d583d2..cd31eb7 100644 --- a/internal/router/metrics.go +++ b/internal/router/metrics.go @@ -31,6 +31,12 @@ type metrics struct { prefixDecisions *prometheus.CounterVec + // cacheRequests counts what the answer cache did, cacheBytes and + // cacheEntries what it holds. + cacheRequests *prometheus.CounterVec + cacheBytes prometheus.Gauge + cacheEntries prometheus.Gauge + tenant *tenantMetrics guard *guardMetrics token *tokenMetrics @@ -91,6 +97,20 @@ func newMetrics() *metrics { Name: "llmgw_admission_rejected_total", Help: "Requests refused by flow control, by reason.", }, []string{"band", "reason"}), + cacheRequests: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "llmgw_cache_requests_total", + Help: "Requests of projects that reuse answers, by what the answer cache did: " + + "hit (answered from it), miss (asked the model, and kept the answer if it could), " + + "bypass (not one whose answer is reused).", + }, []string{"model", "result"}), + cacheBytes: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "llmgw_cache_bytes", + Help: "Memory the answers kept for reuse take in this process, roughly.", + }), + cacheEntries: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "llmgw_cache_entries", + Help: "Answers kept for reuse in this process.", + }), tenant: newTenantMetrics(), guard: newGuardMetrics(), token: newTokenMetrics(), @@ -103,6 +123,7 @@ func (m *metrics) collectors() []prometheus.Collector { m.requests, m.duration, m.ttft, m.tokens, m.scorerValue, m.filtered, m.considered, m.admissionWait, m.admissionRejected, m.prefixDecisions, + m.cacheRequests, m.cacheBytes, m.cacheEntries, }, slices.Concat(m.tenant.collectors(), m.guard.collectors(), m.token.collectors())...) } diff --git a/internal/router/requestlog.go b/internal/router/requestlog.go index 1cf126a..a1d253c 100644 --- a/internal/router/requestlog.go +++ b/internal/router/requestlog.go @@ -28,7 +28,9 @@ type requestLogs struct { type cachedLog struct { l storepkg.RequestLog inCluster bool - at time.Time + // cache is how the project reuses answers. + cache storepkg.AnswerCache + at time.Time } // requestLogOf is a project's setting, read at most every requestLogTTL. @@ -49,7 +51,8 @@ func (r *Router) requestLogOf(ctx context.Context, project string) storepkg.Requ return c.l } -// readProjectPolicy reads a project's keeping and residency, and caches them. +// readProjectPolicy reads a project's keeping, residency and reuse of +// answers, and caches them. func (r *Router) readProjectPolicy(ctx context.Context, project string) (cachedLog, error) { l, err := r.store.RequestLogOf(ctx, project) if err != nil { @@ -59,7 +62,11 @@ func (r *Router) readProjectPolicy(ctx context.Context, project string) (cachedL if err != nil { return cachedLog{}, err } - c := cachedLog{l: l, inCluster: in, at: time.Now()} + ac, err := r.store.AnswerCacheOf(ctx, project) + if err != nil { + return cachedLog{}, err + } + c := cachedLog{l: l, inCluster: in, cache: ac, at: time.Now()} r.reqLogs.mu.Lock() if r.reqLogs.by == nil { r.reqLogs.by = map[string]cachedLog{} diff --git a/internal/router/router.go b/internal/router/router.go index e004443..d129cf6 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -74,6 +74,8 @@ type Router struct { batchWake chan struct{} // reqLogs is each project's keeping of requests, as last read. reqLogs requestLogs + // answers are the answers projects reuse; nil reuses none. + answers *answerCache // extraMetrics are collectors other components publish on /metrics. extraMetrics []prometheus.Collector @@ -264,6 +266,11 @@ func New(cfg config.Server, log *slog.Logger) (*Router, error) { MaxResponseBytes: cfg.Router.MaxBodyBytes, }), } + r.answers = newAnswerCache(cfg.Router.Cache.MaxBytes, cfg.Router.Cache.MaxEntryBytes, + func(bytes int64, n int) { + metrics.cacheBytes.Set(float64(bytes)) + metrics.cacheEntries.Set(float64(n)) + }) r.poller = scrape.NewPoller(reg, scrape.Config{ Interval: cfg.Router.ScrapeInterval, Timeout: cfg.Router.ScrapeTimeout, @@ -765,6 +772,9 @@ type dispatchResult struct { // the candidate, and onCanary whether the candidate served it. canaryOf, canary string onCanary bool + // cached is an answer from the project's answer cache, which no model + // was asked for. + cached bool } type servedModel struct { diff --git a/internal/store/answercache.go b/internal/store/answercache.go new file mode 100644 index 0000000..38fd454 --- /dev/null +++ b/internal/store/answercache.go @@ -0,0 +1,59 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" +) + +// AnswerCache is how a project reuses answers: for how long one is served +// again to the same request, and whether that is any request's answer or +// only one that would come out the same every time. Zero TTL reuses none. +// +// Only the setting lives in the store. The answers are held in each +// gateway's memory, so turning this on writes no text anywhere. +type AnswerCache struct { + TTL time.Duration + // Any reuses answers to requests sampled at a temperature above 0 + // with no seed, which would otherwise each be asked again. + Any bool +} + +// SetAnswerCache turns reusing answers on for a project, or off with ttl 0. +// The TTL is kept in whole seconds, at least one. +func (s *DB) SetAnswerCache(ctx context.Context, projectID string, ttl time.Duration, anyAnswer bool) error { + var secs any + a := 0 + if ttl > 0 { + secs = max(int64(ttl/time.Second), 1) + if anyAnswer { + a = 1 + } + } + res, err := s.db.ExecContext(ctx, + `UPDATE projects SET cache_ttl_seconds = ?, cache_any = ? WHERE id = ?`, secs, a, projectID) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return fmt.Errorf("store: no project %q", projectID) + } + return nil +} + +// AnswerCacheOf is a project's setting; zero when it reuses none. +func (s *DB) AnswerCacheOf(ctx context.Context, projectID string) (AnswerCache, error) { + var secs sql.NullInt64 + var a int + err := s.db.QueryRowContext(ctx, + `SELECT cache_ttl_seconds, cache_any FROM projects WHERE id = ?`, projectID).Scan(&secs, &a) + if errors.Is(err, sql.ErrNoRows) || err == nil && (!secs.Valid || secs.Int64 <= 0) { + return AnswerCache{}, nil + } + if err != nil { + return AnswerCache{}, err + } + return AnswerCache{TTL: time.Duration(secs.Int64) * time.Second, Any: a == 1}, nil +} diff --git a/internal/store/answercache_test.go b/internal/store/answercache_test.go new file mode 100644 index 0000000..f447509 --- /dev/null +++ b/internal/store/answercache_test.go @@ -0,0 +1,57 @@ +package store + +import ( + "testing" + "time" +) + +// answerCacheRoundTrip is the same on both backends: a project's setting +// and a hit's mark on its usage record. +func answerCacheRoundTrip(t *testing.T, s *DB) { + t.Helper() + ctx := t.Context() + proj := tree(t, s) + _, k := mustKey(t, s, "k1", proj) + + if c, err := s.AnswerCacheOf(ctx, proj); err != nil || c.TTL != 0 || c.Any { + t.Fatalf("a new project reuses answers: %+v %v", c, err) + } + if err := s.SetAnswerCache(ctx, proj, 90*time.Minute, true); err != nil { + t.Fatal(err) + } + if c, _ := s.AnswerCacheOf(ctx, proj); c.TTL != 90*time.Minute || !c.Any { + t.Errorf("set: %+v", c) + } + if err := s.SetAnswerCache(ctx, proj, 0, true); err != nil { + t.Fatal(err) + } + if c, _ := s.AnswerCacheOf(ctx, proj); c.TTL != 0 || c.Any { + t.Errorf("off: %+v", c) + } + if err := s.SetAnswerCache(ctx, "no-such-project", time.Hour, false); err == nil { + t.Error("a project that does not exist was set") + } + + now := time.Now() + if err := s.WriteUsage(ctx, []Usage{ + {At: now, KeyID: k.ID, ProjectID: proj, Model: "m", ServedModel: "m", Status: 200, RequestID: "hit", CacheHit: true}, + {At: now, KeyID: k.ID, ProjectID: proj, Model: "m", ServedModel: "m", Status: 200, RequestID: "miss", + PromptTokens: 3, CompletionTokens: 4, CostMicros: 10}, + }); err != nil { + t.Fatal(err) + } + if u, ok, err := s.UsageByRequest(ctx, "hit"); err != nil || !ok || !u.CacheHit { + t.Errorf("a hit: %+v %v", u, err) + } + if u, ok, err := s.UsageByRequest(ctx, "miss"); err != nil || !ok || u.CacheHit { + t.Errorf("a miss: %+v %v", u, err) + } +} + +func TestAProjectsReuseOfAnswersIsKeptAndAHitIsMarked(t *testing.T) { + answerCacheRoundTrip(t, capacityStore(t)) +} + +func TestPostgresKeepsAProjectsReuseOfAnswers(t *testing.T) { + answerCacheRoundTrip(t, openPG(t)) +} diff --git a/internal/store/migrations/postgres/0027_answer_cache.sql b/internal/store/migrations/postgres/0027_answer_cache.sql new file mode 100644 index 0000000..99c4a4e --- /dev/null +++ b/internal/store/migrations/postgres/0027_answer_cache.sql @@ -0,0 +1,11 @@ +-- Ported from the SQLite migration of the same number. +-- Reusing answers, for a project that asks: how long one is reused, and +-- whether any request's answer is, or only those that come out the same +-- every time (temperature 0, or a seed). Off for every project until set. +-- The answers themselves are held in each gateway's memory, never here. +ALTER TABLE projects ADD COLUMN cache_ttl_seconds BIGINT; +ALTER TABLE projects ADD COLUMN cache_any BIGINT NOT NULL DEFAULT 0; + +-- A request answered from that cache: nothing was generated for it, so it +-- carries no tokens and costs nothing, and this says why. +ALTER TABLE usage ADD COLUMN cache_hit BIGINT NOT NULL DEFAULT 0; diff --git a/internal/store/migrations/sqlite/0027_answer_cache.sql b/internal/store/migrations/sqlite/0027_answer_cache.sql new file mode 100644 index 0000000..dc06470 --- /dev/null +++ b/internal/store/migrations/sqlite/0027_answer_cache.sql @@ -0,0 +1,10 @@ +-- Reusing answers, for a project that asks: how long one is reused, and +-- whether any request's answer is, or only those that come out the same +-- every time (temperature 0, or a seed). Off for every project until set. +-- The answers themselves are held in each gateway's memory, never here. +ALTER TABLE projects ADD COLUMN cache_ttl_seconds INTEGER; +ALTER TABLE projects ADD COLUMN cache_any INTEGER NOT NULL DEFAULT 0; + +-- A request answered from that cache: nothing was generated for it, so it +-- carries no tokens and costs nothing, and this says why. +ALTER TABLE usage ADD COLUMN cache_hit INTEGER NOT NULL DEFAULT 0; diff --git a/internal/store/sqlite.go b/internal/store/sqlite.go index babd8ae..da9484b 100644 --- a/internal/store/sqlite.go +++ b/internal/store/sqlite.go @@ -435,18 +435,22 @@ func (s *DB) WriteUsage(ctx context.Context, records []Usage) error { `INSERT INTO usage (ts, key_id, project_id, model, served_model, prompt_tokens, completion_tokens, cost_micros, status, latency_ms, user_id, request_id, replica, queue_ms, ttft_ms, - guard_action, guard_policy, guard_category, guard_ms, end_user, tags) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`) + guard_action, guard_policy, guard_category, guard_ms, end_user, tags, cache_hit) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`) if err != nil { return err } defer stmt.Close() for _, r := range records { + hit := 0 + if r.CacheHit { + hit = 1 + } if _, err := stmt.ExecContext(ctx, millis(r.At), r.KeyID, r.ProjectID, r.Model, r.ServedModel, r.PromptTokens, r.CompletionTokens, r.CostMicros, r.Status, r.LatencyMs, r.UserID, r.RequestID, r.Replica, r.QueueMs, r.TTFTMs, - r.GuardAction, r.GuardPolicy, r.GuardCategory, r.GuardMs, r.EndUser, tagsJSON(r.Tags)); err != nil { + r.GuardAction, r.GuardPolicy, r.GuardCategory, r.GuardMs, r.EndUser, tagsJSON(r.Tags), hit); err != nil { return err } } @@ -856,15 +860,16 @@ func (s *DB) UsageByRequest(ctx context.Context, requestID string) (Usage, bool, var u Usage var ts int64 var tags string + var hit int err := s.db.QueryRowContext(ctx, `SELECT ts, key_id, project_id, COALESCE(user_id, ''), model, served_model, prompt_tokens, completion_tokens, cost_micros, status, latency_ms, request_id, replica, queue_ms, ttft_ms, - guard_action, guard_policy, guard_category, guard_ms, end_user, tags + guard_action, guard_policy, guard_category, guard_ms, end_user, tags, cache_hit FROM usage WHERE request_id = ? ORDER BY ts DESC LIMIT 1`, requestID). Scan(&ts, &u.KeyID, &u.ProjectID, &u.UserID, &u.Model, &u.ServedModel, &u.PromptTokens, &u.CompletionTokens, &u.CostMicros, &u.Status, &u.LatencyMs, &u.RequestID, &u.Replica, &u.QueueMs, &u.TTFTMs, &u.GuardAction, &u.GuardPolicy, &u.GuardCategory, &u.GuardMs, - &u.EndUser, &tags) + &u.EndUser, &tags, &hit) if errors.Is(err, sql.ErrNoRows) { return Usage{}, false, nil } @@ -873,5 +878,6 @@ func (s *DB) UsageByRequest(ctx context.Context, requestID string) (Usage, bool, } u.At = time.UnixMilli(ts) u.Tags = parseTags(tags) + u.CacheHit = hit == 1 return u, true, nil } diff --git a/internal/store/store.go b/internal/store/store.go index d1d338d..d974c35 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -269,6 +269,10 @@ type Usage struct { // and x-llmgw-tags. Both for attributing spend inside a project. EndUser string Tags map[string]string + + // CacheHit says the answer came from the project's answer cache: + // nothing was generated, so the record carries no tokens and no cost. + CacheHit bool } // tagsJSON is how tags are kept: a JSON object, keys sorted, or empty. diff --git a/notes/c17.md b/notes/c17.md new file mode 100644 index 0000000..75558d8 --- /dev/null +++ b/notes/c17.md @@ -0,0 +1,55 @@ +## CHANGELOG + +### Reusing answers + +- **A response cache, per project, off until asked for.** `llmgw keys cache + --project P --ttl 1h` has the same request — same project, endpoint, + model by its catalog name, and body with `stream`, `stream_options`, + `user`, `metadata`, `store` and `service_tier` left out and its keys in + any order — answered from the gateway's memory instead of the model, for + the TTL. `--off` stops it; without either it says how it stands. The + console has it as **Cache answers…** on a project. +- **Only answers that come out the same are reused:** chat and completions + asked at temperature 0 or with a seed, and embeddings, rerank and score. + `--any` reuses sampled answers too. Only a whole 200 from the model asked + for is kept — not an error, a fallback's or a canary's answer, one a + guardrail changed or stopped, or one over `router.cache.max_entry_bytes`. +- **Streams both ways:** a streamed answer is kept put back together into + one completion (tool calls and reasoning included), and a kept answer is + streamed as chunks — with the usage chunk if the client asked — to a + client that streams. `/v1/messages` and `/v1/responses` get it too. +- **Headers:** `x-llmgw-cache: hit|miss|bypass` on every answer of a + project that reuses answers, `Age` on a hit. `Cache-Control: no-cache` + or `no-store`, or `x-llmgw-cache: no`, asks the model anew. +- **Policies still apply:** limits and the input guardrails run before the + cache is looked in, the output policy judges a reused answer as it would + a new one, and a request under `no_store` or continuing a stored + conversation is never served from it nor kept in it. A project's answers + are never looked up by another project's keys. +- **A hit costs nothing:** recorded as a 200 with no tokens and no cost, + marked `cache_hit` on the usage record (migration 0027), and shown by + `llmgw ctl request`; it counts towards the requests-per-minute limit only. +- **In memory only:** `router.cache.max_bytes` (128 MB) for all projects + together in each gateway process, least recently used dropped first; + nothing is written to the store. `llmgw_cache_requests_total{model,result}`, + `llmgw_cache_bytes` and `llmgw_cache_entries`. + +## BACKLOG + +**Done (branch `features-c2`):** exact-match answer cache per project +(`keys cache --ttl/--any/--off`, console **Cache answers…**, `projects` +columns, 30-second setting cache), reproducible answers only unless +`--any`, LRU in memory bounded by `router.cache.max_bytes` and +`max_entry_bytes`, streamed answers kept and replayed as chunks, +`x-llmgw-cache` and `Cache-Control: no-cache`, guardrails before a hit and +the output policy on it, `no_store` and threads bypass, a hit recorded as +`cache_hit` with no tokens and no cost (migration 0027), metrics, docs. +Left: semantic (embedding-similarity) caching through the embeddings +models; a cache shared across control planes (each process keeps its own, +deliberately: a shared one would write every answer to the store); a hit +still waits its turn in admission, because the guardrails judge the request +after admission and the key is the request they let through; logprobs in a +streamed answer leave it unkept; `users forget` does not reach into the +cache (answers age out by the TTL); no listing or flushing of what is +cached (`--off` stops it being served at once, the memory is freed by TTL +and LRU).