smartmate/backend/services/memory/internal/service/retrieve_merge.go

package service

import (
	"context"
	"strings"
	"time"

	memoryutils "github.com/LoveLosita/smartflow/backend/services/memory/internal/utils"
	memorymodel "github.com/LoveLosita/smartflow/backend/services/memory/model"
	"github.com/LoveLosita/smartflow/backend/services/runtime/model"
)

// HybridRetrieve 统一承接读取侧 RAG-first 召回链路。
//
// 步骤化说明：
// 1. 优先走 RAG 语义搜索，按 query 相关性召回候选记忆；
// 2. RAG 报错或 0 命中时回退 MySQL，保证链路韧性；
// 3. 召回结果做三级去重、排序与类型预算裁剪（总量不超过调用方 limit）；
// 4. 旧 legacy 链路完全保留，方便通过配置快速回滚。
func (s *ReadService) HybridRetrieve(
	ctx context.Context,
	req memorymodel.RetrieveRequest,
	effectiveSetting model.MemoryUserSetting,
	limit int,
	now time.Time,
) ([]memorymodel.ItemDTO, retrieveTelemetry, error) {
	telemetry := retrieveTelemetry{}
	if s == nil || s.itemRepo == nil {
		return nil, telemetry, nil
	}
	if !effectiveSetting.MemoryEnabled {
		return nil, telemetry, nil
	}

	// RAG-first：只走语义召回，不再全量拉 MySQL pinned。
	items, semanticTelemetry, err := s.retrieveSemanticCandidates(ctx, req, effectiveSetting, limit, now)
	if err != nil {
		return nil, telemetry, err
	}
	telemetry.SemanticHitCount = semanticTelemetry.HitCount
	telemetry.Degraded = semanticTelemetry.Degraded
	telemetry.RAGFallbackUsed = semanticTelemetry.RAGFallbackUsed

	if len(items) == 0 {
		return nil, telemetry, nil
	}

	beforeDedupCount := len(items)
	items = dedupByID(items)
	items = dedupByHash(items)
	items = dedupByText(items)
	telemetry.DedupDropCount = beforeDedupCount - len(items)
	items = RankItems(items, now)
	items = applyTypeBudget(items, s.cfg, limit)
	if len(items) == 0 {
		return nil, telemetry, nil
	}
	telemetry.FinalCount = len(items)

	_ = s.itemRepo.TouchLastAccessAt(ctx, collectItemDTOIDs(items), now)
	return items, telemetry, nil
}

func (s *ReadService) retrievePinnedCandidates(
	ctx context.Context,
	req memorymodel.RetrieveRequest,
	effectiveSetting model.MemoryUserSetting,
	now time.Time,
) ([]memorymodel.ItemDTO, error) {
	query := buildReadScopedItemQuery(req, now, nil, 0)
	items, err := s.itemRepo.FindPinnedByUser(ctx, query, s.cfg.EffectiveReadPreferenceLimit())
	if err != nil {
		return nil, err
	}
	items = memoryutils.FilterItemsBySetting(items, effectiveSetting)
	return toItemDTOs(items), nil
}

func (s *ReadService) retrieveSemanticCandidates(
	ctx context.Context,
	req memorymodel.RetrieveRequest,
	effectiveSetting model.MemoryUserSetting,
	limit int,
	now time.Time,
) ([]memorymodel.ItemDTO, semanticRetrieveTelemetry, error) {
	telemetry := semanticRetrieveTelemetry{}
	queryText := strings.TrimSpace(req.Query)
	if queryText == "" {
		return nil, telemetry, nil
	}

	candidateLimit := hybridSemanticTopK(s.cfg, limit)
	if s.cfg.RAGEnabled && s.ragRuntime != nil {
		items, err := s.retrieveSemanticCandidatesByRAG(ctx, req, effectiveSetting, candidateLimit, now)
		if shouldReturnSemanticRAGResult(items, err) {
			telemetry.HitCount = len(items)
			return items, telemetry, nil
		}
		telemetry.Degraded = true
		telemetry.RAGFallbackUsed = true
	}
	items, err := s.retrieveSemanticCandidatesByMySQL(ctx, req, effectiveSetting, candidateLimit, now)
	telemetry.HitCount = len(items)
	return items, telemetry, err
}

