Product Comparison Agent

"Should I get the iPhone or Samsung?" "Which laptop is better for video editing?" Comparison questions are complex - the answer depends on what matters to the customer.

Understanding the Comparison Request

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

client = OpenAI()

def parse_comparison_request(query: str) -> Dict:
    """Extract products and use case from comparison query"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """Parse the product comparison request. Return JSON with:
- products: List of specific products mentioned (or product types if generic)
- use_case: What they want to use it for (if mentioned)
- priorities: What features seem important to them
- budget: Any budget constraints mentioned
- comparison_type: "specific" (named products) or "category" (find best in category)

Examples:
"iPhone 15 vs Samsung S24" → specific products
"best laptop for video editing under $1500" → category search"""
            },
            {"role": "user", "content": query}
        ],
        response_format={"type": "json_object"}
    )

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


# Examples
query1 = "Compare the MacBook Pro M3 and Dell XPS 15 for video editing"
parsed1 = parse_comparison_request(query1)
# {
#   "products": ["MacBook Pro M3", "Dell XPS 15"],
#   "use_case": "video editing",
#   "priorities": ["performance", "video rendering"],
#   "budget": null,
#   "comparison_type": "specific"
# }

query2 = "What's the best wireless earbuds under $150 for working out?"
parsed2 = parse_comparison_request(query2)
# {
#   "products": [],
#   "use_case": "working out",
#   "priorities": ["durability", "sweat resistance", "secure fit"],
#   "budget": 150,
#   "comparison_type": "category"
# }

The Comparison Agent

from pinecone import Pinecone

class ProductComparisonAgent:
    def __init__(self, product_index, openai_client, db):
        self.index = product_index
        self.client = openai_client
        self.db = db

    def compare(self, query: str) -> Dict:
        """Handle a product comparison request"""

        # Step 1: Parse the request
        parsed = parse_comparison_request(query)

        # Step 2: Get the products to compare
        if parsed["comparison_type"] == "specific":
            products = self._get_specific_products(parsed["products"])
        else:
            products = self._find_products_for_category(
                use_case=parsed["use_case"],
                budget=parsed.get("budget"),
                priorities=parsed.get("priorities", [])
            )

        if len(products) < 2:
            return self._handle_insufficient_products(parsed, products)

        # Step 3: Extract comparable specs
        comparison_data = self._extract_comparison_data(products)

        # Step 4: Generate comparison with recommendation
        response = self._generate_comparison(
            products=products,
            comparison_data=comparison_data,
            use_case=parsed.get("use_case"),
            priorities=parsed.get("priorities", [])
        )

        return {
            "response": response,
            "products": products,
            "comparison_data": comparison_data
        }

    def _get_specific_products(self, product_names: List[str]) -> List[Dict]:
        """Retrieve specific products by name using semantic search"""

        products = []
        for name in product_names:
            # Semantic search to find the product
            query_embedding = self.client.embeddings.create(
                model="text-embedding-3-small",
                input=name
            ).data[0].embedding

            results = self.index.query(
                vector=query_embedding,
                top_k=1,
                include_metadata=True
            )

            if results.matches:
                product_data = self._enrich_product(results.matches[0].metadata)
                products.append(product_data)

        return products

    def _find_products_for_category(
        self,
        use_case: str,
        budget: Optional[float],
        priorities: List[str]
    ) -> List[Dict]:
        """Find top products for a category based on use case"""

        # Build search query combining use case and priorities
        search_query = f"Best product for {use_case}. " + " ".join(priorities)

        query_embedding = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=search_query
        ).data[0].embedding

        # Build filter
        filter_dict = {"in_stock": True}
        if budget:
            filter_dict["price"] = {"$lte": budget}

        results = self.index.query(
            vector=query_embedding,
            top_k=10,
            include_metadata=True,
            filter=filter_dict
        )

        # Enrich and return top 3-5 products
        products = [
            self._enrich_product(m.metadata)
            for m in results.matches[:5]
        ]

        return products

    def _enrich_product(self, product_data: Dict) -> Dict:
        """Enrich product with full specs and reviews"""

        product_id = product_data.get("product_id") or product_data.get("id")

        # Get full product details from database
        full_product = self.db.get_product(product_id)

        if full_product:
            return {
                **product_data,
                "specifications": full_product.specifications,
                "reviews_summary": self._summarize_reviews(product_id),
                "pros_cons": self._extract_pros_cons(product_id)
            }

        return product_data

Extracting Comparable Specifications

def _extract_comparison_data(self, products: List[Dict]) -> Dict:
    """Extract and align specifications for comparison"""

    # Get all unique spec keys
    all_specs = set()
    for product in products:
        specs = product.get("specifications", {})
        all_specs.update(specs.keys())

    # Build comparison matrix
    comparison = {
        "products": [p["name"] for p in products],
        "prices": [p.get("price", "N/A") for p in products],
        "ratings": [p.get("rating", "N/A") for p in products],
        "specs": {}
    }

    for spec_key in all_specs:
        comparison["specs"][spec_key] = [
            p.get("specifications", {}).get(spec_key, "N/A")
            for p in products
        ]

    # Add pros/cons
    comparison["pros"] = [p.get("pros_cons", {}).get("pros", []) for p in products]
    comparison["cons"] = [p.get("pros_cons", {}).get("cons", []) for p in products]

    return comparison


def _extract_pros_cons(self, product_id: str) -> Dict:
    """Extract pros and cons from reviews using RAG"""

    # Get review chunks for this product
    results = self.index.query(
        vector=self._embed(f"pros and cons of product {product_id}"),
        top_k=10,
        filter={"product_id": product_id, "type": "review"}
    )

    reviews_text = "\n".join([m.metadata.get("text", "") for m in results.matches])

    response = self.client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": """Extract pros and cons from these product reviews.
