Customize the PAI-Rec engine with custom filters, recall, sorting, ranking, feature loading, and feature engineering components. - Platform For AI

You can customize the PAI-Rec engine by creating custom components for filtering, recall, and sorting to build a personalized recommendation system.

Import the engine

go get github.com/alibaba/pairec/v2

Customize a filter

To add a custom filter, follow these steps:

1. Implement a custom type that satisfies the IFilter interface. Because this interface is predefined in PAI-Rec, you only need to provide the implementation.

 // IFilter interface definition
 type IFilter interface {
        // Filter implements the custom filtering logic. It receives items in filterData.Data and should replace them with the filtered results.
        Filter(filterData *FilterData) error
 }

package filter
import (
	"fmt"

	"github.com/alibaba/pairec/v2/filter"
	"github.com/alibaba/pairec/v2/module"
)
 
type MyFilter struct {
}

func (f *MyFilter) Filter(data *filter.FilterData) error {
    fmt.Println("my filter")
    items := data.Data.([]*module.Item)  
    newItems := make([]*module.Item, 0)
    // Process the items and add the eligible ones to newItems.
    ...
    data.Data = newItems
    return nil
}

2. Before the server starts, register your custom filter.

package main

import (
	"pairec_base/src/controller"
	myFilter "pairec_base/src/filter"

	"github.com/alibaba/pairec/v2"
	"github.com/alibaba/pairec/v2/filter"
)

func main() {
        // Register your filter and add it to a hook.
	pairec.AddStartHook(func() error {
		filter.RegisterFilter("myFilter", &myFilter.MyFilter{})
		return nil
	})

	pairec.Route("/api/rec/feed", &controller.FeedController{})
	pairec.Run()
}

3. Update the configuration to use your custom filter. You can apply the filter to a specific scene or to all scenes by using the default key.

// myFilter is the custom filter.
"FilterNames":  {"default":  ["myFilter", "item_exposure_filter"]}

Customize recall

Creating a custom recall component follows the same pattern as creating a filter. Implement the predefined Recall interface.

// Recall interface definition
type Recall interface {
    GetCandidateItems(user *module.User, context *context.RecommendContext) []*module.Item
}

package recall

import (
	"github.com/alibaba/pairec/v2/context"
	"github.com/alibaba/pairec/v2/module"
)

type MyRecall struct {
}

func (r *MyRecall) GetCandidateItems(user *module.User, context *context.RecommendContext) []*module.Item {
	ret := make([]*module.Item, 0)
	// Specific recall logic.
        ...
	return ret
}

Then, register the component by calling recall.RegisterRecall.

package main

import (
	"pairec_base/src/controller"
	recall2 "pairec_base/src/recall"

	"github.com/alibaba/pairec/v2"
	"github.com/alibaba/pairec/v2/service/recall"
)

func main() {
	pairec.AddStartHook(func() error {
		recall.RegisterRecall("myRecall", &recall2.MyRecall{})
		return nil
	})

	pairec.Route("/api/rec/feed", &controller.FeedController{})
	pairec.Run()
}

In your scene configuration, specify the name of the recall component.

"SceneConfs": {
		"home_feed": {
			"default": {
			        // Set to myRecall.
				"RecallNames": ["myRecall"]
			}
		}
	}

Customize sort

Creating a custom sort component follows the same pattern as creating a filter.

1. Implement a type that satisfies the predefined ISort interface.

type ISort interface {   
    // Implement the sort interface. The data is in sortData.Data.
    Sort(sortData *SortData) error
 }

package sort

import (
	"fmt"

	"github.com/alibaba/pairec/v2/module"
	"github.com/alibaba/pairec/v2/sort"
)

type MySort struct {
}

func (s *MySort) Sort(data *sort.SortData) error {
	fmt.Println("my sort")
	items := data.Data.([]*module.Item)
	// Specific sort logic.
	...
	data.Data = items
	return nil
}

2. Before the server starts, register your custom sort component.

package main

import (
	"pairec_base/src/controller"
	sort2 "pairec_base/src/sort"

	"github.com/alibaba/pairec/v2"
	"github.com/alibaba/pairec/v2/sort"
)

func main() {
	pairec.AddStartHook(func() error {
		sort.RegisterSort("mySort", &sort2.MySort{})
		return nil
	})

	pairec.Route("/api/rec/feed", &controller.FeedController{})
	pairec.Run()
}