func (s *ReadService) retrieveSemanticCandidatesByRAG(
	ctx context.Context,
	req memorymodel.RetrieveRequest,
	effectiveSetting model.MemoryUserSetting,
	candidateLimit int,
	now time.Time,
) ([]memorymodel.ItemDTO, error) {
	result, err := s.ragRuntime.RetrieveMemory(ctx, buildReadScopedRAGRequest(req, candidateLimit, s.cfg.Threshold))
	if err != nil {
		return nil, err
	}
	if result == nil || len(result.Items) == 0 {
		return nil, nil
	}

	items := make([]memorymodel.ItemDTO, 0, len(result.Items))
	for _, hit := range result.Items {
		dto, memoryID := buildMemoryDTOFromRetrieveHit(hit)
		if !effectiveSetting.ImplicitMemoryEnabled && !dto.IsExplicit {
			continue
		}
		if !effectiveSetting.SensitiveMemoryEnabled && dto.SensitivityLevel > 0 {
			continue
		}
		if dto.ID <= 0 && memoryID > 0 {
			dto.ID = memoryID
		}
		items = append(items, dto)
	}
	return items, nil
}

func (s *ReadService) retrieveSemanticCandidatesByMySQL(
	ctx context.Context,
	req memorymodel.RetrieveRequest,
	effectiveSetting model.MemoryUserSetting,
	candidateLimit int,
	now time.Time,
) ([]memorymodel.ItemDTO, error) {
	query := buildReadScopedItemQuery(
		req,
		now,
		[]string{model.MemoryItemStatusActive},
		normalizeLimit(candidateLimit, candidateLimit, maxRetrieveLimit),
	)

	items, err := s.itemRepo.FindByQuery(ctx, query)
	if err != nil {
		return nil, err
	}
	items = memoryutils.FilterItemsBySetting(items, effectiveSetting)
	return toItemDTOs(items), nil
}

// dedupByID 按 memory_id 去重，后出现的结果覆盖先出现的结果。
func dedupByID(items []memorymodel.ItemDTO) []memorymodel.ItemDTO {
	if len(items) == 0 {
		return nil
	}

	seen := make(map[int64]struct{}, len(items))
	result := make([]memorymodel.ItemDTO, 0, len(items))
	for i := len(items) - 1; i >= 0; i-- {
		item := items[i]
		if item.ID <= 0 {
			result = append(result, item)
			continue
		}
		if _, exists := seen[item.ID]; exists {
			continue
		}
		seen[item.ID] = struct{}{}
		result = append(result, item)
	}
	reverseItemDTOs(result)
	return result
}

// dedupByHash 按 content_hash 去重；缺失 hash 时跳过，保留 importance 更高的条目。
func dedupByHash(items []memorymodel.ItemDTO) []memorymodel.ItemDTO {
	return dedupByKey(items, func(item memorymodel.ItemDTO) string {
		return fallbackContentHash(item.MemoryType, item.Content, item.ContentHash)
	})
}

// dedupByText 按“类型标签 + 文本”兜底去重，用于覆盖历史数据未带 hash 的场景。
func dedupByText(items []memorymodel.ItemDTO) []memorymodel.ItemDTO {
	return dedupByKey(items, func(item memorymodel.ItemDTO) string {
		text := strings.TrimSpace(item.Content)
		if text == "" {
			text = strings.TrimSpace(item.Title)
		}
		if text == "" {
			return ""
		}
		return renderMemoryTypeLabelForDedup(item.MemoryType) + "::" + normalizeContentForHash(text)
	})
}

