feat: add Search Personalization demo module

- Add search_personalization/ module with UBI tracking, agentic memory,
product catalog, and image utilities
- Add pages/Search_Personalization.py Streamlit page
- Update app.py: 2x2 card grid layout with new Personalization card,
defensive import and logo fallback, responsive CSS
- Update requirements.txt: add python-dotenv, strands-agents, PyYAML
- Update .gitattributes: add LFS for jpg/jpeg/png/pdf
- Update .gitignore: add .env and app_new/
- Add .env.example template

.env.example

@@ -0,0 +1,18 @@
+# AWS Credentials
+AWS_ACCESS_KEY_ID=
+AWS_SECRET_ACCESS_KEY=
+AWS_REGION=us-east-1
+
+# OpenSearch
+OPENSEARCH_ENDPOINT=
+OPENSEARCH_MODEL_ID=
+
+# Agentic Memory
+MEMORY_CONTAINER_ID=
+BEDROCK_MODEL_ID_FAST=
+
+# UBI (User Behavior Insights)
+UBI_EVENTS_ENDPOINT=
+UBI_QUERIES_ENDPOINT=
+UBI_ENABLED=true
+UBI_APPLICATION_NAME=opensearch-ubi

.gitattributes

@@ -33,6 +33,7 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+
 *.jpg filter=lfs diff=lfs merge=lfs -text
 *.jpeg filter=lfs diff=lfs merge=lfs -text
 *.png filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -14,4 +14,7 @@ app.zip
 app_old/
 images_retail/
 app_bkp/
-.streamlit/secrets.toml
+.streamlit/secrets.toml
+
+.env
+app_new/

app.py

@@ -7,7 +7,10 @@ import urllib.request
 import tarfile
 from yaml.loader import SafeLoader
 import warnings
-import utilities.invoke_models as invoke_models
+try:
+    import utilities.invoke_models as invoke_models
+except Exception:
+    invoke_models = None
 import threading
 warnings.filterwarnings("ignore", category=DeprecationWarning)
 
@@ -78,7 +81,8 @@ st.markdown("""
     padding: 24px 16px;
     width: 100%;
     margin: 0 auto;
-    height: 350px;
+    min-height: 280px;
+    height: auto;
     color: white;
     box-shadow: 0 0 10px rgba(0,0,0,0.3);
     transition: transform 0.3s ease-in-out, box-shadow 0.3s ease-in-out, border 0.3s ease-in-out;
@@ -90,6 +94,33 @@ st.markdown("""
     overflow: hidden;
 }
 
