Merge the vector backfill
This commit is contained in:
kami
@@ -189,7 +189,9 @@ func (l *lockedAPI) MorningStatus(ctx context.Context) ([]ipc.MorningRoutineStat
|
||||
func run(args []string) error {
|
||||
cfgPath := flag.String("config", defaultConfigPath(), "path to mavend JSON config")
|
||||
wrappedKeyPath := flag.String("wrapped-key-file", "", "path to wrapped encryption key blob (enables cold-start unlock)")
|
||||
reembed := flag.Bool("reembed", false, "re-embed every stored note and fact with the configured embedder, then serve normally (run once after an embedder swap)")
|
||||
flag.CommandLine.Parse(args)
|
||||
reembedOnStart = *reembed
|
||||
cfg, err := config.Load(*cfgPath)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
+39
-3
@@ -1769,26 +1769,62 @@ func jsonStringImpl(s string) string {
|
||||
return string(b)
|
||||
}
|
||||
|
||||
// reembedOnStart is the -reembed flag (set in run()). Opt-in on purpose: see
|
||||
// runReembed.
|
||||
var reembedOnStart bool
|
||||
|
||||
// checkStoredEmbedder compares the embedder we just loaded with the one that
|
||||
// wrote the vectors already in the DB (Vikunja #378).
|
||||
//
|
||||
// The two models we have both make 384-dim vectors, so a size check catches
|
||||
// nothing: after a swap, recall silently compares vectors from different
|
||||
// spaces and the scores are noise. So we say it out loud. Nothing is changed
|
||||
// here — recall keeps running exactly as before until the backfill lands.
|
||||
// spaces and the scores are noise. So we say it out loud. Recall itself is not
|
||||
// changed here — the fix is `mavend -reembed`.
|
||||
func checkStoredEmbedder(dataStore *store.Store, emb router.Embedder) {
|
||||
if dataStore == nil {
|
||||
return
|
||||
}
|
||||
current := router.EmbedderID(emb)
|
||||
if reembedOnStart {
|
||||
runReembed(dataStore, emb, current)
|
||||
return
|
||||
}
|
||||
stored, mismatch, err := dataStore.CheckEmbedder(context.Background(), current)
|
||||
if err != nil {
|
||||
log.Printf("voice: embedder marker check failed: %v", err)
|
||||
return
|
||||
}
|
||||
if mismatch {
|
||||
log.Printf("voice: WARNING embedder MISMATCH — stored vectors were written by %q but the configured embedder is %q; recall scores are noise until the notes and facts are re-embedded (Vikunja #378)", stored, current)
|
||||
log.Printf("voice: WARNING embedder MISMATCH — stored vectors were written by %q but the configured embedder is %q; recall scores are noise until the notes and facts are re-embedded — run `mavend -reembed` once (Vikunja #378)", stored, current)
|
||||
return
|
||||
}
|
||||
log.Printf("voice: embedder marker ok (%s)", current)
|
||||
}
|
||||
|
||||
// runReembed is the one-shot backfill behind -reembed.
|
||||
//
|
||||
// Why a flag and not automatic on mismatch: the embedder is ONNX on the
|
||||
// laptop's CPU, so a few thousand notes is minutes of work. Doing that silently
|
||||
// inside a normal start would look like the daemon hanging on boot. So the user
|
||||
// runs it once, deliberately, after an embedder swap; the mismatch warning
|
||||
// above tells them to. It re-embeds, logs what it did, and then the daemon
|
||||
// carries on serving as usual — no separate binary, no second start needed.
|
||||
func runReembed(dataStore *store.Store, emb router.Embedder, current string) {
|
||||
log.Printf("voice: re-embedding stored notes and facts with %s — this can take a few minutes, do not interrupt", current)
|
||||
res, err := dataStore.ReembedAll(context.Background(), current,
|
||||
// EmbedPassage, not EmbedQuery: these are stored texts being searched
|
||||
// FOR, which is the side they were written with.
|
||||
func(ctx context.Context, text string) ([]float32, error) {
|
||||
return router.EmbedPassage(ctx, emb, text)
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("voice: re-embed FAILED, nothing was changed and no marker was written — safe to run again: %v", err)
|
||||
return
|
||||
}
|
||||
if res.Skipped {
|
||||
log.Printf("voice: re-embed skipped — the stored vectors were already written by %s", current)
|
||||
return
|
||||
}
|
||||
log.Printf("voice: re-embed done — %d notes in the notes table, %d notes and %d facts in the memory index, %d rows had no text to re-embed, took %s; stored vectors now belong to %s",
|
||||
res.Notes, res.MemNotes, res.Facts, res.NoText, res.Took.Round(time.Second), current)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,161 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// EmbedFunc embeds one piece of stored text. The caller passes
|
||||
// router.EmbedPassage — the STORE side of the query/passage asymmetry, which is
|
||||
// the side every vector in the DB was written with. (Passing the query side
|
||||
// would put the stored vectors in the wrong half of the space and quietly halve
|
||||
// recall.) A func instead of an interface keeps this package free of any
|
||||
// dependency on internal/router.
|
||||
type EmbedFunc func(ctx context.Context, text string) ([]float32, error)
|
||||
|
||||
// BackfillResult is what the re-embed run did, for logging.
|
||||
type BackfillResult struct {
|
||||
Skipped bool // marker already matched — nothing to do
|
||||
Notes int // rows rewritten in the notes table
|
||||
Facts int // fact rows rewritten in memory_vectors
|
||||
MemNotes int // note rows rewritten in memory_vectors
|
||||
NoText int // memory_vectors rows with no text in their meta, left alone
|
||||
Took time.Duration
|
||||
}
|
||||
|
||||
// ReembedAll rewrites every stored vector with the currently configured
|
||||
// embedder and then records that embedder as the one that owns the DB.
|
||||
//
|
||||
// Both places a vector lives are rewritten in the same pass: the `notes` table
|
||||
// `embedding` column and the `memory_vectors` rows (notes AND facts). Doing
|
||||
// only one would leave the two indexes disagreeing, which is worse than leaving
|
||||
// both stale.
|
||||
//
|
||||
// Safe to re-run: if the marker already names the current embedder there is
|
||||
// nothing to fix, so it returns immediately with Skipped set.
|
||||
//
|
||||
// Crash safety: everything — every vector and the marker — happens inside one
|
||||
// transaction. If anything fails or the process dies partway, the transaction
|
||||
// rolls back: no vectors changed and no marker written, so the next run does
|
||||
// the whole job again. The marker is never set unless the full rewrite
|
||||
// committed.
|
||||
func (s *Store) ReembedAll(ctx context.Context, currentID string, embed EmbedFunc) (BackfillResult, error) {
|
||||
start := time.Now()
|
||||
var res BackfillResult
|
||||
|
||||
stored, err := s.Meta(ctx, metaKeyEmbedderID)
|
||||
if err != nil {
|
||||
return res, err
|
||||
}
|
||||
if stored == currentID {
|
||||
res.Skipped = true
|
||||
res.Took = time.Since(start)
|
||||
return res, nil
|
||||
}
|
||||
|
||||
tx, err := s.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("reembed: begin: %w", err)
|
||||
}
|
||||
defer tx.Rollback() // no-op once committed
|
||||
|
||||
// ----- notes table -----
|
||||
type noteRow struct {
|
||||
id int64
|
||||
text string
|
||||
}
|
||||
var notes []noteRow
|
||||
rows, err := tx.QueryContext(ctx, `SELECT id, text FROM notes WHERE text != ''`)
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("reembed: read notes: %w", err)
|
||||
}
|
||||
for rows.Next() {
|
||||
var n noteRow
|
||||
if err := rows.Scan(&n.id, &n.text); err != nil {
|
||||
rows.Close()
|
||||
return res, fmt.Errorf("reembed: note row: %w", err)
|
||||
}
|
||||
notes = append(notes, n)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return res, fmt.Errorf("reembed: notes: %w", err)
|
||||
}
|
||||
|
||||
for _, n := range notes {
|
||||
vec, err := embed(ctx, n.text)
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("reembed: embed note %d: %w", n.id, err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx,
|
||||
`UPDATE notes SET embedding = ? WHERE id = ?`, floatsToBlob(vec), n.id); err != nil {
|
||||
return res, fmt.Errorf("reembed: write note %d: %w", n.id, err)
|
||||
}
|
||||
res.Notes++
|
||||
}
|
||||
|
||||
// ----- memory_vectors (the unified index: notes AND facts) -----
|
||||
// The text to re-embed is the one carried in the row's meta blob, which is
|
||||
// exactly the text that was embedded when the row was written.
|
||||
type vecRow struct {
|
||||
id, text, kind string
|
||||
}
|
||||
var vecs []vecRow
|
||||
rows, err = tx.QueryContext(ctx, `SELECT id, meta FROM memory_vectors`)
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("reembed: read memory vectors: %w", err)
|
||||
}
|
||||
for rows.Next() {
|
||||
var id, metaJSON string
|
||||
if err := rows.Scan(&id, &metaJSON); err != nil {
|
||||
rows.Close()
|
||||
return res, fmt.Errorf("reembed: memory row: %w", err)
|
||||
}
|
||||
meta := map[string]string{}
|
||||
if err := json.Unmarshal([]byte(metaJSON), &meta); err != nil {
|
||||
rows.Close()
|
||||
return res, fmt.Errorf("reembed: meta for %q: %w", id, err)
|
||||
}
|
||||
if meta["text"] == "" {
|
||||
res.NoText++
|
||||
continue
|
||||
}
|
||||
vecs = append(vecs, vecRow{id: id, text: meta["text"], kind: meta["type"]})
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return res, fmt.Errorf("reembed: memory vectors: %w", err)
|
||||
}
|
||||
|
||||
for _, v := range vecs {
|
||||
vec, err := embed(ctx, v.text)
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("reembed: embed %q: %w", v.id, err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx,
|
||||
`UPDATE memory_vectors SET vec = ? WHERE id = ?`, encodeVec(vec), v.id); err != nil {
|
||||
return res, fmt.Errorf("reembed: write %q: %w", v.id, err)
|
||||
}
|
||||
if v.kind == "fact" {
|
||||
res.Facts++
|
||||
} else {
|
||||
res.MemNotes++
|
||||
}
|
||||
}
|
||||
|
||||
// Same transaction as the rewrite, on purpose: the marker can only exist if
|
||||
// every vector above was written.
|
||||
if _, err := tx.ExecContext(ctx,
|
||||
`INSERT INTO meta (key, value) VALUES (?,?)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value`,
|
||||
metaKeyEmbedderID, currentID); err != nil {
|
||||
return res, fmt.Errorf("reembed: write marker: %w", err)
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return res, fmt.Errorf("reembed: commit: %w", err)
|
||||
}
|
||||
res.Took = time.Since(start)
|
||||
return res, nil
|
||||
}
|
||||
@@ -0,0 +1,171 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// markerVec is a recognisable vector: nothing in these tests writes it except
|
||||
// the backfill, so finding it proves the row really was rewritten.
|
||||
var markerVec = []float32{9, 9, 9}
|
||||
|
||||
func newEmbedder(calls *int) EmbedFunc {
|
||||
return func(_ context.Context, _ string) ([]float32, error) {
|
||||
*calls++
|
||||
return markerVec, nil
|
||||
}
|
||||
}
|
||||
|
||||
// seedOldVectors puts one note (notes table + unified index) and one fact
|
||||
// (unified index only) in the DB, both carrying obviously-old vectors.
|
||||
func seedOldVectors(t *testing.T, s *Store) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
old := []float32{0.1, 0.2, 0.3}
|
||||
id, err := s.WriteNote(ctx, time.Now(), "молоко в холодильнике", old, "voice")
|
||||
if err != nil {
|
||||
t.Fatalf("WriteNote: %v", err)
|
||||
}
|
||||
mem := s.VectorMemory()
|
||||
if err := mem.Insert(ctx, "note:1", old, map[string]string{
|
||||
"type": "note", "text": "молоко в холодильнике",
|
||||
}); err != nil {
|
||||
t.Fatalf("Insert note vector: %v", err)
|
||||
}
|
||||
if err := mem.Insert(ctx, "fact:water:1", old, map[string]string{
|
||||
"type": "fact", "text": "я пил воду",
|
||||
}); err != nil {
|
||||
t.Fatalf("Insert fact vector: %v", err)
|
||||
}
|
||||
_ = id
|
||||
}
|
||||
|
||||
func noteVec(t *testing.T, s *Store) []float32 {
|
||||
t.Helper()
|
||||
var blob []byte
|
||||
if err := s.db.QueryRow(`SELECT embedding FROM notes LIMIT 1`).Scan(&blob); err != nil {
|
||||
t.Fatalf("read note embedding: %v", err)
|
||||
}
|
||||
return blobToFloats(blob)
|
||||
}
|
||||
|
||||
func memVec(t *testing.T, s *Store, id string) []float32 {
|
||||
t.Helper()
|
||||
var blob []byte
|
||||
if err := s.db.QueryRow(`SELECT vec FROM memory_vectors WHERE id = ?`, id).Scan(&blob); err != nil {
|
||||
t.Fatalf("read memory vector %s: %v", id, err)
|
||||
}
|
||||
return decodeVec(blob)
|
||||
}
|
||||
|
||||
func sameVec(a, b []float32) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
for i := range a {
|
||||
if a[i] != b[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// The deployed case: old vectors everywhere, no marker. Every vector in both
|
||||
// places must be rewritten and the marker recorded.
|
||||
func TestReembedAllRewritesEveryVector(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
ctx := context.Background()
|
||||
seedOldVectors(t, s)
|
||||
|
||||
calls := 0
|
||||
res, err := s.ReembedAll(ctx, "multilingual-e5-small@384", newEmbedder(&calls))
|
||||
if err != nil {
|
||||
t.Fatalf("ReembedAll: %v", err)
|
||||
}
|
||||
if res.Skipped {
|
||||
t.Fatal("first run should not skip")
|
||||
}
|
||||
if res.Notes != 1 || res.MemNotes != 1 || res.Facts != 1 {
|
||||
t.Fatalf("counts: notes=%d memNotes=%d facts=%d", res.Notes, res.MemNotes, res.Facts)
|
||||
}
|
||||
if calls != 3 {
|
||||
t.Fatalf("embedder called %d times, want 3", calls)
|
||||
}
|
||||
if !sameVec(noteVec(t, s), markerVec) {
|
||||
t.Fatalf("notes table not rewritten: %v", noteVec(t, s))
|
||||
}
|
||||
if !sameVec(memVec(t, s, "note:1"), markerVec) {
|
||||
t.Fatal("unified index note row not rewritten")
|
||||
}
|
||||
if !sameVec(memVec(t, s, "fact:water:1"), markerVec) {
|
||||
t.Fatal("unified index fact row not rewritten")
|
||||
}
|
||||
got, err := s.Meta(ctx, metaKeyEmbedderID)
|
||||
if err != nil {
|
||||
t.Fatalf("Meta: %v", err)
|
||||
}
|
||||
if got != "multilingual-e5-small@384" {
|
||||
t.Fatalf("marker = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Re-running must do nothing at all — not a second pass over the same rows.
|
||||
func TestReembedAllSecondRunIsNoop(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
ctx := context.Background()
|
||||
seedOldVectors(t, s)
|
||||
|
||||
calls := 0
|
||||
if _, err := s.ReembedAll(ctx, "e5@384", newEmbedder(&calls)); err != nil {
|
||||
t.Fatalf("first run: %v", err)
|
||||
}
|
||||
first := calls
|
||||
|
||||
res, err := s.ReembedAll(ctx, "e5@384", newEmbedder(&calls))
|
||||
if err != nil {
|
||||
t.Fatalf("second run: %v", err)
|
||||
}
|
||||
if !res.Skipped {
|
||||
t.Fatal("second run should report Skipped")
|
||||
}
|
||||
if calls != first {
|
||||
t.Fatalf("second run embedded %d more rows, want 0", calls-first)
|
||||
}
|
||||
}
|
||||
|
||||
// A failure partway must leave the DB exactly as it was: no marker, and the old
|
||||
// vectors still in place (one transaction, rolled back).
|
||||
func TestReembedAllPartialFailureLeavesMarkerUnset(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
ctx := context.Background()
|
||||
seedOldVectors(t, s)
|
||||
before := noteVec(t, s)
|
||||
|
||||
calls := 0
|
||||
boom := func(_ context.Context, _ string) ([]float32, error) {
|
||||
calls++
|
||||
if calls == 2 {
|
||||
return nil, errors.New("onnx blew up")
|
||||
}
|
||||
return markerVec, nil
|
||||
}
|
||||
if _, err := s.ReembedAll(ctx, "e5@384", boom); err == nil {
|
||||
t.Fatal("expected an error")
|
||||
}
|
||||
got, err := s.Meta(ctx, metaKeyEmbedderID)
|
||||
if err != nil {
|
||||
t.Fatalf("Meta: %v", err)
|
||||
}
|
||||
if got != "" {
|
||||
t.Fatalf("marker was set to %q after a failed run", got)
|
||||
}
|
||||
if !sameVec(noteVec(t, s), before) {
|
||||
t.Fatal("a failed run left a partially rewritten notes table")
|
||||
}
|
||||
// And the mismatch warning must still fire, so the user knows to re-run.
|
||||
if _, mismatch, err := s.CheckEmbedder(ctx, "e5@384"); err != nil || !mismatch {
|
||||
t.Fatalf("CheckEmbedder after failed backfill: mismatch=%v err=%v", mismatch, err)
|
||||
}
|
||||
}
|
||||
@@ -60,9 +60,9 @@ const EmbedderUnknown = "unknown (written before this marker existed)"
|
||||
// marker was added to catch.
|
||||
// - marker absent and no vectors ⇒ fresh DB, claim it, nothing to fix.
|
||||
//
|
||||
// TODO(#378): on a mismatch, run the one-shot backfill here — re-embed every
|
||||
// stored note and fact text with the current embedder (EmbedPassage side),
|
||||
// write the vectors back, then SetMeta the current id.
|
||||
// On a mismatch the fix is ReembedAll (backfill.go), run explicitly with
|
||||
// `mavend -reembed`. Nothing is re-embedded here: that work is minutes of CPU
|
||||
// on the laptop and must not stall a normal start.
|
||||
func (s *Store) CheckEmbedder(ctx context.Context, currentID string) (stored string, mismatch bool, err error) {
|
||||
stored, err = s.Meta(ctx, metaKeyEmbedderID)
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user