search_personalization/ubi_agent.py

"""
UBI Analysis Agent — A Strands agent for analyzing User Behavior Insights data
from OpenSearch ubi_events and ubi_queries indices to improve website UX.

Usage:
    python -m src.ubi_agent
"""

import logging
from strands import Agent
from strands.models import BedrockModel
from search_personalization.ubi_tools import (
    query_ubi_events,
    query_ubi_queries,
    aggregate_ubi_events,
    aggregate_ubi_queries,
    get_ubi_index_mappings,
    compute_user_preferences,
)

SYSTEM_PROMPT = """You are a UX Analytics Agent for an e-commerce product catalog.

You have access to two OpenSearch indices containing User Behavior Insights (UBI) data:

1. **ubi_events** — Tracks user interactions including:
   - search, add_to_cart, view_product_details, hover, checkout, filter_applied, user_feedback
   - Each event has: action_name, query_id, client_id, session_id, user_id, timestamp,
     message_type, message, and event_attributes (which may contain object.object_id, position.ordinal)

2. **ubi_queries** — Tracks search queries including:
   - user_query (the search text), query_id, client_id, user_id, timestamp,
     query_response_hit_ids (ordered list of product IDs returned)

Your job is to help the user analyze this data to improve the website's user experience. You can:

- Identify popular and underperforming search terms
- Analyze click-through rates (queries that led to product views vs. those that didn't)
- Find products with high view counts but low add-to-cart rates (conversion issues)
- Detect user drop-off patterns in the funnel (search → view → cart → checkout)
- Analyze user session journeys to find friction points
- Compare behavior across user personas (user1=Sarah, user2=Alex, anonymous)
- Identify zero-result or low-engagement searches that need better product matching
- Spot hover patterns that indicate interest without conversion

**User Preference Learning:**
You can also compute and store per-user preference weights using compute_user_preferences.
This analyzes a user's filter history to determine their preferred price range, favorite categories,
and how price-sensitive they are. The results are stored in the user_preferences index and used by
the application to pre-apply filters for returning users.

**Workflow:**
1. When the user asks a question, first determine what data you need
2. Use get_ubi_index_mappings to discover available fields if unsure about the schema
3. Use aggregation tools for high-level patterns and distributions
4. Use query tools to drill into specific events or queries
5. Synthesize findings into actionable UX improvement recommendations

Always ground your analysis in the actual data. Provide specific numbers and examples.
When suggesting improvements, be concrete — e.g., "The search term 'wireless headphones' returned
0 results 15 times in the last 24h — consider adding synonyms or expanding the product catalog."
"""


def create_ubi_agent() -> Agent:
    """Create and return the UBI analysis agent."""
    model = BedrockModel(
        model_id="us.anthropic.claude-opus-4-6-v1",
        region_name="us-east-1",
    )

    agent = Agent(
        model=model,
        tools=[
            query_ubi_events,
            query_ubi_queries,
            aggregate_ubi_events,
            aggregate_ubi_queries,
            get_ubi_index_mappings,
            compute_user_preferences,
        ],
        system_prompt=SYSTEM_PROMPT,
    )
    return agent


def main():
    """Interactive CLI loop for the UBI analysis agent."""
    logging.basicConfig(
        format="%(levelname)s | %(name)s | %(message)s",
        handlers=[logging.StreamHandler()],
    )

    agent = create_ubi_agent()

    print("\n🔍 UBI Analysis Agent")
    print("Analyze user behavior data to improve website UX.")
    print("Type 'quit' or 'exit' to stop.\n")

    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nGoodbye!")
            break

        if not user_input:
            continue
        if user_input.lower() in ("quit", "exit"):
            print("Goodbye!")
            break

        agent(user_input)
        print()  # blank line after response


if __name__ == "__main__":
    main()