Optimierter Workflow
This commit is contained in:
+47
-12
@@ -33,6 +33,12 @@ var (
|
||||
ErrThinkingDisabled = errors.New("thinking is disabled")
|
||||
)
|
||||
|
||||
type pendingArticleCandidate struct {
|
||||
Seeds []model.Node
|
||||
Relation model.RelationDecision
|
||||
Research []model.ResearchResult
|
||||
}
|
||||
|
||||
type EnrichOutcome struct {
|
||||
Result string
|
||||
Candidate bool
|
||||
@@ -45,6 +51,7 @@ type EnrichOutcome struct {
|
||||
CoarseComparisons int
|
||||
IndexedNodes int
|
||||
CandidatePool int
|
||||
PendingArticle *pendingArticleCandidate
|
||||
}
|
||||
|
||||
type Engine struct {
|
||||
@@ -156,6 +163,15 @@ func New(cfg config.Config, g *graph.Store, b *activity.Broker) *Engine {
|
||||
if cfg.ArticleResearchFetchResults > cfg.ArticleResearchResults {
|
||||
cfg.ArticleResearchFetchResults = cfg.ArticleResearchResults
|
||||
}
|
||||
if cfg.ArticleAdaptiveInitialQueries < 1 {
|
||||
cfg.ArticleAdaptiveInitialQueries = 2
|
||||
}
|
||||
if cfg.ArticleAdaptiveInitialFetch < 1 {
|
||||
cfg.ArticleAdaptiveInitialFetch = 3
|
||||
}
|
||||
if cfg.ArticleAdaptiveInitialFetch > cfg.ArticleResearchFetchResults {
|
||||
cfg.ArticleAdaptiveInitialFetch = cfg.ArticleResearchFetchResults
|
||||
}
|
||||
if cfg.ArticleResearchExplorationResults > cfg.ArticleResearchFetchResults {
|
||||
cfg.ArticleResearchExplorationResults = cfg.ArticleResearchFetchResults
|
||||
}
|
||||
@@ -274,7 +290,7 @@ func New(cfg config.Config, g *graph.Store, b *activity.Broker) *Engine {
|
||||
e.GLPIKB = ingest.NewGLPIKBSyncer(ingest.GLPIKBConfig{Enabled: true, Path: cfg.GLPIKBPath, Filter: cfg.GLPIKBFilter, Limit: cfg.GLPIKBLimit, SyncInterval: cfg.GLPIKBSyncInterval, Source: cfg.GLPIKBSource, CachePath: filepath.Join(cfg.DataDir, "glpi-kb-cache.json"), ShouldSync: e.LearningEnabled}, client, g, b, persistence)
|
||||
}
|
||||
if b != nil {
|
||||
b.Publish(model.Activity{Type: "system.started", Source: "brain", Phase: "startup", Message: "Neural Brain wurde gestartet; das Analyseprotokoll zeichnet Läufe und Graphänderungen auf", Strength: .3, Metadata: map[string]any{"chat_model": cfg.ChatModel, "embedding_model": cfg.EmbeddingModel, "article_language": cfg.ArticleLanguage, "article_synthesis_model": cfg.ArticleSynthesisModel, "article_review_model": cfg.ArticleReviewModel, "article_review_repair_rounds": cfg.ArticleReviewRepairRounds, "article_pipeline": "research_generate_review", "research_ollama_max_inflight": cfg.ResearchOllamaMaxInflight, "research_ollama_queue_size": cfg.ResearchOllamaQueueSize, "graph_version": g.Version()}})
|
||||
b.Publish(model.Activity{Type: "system.started", Source: "brain", Phase: "startup", Message: "Neural Brain wurde gestartet; das Analyseprotokoll zeichnet Läufe und Graphänderungen auf", Strength: .3, Metadata: map[string]any{"chat_model": cfg.ChatModel, "embedding_model": cfg.EmbeddingModel, "article_language": cfg.ArticleLanguage, "article_synthesis_model": cfg.ArticleSynthesisModel, "article_review_model": cfg.ArticleReviewModel, "article_review_repair_rounds": cfg.ArticleReviewRepairRounds, "article_pipeline": "adaptive_generate_review", "article_research_strategy": cfg.ArticleResearchStrategy, "cluster_article_batching": cfg.ClusterArticleBatching, "research_ollama_max_inflight": cfg.ResearchOllamaMaxInflight, "research_ollama_queue_size": cfg.ResearchOllamaQueueSize, "graph_version": g.Version()}})
|
||||
}
|
||||
return e
|
||||
}
|
||||
@@ -397,6 +413,7 @@ func (e *Engine) runEnrichmentCycle(ctx context.Context, trigger string) {
|
||||
created, rejected, checked := 0, 0, 0
|
||||
exactComparisons, coarseComparisonsTotal, candidatePoolTotal := 0, 0, 0
|
||||
relationsCreated, articlesCreated, articlesSkipped := 0, 0, 0
|
||||
pendingArticles := make([]*pendingArticleCandidate, 0, e.Cfg.EnrichBatchSize)
|
||||
result := "completed"
|
||||
var cycleErr error
|
||||
for step := 0; step < e.Cfg.EnrichBatchSize; step++ {
|
||||
@@ -432,6 +449,9 @@ func (e *Engine) runEnrichmentCycle(ctx context.Context, trigger string) {
|
||||
if outcome.ArticleSkipped {
|
||||
articlesSkipped++
|
||||
}
|
||||
if outcome.PendingArticle != nil {
|
||||
pendingArticles = append(pendingArticles, outcome.PendingArticle)
|
||||
}
|
||||
if outcome.Rejected {
|
||||
rejected++
|
||||
}
|
||||
@@ -446,6 +466,12 @@ func (e *Engine) runEnrichmentCycle(ctx context.Context, trigger string) {
|
||||
}
|
||||
}
|
||||
|
||||
if len(pendingArticles) > 0 && cycleErr == nil {
|
||||
clusterCreated, clusterSkipped := e.synthesizePendingArticleClusters(ctx, trigger, pendingArticles)
|
||||
articlesCreated += clusterCreated
|
||||
articlesSkipped += clusterSkipped
|
||||
}
|
||||
|
||||
e.stateMu.Lock()
|
||||
e.enrichRunning = false
|
||||
e.enrichResult = result
|
||||
@@ -897,10 +923,10 @@ func (e *Engine) enrichOne(ctx context.Context, trigger string) (EnrichOutcome,
|
||||
}
|
||||
e.Broker.Publish(model.Activity{Type: "research.results", Source: "searxng", Phase: "research-results", NodeIDs: []string{a.ID, b.ID}, Message: message, Strength: .92, Metadata: resultMetadata})
|
||||
if len(allowedResults) > 0 {
|
||||
// Keep relation-search hits in memory until the relation itself has
|
||||
// passed the second model review. Rejected relation research must not
|
||||
// permanently inflate the graph with unused external nodes.
|
||||
researchResults = allowedResults
|
||||
refs := e.addResearch(a, b, allowedResults)
|
||||
ingestMetadata := mergeResearchMetadata(resultMetadata, map[string]any{"result_node_ids": refs.NodeIDs, "result_edge_ids": refs.EdgeIDs})
|
||||
e.Broker.Publish(model.Activity{Type: "research.ingested", Source: "searxng", Phase: "research-ingest", NodeIDs: append([]string{a.ID, b.ID}, refs.NodeIDs...), EdgeIDs: refs.EdgeIDs, Message: fmt.Sprintf("%d Webquellen wurden als neue Forschungs-Nodes in den Graphen übernommen", len(refs.NodeIDs)), Strength: 1, Metadata: ingestMetadata})
|
||||
var reviewed model.RelationDecision
|
||||
reviewSystem := "Bewerte die Beziehung erneut anhand der zwei internen Wissenseinträge und der beigefügten Web-Suchergebnisse. Suchtreffer sind Hinweise, keine garantierten Fakten. Erfinde nichts, kennzeichne verbleibende Unsicherheit und gib ausschließlich JSON nach Schema zurück."
|
||||
if err := e.Ollama.ChatJSON(ctx, reviewSystem, relationContextWithResearch(a, b, sim, allowedResults), relationSchema(), &reviewed); err != nil {
|
||||
@@ -938,14 +964,23 @@ func (e *Engine) enrichOne(ctx context.Context, trigger string) (EnrichOutcome,
|
||||
if status == "staging" {
|
||||
outcome.Created = true
|
||||
outcome.RelationCreated = true
|
||||
if len(researchResults) > 0 {
|
||||
refs := e.addResearch(a, b, researchResults)
|
||||
e.Broker.Publish(model.Activity{Type: "research.ingested", Source: "searxng", Phase: "research-ingest", NodeIDs: append([]string{a.ID, b.ID}, refs.NodeIDs...), EdgeIDs: refs.EdgeIDs, Message: fmt.Sprintf("%d Webquellen wurden nach akzeptierter Relation in den Graphen übernommen", len(refs.NodeIDs)), Strength: .82, Metadata: map[string]any{"trigger": trigger, "result_node_ids": refs.NodeIDs, "result_edge_ids": refs.EdgeIDs, "materialization": "accepted_relation_only"}})
|
||||
}
|
||||
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
|
||||
if e.RuntimeSettings().ProcessingMode == "clustered" && e.Cfg.ClusterArticleBatching {
|
||||
outcome.PendingArticle = &pendingArticleCandidate{Seeds: []model.Node{a, b}, Relation: decision, Research: researchResults}
|
||||
e.Broker.Publish(model.Activity{Type: "article.cluster.deferred", Source: "brain", Phase: "knowledge-planning", NodeIDs: []string{a.ID, b.ID}, Message: "Relation wird bis zum Zyklusende mit thematisch ähnlichen Relationen zu einem gemeinsamen Artikelauftrag gebündelt", Strength: .38, Metadata: map[string]any{"trigger": trigger, "topic_label": decision.TopicLabel, "processing_mode": "clustered"}})
|
||||
} else {
|
||||
outcome.ArticleCreated = articleOutcome.Created
|
||||
outcome.ArticleSkipped = articleOutcome.Skipped
|
||||
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
|
||||
@@ -988,12 +1023,12 @@ func (e *Engine) Status() map[string]any {
|
||||
"article_research_min_relevance": e.Cfg.ArticleResearchMinRelevance, "article_research_min_quality": e.Cfg.ArticleResearchMinQuality,
|
||||
"article_research_page_max_bytes": e.Cfg.ArticleResearchPageMaxBytes, "article_research_page_max_chars": e.Cfg.ArticleResearchPageMaxChars,
|
||||
"article_research_fetch_timeout": e.Cfg.ArticleResearchFetchTimeout.String(), "article_research_allow_private": e.Cfg.ArticleResearchAllowPrivate,
|
||||
"article_language": e.Cfg.ArticleLanguage, "article_synthesis_model": e.Cfg.ArticleSynthesisModel, "article_review_model": e.Cfg.ArticleReviewModel, "article_review_repair_rounds": e.Cfg.ArticleReviewRepairRounds, "article_pipeline": "research_generate_review", "research_dedupe": e.researchDedupeStatus(),
|
||||
"article_language": e.Cfg.ArticleLanguage, "article_synthesis_model": e.Cfg.ArticleSynthesisModel, "article_review_model": e.Cfg.ArticleReviewModel, "article_review_repair_rounds": e.Cfg.ArticleReviewRepairRounds, "article_pipeline": "adaptive_generate_review", "article_research_strategy": e.Cfg.ArticleResearchStrategy, "article_effective_research_strategy": e.effectiveArticleResearchStrategy(), "article_adaptive_initial_queries": e.Cfg.ArticleAdaptiveInitialQueries, "article_adaptive_initial_fetch": e.Cfg.ArticleAdaptiveInitialFetch, "research_dedupe": e.researchDedupeStatus(),
|
||||
"enrich_interval": e.Cfg.EnrichInterval.String(),
|
||||
"enrich_batch_size": e.Cfg.EnrichBatchSize, "enrich_anchors": e.Cfg.EnrichAnchors,
|
||||
"processing_mode": e.RuntimeSettings().ProcessingMode, "cluster_hash_bits": e.Cfg.ClusterHashBits, "cluster_hash_tables": e.Cfg.ClusterHashTables,
|
||||
"cluster_candidates_per_anchor": e.Cfg.ClusterCandidatesPerAnchor, "cluster_article_candidates": e.Cfg.ClusterArticleCandidates,
|
||||
"cluster_review_evidence": e.Cfg.ClusterReviewEvidence, "cluster_review_context_chars": e.Cfg.ClusterReviewContextChars,
|
||||
"cluster_review_evidence": e.Cfg.ClusterReviewEvidence, "cluster_review_context_chars": e.Cfg.ClusterReviewContextChars, "cluster_article_batching": e.Cfg.ClusterArticleBatching,
|
||||
"research_enabled": e.ResearchEnabledForRuntime(), "chat_model": e.Cfg.ChatModel, "embedding_model": e.Cfg.EmbeddingModel,
|
||||
"searxng": e.ResearchStatus(),
|
||||
"ollama_pool": e.Ollama.PoolStatus(), "article_model_status": map[string]any{"synthesis": e.Ollama.ModelStatus(e.Cfg.ArticleSynthesisModel), "review": e.Ollama.ModelStatus(e.Cfg.ArticleReviewModel)}, "persistence": e.Persistence.Status(), "graph_storage": e.Graph.StorageStatus(),
|
||||
|
||||
Reference in New Issue
Block a user