3. Modify the configuration to use the sort component.

"SortNames":   {"default":   ["mySort", "item_score"]}

Customize score boosting

You can apply custom strategies to adjust model scores after scoring. Provide a function that matches the boostFunc signature.

type boostFunc func(score float64, user *module.User, item *module.Item, context *context.RecommendContext) float64

package rank
import (
	"github.com/alibaba/pairec/v2/context"
	"github.com/alibaba/pairec/v2/module"
)

// User-defined function
func BoostScore(score float64, user *module.User, item *module.Item, context *context.RecommendContext) float64 {
	// Implement score boosting logic here.
	vTagId, err := item.IntProperty("vtag_id")
	if err == nil && vTagId == 20214 {
		return score * 1.3
	}
	return score
}

Registration:

package main
import(
      prank "github.com/alibaba/pairec/v2/service/rank"
      "pairec_base/src/rank"
      "github.com/alibaba/pairec/v2"
)

func main(){
        pairec.AddStartHook(func() error {
            prank.SetBoostFunc(rank.BoostScore)
        })

	pairec.Route("/api/rec/feed", &controller.FeedController{})
	pairec.Run()
}

Customize feature loading

In addition to using FeatureConfs, you can load features with a custom function.

The feature loading function must have the following signature:

type LoadFeatureFunc func(user *module.User, items []*module.Item, context *context.RecommendContext)

Use

func RegisterLoadFeatureFunc(sceneName string, f LoadFeatureFunc)

to register the function. The registration is scene-specific.

Example:

package feature
import (
	"github.com/alibaba/pairec/v2/context"
	"github.com/alibaba/pairec/v2/module"
)

func LoadRealTimeFeatures(user *module.User, items []*module.Item, context *context.RecommendContext) {
	// Add a property to the user.
	user.AddProperty("userAge", 30)

	// Add a property to each item.
	for _, item := range items {
		item.AddProperty("count", 5)
	}
}

Registration:

package main
import(
      pfeature "github.com/alibaba/pairec/v2/service/feature"
      "github.com/alibaba/pairec/v2"
)

func main(){
       pairec.AddStartHook(func() error {
            // feed is the scene name.
            pfeature.RegisterLoadFeatureFunc("feed", feature.LoadRealTimeFeatures)
            return nil
        })

	pairec.Route("/api/rec/feed", &controller.FeedController{})
	pairec.Run()
}

Customize feature engineering

After feature loading, you may need to perform feature engineering tasks such as generating new features, creating feature combinations, or processing user and item features together. The engine provides predefined feature processing operators. If these operators do not meet your needs, you can implement custom logic.

You can implement custom feature engineering with a function.

The feature engineering function is defined as follows:

type FeatureFunc func(user *module.User, items []*module.Item, context *context.RecommendContext) []*module.Item

Use

func RegisterFeatureFunc(sceneName string, f FeatureFunc)

to register the function. This registration is also scene-specific.

Example:

Define a feature processing function:

package feature
import (
	"github.com/alibaba/pairec/v2/context"
	"github.com/alibaba/pairec/v2/module"
)

// You can also reduce the number of returned items here.
func MyFeatureFunc(user *module.User, items []*module.Item, context *context.RecommendContext) []*module.Item {
	if len(items) < 400 {
		return items
	}
  
  // Feature processing logic.
  
  return items[:400]
}

package main
import(
      pfeature "github.com/alibaba/pairec/v2/service/feature"
      "github.com/alibaba/pairec/v2"
)

func main(){
       pairec.AddStartHook(func() error {
            // feed is the scene name.
            pfeature.RegisterFeatureFunc("feed", feature.MyFeatureFunc)
            return nil
        })

	pairec.Route("/api/rec/feed", &controller.FeedController{})
	pairec.Run()
}

Customize rank

To integrate a custom rank component, implement the IRank interface. This interface requires two methods: Filter to select items for custom ranking, and Rank to apply your ranking logic.

The IRank interface is defined as follows:

type IRank interface {
	// Filter determines whether the custom rank logic applies to an item.
	Filter(User *module.User, item *module.Item, context *context.RecommendContext) bool

	Rank(User *module.User, items []*module.Item, requestData []map[string]interface{}, context *context.RecommendContext)
}

