+103
-70
@@ -3,13 +3,10 @@ package engine
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"math"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -35,11 +32,14 @@ var (
|
||||
)
|
||||
|
||||
type EnrichOutcome struct {
|
||||
Result string
|
||||
Candidate bool
|
||||
Created bool
|
||||
Rejected bool
|
||||
Comparisons int
|
||||
Result string
|
||||
Candidate bool
|
||||
Created bool
|
||||
RelationCreated bool
|
||||
ArticleCreated bool
|
||||
ArticleSkipped bool
|
||||
Rejected bool
|
||||
Comparisons int
|
||||
}
|
||||
|
||||
type Engine struct {
|
||||
@@ -52,24 +52,27 @@ type Engine struct {
|
||||
GLPIKB *ingest.GLPIKBSyncer
|
||||
Persistence *persist.Coordinator
|
||||
|
||||
mu sync.Mutex
|
||||
stateMu sync.RWMutex
|
||||
lastScan time.Time
|
||||
lastEnrich time.Time
|
||||
lastAttempt time.Time
|
||||
nextEnrich time.Time
|
||||
ollamaOK bool
|
||||
enrichRunning bool
|
||||
enrichTrigger string
|
||||
enrichResult string
|
||||
enrichError string
|
||||
enrichCycles uint64
|
||||
enrichCreated uint64
|
||||
enrichRejected uint64
|
||||
enrichRequests chan string
|
||||
runtimeMu sync.RWMutex
|
||||
runtime RuntimeSettings
|
||||
runtimePath string
|
||||
mu sync.Mutex
|
||||
stateMu sync.RWMutex
|
||||
lastScan time.Time
|
||||
lastEnrich time.Time
|
||||
lastAttempt time.Time
|
||||
nextEnrich time.Time
|
||||
ollamaOK bool
|
||||
enrichRunning bool
|
||||
enrichTrigger string
|
||||
enrichResult string
|
||||
enrichError string
|
||||
enrichCycles uint64
|
||||
enrichCreated uint64
|
||||
enrichRejected uint64
|
||||
relationsCreated uint64
|
||||
articlesCreated uint64
|
||||
articlesSkipped uint64
|
||||
enrichRequests chan string
|
||||
runtimeMu sync.RWMutex
|
||||
runtime RuntimeSettings
|
||||
runtimePath string
|
||||
}
|
||||
|
||||
func New(cfg config.Config, g *graph.Store, b *activity.Broker) *Engine {
|
||||
@@ -84,6 +87,27 @@ func New(cfg config.Config, g *graph.Store, b *activity.Broker) *Engine {
|
||||
if cfg.EnrichAnchors < 1 {
|
||||
cfg.EnrichAnchors = 48
|
||||
}
|
||||
if cfg.ArticleMinSources < 2 {
|
||||
cfg.ArticleMinSources = 3
|
||||
}
|
||||
if cfg.ArticleMaxSources < cfg.ArticleMinSources {
|
||||
cfg.ArticleMaxSources = 8
|
||||
}
|
||||
if cfg.ArticleMinProductionRatio == 0 {
|
||||
cfg.ArticleMinProductionRatio = .70
|
||||
}
|
||||
if cfg.ArticleMaxGenerationDepth < 1 {
|
||||
cfg.ArticleMaxGenerationDepth = 2
|
||||
}
|
||||
if cfg.ArticleMinConfidence == 0 {
|
||||
cfg.ArticleMinConfidence = .74
|
||||
}
|
||||
if cfg.ArticleMinTextChars < 1 {
|
||||
cfg.ArticleMinTextChars = 180
|
||||
}
|
||||
if cfg.ArticleMinAnswerChars < 1 {
|
||||
cfg.ArticleMinAnswerChars = 420
|
||||
}
|
||||
ollamaURLs := append([]string(nil), cfg.OllamaURLs...)
|
||||
if len(ollamaURLs) == 0 && strings.TrimSpace(cfg.OllamaURL) != "" {
|
||||
ollamaURLs = []string{cfg.OllamaURL}
|
||||
@@ -235,6 +259,7 @@ func (e *Engine) runEnrichmentCycle(ctx context.Context, trigger string) {
|
||||
e.Broker.Publish(model.Activity{Type: "think.cycle.started", Source: "brain", Phase: "autonomous", Message: fmt.Sprintf("Autonomer AI-THINK-Zyklus startet · bis zu %d sequenzielle Prüfungen", e.Cfg.EnrichBatchSize), Strength: .72, Metadata: map[string]any{"trigger": trigger, "batch_size": e.Cfg.EnrichBatchSize, "anchors": e.Cfg.EnrichAnchors}})
|
||||
|
||||
created, rejected, checked := 0, 0, 0
|
||||
relationsCreated, articlesCreated, articlesSkipped := 0, 0, 0
|
||||
result := "completed"
|
||||
var cycleErr error
|
||||
for step := 0; step < e.Cfg.EnrichBatchSize; step++ {
|
||||
@@ -258,6 +283,15 @@ func (e *Engine) runEnrichmentCycle(ctx context.Context, trigger string) {
|
||||
if outcome.Created {
|
||||
created++
|
||||
}
|
||||
if outcome.RelationCreated {
|
||||
relationsCreated++
|
||||
}
|
||||
if outcome.ArticleCreated {
|
||||
articlesCreated++
|
||||
}
|
||||
if outcome.ArticleSkipped {
|
||||
articlesSkipped++
|
||||
}
|
||||
if outcome.Rejected {
|
||||
rejected++
|
||||
}
|
||||
@@ -282,15 +316,18 @@ func (e *Engine) runEnrichmentCycle(ctx context.Context, trigger string) {
|
||||
}
|
||||
e.enrichCreated += uint64(created)
|
||||
e.enrichRejected += uint64(rejected)
|
||||
e.relationsCreated += uint64(relationsCreated)
|
||||
e.articlesCreated += uint64(articlesCreated)
|
||||
e.articlesSkipped += uint64(articlesSkipped)
|
||||
e.stateMu.Unlock()
|
||||
|
||||
metadata := map[string]any{"trigger": trigger, "checked": checked, "created": created, "rejected": rejected, "duration_ms": time.Since(started).Milliseconds(), "result": result}
|
||||
metadata := map[string]any{"trigger": trigger, "checked": checked, "created": created, "relations_created": relationsCreated, "articles_created": articlesCreated, "articles_skipped": articlesSkipped, "rejected": rejected, "duration_ms": time.Since(started).Milliseconds(), "result": result}
|
||||
if cycleErr != nil {
|
||||
e.Broker.Publish(model.Activity{Type: "think.cycle.failed", Source: "brain", Phase: "autonomous", Message: "AI-THINK-Zyklus wurde mit Fehler beendet", Strength: .45, Metadata: metadata})
|
||||
slog.Warn("enrichment cycle failed", "trigger", trigger, "error", cycleErr)
|
||||
return
|
||||
}
|
||||
message := fmt.Sprintf("AI-THINK-Zyklus abgeschlossen · %d geprüft · %d erstellt · %d verworfen", checked, created, rejected)
|
||||
message := fmt.Sprintf("AI-THINK-Zyklus abgeschlossen · %d Relationen · %d Artikel · %d verworfen", relationsCreated, articlesCreated, rejected)
|
||||
if result == "no_candidate" {
|
||||
message = "AI-THINK hat im aktuell geprüften Graphbereich keinen Kandidaten oberhalb des Schwellwerts gefunden"
|
||||
}
|
||||
@@ -525,8 +562,26 @@ func (e *Engine) EnrichOne(ctx context.Context) error {
|
||||
if !e.ThinkingEnabled() {
|
||||
return ErrThinkingDisabled
|
||||
}
|
||||
_, err := e.enrichOne(ctx, "direct")
|
||||
return err
|
||||
outcome, err := e.enrichOne(ctx, "direct")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
e.stateMu.Lock()
|
||||
if outcome.RelationCreated {
|
||||
e.relationsCreated++
|
||||
e.enrichCreated++
|
||||
}
|
||||
if outcome.ArticleCreated {
|
||||
e.articlesCreated++
|
||||
}
|
||||
if outcome.ArticleSkipped {
|
||||
e.articlesSkipped++
|
||||
}
|
||||
if outcome.Rejected {
|
||||
e.enrichRejected++
|
||||
}
|
||||
e.stateMu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Engine) enrichOne(ctx context.Context, trigger string) (EnrichOutcome, error) {
|
||||
@@ -547,7 +602,7 @@ func (e *Engine) enrichOne(ctx context.Context, trigger string) (EnrichOutcome,
|
||||
e.setOllamaOK(true)
|
||||
}
|
||||
|
||||
a, b, sim, ok, comparisons := e.Graph.NextPairFiltered(e.Cfg.SimilarityThreshold, e.Cfg.EnrichAnchors, e.thinkingCategories())
|
||||
a, b, sim, ok, comparisons := e.Graph.NextPairFilteredDepth(e.Cfg.SimilarityThreshold, e.Cfg.EnrichAnchors, e.thinkingCategories(), e.Cfg.ArticleMaxGenerationDepth)
|
||||
if !ok {
|
||||
e.stateMu.Lock()
|
||||
e.lastAttempt = time.Now().UTC()
|
||||
@@ -563,7 +618,7 @@ func (e *Engine) enrichOne(ctx context.Context, trigger string) (EnrichOutcome,
|
||||
e.stateMu.Unlock()
|
||||
e.Broker.Publish(model.Activity{Type: "think.started", Source: "brain", Phase: "association", NodeIDs: []string{a.ID, b.ID}, Message: fmt.Sprintf("Verwandtschaft wird geprüft · %.0f%% semantische Nähe", sim*100), Strength: .88, Metadata: map[string]any{"trigger": trigger, "semantic_similarity": sim, "source_label": a.Label, "target_label": b.Label, "model": e.Cfg.ChatModel, "candidate_comparisons": comparisons}})
|
||||
|
||||
system := "Analysiere zwei interne Wissenseinträge. Erfinde keine Fakten. Entscheide, ob eine belastbare Beziehung besteht. Wenn externe Fakten fehlen, setze needs_research=true. Gib ausschließlich JSON nach Schema zurück."
|
||||
system := "Du führst ausschließlich eine Relationserkennung für einen Wissensgraphen durch. Analysiere zwei interne Wissenseinträge, erfinde keine Fakten und entscheide, ob eine belastbare Beziehung besteht. Schreibe keinen Artikel und keine technische Synthese. Wenn externe Fakten zur Relationsentscheidung fehlen, setze needs_research=true. Gib ausschließlich JSON nach Schema zurück."
|
||||
var decision model.RelationDecision
|
||||
if err := e.Ollama.ChatJSON(ctx, system, relationContext(a, b, sim), relationSchema(), &decision); err != nil {
|
||||
e.Broker.Publish(model.Activity{Type: "think.failed", Source: "brain", Phase: "inference", NodeIDs: []string{a.ID, b.ID}, Message: "Qwen-Beziehungsanalyse ist fehlgeschlagen; es wurde nichts gespeichert", Strength: .35, Metadata: map[string]any{"trigger": trigger, "error": err.Error(), "model": e.Cfg.ChatModel}})
|
||||
@@ -603,12 +658,17 @@ func (e *Engine) enrichOne(ctx context.Context, trigger string) (EnrichOutcome,
|
||||
edge.ID = graph.EdgeID(edge.Source, edge.Target, edge.Type, edge.Origin)
|
||||
outcome := EnrichOutcome{Result: status, Candidate: true, Comparisons: comparisons}
|
||||
if status == "staging" {
|
||||
path, err := e.writeAIThink(a, b, decision, researchResults)
|
||||
if err != nil {
|
||||
return outcome, err
|
||||
}
|
||||
outcome.Created = true
|
||||
e.Broker.Publish(model.Activity{Type: "think.created", Source: "brain", Phase: "staging", NodeIDs: []string{a.ID, b.ID}, EdgeIDs: []string{edge.ID}, Message: "Neuer AI-THINK-Beitrag wurde erzeugt und für den gebündelten Staging-Schreibvorgang vorgemerkt", Strength: 1, Metadata: map[string]any{"trigger": trigger, "path": path, "relation_type": safeRelation(decision.RelationType), "confidence": decision.Confidence, "semantic_similarity": sim, "research_result_count": len(researchResults), "title": decision.Title, "write_pending": true}})
|
||||
outcome.RelationCreated = true
|
||||
e.Broker.Publish(model.Activity{Type: "think.relation.created", Source: "brain", Phase: "relation", NodeIDs: []string{a.ID, b.ID}, EdgeIDs: []string{edge.ID}, Message: "Belastbare Wissensrelation wurde als überprüfbare Graph-Edge übernommen", Strength: .86, Metadata: map[string]any{"trigger": trigger, "relation_type": safeRelation(decision.RelationType), "confidence": decision.Confidence, "semantic_similarity": sim, "research_result_count": len(researchResults), "topic_label": decision.TopicLabel}})
|
||||
articleOutcome, err := e.synthesizeKnowledgeArticle(ctx, trigger, []model.Node{a, b}, decision, researchResults)
|
||||
if err != nil {
|
||||
e.Broker.Publish(model.Activity{Type: "article.failed", Source: "brain", Phase: "knowledge-synthesis", NodeIDs: []string{a.ID, b.ID}, EdgeIDs: []string{edge.ID}, Message: "Die Relation bleibt erhalten, aber die Artikelsynthese ist fehlgeschlagen", Strength: .4, Metadata: map[string]any{"trigger": trigger, "error": err.Error()}})
|
||||
outcome.ArticleSkipped = true
|
||||
} else {
|
||||
outcome.ArticleCreated = articleOutcome.Created
|
||||
outcome.ArticleSkipped = articleOutcome.Skipped
|
||||
}
|
||||
} else {
|
||||
outcome.Rejected = true
|
||||
e.Broker.Publish(model.Activity{Type: "think.rejected", Source: "brain", Phase: "validation", NodeIDs: []string{a.ID, b.ID}, Message: "Ähnlichkeit geprüft, aber nicht als belastbare Edge übernommen", Strength: .42, Metadata: map[string]any{"trigger": trigger, "relation_type": safeRelation(decision.RelationType), "confidence": decision.Confidence, "semantic_similarity": sim, "explanation": decision.Explanation}})
|
||||
@@ -625,37 +685,6 @@ func (e *Engine) addResearch(a, b model.Node, results []model.ResearchResult) {
|
||||
e.Graph.UpsertEdge(model.Edge{Source: id, Target: b.ID, Type: "research_evidence", Origin: "research", Status: "staging", Confidence: .55, Weight: .4})
|
||||
}
|
||||
}
|
||||
func (e *Engine) writeAIThink(a, b model.Node, d model.RelationDecision, results []model.ResearchResult) (string, error) {
|
||||
if len(e.Cfg.StagingDirs) == 0 {
|
||||
return "", fmt.Errorf("no BRAIN_STAGING_DIRS configured")
|
||||
}
|
||||
dir := e.Cfg.StagingDirs[0]
|
||||
pair := a.ID + "\x00" + b.ID
|
||||
sum := sha256.Sum256([]byte(pair))
|
||||
short := strings.ToUpper(hex.EncodeToString(sum[:5]))
|
||||
now := time.Now().UTC()
|
||||
id := fmt.Sprintf("KB-AI-THINK-%s-%s", now.Format("20060102"), short)
|
||||
cats := unique(append([]string{"AI-THINK", "AI-Staging"}, common(a.Categories, b.Categories)...))
|
||||
keywords := unique(append(append([]string{}, d.Keywords...), first(a.Keywords, 4)...))
|
||||
keywords = unique(append(keywords, first(b.Keywords, 4)...))
|
||||
var evidence []map[string]any
|
||||
for _, r := range results {
|
||||
evidence = append(evidence, map[string]any{"title": r.Title, "url": r.URL, "excerpt": clamp(r.Content, 300)})
|
||||
}
|
||||
doc := map[string]any{"id": id, "title": nonempty(d.Title, "Zusammenhang: "+a.Label+" ↔ "+b.Label), "text": nonempty(d.Synthesis, d.Explanation), "answer": "Interne AI-THINK-Arbeitsnotiz. Vor produktiver Nutzung im Editor prüfen, korrigieren und freigeben.", "auto_reply": false, "min_score": 0.78, "categories": cats, "keywords": keywords, "source": "Neural Brain / " + e.Cfg.ChatModel + " (AI-THINK)", "source_uri": "brain://edge/" + short, "language": "de-DE", "communication_style": "formal", "ai_think": map[string]any{"status": "staging", "generated_at": now, "source_nodes": []string{a.ExternalID, b.ExternalID}, "source_node_ids": []string{a.ID, b.ID}, "relation_type": safeRelation(d.RelationType), "confidence": d.Confidence, "explanation": d.Explanation, "needs_research": d.NeedsResearch, "research_query": d.ResearchQuery, "research_evidence": evidence}}
|
||||
bts, err := json.MarshalIndent(doc, "", " ")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
path := filepath.Join(dir, strings.ToLower(id)+".json")
|
||||
if e.Persistence.Pending(path) {
|
||||
return path, nil
|
||||
}
|
||||
if _, err := os.Stat(path); err == nil {
|
||||
return path, nil
|
||||
}
|
||||
return e.Persistence.QueueFile(path, append(bts, '\n'), 0o640)
|
||||
}
|
||||
func (e *Engine) Status() map[string]any {
|
||||
s := e.Graph.Snapshot()
|
||||
e.stateMu.RLock()
|
||||
@@ -665,7 +694,11 @@ func (e *Engine) Status() map[string]any {
|
||||
"next_enrich": e.nextEnrich, "ollama_ok": e.ollamaOK, "auto_enrich": e.Cfg.AutoEnrich,
|
||||
"enrich_running": e.enrichRunning, "enrich_trigger": e.enrichTrigger, "enrich_result": e.enrichResult,
|
||||
"enrich_error": e.enrichError, "enrich_cycles": e.enrichCycles, "enrich_created": e.enrichCreated,
|
||||
"enrich_rejected": e.enrichRejected, "enrich_interval": e.Cfg.EnrichInterval.String(),
|
||||
"enrich_rejected": e.enrichRejected, "relations_created": e.relationsCreated, "articles_created": e.articlesCreated,
|
||||
"articles_skipped": e.articlesSkipped, "article_synthesis_enabled": e.Cfg.ArticleSynthesisEnabled,
|
||||
"article_min_sources": e.Cfg.ArticleMinSources, "article_max_sources": e.Cfg.ArticleMaxSources,
|
||||
"article_min_production_ratio": e.Cfg.ArticleMinProductionRatio, "article_max_generation_depth": e.Cfg.ArticleMaxGenerationDepth,
|
||||
"enrich_interval": e.Cfg.EnrichInterval.String(),
|
||||
"enrich_batch_size": e.Cfg.EnrichBatchSize, "enrich_anchors": e.Cfg.EnrichAnchors,
|
||||
"research_enabled": e.Cfg.ResearchEnabled, "chat_model": e.Cfg.ChatModel, "embedding_model": e.Cfg.EmbeddingModel,
|
||||
"ollama_pool": e.Ollama.PoolStatus(), "persistence": e.Persistence.Status(),
|
||||
@@ -708,7 +741,7 @@ func relationContext(a, b model.Node, sim float64) string {
|
||||
return fmt.Sprintf("SEMANTISCHE_NÄHE: %.4f\n\nA\nID: %s\nTitel: %s\nKategorien: %s\nInhalt: %s\n\nB\nID: %s\nTitel: %s\nKategorien: %s\nInhalt: %s", sim, a.ID, a.Label, strings.Join(a.Categories, ", "), a.Summary, b.ID, b.Label, strings.Join(b.Categories, ", "), b.Summary)
|
||||
}
|
||||
func relationSchema() map[string]any {
|
||||
return map[string]any{"type": "object", "properties": map[string]any{"related": map[string]any{"type": "boolean"}, "relation_type": map[string]any{"type": "string", "enum": []string{"related_to", "depends_on", "supports", "contradicts", "extends", "same_topic", "caused_by"}}, "confidence": map[string]any{"type": "number", "minimum": 0, "maximum": 1}, "explanation": map[string]any{"type": "string"}, "needs_research": map[string]any{"type": "boolean"}, "research_query": map[string]any{"type": "string"}, "title": map[string]any{"type": "string"}, "synthesis": map[string]any{"type": "string"}, "keywords": map[string]any{"type": "array", "items": map[string]any{"type": "string"}}}, "required": []string{"related", "relation_type", "confidence", "explanation", "needs_research", "research_query", "title", "synthesis", "keywords"}}
|
||||
return map[string]any{"type": "object", "properties": map[string]any{"related": map[string]any{"type": "boolean"}, "relation_type": map[string]any{"type": "string", "enum": []string{"related_to", "depends_on", "supports", "contradicts", "extends", "same_topic", "caused_by"}}, "confidence": map[string]any{"type": "number", "minimum": 0, "maximum": 1}, "explanation": map[string]any{"type": "string"}, "needs_research": map[string]any{"type": "boolean"}, "research_query": map[string]any{"type": "string"}, "topic_label": map[string]any{"type": "string"}, "keywords": map[string]any{"type": "array", "items": map[string]any{"type": "string"}}}, "required": []string{"related", "relation_type", "confidence", "explanation", "needs_research", "research_query", "topic_label", "keywords"}}
|
||||
}
|
||||
func answerSchema() map[string]any {
|
||||
return map[string]any{"type": "object", "properties": map[string]any{"answer": map[string]any{"type": "string"}, "used_node_ids": map[string]any{"type": "array", "items": map[string]any{"type": "string"}}, "uncertainties": map[string]any{"type": "array", "items": map[string]any{"type": "string"}}}, "required": []string{"answer", "used_node_ids", "uncertainties"}}
|
||||
|
||||
Reference in New Issue
Block a user