Wyvern makes it so much easier to define and auto retrieve features you need as well as connecting to your model, whether it’s a model that’s served from another service or a model that’s hosted locally within the wyvern pipeline service.

Wyvern has a ModelComponent which is the base component that takes in the inference request and a list of entities, and outputs a model output. The model output is a dictionary mapping entity identifiers to model outputs.

Let’s see how ModelComponent actually works

​
Model Inference

ModelComponent has an inference function which is the main entrance to model evaluation.

By default, the base ModelComponent slices entities into smaller batches and call batch_inference on each batch. The default batch size is 30. You should be able to configure the MODEL_BATCH_SIZE env variable to change the batch size.

In order to set up model inference, you only need to define a class that inherits ModelComponent and implement batch_inference.

You can also override this function if you want to customize the inference logic.

​
Register features for the model

Here’s an example of defining features you need for your model inference:

    @cached_property
    def manifest_feature_names(self) -> Set[str]:
        return {
            "RealtimeProductFeature:search_score",
            "RealtimeProductQueryFeature:query_category_similarity",
            "RealtimeProductQueryFeature:query_brand_similarity",
            "RealtimeProductQueryFeature:query_title_similarity",
            "RealtimeProductQueryFeature:query_description_similarity",
        }

The manifest_feature_names is a @cached_property (you can also just do @property but for cached_property gives a little performance advantage) and is a set of feature strings, with each string in the following format:

  • for Wyvern’s real-time features: realtime_feature_component_name + : + feature_name. The realtime_feature_component_name is the name of the RealtimeFeatureComponent you defined. By default, it is the name of your model class.
  • for batch features: FEATURE_VIEW + : + FEATURE_NAME

Wyvern’s framework will use features defined under manifest_feature_names

​
Define the whole model

​
Single Model

pipelines/product_ranking/models.py
import asyncio
from functools import cached_property
from typing import List, Set

from wyvern import (
    CompositeIdentifier,
    ModelComponent,
    ModelInput,
    ModelOutput,
    RankingRequest,
)

from pipelines.product_ranking.schemas import Product


class RankingModel(
    ModelComponent[
        ModelInput[
            Product,
            RankingRequest[Product],
        ],
        ModelOutput[float],
    ],
):
    @cached_property
    def manifest_feature_names(self) -> Set[str]:
        return {
            "RealtimeProductFeature:search_score",
            "RealtimeProductQueryFeature:query_category_similarity",
            "RealtimeProductQueryFeature:query_brand_similarity",
            "RealtimeProductQueryFeature:query_title_similarity",
            "RealtimeProductQueryFeature:query_description_similarity",
        }

    async def batch_inference(
        self,
        request: RankingRequest[Product],
        entities: List[Product],
        **kwargs,
    ) -> List[float]:
        return await asyncio.gather(
            *[self._inference_helper(request, entity) for entity in entities]
        )

    async def _inference_helper(
        self,
        request: RankingRequest[Product],
        entity: Product,
    ) -> float:
        score = 0.0
        composite_identifier = CompositeIdentifier(
            primary_identifier=entity.identifier,
            secondary_identifier=request.query.identifier,
        )
        search_score = self.get_feature(
            entity.identifier,
            "RealtimeProductFeature:search_score",
        )
        query_category_similarity = self.get_feature(
            composite_identifier,
            "RealtimeProductQueryFeature:query_category_similarity",
        )
        query_brand_similarity = self.get_feature(
            composite_identifier,
            "RealtimeProductQueryFeature:query_brand_similarity",
        )
        query_title_similarity = self.get_feature(
            composite_identifier,
            "RealtimeProductQueryFeature:query_title_similarity",
        )
        query_description_similarity = self.get_feature(
            composite_identifier,
            "RealtimeProductQueryFeature:query_description_similarity",
        )
        if search_score:
            score += 2 * search_score
        if query_category_similarity:
            score += 10 * query_category_similarity
        if query_brand_similarity:
            score += 20 * query_brand_similarity
        if query_title_similarity:
            score += 22 * query_title_similarity
        if query_description_similarity:
            score += 13 * query_description_similarity

        return score

As you could see in the example, the RankingModel inherits ModelComponent and defines manifest_feature_names and batch_inference. The batch_inference defines the behavior of model for batch inference. In this example of the _inference_helper, each inference is pretty much handling the formula we’ve talked about, which is:

model_score =
    2 * search_score
    + 10 * query_category_similarity
    + 20 * query_brand_similarity
    + 22 * query_title_similarity
    + 13 * query_description_similarity

​
ModelInput

This is the basic input format of a ModelComponent.

It contains the request information and the list of entities that’s related to the model

Reference: https://docs.wyvern.ai/sdk_ref#modelinput-objects

After the model is defined, now let’s go back to “Wyvern Pipeline” and see how to set up a pipeline in the code in the next page.

