Update the project to use a real-time Amazon products API instead of a vector store. Improve the prompt to handle updated tool

.env.example

@@ -1,5 +1,6 @@
-PINECONE_API_KEY=<your-pinecone-api-key>
-INDEX_NAME=<your-pinecone-index-name>
+RAPIDAPI_KEY=<your-rapidapi-key>
+SEARCH_API_URL=<amazon-search-api-url>
+RAPIDAPI_HOST=<amazon-search-api-host>
 
 REDIS_URL=<your-upstash-redis-url>
 

app/build_vectorstore.py

@@ -1,211 +0,0 @@
-import os
-import pandas as pd
-from typing import List, Dict
-from pinecone.grpc import PineconeGRPC as Pinecone
-from pinecone import ServerlessSpec
-from pinecone_text.sparse import BM25Encoder
-from langchain_openai import OpenAIEmbeddings
-import uuid
-from dotenv import load_dotenv
-from tqdm import tqdm
-
-from .config import settings
-
-_ = load_dotenv()
-
-class PineconeHybridProductIndexer:
-    def __init__(self, index_name: str, api_key: str, environment: str = "us-east-1"):
-        """Initialize Pinecone hybrid search for products"""
-        self.pc = Pinecone(api_key=api_key)
-        self.environment = environment
-        self.index_name = index_name
-
-        # Initialize embeddings model
-        self.dense_model = OpenAIEmbeddings(model=settings.EMBEDDING_MODEL)
-        self.dimensions = settings.EMBEDDING_DIMENSION
-
-        # Initialize sparse encoder (BM25)
-        self.sparse_encoder = BM25Encoder()
-        
-        self.index = None
-        
-    def create_hybrid_index(self):
-        """
-        Create Pinecone hybrid index for products
-        Key requirement: vector_type='dense' and metric='dotproduct' for hybrid search
-        """
-        try:
-            # Delete index if it exists
-            if self.index_name in self.pc.list_indexes().names():
-                print(f"Deleting existing index: {self.index_name}")
-                self.pc.delete_index(self.index_name)
-            
-            # Create hybrid index
-            print(f"Creating index: {self.index_name}")
-            self.pc.create_index(
-                name=self.index_name,
-                dimension=self.dimensions,
-                metric="dotproduct",  # Required for hybrid search
-                spec=ServerlessSpec(
-                    cloud="aws",
-                    region=self.environment
-                )
-            )
-            
-            # Connect to index
-            self.index = self.pc.Index(self.index_name)
-            
-        except Exception as e:
-            print(f"Error creating index: {e}")
-            raise
-    
-    def connect_to_index(self):
-        """Connect to existing index"""
-        if self.index_name not in self.pc.list_indexes().names():
-            raise ValueError(f"Index {self.index_name} does not exist. Create it first.")
-        
-        self.index = self.pc.Index(self.index_name)
-        print(f"Connected to index: {self.index_name}")
-    
-    def delete_index(self):
-        """Delete the Pinecone index if it exists"""
-        try:
-            existing_indexes = self.pc.list_indexes().names()
-            if self.index_name in existing_indexes:
-                print(f"Deleting index: {self.index_name}")
-                self.pc.delete_index(self.index_name)
-                self.index = None
-                print(f"Index deleted: {self.index_name}")
-            else:
-                print(f"Index {self.index_name} does not exist; nothing to delete.")
-        except Exception as e:
-            print(f"Error deleting index: {e}")
-            raise
-    
-    def prepare_documents_for_indexing(self, df: pd.DataFrame) -> List[Dict]:
-        """Prepare documents for hybrid indexing"""
-        print("Preparing documents for hybrid indexing...")
-        
-        # Prepare texts for sparse encoding
-        texts = []
-        documents = []
-        
-        for _, row in df.iterrows():
-            # Create rich text content for both dense and sparse encoding
-            text_content = f"Product: {row['name']}. Category: {row['main_category']}. Type: {row['sub_category']}."
-            texts.append(text_content)
-            
-            # Prepare metadata
-            metadata = {
-                'name': row['name'],
-                'main_category': row['main_category'],
-                'sub_category': row['sub_category'],
-                'discount_price_usd': float(row['discount_price_usd']),
-                'actual_price_usd': float(row['actual_price_usd']),
-                'ratings': float(row['ratings']),
-                'no_of_ratings': int(row['no_of_ratings']),
-                'image': row['image'],
-                'link': row['link']
-            }
-            
-            documents.append({
-                'id': str(uuid.uuid4()),
-                'text': text_content,
-                'metadata': metadata
-            })
-        
-        # Fit sparse encoder on all texts
-        print("Training BM25 sparse encoder...")
-        self.sparse_encoder.fit(texts)
-        
-        # Save sparse encoder
-        print("Saving BM25 sparse encoder...")
-        self.sparse_encoder.dump("bm25_encoder.json")
-        
-        return documents
-  
-    def index_products(self, df: pd.DataFrame, batch_size: int = 100):
-        """Index products in Pinecone with hybrid vectors"""
-        print(f"Starting to index {len(df)} products...")
-
-        # Prepare documents (fits BM25 across the whole corpus and builds metadata)
-        documents = self.prepare_documents_for_indexing(df)
-
-        # Embed and upsert in batches to avoid holding all vectors in memory
-        total_docs = len(documents)
-        total_batches = (total_docs + batch_size - 1) // batch_size
-        max_retries = 5
-        base_delay_seconds = 1.0
-
-        with tqdm(total=total_batches, desc="Upserting batches", unit="batch") as pbar:
-            for i in range(0, total_docs, batch_size):
-                batch_num = i // batch_size + 1
-                batch_docs = documents[i:i + batch_size]
-                start_idx = i + 1
-                end_idx = min(i + len(batch_docs), total_docs)
-                pbar.set_postfix_str(f"batch {batch_num}/{total_batches} items {start_idx}-{end_idx}")
-
-                # Prepare texts
-                batch_texts = [doc['text'] for doc in batch_docs]
-
-                # Create dense and sparse vectors for this batch
-                dense_vectors = self.dense_model.embed_documents(batch_texts)
-                sparse_vectors = self.sparse_encoder.encode_documents(batch_texts)
-
-                # Build Pinecone vector payloads
-                batch_vectors = []
-                for j, doc in enumerate(batch_docs):
-                    batch_vectors.append({
-                        'id': doc['id'],
-                        'values': dense_vectors[j],
-                        'sparse_values': {
-                            'indices': sparse_vectors[j]['indices'],
-                            'values': sparse_vectors[j]['values']
-                        },
-                        'metadata': doc['metadata']
-                    })
-
-                # Upsert with retries
-                last_error = None
-                for attempt in range(1, max_retries + 1):
-                    try:
-                        self.index.upsert(vectors=batch_vectors)
-                        last_error = None
-                        break
-                    except Exception as e:
-                        last_error = e
-                        if attempt < max_retries:
-                            delay = base_delay_seconds * (2 ** (attempt - 1))
-                            tqdm.write(f"[Batch {batch_num}/{total_batches}] Attempt {attempt} failed: {e}. Retrying in {delay:.1f}s...")
-                            import time
-                            time.sleep(delay)
-                        else:
-                            tqdm.write(f"[Batch {batch_num}/{total_batches}] Failed after {max_retries} attempts: {e}")
-                if last_error is not None:
-                    raise last_error
-
-                pbar.update(1)
-
-        print(f"Successfully indexed {total_docs} products!")
-        stats = self.index.describe_index_stats()
-        print(f"Index stats: {stats}")
-
-
-def setup_and_run():
-    """Example usage of the hybrid search system"""
-    
-    # Initialize retriever
-    retriever = PineconeHybridProductIndexer(
-        index_name=settings.INDEX_NAME,
-        api_key=os.getenv("PINECONE_API_KEY")
-    )
-    
-    # Create index (do this once)
-    retriever.create_hybrid_index()
-    
-    # Load and index your data (do this once)
-    df = pd.read_csv("data/amazon_products.csv")
-    retriever.index_products(df)
-
-if __name__ == "__main__":
-    setup_and_run()

