Building Product Search with RAG

You've embedded your products. Now let's build a search experience that doesn't just find products - it understands what customers want and helps them find it.

The Complete RAG Pipeline

Step 1: Query Understanding

Before searching, understand what the customer really wants.

from openai import OpenAI
from typing import Dict, List, Optional
import json

client = OpenAI()

def parse_search_query(query: str) -> Dict:
    """Extract intent and filters from natural language query"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": """Extract search intent and filters from the query.
Return JSON with:
- search_text: The core product search (what they're looking for)
- filters: Any explicit constraints
  - max_price: number or null
  - min_price: number or null
  - category: string or null
  - brand: string or null
- preferences: Any implicit preferences mentioned
- urgency: "high" if they need it soon, "normal" otherwise"""
            },
            {
                "role": "user",
                "content": query
            }
        ],
        response_format={"type": "json_object"}
    )

    return json.loads(response.choices[0].message.content)


# Example
query = "I need comfortable running shoes under $150 for someone with flat feet, preferably Nike"
parsed = parse_search_query(query)
print(json.dumps(parsed, indent=2))

{
  "search_text": "comfortable running shoes for flat feet",
  "filters": {
    "max_price": 150,
    "min_price": null,
    "category": "Shoes > Running",
    "brand": "Nike"
  },
  "preferences": ["comfort", "arch support", "flat feet friendly"],
  "urgency": "normal"
}

Step 2: Semantic Retrieval

from pinecone import Pinecone

pc = Pinecone(api_key="your-api-key")
index = pc.Index("ecommerce-products")

def retrieve_products(
    search_text: str,
    filters: Dict = None,
    top_k: int = 20
) -> List[Dict]:
    """Retrieve relevant products using semantic search"""

    # Embed the search query
    query_embedding = client.embeddings.create(
        model="text-embedding-3-small",
        input=search_text
    ).data[0].embedding

    # Build Pinecone filter from parsed filters
    pinecone_filter = {}
    if filters:
        if filters.get("max_price"):
            pinecone_filter["price"] = {"$lte": filters["max_price"]}
        if filters.get("min_price"):
            pinecone_filter.setdefault("price", {})["$gte"] = filters["min_price"]
        if filters.get("brand"):
            pinecone_filter["brand"] = filters["brand"]
        if filters.get("category"):
            pinecone_filter["category"] = {"$eq": filters["category"]}

    # Always filter for in-stock items
    pinecone_filter["in_stock"] = True

    # Search
    results = index.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=True,
        filter=pinecone_filter if pinecone_filter else None
    )

    return [
        {
            "id": match.id,
            "score": match.score,
            **match.metadata
        }
        for match in results.matches
    ]


# Retrieve products for our query
products = retrieve_products(
    search_text=parsed["search_text"],
    filters=parsed["filters"],
    top_k=20
)
print(f"Retrieved {len(products)} products")

Step 3: Re-ranking for Better Relevance

Initial retrieval is fast but not always perfectly ordered. Re-ranking improves results.

from sentence_transformers import CrossEncoder

# Load a cross-encoder model for re-ranking
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank_products(query: str, products: List[Dict], top_k: int = 5) -> List[Dict]:
    """Re-rank products using a cross-encoder for better relevance"""

    if not products:
        return []

    # Create query-product pairs
    pairs = [
        [query, f"{p['name']} - {p.get('description', '')[:200]}"]
        for p in products
    ]

    # Get re-ranking scores
    scores = reranker.predict(pairs)

    # Add scores and sort
    for i, product in enumerate(products):
        product['rerank_score'] = float(scores[i])

    # Sort by re-rank score (higher is better)
    ranked = sorted(products, key=lambda x: x['rerank_score'], reverse=True)

    return ranked[:top_k]


# Re-rank the retrieved products
reranked = rerank_products(
    query="comfortable running shoes for flat feet",
    products=products,
    top_k=5
)

for p in reranked:
    print(f"{p['rerank_score']:.3f} - {p['name']} (${p['price']})")

0.923 - Nike Air Zoom Structure 25 ($129.99)
0.891 - ASICS Gel-Kayano 30 ($159.99)
0.876 - Brooks Adrenaline GTS 23 ($139.99)
0.854 - New Balance Fresh Foam 860v13 ($134.99)
0.821 - Saucony Guide 16 ($139.99)

Step 4: Generate Helpful Response

def generate_product_response(
    query: str,
    products: List[Dict],
    preferences: List[str] = None
) -> str:
    """Generate a helpful response with product recommendations"""

    # Format products for the prompt
    products_context = "\n\n".join([
        f"""Product: {p['name']}
Price: ${p['price']}
Brand: {p.get('brand', 'N/A')}
Rating: {p.get('rating', 'N/A')} stars
Description: {p.get('description', 'No description')[:300]}
Key Features: {', '.join(p.get('features', [])[:5]) if p.get('features') else 'N/A'}"""
        for p in products[:5]
    ])

    preferences_text = f"\nUser preferences: {', '.join(preferences)}" if preferences else ""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """You are a helpful shopping assistant. Based on the customer's query and the available products, provide a personalized recommendation.

Guidelines:
- Be conversational and helpful
- Explain WHY each product matches their needs
- Highlight relevant features
- Mention any trade-offs (price vs features)
- If a product doesn't perfectly match, be honest about it
- Format with clear sections and bullet points"""
            },
            {
                "role": "user",
                "content": f"""Customer query: {query}
{preferences_text}

Available products:
{products_context}

Please recommend the best options and explain your reasoning."""
            }
        ],
        temperature=0.7
    )

    return response.choices[0].message.content


# Generate response
response = generate_product_response(
    query="I need comfortable running shoes under $150 for someone with flat feet",
    products=reranked,
    preferences=parsed.get("preferences", [])
)
print(response)

Based on your needs for comfortable running shoes for flat feet under $150, here are my top recommendations:

**Best Overall: Nike Air Zoom Structure 25 - $129.99**
This is specifically designed for overpronators (common with flat feet). Key features:
- Dual-density foam provides extra arch support
- Cushioned heel absorbs impact
- Breathable mesh upper
- Fits within your budget with room to spare

**Runner-Up: Brooks Adrenaline GTS 23 - $139.99**
Another excellent stability shoe with:
- GuideRails support system for flat feet
- DNA LOFT cushioning for all-day comfort
- Slightly pricier but exceptional durability

**Budget Option: New Balance Fresh Foam 860v13 - $134.99**
Great value with:
- Medial post for stability
- Fresh Foam midsole
- Wide size options available

My recommendation: Start with the Nike Air Zoom Structure 25. It has the best balance of flat-foot support, comfort, and price for your needs.

Complete RAG Search Function

class ProductSearchRAG:
    def __init__(self, pinecone_index, openai_client, reranker=None):
        self.index = pinecone_index
        self.client = openai_client
        self.reranker = reranker

    def search(self, query: str, top_k: int = 5) -> Dict:
        """Complete RAG search pipeline"""

        # Step 1: Parse query
        parsed = self.parse_query(query)

        # Step 2: Retrieve products
        products = self.retrieve(
            search_text=parsed["search_text"],
            filters=parsed.get("filters"),
            top_k=20  # Retrieve more for re-ranking
        )

        # Step 3: Re-rank (if reranker available)
        if self.reranker and products:
            products = self.rerank(parsed["search_text"], products, top_k)
        else:
            products = products[:top_k]

        # Step 4: Generate response
        response = self.generate_response(
            query=query,
            products=products,
            preferences=parsed.get("preferences", [])
        )

        return {
            "query": query,
            "parsed": parsed,
            "products": products,
            "response": response
        }

    def parse_query(self, query: str) -> Dict:
        # ... (implementation from above)
        pass

    def retrieve(self, search_text: str, filters: Dict, top_k: int) -> List[Dict]:
        # ... (implementation from above)
        pass

    def rerank(self, query: str, products: List[Dict], top_k: int) -> List[Dict]:
        # ... (implementation from above)
        pass

    def generate_response(self, query: str, products: List[Dict], preferences: List[str]) -> str:
        # ... (implementation from above)
        pass


# Usage
rag = ProductSearchRAG(
    pinecone_index=index,
    openai_client=client,
    reranker=CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
)

result = rag.search("birthday gift for a 10 year old who loves science")
print(result["response"])

Handling Edge Cases

def handle_no_results(query: str, parsed: Dict) -> str:
    """Generate helpful response when no products match"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": """The customer's search returned no results. Provide a helpful response:
1. Acknowledge what they're looking for
2. Suggest relaxing their filters (if applicable)
3. Suggest alternative search terms
4. Offer to help them find something similar"""
            },
            {
                "role": "user",
                "content": f"Query: {query}\nParsed filters: {json.dumps(parsed)}"
            }
        ]
    )
    return response.choices[0].message.content


def handle_vague_query(query: str) -> str:
    """Ask clarifying questions for vague queries"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": """The customer's query is too vague to provide good results. Ask 2-3 clarifying questions to help narrow down what they're looking for. Be conversational and helpful."""
            },
            {
                "role": "user",
                "content": f"Query: {query}"
            }
        ]
    )
    return response.choices[0].message.content


# Example
vague_query = "I need a gift"
clarification = handle_vague_query(vague_query)
print(clarification)

I'd love to help you find the perfect gift! Let me ask a few questions:

1. Who is the gift for? (friend, family member, colleague)
2. What's your budget range?
3. Do you know any of their interests or hobbies?

Once I know a bit more, I can suggest some great options!

Key Takeaways

Next up: Product Q&A Agent - Answer detailed questions about specific products using RAG.