Wyvern makes it so much easier to define and auto retrieve features you need as well as connecting to your model, whether it’s a model that’s served from another service or a model that’s hosted locally within the wyvern pipeline service.

Wyvern has a ModelComponent which is the base component that takes in the inference request and a list of entities, and outputs a model output. The model output is a dictionary mapping entity identifiers to model outputs.

Let’s see how ModelComponent actually works

​
Model Inference

ModelComponent has an inference function which is the main entrance to model evaluation.

By default, the base ModelComponent slices entities into smaller batches and call batch_inference on each batch. The default batch size is 30. You should be able to configure the MODEL_BATCH_SIZE env variable to change the batch size.

In order to set up model inference, you only need to define a class that inherits ModelComponent and implement batch_inference.

You can also override this function if you want to customize the inference logic.

​
Register features for the model

Here’s an example of defining features you need for your model inference:

    @cached_property
    def manifest_feature_names(self) -> Set[str]:
        return {
            "RealtimeProductFeature:search_score",
            "RealtimeProductQueryFeature:query_category_similarity",
            "RealtimeProductQueryFeature:query_brand_similarity",
            "RealtimeProductQueryFeature:query_title_similarity",
            "RealtimeProductQueryFeature:query_description_similarity",
        }

The manifest_feature_names is a @cached_property (you can also just do @property but for cached_property gives a little performance advantage) and is a set of feature strings, with each string in the following format:

  • for Wyvern’s real-time features: realtime_feature_component_name + : + feature_name. The realtime_feature_component_name is the name of the RealtimeFeatureComponent you defined. By default, it is the name of your model class.
  • for batch features: FEATURE_VIEW + : + FEATURE_NAME

Wyvern’s framework will use features defined under manifest_feature_names

​
Define the whole model

​
Single Model

pipelines/product_ranking/models.py
import asyncio
from functools import cached_property
from typing import List, Set

from wyvern import (
    CompositeIdentifier,
    ModelComponent,
    ModelInput,
    ModelOutput,
    RankingRequest,
)

from pipelines.product_ranking.schemas import Product


class RankingModel(
    ModelComponent[
        ModelInput[
            Product,
            RankingRequest[Product],
        ],
        ModelOutput[float],
    ],
):
    @cached_property
    def manifest_feature_names(self) -> Set[str]:
        return {
            "RealtimeProductFeature:search_score",
            "RealtimeProductQueryFeature:query_category_similarity",
            "RealtimeProductQueryFeature:query_brand_similarity",
            "RealtimeProductQueryFeature:query_title_similarity",
            "RealtimeProductQueryFeature:query_description_similarity",
        }

    async def batch_inference(
        self,
        request: RankingRequest[Product],
        entities: List[Product],
        **kwargs,
    ) -> List[float]:
        return await asyncio.gather(
            *[self._inference_helper(request, entity) for entity in entities]
        )

    async def _inference_helper(
        self,
        request: RankingRequest[Product],
        entity: Product,
    ) -> float:
        score = 0.0
        composite_identifier = CompositeIdentifier(
            primary_identifier=entity.identifier,
            secondary_identifier=request.query.identifier,
        )
        search_score = self.get_feature(
            entity.identifier,
            "RealtimeProductFeature:search_score",
        )
        query_category_similarity = self.get_feature(
            composite_identifier,
            "RealtimeProductQueryFeature:query_category_similarity",
        )
        query_brand_similarity = self.get_feature(
            composite_identifier,
            "RealtimeProductQueryFeature:query_brand_similarity",
        )
        query_title_similarity = self.get_feature(
            composite_identifier,
            "RealtimeProductQueryFeature:query_title_similarity",
        )
        query_description_similarity = self.get_feature(
            composite_identifier,
            "RealtimeProductQueryFeature:query_description_similarity",
        )
        if search_score:
            score += 2 * search_score
        if query_category_similarity:
            score += 10 * query_category_similarity
        if query_brand_similarity:
            score += 20 * query_brand_similarity
        if query_title_similarity:
            score += 22 * query_title_similarity
        if query_description_similarity:
            score += 13 * query_description_similarity

        return score

As you could see in the example, the RankingModel inherits ModelComponent and defines manifest_feature_names and batch_inference. The batch_inference defines the behavior of model for batch inference. In this example of the _inference_helper, each inference is pretty much handling the formula we’ve talked about, which is:

model_score =
    2 * search_score
    + 10 * query_category_similarity
    + 20 * query_brand_similarity
    + 22 * query_title_similarity
    + 13 * query_description_similarity

​
ModelInput

This is the basic input format of a ModelComponent.

It contains the request information and the list of entities that’s related to the model

Reference: https://docs.wyvern.ai/sdk_ref#modelinput-objects

After the model is defined, now let’s go back to “Wyvern Pipeline” and see how to set up a pipeline in the code in the next page.