chore(vectorstore): create a vectorstor on amazon products dataset.

feat(retriever): Implement a retriever with hybrid search and reranker

.gitignore

@@ -0,0 +1,15 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+.env
+
+# Jupyter notebooks
+*.ipynb
+.ipynb_checkpoints

.python-version

@@ -0,0 +1 @@
+3.12

build_vectorstore.py

@@ -0,0 +1,209 @@
+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
+
+_ = 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="text-embedding-3-large")
+        self.dimensions = 3072
+
+        # 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="amazon-products-catalog",
+        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()

main.py

@@ -0,0 +1,9 @@
+from agno.agent import Agent
+from agno.models.openai import OpenAIChat
+
+from dotenv import load_dotenv
+load_dotenv()
+
+agent = Agent(model=OpenAIChat(id="gpt-4o-mini"))
+
+agent.cli_app("Tell me a 5 second short story about a robot", stream=True)

pyproject.toml

@@ -0,0 +1,19 @@
+[project]
+name = "shopping-ai-agent"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "agno>=2.0.8",
+    "cohere>=5.18.0",
+    "google-genai>=1.38.0",
+    "ipykernel>=6.30.1",
+    "langchain-community>=0.3.29",
+    "langchain-openai>=0.3.33",
+    "pandas>=2.3.2",
+    "pinecone>=7.3.0",
+    "pinecone-client[grpc]>=6.0.0",
+    "pinecone-text>=0.11.0",
+    "python-dotenv>=1.1.1",
+]

retriever.py

@@ -0,0 +1,286 @@
+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.grpc import PineconeGRPC as Pinecone
+from pinecone_text.sparse import BM25Encoder
+from openai import OpenAI
+import cohere
+
+_ = load_dotenv()
+
+# Pydantic Models
+class FilterModel(BaseModel):
+    """Search filters with validation"""
+    min_price: Optional[float] = Field(None, ge=0)
+    max_price: Optional[float] = Field(None, ge=0)
+    categories: Optional[List[str]] = None
+    min_rating: Optional[float] = Field(None, ge=0, le=5)
+    min_reviews: Optional[int] = Field(None, 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 = "text-embedding-3-large",
+        embedding_dimensions: int = 3072,
+        rerank_model: str = "rerank-v3.5",
+        bm25_encoder_path: str = "bm25_encoder.json",
+        environment: str = "us-east-1"
+    ):
+        """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(environment)
+        
+        # Initialize encoders
+        self._initialize_encoders()
+
+    def _initialize_clients(self, environment: str) -> 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=metadata['discount_price_usd'],
+                    original_price=metadata['actual_price_usd'],
+                    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 = 10,
+        alpha: float = 0.5,  # 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="amazon-products-catalog"
+    )
+
+    # 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()