jbergner
94dbd4ccab
All checks were successful
release-tag / release-image (push) Successful in 2m32s
330 lines
14 KiB
Go
330 lines
14 KiB
Go
package engine
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"sort"
|
|
"time"
|
|
|
|
"github.com/local/glpi-neural-brain/internal/graph"
|
|
"github.com/local/glpi-neural-brain/internal/model"
|
|
"github.com/local/glpi-neural-brain/internal/sourceagent"
|
|
"github.com/local/glpi-neural-brain/internal/vectorgraph"
|
|
)
|
|
|
|
type vectorLayerExecution struct {
|
|
Stats graph.VectorSemanticLayerStats
|
|
Mutations graph.MutationStats
|
|
Offloaded bool
|
|
AgentID string
|
|
ComputeMS int64
|
|
FallbackReason string
|
|
LayoutMode string
|
|
LayoutDue bool
|
|
}
|
|
|
|
func (e *Engine) vectorLayerConfig() graph.VectorSemanticLayerConfig {
|
|
workers := 1
|
|
if settings := e.RuntimeSettings(); settings.SpeedMode {
|
|
workers = e.effectiveSpeedCPUWorkers(settings)
|
|
}
|
|
return graph.VectorSemanticLayerConfig{
|
|
Workers: workers, Neighbors: e.Cfg.VectorGraphNeighbors, CandidateLimit: e.Cfg.VectorGraphCandidates,
|
|
HashBits: e.Cfg.ClusterHashBits, HashTables: e.Cfg.ClusterHashTables,
|
|
MinSimilarity: e.Cfg.VectorGraphMinSimilarity, MinAffinity: e.Cfg.VectorGraphMinAffinity,
|
|
Layout: e.Cfg.VectorGraphLayout, LayoutRelax: false, LayoutBlend: e.Cfg.VectorGraphLayoutBlend, LayoutMaxShift: e.Cfg.VectorGraphLayoutMaxShift,
|
|
OrphanPass: e.Cfg.VectorGraphOrphanPass, OrphanNeighbors: e.Cfg.VectorGraphOrphanNeighbors,
|
|
OrphanCandidateLimit: e.Cfg.VectorGraphOrphanCandidates,
|
|
OrphanMinSimilarity: e.Cfg.VectorGraphOrphanMinSimilarity, OrphanMinAffinity: e.Cfg.VectorGraphOrphanMinAffinity,
|
|
}
|
|
}
|
|
|
|
func (e *Engine) vectorPrimaryConfig(layout bool) vectorgraph.Config {
|
|
workers := 1
|
|
if settings := e.RuntimeSettings(); settings.SpeedMode {
|
|
workers = e.effectiveSpeedCPUWorkers(settings)
|
|
}
|
|
return vectorgraph.Config{
|
|
Workers: workers, Neighbors: e.Cfg.VectorGraphNeighbors, CandidateLimit: e.Cfg.VectorGraphCandidates,
|
|
HashBits: e.Cfg.ClusterHashBits, HashTables: e.Cfg.ClusterHashTables, BandBits: 8,
|
|
MinSimilarity: e.Cfg.VectorGraphMinSimilarity, MinAffinity: e.Cfg.VectorGraphMinAffinity,
|
|
Layout: layout, Smoothing: .22,
|
|
}
|
|
}
|
|
|
|
func (e *Engine) vectorOrphanConfig() vectorgraph.Config {
|
|
workers := 1
|
|
if settings := e.RuntimeSettings(); settings.SpeedMode {
|
|
workers = e.effectiveSpeedCPUWorkers(settings)
|
|
}
|
|
return vectorgraph.Config{
|
|
Workers: workers, Neighbors: e.Cfg.VectorGraphOrphanNeighbors, CandidateLimit: e.Cfg.VectorGraphOrphanCandidates,
|
|
HashBits: e.Cfg.ClusterHashBits, HashTables: e.Cfg.ClusterHashTables, BandBits: 8,
|
|
MinSimilarity: e.Cfg.VectorGraphOrphanMinSimilarity, MinAffinity: e.Cfg.VectorGraphOrphanMinAffinity,
|
|
Layout: false,
|
|
}
|
|
}
|
|
|
|
func vectorAgentStartupGrace(initialBootstrap bool, configuredWait time.Duration) time.Duration {
|
|
if !initialBootstrap || configuredWait <= 0 {
|
|
return 0
|
|
}
|
|
return configuredWait
|
|
}
|
|
|
|
func (e *Engine) checkOnlineVectorComputeAgent(ctx context.Context) (bool, error) {
|
|
checkCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
|
defer cancel()
|
|
return e.SourceInbox.HasOnlineComputeAgent(checkCtx, sourceagent.ComputeKindVectorGraph, 3*time.Minute)
|
|
}
|
|
|
|
func (e *Engine) waitForOnlineVectorComputeAgent(ctx context.Context, wait time.Duration) (bool, error) {
|
|
if wait <= 0 {
|
|
return e.checkOnlineVectorComputeAgent(ctx)
|
|
}
|
|
deadline := time.Now().Add(wait)
|
|
for {
|
|
hasAgent, err := e.checkOnlineVectorComputeAgent(ctx)
|
|
if err != nil || hasAgent {
|
|
return hasAgent, err
|
|
}
|
|
remaining := time.Until(deadline)
|
|
if remaining <= 0 {
|
|
return false, nil
|
|
}
|
|
delay := time.Second
|
|
if remaining < delay {
|
|
delay = remaining
|
|
}
|
|
timer := time.NewTimer(delay)
|
|
select {
|
|
case <-ctx.Done():
|
|
if !timer.Stop() {
|
|
<-timer.C
|
|
}
|
|
return false, ctx.Err()
|
|
case <-timer.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (e *Engine) rebuildVectorSemanticLayer(ctx context.Context, runID string, filter graph.NodeFilter) (vectorLayerExecution, error) {
|
|
cfg := e.vectorLayerConfig()
|
|
e.stateMu.RLock()
|
|
lastVectorLayout := e.lastVectorLayout
|
|
e.stateMu.RUnlock()
|
|
layoutDue := e.Cfg.VectorGraphLayout || (e.Cfg.VectorGraphRelaxLayout && (lastVectorLayout.IsZero() || time.Since(lastVectorLayout) >= e.Cfg.VectorGraphLayoutRelaxInterval))
|
|
cfg.Layout = layoutDue
|
|
cfg.LayoutRelax = !e.Cfg.VectorGraphLayout && layoutDue
|
|
layoutMode := "off"
|
|
if e.Cfg.VectorGraphLayout && layoutDue {
|
|
layoutMode = "full"
|
|
} else if cfg.LayoutRelax {
|
|
layoutMode = "relax"
|
|
}
|
|
if !e.Cfg.VectorGraphAgentOffload || e.SourceInbox == nil {
|
|
primary, orphan, focus := e.Graph.BuildVectorSemanticLayerLocal(cfg, filter)
|
|
stats, mutations := e.Graph.ApplyVectorSemanticLayer(cfg, primary, orphan, focus)
|
|
e.noteVectorLayoutApplied(layoutDue, stats.PositionUpdates)
|
|
return vectorLayerExecution{Stats: stats, Mutations: mutations, LayoutMode: layoutMode, LayoutDue: layoutDue}, nil
|
|
}
|
|
|
|
hasAgent, availabilityErr := e.checkOnlineVectorComputeAgent(ctx)
|
|
startupGrace := vectorAgentStartupGrace(!e.bootstrapIsComplete(), e.Cfg.VectorGraphAgentWait)
|
|
if e.SpeedModeEnabled() && !e.Cfg.VectorGraphAgentRequired {
|
|
startupGrace = 0
|
|
}
|
|
if availabilityErr == nil && !hasAgent && startupGrace > 0 {
|
|
_ = e.requestControllerComputeCapacity(ctx, sourceagent.ComputeKindVectorGraph)
|
|
e.Broker.Publish(model.Activity{Type: "vector.graph.agent.waiting", Source: "brain", Phase: "semantic-linking", Message: fmt.Sprintf("Initialer Vector-Graph wartet bis zu %s auf die Compute-Agent-Registrierung", startupGrace), Strength: .34, Metadata: map[string]any{"run_id": runID, "kind": sourceagent.ComputeKindVectorGraph, "startup_grace": startupGrace.String()}})
|
|
hasAgent, availabilityErr = e.waitForOnlineVectorComputeAgent(ctx, startupGrace)
|
|
}
|
|
if availabilityErr != nil || !hasAgent {
|
|
reason := "no_compute_agent"
|
|
if startupGrace > 0 && availabilityErr == nil {
|
|
reason = "no_compute_agent_after_startup_grace"
|
|
}
|
|
if availabilityErr != nil {
|
|
reason = "compute_agent_check_failed: " + availabilityErr.Error()
|
|
}
|
|
if e.Cfg.VectorGraphAgentRequired {
|
|
return vectorLayerExecution{FallbackReason: reason, LayoutMode: layoutMode, LayoutDue: layoutDue}, fmt.Errorf("vector graph agent offload required but unavailable: %s", reason)
|
|
}
|
|
primary, orphan, focus := e.Graph.BuildVectorSemanticLayerLocal(cfg, filter)
|
|
stats, mutations := e.Graph.ApplyVectorSemanticLayer(cfg, primary, orphan, focus)
|
|
e.noteVectorLayoutApplied(layoutDue, stats.PositionUpdates)
|
|
return vectorLayerExecution{Stats: stats, Mutations: mutations, FallbackReason: reason, LayoutMode: layoutMode, LayoutDue: layoutDue}, nil
|
|
}
|
|
|
|
entries := e.Graph.VectorSemanticEntries(filter)
|
|
baseOrphans := e.Graph.KnowledgeOrphanIDsIgnoringOrigin(filter, graph.VectorMathOrigin)
|
|
inputVersion := e.Graph.Version()
|
|
request := sourceagent.VectorGraphComputeRequest{
|
|
Header: sourceagent.VectorGraphComputeHeader{
|
|
GraphVersion: inputVersion, Primary: e.vectorPrimaryConfig(layoutDue),
|
|
OrphanPass: e.Cfg.VectorGraphOrphanPass, Orphan: e.vectorOrphanConfig(), OrphanFocusIDs: baseOrphans,
|
|
},
|
|
Entries: entries,
|
|
}
|
|
e.Broker.Publish(model.Activity{Type: "vector.graph.agent.queued", Source: "brain", Phase: "semantic-linking", Message: fmt.Sprintf("Vector-Graph-CPU-Job für Agent bereitgestellt · %d Embeddings", len(entries)), Strength: .46, Metadata: map[string]any{
|
|
"run_id": runID, "kind": sourceagent.ComputeKindVectorGraph, "indexed": len(entries), "orphan_focus_candidates": len(baseOrphans), "graph_version": inputVersion, "cpu_workers": request.Header.Primary.Workers, "speed_mode": e.SpeedModeEnabled(),
|
|
}})
|
|
jobCtx, cancelJob := context.WithTimeout(ctx, e.Cfg.VectorGraphAgentWait)
|
|
result, err := e.SourceInbox.SubmitVectorGraphJob(jobCtx, request)
|
|
cancelJob()
|
|
if err == nil {
|
|
err = validateAgentVectorGraphResult(request, result)
|
|
}
|
|
if err == nil && e.Graph.Version() != inputVersion {
|
|
err = errors.New("graph changed while vector compute job was running")
|
|
}
|
|
if err != nil {
|
|
reason := err.Error()
|
|
e.Broker.Publish(model.Activity{Type: "vector.graph.agent.fallback", Source: "brain", Phase: "semantic-linking", Message: "Agent-Vectorjob konnte nicht sicher übernommen werden; lokale CPU-Berechnung wird verwendet", Strength: .38, Metadata: map[string]any{"run_id": runID, "reason": reason, "graph_version": inputVersion}})
|
|
if e.Cfg.VectorGraphAgentRequired {
|
|
return vectorLayerExecution{FallbackReason: reason, LayoutMode: layoutMode, LayoutDue: layoutDue}, err
|
|
}
|
|
primary, orphan, focus := e.Graph.BuildVectorSemanticLayerLocal(cfg, filter)
|
|
stats, mutations := e.Graph.ApplyVectorSemanticLayer(cfg, primary, orphan, focus)
|
|
e.noteVectorLayoutApplied(layoutDue, stats.PositionUpdates)
|
|
return vectorLayerExecution{Stats: stats, Mutations: mutations, FallbackReason: reason, LayoutMode: layoutMode, LayoutDue: layoutDue}, nil
|
|
}
|
|
|
|
focus := remainingOrphanFocus(baseOrphans, result.Primary.Links)
|
|
stats, mutations := e.Graph.ApplyVectorSemanticLayer(cfg, result.Primary, result.Orphan, focus)
|
|
e.noteVectorLayoutApplied(layoutDue, stats.PositionUpdates)
|
|
e.Broker.Publish(model.Activity{Type: "vector.graph.agent.completed", Source: "agent", Phase: "semantic-linking", Message: fmt.Sprintf("Agent hat Vector-Graph-CPU-Job abgeschlossen · %d + %d Kanten", len(result.Primary.Links), len(result.Orphan.Links)), Strength: .68, Metadata: withRunMutations(map[string]any{
|
|
"run_id": runID, "agent_id": result.AgentID, "compute_duration_ms": result.DurationMS, "indexed": len(entries),
|
|
"primary_links": len(result.Primary.Links), "orphan_links": len(result.Orphan.Links), "orphan_focus": len(focus), "no_model_call": true, "cpu_workers": request.Header.Primary.Workers, "speed_mode": e.SpeedModeEnabled(),
|
|
}, mutations)})
|
|
return vectorLayerExecution{Stats: stats, Mutations: mutations, Offloaded: true, AgentID: result.AgentID, ComputeMS: result.DurationMS, LayoutMode: layoutMode, LayoutDue: layoutDue}, nil
|
|
}
|
|
|
|
func (e *Engine) noteVectorLayoutApplied(layoutDue bool, updates uint64) {
|
|
if !layoutDue || updates == 0 {
|
|
return
|
|
}
|
|
e.stateMu.Lock()
|
|
e.lastVectorLayout = time.Now().UTC()
|
|
e.stateMu.Unlock()
|
|
}
|
|
|
|
func remainingOrphanFocus(base []string, primary []vectorgraph.Link) []string {
|
|
focus := make(map[string]bool, len(base))
|
|
for _, id := range base {
|
|
focus[id] = true
|
|
}
|
|
for _, link := range primary {
|
|
delete(focus, link.Source)
|
|
delete(focus, link.Target)
|
|
}
|
|
out := make([]string, 0, len(focus))
|
|
for id := range focus {
|
|
out = append(out, id)
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
func validateAgentVectorGraphResult(request sourceagent.VectorGraphComputeRequest, result sourceagent.VectorGraphComputeResult) error {
|
|
entries := request.Entries
|
|
graphVersion := request.Header.GraphVersion
|
|
if result.Kind != sourceagent.ComputeKindVectorGraph {
|
|
return fmt.Errorf("unexpected agent compute kind %q", result.Kind)
|
|
}
|
|
if result.GraphVersion != graphVersion {
|
|
return fmt.Errorf("stale agent graph version %d, expected %d", result.GraphVersion, graphVersion)
|
|
}
|
|
known := make(map[string]bool, len(entries))
|
|
for _, entry := range entries {
|
|
known[entry.ID] = true
|
|
}
|
|
validUnit := func(number float64) bool {
|
|
return !math.IsNaN(number) && !math.IsInf(number, 0) && number >= 0 && number <= 1.000001
|
|
}
|
|
validatePositions := func(name string, positions []vectorgraph.Position, allowed bool) error {
|
|
if !allowed && len(positions) > 0 {
|
|
return fmt.Errorf("%s result contains unexpected positions", name)
|
|
}
|
|
seen := make(map[string]bool, len(positions))
|
|
for _, position := range positions {
|
|
if !known[position.ID] || seen[position.ID] {
|
|
return fmt.Errorf("%s result contains unknown/duplicate position id", name)
|
|
}
|
|
seen[position.ID] = true
|
|
for _, number := range []float64{position.X, position.Y, position.Z} {
|
|
if math.IsNaN(number) || math.IsInf(number, 0) {
|
|
return fmt.Errorf("%s result contains invalid position", name)
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
validateLinks := func(name string, value vectorgraph.Result, maxLinks int, focus map[string]bool) error {
|
|
if value.Stats.Indexed != 0 && value.Stats.Indexed != len(entries) {
|
|
return fmt.Errorf("%s result indexed %d vectors, expected %d", name, value.Stats.Indexed, len(entries))
|
|
}
|
|
if maxLinks >= 0 && len(value.Links) > maxLinks {
|
|
return fmt.Errorf("%s result contains %d links, maximum expected %d", name, len(value.Links), maxLinks)
|
|
}
|
|
seen := make(map[string]bool, len(value.Links))
|
|
for _, link := range value.Links {
|
|
if !known[link.Source] || !known[link.Target] || link.Source == link.Target {
|
|
return fmt.Errorf("%s result contains unknown/invalid endpoint", name)
|
|
}
|
|
key := link.Source + "\x00" + link.Target
|
|
if link.Source > link.Target {
|
|
key = link.Target + "\x00" + link.Source
|
|
}
|
|
if seen[key] {
|
|
return fmt.Errorf("%s result contains duplicate link", name)
|
|
}
|
|
seen[key] = true
|
|
if focus != nil && !focus[link.Source] && !focus[link.Target] {
|
|
return fmt.Errorf("%s result contains link outside orphan focus", name)
|
|
}
|
|
for _, number := range []float64{link.Similarity, link.Affinity, link.Confidence} {
|
|
if !validUnit(number) {
|
|
return fmt.Errorf("%s result contains invalid numeric value", name)
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
primaryK := request.Header.Primary.Neighbors
|
|
if primaryK < 1 {
|
|
primaryK = 4
|
|
}
|
|
primaryMax := len(entries) * primaryK
|
|
if err := validateLinks("primary", result.Primary, primaryMax, nil); err != nil {
|
|
return err
|
|
}
|
|
if err := validatePositions("primary", result.Primary.Positions, request.Header.Primary.Layout); err != nil {
|
|
return err
|
|
}
|
|
|
|
if !request.Header.OrphanPass {
|
|
if len(result.Orphan.Links) > 0 || len(result.Orphan.Positions) > 0 {
|
|
return errors.New("orphan result returned although orphan pass is disabled")
|
|
}
|
|
return nil
|
|
}
|
|
focusIDs := remainingOrphanFocus(request.Header.OrphanFocusIDs, result.Primary.Links)
|
|
focus := make(map[string]bool, len(focusIDs))
|
|
for _, id := range focusIDs {
|
|
focus[id] = true
|
|
}
|
|
orphanK := request.Header.Orphan.Neighbors
|
|
if orphanK < 1 {
|
|
orphanK = 4
|
|
}
|
|
orphanMax := len(focus) * orphanK
|
|
if err := validateLinks("orphan", result.Orphan, orphanMax, focus); err != nil {
|
|
return err
|
|
}
|
|
return validatePositions("orphan", result.Orphan.Positions, false)
|
|
}
|