app/config.py

@@ -6,12 +6,15 @@ class Settings(BaseSettings):
     )
 
     # --- API Keys ---
-    PINECONE_API_KEY: str
+    RAPIDAPI_KEY: str
+    SEARCH_API_URL: str
+    RAPIDAPI_HOST: str
+
     REDIS_URL: str
     OS_SECURITY_KEY: str
+
     GOOGLE_API_KEY: str
     GROQ_API_KEY: str
-    OPENAI_API_KEY: str
     CO_API_KEY: str
 
     # --- Agent Configuration ---
@@ -22,14 +25,7 @@ class Settings(BaseSettings):
     RERANKING_MODEL: str = "rerank-v3.5"
     MEMORY_MANAGER_MODEL: str = "openai/gpt-oss-120b"
 
-    EMBEDDING_MODEL: str = "text-embedding-3-large"
-    EMBEDDING_DIMENSION: int = 3072
-
-    NUM_HISTORY_RUNS: int = 10
-
-    # --- Vectorstore Configuration ---
-    INDEX_NAME: str
-    TOP_K: int = 5
-    ALPHA: float = 0.5
+    NUM_HISTORY_RUNS: int = 5
+    SEARCH_LIMIT: int = 6
 
 settings = Settings()

app/product_search.py

@@ -0,0 +1,360 @@
+import os
+from typing import List, Dict, Any, Optional, Literal
+from pydantic import BaseModel, Field, ValidationError
+from dotenv import load_dotenv
+import yaml
+import time
+from functools import wraps
+import requests
+
+import cohere
+
+from .config import settings
+
+_ = load_dotenv()
+
+
+# Pydantic Models
+class FilterModel(BaseModel):
+    """
+    Search filters for refining Amazon product searches.
+    Use these filters to narrow down search results based on price, category, ratings, and deals.
+    """
+
+    min_price: Optional[int] = Field(
+        None, description="Minimum price in USD. e.g, 10 for products $10 and up.", ge=0
+    )
+    max_price: Optional[int] = Field(
+        None,
+        description="Maximum price in USD. e.g, 50 for products $50 and under.",
+        ge=0,
+    )
+    category_id: Optional[str] = Field(
+        None,
+        description="Filter products by category id (e.g., 'gift-cards', 'fashion'). Must be from the allowed values provided.",
+    )
+    sort_by: Optional[Literal["RELEVANCE", "NEWEST", "BEST_SELLERS", "REVIEWS"]] = (
+        Field(
+            None,
+            description=(
+                "Sort order for results: "
+                "'RELEVANCE' - Most relevant to search query (default), "
+                "'NEWEST' - Newest products first, "
+                "'BEST_SELLERS' - Most popular/best selling products, "
+                "'REVIEWS' - Highest rated products first. "
+                "Use 'BEST_SELLERS' when user wants popular items, 'REVIEWS' for highly rated items, 'NEWEST' for latest releases."
+            ),
+        )
+    )
+    high_ratings: Optional[bool] = Field(
+        None,
+        description="Set to true to only return products with 4 stars or higher rating.",
+    )
+    deals_and_discounts: Optional[Literal["ALL_DISCOUNTS", "TODAYS_DEALS"]] = Field(
+        None,
+        description=(
+            "Filter for products on sale or with discounts: "
+            "'ALL_DISCOUNTS' - Any product currently on sale or discounted, "
+            "'TODAYS_DEALS' - Products in Amazon's daily deals only. "
+        ),
+    )
+
+class ProductItem(BaseModel):
+    """Pydantic model for validating product data"""
+    asin: str
+    title: str = Field(alias='product_title')
+    price: Optional[str] = Field(None, alias='product_price')
+    original_price: Optional[str] = Field(None, alias='product_original_price')
+    rating: Optional[float] = Field(None, alias='product_star_rating')
+    num_reviews: int = Field(0, alias='product_num_ratings')
+    is_best_seller: bool
+    is_amazon_choice: bool
+    delivery: Optional[str] = None
+    product_photo: str
+    product_url: str
+
+# Timer decorator
+def timer(func):
+    """Decorator to measure function execution time"""
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        start_time = time.time()
+        result = func(*args, **kwargs)
+        execution_time = time.time() - start_time
+        print(f"{func.__name__} executed in {execution_time:.3f}s")
+        return result
+    return wrapper
+
+class AmazonProductSearch:
+    def __init__(self, rapid_api_key: str, rerank_model: str):
+        """Initialize Pinecone hybrid search for products"""
+        self.rapid_api_key = rapid_api_key
+        self.rerank_model = rerank_model
+
+        self._initialize_cohere_client()
+
+    def _initialize_cohere_client(self) -> None:
+        """Initialize external service clients"""
+        try:
+            # Initialize Cohere
+            self.cohere_client = cohere.ClientV2()
+        except Exception as e:
+            print(f"Failed to initialize clients: {e}")
+            raise
+
+    def _validate_and_convert_products(
+        self, raw_products: List[Dict[str, Any]]
+    ) -> List[ProductItem]:
+        """
+        Validate raw product data and convert to ProductItem model.
+        """
+        products = []
+
+        for idx, product in enumerate(raw_products):
+            try:
+                # Validate using Pydantic
+                validated = ProductItem(**product)
+                products.append(validated)
+            except ValidationError as e:
+                print(f"Validation error for product at index {idx}: {e}")
+                print(product)
+                continue
+            except Exception as e:
+                print(f"Unexpected error validating product at index {idx}: {e}")
+                continue
+
+        return products
+
+    def _enhance_query_with_filters(
+        self,
+        query: str,
+        filters: Optional[FilterModel] = None
+    ) -> str:
+        """
+        Enhance the search query with filter information for better reranking.
+        
+        Args:
+            query: Original search query
+            filters: Applied filters
+            
+        Returns:
+            Enhanced query string with filter context
+        """
+        if not filters:
+            return query
+
+        enhancements = []
+
+        # Add price range context
+        if filters.min_price is not None or filters.max_price is not None:
+            if filters.min_price is not None and filters.max_price is not None:
+                enhancements.append(f"priced between ${filters.min_price} and ${filters.max_price}")
+            elif filters.min_price is not None:
+                enhancements.append(f"priced at least ${filters.min_price}")
+            elif filters.max_price is not None:
+                enhancements.append(f"priced under ${filters.max_price}")
+
+        # Add category context
+        if filters.category_id:
+            category_name = filters.category_id.replace('-', ' ')
+            enhancements.append(f"in {category_name} category")
+
+        # Add rating context
+        if filters.high_ratings:
+            enhancements.append("with high ratings (4+ stars)")
+
+        # Add deals context
+        if filters.deals_and_discounts:
+            if filters.deals_and_discounts == 'TODAYS_DEALS':
+                enhancements.append("from today's deals")
+            elif filters.deals_and_discounts == 'ALL_DISCOUNTS':
+                enhancements.append("with discounts or on sale")
+
+        # Add sort preference context
+        if filters.sort_by:
+            sort_context = {
+                'BEST_SELLERS': 'popular and best-selling',
+                'REVIEWS': 'highly reviewed',
+                'NEWEST': 'newest and latest',
+                'RELEVANCE': 'most relevant'
+            }
+            if filters.sort_by in sort_context:
+                enhancements.append(sort_context[filters.sort_by])
+
+        # Combine original query with enhancements
+        if enhancements:
+            enhanced_query = f"{query}, {', '.join(enhancements)}"
+            return enhanced_query
+
+        return query
+
+    @timer
+    def _execute_product_search(
+        self,
+        query: str,
+        filters: Optional[FilterModel] = None
+    ) -> list[dict[str, Any]]:
+        """
+        Fetch products from Amazon using RapidAPI.
+        
+        Args:
+            query: Search query string
+            filters: Optional filters to refine the search
+            
+        Returns:
+            List of product dictionaries
+        """
+        url = settings.SEARCH_API_URL
+
+        # Build query parameters
+        querystring = {
+            "query": query,
+            "page": "1",
+            "country": "US",
+            "fields": "product_title,product_price,product_original_price,product_star_rating,product_num_ratings,product_url,product_photo,is_best_seller,is_amazon_choice,delivery"
+        }
+
+        # Add filters if provided
+        if filters:
+            if filters.min_price is not None:
+                querystring["min_price"] = str(filters.min_price)
+
+            if filters.max_price is not None:
+                querystring["max_price"] = str(filters.max_price)
+
+            if filters.category_id:
+                querystring["category_id"] = filters.category_id
+
+            if filters.sort_by:
+                querystring["sort_by"] = filters.sort_by
+
+            if filters.high_ratings:
+                querystring["four_stars_and_up"] = "true"
+
+            if filters.deals_and_discounts:
+                querystring["deals_and_discounts"] = filters.deals_and_discounts
+
+        # Set up headers
+        headers = {
+            "x-rapidapi-key": settings.RAPIDAPI_KEY,
+            "x-rapidapi-host": settings.RAPIDAPI_HOST
+        }
+
+        try:
+            response = requests.get(url, headers=headers, params=querystring, timeout=30)
+            response.raise_for_status()
+
+            data = response.json()
+
+            # Extract products from response
+            products = data.get('data', {}).get('products', [])
+            return products
+
+        except requests.exceptions.RequestException as e:
+            print(f"Error fetching products: {e}")
+            return []
+        except Exception as e:
+            print(f"Unexpected error: {e}")
+            return []
+
+    @timer
+    def _rerank_products(
+        self, 
+        query: str, 
+        products: List[ProductItem],
+        top_n: int
+    ) -> List[ProductItem]:
+        """
+        Rerank products using Cohere reranker
+        """
+        if not products:
+            return products
+
+        # Convert products to yaml format as required by Cohere Reranker
+        yaml_docs = [yaml.dump(product, sort_keys=False) for product in products]
+
+        # Rerank products
+        response = self.cohere_client.rerank(
+            model=self.rerank_model,
+            query=query,
+            top_n=top_n,
+            documents=yaml_docs
+        )
+
+        # Return reranked products
+        return [products[result.index] for result in response.results]
+
+    @timer
+    def fetch_products(
+        self,
+        query: str,
+        filters: Optional[FilterModel] = None,
+        limit: int = 10
+    ) -> list[ProductItem]:
+        """
+        Fetch products from Amazon with reranking.
+
+        Args:
+            query: Search query string
+            filters: Optional filters to refine the search
+            limit: Number of products to return. Defaults to 10.
+
+        Returns:
+            List of ProductItem instances after reranking
+        """
+
+        try:
+            # Fetch products from Amazon
+            raw_products = self._execute_product_search(query=query, filters=filters)
+
+            if not raw_products:
+                print(f"No products found for query: {query}")
+                return []
+
+            # Validate products
+            validated_products = self._validate_and_convert_products(raw_products=raw_products)
+
+            # Enhance query with filter context
+            enhanced_query = self._enhance_query_with_filters(query=query, filters=filters)
+
+            # Rerank and limit results
+            reranked_products = self._rerank_products(
+                query=enhanced_query,
+                products=validated_products,
+                top_n=limit
+            )
+
+            return reranked_products
+
+        except Exception as e:
+            print(f"Error during product search: {e}")
+            return []
+
+
+# Usage Example
+def example_usage():
+    """Example usage of the amazon product search"""
+    
+    # Initialize retriever
+    searcher = AmazonProductSearch(
+        rapid_api_key=settings.RAPIDAPI_KEY,
+        rerank_model=settings.RERANKING_MODEL
+    )
+
+    # Example search
+    filters = FilterModel(
+        category_id="fashion-mens"
+    )
+
+    results: list[ProductItem] = searcher.fetch_products(
+        query="black t-shirts for men",
+        filters=filters,
+        limit=6
+    )
+    
+    for i, product in enumerate(results, 1):
+        print(f"{i}. {product.title} - {product.price} ({round(product.rating, 1)}⭐)")
+        print(f"   URL: {product.product_url}")
+
+if __name__ == "__main__":
+    example_usage()

