Files
glpi-ai-agent/internal/knowledge/persistent_index.go
groot 827c348e69
All checks were successful
release-tag / release-image (push) Successful in 1m40s
Persistenz und Performance optimiert
2026-07-29 08:56:04 +02:00

695 lines
20 KiB
Go

package knowledge
import (
"context"
"crypto/sha256"
"encoding/gob"
"encoding/hex"
"encoding/json"
"fmt"
"log/slog"
"os"
"path/filepath"
"sort"
"strings"
"time"
"github.com/example/glpi-ai-agent/internal/model"
)
const persistentSnapshotVersion = 1
type fileRecord struct {
Key string
Path string
ID string
Size int64
ModTimeUnixNano int64
RawHash string
Managed bool
Included bool
Ignored bool
Unmapped []string
}
type persistentSnapshot struct {
Version int
Fingerprint string
SavedAt time.Time
Docs []model.KnowledgeDoc
Files map[string]string
Managed map[string]bool
StaticDocs map[string]model.KnowledgeDoc
TitleVectors map[string][]float64
ChunkVectors map[string][][]float64
Chunks map[string][]string
Manifest map[string]fileRecord
LoadStats LoadStats
}
func (s *Store) indexFingerprint() string {
allowed := make([]string, 0, len(s.allowedSources))
for source := range s.allowedSources {
allowed = append(allowed, source)
}
sort.Strings(allowed)
mapKeys := make([]string, 0, len(s.categoryMap))
for k := range s.categoryMap {
mapKeys = append(mapKeys, k)
}
sort.Strings(mapKeys)
mapping := make([]struct {
Key string
IDs []int64
}, 0, len(mapKeys))
for _, k := range mapKeys {
mapping = append(mapping, struct {
Key string
IDs []int64
}{k, append([]int64(nil), s.categoryMap[k]...)})
}
payload := struct {
RAG bool
EmbeddingIdentity string
EmbeddingProfile string
ChunkWords int
ChunkOverlap int
MaxChunksPerDoc int
CategoryMode string
IgnoreGlobs []string
AllowedSources []string
CategoryMap any
}{
RAG: s.rag, EmbeddingIdentity: s.scoring.EmbeddingIdentity, EmbeddingProfile: s.scoring.EmbeddingProfile,
ChunkWords: s.scoring.ChunkWords, ChunkOverlap: s.scoring.ChunkOverlap, MaxChunksPerDoc: s.scoring.MaxChunksPerDoc,
CategoryMode: s.loadOptions.CategoryMode, IgnoreGlobs: append([]string(nil), s.loadOptions.IgnoreGlobs...), AllowedSources: allowed, CategoryMap: mapping,
}
b, _ := json.Marshal(payload)
h := sha256.Sum256(b)
return hex.EncodeToString(h[:])
}
func (s *Store) loadPersistentSnapshot() (bool, error) {
f, err := os.Open(s.snapshotPath)
if err != nil {
if os.IsNotExist(err) {
return false, nil
}
return false, fmt.Errorf("open persistent knowledge index: %w", err)
}
defer f.Close()
var snap persistentSnapshot
if err := gob.NewDecoder(f).Decode(&snap); err != nil {
return false, fmt.Errorf("decode persistent knowledge index: %w", err)
}
if snap.Version != persistentSnapshotVersion {
return false, fmt.Errorf("persistent knowledge index version %d is unsupported", snap.Version)
}
if snap.Fingerprint != s.indexFingerprint() {
return false, nil
}
if snap.Files == nil {
snap.Files = map[string]string{}
}
if snap.Managed == nil {
snap.Managed = map[string]bool{}
}
if snap.StaticDocs == nil {
snap.StaticDocs = map[string]model.KnowledgeDoc{}
}
if snap.TitleVectors == nil {
snap.TitleVectors = map[string][]float64{}
}
if snap.ChunkVectors == nil {
snap.ChunkVectors = map[string][][]float64{}
}
if snap.Chunks == nil {
snap.Chunks = map[string][]string{}
}
if snap.Manifest == nil {
snap.Manifest = map[string]fileRecord{}
}
// Paths in a snapshot may come from another host/project location. Rebind
// them to the current knowledge roots using the stable manifest key.
reboundFiles := map[string]string{}
for key, rec := range snap.Manifest {
name := filepath.Base(key)
if strings.HasPrefix(key, "managed/") {
rec.Path = filepath.Join(s.managedDir, name)
} else {
rec.Path = filepath.Join(s.dir, name)
}
snap.Manifest[key] = rec
if rec.Included && rec.ID != "" {
if rec.Managed || reboundFiles[rec.ID] == "" {
reboundFiles[rec.ID] = rec.Path
}
}
}
snap.Files = reboundFiles
s.mu.Lock()
s.docs = snap.Docs
s.files = snap.Files
s.managed = snap.Managed
s.staticDocs = snap.StaticDocs
s.titleVectors = snap.TitleVectors
s.chunkVectors = snap.ChunkVectors
s.chunks = snap.Chunks
s.manifest = snap.Manifest
s.external = map[string]string{}
s.loadStats = snap.LoadStats
s.initStatus = InitStatus{
State: "ready", Phase: "ready-cache", TotalFiles: len(snap.Manifest), ProcessedFiles: len(snap.Manifest), LoadedDocs: len(snap.Docs),
IndexedDocs: len(snap.Docs), CacheHits: len(snap.Docs), PendingEmbeddings: 0, StartedAt: time.Now(), FinishedAt: time.Now(),
SnapshotLoaded: true, SnapshotPath: s.snapshotPath, SnapshotSavedAt: snap.SavedAt,
}
s.mu.Unlock()
slog.Info("persistent knowledge index loaded", "documents", len(snap.Docs), "files", len(snap.Manifest), "saved_at", snap.SavedAt, "path", s.snapshotPath)
return true, nil
}
func (s *Store) persistSnapshot() error {
if s == nil {
return nil
}
if err := os.MkdirAll(filepath.Dir(s.snapshotPath), 0o750); err != nil {
return err
}
s.mu.RLock()
localDocs := make([]model.KnowledgeDoc, 0, len(s.docs))
files := map[string]string{}
managed := map[string]bool{}
staticDocs := make(map[string]model.KnowledgeDoc, len(s.staticDocs))
titles := map[string][]float64{}
vectors := map[string][][]float64{}
chunks := map[string][]string{}
for id, d := range s.staticDocs {
staticDocs[id] = d
}
for _, d := range s.docs {
if s.external[d.ID] != "" {
continue
}
localDocs = append(localDocs, d)
if path := s.files[d.ID]; path != "" {
files[d.ID] = path
}
if s.managed[d.ID] {
managed[d.ID] = true
}
if v := s.titleVectors[d.ID]; len(v) > 0 {
titles[d.ID] = v
}
if vv := s.chunkVectors[d.ID]; len(vv) > 0 {
vectors[d.ID] = vv
}
if cc := s.chunks[d.ID]; len(cc) > 0 {
chunks[d.ID] = cc
}
}
manifest := make(map[string]fileRecord, len(s.manifest))
for k, v := range s.manifest {
manifest[k] = v
}
stats := s.loadStats
stats.UnmappedCategories = append([]string(nil), stats.UnmappedCategories...)
s.mu.RUnlock()
snap := persistentSnapshot{Version: persistentSnapshotVersion, Fingerprint: s.indexFingerprint(), SavedAt: time.Now(), Docs: localDocs, Files: files, Managed: managed, StaticDocs: staticDocs, TitleVectors: titles, ChunkVectors: vectors, Chunks: chunks, Manifest: manifest, LoadStats: stats}
tmp := s.snapshotPath + ".tmp"
f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o640)
if err != nil {
return err
}
encErr := gob.NewEncoder(f).Encode(&snap)
closeErr := f.Close()
if encErr != nil {
_ = os.Remove(tmp)
return encErr
}
if closeErr != nil {
_ = os.Remove(tmp)
return closeErr
}
if err := os.Rename(tmp, s.snapshotPath); err != nil {
_ = os.Remove(tmp)
return err
}
s.setInitStatus(func(st *InitStatus) {
st.SnapshotPath = s.snapshotPath
st.SnapshotSavedAt = snap.SavedAt
})
return nil
}
// StartIncrementalSync keeps a warm persistent index usable while source files
// are checked for changes in the background. The first delta scan runs
// immediately. A zero interval disables subsequent periodic scans.
func (s *Store) StartIncrementalSync(ctx context.Context, interval time.Duration) {
if s == nil || strings.EqualFold(s.scoring.IndexMode, "readonly") {
return
}
go func() {
s.syncLocalSafely(ctx)
if interval <= 0 {
return
}
t := time.NewTicker(interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
s.syncLocalSafely(ctx)
}
}
}()
}
func (s *Store) syncLocalSafely(ctx context.Context) {
if err := s.SyncLocal(ctx); err != nil {
slog.Error("incremental knowledge scan failed; keeping previous index", "error", err)
s.setInitStatus(func(st *InitStatus) { st.LastScanAt = time.Now(); st.LastScanError = err.Error(); st.Phase = "ready" })
}
}
// SyncLocal performs a metadata-first delta scan. Unchanged files are never
// opened or parsed. Files whose size/mtime changed are hashed; unchanged bytes
// are reused without parsing/embedding. Only genuinely changed retrieval text
// is sent to the embedding provider.
func (s *Store) SyncLocal(ctx context.Context) error {
if s == nil {
return fmt.Errorf("knowledge store is not initialized")
}
if strings.EqualFold(s.scoring.IndexMode, "readonly") {
return nil
}
s.initMu.Lock()
defer s.initMu.Unlock()
if !s.Ready() {
return fmt.Errorf("knowledge store is not ready")
}
s.setInitStatus(func(st *InitStatus) {
st.Phase = "syncing"
st.LastScanError = ""
st.ChangedFiles = 0
st.DeletedFiles = 0
st.ReusedFiles = 0
})
s.mu.RLock()
oldManifest := make(map[string]fileRecord, len(s.manifest))
for k, v := range s.manifest {
oldManifest[k] = v
}
oldDocs := make(map[string]model.KnowledgeDoc, len(s.docs))
for _, d := range s.docs {
oldDocs[d.ID] = d
}
oldStatic := make(map[string]model.KnowledgeDoc, len(s.staticDocs))
for k, v := range s.staticDocs {
oldStatic[k] = v
}
oldManaged := make(map[string]model.KnowledgeDoc)
for _, d := range s.docs {
if s.managed[d.ID] {
oldManaged[d.ID] = d
}
}
oldTitle := s.titleVectors
oldChunkVec := s.chunkVectors
oldChunks := s.chunks
externalDocs := make([]model.KnowledgeDoc, 0)
externalMap := make(map[string]string, len(s.external))
for id, src := range s.external {
externalMap[id] = src
}
for _, d := range s.docs {
if s.external[d.ID] != "" {
externalDocs = append(externalDocs, d)
}
}
s.mu.RUnlock()
resStatic, err := s.scanDeltaDir(ctx, s.dir, "static", false, s.loadOptions, oldManifest, oldStatic)
if err != nil {
return err
}
resManaged, err := s.scanDeltaDir(ctx, s.managedDir, "managed", true, LoadOptions{CategoryMode: "strict"}, oldManifest, oldManaged)
if err != nil {
return err
}
manifest := make(map[string]fileRecord, len(resStatic.manifest)+len(resManaged.manifest))
for k, v := range resStatic.manifest {
manifest[k] = v
}
for k, v := range resManaged.manifest {
manifest[k] = v
}
// Rebuild effective local documents. Managed entries intentionally override
// static entries with the same id, matching the original startup behavior.
staticDocs := resStatic.docs
managedDocs := resManaged.docs
ids := make([]string, 0, len(staticDocs)+len(managedDocs))
merged := map[string]model.KnowledgeDoc{}
for id, d := range staticDocs {
merged[id] = d
ids = append(ids, id)
}
for id, d := range managedDocs {
if _, ok := merged[id]; !ok {
ids = append(ids, id)
}
merged[id] = d
}
sort.Strings(ids)
localDocs := make([]model.KnowledgeDoc, 0, len(ids))
for _, id := range ids {
localDocs = append(localDocs, merged[id])
}
newTitle := make(map[string][]float64, len(localDocs)+len(externalDocs))
newChunkVec := make(map[string][][]float64, len(localDocs)+len(externalDocs))
newChunks := make(map[string][]string, len(localDocs)+len(externalDocs))
needEmbed := make([]model.KnowledgeDoc, 0)
reused := 0
for _, d := range localDocs {
parts := chunkText(d.Text, s.scoring.ChunkWords, s.scoring.ChunkOverlap, s.scoring.MaxChunksPerDoc)
newChunks[d.ID] = parts
if old, ok := oldDocs[d.ID]; ok && hashDoc(old, s.scoring) == hashDoc(d, s.scoring) && len(oldTitle[d.ID]) > 0 && len(oldChunkVec[d.ID]) == len(parts) {
newTitle[d.ID] = oldTitle[d.ID]
newChunkVec[d.ID] = oldChunkVec[d.ID]
reused++
} else if s.rag {
needEmbed = append(needEmbed, d)
}
}
if s.rag && len(needEmbed) > 0 {
if s.embedder == nil {
return fmt.Errorf("RAG is enabled but no embedding provider is configured")
}
const docsPerBatch = 20
for start := 0; start < len(needEmbed); start += docsPerBatch {
if err := ctx.Err(); err != nil {
return err
}
end := start + docsPerBatch
if end > len(needEmbed) {
end = len(needEmbed)
}
emb, err := s.embedDocuments(ctx, needEmbed[start:end])
if err != nil {
return err
}
for _, d := range needEmbed[start:end] {
newTitle[d.ID] = emb[d.ID].title
newChunkVec[d.ID] = emb[d.ID].chunks
}
}
}
// Preserve connector-backed documents and vectors; their own syncers manage them.
for _, d := range externalDocs {
if _, collision := merged[d.ID]; collision {
return fmt.Errorf("local knowledge id %q collides with external source %q", d.ID, externalMap[d.ID])
}
newTitle[d.ID] = oldTitle[d.ID]
newChunkVec[d.ID] = oldChunkVec[d.ID]
newChunks[d.ID] = oldChunks[d.ID]
}
files := map[string]string{}
managedMap := map[string]bool{}
for _, rec := range manifest {
if !rec.Included || rec.ID == "" {
continue
}
if rec.Managed {
managedMap[rec.ID] = true
files[rec.ID] = rec.Path
} else if !managedMap[rec.ID] {
files[rec.ID] = rec.Path
}
}
stats := mergeLoadStats(resStatic.stats, resManaged.stats)
allDocs := append(localDocs, externalDocs...)
changed := resStatic.changed + resManaged.changed
deleted := countDeleted(oldManifest, manifest)
s.mu.Lock()
s.docs = allDocs
s.files = files
s.managed = managedMap
s.staticDocs = staticDocs
s.titleVectors = newTitle
s.chunkVectors = newChunkVec
s.chunks = newChunks
s.manifest = manifest
s.loadStats = stats
s.initStatus.State = "ready"
s.initStatus.Phase = "ready"
s.initStatus.TotalFiles = len(manifest)
s.initStatus.ProcessedFiles = len(manifest)
s.initStatus.LoadedDocs = len(localDocs)
s.initStatus.IndexedDocs = len(localDocs)
s.initStatus.PendingEmbeddings = 0
s.initStatus.LastScanAt = time.Now()
s.initStatus.ChangedFiles = changed
s.initStatus.DeletedFiles = deleted
s.initStatus.ReusedFiles = reused
s.initStatus.LastScanError = ""
s.mu.Unlock()
if changed > 0 || deleted > 0 {
if err := s.persistSnapshot(); err != nil {
return fmt.Errorf("persist incremental knowledge index: %w", err)
}
}
slog.Info("incremental knowledge scan complete", "files", len(manifest), "documents", len(localDocs), "changed_files", changed, "deleted_files", deleted, "reused_vectors", reused, "embedded_documents", len(needEmbed))
return nil
}
type deltaScanResult struct {
docs map[string]model.KnowledgeDoc
manifest map[string]fileRecord
stats LoadStats
changed int
}
func (s *Store) scanDeltaDir(ctx context.Context, dir, origin string, managed bool, opts LoadOptions, old map[string]fileRecord, oldDocs map[string]model.KnowledgeDoc) (deltaScanResult, error) {
res := deltaScanResult{docs: map[string]model.KnowledgeDoc{}, manifest: map[string]fileRecord{}}
entries, err := os.ReadDir(dir)
if err != nil {
return res, fmt.Errorf("read knowledge directory %q: %w", dir, err)
}
for _, e := range entries {
if err := ctx.Err(); err != nil {
return res, err
}
if e.IsDir() || !strings.HasSuffix(strings.ToLower(e.Name()), ".json") {
continue
}
key := origin + "/" + e.Name()
path := filepath.Join(dir, e.Name())
info, err := e.Info()
if err != nil {
return res, err
}
if matchesAnyGlob(e.Name(), opts.IgnoreGlobs) {
res.manifest[key] = fileRecord{Key: key, Path: path, Size: info.Size(), ModTimeUnixNano: info.ModTime().UnixNano(), Managed: managed, Ignored: true}
res.stats.IgnoredFiles++
continue
}
if prev, ok := old[key]; ok && prev.Size == info.Size() && prev.ModTimeUnixNano == info.ModTime().UnixNano() {
prev.Path = path
res.manifest[key] = prev
if prev.Included && prev.ID != "" {
if d, ok := oldDocs[prev.ID]; ok {
res.docs[prev.ID] = d
}
}
for _, u := range prev.Unmapped {
res.stats.UnmappedCategories = appendUniqueString(res.stats.UnmappedCategories, u)
}
if len(prev.Unmapped) > 0 {
res.stats.UnmappedCategoryFiles++
}
if prev.Ignored {
res.stats.IgnoredFiles++
}
continue
}
b, err := os.ReadFile(path)
if err != nil {
return res, err
}
h := sha256.Sum256(b)
rawHash := hex.EncodeToString(h[:])
if prev, ok := old[key]; ok && prev.RawHash != "" && prev.RawHash == rawHash {
prev.Path = path
prev.Size = info.Size()
prev.ModTimeUnixNano = info.ModTime().UnixNano()
res.manifest[key] = prev
if prev.Included && prev.ID != "" {
if d, ok := oldDocs[prev.ID]; ok {
res.docs[prev.ID] = d
}
}
continue
}
d, unmapped, skip, err := decodeKnowledgeDoc(b, opts.CategoryMode, s.categoryMap)
if err != nil {
return res, fmt.Errorf("%s: %w", e.Name(), err)
}
rec := fileRecord{Key: key, Path: path, Size: info.Size(), ModTimeUnixNano: info.ModTime().UnixNano(), RawHash: rawHash, Managed: managed, Unmapped: append([]string(nil), unmapped...)}
if len(unmapped) > 0 {
res.stats.UnmappedCategoryFiles++
res.stats.UnmappedCategories = mergeStrings(res.stats.UnmappedCategories, unmapped)
}
if skip {
rec.Ignored = true
res.stats.IgnoredFiles++
res.manifest[key] = rec
res.changed++
continue
}
if d.ID == "" || d.Title == "" {
return res, fmt.Errorf("%s: id/title required", e.Name())
}
if !safeID(d.ID) {
return res, fmt.Errorf("%s: invalid id %q", e.Name(), d.ID)
}
d.Source = strings.ToLower(strings.TrimSpace(d.Source))
if d.Source == "" {
return res, fmt.Errorf("%s: source required", e.Name())
}
if _, allowed := s.allowedSources[d.Source]; !allowed {
res.manifest[key] = rec
res.changed++
continue
}
d.Language = strings.TrimSpace(d.Language)
d.CommunicationStyle = strings.ToLower(strings.TrimSpace(d.CommunicationStyle))
rec.ID = d.ID
rec.Included = true
res.manifest[key] = rec
if _, dup := res.docs[d.ID]; dup {
return res, fmt.Errorf("duplicate knowledge id %q in %s", d.ID, dir)
}
res.docs[d.ID] = d
res.changed++
}
return res, nil
}
func countDeleted(old, cur map[string]fileRecord) int {
n := 0
for k := range old {
if _, ok := cur[k]; !ok {
n++
}
}
return n
}
func mergeLoadStats(a, b LoadStats) LoadStats {
return LoadStats{IgnoredFiles: a.IgnoredFiles + b.IgnoredFiles, UnmappedCategoryFiles: a.UnmappedCategoryFiles + b.UnmappedCategoryFiles, UnmappedCategories: mergeStrings(a.UnmappedCategories, b.UnmappedCategories)}
}
func buildManifestForDocs(docs []model.KnowledgeDoc, files []string, origin string, managed bool) (map[string]fileRecord, error) {
out := make(map[string]fileRecord, len(docs))
for i, d := range docs {
if i >= len(files) {
return nil, fmt.Errorf("knowledge manifest mismatch: %d docs, %d files", len(docs), len(files))
}
path := files[i]
info, err := os.Stat(path)
if err != nil {
return nil, err
}
b, err := os.ReadFile(path)
if err != nil {
return nil, err
}
h := sha256.Sum256(b)
key := origin + "/" + filepath.Base(path)
out[key] = fileRecord{Key: key, Path: path, ID: d.ID, Size: info.Size(), ModTimeUnixNano: info.ModTime().UnixNano(), RawHash: hex.EncodeToString(h[:]), Managed: managed, Included: true, Unmapped: append([]string(nil), d.UnmappedExternalCategories...)}
}
return out, nil
}
func (s *Store) persistExternalVectorCache(source string) error {
if s == nil || !s.rag {
return nil
}
if err := os.MkdirAll(filepath.Dir(s.externalCachePath), 0o750); err != nil {
return err
}
s.mu.RLock()
cf := cacheFile{Version: 3, Hashes: map[string]string{}, TitleVectors: map[string][]float64{}, ChunkVectors: map[string][][]float64{}}
for _, d := range s.docs {
if s.external[d.ID] != source || len(s.titleVectors[d.ID]) == 0 {
continue
}
cf.Hashes[d.ID] = hashDoc(d, s.scoring)
cf.TitleVectors[d.ID] = append([]float64(nil), s.titleVectors[d.ID]...)
cf.ChunkVectors[d.ID] = cloneChunkVectors(s.chunkVectors[d.ID])
}
s.mu.RUnlock()
b, err := json.Marshal(cf)
if err != nil {
return err
}
tmp := s.externalCachePath + ".tmp"
if err := os.WriteFile(tmp, b, 0o640); err != nil {
return err
}
if err := os.Rename(tmp, s.externalCachePath); err != nil {
_ = os.Remove(tmp)
return err
}
return nil
}
// augmentManifestAllFiles records even filtered/ignored JSON files so periodic
// delta scans do not repeatedly open files that are intentionally not part of
// the active corpus.
func augmentManifestAllFiles(dir, origin string, managed bool, opts LoadOptions, manifest map[string]fileRecord) error {
entries, err := os.ReadDir(dir)
if err != nil {
return err
}
for _, e := range entries {
if e.IsDir() || !strings.HasSuffix(strings.ToLower(e.Name()), ".json") {
continue
}
key := origin + "/" + e.Name()
if _, ok := manifest[key]; ok {
continue
}
info, err := e.Info()
if err != nil {
return err
}
path := filepath.Join(dir, e.Name())
rec := fileRecord{Key: key, Path: path, Size: info.Size(), ModTimeUnixNano: info.ModTime().UnixNano(), Managed: managed, Included: false, Ignored: matchesAnyGlob(e.Name(), opts.IgnoreGlobs)}
if !rec.Ignored {
b, err := os.ReadFile(path)
if err != nil {
return err
}
h := sha256.Sum256(b)
rec.RawHash = hex.EncodeToString(h[:])
}
manifest[key] = rec
}
return nil
}