+@media (max-width: 768px) {
+    .card {
+        min-height: 200px;
+        padding: 16px 12px;
+    }
+    .card-header {
+        font-size: 60px !important;
+        height: 40px !important;
+        width: 40px !important;
+    }
+    .card-text {
+        font-size: 24px !important;
+    }
+    .card-description {
+        font-size: 13px !important;
+    }
+}
+
+@media (min-width: 769px) and (max-width: 1200px) {
+    .card {
+        min-height: 250px;
+    }
+    .card-text {
+        font-size: 28px !important;
+    }
+}
+
 .card:hover {
     transform: scale(1.06);
     transition: all 0.3s ease-in-out;
@@ -178,7 +209,10 @@ st.markdown("""
 # Header with logo and title
 col_logo, col_title = st.columns([32, 68])
 with col_logo:
-    st.image("/data/app_bkp/images/OS_AI_1_cropped.png", use_container_width=True)
+    try:
+        st.image("/data/app_bkp/images/OS_AI_1_cropped.png", use_container_width=True)
+    except Exception:
+        st.markdown("### 🔍 OpenSearch AI")
 
 spacer_col = st.columns(1)[0]
 with spacer_col:
@@ -186,7 +220,7 @@ with spacer_col:
 
 
 
-col1, col2, col3 = st.columns([1, 1, 1])
+col1, col2 = st.columns([1, 1])
 
 with col1:
     st.markdown("""
@@ -212,13 +246,23 @@ with col2:
             </div>
         </a>
     """, unsafe_allow_html=True)
-st.markdown("""
-    </div>
-""", unsafe_allow_html=True)
-
 
+col3, col4 = st.columns([1, 1])
 
 with col3:
+    st.markdown("""
+        <a href="/Search_Personalization" target="_self" style="text-decoration: none;">
+            <div class="card">
+                <div class="card-header">🪄</div>
+                <div class="card-text" style="font-size: 31px; color: #e46e08;">Personalization</div>
+                <div class="card-description" style="font-size: 16px; color: #ccc; margin-top: 6px;text-align: center;white-space: normal;">
+                    Explore how agentic memory drives real-time search personalization via UBI clickstream </div>
+            </div>
+        </a>
+    """, unsafe_allow_html=True)
+
+
+with col4:
     st.markdown("""
         <a href="/AI_Shopping_Assistant" target="_self" style="text-decoration: none;">
             <div class="card">
@@ -229,9 +273,6 @@ with col3:
             </div>
         </a>
     """, unsafe_allow_html=True)
-st.markdown("""
-    </div>
-""", unsafe_allow_html=True)
 
 st.markdown("""
     <style>

pages/Search_Personalization.py

@@ -0,0 +1,937 @@
+"""
+Search Personalization — OpenSearch UBI Boutique
+A personalized product search experience powered by OpenSearch agentic memory.
+
+This page demonstrates how user behavior (clickstream) data stored in OpenSearch
+can drive real-time search personalization through an agentic memory pipeline.
+"""
+import streamlit as st
+from typing import Dict, List, Any, Optional
+from search_personalization.data_loader import get_opensearch_client
+from search_personalization.image_utils import get_image_path, get_thumbnail_path, preload_image_cache
+from opensearchpy import OpenSearchException
+import os
+from dotenv import load_dotenv
+import uuid
+from datetime import datetime, timedelta
+
+# Import UBI tracking functions
+from search_personalization.ubi_tracker import (
+    track_query, track_event, is_ubi_enabled, validate_ubi_config
+)
+
+# Import agentic memory pipeline
+from search_personalization.agentic_memory.demo_runner import run_pipeline_for_persona
+
+# Import user preferences
+from search_personalization.user_preferences import get_user_preferences
+
+# Load environment variables
+load_dotenv()
+
+# UBI Configuration
+UBI_CONFIG = {
+    'events_endpoint': os.getenv('UBI_EVENTS_ENDPOINT'),
+    'queries_endpoint': os.getenv('UBI_QUERIES_ENDPOINT'),
+    'aws_region': os.getenv('AWS_REGION', 'us-east-1'),
+    'debug_mode': os.getenv('UBI_DEBUG_MODE', 'false').lower() == 'true',
+    'enabled': os.getenv('UBI_ENABLED', 'true').lower() == 'true',
+    'application_name': os.getenv('UBI_APPLICATION_NAME', 'opensearch-ubi')
+}
+
+# User Personas for clickstream tracking
+USER_PERSONAS = [
+    {
+        "user_id": "anonymous",
+        "name": "Anonymous User",
+        "age": 30,
+        "gender": "unspecified",
+        "location": "General",
+        "occupation": "General User",
+        "interests": ["General Shopping"],
+        "shopping_behavior": "Standard search without personalization",
+        "key": "anonymous",
+        "emoji": "👤"
+    },
+    {
+        "name": "Sarah",
+        "user_id": "user1",
+        "age": 32,
+        "gender": "female",
+        "location": "San Francisco, CA",
+        "occupation": "Marketing Manager",
+        "interests": ["Technology", "Fitness", "Travel"],
+        "shopping_behavior": "Focused on quality and trends.",
+        "key": "business_professional",
+        "emoji": "👩‍💼"
+    },
+    {
+        "name": "Alex",
+        "user_id": "user2",
+        "age": 21,
+        "gender": "male",
+        "location": "Austin, TX",
+        "occupation": "Computer Science Student",
+        "interests": ["Gaming", "Programming", "Music"],
+        "shopping_behavior": "Budget-conscious, looks for deals and discounts",
+        "key": "student",
+        "emoji": "🎓"
+    }
+]
+
+# Validate UBI configuration on startup
+_ubi_valid, _ubi_error = validate_ubi_config()
+if not _ubi_valid:
+    print(f"WARNING: {_ubi_error}")
+
+SESSION_DURATION = timedelta(minutes=1)
+
+# Base path for product images (relative to the search_personalization module)
+_MODULE_DIR = os.path.dirname(os.path.abspath(__file__))
+_PRODUCT_SAMPLES_DIR = os.path.join(
+    os.path.dirname(_MODULE_DIR), "search_personalization", "product_samples"
+)
+
+
+def _ensure_user_session(user_id: str) -> str:
+    """Ensure the given user has a valid (non-expired) session."""
+    now = datetime.utcnow()
+    sessions = st.session_state.get('sp_user_sessions', {})
+    user_session = sessions.get(user_id)
+
+    if user_session is None or (now - user_session['started_at']) >= SESSION_DURATION:
+        sessions[user_id] = {
+            'session_id': str(uuid.uuid4()),
+            'started_at': now,
+        }
+        st.session_state.sp_user_sessions = sessions
+
+    return sessions[user_id]['session_id']
+
+
+def _init_session_state() -> None:
+    """Initialize session state variables prefixed with sp_ to avoid collision."""
+    defaults = {
+        'sp_cart': [],
+        'sp_products': [],
+        'sp_filtered_products': [],
+        'sp_search_query': '',
+        'sp_query_id': 'browsing',
+        'sp_current_page': 1,
+        'sp_selected_product': None,
+        'sp_page_view': 'categories',
+        'sp_selected_category': None,
+        'sp_products_loaded': False,
+        'sp_load_error': None,
+        'sp_show_opensearch_query': False,
+        'sp_show_agent_reasoning': False,
+        'sp_last_opensearch_query': None,
+        'sp_selected_user_id': 'anonymous',
+        'sp_pending_query_tracking': False,
+        'sp_last_search_product_ids': [],
+        'sp_ubi_client_id': str(uuid.uuid4()),
+        'sp_user_sessions': {},
+        'sp_user_preferences': None,
+        'sp_preferences_applied': False,
+    }
+    for key, default in defaults.items():
+        if key not in st.session_state:
+            st.session_state[key] = default
+
+    # Ensure current user has an active session
+    _ensure_user_session(st.session_state.sp_selected_user_id)
+    st.session_state.sp_ubi_session_id = (
+        st.session_state.sp_user_sessions[st.session_state.sp_selected_user_id]['session_id']
+    )
+
+    # Load user preferences on first run
+    if (st.session_state.sp_user_preferences is None
+            and st.session_state.sp_selected_user_id != 'anonymous'):
+        _load_and_apply_preferences(st.session_state.sp_selected_user_id)
+
+
+def _load_and_apply_preferences(user_id: str) -> None:
+    """Load preferences from OpenSearch for the user."""
+    if user_id == 'anonymous':
+        st.session_state.sp_user_preferences = None
+        st.session_state.sp_preferences_applied = False
+        return
+    prefs = get_user_preferences(user_id)
+    st.session_state.sp_user_preferences = prefs
+    st.session_state.sp_preferences_applied = False
+
+
+def _load_products() -> None:
+    """Load all products from OpenSearch."""
+    if st.session_state.sp_products_loaded:
+        return
+    try:
+        client = get_opensearch_client()
+        response = client.search(
+            index='products',
+            body={
+                "query": {"match_all": {}},
+                "size": 10000,
+                "_source": {"excludes": ["image_binary", "product_description_vector"]}
+            }
+        )
+        products = [hit['_source'] for hit in response['hits']['hits']]
+        st.session_state.sp_products = products
+        st.session_state.sp_filtered_products = products
+        st.session_state.sp_products_loaded = True
+        st.session_state.sp_load_error = None
+        if products:
+            preload_image_cache(products)
+    except Exception as e:
+        st.session_state.sp_load_error = str(e)
+        st.session_state.sp_products_loaded = False
+        st.session_state.sp_products = []
+        st.session_state.sp_filtered_products = []
+
+
+def _get_product_by_id(product_id: str) -> Optional[Dict[str, Any]]:
+    """Get product by ID from session state."""
+    for product in st.session_state.sp_products:
+        if product.get('id') == product_id:
+            return product
+    return None
+
+
+def _add_to_cart(product_id: str) -> None:
+    """Add a product to the shopping cart and track with UBI."""
+    st.session_state.sp_cart.append(product_id)
+    if is_ubi_enabled():
+        try:
+            product = _get_product_by_id(product_id)
+            if product:
+                query_id = st.session_state.get('sp_query_id', 'browsing')
+                event_attributes = {
+                    'object': {
+                        'object_id': product_id,
+                        'object_id_field': 'id',
+                        'description': product.get('description', '')
+                    },
+                    'position': {'ordinal': 0},
+                    'product_name': product.get('name', ''),
+                    'product_price': product.get('price', 0),
+                    'product_category': product.get('category', '')
+                }
+                track_event(
+                    action_name='add_to_cart',
+                    query_id=query_id,
+                    object_id=product_id,
+                    event_attributes=event_attributes,
+                    message=f"Added {product.get('name', 'product')} to cart",
+                    message_type='CONVERSION',
+                    user_id=st.session_state.sp_selected_user_id
+                )
+        except Exception:
+            pass
+
+
+def _get_cart_items() -> List[Dict[str, Any]]:
+    """Get cart items with quantities."""
+    from collections import Counter
+    product_counts = Counter(st.session_state.sp_cart)
+    cart_items = []
+    for product_id, quantity in product_counts.items():
+        product = _get_product_by_id(product_id)
+        if product:
+            cart_items.append({
+                'product': product,
+                'quantity': quantity,
+                'product_id': product_id
+            })
+    return cart_items
+
+
+def _calculate_cart_total() -> float:
+    """Calculate the total price of all items in the cart."""
+    cart_items = _get_cart_items()
+    return sum(item['product'].get('price', 0) * item['quantity'] for item in cart_items)
+
+
+# --- Mapping from user_id to persona name for agentic memory ---
+_PERSONA_NAME_MAP = {"user1": "sarah", "user2": "alex"}
+
+
+def _search_products(search_query: str) -> tuple:
+    """Search products in OpenSearch using neural knn search."""
+    try:
+        client = get_opensearch_client()
+        if not search_query or search_query.strip() == '':
+            query_body = {"query": {"match_all": {}}, "size": 10}
+        else:
+            query_body = {
+                "query": {
+                    "neural": {
+                        "product_description_vector": {
+                            "query_text": search_query.strip(),
+                            "model_id": os.getenv('OPENSEARCH_MODEL_ID', ''),
+                            "k": 50
+                        }
+                    }
+                },
+                "size": 10
+            }
+        import json
+        st.session_state.sp_last_opensearch_query = json.dumps(query_body, indent=2)
+        response = client.search(index='products', body=query_body)
+        products = []
+        product_ids = []
+        for hit in response['hits']['hits']:
+            product = hit['_source']
+            products.append(product)
+            product_ids.append(product.get('id', hit.get('_id')))
+        total_count = response['hits']['total']['value']
+        return products, total_count, product_ids
+    except Exception as e:
+        st.error(f"Search error: {str(e)}")
+        return [], 0, []
+
+
+def _agentic_search(search_query: str, user_id: str) -> tuple:
+    """Run the agentic memory pipeline for personalized search."""
+    import json as _json
+
+    persona_name = _PERSONA_NAME_MAP[user_id]
+    session_id = st.session_state.get("sp_ubi_session_id", "web-session")
+
+    trace = run_pipeline_for_persona(
+        query=search_query,
+        persona_name=persona_name,
+        session_id=session_id,
+    )
+
+    # Extract explanation
+    _explanation_data = trace.get("agents", {}).get("explanation", {})
+    explanation = _explanation_data.get("output")
+    if explanation is None and "_future" in _explanation_data:
+        _future = _explanation_data["_future"]
+        try:
+            _future.result(timeout=3.0)
+            explanation = _future.result() if _future.done() else None
+        except Exception:
+            explanation = None
+
+    # Extract enriched query and inferred attributes
+    query_agent_output = trace["agents"].get("query", {}).get("output", "")
+    enriched_query_text = search_query
+    inferred_attrs = {}
+    try:
+        _qstart = query_agent_output.find("{")
+        _qend = query_agent_output.rfind("}") + 1
+        if _qstart >= 0 and _qend > _qstart:
+            _parsed_query = _json.loads(query_agent_output[_qstart:_qend])
+            enriched_query_text = _parsed_query.get("enriched_query", search_query)
+            inferred_attrs = _parsed_query.get("inferred_attributes", {})
+    except (_json.JSONDecodeError, TypeError):
+        pass
+
+    # Store the OpenSearch query for display
+    actual_os_query = trace.get("opensearch_query")
+    if actual_os_query:
+        st.session_state.sp_last_opensearch_query = _json.dumps(actual_os_query, indent=2)
+
+    # Parse ranked product IDs from ranking agent
+    ranking_output = trace["agents"].get("ranking", {}).get("output", "")
+    ranked_product_ids = []
+    personalization_summary = None
+    try:
+        start = ranking_output.find("{")
+        end = ranking_output.rfind("}") + 1
+        if start >= 0 and end > start:
+            parsed = _json.loads(ranking_output[start:end])
+            for item in parsed.get("results", []):
+                pid = item.get("product_id") or item.get("id")
+                if pid:
+                    ranked_product_ids.append(pid)
+            personalization_summary = parsed.get("personalization_summary")
+    except (_json.JSONDecodeError, TypeError):
+        pass
+
+    st.session_state._sp_personalization = personalization_summary
+
+    # Store reasoning trace
+    _reranked_count = len(ranked_product_ids)
+    st.session_state._sp_reasoning = {
+        "query_enrichment": {
+            "original_query": search_query,
+            "enriched_query": enriched_query_text,
+            "inferred_attributes": inferred_attrs,
+        },
+        "reranking": {
+            "model": "Cohere Rerank 3.5 (via Bedrock)",
+            "results_reranked": _reranked_count,
+        },
+        "aversion_filter": {
+            "aversions_from_profile": inferred_attrs.get("aversions", []),
+        },
+        "personalization_summary": personalization_summary,
+        "pipeline_duration_ms": trace.get("total_duration_ms"),
+    }
+
+    # Resolve product IDs to full product dicts
+    products = []
+    product_ids = []
+    for pid in ranked_product_ids:
+        product = _get_product_by_id(pid)
+        if product:
+            products.append(product)
+            product_ids.append(pid)
+
+    # Fallback to regular search if pipeline returned nothing
+    if not products:
+        products, _, product_ids = _search_products(enriched_query_text)
+        return products, len(products), product_ids, explanation
+
+    return products, len(products), product_ids, explanation
+
+
+# ─── UI COMPONENTS ───────────────────────────────────────────────────────────
+
+
+def _render_header() -> None:
+    """Render the header with title on top, then search bar + persona + cart below."""
+    # Home button
+    st.page_link("app.py", label=":orange[Home]", icon="🏠")
+    # Title on its own row
+    st.markdown(
+        '<h1 style="font-size:1.8rem;font-weight:bold;color:#ffffff;margin:0 0 0.5rem 0;">'
+        'Search personalization with agentic memory and UBI data</h1>',
+        unsafe_allow_html=True
+    )
+
+    # Search bar, persona selector, cart on second row
+    header_col2, header_col3, header_col4 = st.columns([4, 1.5, 0.5])
+
+    with header_col2:
+        search_query = st.text_input(
+            "Search", placeholder="What are you looking for?",
+            label_visibility="collapsed", key="sp_header_search"
+        )
+        if search_query != st.session_state.sp_search_query:
+            st.session_state.sp_search_query = search_query
+            if search_query and search_query.strip():
+                st.session_state.sp_page_view = 'grid'
+                st.session_state.sp_selected_category = None
+            else:
+                st.session_state.sp_page_view = 'categories'
+                st.session_state.sp_selected_category = None
+            new_query_id = str(uuid.uuid4())
+            st.session_state.sp_query_id = new_query_id
+            st.session_state.sp_pending_query_tracking = True
+            if is_ubi_enabled():
+                try:
+                    query_text = search_query.strip() if search_query and search_query.strip() else "match_all"
+                    track_event(
+                        action_name="search", query_id=new_query_id,
+                        event_attributes={"query_text": query_text},
+                        message=f"User searched for: {query_text}",
+                        user_id=st.session_state.sp_selected_user_id
+                    )
+                except Exception:
+                    pass
+
+    with header_col3:
+        persona_options = [f"{p['emoji']} {p['name']}" for p in USER_PERSONAS]
+        current_idx = next(
+            (i for i, p in enumerate(USER_PERSONAS)
+             if p['user_id'] == st.session_state.sp_selected_user_id), 0
+        )
+        selected_persona = st.selectbox(
+            "User", options=persona_options, index=current_idx,
+            label_visibility="collapsed", key="sp_persona_selector",
+            help="Select a user persona for clickstream tracking"
+        )
+        selected_index = persona_options.index(selected_persona)
+        new_user_id = USER_PERSONAS[selected_index]['user_id']
+        if new_user_id != st.session_state.sp_selected_user_id:
+            st.session_state.sp_selected_user_id = new_user_id
+            st.session_state.sp_ubi_session_id = _ensure_user_session(new_user_id)
+            st.session_state.sp_query_id = str(uuid.uuid4())
+            # Clear agentic pipeline cache
+            st.session_state.pop("_sp_cache_key", None)
+            st.session_state.pop("_sp_cache", None)
+            st.session_state.pop("_sp_reasoning", None)
+            st.session_state.sp_last_opensearch_query = None
+            _load_and_apply_preferences(new_user_id)
+
+    with header_col4:
+        cart_count = len(st.session_state.sp_cart)
+        if st.button(f"🛒 {cart_count}", key="sp_cart_button", help="View Cart"):
+            st.session_state.sp_page_view = 'cart'
+            st.rerun()
+
+
+def _render_product_card(product: Dict[str, Any], position: int = 0) -> None:
+    """Render a single product card with actions."""
+    product_id = product.get('id', '')
+    product_name = product.get('name', 'Unknown Product')
+    price = product.get('price', 0)
+    current_stock = product.get('current_stock', 0)
+    category = product.get('category', 'Unknown')
+    image_path = get_thumbnail_path(product)
+    query_id = st.session_state.get('sp_query_id', 'browsing')
+    user_id = st.session_state.get('sp_selected_user_id', 'anonymous')
+
+    col_image, col_content = st.columns([1, 4])
+
+    with col_image:
+        try:
+            st.image(image_path, use_container_width=True)
+        except Exception:
+            st.image(
+                os.path.join(_PRODUCT_SAMPLES_DIR, "product_image_coming_soon.png"),
+                use_container_width=True
+            )
+
+    with col_content:
+        st.markdown(f"**{product_name}**")
+        price_col, stock_col = st.columns([1, 2])
+        with price_col:
+            st.markdown(f"${price:,.2f}")
+        with stock_col:
+            if current_stock < 10:
+                st.markdown(f"⚠️ {current_stock} items left")
+            else:
+                st.caption(f"{current_stock} items left")
+
+        btn1, btn2, btn3, btn4 = st.columns([1, 1, 1, 2])
+        with btn1:
+            if st.button("🛒 Add", key=f"sp_cart_{product_id}", type="primary"):
+                _add_to_cart(product_id)
+                st.rerun()
+        with btn2:
+            if st.button("❤️", key=f"sp_like_{product_id}"):
+                if is_ubi_enabled():
+                    try:
+                        track_event(
+                            action_name='user_feedback', query_id=query_id,
+                            object_id=product_id,
+                            event_attributes={
+                                'object': {'object_id': product_id, 'object_id_field': 'id'},
+                                'position': {'ordinal': position},
+                                'feedback': 'thumbs_up'
+                            },
+                            message=f"Thumbs up: {product_name}",
+                            message_type='LIKE', user_id=user_id
+                        )
+                    except Exception:
+                        pass
+                st.toast(f"You liked {product_name}!")
+
+        with btn3:
+            if st.button("👎", key=f"sp_dislike_{product_id}"):
+                if is_ubi_enabled():
+                    try:
+                        track_event(
+                            action_name='user_feedback', query_id=query_id,
+                            object_id=product_id,
+                            event_attributes={
+                                'object': {'object_id': product_id, 'object_id_field': 'id'},
+                                'position': {'ordinal': position},
+                                'feedback': 'thumbs_down'
+                            },
+                            message=f"Thumbs down: {product_name}",
+                            message_type='REJECT', user_id=user_id
+                        )
+                    except Exception:
+                        pass
+                st.toast(f"You disliked {product_name}!")
+        with btn4:
+            if st.button("View Details", key=f"sp_details_{product_id}", type="secondary"):
+                if is_ubi_enabled():
+                    try:
+                        track_event(
+                            action_name='view_product_details', query_id=query_id,
+                            object_id=product_id,
+                            event_attributes={
+                                'object': {'object_id': product_id, 'object_id_field': 'id'},
+                                'position': {'ordinal': position},
+                            },
+                            message=f"Viewed: {product_name}",
+                            user_id=user_id
+                        )
+                    except Exception:
+                        pass
+                st.session_state.sp_selected_product = product_id
+                st.session_state.sp_page_view = 'details'
+                st.rerun()
+
+    st.divider()
+
+
+def _render_categories() -> None:
+    """Render the category landing page."""
+    categories = sorted(set(
+        p.get('category', '').lower()
+        for p in st.session_state.sp_products if p.get('category')
+    ))
+    if not categories:
+        st.info("No categories available.")
+        return
+
+    category_meta = {
+        'accessories': {'emoji': '👜', 'desc': 'Bags, watches, belts and more'},
+        'apparel': {'emoji': '👕', 'desc': 'Shirts, jackets, pants and dresses'},
+        'electronics': {'emoji': '💻', 'desc': 'Gadgets, devices and tech gear'},
+        'footwear': {'emoji': '👟', 'desc': 'Shoes, boots, sneakers and sandals'},
+        'jewelry': {'emoji': '💍', 'desc': 'Rings, necklaces, bracelets and earrings'},
+    }
+
+    st.markdown("### Shop by Category")
+    cols_per_row = 3
+    for row_start in range(0, len(categories), cols_per_row):
+        row_cats = categories[row_start:row_start + cols_per_row]
+        cols = st.columns(cols_per_row)
+        for idx, cat in enumerate(row_cats):
+            meta = category_meta.get(cat, {'emoji': '📦', 'desc': ''})
+            with cols[idx]:
+                # Show sample thumbnail
+                thumb_dir = os.path.join(_PRODUCT_SAMPLES_DIR, ".thumbnails", cat)
+                sample_image = None
+                if os.path.isdir(thumb_dir):
+                    images = [f for f in os.listdir(thumb_dir) if f.endswith('.jpg')]
+                    if images:
+                        sample_image = os.path.join(thumb_dir, images[0])
+                if sample_image:
+                    st.image(sample_image, use_container_width=True)
+                else:
+                    st.markdown(f"<div style='font-size:4rem;text-align:center;'>{meta['emoji']}</div>",
+                                unsafe_allow_html=True)
+                st.markdown(f"**{meta['emoji']} {cat.title()}**")
+                st.caption(meta['desc'])
+                if st.button(f"Browse {cat.title()}", key=f"sp_cat_{cat}", use_container_width=True):
+                    st.session_state.sp_selected_category = cat
+                    st.session_state.sp_page_view = 'grid'
+                    st.rerun()
+
+
+def _render_product_grid(products: List[Dict[str, Any]]) -> None:
+    """Render the product list."""
+    if not products:
+        st.markdown("### 🔍 No results found")
+        st.caption("Try adjusting your search query to see more products.")
+        return
+    st.markdown(f"**Showing {len(products)} products**")
+    for i, product in enumerate(products):
+        _render_product_card(product, position=i + 1)
+
+
+def _render_product_details(product_id: str) -> None:
+    """Render the product details page."""
+    product = _get_product_by_id(product_id)
+    if not product:
+        st.error("Product not found")
+        if st.button("← Back to Products", type="primary"):
+            st.session_state.sp_page_view = 'grid'
+            st.session_state.sp_selected_product = None
+            st.rerun()
+        return
+
+    if st.button("← Back to Products", key="sp_back_details", type="secondary"):
+        st.session_state.sp_page_view = 'grid'
+        st.session_state.sp_selected_product = None
+        st.rerun()
+
+    st.divider()
+    product_name = product.get('name', 'Unknown')
+    price = product.get('price', 0)
+    description = product.get('description', '')
+    category = product.get('category', '')
+    style = product.get('style', '')
+    current_stock = product.get('current_stock', 0)
+    image_path = get_thumbnail_path(product)
+
+    col_img, col_info = st.columns([1, 1], gap="large")
+    with col_img:
+        try:
+            st.image(image_path, use_container_width=True)
+        except Exception:
+            st.image(os.path.join(_PRODUCT_SAMPLES_DIR, "product_image_coming_soon.png"),
+                     use_container_width=True)
+    with col_info:
+        st.markdown(f"## {product_name}")
+        st.markdown(description)
+        st.divider()
+        st.markdown(f"### ${price:,.2f}")
+        if current_stock > 10:
+            st.success(f"✓ In Stock ({current_stock} available)")
+        elif current_stock > 0:
+            st.warning(f"⚠️ Low Stock ({current_stock} left)")
+        else:
+            st.error("✗ Out of Stock")
+        if current_stock > 0:
+            if st.button("🛒 Add to Cart", key="sp_detail_add", type="primary", use_container_width=True):
+                _add_to_cart(product_id)
+                st.success(f"✓ Added to cart!")
+                st.rerun()
+        st.divider()
+        c1, c2, c3 = st.columns(3)
+        c1.metric("Category", category.title())
+        c2.metric("Style", style.title())
+        gender = product.get('gender_affinity', '')
+        c3.metric("Gender", "Women" if gender == "F" else "Men" if gender == "M" else "Unisex")
+
+
+def _render_cart() -> None:
+    """Render the shopping cart view."""
+    if st.button("← Continue Shopping", key="sp_back_shop", type="secondary"):
+        st.session_state.sp_page_view = 'categories'
+        st.session_state.sp_selected_category = None
+        st.rerun()
+
+    st.divider()
+    cart_count = len(st.session_state.sp_cart)
+    st.markdown(f"### 🛒 Shopping Cart ({cart_count} items)")
+
+    if cart_count == 0:
+        st.info("Your cart is empty. Add some products to get started!")
+        if st.button("Browse Products", type="primary", use_container_width=True):
+            st.session_state.sp_page_view = 'categories'
+            st.rerun()
+        return
+
+    cart_items = _get_cart_items()
+    for idx, item in enumerate(cart_items):
+        product = item['product']
+        quantity = item['quantity']
+        product_id = item['product_id']
+        image_path = get_thumbnail_path(product)
+
+        col_img, col_det, col_act = st.columns([1, 3, 1])
+        with col_img:
+            try:
+                st.image(image_path, use_container_width=True)
+            except Exception:
+                st.image(os.path.join(_PRODUCT_SAMPLES_DIR, "product_image_coming_soon.png"),
+                         use_container_width=True)
+        with col_det:
+            st.markdown(f"**{product.get('name', 'Unknown')}**")
+            st.caption(f"Category: {product.get('category', '').title()}")
+            st.markdown(f"${product.get('price', 0):,.2f} × {quantity} = "
+                        f"**${product.get('price', 0) * quantity:,.2f}**")
+        with col_act:
+            if st.button("➖", key=f"sp_rm1_{idx}"):
+                if product_id in st.session_state.sp_cart:
+                    st.session_state.sp_cart.remove(product_id)
+                st.rerun()
+            if st.button("🗑️", key=f"sp_rmall_{idx}"):
+                st.session_state.sp_cart = [
+                    pid for pid in st.session_state.sp_cart if pid != product_id
+                ]
+                st.rerun()
+        st.divider()
+
+    # Cart summary
+    total = _calculate_cart_total()
+    st.markdown(f"### Total: ${total:,.2f}")
+
+    col_clear, col_checkout = st.columns(2)
+    with col_clear:
+        if st.button("🗑️ Clear Cart", key="sp_clear_cart"):
+            st.session_state.sp_cart = []
+            st.rerun()
+    with col_checkout:
+        if st.button("💳 Checkout", key="sp_checkout", type="primary"):
+            if is_ubi_enabled():
+                try:
+                    track_event(
+                        action_name='checkout',
+                        query_id=st.session_state.get('sp_query_id', 'browsing'),
+                        event_attributes={
+                            'cart_total': total,
+                            'cart_item_count': cart_count,
+                        },
+                        message=f"Checkout: {cart_count} items (${total:.2f})",
+                        message_type='CONVERSION',
+                        user_id=st.session_state.sp_selected_user_id
+                    )
+                except Exception:
+                    pass
+            st.session_state.sp_cart = []
+            st.success("✓ Thank you for your purchase!")
+            st.rerun()
+
+
+def _render_developer_sidebar() -> None:
+    """Render developer tools in the sidebar."""
+    with st.sidebar:
+        st.subheader("🔧 Developer Tools")
+        show_query = st.checkbox("Show OpenSearch Query", key="sp_show_query")
+        if show_query and st.session_state.sp_last_opensearch_query:
+            st.markdown("**Current Query:**")
+            st.code(st.session_state.sp_last_opensearch_query, language="json")
+        elif show_query:
+            st.info("No query executed yet.")
+
+        show_reasoning = st.checkbox("Show Agent Reasoning", key="sp_show_reasoning")
+        if show_reasoning:
+            reasoning = st.session_state.get("_sp_reasoning")
+            if reasoning:
+                st.markdown("---")
+                st.markdown("#### 🧠 Agent Reasoning")
+                enrichment = reasoning.get("query_enrichment", {})
+                with st.expander("1. Query Enrichment", expanded=True):
+                    st.markdown(f"**Original:** `{enrichment.get('original_query', '')}`")
+                    st.markdown(f"**Enriched:** `{enrichment.get('enriched_query', '')}`")
+                    attrs = enrichment.get("inferred_attributes", {})
+                    if attrs:
+                        for k, v in attrs.items():
+                            if v:
+                                st.markdown(f"- {k}: `{v}`")
+                reranking = reasoning.get("reranking", {})
+                with st.expander("2. Reranking", expanded=True):
+                    st.markdown(f"**Model:** {reranking.get('model', 'N/A')}")
+                    st.markdown(f"**Results reranked:** {reranking.get('results_reranked', 0)}")
+                aversion = reasoning.get("aversion_filter", {})
+                with st.expander("3. Aversions", expanded=True):
+                    aversions = aversion.get("aversions_from_profile", [])
+                    if aversions:
+                        st.markdown(f"**Profile aversions:** {', '.join(f'`{a}`' for a in aversions)}")
+                    else:
+                        st.markdown("No aversions detected.")
+                duration = reasoning.get("pipeline_duration_ms")
+                if duration:
+                    st.caption(f"⏱️ Pipeline: {duration}ms")
+            else:
+                st.info("Select Sarah or Alex and search to see reasoning.")
+
+
+# ─── MAIN PAGE ────────────────────────────────────────────────────────────────
+
+
+def main() -> None:
+    """Main entry point for the Search Personalization page."""
+    _init_session_state()
+
+    # Refresh the current user's session
+    st.session_state.sp_ubi_session_id = _ensure_user_session(
+        st.session_state.sp_selected_user_id
+    )
+
+    # Render header (search + persona + cart)
+    _render_header()
+    st.divider()
+
+    # Route to the correct view
+    if st.session_state.sp_page_view == 'details' and st.session_state.sp_selected_product:
+        _render_product_details(st.session_state.sp_selected_product)
+        _render_developer_sidebar()
+        return
+
+    if st.session_state.sp_page_view == 'cart':
+        _render_cart()
+        return
+
+    # Load products
+    if not st.session_state.sp_products_loaded and not st.session_state.sp_load_error:
+        with st.spinner("🔄 Loading products..."):
+            _load_products()
+    else:
+        _load_products()
+
+    # Categories landing
+    if st.session_state.sp_page_view == 'categories' and not st.session_state.sp_search_query:
+        if st.session_state.sp_products_loaded and st.session_state.sp_products:
+            _render_categories()
+            _render_developer_sidebar()
+            return
+
+    # Error states
+    if st.session_state.sp_load_error:
+        st.error("❌ Unable to connect to the product catalog.")
+        st.info("Check your AWS credentials and OpenSearch endpoint.")
+        if st.button("🔄 Retry", key="sp_retry", type="primary"):
+            st.session_state.sp_products_loaded = False
+            st.session_state.sp_load_error = None
+            st.rerun()
+        return
+
+    if not st.session_state.sp_products_loaded:
+        st.info("⏳ Loading products...")
+        return
+
+    if not st.session_state.sp_products:
+        st.warning("No products available in the catalog.")
+        return
+
+    # ─── Search + product grid ───
+    search_query = st.session_state.sp_search_query
+    page_size = 10
+
+    # Back to categories button when browsing a category
+    if st.session_state.sp_selected_category and not (search_query and search_query.strip()):
+        if st.button("← Back to Categories", key="sp_back_cats", type="secondary"):
+            st.session_state.sp_selected_category = None
+            st.session_state.sp_page_view = 'categories'
+            st.rerun()
+        st.markdown(f"## {st.session_state.sp_selected_category.title()}")
+
+    # Perform search
+    if search_query and search_query.strip():
+        persona_user_id = st.session_state.sp_selected_user_id
+        if persona_user_id in _PERSONA_NAME_MAP:
+            cache_key = f"{persona_user_id}:{search_query}"
+            if st.session_state.get("_sp_cache_key") != cache_key:
+                products_to_display, total_count, product_ids, explanation = (
+                    _agentic_search(search_query, persona_user_id)
+                )
+                st.session_state._sp_cache_key = cache_key
+                st.session_state._sp_cache = (products_to_display, total_count, product_ids, explanation)
+            else:
+                products_to_display, total_count, product_ids, explanation = st.session_state._sp_cache
+        else:
+            products_to_display, total_count, product_ids = _search_products(search_query)
+            explanation = None
+    else:
+        all_products = st.session_state.sp_products
+        if st.session_state.sp_selected_category:
+            all_products = [
+                p for p in all_products
+                if p.get('category', '').lower() == st.session_state.sp_selected_category
+            ]
+        total_count = len(all_products)
+        start = (st.session_state.sp_current_page - 1) * page_size
+        products_to_display = all_products[start:start + page_size]
+        product_ids = [p.get('id', '') for p in products_to_display]
+        explanation = None
+
+    # UBI query tracking (non-blocking)
+    if is_ubi_enabled() and st.session_state.get('sp_pending_query_tracking', False):
+        st.session_state.sp_pending_query_tracking = False
+        import threading
+        _q_text = search_query.strip() if search_query and search_query.strip() else "match_all"
+        _q_id = st.session_state.get('sp_query_id', 'browsing')
+        _q_uid = st.session_state.sp_selected_user_id
+        _q_hits = list(product_ids)
+        def _track():
+            try:
+                track_query(_q_text, _q_id, user_id=_q_uid, query_response_hit_ids=_q_hits)
+            except Exception:
+                pass
+        threading.Thread(target=_track, daemon=True).start()
+
+    # Show search info
+    if search_query and search_query.strip():
+        if not products_to_display:
+            st.info(f"🔍 No results found for '{search_query}'")
+        else:
+            if explanation:
+                st.info(f"💡 {explanation}")
+            personalization = st.session_state.get("_sp_personalization")
+            persona_user_id = st.session_state.sp_selected_user_id
+            if persona_user_id in _PERSONA_NAME_MAP and personalization and "none" not in personalization.lower():
+                persona_name = _PERSONA_NAME_MAP.get(persona_user_id, "").title()
+                st.caption(f"📊 Personalized results for {persona_name}")
+
+    # Render product grid
+    _render_product_grid(products_to_display)
+
+    # Developer tools sidebar
+    _render_developer_sidebar()
+
+
+# Run the page
+main()

requirements.txt

@@ -19,3 +19,8 @@ einops
 seaborn
 opentelemetry-sdk
 opentelemetry-exporter-otlp
+
+python-dotenv
+strands-agents
+strands-agents-tools
+PyYAML

search_personalization/IAM_SETUP.md

@@ -0,0 +1,170 @@
+# IAM Setup for Search Personalization Module
+
+This document describes the least-privilege IAM configuration required to run the
+Search Personalization module on Hugging Face Spaces (or any environment without
+`~/.aws/config` profiles).
+
+## AWS Account
+
+- **Account ID**: `099713751195`
+- **Region**: `us-east-1` (primary), `us-west-2` (Cohere Rerank only)
+
+> **Important**: The domain name in the IAM policy ARNs is `mydomain`. Verify your
+> actual OpenSearch domain name matches. You can find it in the AWS Console under
+> OpenSearch Service → Domains, or by inspecting the `OPENSEARCH_ENDPOINT` URL
+> (the segment between `search-` and the hash is the domain name).
+
+## Resources Accessed
+
+| AWS Service | Resource | Operations |
+|-------------|----------|------------|
+| **OpenSearch** (`mydomain`) | `products` index | Neural kNN search, read mapping + aggregations |
+| **OpenSearch** (`mydomain`) | `user_preferences` index | Read/write preferences (upsert by user_id), create index |
+| **OpenSearch** (`mydomain`) | `personas` index | Search |
+| **OpenSearch** (`mydomain`) | `_plugins/_ml/memory_containers/*` | Read long-term/session/working/history memories, write memories |
+| **Bedrock** (us-east-1) | `us.amazon.nova-micro-v1:0` | Agent reasoning (Query, Ranking, Explanation agents via Strands — uses streaming) |
+| **Bedrock** (us-west-2) | `cohere.rerank-v3-5:0` | Semantic reranking of search results (non-streaming) |
+| **OSIS** (us-east-1) | `ubi-events-pipeline-*` | POST UBI clickstream events (`/ubi/events`) |
+| **OSIS** (us-east-1) | `ubi-queries-pipeline-*` | POST UBI query tracking data (`/ubi/queries`) |
+
+## Step 1: Create IAM User
+
+```bash
+aws iam create-user --user-name hf-opensearch-personalization
+```
+
+## Step 2: Attach the IAM Policy
+
+The policy file is at `search_personalization/iam-policy.json`.
+
+```bash
+aws iam put-user-policy \
+  --user-name hf-opensearch-personalization \
+  --policy-name OpenSearchPersonalizationAccess \
+  --policy-document file://search_personalization/iam-policy.json
+```
+
+## Step 3: Create Access Keys
+
+```bash
+aws iam create-access-key --user-name hf-opensearch-personalization
+```
+
+Save the `AccessKeyId` and `SecretAccessKey` — you'll need them for HF Spaces.
+
+## Step 4: Configure OpenSearch Fine-Grained Access Control (FGAC)
+
+Your domain uses fine-grained access control. You need to map the IAM user to a
+backend role in OpenSearch Dashboards.
+
+### 4a. Create an OpenSearch Security Role
+
+In OpenSearch Dashboards → Security → Roles → Create role:
+
+**Role name**: `hf_demo_role`
+
+**Cluster permissions**:
+```
+cluster:admin/opensearch/ml/*
+```
+(Required for the `/_plugins/_ml/memory_containers` API)
+
+**Index permissions**:
+
+| Index pattern | Allowed actions |
+|---------------|----------------|
+| `products` | `indices:data/read/*` |
+| `personas` | `indices:data/read/*` |
+| `user_preferences` | `indices:data/read/*`, `indices:data/write/*`, `indices:admin/create` |
+| `.plugins-ml-memory*` | `indices:data/read/*`, `indices:data/write/*` |
+
+> Note: Memory containers store data in hidden indices prefixed with
+> `.plugins-ml-memory`. The exact pattern depends on your OpenSearch version.
+
+### 4b. Map the IAM User to the Role
+
+In OpenSearch Dashboards → Security → Roles → `hf_demo_role` → Mapped users → Map user:
+
+Add as **User** type (not Backend role):
+```
+arn:aws:iam::099713751195:user/hf-opensearch-personalization
+```
+
+> Note: IAM users authenticated via SigV4 are mapped as "User" type in FGAC.
+> "Backend role" type is for IAM *roles* (assumed via STS). This is a common
+> source of confusion in OpenSearch FGAC.
+
+## Step 5: Configure OSIS Pipeline Access
+
+Your OSIS pipelines must also allow this IAM user to ingest data. The pipeline's
+IAM role needs to trust this user, OR the pipeline access policy must include:
+
+```json
+{
+  "Effect": "Allow",
+  "Principal": {
+    "AWS": "arn:aws:iam::099713751195:user/hf-opensearch-personalization"
+  },
+  "Action": "osis:Ingest",
+  "Resource": [
+    "arn:aws:osis:us-east-1:099713751195:pipeline/ubi-events-pipeline-*",
+    "arn:aws:osis:us-east-1:099713751195:pipeline/ubi-queries-pipeline-*"
+  ]
+}
+```
+
+## Step 6: Set HF Space Secrets
+
+In your Hugging Face Space settings (Settings → Repository Secrets), add:
+
+| Secret Name | Value |
+|-------------|-------|
+| `AWS_ACCESS_KEY_ID` | The access key from Step 3 |
+| `AWS_SECRET_ACCESS_KEY` | The secret key from Step 3 |
+| `AWS_REGION` | `us-east-1` |
+| `OPENSEARCH_ENDPOINT` | `https://search-mydomain-lsbvasigo2g6v5kmbtlov4573i.aos.us-east-1.on.aws` |
+| `OPENSEARCH_MODEL_ID` | `PSZ-g54B5Zk16lKMlvWx` |
+| `OPENSEARCH_INGEST_PIPELINE` | `product-multimodal-pipeline` |
+| `EMBEDDING_MODEL_ID` | `PSZ-g54B5Zk16lKMlvWx` |
+| `EMBEDDING_DIMENSION` | `1024` |
+| `MEMORY_CONTAINER_ID` | `HTcGj54B5Zk16lKMsL49` |
+| `BEDROCK_MODEL_ID_FAST` | `us.amazon.nova-micro-v1:0` |
+| `BEDROCK_MODEL_ID` | `us.anthropic.claude-sonnet-4-6` |
+| `UBI_EVENTS_ENDPOINT` | `https://ubi-events-pipeline-6qdwk2cov2ab44jrvriugly2au.us-east-1.osis.amazonaws.com` |
+| `UBI_QUERIES_ENDPOINT` | `https://ubi-queries-pipeline-kxb6c3jrjlfbgcsq6tttgpeuum.us-east-1.osis.amazonaws.com` |
+| `UBI_ENABLED` | `true` |
+| `UBI_APPLICATION_NAME` | `os-product-catalog` |
+
+**Do NOT set** `AWS_PROFILE` in HF Spaces — it's only for local development.
+
+## Step 7: Verify (Optional)
+
+After setting up, test connectivity from the Space logs:
+
+```python
+import boto3
+session = boto3.Session()
+sts = session.client('sts')
+print(sts.get_caller_identity())
+# Should show: arn:aws:iam::099713751195:user/hf-opensearch-personalization
+```
+
+## Security Notes
+
+- The IAM user has NO admin access — it can only search/index specific indices and invoke specific models
+- It cannot create/delete the OpenSearch domain itself
+- It cannot access other AWS services (EC2, S3, Lambda, etc.)
+- It cannot modify IAM policies or create new users
+- Rotate keys periodically via `aws iam create-access-key` + `aws iam delete-access-key`
+- Monitor usage via CloudTrail filtered to this IAM user
+
+## Important: What This Policy Does NOT Allow
+
+- ❌ Creating or deleting OpenSearch domains
+- ❌ Modifying domain configuration or access policies
+- ❌ Accessing any index other than `products`, `personas`, `user_preferences`
+- ❌ Invoking any Bedrock model other than Nova Micro and Cohere Rerank
+- ❌ Any S3, EC2, Lambda, IAM, or other service access
+- ❌ Deleting or recreating the `products` or `personas` indices (read-only)
+- ❌ Deleting the `user_preferences` index (only write/read documents)
+- ❌ Creating or deleting OSIS pipelines (only ingest data)

search_personalization/__init__.py

@@ -0,0 +1 @@
+"""Product Catalog Service - RESTful API for product management."""

search_personalization/agentic_memory/__init__.py

@@ -0,0 +1,4 @@
+"""
+Agentic Memory Demo — Multi-agent e-commerce query understanding
+with OpenSearch agentic memory for personalized search.
+"""

search_personalization/agentic_memory/agents/query_agent.py

@@ -0,0 +1,96 @@
+"""
+Query Understanding Agent — Enriches raw search queries using user profile context.
+Returns only the enriched query attributes; the orchestrator handles the actual search.
+"""
+
+from typing import Optional
+
+from strands import Agent
+from strands.models import BedrockModel
+
+from search_personalization.agentic_memory.config import BEDROCK_MODEL_ID_FAST, AWS_REGION
+from search_personalization.agentic_memory.schema_cache import get_schema_prompt_block
+
+_DEFAULT_SCHEMA_BLOCK = """PRODUCT INDEX SCHEMA:
+- category (keyword, 5 values): apparel, accessories, footwear, electronics, jewelry
+- style (keyword): jacket, formal, shirt, glasses, scarf, backpack, sneaker, belt, handbag, bracelet, camera, sandals, boot, necklace, socks, watch, earrings, bag, speaker, headphones, cable, television, computer, keyboard
+- gender_affinity (keyword): F, M
+- price (numeric)
+- name (text) — product name
+- description (text) — product description"""
+
+_SYSTEM_PROMPT_TEMPLATE = """You are the Query Understanding Agent. Enrich a search query using the user's behavioral profile.
+
+{schema_block}
+
+FIELD USAGE:
+- category → hard filter (exact match, 5 values)
+- gender_affinity → hard filter ("F" or "M")
+- max_price → hard filter (numeric ceiling) — ONLY when user explicitly states price in query
+- enriched_query → drives vector/semantic search (no filter values here)
+
+ENRICHED QUERY RULES:
+1. Format: space-separated keywords/phrases, NOT a sentence. Example: "shoes sneakers athletic comfortable black gray"
+2. Start with the user's original query words (strip any price mentions like "under $50").
+3. Append descriptors reflecting the user's DOMINANT style pattern (majority of behavior, not outliers). Example: 3 sneaker purchases + 1 dress shirt = sneaker person, NOT formal person.
+4. If the profile shows strong preference for a product sub-type (e.g., bought 3 sneakers), include that sub-type. This is the most important signal.
+5. Express aversions as positive opposites (e.g., aversion "flashy" → add "understated"; "loud colors" → "muted tones"). Do NOT use negation words.
+6. NEVER include in enriched_query: price values ("under $50", "$50", "budget"), category names, gender letters, or any term related to the aversions (e.g., if aversions include "formal"/"fancy", never add "formal", "dress", "elegant", or similar).
+7. Use conversation history to resolve references (e.g., "in blue" after "shoes" → "shoes blue").
+
+INFERRED ATTRIBUTES:
+- category: REQUIRED. Map query to: apparel, accessories, footwear, electronics, jewelry.
+- gender_affinity: "F" or "M" if any gender signal exists in profile. Null otherwise.
+- max_price: numeric ONLY if the user's raw query text explicitly mentions a price constraint (e.g., "under $50", "below $100"). If the query does NOT contain a price mention, max_price MUST be null even if the profile suggests a budget ceiling.
+- style: closest schema value or null.
+- colors: recurring color preferences from profile.
+- preferred_materials: recurring materials from profile.
+- use_context: primary shopping context (e.g., "campus daily wear", "business-casual work").
+- aversions: raw aesthetic dislikes from profile (e.g., ["flashy", "formal"]). Never include the searched product type.
+
+OUTPUT (strict JSON, nothing else):
+{{"enriched_query": "keywords only, e.g.: shoes sneakers athletic comfortable muted", "inferred_attributes": {{"colors": [], "style": null, "category": null, "gender_affinity": null, "max_price": null, "aversions": [], "preferred_materials": [], "use_context": null}}}}
+"""
+
+
+def create_query_agent() -> Agent:
+    """Create the Query Understanding Agent (no tools — pure reasoning)."""
+    schema_block = get_schema_prompt_block() or _DEFAULT_SCHEMA_BLOCK
+    system_prompt = _SYSTEM_PROMPT_TEMPLATE.format(schema_block=schema_block)
+
+    model = BedrockModel(
+        model_id=BEDROCK_MODEL_ID_FAST,
+        region_name=AWS_REGION,
+    )
+    return Agent(
+        model=model,
+        tools=[],
+        system_prompt=system_prompt,
+    )
+
+
+def invoke_query_agent(
+    query: str,
+    persona_id: str,
+    session_id: str,
+    profile: str = "",
+    session_context: str = "",
+) -> str:
+    """
+    Invoke the Query Understanding Agent.
+
+    Returns:
+        JSON string with enriched_query and inferred_attributes only.
+    """
+    agent = create_query_agent()
+
+    profile_section = f"\n\nUser Profile:\n{profile}" if profile else "\n\nNo profile available — use query as-is."
+    history_section = f"\n\nConversation History:\n{session_context}" if session_context else ""
+
+    prompt = (
+        f"Process this search:\n\n"
+        f"Raw query: \"{query}\"{profile_section}{history_section}\n\n"
+        f"Enrich the query and output the JSON."
+    )
+    result = agent(prompt)
+    return str(result)

search_personalization/agentic_memory/agents/ranking_agent.py

@@ -0,0 +1,70 @@
+"""
+Ranking Agent — Accepts enriched query + search results directly from the orchestrator
+and produces personalized, reranked product results.
+"""
+
+from strands import Agent
+from strands.models import BedrockModel
+
+from search_personalization.agentic_memory.config import BEDROCK_MODEL_ID_FAST, AWS_REGION
+from search_personalization.agentic_memory.memory_tools import rerank_results
+
+
+SYSTEM_PROMPT = """You are the Ranking Agent. Rerank search results and produce personalized ranked results.
+
+WORKFLOW:
+1. You receive the enriched query and search results in the prompt.
+2. Call rerank_results with the enriched query string and the search results for semantic reranking.
+3. Filter out any products matching known aversions from the inferred_attributes.
+4. Output the final ranked JSON.
+
+RULES:
+- Top 7 results max.
+- Use the enriched_query as the rerank query (it has user preferences baked in).
+- Exclude products that match aversions listed in inferred_attributes.
+- Output ONLY the JSON object. No text before or after.
+
+OUTPUT FORMAT (strict JSON, nothing else):
+{"results": [{"rank": 1, "product_id": "", "name": "", "price": 0, "score": 0}], "enriched_query": "the enriched query used", "personalization_summary": "one sentence"}
+"""
+
+
+def create_ranking_agent() -> Agent:
+    """Create the Ranking Agent."""
+    model = BedrockModel(
+        model_id=BEDROCK_MODEL_ID_FAST,
+        region_name=AWS_REGION,
+    )
+    return Agent(
+        model=model,
+        tools=[rerank_results],
+        system_prompt=SYSTEM_PROMPT,
+    )
+
+
+def invoke_ranking_agent(
+    persona_id: str,
+    session_id: str,
+    search_results: str,
+) -> str:
+    """
+    Invoke the Ranking Agent.
+
+    Args:
+        persona_id: The user ID.
+        session_id: The current session ID.
+        search_results: JSON string from the Query Agent containing enriched_query,
+                        inferred_attributes, and search_results.
+
+    Returns:
+        The agent's response with ranked, personalized results.
+    """
+    agent = create_ranking_agent()
+    prompt = (
+        f"Produce personalized ranked results for persona_id '{persona_id}' "
+        f"in session '{session_id}'.\n\n"
+        f"Here are the enriched query and search results:\n{search_results}\n\n"
+        f"Rerank them, filter aversions, and return the final ranked results."
+    )
+    result = agent(prompt)
+    return str(result)

search_personalization/agentic_memory/agents/results_explanation_agent.py

@@ -0,0 +1,120 @@
+"""
+Results Explanation Agent — Explains search result gaps to users.
+
+Runs ASYNCHRONOUSLY after ranking to avoid adding latency to the main pipeline.
+Only generates an explanation when there's a clear mismatch between
+user intent and returned results.
+"""
+
+import json
+from typing import Optional
+
+from strands import Agent
+from strands.models import BedrockModel
+
+from search_personalization.agentic_memory.config import BEDROCK_MODEL_ID_FAST, AWS_REGION
+
+_SYSTEM_PROMPT = """You detect mismatches between search results and the user's intent (query + profile preferences). Output JSON only.
+
+RULES:
+1. Output {"explanation": null} if results align well with BOTH the query AND the user's profile preferences.
+2. A mismatch occurs when:
+   - Price: user asked "under $50" but ALL or MOST results exceed that price.
+   - Style/sub-type: user prefers sneakers (from profile) but results are mostly boots/dress shoes due to price or availability constraints.
+   - Color: user asked for "red shoes" but no red products appear.
+   - Product type: user asked for "dress" but catalog has no dresses.
+3. Shoes includes ALL footwear: sneakers, boots, lace-ups, sandals, heels, loafers, dress shoes. Results being a different footwear sub-type than the user's preference IS a mismatch worth explaining.
+4. When there's a style mismatch due to constraints (e.g., not enough sneakers under $50), explain WHY different styles appear — reference the constraint that caused it.
+5. If only 1-2 results don't match preferences but the majority do, output null.
+6. Do NOT explain when results match both query and preferences well.
+7. One sentence max. Be helpful, not apologetic.
+
+EXAMPLES:
+- Query "shoes under $50", profile prefers sneakers, results are boots/oxfords → {"explanation": "Limited sneaker options under $50 — showing other footwear styles within your budget."}
+- Query "shoes under $50", profile prefers sneakers, results ARE sneakers → {"explanation": null}
+- Query "backpack", results are backpacks → {"explanation": null}
+- Query "red jacket", results have no red items → {"explanation": "No red jackets available — showing closest color matches."}
+
+OUTPUT: {"explanation": "one sentence" or null}
+"""
+
+
+def should_explain(query: str, results: list, price_filter_relaxed: bool = False) -> bool:
+    """Determine whether the explanation agent should run."""
+    if price_filter_relaxed:
+        return True
+    if not results:
+        return True
+    return True
+
+
+def invoke_results_explanation_agent(
+    query: str,
+    results: list,
+    price_filter_relaxed: bool = False,
+    original_max_price: float = None,
+    inferred_attributes: dict = None,
+    **kwargs,
+) -> Optional[str]:
+    """
+    Generate a brief explanation when results don't match user intent or profile preferences.
+
+    Returns:
+        A short explanation string, or None if results are fine.
+    """
+    if not should_explain(query, results, price_filter_relaxed):
+        return None
+
+    # Price filter relaxation — deterministic message, no LLM needed
+    if price_filter_relaxed and original_max_price:
+        return f"Not enough options under ${original_max_price:.0f} — showing best matches across all price ranges."
+
+    model = BedrockModel(
+        model_id=BEDROCK_MODEL_ID_FAST,
+        region_name=AWS_REGION,
+    )
+    agent = Agent(
+        model=model,
+        tools=[],
+        system_prompt=_SYSTEM_PROMPT,
+    )
+
+    # Compact prompt — query, profile preferences, and result summaries
+    result_summaries = ", ".join(
+        f"{r.get('name', '?')} (${r.get('price', '?')}, style: {r.get('style', '?')})"
+        for r in results[:5]
+    )
+
+    # Build profile preference context
+    profile_context = ""
+    if inferred_attributes:
+        pref_parts = []
+        if inferred_attributes.get("style"):
+            pref_parts.append(f"preferred style: {inferred_attributes['style']}")
+        if inferred_attributes.get("colors"):
+            pref_parts.append(f"preferred colors: {', '.join(inferred_attributes['colors'][:3])}")
+        if inferred_attributes.get("use_context"):
+            pref_parts.append(f"use context: {inferred_attributes['use_context']}")
+        if inferred_attributes.get("max_price"):
+            pref_parts.append(f"budget: ${inferred_attributes['max_price']}")
+        if pref_parts:
+            profile_context = f"\nUser profile preferences: {'; '.join(pref_parts)}"
+
+    prompt = (
+        f"Query: \"{query}\"{profile_context}\n"
+        f"Results: {result_summaries}\n"
+        f"JSON:"
+    )
+
+    try:
+        result = agent(prompt)
+        raw = str(result)
+        start = raw.find("{")
+        end = raw.rfind("}") + 1
+        if start >= 0 and end > start:
+            parsed = json.loads(raw[start:end])
+            return parsed.get("explanation")
+    except Exception:
+        pass
+
+    return None

search_personalization/agentic_memory/config.py

@@ -0,0 +1,46 @@
+"""
+Configuration for the Agentic Memory demo.
+Manages OpenSearch connection, memory container IDs, and model settings.
+"""
+
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# OpenSearch cluster
+OPENSEARCH_ENDPOINT = os.getenv("OPENSEARCH_ENDPOINT", "")
+AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
+
+# Bedrock model for agents
+BEDROCK_MODEL_ID_FAST = os.getenv(
+    "BEDROCK_MODEL_ID_FAST",
+    "us.amazon.nova-micro-v1:0"
+)
+BEDROCK_MODEL_ID = os.getenv(
+    "BEDROCK_MODEL_ID",
+    "us.anthropic.claude-sonnet-4-6"
+)
+
+# Embedding model for memory containers
+EMBEDDING_MODEL_ID = os.getenv("EMBEDDING_MODEL_ID", "")
+EMBEDDING_DIMENSION = int(os.getenv("EMBEDDING_DIMENSION", "1024"))
+
+# LLM model for memory processing (extraction/summarization)
+MEMORY_LLM_MODEL_ID = os.getenv("MEMORY_LLM_MODEL_ID", "")
+
+# Single memory container (holds all memory types: long-term, sessions, working, history)
+CONTAINER_NAME = "product-search-agent"
+CONTAINER_ID = os.getenv("MEMORY_CONTAINER_ID")
+
+# Persona IDs
+PERSONAS = {
+    "sarah": "user1",
+    "alex": "user2",
+}
+
+# Persona genders (used for gender_affinity matching in product personalization)
+PERSONA_GENDERS = {
+    "user1": "female",
+    "user2": "male",
+}

search_personalization/agentic_memory/demo_runner.py

@@ -0,0 +1,424 @@
+"""
+Demo Runner — Orchestrates the 3-agent pipeline for the agentic memory demo.
+
+Architecture:
+  - Long-term memory: user preferences fetched from OpenSearch agentic memory
+  - Query Understanding Agent: enriches query, searches catalog, returns results directly
+  - Ranking Agent: receives results directly, reranks with personalization
+  - Working memory write: async fire-and-forget for observability/tracing
+
+Usage:
+    python -m src.agentic_memory.demo_runner
+    python -m src.agentic_memory.demo_runner --query "black boots for a concert"
+    python -m src.agentic_memory.demo_runner --interactive
+"""
+
+import json
+import time
+import uuid
+import threading
+from datetime import datetime
+from typing import Optional
+
+from search_personalization.agentic_memory.config import PERSONAS
+from search_personalization.agentic_memory.memory_tools import get_user_profile
+from search_personalization.agentic_memory.agents.query_agent import invoke_query_agent
+from search_personalization.agentic_memory.memory_tools import write_session_memory, write_history, read_session_memory
+
+
+DEFAULT_QUERY = "backpack"
+
+
+def run_pipeline_for_persona(
+    query: str,
+    persona_name: str,
+    session_id: Optional[str] = None,
+    conversation_history: Optional[str] = None,
+) -> dict:
+    """
+    Run the 3-step pipeline for a single persona:
+      1. Profile Cache (instant)
+      2. Query Understanding Agent (enrich + search)
+      3. Ranking Agent (rerank with direct results)
+    """
+    persona_id = PERSONAS[persona_name]
+    if not session_id:
+        session_id = f"sess-{uuid.uuid4().hex[:8]}"
+
+    print(f"\n{'='*60}")
+    print(f"  PERSONA: {persona_name.upper()} ({persona_id})")
+    print(f"  Query: \"{query}\"")
+    print(f"  Session: {session_id}")
+    print(f"{'='*60}")
+
+    pipeline_start = time.time()
+    pipeline_trace = {
+        "persona_id": persona_id,
+        "persona_name": persona_name,
+        "query": query,
+        "session_id": session_id,
+        "timestamp": datetime.utcnow().isoformat(),
+        "agents": {},
+    }
+
+    # --- Step 1: Profile + session in parallel ---
+    from concurrent.futures import ThreadPoolExecutor
+
+    start = time.time()
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        profile_future = pool.submit(get_user_profile, persona_id)
+        if conversation_history:
+            session_context = conversation_history
+            profile_raw = profile_future.result()
+        else:
+            session_future = pool.submit(read_session_memory, persona_id, session_id)
+            profile_raw = profile_future.result()
+            session_context = session_future.result()
+
+    # Format USER_PREFERENCE memories into a profile string for the agent
+    profile_data = json.loads(profile_raw)
+    memories = profile_data.get("memories", [])
+    preference_statements = [
+        m.get("_source", {}).get("memory", "")
+        for m in memories
+        if m.get("_source", {}).get("strategy_type") == "USER_PREFERENCE"
+    ]
+    profile_lines = [f"- {s}" for s in preference_statements if s]
+    profile = (
+        f"Profile based on {len(preference_statements)} behavioral observations:\n"
+        + "\n".join(profile_lines)
+    ) if profile_lines else ""
+
+    memory_duration = time.time() - start
+    print(f"\n  [memory] Profile + session loaded ({memory_duration*1000:.1f}ms)")
+    pipeline_trace["agents"]["memory"] = {
+        "duration_ms": round(memory_duration * 1000, 1),
+        "source": "opensearch-long-term-memory",
+        "preference_records": len(preference_statements),
+    }
+
+    # --- Step 2: Query Understanding Agent (enrich only) ---
+    print(f"  [agent] Query Understanding Agent — enrich...")
+    start = time.time()
+    query_result = invoke_query_agent(query, persona_id, session_id, profile=profile or "", session_context=session_context)
+    query_duration = time.time() - start
+    pipeline_trace["agents"]["query"] = {
+        "duration_ms": int(query_duration * 1000),
+        "output": query_result,
+    }
+    print(f"          Done ({query_duration:.1f}s)")
+
+    # --- Step 2b: Embed enriched query + knn search ---
+    print(f"  [search] Embed + catalog search...")
+    start = time.time()
+    try:
+        enrichment = json.loads(query_result)
+    except (json.JSONDecodeError, TypeError):
+        # Agent output may contain surrounding text — extract JSON object
+        enrichment = None
+        try:
+            _start = query_result.find("{")
+            _end = query_result.rfind("}") + 1
+            if _start >= 0 and _end > _start:
+                enrichment = json.loads(query_result[_start:_end])
+        except (json.JSONDecodeError, TypeError):
+            pass
+        if not enrichment:
+            enrichment = {"enriched_query": query, "inferred_attributes": {}}
+
+    attrs = enrichment.get("inferred_attributes", {})
+    enriched_query = enrichment.get("enriched_query", query)
+
+    # Let the agent handle all semantic resolution — no deterministic style matching.
+    # Style is a semantic field: "shoes" should surface boots, sneakers, sandals, etc.
+    # Category and gender_affinity are hard kNN pre-filters.
+    # max_price is only applied as a kNN filter when the user explicitly mentions
+    # price in their original query. The agent may infer max_price from profile
+    # (useful for reranking/explanation) but we don't use it as a hard filter
+    # unless the user asked for it.
+    style_filter = None
+    category_filter = attrs.get("category")
+    max_price_filter = attrs.get("max_price")
+    # Ensure max_price is numeric if present
+    if max_price_filter is not None:
+        try:
+            max_price_filter = float(max_price_filter)
+        except (ValueError, TypeError):
+            max_price_filter = None
+
+    # Only use max_price as a kNN pre-filter if the user explicitly stated price
+    import re
+    _price_match = re.search(
+        r'\$\s*(\d+(?:\.\d+)?)|(?:under|below|less\s+than|cheaper\s+than|max)\s+\$?\s*(\d+(?:\.\d+)?)|(\d+(?:\.\d+)?)\s*dollars',
+        query, re.IGNORECASE
+    )
+    if _price_match:
+        # Extract the price value from whichever group matched
+        _explicit_price = next((g for g in _price_match.groups() if g is not None), None)
+        if _explicit_price:
+            knn_price_filter = float(_explicit_price)
+        else:
+            knn_price_filter = max_price_filter
+    else:
+        knn_price_filter = None
+
+    from search_personalization.agentic_memory.memory_tools import search_product_catalog, rerank_results
+    gender_filter = attrs.get("gender_affinity")
+    search_results = search_product_catalog(
+        query=enriched_query,
+        category=category_filter,
+        max_price=knn_price_filter,
+        gender_affinity=gender_filter,
+    )
+
+    search_duration = time.time() - start
+    print(f"          Done ({search_duration:.1f}s)")
+
+    # Extract the actual OpenSearch query from the search results for display
+    try:
+        _search_parsed = json.loads(search_results) if isinstance(search_results, str) else search_results
+        actual_opensearch_query = _search_parsed.get("_opensearch_query")
+    except (json.JSONDecodeError, TypeError):
+        actual_opensearch_query = None
+
+    # --- Step 3: Rerank + filter aversions (no LLM needed) ---
+    print(f"  [rerank] Reranking + filtering...")
+    start = time.time()
+    # Build a rerank query that blends user intent with style/use-context signals.
+    # The search step already used the enriched query (with profile preferences),
+    # so reranking surfaces results that best match what the user actually asked for
+    # plus their functional context.
+    rerank_parts = [query]
+    use_ctx = attrs.get("use_context")
+    preferred_materials = attrs.get("preferred_materials", [])
+    if use_ctx:
+        rerank_parts.append(use_ctx)
+    if preferred_materials:
+        rerank_parts.append(", ".join(preferred_materials[:2]))
+    rerank_query = ", ".join(rerank_parts)
+    reranked_raw = rerank_results(query=rerank_query, documents=search_results, top_n=10)
+
+    # Aversions are handled semantically — the query agent expresses them as positive
+    # opposites in the enriched_query (e.g., "flashy" → "subtle understated") so the
+    # vector search naturally favors non-aversive products. No post-filtering needed.
+    raw_aversions = attrs.get("aversions", [])
+    reranked = json.loads(reranked_raw) if isinstance(reranked_raw, str) else reranked_raw
+    results_list = reranked.get("reranked_results", [])
+
+    # Price post-filter: apply after reranking so semantic relevance isn't starved.
+    # If too few results survive the price filter, keep the full set and let the
+    # explanation agent inform the user about the price gap.
+    price_filter_relaxed = False
+    # Only apply price filtering if the user explicitly mentioned price in their query
+    import re
+    _user_mentioned_price = bool(re.search(r'(\$\d|under|below|less than|cheap|budget|affordable|\d+\s*dollars)', query.lower()))
+    if max_price_filter and results_list and _user_mentioned_price:
+        price_filtered = [r for r in results_list if (r.get("price") or 0) <= max_price_filter]
+        if len(price_filtered) >= 3:
+            results_list = price_filtered
+        else:
+            # Not enough results within budget — keep full set, flag for explanation
+            price_filter_relaxed = True
+            print(f"  [price] Only {len(price_filtered)} results under ${max_price_filter:.0f}, showing all with explanation")
+
+    # Build final output
+    final_results = {
+        "results": [
+            {"rank": i + 1, "product_id": r.get("id"), "name": r.get("name"), "price": r.get("price"), "score": r.get("rerank_score")}
+            for i, r in enumerate(results_list[:7])
+        ],
+        "enriched_query": enriched_query,
+        "personalization_summary": (
+            f"Use context: {use_ctx or 'none'}, "
+            f"materials: {preferred_materials or 'none'}, "
+            f"filtered aversions: {raw_aversions}"
+        ) if (raw_aversions or use_ctx or preferred_materials) else "No personalization applied",
+    }
+    ranking_result = json.dumps(final_results, indent=2, default=str)
+    ranking_duration = time.time() - start
+    pipeline_trace["agents"]["ranking"] = {
+        "duration_ms": int(ranking_duration * 1000),
+        "output": ranking_result,
+    }
+    print(f"          Done ({ranking_duration:.1f}s)")
+
+    # Combine for working memory
+    combined_payload = json.dumps({**enrichment, "search_results": results_list}, default=str)
+
+    # --- Async: write working memory for observability (fire-and-forget) ---
+    def _write_working_memory_async():
+        try:
+            from search_personalization.agentic_memory.memory_tools import write_working_memory
+            write_working_memory(persona_id, session_id, combined_payload)
+        except Exception:
+            pass
+
+    threading.Thread(target=_write_working_memory_async, daemon=True).start()
+
+    # --- Async: Results Explanation Agent (runs in parallel, doesn't block) ---
+    from concurrent.futures import Future
+    explanation_future: Future = Future()
+
+    def _run_explanation_async():
+        try:
+            from search_personalization.agentic_memory.agents.results_explanation_agent import (
+                invoke_results_explanation_agent,
+            )
+            explanation = invoke_results_explanation_agent(
+                query=query,
+                results=results_list[:5],
+                price_filter_relaxed=price_filter_relaxed,
+                original_max_price=max_price_filter if price_filter_relaxed else None,
+                inferred_attributes=attrs,
+            )
+            explanation_future.set_result(explanation)
+        except Exception:
+            explanation_future.set_result(None)
+
+    explanation_thread = threading.Thread(target=_run_explanation_async, daemon=True)
+    explanation_thread.start()
+
+    pipeline_trace["total_duration_ms"] = int((time.time() - pipeline_start) * 1000)
+    pipeline_trace["opensearch_query"] = actual_opensearch_query
+    print(f"\n  Total pipeline: {pipeline_trace['total_duration_ms']}ms")
+
+    # Price relaxation explanation is synchronous (no LLM needed, just a string).
+    # LLM-based explanations run fully async — the future is returned in the trace
+    # so the caller can resolve it without blocking the pipeline.
+    if price_filter_relaxed and max_price_filter:
+        pipeline_trace["agents"]["explanation"] = {
+            "output": f"Not enough options under ${max_price_filter:.0f} — showing best matches across all price ranges."
+        }
+    else:
+        pipeline_trace["agents"]["explanation"] = {"output": None, "_future": explanation_future}
+
+    # --- Post-pipeline: session memory write (async fire-and-forget) ---
+    def _write_session_memory_async():
+        try:
+            top_results = ", ".join(
+                r.get("name", "") for r in results_list[:3]
+            )
+            session_content = (
+                f"User searched: \"{query}\". "
+                f"Category: {attrs.get('category') or 'general'}. "
+                f"Top results: {top_results}."
+            )
+            write_session_memory(persona_id, session_id, session_content)
+        except Exception:
+            pass
+
+    threading.Thread(target=_write_session_memory_async, daemon=True).start()
+
+    # History write can remain async (not needed for multi-turn)
+    def _write_history_async():
+        try:
+            write_history(persona_id, session_id,
+                f"Session {session_id} | Query: \"{query}\" | "
+                f"Persona: {persona_name} | Timestamp: {pipeline_trace['timestamp']}")
+        except Exception:
+            pass
+
+    threading.Thread(target=_write_history_async, daemon=True).start()
+
+    return pipeline_trace
+
+
+def run_comparison_demo(query: str):
+    """Run the same query through both personas side by side."""
+    print("\n" + "=" * 70)
+    print("  AGENTIC MEMORY DEMO — PERSONALIZED SEARCH")
+    print("=" * 70)
+    print(f"\n  Query: \"{query}\"")
+    print(f"  Personas: Sarah (classic/professional style) vs Alex (casual/streetwear style)")
+    print(f"  Architecture: Cache → Query Understanding → Ranking (3 agents)")
+
+    demo_session_prefix = f"demo-{uuid.uuid4().hex[:6]}"
+
+    sarah_trace = run_pipeline_for_persona(
+        query=query, persona_name="sarah", session_id=f"{demo_session_prefix}-sarah",
+    )
+    alex_trace = run_pipeline_for_persona(
+        query=query, persona_name="alex", session_id=f"{demo_session_prefix}-alex",
+    )
+
+    print("\n" + "=" * 70)
+    print("  COMPARISON SUMMARY")
+    print("=" * 70)
+    print(f"\n  Same query: \"{query}\"")
+    print(f"\n  Sarah's results:")
+    print(f"    {_extract_summary(sarah_trace['agents'].get('ranking', {}).get('output', ''))}")
+    print(f"\n  Alex's results:")
+    print(f"    {_extract_summary(alex_trace['agents'].get('ranking', {}).get('output', ''))}")
+    print(f"\n  Key insight: Same query — memory + decoupled ranking is the difference.")
+    print("=" * 70)
+
+    return {"sarah": sarah_trace, "alex": alex_trace}
+
+
+def _extract_summary(output: str) -> str:
+    if not output:
+        return "(no output)"
+    clean = output.replace("\n", " ").strip()
+    return clean[:300] + "..." if len(clean) > 300 else clean
+
+
+def run_interactive():
+    """Interactive mode."""
+    print("\n" + "=" * 70)
+    print("  AGENTIC MEMORY DEMO — INTERACTIVE MODE")
+    print("=" * 70)
+    print("\n  Type a query to run through both personas")
+    print("  'sarah: <query>' or 'alex: <query>' for one persona")
+    print("  'quit' to stop\n")
+
+    session_ids = {
+        "sarah": f"interactive-{uuid.uuid4().hex[:6]}-sarah",
+        "alex": f"interactive-{uuid.uuid4().hex[:6]}-alex",
+    }
+
+    while True:
+        try:
+            user_input = input("Query: ").strip()
+        except (EOFError, KeyboardInterrupt):
+            print("\nGoodbye!")
+            break
+
+        if not user_input or user_input.lower() in ("quit", "exit"):
+            break
+
+        if user_input.lower().startswith("sarah:"):
+            run_pipeline_for_persona(user_input[6:].strip(), "sarah", session_ids["sarah"])
+        elif user_input.lower().startswith("alex:"):
+            run_pipeline_for_persona(user_input[5:].strip(), "alex", session_ids["alex"])
+        else:
+            run_comparison_demo(user_input)
+        print()
+
+
+def main():
+    import argparse
+    parser = argparse.ArgumentParser(description="Run the agentic memory demo pipeline")
+    parser.add_argument("--query", type=str, default=DEFAULT_QUERY)
+    parser.add_argument("--interactive", action="store_true")
+    parser.add_argument("--persona", type=str, choices=["sarah", "alex"])
+    args = parser.parse_args()
+
+    # Refresh schema cache and start daily scheduler (7am)
+    from search_personalization.agentic_memory.schema_cache import refresh_cache, start_scheduler
+    try:
+        print("  Refreshing schema cache...")
+        refresh_cache()
+    except Exception as e:
+        print(f"  Schema cache refresh failed (will use defaults): {e}")
+    start_scheduler(hour=7, minute=0)
+
+    if args.interactive:
+        run_interactive()
+    elif args.persona:
+        run_pipeline_for_persona(query=args.query, persona_name=args.persona)
+    else:
+        run_comparison_demo(args.query)
+
+
+if __name__ == "__main__":
+    main()

search_personalization/agentic_memory/memory_client.py

@@ -0,0 +1,168 @@
+"""
+OpenSearch Agentic Memory client.
+Wraps the /_plugins/_ml/memory_containers/ API (introduced in OpenSearch 3.3).
+"""
+
+from typing import Any, Optional
+from opensearchpy import OpenSearch
+from search_personalization.data_loader import get_opensearch_client
+from search_personalization.agentic_memory.config import EMBEDDING_MODEL_ID, MEMORY_LLM_MODEL_ID, EMBEDDING_DIMENSION
+
+
+def _client() -> OpenSearch:
+    """Get the shared OpenSearch client."""
+    return get_opensearch_client()
+
+
+# --- Container Management ---
+
+
+def create_container(
+    name: str,
+    description: str,
+    strategies: list[dict],
+) -> dict:
+    """
+    Create a memory container.
+
+    Args:
+        name: Human-readable container name.
+        description: What this container stores.
+        strategies: List of strategy dicts, e.g. [{"type": "USER_PREFERENCE", "namespace": ["user_id"]}]
+    """
+    client = _client()
+    body = {
+        "name": name,
+        "description": description,
+        "configuration": {
+            "embedding_model_type": "TEXT_EMBEDDING",
+            "embedding_model_id": EMBEDDING_MODEL_ID,
+            "embedding_dimension": EMBEDDING_DIMENSION,
+            "llm_id": MEMORY_LLM_MODEL_ID,
+            "strategies": strategies,
+        },
+    }
+    return client.transport.perform_request(
+        "POST", "/_plugins/_ml/memory_containers/_create", body=body
+    )
+
+
+def list_containers() -> dict:
+    """Search all memory containers."""
+    client = _client()
+    body = {"query": {"match_all": {}}}
+    return client.transport.perform_request(
+        "POST", "/_plugins/_ml/memory_containers/_search", body=body
+    )
+
+
+def get_container(container_id: str) -> dict:
+    """Get a specific container by ID."""
+    client = _client()
+    return client.transport.perform_request(
+        "GET", f"/_plugins/_ml/memory_containers/{container_id}"
+    )
+
+
+def delete_container(container_id: str) -> dict:
+    """Delete a memory container."""
+    client = _client()
+    return client.transport.perform_request(
+        "DELETE", f"/_plugins/_ml/memory_containers/{container_id}"
+    )
+
+
+# --- Memory Operations ---
+
+
+def write_memory(
+    container_id: str,
+    namespace: dict,
+    content: str,
+    payload_type: str = "conversational",
+    infer: bool = True,
+) -> dict:
+    """
+    Write a memory to a container.
+
+    Args:
+        container_id: The memory container ID.
+        namespace: Namespace dict, e.g. {"user_id": "user1", "session_id": "s1"}.
+        content: The text content to store.
+        payload_type: "conversational" or "data".
+        infer: Whether to use LLM to extract knowledge.
+    """
+    client = _client()
+    body: dict[str, Any] = {
+        "messages": [
+            {
+                "role": "user",
+                "content": [{"text": content, "type": "text"}],
+            }
+        ],
+        "namespace": namespace,
+        "payload_type": payload_type,
+        "infer": infer,
+    }
+    return client.transport.perform_request(
+        "POST", f"/_plugins/_ml/memory_containers/{container_id}/memories", body=body
+    )
+
+
+def read_memory(
+    container_id: str,
+    namespace: dict,
+    memory_type: str = "long-term",
+    size: int = 10,
+) -> dict:
+    """
+    Search memories in a container by namespace.
+
+    Args:
+        container_id: The memory container ID.
+        namespace: Namespace dict to filter by.
+        memory_type: One of "sessions", "working", "long-term", "history".
+        size: Max results.
+    """
+    client = _client()
+    must_clauses = [{"term": {f"namespace.{k}": v}} for k, v in namespace.items()]
+    body = {"query": {"bool": {"must": must_clauses}}, "size": size}
+    return client.transport.perform_request(
+        "GET",
+        f"/_plugins/_ml/memory_containers/{container_id}/memories/{memory_type}/_search",
+        body=body,
+    )
+
+
+def search_memory(
+    container_id: str,
+    namespace: dict,
+    query: str,
+    memory_type: str = "long-term",
+    size: int = 5,
+) -> dict:
+    """
+    Semantic search within a memory container namespace.
+
+    Args:
+        container_id: The memory container ID.
+        namespace: Namespace dict to filter by.
+        query: Natural language query for semantic search.
+        memory_type: One of "sessions", "working", "long-term", "history".
+        size: Max results.
+    """
+    client = _client()
+    must_clauses = [{"term": {f"namespace.{k}": v}} for k, v in namespace.items()]
+    body = {
+        "query": {
+            "bool": {
+                "must": must_clauses + [{"match": {"content": query}}],
+            }
+        },
+        "size": size,
+    }
+    return client.transport.perform_request(
+        "GET",
+        f"/_plugins/_ml/memory_containers/{container_id}/memories/{memory_type}/_search",
+        body=body,
+    )

search_personalization/agentic_memory/memory_tools.py

@@ -0,0 +1,347 @@
+"""
+Strands @tool functions for agents to read/write OpenSearch agentic memory.
+These tools are shared across the Query Understanding and Ranking agents.
+"""
+
+import json
+import os
+from typing import Optional
+from strands import tool
+
+from search_personalization.agentic_memory.config import CONTAINER_ID
+from search_personalization.agentic_memory.memory_client import read_memory, write_memory
+
+# In-memory cache for rerank results (keyed by hash of query + documents)
+_rerank_cache: dict[str, str] = {}
+
+
+def _get_container_id() -> str:
+    """Get the single memory container ID, raising if not configured."""
+    if not CONTAINER_ID:
+        raise ValueError(
+            "MEMORY_CONTAINER_ID not configured. Run setup or set the env var."
+        )
+    return CONTAINER_ID
+
+
+@tool
+def get_user_profile(persona_id: str) -> str:
+    """Retrieve the user's long-term memory profile from OpenSearch agentic memory.
+    Returns extracted preferences including color preferences, style, size, price sensitivity,
+    and any explicit aversions learned from purchase history, returns, and reviews.
+
+    Args:
+        persona_id: The user ID (e.g., user1, user2).
+    """
+    from search_personalization.agentic_memory.memory_client import _client
+
+    container_id = _get_container_id()
+    namespace = {"user_id": persona_id}
+
+    # Query long-term memory filtered to USER_PREFERENCE strategy only
+    client = _client()
+    body = {
+        "query": {"bool": {"must": [
+            {"term": {"namespace.user_id": persona_id}},
+            {"term": {"strategy_type": "USER_PREFERENCE"}},
+        ]}},
+        "size": 20,
+    }
+    result = client.transport.perform_request(
+        "GET",
+        f"/_plugins/_ml/memory_containers/{container_id}/memories/long-term/_search",
+        body=body,
+    )
+
+    memories = result.get("hits", {}).get("hits", [])
+    trace = {
+        "memory_reads": [
+            {
+                "namespace": namespace,
+                "type": "long-term",
+                "strategy_filter": "USER_PREFERENCE",
+                "records_retrieved": len(memories),
+            }
+        ]
+    }
+
+    return json.dumps(
+        {"persona_id": persona_id, "memories": memories, "trace": trace},
+        indent=2,
+        default=str,
+    )
+
+
+@tool
+def read_session_memory(persona_id: str, session_id: str) -> str:
+    """Retrieve current session context for multi-turn query understanding.
+    Returns previous queries and interactions within this shopping session.
+
+    Args:
+        persona_id: The user ID (e.g., user1, user2).
+        session_id: The current session identifier.
+    """
+    container_id = _get_container_id()
+    namespace = {"user_id": persona_id, "session_id": session_id}
+
+    result = read_memory(
+        container_id=container_id,
+        namespace=namespace,
+        memory_type="sessions",
+    )
+
+    memories = result.get("hits", {}).get("hits", [])
+    trace = {
+        "memory_reads": [
+            {
+                "namespace": namespace,
+                "type": "session",
+                "records_retrieved": len(memories),
+            }
+        ]
+    }
+
+    return json.dumps(
+        {"persona_id": persona_id, "session_id": session_id, "memories": memories, "trace": trace},
+        indent=2,
+        default=str,
+    )
+
+
+@tool
+def write_session_memory(persona_id: str, session_id: str, content: str) -> str:
+    """Write a conversation turn to session memory for multi-turn context.
+
+    Args:
+        persona_id: The user ID.
+        session_id: The current session identifier.
+        content: The conversation turn content to store (query + results summary).
+    """
+    container_id = _get_container_id()
+    namespace = {"user_id": persona_id, "session_id": session_id}
+
+    result = write_memory(
+        container_id=container_id,
+        namespace=namespace,
+        content=content,
+    )
+
+    trace = {
+        "memory_writes": [
+            {
+                "namespace": namespace,
+                "type": "session",
+                "action": "write_session_turn",
+            }
+        ]
+    }
+
+    return json.dumps({"status": "written", "trace": trace}, indent=2, default=str)
+
+
+@tool
+def write_working_memory(persona_id: str, session_id: str, enriched_query: str) -> str:
+    """Write the enriched query decomposition to working memory for the Ranking Agent to consume.
+    This is the inter-agent communication channel.
+
+    Args:
+        persona_id: The user ID.
+        session_id: The current session identifier.
+        enriched_query: JSON string containing the enriched, decomposed query with all inferred attributes.
+    """
+    container_id = _get_container_id()
+    namespace = {"user_id": persona_id, "session_id": session_id}
+
+    result = write_memory(
+        container_id=container_id,
+        namespace=namespace,
+        content=enriched_query,
+        infer=False,
+    )
+
+    trace = {
+        "memory_writes": [
+            {
+                "namespace": namespace,
+                "type": "working-memory",
+                "action": "write_enriched_query",
+            }
+        ]
+    }
+
+    return json.dumps({"status": "written", "trace": trace}, indent=2, default=str)
+
+
+@tool
+def write_history(persona_id: str, session_id: str, content: str) -> str:
+    """Append a completed interaction to the user's history for long-term pattern learning.
+
+    Args:
+        persona_id: The user ID.
+        session_id: The session identifier.
+        content: Summary of the completed interaction (query, enrichment, results, outcome).
+    """
+    container_id = _get_container_id()
+    namespace = {"user_id": persona_id, "session_id": session_id}
+
+    result = write_memory(
+        container_id=container_id,
+        namespace=namespace,
+        content=content,
+    )
+
+    trace = {
+        "memory_writes": [
+            {
+                "namespace": namespace,
+                "type": "history",
+                "action": "append_interaction",
+            }
+        ]
+    }
+
+    return json.dumps({"status": "written", "trace": trace}, indent=2, default=str)
+
+
+@tool
+def search_product_catalog(query: str, category: Optional[str] = None, max_price: Optional[float] = None, gender_affinity: Optional[str] = None, size: int = 10) -> str:
+    """Search the product catalog using server-side neural query (OpenSearch does the embedding).
+    Style/product-type matching is handled semantically by the vector search — do NOT use term filters
+    for fields with semantic meaning.
+
+    Only truly categorical/numeric fields are used as hard filters:
+    - category: broad taxonomy (5 values) — safe for exact filtering
+    - max_price: numeric range — safe for range filtering
+    - gender_affinity: hard filter ("M" or "F") to ensure persona-appropriate results
+
+    Args:
+        query: Natural language search query (e.g., "navy leather boots").
+        category: Optional category filter (apparel, footwear, accessories, jewelry, electronics).
+        max_price: Optional maximum price filter.
+        gender_affinity: Optional gender affinity filter ("M" or "F"). Hard-filters to persona-appropriate products.
+        size: Max number of results to return.
+    """
+    import os
+    from search_personalization.data_loader import get_opensearch_client
+
+    client = get_opensearch_client()
+    model_id = os.getenv('OPENSEARCH_MODEL_ID', 'default_model_id')
+
+    # Hard filters: only for truly categorical/numeric fields
+    filter_clauses = []
+    if category:
+        filter_clauses.append({"term": {"category": category}})
+    if max_price:
+        filter_clauses.append({"range": {"price": {"lte": max_price}}})
+    if gender_affinity:
+        filter_clauses.append({"term": {"gender_affinity": gender_affinity.upper()}})
+
+    # Use server-side neural query — OpenSearch calls the embedding model via its ML connector
+    neural_query: dict = {
+        "neural": {
+            "product_description_vector": {
+                "query_text": query,
+                "model_id": model_id,
+                "k": size,
+            }
+        }
+    }
+
+    if filter_clauses:
+        # Use neural query's built-in filter (pre-filter before kNN neighbor selection)
+        neural_query["neural"]["product_description_vector"]["filter"] = {"bool": {"filter": filter_clauses}}
+
+    query_body = {
+        "size": size,
+        "query": neural_query,
+    }
+
+    resp = client.search(index="products", body=query_body)
+    hits = resp["hits"]["hits"]
+
+    results = []
+    for h in hits:
+        src = h["_source"]
+        results.append({
+            "id": src.get("id"),
+            "name": src.get("name"),
+            "category": src.get("category"),
+            "style": src.get("style"),
+            "price": src.get("price"),
+            "description": src.get("description", "")[:150],
+            "score": h.get("_score"),
+        })
+
+    return json.dumps({"total_hits": resp["hits"]["total"]["value"], "results": results, "_opensearch_query": query_body}, indent=2, default=str)
+
+
+@tool
+def rerank_results(query: str, documents: str, top_n: int = 10) -> str:
+    """Rerank search results using Cohere Rerank 3.5 via Bedrock.
+    Call this AFTER search_product_catalog to reorder results by relevance
+    to the enriched query. Pass the enriched query (with user preferences baked in)
+    as the query parameter for personalized reranking.
+
+    Args:
+        query: The enriched query string to rerank against (e.g., "spacious tan leather backpack, neutral colors, business-casual, $75-$175").
+        documents: JSON string — either the full search_product_catalog output or just the results array.
+        top_n: Number of top results to return after reranking.
+    """
+    import hashlib
+    import boto3
+
+    # Cache lookup — keyed by query + document content
+    cache_key = hashlib.sha256((query + documents if isinstance(documents, str) else query + json.dumps(documents, sort_keys=True)).encode()).hexdigest()
+    if cache_key in _rerank_cache:
+        return _rerank_cache[cache_key]
+
+    client = boto3.client("bedrock-runtime", region_name=os.getenv("AWS_REGION", "us-east-1"))
+
+    # Robustly parse documents — handle both full output and just the array
+    if isinstance(documents, str):
+        parsed = json.loads(documents)
+    else:
+        parsed = documents
+
+    if isinstance(parsed, dict) and "results" in parsed:
+        docs = parsed["results"]
+    elif isinstance(parsed, list):
+        docs = parsed
+    else:
+        docs = parsed.get("reranked_results", []) if isinstance(parsed, dict) else []
+
+    if not docs:
+        return json.dumps({"error": "No documents to rerank"})
+
+    # Build document strings for Cohere Rerank API (must be plain strings)
+    doc_texts = []
+    for d in docs:
+        text = f"{d.get('name', '')}. {d.get('description', '')}. Style: {d.get('style', '')}. Category: {d.get('category', '')}. Price: ${d.get('price', '')}"
+        doc_texts.append(text)
+
+    body = json.dumps({
+        "query": query,
+        "documents": doc_texts,
+        "top_n": min(top_n, len(doc_texts)),
+        "api_version": 2,
+    })
+
+    response = client.invoke_model(
+        modelId="cohere.rerank-v3-5:0",
+        body=body,
+        contentType="application/json",
+        accept="application/json",
+    )
+
+    result = json.loads(response["body"].read())
+    reranked = []
+    for item in result.get("results", []):
+        idx = item["index"]
+        if idx < len(docs):
+            original = docs[idx].copy()
+            original["rerank_score"] = item["relevance_score"]
+            reranked.append(original)
+
+    result_json = json.dumps({"reranked_results": reranked}, indent=2, default=str)
+    _rerank_cache[cache_key] = result_json
+    return result_json

search_personalization/agentic_memory/personas/alex.json

@@ -0,0 +1,123 @@
+{
+  "persona_id": "user2",
+  "name": "Alex",
+  "description": "21-year-old Computer Science student in Austin, TX. Active lifestyle, prefers comfortable sporty clothing and sneakers. Dislikes fancy or flashy items.",
+  "raw_signals": [
+    {
+      "type": "purchase",
+      "item": "Comfortable Black Running Sneakers",
+      "category": "footwear",
+      "style": "sneaker",
+      "color": "black",
+      "date": "2026-01-05"
+    },
+    {
+      "type": "purchase",
+      "item": "Lightweight White Training Sneakers",
+      "category": "footwear",
+      "style": "sneaker",
+      "color": "white",
+      "date": "2026-01-18"
+    },
+    {
+      "type": "purchase",
+      "item": "Sporty Gray Jogger Pants",
+      "category": "apparel",
+      "style": "jacket",
+      "color": "gray",
+      "date": "2026-02-02"
+    },
+    {
+      "type": "purchase",
+      "item": "Athletic Navy Zip-Up Hoodie",
+      "category": "apparel",
+      "style": "jacket",
+      "color": "navy",
+      "date": "2026-02-15"
+    },
+    {
+      "type": "purchase",
+      "item": "Red Cushioned Sport Sneakers",
+      "category": "footwear",
+      "style": "sneaker",
+      "color": "red",
+      "date": "2026-03-01"
+    },
+    {
+      "type": "purchase",
+      "item": "Quick-Dry Athletic Shorts",
+      "category": "apparel",
+      "style": "shirt",
+      "color": "black",
+      "date": "2026-03-10"
+    },
+    {
+      "type": "purchase",
+      "item": "Sporty Digital Watch",
+      "category": "accessories",
+      "style": "watch",
+      "color": "black",
+      "date": "2026-03-20"
+    },
+    {
+      "type": "purchase",
+      "item": "Classic Navy Dress Shirt",
+      "category": "apparel",
+      "style": "shirt",
+      "color": "navy",
+      "date": "2026-04-05"
+    },
+    {
+      "type": "return",
+      "item": "Sparkling Elegant Bracelet",
+      "category": "jewelry",
+      "style": "bracelet",
+      "reason": "Not my style at all — too fancy and flashy. I prefer simple, functional stuff."
+    },
+    {
+      "type": "return",
+      "item": "Shiny Gold Statement Necklace",
+      "category": "jewelry",
+      "style": "necklace",
+      "reason": "Way too flashy for me. I don't wear jewelry like this."
+    },
+    {
+      "type": "review",
+      "item": "Comfortable Black Running Sneakers",
+      "category": "footwear",
+      "text": "Super comfortable for walking around campus and hitting the gym. Exactly what I needed — lightweight and breathable."
+    },
+    {
+      "type": "review",
+      "item": "Red Cushioned Sport Sneakers",
+      "category": "footwear",
+      "text": "Great cushioning for running. I wear these for my morning jogs and they hold up well."
+    },
+    {
+      "type": "review",
+      "item": "Athletic Navy Zip-Up Hoodie",
+      "category": "apparel",
+      "text": "Perfect for throwing on before and after workouts. Comfortable and simple."
+    },
+    {
+      "type": "support_chat",
+      "text": "I wear size M in tops and 10 in shoes. Mostly looking for sporty comfortable stuff for campus and the gym."
+    },
+    {
+      "type": "browse_pattern",
+      "description": "Viewed 25 sneakers, 10 athletic wear items, 2 formal shirts. Filtered by 'sneaker' style 15 times. Never browsed jewelry or accessories beyond watches."
+    },
+    {
+      "type": "wishlist_add",
+      "item": "Breathable Mesh Running Sneakers",
+      "category": "footwear",
+      "style": "sneaker"
+    },
+    {
+      "type": "wishlist_add",
+      "item": "Lightweight Training Tank Top",
+      "category": "apparel",
+      "style": "shirt"
+    }
+  ]
+}

search_personalization/agentic_memory/personas/sarah.json

@@ -0,0 +1,87 @@
+{
+  "persona_id": "user1",
+  "name": "Sarah",
+  "description": "32-year-old Marketing Manager in San Francisco. Quality-focused, trend-aware shopper who values polished, professional style.",
+  "raw_signals": [
+    {
+      "type": "purchase",
+      "item": "Navy Casual Jacket",
+      "category": "apparel",
+      "style": "jacket",
+      "color": "navy",
+      "date": "2026-01-10"
+    },
+    {
+      "type": "purchase",
+      "item": "Sleek Saddle Leather Knee Boots",
+      "category": "footwear",
+      "style": "boot",
+      "color": "saddle brown",
+      "date": "2026-01-25"
+    },
+    {
+      "type": "purchase",
+      "item": "Sophisticated Everyday Elegance Bracelet",
+      "category": "jewelry",
+      "style": "bracelet",
+      "color": "gold",
+      "date": "2026-02-05"
+    },
+    {
+      "type": "purchase",
+      "item": "Spacious Tan Backpack for Her Travels",
+      "category": "accessories",
+      "style": "backpack",
+      "color": "tan",
+      "date": "2026-02-20"
+    },
+    {
+      "type": "purchase",
+      "item": "Timeless Grace Bracelet",
+      "category": "jewelry",
+      "style": "bracelet",
+      "color": "silver",
+      "date": "2026-03-01"
+    },
+    {
+      "type": "return",
+      "item": "Stylish Orange Backpack",
+      "category": "accessories",
+      "style": "backpack",
+      "reason": "Too flashy for my work environment, I prefer neutral tones that go with everything"
+    },
+    {
+      "type": "return",
+      "item": "Cozy Orange Ski Jacket",
+      "category": "apparel",
+      "style": "jacket",
+      "reason": "The color is too loud for me. I wanted something more understated"
+    },
+    {
+      "type": "review",
+      "item": "Sleek Saddle Leather Knee Boots",
+      "category": "footwear",
+      "text": "Perfect quality leather, exactly the classic style I look for. Goes with everything from jeans to work outfits."
+    },
+    {
+      "type": "review",
+      "item": "Navy Casual Jacket",
+      "category": "apparel",
+      "text": "Love the fit and the fabric quality. Navy is my go-to color — professional but not boring."
+    },
+    {
+      "type": "support_chat",
+      "text": "I recently changed jobs and need more business-casual pieces. Also my shoe size is now 8 instead of 7.5 — can you update my profile?"
+    },
+    {
+      "type": "browse_pattern",
+      "description": "Viewed 15 leather boots, 8 jackets in navy/gray/black, 0 sneakers."
+    },
+    {
+      "type": "wishlist_add",
+      "item": "Cozy Dark Gray Women's Jacket",
+      "category": "apparel",
+      "style": "jacket"
+    }
+  ]
+}

search_personalization/agentic_memory/schema_cache.py

@@ -0,0 +1,104 @@
+"""
+Schema Cache — Discovers the product index schema from OpenSearch and caches it.
+
+Runs on a schedule (daily at 7am) to keep the Query Understanding Agent prompt
+in sync with the actual index mapping without per-query overhead.
+"""
+
+import json
+import threading
+import time
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Optional
+
+from search_personalization.data_loader import get_opensearch_client
+
+_CACHE_FILE = Path(__file__).parent / "schema_cache" / "products_schema.json"
+_FILTERABLE_TYPES = {"keyword", "float", "integer", "boolean"}
+_INDEX_NAME = "products"
+_EXCLUDED_FIELDS = {"id", "image", "image_binary", "product_description_vector"}
+_MAX_TERMS = 50
+
+
+def discover_schema() -> dict:
+    """Fetch index mapping and top terms for filterable fields from OpenSearch."""
+    client = get_opensearch_client()
+
+    # Get mapping
+    mapping_resp = client.indices.get_mapping(index=_INDEX_NAME)
+    properties = mapping_resp[_INDEX_NAME]["mappings"]["properties"]
+
+    # Identify filterable fields
+    schema = {}
+    keyword_fields = []
+    for field_name, field_def in properties.items():
+        if field_name in _EXCLUDED_FIELDS:
+            continue
+        field_type = field_def.get("type")
+        if field_type in _FILTERABLE_TYPES:
+            schema[field_name] = {"type": field_type, "values": None}
+            if field_type == "keyword":
+                keyword_fields.append(field_name)
+
+    # Get top terms for keyword fields via aggregations
+    if keyword_fields:
+        aggs = {f: {"terms": {"field": f, "size": _MAX_TERMS}} for f in keyword_fields}
+        agg_resp = client.search(index=_INDEX_NAME, body={"size": 0, "aggs": aggs})
+        for field_name in keyword_fields:
+            buckets = agg_resp["aggregations"][field_name].get("buckets", [])
+            schema[field_name]["values"] = [b["key"] for b in buckets]
+
+    return {"index": _INDEX_NAME, "discovered_at": datetime.now().isoformat(), "fields": schema}
+
+
+def refresh_cache() -> dict:
+    """Run discovery and write result to disk cache."""
+    schema = discover_schema()
+    _CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    _CACHE_FILE.write_text(json.dumps(schema, indent=2))
+    return schema
+
+
+def get_cached_schema() -> Optional[dict]:
+    """Load cached schema from disk. Returns None if not cached."""
+    if _CACHE_FILE.exists():
+        return json.loads(_CACHE_FILE.read_text())
+    return None
+
+
+def get_schema_prompt_block() -> str:
+    """Render the cached schema as a prompt block for the Query Agent."""
+    schema = get_cached_schema()
+    if not schema:
+        return ""
+
+    lines = [f"PRODUCT INDEX SCHEMA (discovered from '{schema['index']}' index, use these exact values for filters):"]
+    for field_name, field_def in schema["fields"].items():
+        if field_def["values"]:
+            lines.append(f"- {field_name} ({field_def['type']}): {', '.join(str(v) for v in field_def['values'])}")
+        else:
+            lines.append(f"- {field_name} ({field_def['type']})")
+    return "\n".join(lines)
+
+
+def _run_scheduler(hour: int = 7, minute: int = 0):
+    """Background thread that refreshes the cache daily at the specified time."""
+    while True:
+        now = datetime.now()
+        target = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
+        if target <= now:
+            target += timedelta(days=1)
+        wait_seconds = (target - now).total_seconds()
+        time.sleep(wait_seconds)
+        try:
+            refresh_cache()
+            print(f"[schema_cache] Refreshed at {datetime.now().isoformat()}")
+        except Exception as e:
+            print(f"[schema_cache] Refresh failed: {e}")
+
+
+def start_scheduler(hour: int = 7, minute: int = 0):
+    """Start the daily schema refresh as a daemon thread."""
+    t = threading.Thread(target=_run_scheduler, args=(hour, minute), daemon=True)
+    t.start()

search_personalization/data_loader.py

@@ -0,0 +1,383 @@
+"""Data loader for products from YAML file and OpenSearch indexing."""
+import os
+import yaml
+from pathlib import Path
+from typing import List, Dict, Any, Optional
+from opensearchpy import OpenSearch, RequestsHttpConnection, AWSV4SignerAuth, helpers
+import boto3
+from dotenv import load_dotenv
+
+
+# Load environment variables
+load_dotenv()
+
+# OpenSearch configuration from environment variables
+OPENSEARCH_ENDPOINT = os.getenv("OPENSEARCH_ENDPOINT", "")
+AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
+
+
+def load_products_from_yaml(yaml_path: str = "../search_personalization/product_samples/products.yaml") -> List[Dict[str, Any]]:
+    """
+    Load products from YAML file.
+    
+    Args:
+        yaml_path: Path to the products YAML file (relative to src directory)
+        
+    Returns:
+        List of product dictionaries
+    """
+    # Get path relative to this file
+    file_path = Path(__file__).parent / yaml_path
+    
+    if not file_path.exists():
+        raise FileNotFoundError(f"Products YAML file not found at: {file_path}")
+    
+    print(f"Loading products from: {file_path}")
+    
+    with open(file_path, 'r') as f:
+        products = yaml.safe_load(f)
+    
+    return products if products else []
+
+
+def get_sample_products(count: int = 10) -> List[Dict[str, Any]]:
+    """
+    Get a sample of products from the YAML file.
+    
+    Args:
+        count: Number of products to return
+        
+    Returns:
+        List of product dictionaries
+    """
+    products = load_products_from_yaml()
+    return products[:count]
+
+
+_opensearch_client: Optional[OpenSearch] = None
+
+
+def get_opensearch_client(endpoint: str = OPENSEARCH_ENDPOINT) -> OpenSearch:
+    """
+    Return a cached OpenSearch client singleton (created once, reused).
+    
+    Authenticates using AWS SigV4 signing with the IAM role resolved from
+    the configured AWS_PROFILE (or default credential chain).
+    
+    Args:
+        endpoint: OpenSearch cluster endpoint URL
+        
+    Returns:
+        OpenSearch client instance
+    """
+    global _opensearch_client
+    if _opensearch_client is not None:
+        return _opensearch_client
+
+    # Extract host from endpoint URL
+    host = endpoint.replace('https://', '').replace('http://', '')
+    
+    # Authenticate via AWS SigV4 signing using IAM role
+    aws_profile = os.getenv("AWS_PROFILE")
+    session = boto3.Session(profile_name=aws_profile) if aws_profile else boto3.Session()
+    credentials = session.get_credentials()
+    if not credentials:
+        raise ValueError("AWS credentials not found")
+    http_auth = AWSV4SignerAuth(credentials, AWS_REGION, 'es')
+    
+    # Create OpenSearch client
+    _opensearch_client = OpenSearch(
+        hosts=[{'host': host, 'port': 443}],
+        http_auth=http_auth,
+        use_ssl=True,
+        verify_certs=True,
+        connection_class=RequestsHttpConnection,
+        timeout=30,
+        max_retries=3,
+        retry_on_timeout=True
+    )
+    
+    return _opensearch_client
+
+
+def create_products_index(client: OpenSearch, index_name: str = 'products') -> None:
+    """
+    Create the products index with appropriate mappings.
+    
+    Args:
+        client: OpenSearch client
+        index_name: Name of the index to create
+    """
+    ingest_pipeline = os.getenv('OPENSEARCH_INGEST_PIPELINE', 'product-multimodal-pipeline')
+    index_body = {
+        'settings': {
+            'default_pipeline': ingest_pipeline,
+            'index.knn': True,
+            'number_of_shards': 1,
+            'number_of_replicas': 1
+        },
+        'mappings': {
+            'properties': {
+                'id': {'type': 'keyword'},
+                'name': {'type': 'text', 'analyzer': 'standard'},
+                'description': {'type': 'text', 'analyzer': 'standard'},
+                'price': {'type': 'float'},
+                'category': {'type': 'keyword'},
+                'style': {'type': 'keyword'},
+                'current_stock': {'type': 'integer'},
+                'image': {'type': 'keyword'},
+                'gender_affinity': {'type': 'keyword'},
+                'where_visible': {'type': 'keyword'},
+                'promoted': {'type': 'boolean'},
+                'product_description_vector': {
+                    'type': 'knn_vector',
+                    'dimension': 1024,
+                    'method': {
+                        'name': 'hnsw',
+                        'space_type': 'cosinesimil',
+                        'engine': 'faiss',
+                        'parameters': {
+                            'ef_construction': 128,
+                            'm': 16
+                        }
+                    }
+                }
+            }
+        }
+    }
+    
+    # Delete index if it exists
+    if client.indices.exists(index=index_name):
+        print(f"Deleting existing index: {index_name}")
+        client.indices.delete(index=index_name)
+    
+    # Create new index
+    print(f"Creating index: {index_name}")
+    client.indices.create(index=index_name, body=index_body)
+    print(f"Index '{index_name}' created successfully")
+
+
+def _get_image_base64(product: Dict[str, Any]) -> Optional[str]:
+    """Load product image as base64 string for multimodal embedding."""
+    import base64
+    category = product.get('category', '')
+    image_filename = product.get('image', '')
+    if not category or not image_filename:
+        return None
+    image_path = Path(__file__).parent.parent / 'product_samples' / category / image_filename
+    if not image_path.exists():
+        return None
+    with open(image_path, 'rb') as f:
+        return base64.b64encode(f.read()).decode('utf-8')
+
+
+def index_products_to_opensearch(
+    products: List[Dict[str, Any]], 
+    client: OpenSearch = None,
+    index_name: str = 'products'
+) -> tuple[int, int]:
+    """
+    Index products into OpenSearch using bulk API.
+    
+    Args:
+        products: List of product dictionaries
+        client: OpenSearch client (creates new one if None)
+        index_name: Name of the index
+        
+    Returns:
+        Tuple of (success_count, failed_count)
+    """
+    from opensearchpy import helpers
+    
+    if client is None:
+        client = get_opensearch_client()
+    
+    # Prepare bulk actions with image_binary for multimodal embedding
+    actions = []
+    for product in products:
+        doc = dict(product)
+        image_b64 = _get_image_base64(product)
+        if image_b64:
+            doc['image_binary'] = image_b64
+        action = {
+            '_index': index_name,
+            '_id': product['id'],
+            '_source': doc
+        }
+        actions.append(action)
+    
+    # Bulk index
+    print(f"Indexing {len(products)} products...")
+    success, failed = helpers.bulk(client, actions, raise_on_error=False)
+    
+    print(f"Successfully indexed: {success} products")
+    if failed:
+        print(f"Failed to index: {len(failed)} products")
+    
+    # Refresh index to make documents searchable
+    client.indices.refresh(index=index_name)
+    print(f"Index '{index_name}' refreshed")
+    
+    return success, len(failed) if failed else 0
+
+
+def load_yaml_to_opensearch(
+    yaml_path: str = "../search_personalization/product_samples/products.yaml",
+    endpoint: str = OPENSEARCH_ENDPOINT,
+    index_name: str = 'products',
+    recreate_index: bool = True
+) -> None:
+    """
+    Load products from YAML and index them into OpenSearch.
+    
+    Args:
+        yaml_path: Path to the products YAML file
+        endpoint: OpenSearch cluster endpoint URL
+        index_name: Name of the index
+        recreate_index: Whether to recreate the index (deletes existing)
+    """
+    # Load products from YAML
+    print("Loading products from YAML...")
+    products = load_products_from_yaml(yaml_path)
+    print(f"Loaded {len(products)} products")
+    
+    # Get OpenSearch client
+    print(f"Connecting to OpenSearch at {endpoint}...")
+    client = get_opensearch_client(endpoint)
+    
+    # Test connection
+    info = client.info()
+    print(f"Connected to cluster: {info['cluster_name']}")
+    
+    # Create index if needed
+    if recreate_index:
+        create_products_index(client, index_name)
+    
+    # Index products
+    success, failed = index_products_to_opensearch(products, client, index_name)
+    
+    print(f"\n✓ Indexing complete: {success} successful, {failed} failed")
+
+
+# --- Persona Index ---
+
+PERSONAS = [
+    {
+        "user_id": "anonymous",
+        "name": "Anonymous User",
+        "age": 30,
+        "gender": "unspecified",
+        "location": "General",
+        "occupation": "General User",
+        "interests": ["General Shopping"],
+        "shopping_behavior": "Standard search without personalization",
+        "key": "anonymous",
+        "emoji": "👤"
+    },
+    {
+        "name": "Sarah",
+        "user_id": "user1",
+        "age": 32,
+        "gender": "female",
+        "location": "San Francisco, CA",
+        "occupation": "Marketing Manager",
+        "interests": ["Technology", "Fitness", "Travel"],
+        "shopping_behavior": "Focused on quality and trends.",
+        "key": "business_professional",
+        "emoji": "👩‍💼"
+    },
+    {
+        "name": "Alex",
+        "user_id": "user2",
+        "age": 21,
+        "gender": "male",
+        "location": "Austin, TX",
+        "occupation": "Computer Science Student",
+        "interests": ["Gaming", "Programming", "Music"],
+        "shopping_behavior": "Budget-conscious, looks for deals and discounts",
+        "key": "student",
+        "emoji": "🎓"
+    }
+]
+
+
+def create_personas_index(client: OpenSearch, index_name: str = 'personas') -> None:
+    """Create the personas index with appropriate mappings."""
+    index_body = {
+        'settings': {
+            'number_of_shards': 1,
+            'number_of_replicas': 1
+        },
+        'mappings': {
+            'properties': {
+                'user_id': {'type': 'keyword'},
+                'name': {'type': 'text'},
+                'age': {'type': 'integer'},
+                'gender': {'type': 'keyword'},
+                'location': {'type': 'text', 'fields': {'keyword': {'type': 'keyword'}}},
+                'occupation': {'type': 'text', 'fields': {'keyword': {'type': 'keyword'}}},
+                'interests': {'type': 'keyword'},
+                'shopping_behavior': {'type': 'text'},
+                'key': {'type': 'keyword'},
+                'emoji': {'type': 'keyword'}
+            }
+        }
+    }
+
+    if client.indices.exists(index=index_name):
+        print(f"Deleting existing index: {index_name}")
+        client.indices.delete(index=index_name)
+
+    print(f"Creating index: {index_name}")
+    client.indices.create(index=index_name, body=index_body)
+    print(f"Index '{index_name}' created successfully")
+
+
+def index_personas(
+    client: OpenSearch = None,
+    index_name: str = 'personas'
+) -> tuple[int, int]:
+    """Index persona documents into OpenSearch."""
+    if client is None:
+        client = get_opensearch_client()
+
+    actions = [
+        {'_index': index_name, '_id': p['user_id'], '_source': p}
+        for p in PERSONAS
+    ]
+
+    print(f"Indexing {len(PERSONAS)} personas...")
+    success, failed = helpers.bulk(client, actions, raise_on_error=False)
+
+    print(f"Successfully indexed: {success} personas")
+    if failed:
+        print(f"Failed to index: {len(failed)} personas")
+
+    client.indices.refresh(index=index_name)
+    print(f"Index '{index_name}' refreshed")
+
+    return success, len(failed) if failed else 0
+
+
+def load_personas_to_opensearch(
+    endpoint: str = OPENSEARCH_ENDPOINT,
+    index_name: str = 'personas',
+    recreate_index: bool = True
+) -> None:
+    """Create the personas index and load all persona documents."""
+    print(f"Connecting to OpenSearch at {endpoint}...")
+    client = get_opensearch_client(endpoint)
+
+    info = client.info()
+    print(f"Connected to cluster: {info['cluster_name']}")
+
+    if recreate_index:
+        create_personas_index(client, index_name)
+
+    success, failed = index_personas(client, index_name)
+    print(f"\n✓ Persona indexing complete: {success} successful, {failed} failed")
+
+
+if __name__ == '__main__':
+    # Run the data loading when script is executed directly
+    load_yaml_to_opensearch()

search_personalization/iam-policy.json

@@ -0,0 +1,67 @@
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Sid": "OpenSearchRead",
+      "Effect": "Allow",
+      "Action": [
+        "es:ESHttpGet",
+        "es:ESHttpHead",
+        "es:ESHttpPost"
+      ],
+      "Resource": [
+        "arn:aws:es:us-east-1:099713751195:domain/mydomain/products*",
+        "arn:aws:es:us-east-1:099713751195:domain/mydomain/personas*",
+        "arn:aws:es:us-east-1:099713751195:domain/mydomain/user_preferences*",
+        "arn:aws:es:us-east-1:099713751195:domain/mydomain/_plugins/_ml/memory_containers*"
+      ]
+    },
+    {
+      "Sid": "OpenSearchWriteUserPreferences",
+      "Effect": "Allow",
+      "Action": [
+        "es:ESHttpPut",
+        "es:ESHttpPost"
+      ],
+      "Resource": [
+        "arn:aws:es:us-east-1:099713751195:domain/mydomain/user_preferences*"
+      ]
+    },
+    {
+      "Sid": "OpenSearchWriteMemoryContainers",
+      "Effect": "Allow",
+      "Action": [
+        "es:ESHttpPost"
+      ],
+      "Resource": [
+        "arn:aws:es:us-east-1:099713751195:domain/mydomain/_plugins/_ml/memory_containers*"
+      ]
+    },
+    {
+      "Sid": "BedrockInvokeModels",
+      "Effect": "Allow",
+      "Action": [
+        "bedrock:InvokeModel",
+        "bedrock:InvokeModelWithResponseStream"
+      ],
+      "Resource": [
+        "arn:aws:bedrock:us-east-1:099713751195:inference-profile/us.amazon.nova-micro-v1:0",
+        "arn:aws:bedrock:us-east-1::foundation-model/amazon.nova-micro-v1:0",
+        "arn:aws:bedrock:us-west-2::foundation-model/cohere.rerank-v3-5:0"
+      ]
+    },
+    {
+      "Sid": "OSISIngestUBI",
+      "Effect": "Allow",
+      "Action": [
+        "osis:Ingest"
+      ],
+      "Resource": [
+        "arn:aws:osis:us-east-1:099713751195:pipeline/ubi-events-pipeline",
+        "arn:aws:osis:us-east-1:099713751195:pipeline/ubi-events-pipeline/*",
+        "arn:aws:osis:us-east-1:099713751195:pipeline/ubi-queries-pipeline",
+        "arn:aws:osis:us-east-1:099713751195:pipeline/ubi-queries-pipeline/*"
+      ]
+    }
+  ]
+}