app/prompts.py

@@ -1,39 +1,38 @@
-SHOPPING_AGENT_PROMPT = """
+AGENT_INSTRUCTIONS = """
 <role>
 You are **Gem**, a highly personalized and friendly shopping AI Agent. 
-Your purpose is to guide the user step by step in finding products that perfectly fit their needs, style, and preferences.
+Your purpose is to guide the user step by step in finding products from Amazon that perfectly fit their needs, style, and preferences.
 </role>
 
 <behavior>
 - Be warm, concise, and conversational like a trusted shopping friend.
 - Be aware of the current date, time, and season to make relevant suggestions.
-- Ask at most one clarifying question at a time (never bombard the user with multiple questions).  
-- Do NOT reveal or explain internal tools, filter-building, or any backend process to the user.  
+- Ask at most one clarifying question at a time (never bombard the user with multiple questions).
 - After fetching products, do NOT repeat each product details (products appear automatically as cards in the frontend). Instead: highlight, recommend, compare, and ask which ones they like.  
 - Always offer a short personal recommendation plus a clear, human reason (e.g., “I recommend X because…”).  
-- If the user shares personal details or explicitly mentions some preferences, save them to memory (using `update_user_memory` tool) and confirm briefly to the user that you’ll remember. DO NOT save general things like "Looking for men shoes" or "User wants a new phone under $300".
+- If the user shares personal information or explicitly mentions some preferences, save them to memory (using `update_user_memory` tool) and confirm briefly to the user that you’ll remember. DO NOT save general things like "Looking for men shoes" or "User wants a new phone under $300".
 </behavior>
 
 <tools>