Then, use

func RegisterRank(sceneName string, ranks ...IRank)

to register different custom rank components for each scene.

Note

You can register multiple rank components for a single scene. This allows you to apply different ranking algorithms to different types of items.

The following example demonstrates this:

type MyRank struct {
    index int
}

func NewMyRank() *MyRank {
    return &MyRank{
        index: 0,
    }
}
func (r *MyRank) Filter(User *module.User, item *module.Item, context *context.RecommendContext) bool {
    r.index++

    item.AddProperty("other", "other")
    if r.index%2 == 0 {
        item.AddProperty("index", r.index)
        return true
    }
    return false
}

func (r *MyRank) Rank(User *module.User, items []*module.Item, requestData []map[string]interface{}, context *context.RecommendContext) {
    fmt.Println("rank len", len(items))
    for _, item := range items {
        if f, err := item.FloatProperty("index"); err == nil {
            item.Score = float64(f * 10)
        }
    }

    r.index = 0
}

If an item does not match any custom rank component, the engine uses the model configuration from RankConf.

Complete example

The recall.go file provides an example of a custom recall component.

package recall

import (
	"fmt"
	"math/rand"

	"github.com/alibaba/pairec/v2/context"
	"github.com/alibaba/pairec/v2/module"
)

type MyRecall struct {
}

func (r *MyRecall) GetCandidateItems(user *module.User, context *context.RecommendContext) []*module.Item {
	fmt.Println("MyRecall is running!")

	ret := make([]*module.Item, 0)

	for i := 1; i < 100; i++ {
		item := module.NewItem(fmt.Sprintf("item_%d", i))
		item.Score = rand.Float64()
		item.AddProperty("title", fmt.Sprintf("News_%d", i))
		item.AddProperty("count", i)
		item.RetrieveId = "myRecall"
		ret = append(ret, item)
	}

	return ret
}

The filter.go file provides an example of a custom filter.

package filter

import (
	"fmt"

	"github.com/alibaba/pairec/v2/filter"
	"github.com/alibaba/pairec/v2/module"
)

type MyFilter struct {
}

func (f *MyFilter) Filter(data *filter.FilterData) error {
	fmt.Println("MyFilter is running!")

	items := data.Data.([]*module.Item)
	newItems := make([]*module.Item, 0)

	for _, item := range items {
		if item.Score > 0.2 {
			newItems = append(newItems, item)
		}
	}

	fmt.Printf("MyFilter: kept %d items\n", len(newItems))
	data.Data = newItems
	return nil
}

The rank.go file provides an example of a custom rank component. You can define multiple rank components, such as MyRank and MyRank1.

package rank

import (
	"fmt"

	"github.com/alibaba/pairec/v2/context"
	"github.com/alibaba/pairec/v2/module"
)

type MyRank struct {
}

func (rank *MyRank) Filter(User *module.User, item *module.Item, context *context.RecommendContext) bool {
	//fmt.Println("MyRank Filter ")
	if item != nil && item.Score <= 0.5 {
		return true
	}
	return false
}
func (rank *MyRank) Rank(User *module.User, items []*module.Item, requestData []map[string]interface{}, context *context.RecommendContext) {
	fmt.Println("MyRank is running")

	for _, item := range items {
		item.Score = BoostScore(item.Score, User, item, context)
	}
}

rank1.go

package rank

import (
	"fmt"

	"github.com/alibaba/pairec/v2/context"
	"github.com/alibaba/pairec/v2/module"
)

type MyRank1 struct{}

func (r *MyRank1) Filter(User *module.User, item *module.Item, context *context.RecommendContext) bool {
	//fmt.Println("MyRank1 Filter is running")
	if item != nil && 0.5 < item.Score && item.Score < 1 {
		return true
	}
	return false
}
func (r *MyRank1) Rank(User *module.User, items []*module.Item, requestData []map[string]interface{}, context *context.RecommendContext) {
	fmt.Println("MyRank1 is running")
	for _, item := range items {
		if item.Score >= 0.9 {
			item.Score *= 1.1
		} else if item.Score >= 0.8 {
			item.Score *= 1.2
		} else if item.Score >= 0.7 {
			item.Score *= 1.3
		} else {
			item.Score *= 1.4
		}
	}
}

