Remove dup output

Author: zhenghaoz

cmd/gorse-benchmark/main.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 
 	mapset "github.com/deckarep/golang-set/v2"
+	"github.com/gorse-io/gorse/common/parallel"
 	"github.com/gorse-io/gorse/config"
 	"github.com/gorse-io/gorse/dataset"
 	"github.com/gorse-io/gorse/logics"
@@ -18,7 +19,10 @@ import (
 	"github.com/gorse-io/gorse/storage"
 	"github.com/gorse-io/gorse/storage/data"
 	"github.com/samber/lo"
+	"github.com/samber/lo/mutable"
+	"github.com/sashabaranov/go-openai"
 	"github.com/spf13/cobra"
+	"go.uber.org/atomic"
 	"golang.org/x/term"
 	"modernc.org/sortutil"
 )
@@ -65,8 +69,8 @@ var llmCmd = &cobra.Command{
 		fmt.Printf("  Negative Feedbacks: %d\n", dataset.CountNegative())
 		// Split dataset
 		train, test := dataset.Split(0.8, 42)
-		EvaluateLLM(cfg, train, test, aux.GetItems())
 		// EvaluateFM(train, test)
+		EvaluateLLM(cfg, train, test, aux.GetItems())
 	},
 }
 
@@ -109,29 +113,40 @@ func EvaluateLLM(cfg *config.Config, train, test dataset.CTRSplit, items []data.
 		}
 	}
 
-	var sumAUC float32
-	var validUsers float32
-	for userId, testItems := range userTest {
+	var sumAUC atomic.Float32
+	var validUsers atomic.Float32
+	parallel.Detachable(len(userTest), runtime.NumCPU(), 100, func(pCtx *parallel.Context, userIdx int) {
+		userId := int32(userIdx)
+		testItems := userTest[userId]
+		if len(userTrain[userId]) > 100 || len(userTrain[userId]) == 0 {
+			return
+		}
 		if _, ok := userPositive[userId]; !ok {
-			continue
+			return
 		}
 		if _, ok := userNegative[userId]; !ok {
-			continue
+			return
 		}
 		candidates := make([]*data.Item, 0, len(testItems))
 		for _, itemId := range testItems {
 			candidates = append(candidates, &items[itemId])
 		}
+		mutable.Reverse(candidates)
 		feedback := make([]*logics.FeedbackItem, 0, len(testItems))
 		for _, itemId := range userTrain[userId] {
 			feedback = append(feedback, &logics.FeedbackItem{
 				Item: items[itemId],
 			})
 		}
+		pCtx.Detach()
 		result, err := chat.Rank(&data.User{}, feedback, candidates)
 		if err != nil {
+			if apiError, ok := err.(*openai.APIError); ok && apiError.HTTPStatusCode == 421 {
+				return
+			}
 			log.Fatalf("failed to rank items for user %d: %v", userId, err)
 		}
+		pCtx.Attach()
 		var posPredictions, negPredictions []float32
 		for i, name := range result {
 			itemId := test.GetIndex().EncodeItem(name) - int32(test.CountUsers())
@@ -143,18 +158,20 @@ func EvaluateLLM(cfg *config.Config, train, test dataset.CTRSplit, items []data.
 				log.Fatalf("item %s not found in test set for user %d", name, userId)
 			}
 		}
-		sumAUC += AUC(posPredictions, negPredictions) * float32(len(posPredictions))
-		validUsers += float32(len(posPredictions))
-		fmt.Println("User", userId, "AUC:", AUC(posPredictions, negPredictions))
-		if validUsers >= 100 {
-			break
+		if len(negPredictions) == 0 || len(posPredictions) == 0 {
+			return
 		}
-	}
-	if validUsers == 0 {
+		sumAUC.Add(AUC(posPredictions, negPredictions) * float32(len(posPredictions)))
+		validUsers.Add(float32(len(posPredictions)))
+		fmt.Printf("User %d AUC: %f pos: %d/%d, neg: %d/%d\n", userId, AUC(posPredictions, negPredictions),
+			len(posPredictions), userPositive[userId].Cardinality(),
+			len(negPredictions), userNegative[userId].Cardinality())
+	})
+	if validUsers.Load() == 0 {
 		return 0
 	}
 
-	score := sumAUC / validUsers
+	score := sumAUC.Load() / validUsers.Load()
 	fmt.Println("LLM GAUC:", score)
 	return score
 }
@@ -169,6 +186,15 @@ func EvaluateFM(train, test dataset.CTRSplit) float32 {
 			SetJobs(runtime.NumCPU()).
 			SetPatience(10))
 
+	userTrain := make(map[int32]int, train.CountUsers())
+	for i := 0; i < train.Count(); i++ {
+		indices, _, target := train.Get(i)
+		userId := indices[0]
+		if target > 0 {
+			userTrain[userId]++
+		}
+	}
+
 	var posFeatures, negFeatures []lo.Tuple2[[]int32, []float32]
 	var posUsers, negUsers []int32
 	for i := 0; i < test.Count(); i++ {
@@ -196,6 +222,9 @@ func EvaluateFM(train, test dataset.CTRSplit) float32 {
 	var sumAUC float32
 	var validUsers float32
 	for user, pos := range userPosPrediction {
+		if userTrain[user] > 100 || userTrain[user] == 0 {
+			continue
+		}
 		if neg, ok := userNegPrediction[user]; ok {
 			sumAUC += AUC(pos, neg) * float32(len(pos))
 			validUsers += float32(len(pos))

logics/chat.go

@@ -111,9 +111,11 @@ func (r *ChatRanker) Rank(user *data.User, feedback []*FeedbackItem, items []*da
 		s.Add(item.ItemId)
 	}
 	var result []string
+	m := mapset.NewSet[string]()
 	for _, itemId := range parsed {
-		if s.Contains(itemId) {
+		if s.Contains(itemId) && !m.Contains(itemId) {
 			result = append(result, itemId)
+			m.Add(itemId)
 		}
 	}
 	return result, nil