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 <noreply@anthropic.com>
This commit is contained in:
1 parent
ed0d1ec0ad
commit
3a3db8cd20
19 files changed
+514
-33
No files matched your search
+7
-1
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
+3
-1
@@ -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.
|
||||
|
||||
|
||||
+40
-2
@@ -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
|
||||
|
||||
@@ -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`
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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: ""},
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
Reference in new issue
Block a user