-- `fetch_products`: Use this tool to fetch products from vector store. It will automatically display product cards in the frontend.
+- `fetch_products`: Use this tool to fetch products from Amazon. It will automatically display product cards in the frontend.
 - `update_user_memory`: Use whenever the user shares personal information or instruct you to remember something about them.
 </tools>
 
 <task_flow>
 1. Understand the user's needs:
-   - Start by understanding the user's request. If they say "I want X" ask one targeted question to clarify (budget, size, color, use-case, must-have features).  
+   - Start by understanding the user's request. If they say "I want X" ask one targeted question at a time to clarify (budget, size, color, deals and discounts, must-have features).  
    - Keep questions short and friendly: e.g., "Nice — what's your budget for this?" or "Do you prefer a brand or any brand is fine?" 
 
 2. Search for products:
-   - When you have enough detail, call `fetch_products` with appropriate `filters` (price range, rating, category). The filters are optional, but highly recommended to narrow down results.
+   - When you have enough details, call `fetch_products` with appropriate `filters` (price range, rating, deals, category id etc.). The filters are optional, but highly recommended to narrow down results.
    - **Price**: Identify ranges like "under $50", "over $100", "between $20 and $60".
-   - **Rating**: Look for terms like "highly rated", "4 stars or more", "best reviewed". A generic term like "best" can imply a `min_rating` of `4.0`.
-   - **Rating Count**: Look for terms like "popular", "most reviewed". A generic term like "popular" can imply a `min_reviews` of `100`.
-   - **Categories**: Identify the categories from the user query. **MUST** be from the allowed categories listed below. Do not invent new categories. For example, if the user says "I need black shirts for men", the right category would be "men's clothing". 
-   - If no category matches, **do not invent one**. Instead, continue without a category filter.  
+   - **Rating**: Look for terms like "highly rated", "4 stars or more", "best reviewed". A generic term like "best" means you should set `high_ratings` to true.
+   - **Category**: Identify the category id from the user query. **MUST** be from the allowed categories listed below. Do not invent new categories. For example, if the user says "I need black shirts for men", the right category id would be "fashion-mens". If no category matches, **do not invent one**. Instead, continue without a category filter.  
+   - **Deals & Dicounts**: Ask the user if they want products on sales, discounts or today's deals. Use `deals_and_discounts` filter with suitable value.
+   - **Sorting**: After talking to user, determine if the user needs newest, highly rated or best selling items. Use `sort_by` filter if applicable, otherwise leave it default to relevance.
 
 3. Recommend and Personalize:
