本文介紹如何利用PAI-Rec引擎進行個人化推薦系統的二次開發，提升推薦演算法的靈活性與效能。 - Platform For AI

本文介紹如何基於PAI-Rec引擎去進行二次個人化開發，自訂過濾，召回，排序的介紹。

匯入引擎

go get github.com/alibaba/pairec/v2

自訂過濾

增加自訂Filter需要以下步驟

1.實現自訂Filter類，實現IFilter介面, IFilter介面已經在PAI-Rec中已經定義完成，只需實作類別

 // IFilter 介面定義
 type IFilter interface {
        // 主要實現 Filter 方法，傳入的資料在 filterData.Data 裡找到
        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)
    // 處理items, 合格加入到 newItems中
    ...
    data.Data = newItems
    return nil
}

2.在server啟動前，註冊自訂的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() {
        // 先註冊自己的 filter，加到 hook 中
	pairec.AddStartHook(func() error {
		filter.RegisterFilter("myFilter", &myFilter.MyFilter{})
		return nil
	})

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

3.修改配置，增加filter的訪問，這裡是配置時情境 + filter 列表的形式，情境可以設定具體的值，也可以設定預設值default

// myfilter 是自訂的 filter
"FilterNames":  {"default":  ["myfilter", "item_exposure_filter"]}

自訂召回

和過濾類似，需要實現 Recall 介面， Recall 介面已經定義完成，只需實作類別

// Recall 介面定義
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)
	// recall 具體邏輯
        ...
	return ret
}

然後使用 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()
}

在情境類別的配置中，可以設定相應的 recall 名稱。

"SceneConfs": {
		"home_feed": {
			"default": {
			        // 設定為 myRecall
				"RecallNames": ["myRecall"]
			}
		}
	}

自訂 Sort 支援

和 filter 類似，sort 也使用同樣的方式實現自訂

1.實現自訂 ISort 類， ISort 已經定義好

type ISort interface {   
    // 實現 sort 介面，資料在 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)
	// sort 具體邏輯處理
	...
	data.Data = items
	return nil
}

2.在 server 啟動前，註冊自訂的 Sort

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.修改配置，增加 Sort 的訪問

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

提權操作自訂

很多情況下模型打分之後，使用者需要自訂策略實現對模型得分的提降權操作，使用者需要提供自訂的策略邏輯。需要實現boostFunc定義。

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"
)

// 使用者自訂
func BoostScore(score float64, user *module.User, item *module.Item, context *context.RecommendContext) float64 {
	//編輯提權
	vTagId, err := item.IntProperty("vtag_id")
	if err == nil && vTagId == 20214 {
		return score * 1.3
	}
	return score
}

註冊：

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()
}

特徵自訂載入

特徵載入過程，增加了自訂特徵載入的支援。除了在配置中FeatureConfs中載入特徵，也支援自訂的 function 載入特徵。

載入特徵 function 需要實現

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

通過

func RegisterLoadFeatureFunc(sceneName string, f LoadFeatureFunc)

進行註冊。這個是分情境註冊。

例如：

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) {
	// 向 user 增加
	user.AddProperty("userAge", 30)

	// 向 item 增加
	for _, item := range items {
		item.AddProperty("count", 5)
	}
}

註冊：

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

func main(){
       pairec.AddStartHook(func() error {
            // feed 是情境名稱
            pfeature.RegisterLoadFeatureFunc("feed", feature.LoadRealTimeFeatures)
            return nil
        })

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

特徵工程自訂處理

在特徵載入完成之後，很多時候會做特徵處理的工作，比如產生新的特徵，產生組合特徵，特徵處理時，可能會有 user 和 item 特徵的綜合處理等等。在特徵載入之後，引擎也提供一些預定義的特徵處理運算元用於特徵工程，在不滿足的情況下，可以自訂處理實現。

特徵工程的自訂處理也是通過自訂的 function 來實現的。

特徵工程 function 定義

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

通過

func RegisterFeatureFunc(sceneName string, f FeatureFunc)

進行註冊。這個也是分情境的註冊。

舉例：

定義一個特徵處理函數 funciton

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

// 這裡還可以決定進一步減少返回的 item 數量
func MyFeatureFunc(user *module.User, items []*module.Item, context *context.RecommendContext) []*module.Item {
	if len(items) < 400 {
		return items
	}
  
  // 特徵的處理
  
  return items[:400]
}

在 main.go 裡進行註冊

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

func main(){
       pairec.AddStartHook(func() error {
            // feed 是情境名稱
            pfeature.RegisterFeatureFunc("feed", feature.MyFeatureFunc)
            return nil
        })

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

自訂 Rank(演算法調用）

rank 需要實現 IRank 介面才能註冊到流程中。包含兩部分，一個是過濾出想自訂調用演算法的 item，然後對過濾出來的 item 調用自訂 rank function。

IRank 介面定義如下：

type IRank interface {
	// Filter the custom rank of 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)
}

然後通過

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

根據每個情境註冊不同的自訂的 rank。

說明

這裡 ranks 可以設定多個，可以針對不同的 item，調用不同的 rank。

這裡舉個簡單的例子說明：

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
}

如果沒有匹配到自訂 Rank 的 item 列表，還是會調用 RankConf 裡的模型配置。

完整樣本

recall.go 自訂recall 樣本

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("新聞%d", i))
		item.AddProperty("count", i)
		item.RetrieveId = "myRecall"
		ret = append(ret, item)
	}

	return ret
}

filter.go 自訂 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
}

rank.go 自訂 rank 樣本，可以有很多不同的 rank ，rank 和 rank1 分別是不同的樣本

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
		}
	}
}

boost.go 基於 rank 的 boost 樣本（如果是自訂rank，需要自己手動調用，非自訂rank會自動調用）

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
}

feature.go 自訂 feature 樣本

package feature

import (
	"fmt"

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

// LoadRealTimeFeatures 自訂特徵載入函數
func LoadRealTimeFeatures(user *module.User, items []*module.Item, context *context.RecommendContext) {
	// 向 user 增加
	user.AddProperty("userAge", 30)

	// 向 item 增加
	for _, item := range items {
		item.AddProperty("count", 5)
	}
}

// MyFeatureFunc 自訂特徵處理
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, len(items), 0)
	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]
	}
}

main.go 主函數，註冊各種自訂

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()
}

config.json 配置樣本

{
	"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"],
				"SortName":["mySort"]
			}
		}
	},
	"LogConf": {
	  "RetensionDays": 3,
	  "DiskSize": 20,
	  "LogLevel": "INFO"
	},
	"RankConf": {
	},
	"FeatureConfs": {
	}
}