diff --git a/sdk/python/feast/cli/cli.py b/sdk/python/feast/cli/cli.py index 7826dfb650a..3b4e7b782b8 100644 --- a/sdk/python/feast/cli/cli.py +++ b/sdk/python/feast/cli/cli.py @@ -480,6 +480,7 @@ def materialize_incremental_command( "ray_rag", "rag", "pytorch_nlp", + "recommendation", ], case_sensitive=False, ), diff --git a/sdk/python/feast/templates/recommendation/__init__.py b/sdk/python/feast/templates/recommendation/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/sdk/python/feast/templates/recommendation/bootstrap.py b/sdk/python/feast/templates/recommendation/bootstrap.py new file mode 100644 index 00000000000..50afc3b07d4 --- /dev/null +++ b/sdk/python/feast/templates/recommendation/bootstrap.py @@ -0,0 +1,149 @@ +from feast.file_utils import replace_str_in_file + + +def bootstrap() -> None: + import pathlib + from datetime import datetime, timedelta + + import numpy as np + import pandas as pd + + repo_path = pathlib.Path(__file__).parent.absolute() / "feature_repo" + project_name = pathlib.Path(__file__).parent.absolute().name + data_path = repo_path / "data" + data_path.mkdir(exist_ok=True) + + products = [ + ( + "P001", + "Wireless Noise-Cancelling Headphones", + "Premium over-ear headphones with active noise cancellation and 30-hour battery life.", + "Electronics", + 299.99, + 4.7, + ), + ( + "P002", + "Bluetooth Portable Speaker", + "Waterproof portable speaker with deep bass and 12-hour playtime.", + "Electronics", + 79.99, + 4.5, + ), + ( + "P003", + "Mechanical Gaming Keyboard", + "RGB mechanical keyboard with Cherry MX switches and programmable keys.", + "Electronics", + 149.99, + 4.6, + ), + ( + "P004", + "Ergonomic Wireless Mouse", + "Vertical ergonomic mouse designed to reduce wrist strain.", + "Electronics", + 49.99, + 4.3, + ), + ( + "P005", + "Python Machine Learning Cookbook", + "Practical recipes for building ML models with scikit-learn and TensorFlow.", + "Books", + 39.99, + 4.4, + ), + ( + "P006", + "Data Engineering Fundamentals", + "Comprehensive guide to building modern data pipelines and architectures.", + "Books", + 44.99, + 4.6, + ), + ( + "P007", + "Introduction to Deep Learning", + "Beginner-friendly deep learning textbook with hands-on PyTorch examples.", + "Books", + 54.99, + 4.7, + ), + ( + "P008", + "Trail Running Shoes", + "Lightweight trail running shoes with superior grip and cushioning.", + "Sports", + 129.99, + 4.6, + ), + ( + "P009", + "Premium Yoga Mat", + "Non-slip extra-thick yoga mat with carrying strap.", + "Sports", + 34.99, + 4.4, + ), + ( + "P010", + "Resistance Bands Set", + "Set of 5 resistance bands with varying tension levels for home workouts.", + "Sports", + 24.99, + 4.3, + ), + ( + "P011", + "Robot Vacuum Cleaner", + "Self-navigating robot vacuum with app control and auto-charging.", + "Home", + 349.99, + 4.5, + ), + ( + "P012", + "Air Purifier with HEPA Filter", + "Room air purifier with true HEPA filter and air quality sensor.", + "Home", + 199.99, + 4.6, + ), + ] + + embedding_dim = 384 + embeddings = np.zeros((len(products), embedding_dim), dtype=np.float32) + embeddings[np.arange(len(products)), np.arange(len(products))] = 1.0 + + end_date = datetime.now().replace(microsecond=0, second=0, minute=0) + start_date = end_date - timedelta(days=15) + timestamps = [start_date + timedelta(hours=i) for i in range(len(products))] + + columns = [ + "product_id", + "product_name", + "description", + "category", + "price", + "rating", + ] + df = pd.DataFrame(products, columns=columns) + df["price"] = df["price"].astype(np.float32) + df["rating"] = df["rating"].astype(np.float32) + df["embedding"] = [embedding.tolist() for embedding in embeddings] + df["event_timestamp"] = timestamps + df["created"] = end_date + + products_path = data_path / "products.parquet" + df.to_parquet(path=str(products_path), allow_truncated_timestamps=True) + + example_py_file = repo_path / "feature_definitions.py" + replace_str_in_file(example_py_file, "%PROJECT_NAME%", str(project_name)) + replace_str_in_file( + example_py_file, "%PARQUET_PATH%", str(products_path.relative_to(repo_path)) + ) + + +if __name__ == "__main__": + bootstrap() diff --git a/sdk/python/feast/templates/recommendation/feature_repo/__init__.py b/sdk/python/feast/templates/recommendation/feature_repo/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/sdk/python/feast/templates/recommendation/feature_repo/feature_definitions.py b/sdk/python/feast/templates/recommendation/feature_repo/feature_definitions.py new file mode 100644 index 00000000000..bdb06191b55 --- /dev/null +++ b/sdk/python/feast/templates/recommendation/feature_repo/feature_definitions.py @@ -0,0 +1,43 @@ +from datetime import timedelta + +from feast import Entity, FeatureService, FeatureView, Field, FileSource, Project +from feast.types import Array, Float32, String + +project = Project( + name="%PROJECT_NAME%", + description="A project for product recommendations using vector similarity search", +) + +product = Entity(name="product", join_keys=["product_id"]) + +products_source = FileSource( + name="products_source", + path="%PARQUET_PATH%", + timestamp_field="event_timestamp", + created_timestamp_column="created", +) + +# test_workflow.py writes model embeddings before materialization. +product_embeddings = FeatureView( + name="product_embeddings", + entities=[product], + ttl=timedelta(days=365), + schema=[ + Field( + name="embedding", dtype=Array(Float32), vector_index=True, vector_length=384 + ), + Field(name="product_name", dtype=String), + Field(name="description", dtype=String), + Field(name="category", dtype=String), + Field(name="price", dtype=Float32), + Field(name="rating", dtype=Float32), + ], + online=True, + source=products_source, + tags={"team": "recommendations"}, +) + +recommendation_service = FeatureService( + name="recommendation_service", + features=[product_embeddings], +) diff --git a/sdk/python/feast/templates/recommendation/feature_repo/feature_store.yaml b/sdk/python/feast/templates/recommendation/feature_repo/feature_store.yaml new file mode 100644 index 00000000000..b213490d5b9 --- /dev/null +++ b/sdk/python/feast/templates/recommendation/feature_repo/feature_store.yaml @@ -0,0 +1,13 @@ +project: my_project +# By default, the registry is a file (but can be turned into a more scalable SQL-backed registry) +registry: data/registry.db +# The provider primarily specifies default offline / online stores & storing the registry in a given cloud +provider: local +online_store: + type: sqlite + path: data/online_store.db + vector_enabled: true +entity_key_serialization_version: 3 +# By default, no_auth for authentication and authorization, other possible values kubernetes and oidc. Refer the documentation for more details. +auth: + type: no_auth diff --git a/sdk/python/feast/templates/recommendation/feature_repo/test_workflow.py b/sdk/python/feast/templates/recommendation/feature_repo/test_workflow.py new file mode 100644 index 00000000000..46abafb5ccd --- /dev/null +++ b/sdk/python/feast/templates/recommendation/feature_repo/test_workflow.py @@ -0,0 +1,70 @@ +import subprocess +import sys +from datetime import datetime +from pathlib import Path + +import numpy as np +import pandas as pd + +from feast import FeatureStore + + +def run_demo() -> None: + try: + from sentence_transformers import SentenceTransformer + except ImportError: + print("sentence-transformers is required: pip install sentence-transformers") + sys.exit(1) + + model = SentenceTransformer("all-MiniLM-L6-v2") + products_path = Path("data/products.parquet") + products = pd.read_parquet(products_path) + descriptions = (products["product_name"] + ". " + products["description"]).tolist() + embeddings = model.encode(descriptions, normalize_embeddings=True) + products["embedding"] = [ + embedding.astype(np.float32).tolist() for embedding in embeddings + ] + products.to_parquet(products_path, allow_truncated_timestamps=True) + + print("\nApplying feature definitions") + subprocess.run(["feast", "apply"], check=True) + + print("\nLoading features into the online store") + store = FeatureStore(repo_path=".") + store.materialize_incremental(end_date=datetime.now()) + + query = "gaming laptop accessories" + print(f"\nSearching for: {query}") + query_embedding = model.encode([query], normalize_embeddings=True)[0].tolist() + + results = store.retrieve_online_documents_v2( + features=[ + "product_embeddings:embedding", + "product_embeddings:product_id", + "product_embeddings:product_name", + "product_embeddings:category", + "product_embeddings:price", + "product_embeddings:rating", + ], + query=query_embedding, + top_k=5, + ).to_dict() + + if results and len(results.get("product_id", [])) > 0: + num_results = len(results["product_id"]) + print(f" Top {num_results} recommendations:") + for i in range(num_results): + name = results["product_name"][i] + category = results["category"][i] + price = results["price"][i] + rating = results["rating"][i] + print(f" {i + 1}. {name} [{category}] - ${price:.2f} (rating: {rating})") + else: + print(" No results found.") + + print("\nTearing down the feature store") + subprocess.run(["feast", "teardown"], check=True) + + +if __name__ == "__main__": + run_demo() diff --git a/sdk/python/tests/unit/local_feast_tests/test_init.py b/sdk/python/tests/unit/local_feast_tests/test_init.py index 1f3c0993845..0d3649f1185 100644 --- a/sdk/python/tests/unit/local_feast_tests/test_init.py +++ b/sdk/python/tests/unit/local_feast_tests/test_init.py @@ -3,6 +3,9 @@ from pathlib import Path from textwrap import dedent +import pandas as pd + +from feast.repo_operations import init_repo from feast.utils import _utc_now from tests.utils.cli_repo_creator import CliRunner @@ -81,3 +84,32 @@ def test_postgres_template_registry_path_is_parameterized() -> None: contents = template_fs_yaml.read_text(encoding="utf-8") expected = "path: postgresql://DB_USERNAME:DB_PASSWORD@DB_HOST:DB_PORT/DB_NAME" assert expected in contents + + +def test_recommendation_template_init(tmp_path: Path) -> None: + repo_path = tmp_path / "recommendation_project" + + init_repo( + "recommendation_project", + template="recommendation", + repo_path=str(repo_path), + ) + + feature_repo = repo_path / "feature_repo" + config = (feature_repo / "feature_store.yaml").read_text(encoding="utf-8") + assert "project: recommendation_project" in config + assert "type: sqlite" in config + assert "vector_enabled: true" in config + + definitions = (feature_repo / "feature_definitions.py").read_text(encoding="utf-8") + assert "vector_index=True" in definitions + assert "vector_length=384" in definitions + assert 'name="recommendation_project"' in definitions + assert 'path="data/products.parquet"' in definitions + + products_path = feature_repo / "data" / "products.parquet" + assert products_path.is_file() + products = pd.read_parquet(products_path) + assert len(products) == 12 + assert all(len(embedding) == 384 for embedding in products["embedding"]) + assert not (repo_path / "bootstrap.py").exists()