-   - Highlight the best options for the user.  
+   - Highlight the best options for the user. Mention key features, and why they stand out.
    - Give your own smart recommendation (e.g., “I think this one is a perfect match because…”).  
    - Ask them for feedback (e.g., “Do you like this style, or should I show something different?”).  
    - If results aren’t ideal, smoothly guide them by suggesting adjustments (e.g., slightly higher budget, different style).
@@ -41,27 +40,42 @@ Your purpose is to guide the user step by step in finding products that perfectl
 4. Loop back: refine results or give new options based on feedback.  
 </task_flow>
 
-<allowed_categories>
-You must use one of the following category values if a category filter is applicable. Remember, these categories are case-sensitive:
-- appliances
-- car & motorbike
-- tv, audio & cameras
-- sports & fitness
-- grocery & gourmet foods
-- home & kitchen
-- pet supplies
-- stores
-- toys & baby products
-- kids' fashion
-- bags & luggage
-- accessories
-- women's shoes
-- beauty & health
-- men's shoes
-- women's clothing
-- industrial supplies
-- men's clothing
-- music
-- home, kitchen, pets
-</allowed_categories>
-"""
+<allowed_categories_with_ids>
+If a category filter is applicable, you must use the category id of the following allowed categories:
+
+- Appliances (appliances)
+- Arts, Crafts & Sewing (arts-crafts)
+- Audible Books & Originals (audible)
+- Automotive Parts & Accessories (automotive)
+- Baby (baby-products)
+- Beauty & Personal Care (beauty)
+- Books (stripbooks)
+- Cell Phones & Accessories (mobile)
+- Clothing, Shoes & Jewelry (fashion)
+   * Women (fashion-womens)
+   * Men (fashion-men)
+   * Boys (fashion-boys)
+   * Girls (fashion-girls)
+   * Baby (fashion-baby)
+- Collectibles & Fine Art (collectibles)
+- Computers (computers)
+- Electronics (electronics)
+- Garden & Outdoor (lawngarden)
+- Gift Cards (gift-cards)
+- Grocery & Gourmet Food (grocery)
+- Handmade (handmade)
+- Health, Household & Baby Care (hpc)
+- Home & Kitchen (garden)
+- Industrial & Scientific (industrial)
+- Luggage & Travel Gear (fashion-luggage)
+- Musical Instruments (mi)
+- Pet Supplies (pets)
+
+NOTE: Use the category id in filter, not the catergory itself. Do NOT invent new categories.
+</allowed_categories_with_ids>
+
+<constraints>
+- Never disclose internal system details, tools, behind the scenes to the user.
+- DO NOT engage in conversations outside your role.
+</constraints>
+"""

