From 3a3db8cd201a940c6e25e296d1c6436b452b048f Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sun, 25 Oct 2026 17:47:11 +0100 Subject: [PATCH] Answers to similar questions, not only the same one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit llmgw keys cache --similar 0.95 also gives a chat question the answer to one like it: the last user message embedded by router.cache.embedding_model (asked on the gateway's account, as a judge is, within 2 s), compared by cosine with the questions answered in the same conversation — system prompt, earlier turns, tools and sampling the same. The exact cache is asked first; an exact hit embeds nothing. Said as x-llmgw-cache: similar with x-llmgw-cache-similarity, charged as a hit. The vectors live with their answers in the gateway's memory, counted in max_bytes and evicted with them, not in a pgvector table as BACKLOG I1 planned: the cache keeps answer text out of the store on purpose. A question over 2,000 characters is reused only when the same. Migration 0051 holds the setting. BACKLOG I1. Co-Authored-By: Claude Opus 5.5 --- BACKLOG.md | 8 +- CHANGELOG.md | 5 + docs/data.md | 4 +- docs/install.md | 42 ++++- docs/reference.md | 10 ++ internal/cli/answercache.go | 25 ++- internal/cli/tui_people.go | 15 +- internal/config/config.go | 5 + internal/docgen/reference_gen.go | 1 + internal/router/cache.go | 90 +++++++++-- internal/router/cache_test.go | 6 +- internal/router/metrics.go | 2 +- internal/router/router.go | 4 + internal/router/similar.go | 129 +++++++++++++++ internal/router/similar_test.go | 150 ++++++++++++++++++ internal/store/answercache.go | 25 ++- internal/store/answercache_test.go | 15 +- .../postgres/0051_cache_similar.sql | 6 + .../migrations/sqlite/0051_cache_similar.sql | 5 + 19 files changed, 514 insertions(+), 33 deletions(-) create mode 100644 internal/router/similar.go create mode 100644 internal/router/similar_test.go create mode 100644 internal/store/migrations/postgres/0051_cache_similar.sql create mode 100644 internal/store/migrations/sqlite/0051_cache_similar.sql diff --git a/BACKLOG.md b/BACKLOG.md index cd66be8..8a73c09 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -1496,7 +1496,13 @@ Azure/Anthropic/Bedrock/Vertex adapters, request hooks a customer plugs in, routing by what a request holds. Not taken: a realtime (WebSocket audio) API, prompt management, an `npx` install. -### I1. Answers to similar questions, not only the same one (M) +### I1. Answers to similar questions, not only the same one (M) — done 2026-10-25 +`keys cache --similar`, `router.cache.embedding_model`, `internal/router/similar.go`. Changed +from the plan: the vectors are in the gateway's memory beside the exact answers, not a pgvector +table. The exact cache keeps answer text out of the store on purpose (data.md), and a table of +answers would undo that; it also works on SQLite. The usage row marks a similar hit as a hit +(no column of its own). Open: the 40-rewordings and eval-suite runs on the PoC. +The original item follows. The answer cache (`internal/router/cache.go`, C17) answers a request whose body is the same byte for byte. A help desk asked "how do I reset my password" in forty wordings asks the model forty times. diff --git a/CHANGELOG.md b/CHANGELOG.md index 58462a2..462c766 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,11 @@ APIs change between minors without a deprecation period. ### Added +- **Answers to similar questions.** `llmgw keys cache --similar 0.95` also + gives a chat question the answer already given to one like it in the same + conversation, compared by `router.cache.embedding_model`, as + `x-llmgw-cache: similar` with the similarity. Held in memory beside the + exact answers; nothing written to the store but the setting. - **A cached prompt costs the cached rate.** `cached_per_1k` in a price charges the prompt tokens an engine or provider read from its cache (the answer's `cached_tokens`) at their own rate; without it they cost what any diff --git a/docs/data.md b/docs/data.md index bd40ca1..c90b000 100644 --- a/docs/data.md +++ b/docs/data.md @@ -72,7 +72,9 @@ 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 +policy. A project that reuses answers to similar questions (`--similar`) +also has each question's last user message embedded, and the vector kept +beside its answer, in the same memory and for as long. 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. diff --git a/docs/install.md b/docs/install.md index b43eb6c..f7ccd7b 100644 --- a/docs/install.md +++ b/docs/install.md @@ -2158,8 +2158,46 @@ router: 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 +### Similar questions + +A help desk asked "how do I reset my password" in forty wordings asks the +model forty times under the exact cache. `--similar` also gives a chat +question the answer to one like it: + +``` +llmgw keys cache --project Acme/support/desk --ttl 24h --similar 0.95 +``` + +```yaml +router: + cache: + embedding_model: bge-m3 # an embedding model in the catalog +``` + +- **What is compared:** the last user message, embedded by + `router.cache.embedding_model`, against the questions already answered + in the same conversation — the same system prompt, the same earlier turns, + tools, `response_format` and sampling settings, the same model and + project. A like question in another conversation, or after a tool's + result, is not taken for this one. The exact cache is asked first, and an + exact hit embeds nothing. +- **How alike:** the cosine similarity of the two embeddings, at least the + project's `--similar`. 0.95 is strict; below 0.9 different questions start + to meet. A question that differs only in a number or a name ("2+2" and + "2+3") can be closer than that, so a project whose questions turn on + figures should not use it. A question over 2,000 characters is reused only + when it is the same. +- **Said:** `x-llmgw-cache: similar`, with `x-llmgw-cache-similarity` + and `Age`. Recorded and charged as a hit. The same rules as a hit decide + which answers are reused and kept; guardrails judge the request before. +- **The embedding** is asked of the model by the gateway on its own account, + as a judge is, within 2 seconds; every miss waits for it. Without the + model, or when it does not answer, only the same request is reused. The + vectors are kept in memory with their answers and counted in + `router.cache.max_bytes`. + +`llmgw_cache_requests_total{model,result}` counts hits, similar 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 diff --git a/docs/reference.md b/docs/reference.md index 359d7ca..e01bd62 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -53,6 +53,7 @@ rather than repeated. | `retain_days` | `int` | [Batch](#batch) | | `max_bytes` | `int64` | [Cache](#cache) | | `max_entry_bytes` | `int64` | [Cache](#cache) | +| `embedding_model` | `string` | [Cache](#cache) | | `model` | `string` | [Canary](#canary) | | `percent` | `int` | [Canary](#canary) | | `min_requests` | `int` | [Canary](#canary) | @@ -769,6 +770,15 @@ Past it, the answers used longest ago go first. MaxEntryBytes bounds one answer; a longer one is not kept. 1 MB unless set. +### `embedding_model` + +`string` + +EmbeddingModel embeds the questions of projects that reuse answers to +similar ones (llmgw keys cache --similar): an embedding model in the +catalog. Unset, those projects reuse only the same request's answer. +The gateway asks it on its own account, as it asks a judge. + ## Canary ### `model` diff --git a/internal/cli/answercache.go b/internal/cli/answercache.go index f6ee577..24bd43f 100644 --- a/internal/cli/answercache.go +++ b/internal/cli/answercache.go @@ -5,6 +5,8 @@ import ( "time" "github.com/spf13/cobra" + + "gitea.mynodes.xyz/ilker/llmgw/internal/store" ) func newAnswerCacheCmd(open opener) *cobra.Command { @@ -12,6 +14,7 @@ func newAnswerCacheCmd(open opener) *cobra.Command { project string ttl time.Duration anyAnswer bool + similar float64 off bool ) cmd := &cobra.Command{ @@ -28,15 +31,25 @@ func newAnswerCacheCmd(open opener) *cobra.Command { "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.", + "it stands. The answers are held in memory, router.cache bounds how much.\n\n" + + "--similar also answers a chat question that is like one already answered:\n" + + "its last user message embedded by router.cache.embedding_model, and given\n" + + "the answer whose question reaches that cosine similarity (0.95 is strict,\n" + + "0.9 looser), as x-llmgw-cache: similar. The system prompt, the earlier\n" + + "turns, tools and the sampling settings must still be the same.", 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/support/desk --ttl 24h --similar 0.95\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 off && (ttl != 0 || anyAnswer || similar != 0): + return fmt.Errorf("--off, or --ttl with --any or --similar: not both") + case similar < 0 || similar > 1: + return fmt.Errorf("--similar is a cosine similarity between 0 and 1, like 0.95") + case similar != 0 && ttl == 0: + return fmt.Errorf("--similar goes with --ttl: say how long answers are reused") case ttl < 0: return fmt.Errorf("--ttl is how long an answer is reused, like 1h") case ttl > 0 && ttl < time.Second: @@ -54,7 +67,7 @@ func newAnswerCacheCmd(open opener) *cobra.Command { return err } if off || ttl > 0 { - if err := s.SetAnswerCache(cmd.Context(), p.ID, ttl, anyAnswer); err != nil { + if err := s.SetAnswerCache(cmd.Context(), p.ID, store.AnswerCache{TTL: ttl, Any: anyAnswer, Similar: similar}); err != nil { return err } } @@ -71,6 +84,9 @@ func newAnswerCacheCmd(open opener) *cobra.Command { 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 c.TTL > 0 && c.Similar > 0 { + fmt.Fprintf(cmd.OutOrStdout(), "Chat questions reaching %.2f similarity to one answered get its answer too.\n", c.Similar) + } if off || ttl > 0 { fmt.Fprintln(cmd.OutOrStdout(), "Gateways see the change within 30 seconds.") } @@ -80,6 +96,7 @@ func newAnswerCacheCmd(open opener) *cobra.Command { 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().Float64Var(&similar, "similar", 0, "also answer chat questions this similar to one answered, like 0.95") cmd.Flags().BoolVar(&off, "off", false, "stop reusing answers for this project") return cmd } diff --git a/internal/cli/tui_people.go b/internal/cli/tui_people.go index 15a8e5d..46e848e 100644 --- a/internal/cli/tui_people.go +++ b/internal/cli/tui_people.go @@ -573,13 +573,20 @@ func (m consoleModel) orgMenu(r orgRow) *menuDialog { if c.Any { which = "any" } + similar := "off" + if c.Similar > 0 { + similar = strconv.FormatFloat(c.Similar, 'f', -1, 64) + } 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.", + "Reproducible: only temperature 0 or a seed, and embeddings. Any: sampled answers too.\n" + + "Similar: a chat question this alike to one answered gets its answer (needs\n" + + "router.cache.embedding_model).", fields: []formField{ textF("reuse for", ttl, "like 1h; off reuses none"), choiceF("answers", []string{"reproducible", "any"}, which), + textF("similar", similar, "like 0.95; off reuses only the same request"), }, submit: func(m *consoleModel, v []string) (tea.Cmd, error) { args := []string{"keys", "cache", "--project=" + r.id} @@ -593,6 +600,12 @@ func (m consoleModel) orgMenu(r orgRow) *menuDialog { if v[1] == "any" { args = append(args, "--any") } + if v[2] != "off" && v[2] != "" && v[2] != "0" { + if f, err := strconv.ParseFloat(v[2], 64); err != nil || f <= 0 || f > 1 { + return nil, fmt.Errorf("similar is a number like 0.95, or off") + } + args = append(args, "--similar="+v[2]) + } return m.runCLI("Cache answers", args...), nil }}) return nil diff --git a/internal/config/config.go b/internal/config/config.go index b21ea99..ea39429 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -798,6 +798,11 @@ type Cache struct { // MaxEntryBytes bounds one answer; a longer one is not kept. 1 MB // unless set. MaxEntryBytes int64 `koanf:"max_entry_bytes"` + // EmbeddingModel embeds the questions of projects that reuse answers to + // similar ones (llmgw keys cache --similar): an embedding model in the + // catalog. Unset, those projects reuse only the same request's answer. + // The gateway asks it on its own account, as it asks a judge. + EmbeddingModel string `koanf:"embedding_model"` } // Default returns the configuration used when nothing is specified. diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index fc0f173..8ac2744 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -53,6 +53,7 @@ var Reference = []Block{ {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: ""}, + {Key: "embedding_model", Type: "string", Doc: "EmbeddingModel embeds the questions of projects that reuse answers to\nsimilar ones (llmgw keys cache --similar): an embedding model in the\ncatalog. Unset, those projects reuse only the same request's answer.\nThe gateway asks it on its own account, as it asks a judge.", 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: ""}, diff --git a/internal/router/cache.go b/internal/router/cache.go index 895746a..86ede02 100644 --- a/internal/router/cache.go +++ b/internal/router/cache.go @@ -72,6 +72,10 @@ type cacheState struct { // project is whose answer it is, and ttl how long it is reused. project string ttl time.Duration + // group and vec place it among answers to similar questions, for a + // project that reuses those; empty otherwise. + group string + vec []float32 } // cachedAnswer is one reusable answer. @@ -89,16 +93,23 @@ type cachedAnswer struct { // project's TTL shortens the life of what it already has. expires time.Time size int64 + // group is the conversation the question was asked in, everything of + // the request but its last user message; vec is that message embedded, + // of unit length. Empty for an answer reused only by the same request. + group string + vec []float32 } // 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 + mu sync.Mutex + lru *list.List // of *cachedAnswer, most recently used at the front + by map[string]*list.Element + // groups are the answers to similar questions, by conversation. + groups map[string]map[*list.Element]struct{} + bytes int64 // swept is when expired answers were last cleared out. swept time.Time // size reports bytes and answers held, after each change. @@ -117,7 +128,7 @@ func newAnswerCache(maxBytes, maxEntry int64, size func(int64, int)) *answerCach size = func(int64, int) {} } return &answerCache{maxBytes: maxBytes, maxEntry: maxEntry, lru: list.New(), - by: map[string]*list.Element{}, size: size} + by: map[string]*list.Element{}, groups: map[string]map[*list.Element]struct{}{}, size: size} } // get is the answer kept under key, when it is younger than ttl and has @@ -144,7 +155,8 @@ func (c *answerCache) get(key string, ttl time.Duration, now time.Time) *cachedA 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) + a.size = int64(len(a.body) + 2*len(a.key) + len(a.project) + len(a.model) + len(a.path) + 160 + + len(a.group) + 4*len(a.vec)) if a.size > c.maxEntry { return false } @@ -168,7 +180,14 @@ func (c *answerCache) put(a *cachedAnswer, now time.Time) bool { for c.bytes+a.size > c.maxBytes && c.lru.Len() > 0 { c.remove(c.lru.Back()) } - c.by[a.key] = c.lru.PushFront(a) + el := c.lru.PushFront(a) + c.by[a.key] = el + if a.group != "" { + if c.groups[a.group] == nil { + c.groups[a.group] = map[*list.Element]struct{}{} + } + c.groups[a.group][el] = struct{}{} + } c.bytes += a.size c.size(c.bytes, len(c.by)) return true @@ -178,9 +197,38 @@ func (c *answerCache) remove(el *list.Element) { a := el.Value.(*cachedAnswer) c.lru.Remove(el) delete(c.by, a.key) + if g := c.groups[a.group]; g != nil { + delete(g, el) + if len(g) == 0 { + delete(c.groups, a.group) + } + } c.bytes -= a.size } +// nearest is the answer in group whose question is most like vec, when +// that reaches least and the answer is younger than ttl, and how alike. +func (c *answerCache) nearest(group string, vec []float32, least float64, ttl time.Duration, now time.Time) (*cachedAnswer, float64) { + c.mu.Lock() + defer c.mu.Unlock() + var best *list.Element + bestSim := least + for el := range c.groups[group] { + a := el.Value.(*cachedAnswer) + if now.After(a.expires) || now.Sub(a.at) >= ttl || len(a.vec) != len(vec) { + continue + } + if s := dot(a.vec, vec); s >= bestSim { + best, bestSim = el, s + } + } + if best == nil { + return nil, 0 + } + c.lru.MoveToFront(best) + return best.Value.(*cachedAnswer), bestSim +} + // 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. @@ -238,6 +286,15 @@ func (r *Router) cacheLookup(ctx context.Context, w http.ResponseWriter, req *pa } result = "miss" req.cache = cacheState{key: key, project: req.ProjectID, ttl: set.TTL} + if set.Similar > 0 { + if a, sim := r.similarLookup(ctx, req, model, set); a != nil { + result = "similar" + w.Header().Set("Age", strconv.Itoa(int(now.Sub(a.at).Seconds()))) + w.Header().Set(similarityHeader, strconv.FormatFloat(sim, 'f', 3, 64)) + req.cache = cacheState{} + return a + } + } return nil } @@ -288,15 +345,29 @@ var answerFields = []string{"model", "stream", "stream_options", "user", "metada // 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) { + m, ok := decodeRequest(body) + if !ok { + return "", false + } + return keyOf(project, path, model, m) +} + +// decodeRequest is a request as a map, its numbers kept as written and +// what does not change the answer taken out. +func decodeRequest(body []byte) (map[string]any, 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 + return nil, false } for _, f := range answerFields { delete(m, f) } + return m, true +} + +func keyOf(project, path, model string, m map[string]any) (string, bool) { // encoding/json writes map keys sorted, at every depth. norm, err := json.Marshal(canonicalNumbers(m)) if err != nil { @@ -420,7 +491,8 @@ func (r *Router) cacheKeep(req *parsedRequest, res dispatchResult, status int, c } 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) + path: req.Path, body: body, at: now, expires: now.Add(req.cache.ttl), + group: req.cache.group, vec: req.cache.vec}, now) } // serveCached answers a request from the cache, as the model would have: diff --git a/internal/router/cache_test.go b/internal/router/cache_test.go index 56369dd..37a2c76 100644 --- a/internal/router/cache_test.go +++ b/internal/router/cache_test.go @@ -27,7 +27,7 @@ import ( 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 { + if err := rt.store.SetAnswerCache(t.Context(), mustProjectOf(t, rt, "key-1"), store.AnswerCache{TTL: time.Hour, Any: anyAnswer}); err != nil { t.Fatal(err) } return rt, secret @@ -135,7 +135,7 @@ func TestAProjectThatReusesAnswersAnswersTheSameRequestWithoutTheModel(t *testin } // Turned off, nothing is reused, and nothing is said about it. - _ = rt.store.SetAnswerCache(t.Context(), mustProjectOf(t, rt, "key-1"), 0, false) + _ = rt.store.SetAnswerCache(t.Context(), mustProjectOf(t, rt, "key-1"), store.AnswerCache{}) 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 { @@ -164,7 +164,7 @@ func TestKeysOfOneProjectNeverSeeAnotherProjectsAnswers(t *testing.T) { if err != nil { t.Fatal(err) } - _ = rt.store.SetAnswerCache(t.Context(), other, time.Hour, true) + _ = rt.store.SetAnswerCache(t.Context(), other, store.AnswerCache{TTL: time.Hour, Any: 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", diff --git a/internal/router/metrics.go b/internal/router/metrics.go index 10df852..2837ae5 100644 --- a/internal/router/metrics.go +++ b/internal/router/metrics.go @@ -139,7 +139,7 @@ func newMetrics() *metrics { 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), " + + "hit (answered from it), similar (answered from it for a question like this one), miss (asked the model, and kept the answer if it could), " + "bypass (not one whose answer is reused).", }, []string{"model", "result"}), kbRetrievals: prometheus.NewCounterVec(prometheus.CounterOpts{ diff --git a/internal/router/router.go b/internal/router/router.go index 2212703..045f7f2 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -91,6 +91,9 @@ type Router struct { reqLogs requestLogs // answers are the answers projects reuse; nil reuses none. answers *answerCache + // cacheEmbedModel embeds the questions of projects that reuse answers + // to similar ones; empty, they reuse only the same request's. + cacheEmbedModel string // extraMetrics are collectors other components publish on /metrics. extraMetrics []prometheus.Collector @@ -345,6 +348,7 @@ func New(cfg config.Server, log *slog.Logger) (*Router, error) { MaxResponseBytes: cfg.Router.MaxBodyBytes, }), } + r.cacheEmbedModel = cfg.Router.Cache.EmbeddingModel r.answers = newAnswerCache(cfg.Router.Cache.MaxBytes, cfg.Router.Cache.MaxEntryBytes, func(bytes int64, n int) { metrics.cacheBytes.Set(float64(bytes)) diff --git a/internal/router/similar.go b/internal/router/similar.go new file mode 100644 index 0000000..f2afc64 --- /dev/null +++ b/internal/router/similar.go @@ -0,0 +1,129 @@ +package router + +import ( + "context" + "encoding/json" + "fmt" + "math" + "time" + + storepkg "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// Answers to similar questions, for a project that asks: a help desk asked +// "how do I reset my password" in forty wordings asks the model once. The +// last user message is embedded, and an answer kept for a question in the +// same conversation whose embedding reaches the project's similarity is +// given again. Everything else of the request — the system prompt, the +// earlier turns, tools, the sampling settings — must be the same, so a like +// question in another conversation is not taken for this one. +// +// Kept beside the exact answers, in this process's memory, for the reasons +// those are; an answer pushed out of the cache takes its question with it. + +// similarityHeader says how alike a reused answer's question was. +const similarityHeader = "x-llmgw-cache-similarity" + +// maxSimilarQuestion bounds the questions compared: a longer one is a +// document pasted in, where a changed figure or name leaves the embedding +// almost as it was. Those are reused only when they are the same. +const maxSimilarQuestion = 2000 + +// similarTimeout bounds asking for the embedding, which every miss waits on. +const similarTimeout = 2 * time.Second + +// similarLookup is the answer to a question like this request's, and how +// alike. On a miss it leaves in req.cache where this request's answer is to +// be kept among the similar ones. +func (r *Router) similarLookup(ctx context.Context, req *parsedRequest, model string, set storepkg.AnswerCache) (*cachedAnswer, float64) { + if r.cacheEmbedModel == "" || req.Path != "/v1/chat/completions" { + return nil, 0 + } + group, question, ok := similarGroup(req.ProjectID, req.Path, model, req.UpstreamBody) + if !ok || question == "" || len(question) > maxSimilarQuestion { + return nil, 0 + } + vec, err := r.embedQuestion(ctx, question) + if err != nil { + r.log.Warn("a question could not be embedded for the cache; only the same request is reused", + "model", r.cacheEmbedModel, "err", err) + return nil, 0 + } + if a, sim := r.answers.nearest(group, vec, set.Similar, set.TTL, time.Now()); a != nil && a.project == req.ProjectID { + return a, sim + } + req.cache.group, req.cache.vec = group, vec + return nil, 0 +} + +// similarGroup is the conversation a chat request asks its question in — +// the request with the last message's content left out, keyed like an +// answer — and the question. Not one when the last message is not the +// user's: a tool's result is not a question asked again in other words. +func similarGroup(project, path, model string, body []byte) (group, question string, ok bool) { + m, ok := decodeRequest(body) + if !ok { + return "", "", false + } + msgs, _ := m["messages"].([]any) + if len(msgs) == 0 { + return "", "", false + } + last, _ := msgs[len(msgs)-1].(map[string]any) + if last == nil || last["role"] != "user" { + return "", "", false + } + question = lastUserText(body) + last["content"] = "" + group, ok = keyOf(project, path+"#similar", model, m) + return group, question, ok +} + +// embedQuestion has router.cache.embedding_model embed a question, on the +// gateway's own account as a judge is asked, into a vector of unit length. +func (r *Router) embedQuestion(ctx context.Context, text string) ([]float32, error) { + if _, m := r.resolve(r.cacheEmbedModel); m.QueryPrefix != "" { + text = m.QueryPrefix + text + } + ctx, cancel := context.WithTimeout(ctx, similarTimeout) + defer cancel() + body, _ := json.Marshal(map[string]any{"model": r.cacheEmbedModel, "input": []string{text}}) + out, err := r.judgeOnce(ctx, "/v1/embeddings", body, nil, nil) + if err != nil { + return nil, err + } + var res struct { + Data []struct { + Embedding []float32 `json:"embedding"` + } `json:"data"` + } + if err := json.Unmarshal(out, &res); err != nil || len(res.Data) != 1 || len(res.Data[0].Embedding) == 0 { + return nil, fmt.Errorf("%s answered no embedding", r.cacheEmbedModel) + } + return unit(res.Data[0].Embedding), nil +} + +// unit is v scaled to length 1, so that cosine similarity is a dot product. +func unit(v []float32) []float32 { + var n float64 + for _, x := range v { + n += float64(x) * float64(x) + } + if n == 0 { + return v + } + n = math.Sqrt(n) + out := make([]float32, len(v)) + for i, x := range v { + out[i] = float32(float64(x) / n) + } + return out +} + +func dot(a, b []float32) float64 { + var s float64 + for i := range a { + s += float64(a[i]) * float64(b[i]) + } + return s +} diff --git a/internal/router/similar_test.go b/internal/router/similar_test.go new file mode 100644 index 0000000..fe29f1a --- /dev/null +++ b/internal/router/similar_test.go @@ -0,0 +1,150 @@ +package router + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "strings" + "sync/atomic" + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// similarRouter is cachingRouter whose project also reuses answers to +// questions 0.95 alike, embedded by "emb": an engine that places each +// question where the test says, beside the chat engine. +func similarRouter(t *testing.T, places map[string][]float32) (*Router, string, *fakeEngine, *atomic.Int64) { + t.Helper() + e := newFakeEngine(t, "qwen2.5-14b-instruct") + target, _ := url.Parse(e.srv.URL) + chat := httputil.NewSingleHostReverseProxy(target) + var embeds atomic.Int64 + mux := http.NewServeMux() + mux.HandleFunc("POST /v1/embeddings", func(w http.ResponseWriter, r *http.Request) { + embeds.Add(1) + var in struct { + Input []string `json:"input"` + } + b, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(b, &in) + v, ok := places[in.Input[0]] + if !ok { + v = []float32{0, 0, 1} + } + _ = json.NewEncoder(w).Encode(map[string]any{"data": []any{map[string]any{"index": 0, "embedding": v}}}) + }) + mux.Handle("/", chat) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + + rt, secret := pricedRouter(t, srv.URL, "emb") + rt.cacheEmbedModel = "emb" + if err := rt.store.SetAnswerCache(t.Context(), mustProjectOf(t, rt, "key-1"), + store.AnswerCache{TTL: time.Hour, Similar: 0.95}); err != nil { + t.Fatal(err) + } + return rt, secret, e, &embeds +} + +func TestASimilarQuestionGetsTheAnswerAlreadyGiven(t *testing.T) { + rt, secret, e, embeds := similarRouter(t, map[string][]float32{ + "how do I reset my password": {1, 0.1, 0}, + "how can I reset my password?": {1, 0.12, 0}, + "what is the capital of Sweden": {0, 1, 0}, + "how do I reset my password, then": {1, 0.11, 0}, + }) + h := rt.Handler() + hdr := func(id string) map[string]string { + return map[string]string{"Authorization": "Bearer " + secret, "X-Request-Id": id} + } + + first := post(t, h, ask(0, "how do I reset my password"), hdr("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()) + } + second := post(t, h, ask(0, "how can I reset my password?"), hdr("second")) + if second.Code != 200 || second.Header().Get(cacheHeader) != "similar" { + t.Fatalf("second: %d %q %s", second.Code, second.Header().Get(cacheHeader), second.Body.String()) + } + if s := second.Header().Get(similarityHeader); !strings.HasPrefix(s, "0.99") && s != "1.000" { + t.Errorf("similarity %q", s) + } + if second.Body.String() != first.Body.String() { + t.Errorf("the similar question got another answer:\n%s\n%s", first.Body, second.Body) + } + if u := usageOf(t, rt, "second"); !u.CacheHit || u.CostMicros != 0 { + t.Errorf("a similar hit is recorded as %+v", u) + } + + other := post(t, h, ask(0, "what is the capital of Sweden"), hdr("other")) + if other.Header().Get(cacheHeader) != "miss" { + t.Errorf("an unlike question: %q", other.Header().Get(cacheHeader)) + } + + // The same question after another system prompt is another conversation. + sys := ask(0, "how do I reset my password, then") + sys["messages"] = append([]any{map[string]any{"role": "system", "content": "You are a pirate."}}, sys["messages"].([]any)...) + if rec := post(t, h, sys, hdr("system")); rec.Header().Get(cacheHeader) != "miss" { + t.Errorf("a like question under another system prompt: %q", rec.Header().Get(cacheHeader)) + } + // A request answered exactly is not embedded at all. + before := embeds.Load() + if rec := post(t, h, ask(0, "how do I reset my password"), hdr("exact")); rec.Header().Get(cacheHeader) != "hit" { + t.Errorf("the same question: %q", rec.Header().Get(cacheHeader)) + } + if embeds.Load() != before { + t.Error("an exact hit waited on the embedding model") + } + if got := e.calls.Load(); got != 3 { + t.Errorf("the chat model was asked %d times, want 3 (first, other, system)", got) + } +} + +func TestWithoutAnEmbeddingModelOnlyTheSameRequestIsReused(t *testing.T) { + rt, secret, e, embeds := similarRouter(t, map[string][]float32{ + "how do I reset my password": {1, 0.1, 0}, + "how can I reset my password?": {1, 0.12, 0}, + }) + rt.cacheEmbedModel = "" + h := rt.Handler() + hdr := map[string]string{"Authorization": "Bearer " + secret} + post(t, h, ask(0, "how do I reset my password"), hdr) + if rec := post(t, h, ask(0, "how can I reset my password?"), hdr); rec.Header().Get(cacheHeader) != "miss" { + t.Errorf("without an embedding model: %q", rec.Header().Get(cacheHeader)) + } + if e.calls.Load() != 2 || embeds.Load() != 0 { + t.Errorf("chat %d, embeddings %d", e.calls.Load(), embeds.Load()) + } +} + +func TestTheConversationAQuestionIsAskedInIsAllButItsLastMessage(t *testing.T) { + body := func(last string, earlier ...string) []byte { + var msgs []any + for _, s := range earlier { + msgs = append(msgs, map[string]any{"role": "user", "content": s}, map[string]any{"role": "assistant", "content": "ok"}) + } + msgs = append(msgs, map[string]any{"role": "user", "content": last}) + b, _ := json.Marshal(map[string]any{"model": "m", "messages": msgs, "temperature": 0}) + return b + } + g1, q1, ok1 := similarGroup("p", "/v1/chat/completions", "m", body("a?", "hi")) + g2, _, ok2 := similarGroup("p", "/v1/chat/completions", "m", body("b?", "hi")) + g3, _, _ := similarGroup("p", "/v1/chat/completions", "m", body("a?", "hello")) + g4, _, _ := similarGroup("q", "/v1/chat/completions", "m", body("a?", "hi")) + if !ok1 || !ok2 || q1 != "a?" || g1 != g2 { + t.Errorf("two questions after the same turns are in different conversations") + } + if g1 == g3 || g1 == g4 { + t.Errorf("other turns or another project share a conversation") + } + tool, _ := json.Marshal(map[string]any{"model": "m", "messages": []any{ + map[string]any{"role": "user", "content": "x"}, map[string]any{"role": "tool", "content": "42"}}}) + if _, _, ok := similarGroup("p", "/v1/chat/completions", "m", tool); ok { + t.Error("a tool's result is taken for a question") + } +} diff --git a/internal/store/answercache.go b/internal/store/answercache.go index 38fd454..70068ea 100644 --- a/internal/store/answercache.go +++ b/internal/store/answercache.go @@ -19,21 +19,29 @@ type AnswerCache struct { // Any reuses answers to requests sampled at a temperature above 0 // with no seed, which would otherwise each be asked again. Any bool + // Similar is the cosine similarity a chat question must reach to one + // already answered, in the same conversation so far, to be given that + // answer; 0 answers only the same request again. + Similar float64 } // 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 { +func (s *DB) SetAnswerCache(ctx context.Context, projectID string, c AnswerCache) error { var secs any - a := 0 - if ttl > 0 { - secs = max(int64(ttl/time.Second), 1) - if anyAnswer { + a, similar := 0, 0.0 + if c.TTL > 0 { + secs = max(int64(c.TTL/time.Second), 1) + if c.Any { a = 1 } + similar = c.Similar + } + if similar < 0 || similar > 1 { + return fmt.Errorf("store: similarity %v is not between 0 and 1", similar) } res, err := s.db.ExecContext(ctx, - `UPDATE projects SET cache_ttl_seconds = ?, cache_any = ? WHERE id = ?`, secs, a, projectID) + `UPDATE projects SET cache_ttl_seconds = ?, cache_any = ?, cache_similar = ? WHERE id = ?`, secs, a, similar, projectID) if err != nil { return err } @@ -47,13 +55,14 @@ func (s *DB) SetAnswerCache(ctx context.Context, projectID string, ttl time.Dura func (s *DB) AnswerCacheOf(ctx context.Context, projectID string) (AnswerCache, error) { var secs sql.NullInt64 var a int + var similar float64 err := s.db.QueryRowContext(ctx, - `SELECT cache_ttl_seconds, cache_any FROM projects WHERE id = ?`, projectID).Scan(&secs, &a) + `SELECT cache_ttl_seconds, cache_any, cache_similar FROM projects WHERE id = ?`, projectID).Scan(&secs, &a, &similar) 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 + return AnswerCache{TTL: time.Duration(secs.Int64) * time.Second, Any: a == 1, Similar: similar}, nil } diff --git a/internal/store/answercache_test.go b/internal/store/answercache_test.go index f447509..9b89f3e 100644 --- a/internal/store/answercache_test.go +++ b/internal/store/answercache_test.go @@ -16,19 +16,28 @@ func answerCacheRoundTrip(t *testing.T, s *DB) { 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 { + if err := s.SetAnswerCache(ctx, proj, AnswerCache{TTL: 90 * time.Minute, Any: 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 { + if err := s.SetAnswerCache(ctx, proj, AnswerCache{TTL: 0, Any: 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 { + if err := s.SetAnswerCache(ctx, proj, AnswerCache{TTL: time.Hour, Similar: 0.95}); err != nil { + t.Fatal(err) + } + if c, _ := s.AnswerCacheOf(ctx, proj); c.Similar != 0.95 { + t.Errorf("similar: %+v", c) + } + if err := s.SetAnswerCache(ctx, proj, AnswerCache{TTL: time.Hour, Similar: 1.5}); err == nil { + t.Error("a similarity above 1 was set") + } + if err := s.SetAnswerCache(ctx, "no-such-project", AnswerCache{TTL: time.Hour, Any: false}); err == nil { t.Error("a project that does not exist was set") } diff --git a/internal/store/migrations/postgres/0051_cache_similar.sql b/internal/store/migrations/postgres/0051_cache_similar.sql new file mode 100644 index 0000000..25f0f1d --- /dev/null +++ b/internal/store/migrations/postgres/0051_cache_similar.sql @@ -0,0 +1,6 @@ +-- Ported from the SQLite migration of the same number. +-- Answers to similar questions, for a project that asks: the cosine +-- similarity a question must reach to another's to be given its answer. +-- 0 is off, and only the same request is answered again. The questions and +-- answers are held in each gateway's memory, never here. +ALTER TABLE projects ADD COLUMN cache_similar DOUBLE PRECISION NOT NULL DEFAULT 0; diff --git a/internal/store/migrations/sqlite/0051_cache_similar.sql b/internal/store/migrations/sqlite/0051_cache_similar.sql new file mode 100644 index 0000000..7c0981e --- /dev/null +++ b/internal/store/migrations/sqlite/0051_cache_similar.sql @@ -0,0 +1,5 @@ +-- Answers to similar questions, for a project that asks: the cosine +-- similarity a question must reach to another's to be given its answer. +-- 0 is off, and only the same request is answered again. The questions and +-- answers are held in each gateway's memory, never here. +ALTER TABLE projects ADD COLUMN cache_similar REAL NOT NULL DEFAULT 0;