Return JSON with:
- pros: List of 3-5 most mentioned positives
- cons: List of 3-5 most mentioned negatives
Be specific and cite frequency if notable."""
            },
            {"role": "user", "content": reviews_text}
        ],
        response_format={"type": "json_object"}
    )

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

Generating the Comparison

def _generate_comparison(
    self,
    products: List[Dict],
    comparison_data: Dict,
    use_case: Optional[str],
    priorities: List[str]
) -> str:
    """Generate a helpful comparison with recommendation"""

    # Format products for LLM
    products_context = self._format_products_for_comparison(products, comparison_data)

    use_case_context = f"\nCustomer's use case: {use_case}" if use_case else ""
    priorities_context = f"\nCustomer's priorities: {', '.join(priorities)}" if priorities else ""

    response = self.client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """You are a product comparison expert. Create a helpful, personalized comparison.

Structure your response:
1. **Quick Summary** - One sentence on the key difference
2. **Comparison Table** - Format as markdown table with key specs
3. **Deep Dive** - For each major category (performance, value, etc.), explain which wins and why
4. **Pros & Cons** - Bullet points for each product
5. **Recommendation** - Clear recommendation based on the customer's use case

Guidelines:
- Focus on specs that matter for the use case
- Explain technical specs in plain language
- Be honest about trade-offs
- If it's close, say so
- Give a clear recommendation with reasoning"""
            },
            {
                "role": "user",
                "content": f"""Compare these products:

{products_context}
{use_case_context}
{priorities_context}

Provide a comparison and recommendation."""
            }
        ],
        temperature=0.7
    )

    return response.choices[0].message.content


def _format_products_for_comparison(self, products: List[Dict], comparison_data: Dict) -> str:
    """Format products into structured text for LLM"""

    formatted = []

    for i, product in enumerate(products):
        product_info = f"""
### {product['name']}
**Price:** ${product.get('price', 'N/A')}
**Rating:** {product.get('rating', 'N/A')}/5 ({product.get('review_count', 0)} reviews)

**Specifications:**
"""
        for spec, values in comparison_data["specs"].items():
            product_info += f"- {spec}: {values[i]}\n"

        if comparison_data["pros"][i]:
            product_info += "\n**Pros (from reviews):**\n"
            for pro in comparison_data["pros"][i][:3]:
                product_info += f"- {pro}\n"

        if comparison_data["cons"][i]:
            product_info += "\n**Cons (from reviews):**\n"
            for con in comparison_data["cons"][i][:3]:
                product_info += f"- {con}\n"

        formatted.append(product_info)

    return "\n---\n".join(formatted)

Example Comparison Output

Handling Edge Cases

def _handle_insufficient_products(self, parsed: Dict, products: List[Dict]) -> Dict:
    """Handle when we can't find enough products to compare"""

    if len(products) == 0:
        return {
            "response": f"""I couldn't find the products you mentioned. Could you:
1. Check the spelling of the product names
2. Be more specific (e.g., "MacBook Pro 14-inch M3" instead of "MacBook")
3. Or describe what you're looking for and I'll find options to compare""",
            "products": [],
            "comparison_data": None
        }

    if len(products) == 1:
        product = products[0]
        return {
            "response": f"""I found **{product['name']}**, but couldn't find the other product(s) to compare.

Would you like me to:
1. Tell you about {product['name']}
2. Find similar products to compare it with
3. Search for a different product""",
            "products": products,
            "comparison_data": None
        }


def _handle_incompatible_products(self, products: List[Dict]) -> Optional[str]:
    """Warn if comparing very different product types"""

    categories = [p.get("category", "").split(">")[0] for p in products]
    unique_categories = set(categories)

    if len(unique_categories) > 1:
        return f"""Note: You're comparing products from different categories ({', '.join(unique_categories)}).
I'll do my best, but they serve different purposes. Let me know if you'd like suggestions for similar products instead."""

    return None

Key Takeaways

Next up: Hybrid Search Strategies - Combine keyword and semantic search for the best of both worlds.