search_personalization/image_utils.py

@@ -0,0 +1,228 @@
+"""
+Image mapping utilities for product catalog.
+
+This module provides functions to map product image filenames to their
+actual file paths in the product_samples directory structure, with
+fallback handling for missing images and caching for performance.
+"""
+import os
+from pathlib import Path
+from typing import Dict, Optional
+from functools import lru_cache
+
+
+# Base path for product images (relative to project root)
+PRODUCT_SAMPLES_BASE = "search_personalization/product_samples"
+PLACEHOLDER_IMAGE = "search_personalization/product_samples/product_image_coming_soon.png"
+THUMBNAIL_DIR = Path("search_personalization/product_samples/.thumbnails")
+
+# In-memory cache for image path existence checks
+_image_cache: Dict[str, str] = {}
+
+
+def get_thumbnail_path(product: Dict) -> str:
+    """
+    Get a small thumbnail for fast UI rendering (~200px wide, ~70% quality).
+    Generates on first access, then serves from disk cache.
+
+    Args:
+        product: Product dictionary containing 'category' and 'image' fields
+
+    Returns:
+        Path to the thumbnail image, or placeholder if original doesn't exist.
+    """
+    original = get_image_path(product)
+    if original == PLACEHOLDER_IMAGE:
+        return PLACEHOLDER_IMAGE
+
+    rel = Path(original).relative_to(PRODUCT_SAMPLES_BASE)
+    thumb = THUMBNAIL_DIR / rel
+    if thumb.exists():
+        return str(thumb)
+
+    # Generate thumbnail
+    try:
+        from PIL import Image
+        thumb.parent.mkdir(parents=True, exist_ok=True)
+        img = Image.open(original)
+        img.thumbnail((200, 200))
+        img.save(thumb, "JPEG", quality=70)
+        return str(thumb)
+    except Exception:
+        return original
+
+
+def get_image_path(product: Dict, use_cache: bool = True) -> str:
+    """
+    Map product image filename to actual file path in product_samples directory.
+    
+    This function constructs the full path to a product image based on the
+    product's category and image filename. If the image file doesn't exist,
+    it returns the path to a placeholder image.
+    
+    Args:
+        product: Product dictionary containing 'category' and 'image' fields
+        use_cache: Whether to use cached image path results (default: True)
+    
+    Returns:
+        String path to the image file, either the actual product image or
+        the placeholder image if the product image is not found
+    
+    Examples:
+        >>> product = {"category": "accessories", "image": "abc123.jpg"}
+        >>> get_image_path(product)
+        'product_samples/accessories/abc123.jpg'
+        
+        >>> product = {"category": "apparel", "image": "missing.jpg"}
+        >>> get_image_path(product)  # If missing.jpg doesn't exist
+        'product_samples/product_image_coming_soon.png'
+    
+    Requirements: 9.4
+    """
+    # Extract category and image from product
+    category = product.get('category', '')
+    image_filename = product.get('image', '')
+    
+    # If no image filename provided, return placeholder
+    if not image_filename:
+        return PLACEHOLDER_IMAGE
+    
+    # If no category provided, return placeholder
+    if not category:
+        return PLACEHOLDER_IMAGE
+    
+    # Create cache key
+    cache_key = f"{category}/{image_filename}"
+    
+    # Check cache first if enabled
+    if use_cache and cache_key in _image_cache:
+        return _image_cache[cache_key]
+    
+    # Construct the expected image path
+    image_path = os.path.join(PRODUCT_SAMPLES_BASE, category, image_filename)
+    
+    # Check if the image file exists
+    if os.path.exists(image_path) and os.path.isfile(image_path):
+        # Cache the result
+        if use_cache:
+            _image_cache[cache_key] = image_path
+        return image_path
+    
+    # Image doesn't exist, return placeholder
+    if use_cache:
+        _image_cache[cache_key] = PLACEHOLDER_IMAGE
+    return PLACEHOLDER_IMAGE
+
+
+@lru_cache(maxsize=1000)
+def get_image_path_cached(category: str, image_filename: str) -> str:
+    """
+    Cached version of image path lookup using functools.lru_cache.
+    
+    This function provides an alternative caching mechanism using Python's
+    built-in LRU cache decorator. It's more efficient for repeated lookups
+    of the same category/image combinations.
+    
+    Args:
+        category: Product category (e.g., 'accessories', 'apparel', 'footwear')
+        image_filename: Image filename (e.g., 'abc123.jpg')
+    
+    Returns:
+        String path to the image file or placeholder
+    
+    Note:
+        This function uses @lru_cache which caches up to 1000 unique
+        category/filename combinations. The cache persists for the
+        lifetime of the application.
+    """
+    # If no image filename provided, return placeholder
+    if not image_filename:
+        return PLACEHOLDER_IMAGE
+    
+    # If no category provided, return placeholder
+    if not category:
+        return PLACEHOLDER_IMAGE
+    
+    # Construct the expected image path
+    image_path = os.path.join(PRODUCT_SAMPLES_BASE, category, image_filename)
+    
+    # Check if the image file exists
+    if os.path.exists(image_path) and os.path.isfile(image_path):
+        return image_path
+    
+    # Image doesn't exist, return placeholder
+    return PLACEHOLDER_IMAGE
+
+
+def clear_image_cache() -> None:
+    """
+    Clear the in-memory image path cache.
+    
+    This function clears both the manual cache dictionary and the
+    LRU cache used by get_image_path_cached. Useful for testing
+    or when the file system changes.
+    """
+    global _image_cache
+    _image_cache.clear()
+    get_image_path_cached.cache_clear()
+
+
+def get_cache_info() -> Dict[str, any]:
+    """
+    Get information about the current cache state.
+    
+    Returns:
+        Dictionary containing cache statistics:
+        - manual_cache_size: Number of entries in the manual cache
+        - lru_cache_info: Statistics from the LRU cache
+    """
+    return {
+        'manual_cache_size': len(_image_cache),
+        'lru_cache_info': get_image_path_cached.cache_info()._asdict()
+    }
+
+
+def preload_image_cache(products: list) -> int:
+    """
+    Preload the image cache with paths for a list of products.
+    
+    This function can be used to warm up the cache when loading
+    products from OpenSearch, improving performance for subsequent
+    image path lookups.
+    
+    Args:
+        products: List of product dictionaries
+    
+    Returns:
+        Number of image paths successfully cached
+    
+    Example:
+        >>> products = [
+        ...     {"category": "accessories", "image": "img1.jpg"},
+        ...     {"category": "apparel", "image": "img2.jpg"}
+        ... ]
+        >>> count = preload_image_cache(products)
+        >>> print(f"Cached {count} image paths")
+    """
+    cached_count = 0
+    
+    for product in products:
+        category = product.get('category', '')
+        image_filename = product.get('image', '')
+        
+        if category and image_filename:
+            # Use the cached version to populate the LRU cache
+            get_image_path_cached(category, image_filename)
+            cached_count += 1
+    
+    return cached_count
+
+
+def validate_placeholder_exists() -> bool:
+    """
+    Check if the placeholder image file exists.
+    
+    Returns:
+        True if placeholder image exists, False otherwise
+    """
+    return os.path.exists(PLACEHOLDER_IMAGE) and os.path.isfile(PLACEHOLDER_IMAGE)