The boost.go file contains an example of a rank-based boost function. Call this function manually from within a custom rank component. The engine calls it automatically for standard rank components.

package rank

import (
	"github.com/alibaba/pairec/v2/context"
	"github.com/alibaba/pairec/v2/module"
)

func BoostScore(score float64, user *module.User, item *module.Item, ctx *context.RecommendContext) float64 {
	//fmt.Println("BoostScore is running!")
	return score * 1.5
}

The feature.go file provides examples of custom feature functions.

package feature

import (
	"fmt"

	"github.com/alibaba/pairec/v2/context"
	"github.com/alibaba/pairec/v2/module"
)

// LoadRealTimeFeatures: A custom feature loading function.
func LoadRealTimeFeatures(user *module.User, items []*module.Item, context *context.RecommendContext) {
	// Add a property to the user.
	user.AddProperty("userAge", 30)

	// Add a property to each item.
	for _, item := range items {
		item.AddProperty("count", 5)
	}
}

// MyFeatureFunc: Custom feature processing.
func MyFeatureFunc(user *module.User, items []*module.Item, context *context.RecommendContext) []*module.Item {
	fmt.Println("MyFeatureFunc is running")
	if len(items) < context.Size {
		fmt.Printf("items less size :%d \n", len(items))
		return items
	}
	item2 := make([]*module.Item, 0, len(items))
	for _, item := range items {
		if v, ok := item.Properties["userAge"]; ok {
			if age, o := v.(int); o && age > 18 {
				item2 = append(item2, item)
			}
		}

	}
	if len(item2) < context.Size {
		fmt.Printf("item2 : %d \n", len(item2))
		return item2
	} else {
		fmt.Printf("items[:context.Size] : %d \n", len(items))
		return items[:context.Size]
	}
}

The main.go file is the main function that registers all custom components.

package main

import (
	"pairec_base/src/controller"
	feature2 "pairec_base/src/feature"
	filter2 "pairec_base/src/filter"
	rank2 "pairec_base/src/rank"
	recall2 "pairec_base/src/recall"
	sort2 "pairec_base/src/sort"

	"github.com/alibaba/pairec/v2"
	"github.com/alibaba/pairec/v2/filter"
	"github.com/alibaba/pairec/v2/service/feature"
	"github.com/alibaba/pairec/v2/service/rank"
	"github.com/alibaba/pairec/v2/service/recall"
	"github.com/alibaba/pairec/v2/sort"
)

func main() {
	pairec.AddStartHook(func() error {
		recall.RegisterRecall("myRecall", &recall2.MyRecall{})
		filter.RegisterFilter("myFilter", &filter2.MyFilter{})
		sort.RegisterSort("mySort", &sort2.MySort{})
		rank.RegisterRank("home_feed", &rank2.MyRank{}, &rank2.MyRank1{})
		rank.SetBoostFunc(rank2.BoostScore)
		feature.RegisterLoadFeatureFunc("home_feed", feature2.LoadRealTimeFeatures)
		feature.RegisterFeatureFunc("home_feed", feature2.MyFeatureFunc)
		return nil
	})

	pairec.Route("/api/rec/feed", &controller.FeedController{})
	pairec.Run()
}

The following code shows an example configuration in the config.json file.

{
	"RunMode": "product",
	"ListenConf": {
	  "HttpAddr": "",
	  "HttpPort": 8000
	},
	"FilterConfs": [
	],
	"RecallConfs": [
		{
			"Name": "mock_recall",
			"RecallType": "MockRecall",
			"RecallCount": 200
		}
	],
	"SortNames": {
	  "default": [
		  "mySort"
	  ]
	},
	"FilterNames": {
	  "default": [
		"myFilter"
	  ]
	},
	"AlgoConfs": [
	],
	"KafkaConfs": {
	},
	"RedisConfs": {
	},
	"SceneConfs": {
		"home_feed": {
			"default": {
				"RecallNames": ["myRecall"],
				"FilterNames": ["myFilter"],
				"SortNames":["mySort"]
			}
		}
	},
	"LogConf": {
	  "RetensionDays": 3,
	  "DiskSize": 20,
	  "LogLevel": "INFO"
	},
	"RankConf": {
	},
	"FeatureConfs": {
	}
}