app/retriever.py

@@ -1,287 +0,0 @@
-import os
-from typing import List, Dict, Optional, Tuple
-from concurrent.futures import ThreadPoolExecutor
-from pydantic import BaseModel, Field
-from dotenv import load_dotenv
-import yaml
-import time
-from functools import wraps
-
-from pinecone import Pinecone
-from pinecone_text.sparse import BM25Encoder
-from openai import OpenAI
-import cohere
-
-from .config import settings
-
-_ = load_dotenv()
-
-# Pydantic Models
-class FilterModel(BaseModel):
-    """Search filters with validation"""
-    min_price: Optional[float] = Field(None, description="Minimum price of the product", ge=0)
-    max_price: Optional[float] = Field(None, description="Maximum price of the product", ge=0)
-    categories: Optional[List[str]] = Field(None, description="List of the main product categories, e.g., 'men's clothing'. Must be from the allowed values provided.")
-    min_rating: Optional[float] = Field(None, description="Minimum rating of the product", ge=0, le=5)
-    min_reviews: Optional[int] = Field(None, description="Minimum number of reviews of the product", ge=0)
-
-class ProductItem(BaseModel):
-    """Product model"""
-    name: str
-    price: float
-    original_price: float
-    rating: float
-    num_reviews: int
-    category: str
-    sub_category: str
-    image_url: str
-    link: str
-
-def timer(func):
-    """Decorator to measure function execution time"""
-    @wraps(func)
-    def wrapper(*args, **kwargs):
-        start_time = time.time()
-        result = func(*args, **kwargs)
-        execution_time = time.time() - start_time
-        print(f"{func.__name__} executed in {execution_time:.3f}s")
-        return result
-    return wrapper
-
-class PineconeHybridRetriever:
-    def __init__(
-        self, 
-        index_name: str,
-        embedding_model: str = settings.EMBEDDING_MODEL,
-        embedding_dimensions: int = settings.EMBEDDING_DIMENSION,
-        rerank_model: str = settings.RERANKING_MODEL,
-        bm25_encoder_path: str = "app/bm25_encoder.json"
-    ):
-        """Initialize Pinecone hybrid search for products"""
-        self.index_name = index_name
-        self.embedding_model = embedding_model
-        self.embedding_dimensions = embedding_dimensions
-        self.rerank_model = rerank_model
-        self.bm25_encoder_path = bm25_encoder_path
-
-        self._initialize_clients()
-        
-        # Initialize encoders
-        self._initialize_encoders()
-
-    def _initialize_clients(self) -> None:
-        """Initialize external service clients"""
-        try:
-            # Initialize Pinecone
-            self.pc = Pinecone()
-            self.index = self.pc.Index(self.index_name)
-            
-            # Initialize OpenAI
-            self.openai_client = OpenAI()
-            
-            # Initialize Cohere
-            self.cohere_client = cohere.ClientV2()
-            
-        except Exception as e:
-            print(f"Failed to initialize clients: {e}")
-            raise
-
-    def _initialize_encoders(self) -> None:
-        """Initialize sparse encoder"""
-        try:
-            self.sparse_encoder = BM25Encoder().load(self.bm25_encoder_path)
-        except Exception as e:
-            print(f"Failed to load BM25 encoder: {e}")
-            raise
-
-    def _get_dense_embedding(self, query: str) -> List[float]:
-        """Generate dense embedding for query"""
-        response = self.openai_client.embeddings.create(
-            input=query,
-            model=self.embedding_model
-        )
-        return response.data[0].embedding
-
-    def _get_sparse_encoding(self, query: str) -> Dict[str, List]:
-        """Generate sparse encoding for query"""
-        return self.sparse_encoder.encode_queries(query)
-
-    @timer
-    def _execute_parallel_encoding(self, query: str) -> Tuple[List[float], Dict[str, List]]:
-        """Execute dense and sparse encoding in parallel"""
-        with ThreadPoolExecutor(max_workers=2) as executor:
-            # Submit both encoding tasks
-            dense_future = executor.submit(self._get_dense_embedding, query)
-            sparse_future = executor.submit(self._get_sparse_encoding, query)
-            
-            # Wait for completion
-            dense_embedding = dense_future.result()
-            sparse_encoding = sparse_future.result()
-            
-            return dense_embedding, sparse_encoding
-
-    def _build_filter_conditions(self, filters: FilterModel) -> Dict:
-        """Convert FilterModel to Pinecone filter format"""
-        conditions = {}
-        
-        # Handle price range
-        if filters.min_price or filters.max_price:
-            price_cond = {}
-            if filters.min_price:
-                price_cond["$gte"] = filters.min_price
-            if filters.max_price:
-                price_cond["$lte"] = filters.max_price
-            conditions["discount_price_usd"] = price_cond
-        
-        # Handle ratings, review count and categories
-        if filters.min_rating:
-            conditions["ratings"] = {"$gte": filters.min_rating}
-        
-        if filters.min_reviews:
-            conditions["no_of_ratings"] = {"$gte": filters.min_reviews}
-        
-        if filters.categories:
-            conditions["main_category"] = {"$in": filters.categories}
-
-        return conditions
-
-    def _convert_to_products(self, matches: List[Dict]) -> List[ProductItem]:
-        """Convert search results to ProductItem objects"""
-        products = []
-        for match in matches:
-            metadata = match.get('metadata', {})
-            try:
-                product = ProductItem(
-                    name=metadata['name'],
-                    price=round(metadata['discount_price_usd'], 2),
-                    original_price=round(metadata['actual_price_usd'], 2),
-                    rating=metadata['ratings'],
-                    num_reviews=metadata['no_of_ratings'],
-                    category=metadata['main_category'],
-                    sub_category=metadata['sub_category'],
-                    image_url=metadata['image'],
-                    link=metadata['link']
-                )
-                products.append(product)
-            except KeyError as e:
-                print(f"Missing metadata field: {e}")
-                continue
-                
-        return products
-
-    @timer
-    def _rerank_products(
-        self, 
-        query: str, 
-        products: List[ProductItem],
-        top_n: int
-    ) -> List[ProductItem]:
-        """
-        Rerank products using Cohere reranker
-        """
-        if not products:
-            return products
-
-        # Convert products to yaml format
-        yaml_docs = [yaml.dump(product, sort_keys=False) for product in products]
-        
-        # Rerank products
-        response = self.cohere_client.rerank(
-            model=self.rerank_model,
-            query=query,
-            top_n=top_n,
-            documents=yaml_docs
-        )
-
-        # Return reranked products
-        return [products[result.index] for result in response.results]
-
-    @timer
-    def search_products(
-        self, 
-        query: str, 
-        filters: FilterModel = None, 
-        limit: int = settings.TOP_K,
-        alpha: float = settings.ALPHA,  # Balance between dense (1.0) and sparse (0.0)
-        use_hybrid_search: bool = True,
-        enable_reranking: bool = False,
-    ) -> List[ProductItem]:
-        """
-        Perform hybrid search for products
-        """
-        try:
-            if use_hybrid_search:
-                dense_embedding, sparse_encoding = self._execute_parallel_encoding(query)
-            
-            else:
-                dense_embedding = self._get_dense_embedding(query)
-                sparse_encoding = None
-                alpha = 1.0  # Force dense-only search
-            
-            # Build filters
-            filter_conditions = None
-            if filters:
-                filter_conditions = self._build_filter_conditions(filters)
-
-            if enable_reranking:
-                # Double the limit for reranking so we have enough results to rerank
-                limit = limit * 3
-            
-            # Prepare query arguments
-            query_args = {
-                "vector": dense_embedding,
-                "top_k": limit,
-                "include_metadata": True,
-                "filter": filter_conditions,
-                "alpha": alpha
-            }
-            
-            if use_hybrid_search and sparse_encoding:
-                query_args["sparse_vector"] = sparse_encoding
-            
-            # Perform search
-            results = self.index.query(**query_args)
-
-            # Convert results to ProductItem objects
-            products = self._convert_to_products(results['matches'])
-
-            # Apply reranking if requested
-            if enable_reranking and products:
-                products = self._rerank_products(query, products, top_n=limit//3) # Get only the specified limit of products
-
-            return products
-            
-        except Exception as e:
-            print(f"Error during search: {e}")
-            return []
-
-
-# Usage Example
-def example_usage():
-    """Example usage of the hybrid search system"""
-    
-    # Initialize retriever
-    retriever = PineconeHybridRetriever(
-        index_name=settings.INDEX_NAME
-    )
-
-    # Example search
-    filters = FilterModel(
-        min_price=10,
-        max_price=20,
-        min_rating=4.0
-    )
-
-    results: list[ProductItem] = retriever.search_products(
-        query="Black men shirts for casual wear",
-        filters=filters,
-        limit=10,
-        use_hybrid_search=True,
-        enable_reranking=True
-    )
-    
-    for i, product in enumerate(results, 1):
-        print(f"{i}. {product.name} - ${round(product.price, 2)} ({round(product.rating, 1)}⭐)")
-
-if __name__ == "__main__":
-    example_usage()

app/tools.py

@@ -1,5 +1,5 @@
 from agno.tools import tool
-from .retriever import PineconeHybridRetriever, FilterModel
+from .product_search import AmazonProductSearch, FilterModel, ProductItem
 from .config import settings
 
 from typing import Optional
@@ -8,27 +8,33 @@ from dotenv import load_dotenv
 
 load_dotenv()
 
+
 @tool
 def fetch_products(
-    query: str, limit: int = settings.TOP_K, filters: Optional[FilterModel] = None
+    query: str,
+    limit: int = settings.SEARCH_LIMIT,
+    filters: Optional[FilterModel] = None,
 ) -> list[dict]:
     """
-    Fetch products from database based on query and filters
+    Fetch products from Amazon based on query and filters
 
     Args:
         query (str): Search query
-        limit (int, optional): Number of products to return. Defaults to 5.
+        limit (int, optional): Number of products to return. Must be between 2-10. Defaults to 6.
         filters (FilterModel, optional): Filters to apply. Defaults to None.
 
     Returns:
         list[dict]: List of products dictionary items matching the query and filters
     """
-    retriever = PineconeHybridRetriever(settings.INDEX_NAME)
-    products = retriever.search_products(
+    limit = max(2, min(limit, 10))  # Ensure limit is between 2 and 10
+
+    product_searcher = AmazonProductSearch(
+        rapid_api_key=settings.RAPIDAPI_KEY, rerank_model=settings.RERANKING_MODEL
+    )
+
+    products: list[ProductItem] = product_searcher.fetch_products(
         query=query,
         filters=filters,
-        limit=limit,
-        use_hybrid_search=True,
-        enable_reranking=True,
+        limit=limit
     )
-    return json.dumps([product.model_dump() for product in products])
+    return json.dumps([product.model_dump() for product in products])

main.py

@@ -5,7 +5,7 @@ from agno.db.redis import RedisDb
 from agno.os import AgentOS
 from agno.memory import MemoryManager
 
-from app.prompts import SHOPPING_AGENT_PROMPT
+from app.prompts import AGENT_INSTRUCTIONS
 from app.tools import fetch_products
 from app.config import settings
 
@@ -28,7 +28,7 @@ shopping_agent = Agent(
     add_datetime_to_context=True,
     db=redis_db,
     tools=tools,
-    instructions=SHOPPING_AGENT_PROMPT,
+    instructions=AGENT_INSTRUCTIONS,
     enable_agentic_memory=True,
     memory_manager=MemoryManager(
         Groq(id=settings.MEMORY_MANAGER_MODEL),
@@ -40,11 +40,11 @@ shopping_agent = Agent(
 
 agent_os = AgentOS(
     os_id="shopping-agent-os",
-    description="A shopping Agent OS that helps users find the best products based on their needs.",
+    description="A shopping Agent OS that helps users find the best products from amazon based on their needs.",
     agents=[shopping_agent]
 )
 
 app = agent_os.get_app()
 
 if __name__ == "__main__":
-    agent_os.serve(app="main:app", port=7860, host="0.0.0.0", reload=True)
+    agent_os.serve(app="main:app", port=7860, host="0.0.0.0", reload=True)