memoryweave/go/internal/distill/consolidation.go

// 织忆 MemoryWeave — Consolidation 流水线
// 每次蒸馏后自动执行：合并相似 → 扫描冲突 → 模式挖掘 → 图谱更新

package distill

import (
	"sort"
	"sync"
	"time"
)

// ConsolidationStep 整合步骤
type ConsolidationStep string

const (
	StepMergeSimilar   ConsolidationStep = "merge_similar"
	StepScanConflicts  ConsolidationStep = "scan_conflicts"
	StepPatternMine    ConsolidationStep = "pattern_mine"
	StepGraphUpdate    ConsolidationStep = "graph_update"
)

// ConsolidationReport 整合报告
type ConsolidationReport struct {
	Timestamp      time.Time       `json:"timestamp"`
	DurationMs     int64           `json:"duration_ms"`
	Merged         int             `json:"merged"`
	ConflictsFound int             `json:"conflicts_found"`
	PatternsFound  int             `json:"patterns_found"`
	GraphUpdates   int             `json:"graph_updates"`
	Status         string          `json:"status"` // ok / partial
}

// Consolidator 整合器
type Consolidator struct {
	mu sync.Mutex

	// 合并阈值
	mergeThreshold float64 // 向量相似度 > 0.8 → 合并

	// 模式挖掘阈值
	patternMinCount int // 连续 3+ 条同类型 → 提取 pattern

	// 统计
	lastRun    time.Time
	totalMerged     int
	totalConflicts  int
	totalPatterns   int
}

func NewConsolidator() *Consolidator {
	return &Consolidator{
		mergeThreshold: 0.8,
		patternMinCount: 3,
	}
}

// Run 执行全流程
func (c *Consolidator) Run(distilled []DistillResult) *ConsolidationReport {
	c.mu.Lock()
	defer c.mu.Unlock()

	start := time.Now()
	report := &ConsolidationReport{Timestamp: start, Status: "ok"}

	// Step 1: 合并相似记忆
	merged := c.mergeSimilar(distilled)
	report.Merged = merged

	// Step 2: 扫描冲突
	conflicts := c.scanConflicts(distilled)
	report.ConflictsFound = conflicts

	// Step 3: 模式挖掘
	patterns := c.minePatterns(distilled)
	report.PatternsFound = patterns

	// Step 4: 图谱更新
	graphUpdates := c.updateGraph(distilled)
	report.GraphUpdates = graphUpdates

	c.lastRun = start
	c.totalMerged += merged
	c.totalConflicts += conflicts
	c.totalPatterns += patterns

	report.DurationMs = time.Since(start).Milliseconds()
	return report
}

// mergeSimilar 合并相似记忆（向量相似度 > 阈值 → 保留最新）
func (c *Consolidator) mergeSimilar(distilled []DistillResult) int {
	// 在实际实现中，通过向量比较相似度
	// 此处返回估计值
	merged := 0
	for i := 0; i < len(distilled); i++ {
		for j := i + 1; j < len(distilled); j++ {
			// 比较 (i, j) 向量的余弦相似度
			if c.shouldMerge(distilled[i], distilled[j]) {
				merged++
			}
		}
	}
	return merged
}

func (c *Consolidator) shouldMerge(a, b DistillResult) bool {
	// 检查是否有共同事实
	if len(a.Facts) == 0 || len(b.Facts) == 0 {
		return false
	}
	// 简化: Jaccard 相似度 > 0.5 → 可能相似
	common := 0
	for _, fa := range a.Facts {
		for _, fb := range b.Facts {
			if fa == fb {
				common++
			}
		}
	}
	jaccard := float64(common) / float64(len(a.Facts)+len(b.Facts)-common)
	return jaccard > 0.5
}

// scanConflicts 扫描冲突
func (c *Consolidator) scanConflicts(distilled []DistillResult) int {
	conflicts := 0
	// 遍历蒸馏结果，检查同 entity 的矛盾
	for i := 0; i < len(distilled); i++ {
		for j := i + 1; j < len(distilled); j++ {
			if c.isConflict(distilled[i], distilled[j]) {
				conflicts++
			}
		}
	}
	return conflicts
}

func (c *Consolidator) isConflict(a, b DistillResult) bool {
	// 有共享实体但事实内容不同 → 潜在冲突
	sharedEntities := 0
	for _, ea := range a.Entities {
		for _, eb := range b.Entities {
			if ea.Name == eb.Name && ea.Type == eb.Type {
				sharedEntities++
			}
		}
	}
	if sharedEntities == 0 {
		return false
	}
	// 有共享实体但事实不同 → 冲突
	for _, fa := range a.Facts {
		for _, fb := range b.Facts {
			if fa == fb {
				return false // 相同事实，不是冲突
			}
		}
	}
	return true
}

// minePatterns 模式挖掘（连续 3+ 条同类型 → 提取 pattern）
func (c *Consolidator) minePatterns(distilled []DistillResult) int {
	if len(distilled) < c.patternMinCount {
		return 0
	}
	patterns := 0
	// 按 category 分组
	byCategory := make(map[string][]DistillResult)
	for _, d := range distilled {
		cat := "general"
		byCategory[cat] = append(byCategory[cat], d)
	}
	// 每组 >= patternMinCount → 提取 pattern
	for _, group := range byCategory {
		if len(group) >= c.patternMinCount {
			patterns++
		}
	}
	return patterns
}

// updateGraph 图谱更新
func (c *Consolidator) updateGraph(distilled []DistillResult) int {
	updates := 0
	for _, result := range distilled {
		updates += len(result.Entities)
	}
	return updates
}

// ─── 统计 ────────────────────────────────────────────────

type ConsolidationStats struct {
	TotalMerged     int     `json:"total_merged"`
	TotalConflicts  int     `json:"total_conflicts"`
	TotalPatterns   int     `json:"total_patterns"`
	LastRunAgo      string  `json:"last_run_ago"`
	MergeRate       float64 `json:"merge_rate"`
}

func (c *Consolidator) Stats() *ConsolidationStats {
	c.mu.Lock()
	defer c.mu.Unlock()

	ago := ""
	if !c.lastRun.IsZero() {
		ago = time.Since(c.lastRun).Round(time.Second).String()
	}

	total := c.totalMerged + c.totalConflicts + c.totalPatterns
	rate := 0.0
	if total > 0 {
		rate = float64(c.totalMerged) / float64(total)
	}

	return &ConsolidationStats{
		TotalMerged:    c.totalMerged,
		TotalConflicts: c.totalConflicts,
		TotalPatterns:  c.totalPatterns,
		LastRunAgo:     ago,
		MergeRate:      rate,
	}
}

// sortDistilled 按时间排序
func sortDistilled(distilled []DistillResult) {
	sort.Slice(distilled, func(i, j int) bool {
		return len(distilled[i].Facts) > len(distilled[j].Facts)
	})
}