func dedupByKey(items []memorymodel.ItemDTO, keyBuilder func(item memorymodel.ItemDTO) string) []memorymodel.ItemDTO {
	if len(items) == 0 {
		return nil
	}

	selectedIndex := make(map[string]int, len(items))
	for index, item := range items {
		key := strings.TrimSpace(keyBuilder(item))
		if key == "" {
			continue
		}
		if previous, exists := selectedIndex[key]; exists {
			if preferCurrentItem(items[previous], item) {
				selectedIndex[key] = index
			}
			continue
		}
		selectedIndex[key] = index
	}

	result := make([]memorymodel.ItemDTO, 0, len(items))
	for index, item := range items {
		key := strings.TrimSpace(keyBuilder(item))
		if key == "" {
			result = append(result, item)
			continue
		}
		if selectedIndex[key] == index {
			result = append(result, item)
		}
	}
	return result
}

func preferCurrentItem(previous memorymodel.ItemDTO, current memorymodel.ItemDTO) bool {
	if current.Importance != previous.Importance {
		return current.Importance > previous.Importance
	}
	if current.Confidence != previous.Confidence {
		return current.Confidence > previous.Confidence
	}
	return true
}

// applyTypeBudget 在排序结果上应用四类记忆预算，并以 callerLimit 作为总量硬上限。
//
// 说明：
// 1. 每种类型先保底自己的预算上限，避免 fact 抢掉 constraint 的位置；
// 2. 裁剪时保持当前排序顺序，不在这里重新打分；
// 3. 最终总量不超过 min(callerLimit, cfg.TotalReadBudget())。
func applyTypeBudget(items []memorymodel.ItemDTO, cfg memorymodel.Config, callerLimit int) []memorymodel.ItemDTO {
	if len(items) == 0 {
		return nil
	}

	hardCap := cfg.TotalReadBudget()
	if callerLimit > 0 && callerLimit < hardCap {
		hardCap = callerLimit
	}

	budgetByType := map[string]int{
		memorymodel.MemoryTypeConstraint: cfg.EffectiveReadConstraintLimit(),
		memorymodel.MemoryTypePreference: cfg.EffectiveReadPreferenceLimit(),
		memorymodel.MemoryTypeFact:       cfg.EffectiveReadFactLimit(),
	}
	usedByType := make(map[string]int, len(budgetByType))
	result := make([]memorymodel.ItemDTO, 0, minInt(len(items), hardCap))
	for _, item := range items {
		if len(result) >= hardCap {
			break
		}

		memoryType := resolveBudgetMemoryType(item.MemoryType)
		if usedByType[memoryType] >= budgetByType[memoryType] {
			continue
		}
		usedByType[memoryType]++
		result = append(result, item)
	}
	return result
}

// hybridSemanticTopK 计算语义召回的候选集大小。
// 使用 callerLimit 的 2 倍作为 TopK，保证去重后仍有足够结果填充预算。
func hybridSemanticTopK(cfg memorymodel.Config, limit int) int {
	return limit * 2
}

func resolveBudgetMemoryType(memoryType string) string {
	normalized := memorymodel.NormalizeMemoryType(memoryType)
	if normalized == "" {
		return memorymodel.MemoryTypeFact
	}
	return normalized
}

func renderMemoryTypeLabelForDedup(memoryType string) string {
	switch memorymodel.NormalizeMemoryType(memoryType) {
	case memorymodel.MemoryTypePreference:
		return "偏好"
	case memorymodel.MemoryTypeConstraint:
		return "约束"
	case memorymodel.MemoryTypeFact:
		return "事实"
	default:
		return "记忆"
	}
}

func collectItemDTOIDs(items []memorymodel.ItemDTO) []int64 {
	if len(items) == 0 {
		return nil
	}

	ids := make([]int64, 0, len(items))
	for _, item := range items {
		if item.ID <= 0 {
			continue
		}
		ids = append(ids, item.ID)
	}
	return ids
}

func reverseItemDTOs(items []memorymodel.ItemDTO) {
	for left, right := 0, len(items)-1; left < right; left, right = left+1, right-1 {
		items[left], items[right] = items[right], items[left]
	}
}

func minInt(left, right int) int {
	if left < right {
		return left
	}
	return right
}