tessl/pypi-google-cloud-retail
Python client library for Google Cloud Retail API enabling end-to-end personalized recommendation systems
—
Pending
ai-recommendations.mddocs/
AI-Powered Recommendations
Machine learning-based prediction and recommendation services for personalized customer experiences. The Prediction and Model Services provide intelligent product recommendations, model management, and ML model training capabilities powered by Google's retail AI.
Capabilities
Prediction Service
AI-powered recommendations and predictions based on user behavior and product catalog.
class PredictionServiceClient:
def predict(self, request: PredictRequest) -> PredictResponse:
"""
Gets personalized product recommendations using trained ML models.
Args:
request: Prediction parameters including placement, user context, and model settings
Returns:
PredictResponse: Recommended products with scores and metadata
Raises:
InvalidArgument: If prediction parameters are invalid
NotFound: If placement or model not found
"""Model Management
Comprehensive ML model lifecycle management including creation, training, and deployment.
class ModelServiceClient:
def create_model(self, request: CreateModelRequest) -> Operation:
"""
Creates a new recommendation model (long-running operation).
Args:
request: Model configuration including type, training parameters, and data requirements
Returns:
Operation: Resolves to Model when training completes
"""
def get_model(self, request: GetModelRequest) -> Model:
"""
Retrieves model information and training status.
Args:
request: Contains model resource name
Returns:
Model: Model configuration, status, and performance metrics
"""
def list_models(self, request: ListModelsRequest) -> ListModelsResponse:
"""
Lists all models in a catalog with filtering and pagination.
Args:
request: Contains parent catalog and optional filtering parameters
Returns:
ListModelsResponse: List of models with pagination support
"""
def update_model(self, request: UpdateModelRequest) -> Model:
"""
Updates an existing model's configuration.
Args:
request: Contains model updates and field mask
Returns:
Model: Updated model information
"""
def delete_model(self, request: DeleteModelRequest) -> None:
"""
Deletes a model and stops serving predictions.
Args:
request: Contains model resource name
"""
def pause_model(self, request: PauseModelRequest) -> Model:
"""
Pauses model serving without deleting the model.
Args:
request: Contains model resource name
Returns:
Model: Updated model with paused status
"""
def resume_model(self, request: ResumeModelRequest) -> Model:
"""
Resumes a paused model for serving predictions.
Args:
request: Contains model resource name
Returns:
Model: Updated model with active status
"""
def tune_model(self, request: TuneModelRequest) -> Operation:
"""
Fine-tunes an existing model with new data (long-running operation).
Args:
request: Contains model name and tuning parameters
Returns:
Operation: Resolves to TuneModelResponse when tuning completes
"""
class PredictionServiceAsyncClient:
async def predict(self, request: PredictRequest) -> PredictResponse:
"""Async version of predict."""
class ModelServiceAsyncClient:
async def create_model(self, request: CreateModelRequest) -> Operation:
"""Async version of create_model."""
async def get_model(self, request: GetModelRequest) -> Model:
"""Async version of get_model."""
async def list_models(self, request: ListModelsRequest) -> ListModelsResponse:
"""Async version of list_models."""
async def update_model(self, request: UpdateModelRequest) -> Model:
"""Async version of update_model."""
async def delete_model(self, request: DeleteModelRequest) -> None:
"""Async version of delete_model."""
async def pause_model(self, request: PauseModelRequest) -> Model:
"""Async version of pause_model."""
async def resume_model(self, request: ResumeModelRequest) -> Model:
"""Async version of resume_model."""
async def tune_model(self, request: TuneModelRequest) -> Operation:
"""Async version of tune_model."""Data Types
Prediction Request and Response
Configuration for getting personalized recommendations and the resulting predictions.
class PredictRequest:
placement: str # Placement resource name (required)
user_event: UserEvent # User context and current session information
page_size: int # Maximum predictions to return (default: 100, max: 1000)
page_token: str # Token for pagination
filter: str # Filter expression for recommended products
validate_only: bool # Validate request without serving predictions
params: Dict[str, Value] # Additional model parameters
labels: Dict[str, str] # Custom labels for analytics
class PredictResponse:
results: List[PredictResponsePredictionResult] # Prediction results
attribution_token: str # Token for attributing user actions
missing_ids: List[str] # Product IDs that couldn't be recommended
dry_run: bool # Whether this was a validation-only request
metadata: List[PredictResponsePredictionResult] # Additional metadata
class PredictResponsePredictionResult:
id: str # Recommended product ID
metadata: Dict[str, Value] # Prediction metadata and scoresModel Configuration
Comprehensive model definition with training parameters and performance metrics.
class Model:
name: str # Model resource name (read-only)
display_name: str # Human-readable model name
training_state: ModelTrainingState # TRAINING, PAUSED, READY
serving_state: ModelServingState # INACTIVE, ACTIVE, TUNED
create_time: Timestamp # Model creation time (read-only)
update_time: Timestamp # Last update time (read-only)
type: str # Model type (recommended-for-you, others-you-may-like, etc.)
optimization_objective: str # CTR, CVR, REVENUE_PER_ORDER
periodic_tuning_state: ModelPeriodicTuningState # Automatic retraining settings
last_tune_time: Timestamp # Last tuning/training time
tuning_operation: str # Current tuning operation name
data_state: ModelDataState # Data requirements satisfaction
filtering_option: RecommendationsFilteringOption # Filtering behavior
serving_config_lists: List[ModelServingConfigList] # Associated serving configurations
model_features_config: ModelModelFeaturesConfig # Feature configuration
class ModelServingConfigList:
serving_config_ids: List[str] # IDs of serving configurations using this model
class ModelModelFeaturesConfig:
frequently_bought_together_config: ModelFrequentlyBoughtTogetherFeaturesConfig # FBT feature settingsModel Training States and Metadata
Training status, performance metrics, and operational metadata.
class ModelTrainingState(Enum):
TRAINING_STATE_UNSPECIFIED = 0
PAUSED = 1 # Training paused, can be resumed
TRAINING = 2 # Currently training
class ModelServingState(Enum):
SERVING_STATE_UNSPECIFIED = 0
INACTIVE = 1 # Not serving predictions
ACTIVE = 2 # Actively serving predictions
TUNED = 3 # Recently tuned, ready for serving
class ModelDataState(Enum):
DATA_STATE_UNSPECIFIED = 0
DATA_OK = 1 # Sufficient data for training
DATA_ERROR = 2 # Insufficient or invalid data
class ModelPeriodicTuningState(Enum):
PERIODIC_TUNING_STATE_UNSPECIFIED = 0
PERIODIC_TUNING_DISABLED = 1 # No automatic retraining
ALL_TUNING_DISABLED = 2 # All tuning disabled
PERIODIC_TUNING_ENABLED = 3 # Automatic retraining enabledRequest Types
Model management operation requests with comprehensive configuration options.
class CreateModelRequest:
parent: str # Catalog resource name (required)
model: Model # Model configuration (required)
dry_run: bool # Validate request without creating model
class GetModelRequest:
name: str # Model resource name (required)
class ListModelsRequest:
parent: str # Catalog resource name (required)
page_size: int # Maximum models to return
page_token: str # Token for pagination
class UpdateModelRequest:
model: Model # Model with updates (required)
update_mask: FieldMask # Fields to update
class DeleteModelRequest:
name: str # Model resource name (required)
class PauseModelRequest:
name: str # Model resource name (required)
class ResumeModelRequest:
name: str # Model resource name (required)
class TuneModelRequest:
name: str # Model resource name (required)
class TuneModelResponse:
# Response contains tuning results and performance metricsModel Types and Optimization
Different recommendation model types and optimization objectives.
# Model Types (specified in Model.type field)
MODEL_TYPE_RECOMMENDED_FOR_YOU = "recommended-for-you" # Personalized recommendations
MODEL_TYPE_OTHERS_YOU_MAY_LIKE = "others-you-may-like" # Similar item recommendations
MODEL_TYPE_FREQUENTLY_BOUGHT_TOGETHER = "frequently-bought-together" # Bundle recommendations
MODEL_TYPE_PAGE_OPTIMIZATION = "page-optimization" # Page-level optimization
MODEL_TYPE_SIMILAR_ITEMS = "similar-items" # Content-based similarity
MODEL_TYPE_BUY_IT_AGAIN = "buy-it-again" # Repurchase recommendations
MODEL_TYPE_ON_SALE_ITEMS = "on-sale-items" # Promotional item recommendations
# Optimization Objectives
OPTIMIZATION_OBJECTIVE_REVENUE_PER_ORDER = "revenue-per-order" # Maximize revenue
OPTIMIZATION_OBJECTIVE_CVR = "cvr" # Maximize conversion rate
OPTIMIZATION_OBJECTIVE_CTR = "ctr" # Maximize click-through rateUsage Examples
Getting Product Recommendations
from google.cloud import retail
client = retail.PredictionServiceClient()
# Get personalized recommendations for a user
user_event = retail.UserEvent(
event_type="page-view",
visitor_id="visitor-123",
user_info=retail.UserInfo(
user_id="user-456",
ip_address="192.168.1.100"
),
product_details=[
retail.ProductDetail(
product=retail.Product(id="viewed-product-123"),
quantity=1
)
]
)
request = retail.PredictRequest(
placement="projects/my-project/locations/global/catalogs/default_catalog/placements/product_detail_page_recommended_for_you",
user_event=user_event,
page_size=10,
filter='(categories: ANY("Electronics"))'
)
response = client.predict(request=request)
print(f"Recommendations for user:")
for result in response.results:
print(f"- Product ID: {result.id}")
if 'score' in result.metadata:
print(f" Confidence Score: {result.metadata['score'].number_value:.3f}")Creating a Recommendation Model
model_client = retail.ModelServiceClient()
# Create a new recommendation model
model = retail.Model(
display_name="Homepage Recommendations Model",
type="recommended-for-you",
optimization_objective="revenue-per-order",
filtering_option=retail.RecommendationsFilteringOption.RECOMMENDATIONS_FILTERING_ENABLED,
periodic_tuning_state=retail.Model.PeriodicTuningState.PERIODIC_TUNING_ENABLED
)
request = retail.CreateModelRequest(
parent="projects/my-project/locations/global/catalogs/default_catalog",
model=model
)
# This is a long-running operation
operation = model_client.create_model(request=request)
print(f"Model creation operation: {operation.name}")
# Wait for model training to complete (this can take several hours)
print("Waiting for model training to complete...")
trained_model = operation.result()
print(f"Model created: {trained_model.name}")
print(f"Training state: {trained_model.training_state}")
print(f"Serving state: {trained_model.serving_state}")Managing Model Lifecycle
# List all models in catalog
request = retail.ListModelsRequest(
parent="projects/my-project/locations/global/catalogs/default_catalog",
page_size=20
)
response = model_client.list_models(request=request)
print("Available models:")
for model in response.models:
print(f"- {model.display_name}")
print(f" Name: {model.name}")
print(f" Type: {model.type}")
print(f" Training State: {model.training_state}")
print(f" Serving State: {model.serving_state}")
print(f" Last Tune Time: {model.last_tune_time}")
print()
# Get detailed model information
model_name = "projects/my-project/locations/global/catalogs/default_catalog/models/model-123"
request = retail.GetModelRequest(name=model_name)
model = model_client.get_model(request=request)
print(f"Model Details:")
print(f"Display Name: {model.display_name}")
print(f"Optimization Objective: {model.optimization_objective}")
print(f"Data State: {model.data_state}")
print(f"Filtering Option: {model.filtering_option}")
# Pause model serving
if model.serving_state == retail.Model.ServingState.ACTIVE:
pause_request = retail.PauseModelRequest(name=model_name)
paused_model = model_client.pause_model(request=pause_request)
print(f"Model paused. New serving state: {paused_model.serving_state}")
# Resume model serving
if model.serving_state == retail.Model.ServingState.INACTIVE:
resume_request = retail.ResumeModelRequest(name=model_name)
resumed_model = model_client.resume_model(request=resume_request)
print(f"Model resumed. New serving state: {resumed_model.serving_state}")Model Tuning and Retraining
# Fine-tune an existing model with recent data
tune_request = retail.TuneModelRequest(
name="projects/my-project/locations/global/catalogs/default_catalog/models/model-123"
)
# This is a long-running operation
tune_operation = model_client.tune_model(request=tune_request)
print(f"Model tuning operation: {tune_operation.name}")
# Monitor tuning progress
print("Waiting for model tuning to complete...")
tune_result = tune_operation.result()
print(f"Model tuning completed: {tune_result}")
# Check updated model status
updated_model = model_client.get_model(
request=retail.GetModelRequest(
name="projects/my-project/locations/global/catalogs/default_catalog/models/model-123"
)
)
print(f"Updated serving state: {updated_model.serving_state}")
print(f"Last tune time: {updated_model.last_tune_time}")Advanced Prediction Configuration
# Get recommendations with advanced filtering and custom parameters
user_event = retail.UserEvent(
event_type="add-to-cart",
visitor_id="visitor-123",
user_info=retail.UserInfo(
user_id="user-456",
ip_address="192.168.1.100",
user_agent="Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X)"
),
product_details=[
retail.ProductDetail(
product=retail.Product(id="cart-product-456"),
quantity=2
)
],
purchase_transaction=retail.PurchaseTransaction(
id="txn-789",
revenue=199.98,
tax=16.00,
cost=150.00,
currency_code="USD"
)
)
request = retail.PredictRequest(
placement="projects/my-project/locations/global/catalogs/default_catalog/placements/shopping_cart_frequently_bought_together",
user_event=user_event,
page_size=5,
filter='(price_info.price: [10, 500]) AND (availability: "IN_STOCK")',
params={
'returnProduct': {'bool_value': True},
'strictFiltering': {'bool_value': True}
},
labels={
'page_type': 'shopping_cart',
'ab_test_group': 'treatment_a'
}
)
response = client.predict(request=request)
print("Frequently bought together recommendations:")
for result in response.results:
print(f"- Product ID: {result.id}")
# Extract custom metadata
for key, value in result.metadata.items():
print(f" {key}: {value}")Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-retail