diff --git a/PRODUCT_RECOMMENDER.md b/PRODUCT_RECOMMENDER.md new file mode 100644 index 0000000..c73de47 --- /dev/null +++ b/PRODUCT_RECOMMENDER.md @@ -0,0 +1,50 @@ +# Product Catalog Recommender Domain + +This example adapts the AO Labs recommender pattern to an e-commerce product catalog. It uses a local product dataset, deterministic feature encoding, and a fallback ranker so reviewers can validate the domain without API keys or private AO packages. + +## Domain + +The catalog covers electronics, home, outdoor, apparel, wellness, and office products. Each product includes: + +- category +- price tier +- sustainability flag +- brand style +- shipping speed +- use case +- shopper intent +- description + +## AO Input Encoding + +The product recommender keeps the same 8-bit input shape used by the base recommender architecture: + +| Segment | Bits | Source | +| --- | ---: | --- | +| Category | 3 | Product category | +| Price tier | 2 | Product price tier | +| Sustainability | 1 | Whether the product matches a sustainability preference | +| Use case | 2 | Shopper context | + +This maps to `arch_i = [3, 2, 1, 2]` in `arch__ProductRecommender.py`. + +## Run the Demo + +Streamlit UI: + +```bash +streamlit run product_recommender.py +``` + +CLI fallback: + +```bash +python3 product_recommender.py +``` + +## Validate + +```bash +python3 -m unittest tests/test_product_domain.py +python3 -m py_compile product_domain.py product_recommender.py arch__ProductRecommender.py tests/test_product_domain.py +``` diff --git a/README.md b/README.md index 26cb9cc..95e1e1c 100644 --- a/README.md +++ b/README.md @@ -55,10 +55,13 @@ You're done! Access the app at `localhost:8501` in your browser. The recommender system works by loading a set of random video links. Once the user hits the Run button, a video will be shown, and the system will suggest whether it recommends the video or not. The user can then provide feedback using "pain" or "pleasure" signals to guide the recommendation process. Based on this feedback, the system adjusts its responses and suggests another video. This cycle continues, allowing for more accurate and personalized recommendations over time. +## Additional Domain Examples + +- [Product Catalog Recommender](PRODUCT_RECOMMENDER.md): e-commerce recommendations with a local product catalog, AO-compatible 8-bit encoding, Streamlit UI, and deterministic fallback tests. + ## Contributing Fork the repository, make your changes, and submit a pull request for review. - diff --git a/arch__ProductRecommender.py b/arch__ProductRecommender.py new file mode 100644 index 0000000..99f03e1 --- /dev/null +++ b/arch__ProductRecommender.py @@ -0,0 +1,17 @@ +# -*- coding: utf-8 -*- +""" +AO architecture for the product catalog recommender example. +""" + +import ao_arch as ar + + +description = "Product Catalog Recommender" + +# category + price tier + sustainability preference match + shopper use case +arch_i = [3, 2, 1, 2] +arch_z = [10] +arch_c = [] +connector_function = "full_conn" + +arch = ar.Arch(arch_i, arch_z, arch_c, connector_function, description) diff --git a/product_domain.py b/product_domain.py new file mode 100644 index 0000000..555bdb8 --- /dev/null +++ b/product_domain.py @@ -0,0 +1,255 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Iterable + + +CATEGORIES = [ + "electronics", + "home", + "outdoor", + "apparel", + "wellness", + "office", +] +PRICE_TIERS = ["budget", "midrange", "premium"] +USE_CASES = ["daily", "gift", "travel", "work"] + + +@dataclass(frozen=True) +class Product: + name: str + category: str + price_tier: str + sustainable: bool + brand_style: str + shipping_speed: str + use_case: str + shopper_intent: str + description: str + + +@dataclass(frozen=True) +class ShopperContext: + category: str = "electronics" + max_price_tier: str = "premium" + prefers_sustainable: bool = False + use_case: str = "daily" + style: str = "minimal" + shipping_speed: str = "standard" + + +PRODUCT_CATALOG = [ + Product( + name="Refurbished Noise-Canceling Headphones", + category="electronics", + price_tier="midrange", + sustainable=True, + brand_style="minimal", + shipping_speed="fast", + use_case="travel", + shopper_intent="focus", + description="Certified refurbished headphones for calls, travel, and deep work.", + ), + Product( + name="Compact USB-C Travel Charger", + category="electronics", + price_tier="budget", + sustainable=False, + brand_style="minimal", + shipping_speed="fast", + use_case="travel", + shopper_intent="utility", + description="Small multi-port charger for laptops, phones, and tablets.", + ), + Product( + name="Modular Desk Lamp", + category="home", + price_tier="midrange", + sustainable=True, + brand_style="modern", + shipping_speed="standard", + use_case="work", + shopper_intent="comfort", + description="Repairable LED desk lamp with warm and cool light modes.", + ), + Product( + name="Organic Cotton Throw Blanket", + category="home", + price_tier="premium", + sustainable=True, + brand_style="warm", + shipping_speed="standard", + use_case="gift", + shopper_intent="comfort", + description="Soft, durable blanket made from certified organic cotton.", + ), + Product( + name="Trail Daypack", + category="outdoor", + price_tier="midrange", + sustainable=True, + brand_style="rugged", + shipping_speed="fast", + use_case="travel", + shopper_intent="adventure", + description="Lightweight daypack with recycled fabric and hydration sleeve.", + ), + Product( + name="Insulated Steel Bottle", + category="outdoor", + price_tier="budget", + sustainable=True, + brand_style="minimal", + shipping_speed="fast", + use_case="daily", + shopper_intent="utility", + description="Leak-proof bottle for commuting, gym bags, and weekend hikes.", + ), + Product( + name="Merino Travel Hoodie", + category="apparel", + price_tier="premium", + sustainable=True, + brand_style="minimal", + shipping_speed="standard", + use_case="travel", + shopper_intent="comfort", + description="Odor-resistant hoodie designed for one-bag travel.", + ), + Product( + name="Everyday Canvas Tote", + category="apparel", + price_tier="budget", + sustainable=True, + brand_style="casual", + shipping_speed="fast", + use_case="daily", + shopper_intent="utility", + description="Reusable tote for errands, groceries, and books.", + ), + Product( + name="Adjustable Yoga Mat", + category="wellness", + price_tier="midrange", + sustainable=True, + brand_style="calm", + shipping_speed="standard", + use_case="daily", + shopper_intent="routine", + description="Non-slip mat with alignment guides for home workouts.", + ), + Product( + name="Sleep Wind-Down Kit", + category="wellness", + price_tier="premium", + sustainable=False, + brand_style="calm", + shipping_speed="standard", + use_case="gift", + shopper_intent="comfort", + description="Eye mask, lavender spray, and guided sleep cards.", + ), + Product( + name="Recycled Paper Notebook Set", + category="office", + price_tier="budget", + sustainable=True, + brand_style="minimal", + shipping_speed="fast", + use_case="work", + shopper_intent="organization", + description="Three ruled notebooks made from post-consumer paper.", + ), + Product( + name="Ergonomic Monitor Stand", + category="office", + price_tier="midrange", + sustainable=False, + brand_style="modern", + shipping_speed="fast", + use_case="work", + shopper_intent="comfort", + description="Aluminum stand with storage space for keyboard and notes.", + ), +] + + +def _binary_index(value: str, options: list[str], width: int) -> list[int]: + if value not in options: + raise ValueError(f"Unknown value {value!r}; expected one of {options}") + + index = options.index(value) + return [int(bit) for bit in format(index, f"0{width}b")] + + +def encode_product(product: Product, context: ShopperContext) -> list[int]: + """Encode product + shopper context into the existing 8-bit AO input shape.""" + + category_bits = _binary_index(product.category, CATEGORIES, 3) + price_bits = _binary_index(product.price_tier, PRICE_TIERS, 2) + sustainability_bit = [int(product.sustainable and context.prefers_sustainable)] + use_case_bits = _binary_index(context.use_case, USE_CASES, 2) + return category_bits + price_bits + sustainability_bit + use_case_bits + + +def filter_products( + catalog: Iterable[Product], + context: ShopperContext, +) -> list[Product]: + max_price_index = PRICE_TIERS.index(context.max_price_tier) + + return [ + product + for product in catalog + if product.category == context.category + and PRICE_TIERS.index(product.price_tier) <= max_price_index + ] + + +def score_product( + product: Product, + context: ShopperContext, + feedback: dict[str, int] | None = None, +) -> float: + score = 0.0 + + if product.category == context.category: + score += 0.28 + if PRICE_TIERS.index(product.price_tier) <= PRICE_TIERS.index(context.max_price_tier): + score += 0.16 + if product.use_case == context.use_case: + score += 0.18 + if product.sustainable and context.prefers_sustainable: + score += 0.14 + if product.brand_style == context.style: + score += 0.10 + if product.shipping_speed == context.shipping_speed: + score += 0.08 + if context.use_case in {product.use_case, product.shopper_intent}: + score += 0.06 + + if feedback: + score += max(-0.2, min(0.2, feedback.get(product.name, 0) * 0.05)) + + return round(score, 4) + + +def recommend_products( + context: ShopperContext, + catalog: Iterable[Product] = PRODUCT_CATALOG, + feedback: dict[str, int] | None = None, + limit: int = 3, +) -> list[tuple[Product, float, list[int]]]: + candidates = filter_products(catalog, context) + if not candidates: + candidates = list(catalog) + + ranked = sorted( + ( + (product, score_product(product, context, feedback), encode_product(product, context)) + for product in candidates + ), + key=lambda item: (-item[1], item[0].name), + ) + return ranked[:limit] diff --git a/product_recommender.py b/product_recommender.py new file mode 100644 index 0000000..a3989c6 --- /dev/null +++ b/product_recommender.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +from product_domain import ( + CATEGORIES, + PRICE_TIERS, + USE_CASES, + PRODUCT_CATALOG, + ShopperContext, + recommend_products, +) + + +def running_inside_streamlit() -> bool: + try: + from streamlit.runtime.scriptrunner import get_script_run_ctx + + return get_script_run_ctx() is not None + except ModuleNotFoundError: + return False + + +def run_cli_demo() -> None: + context = ShopperContext( + category="electronics", + max_price_tier="midrange", + prefers_sustainable=True, + use_case="travel", + style="minimal", + shipping_speed="fast", + ) + + for product, score, bits in recommend_products(context, PRODUCT_CATALOG): + print(f"{score:.2f} {bits} {product.name} - {product.description}") + + +def run_streamlit_app() -> None: + import streamlit as st + + st.set_page_config(page_title="AO Product Recommender", page_icon="🛒") + st.title("AO Product Catalog Recommender") + st.caption("A local e-commerce recommender domain for continuously trainable AO agents.") + + category = st.selectbox("Shopping category", CATEGORIES) + max_price_tier = st.selectbox("Maximum price tier", PRICE_TIERS, index=1) + use_case = st.selectbox("Shopping use case", USE_CASES) + prefers_sustainable = st.checkbox("Prefer sustainable products", value=True) + style = st.selectbox( + "Preferred style", + ["minimal", "modern", "warm", "rugged", "casual", "calm"], + ) + shipping_speed = st.selectbox("Shipping preference", ["standard", "fast"], index=1) + + if "product_feedback" not in st.session_state: + st.session_state.product_feedback = {} + + context = ShopperContext( + category=category, + max_price_tier=max_price_tier, + prefers_sustainable=prefers_sustainable, + use_case=use_case, + style=style, + shipping_speed=shipping_speed, + ) + + st.subheader("Recommendations") + for product, score, bits in recommend_products( + context, + PRODUCT_CATALOG, + feedback=st.session_state.product_feedback, + ): + with st.container(border=True): + st.markdown(f"### {product.name}") + st.write(product.description) + st.write( + { + "score": score, + "ao_binary_input": bits, + "category": product.category, + "price_tier": product.price_tier, + "sustainable": product.sustainable, + "shipping_speed": product.shipping_speed, + "use_case": product.use_case, + } + ) + + col_like, col_dislike = st.columns(2) + with col_like: + if st.button("Recommend more", key=f"like-{product.name}"): + st.session_state.product_feedback[product.name] = ( + st.session_state.product_feedback.get(product.name, 0) + 1 + ) + st.rerun() + with col_dislike: + if st.button("Show less", key=f"dislike-{product.name}"): + st.session_state.product_feedback[product.name] = ( + st.session_state.product_feedback.get(product.name, 0) - 1 + ) + st.rerun() + + +if __name__ == "__main__": + if running_inside_streamlit(): + run_streamlit_app() + else: + run_cli_demo() diff --git a/tests/test_product_domain.py b/tests/test_product_domain.py new file mode 100644 index 0000000..4bea78b --- /dev/null +++ b/tests/test_product_domain.py @@ -0,0 +1,92 @@ +import unittest + +from product_domain import ( + PRODUCT_CATALOG, + Product, + ShopperContext, + encode_product, + filter_products, + recommend_products, + score_product, +) + + +class ProductDomainTest(unittest.TestCase): + def test_encode_product_matches_ao_input_shape(self) -> None: + context = ShopperContext( + category="electronics", + max_price_tier="midrange", + prefers_sustainable=True, + use_case="travel", + ) + + bits = encode_product(PRODUCT_CATALOG[0], context) + + self.assertEqual(len(bits), 8) + self.assertTrue(all(bit in {0, 1} for bit in bits)) + + def test_filter_products_respects_category_and_max_price(self) -> None: + context = ShopperContext(category="electronics", max_price_tier="budget") + + products = filter_products(PRODUCT_CATALOG, context) + + self.assertEqual([product.name for product in products], ["Compact USB-C Travel Charger"]) + + def test_sustainable_travel_context_ranks_headphones_first(self) -> None: + context = ShopperContext( + category="electronics", + max_price_tier="midrange", + prefers_sustainable=True, + use_case="travel", + style="minimal", + shipping_speed="fast", + ) + + recommendations = recommend_products(context, PRODUCT_CATALOG) + + self.assertEqual(recommendations[0][0].name, "Refurbished Noise-Canceling Headphones") + self.assertGreater(recommendations[0][1], recommendations[1][1]) + + def test_feedback_can_change_ranking(self) -> None: + context = ShopperContext( + category="electronics", + max_price_tier="midrange", + prefers_sustainable=True, + use_case="travel", + style="minimal", + shipping_speed="fast", + ) + + feedback = { + "Refurbished Noise-Canceling Headphones": -5, + "Compact USB-C Travel Charger": 5, + } + recommendations = recommend_products(context, PRODUCT_CATALOG, feedback=feedback) + + self.assertEqual(recommendations[0][0].name, "Compact USB-C Travel Charger") + + def test_unknown_category_raises_helpful_error(self) -> None: + product = Product( + name="Unknown", + category="not-a-category", + price_tier="budget", + sustainable=False, + brand_style="minimal", + shipping_speed="fast", + use_case="daily", + shopper_intent="utility", + description="Invalid test product", + ) + + with self.assertRaisesRegex(ValueError, "Unknown value"): + encode_product(product, ShopperContext()) + + def test_score_is_deterministic(self) -> None: + context = ShopperContext(category="office", max_price_tier="midrange", use_case="work") + product = next(item for item in PRODUCT_CATALOG if item.name == "Ergonomic Monitor Stand") + + self.assertEqual(score_product(product, context), score_product(product, context)) + + +if __name__ == "__main__": + unittest.main()