Magento: Intelligent Catalog Assistant

Magento (Adobe Commerce) is the reference for complex e-commerce catalogs: thousands of SKUs, configurable attributes, B2B catalogs with custom pricing. An AI assistant capable of navigating this complexity offers a major competitive advantage. This guide shows you how to deploy a RAG chatbot optimized for Magento's specifics.

Why Magento Requires a Specific Approach

The Complexity of Magento Catalogs

Magento handles much richer data structures than other platforms:

Magento-Specific Use Cases

• B2B catalog navigation: "Show me all DN50 stainless steel valves compatible with our system"
• Product configuration: "Which SKU corresponds to a 27-inch, 4K monitor with USB-C port?"
• Customer pricing: "What's my price for 500 units of this component?"
• Multi-warehouse stock: "Where is this product available for fast delivery in California?"

Magento + RAG Architecture

Magento GraphQL Connector

DEVELOPERpython
import requests
from typing import List, Dict, Optional

class MagentoGraphQLConnector:
    def __init__(self, base_url: str, bearer_token: str):
        self.base_url = base_url
        self.graphql_url = f"{base_url}/graphql"
        self.headers = {
            'Authorization': f'Bearer {bearer_token}',
            'Content-Type': 'application/json'
        }

    def get_all_products(
        self,
        store_view: str = 'default',
        page_size: int = 100
    ) -> List[Dict]:
        """Retrieve all products via GraphQL."""
        products = []
        current_page = 1

        while True:
            query = self._build_products_query(current_page, page_size)

            response = requests.post(
                self.graphql_url,
                json={'query': query},
                headers={**self.headers, 'Store': store_view}
            )
            response.raise_for_status()

            data = response.json()
            items = data['data']['products']['items']

            if not items:
                break

            for item in items:
                formatted = self._format_product(item)
                products.append(formatted)

            total_pages = data['data']['products']['page_info']['total_pages']
            if current_page >= total_pages:
                break

            current_page += 1

        return products

    def _build_products_query(self, page: int, page_size: int) -> str:
        """Build GraphQL query for products."""
        return f"""
        {{
            products(
                search: ""
                pageSize: {page_size}
                currentPage: {page}
            ) {{
                total_count
                page_info {{
                    total_pages
                    current_page
                }}
                items {{
                    id
                    sku
                    name
                    url_key
                    description {{
                        html
                    }}
                    short_description {{
                        html
                    }}
                    price_range {{
                        minimum_price {{
                            regular_price {{
                                value
                                currency
                            }}
                            final_price {{
                                value
                                currency
                            }}
                        }}
                    }}
                    stock_status
                    categories {{
                        id
                        name
                        path
                    }}
                    ... on ConfigurableProduct {{
                        configurable_options {{
                            attribute_code
                            label
                            values {{
                                label
                                value_index
                            }}
                        }}
                        variants {{
                            product {{
                                sku
                                name
                                stock_status
                            }}
                            attributes {{
                                code
                                label
                                value_index
                            }}
                        }}
                    }}
                    custom_attributes {{
                        attribute_code
                        value
                    }}
                    image {{
                        url
                    }}
                }}
            }}
        }}
        """

    def _format_product(self, product: dict) -> dict:
        """Format a Magento product for RAG."""
        content_parts = [
            f"Product: {product['name']}",
            f"SKU: {product['sku']}",
        ]

        # Description
        if product.get('description', {}).get('html'):
            content_parts.append(
                f"Description: {self._clean_html(product['description']['html'])}"
            )

        # Price
        price_info = product.get('price_range', {}).get('minimum_price', {})
        if price_info:
            regular = price_info.get('regular_price', {}).get('value')
            final = price_info.get('final_price', {}).get('value')
            currency = price_info.get('regular_price', {}).get('currency', 'USD')

            if regular and final and regular != final:
                content_parts.append(f"Price: {final} {currency} (was {regular} {currency})")
            elif final:
                content_parts.append(f"Price: {final} {currency}")

        # Categories
        categories = [cat['name'] for cat in product.get('categories', [])]
        if categories:
            content_parts.append(f"Categories: {' > '.join(categories)}")

        # Custom attributes
        for attr in product.get('custom_attributes', []):
            if attr.get('value'):
                content_parts.append(f"{attr['attribute_code']}: {attr['value']}")

        # Configurable options
        if product.get('configurable_options'):
            options_text = []
            for option in product['configurable_options']:
                values = [v['label'] for v in option['values']]
                options_text.append(f"{option['label']}: {', '.join(values)}")
            content_parts.append(f"Available options: {'; '.join(options_text)}")

        # Variants
        if product.get('variants'):
            variants_text = []
            for variant in product['variants'][:10]:  # Limit
                variant_attrs = [f"{a['label']}" for a in variant['attributes']]
                status = "Available" if variant['product']['stock_status'] == 'IN_STOCK' else "Out of stock"
                variants_text.append(f"{' / '.join(variant_attrs)} ({status})")
            content_parts.append(f"Variants: {'; '.join(variants_text)}")

        return {
            "id": f"magento_{product['id']}",
            "title": product['name'],
            "content": "\n".join(content_parts),
            "metadata": {
                "source": "magento",
                "product_id": product['id'],
                "sku": product['sku'],
                "url_key": product['url_key'],
                "price": price_info.get('final_price', {}).get('value'),
                "in_stock": product.get('stock_status') == 'IN_STOCK',
                "categories": [c['id'] for c in product.get('categories', [])],
                "category_names": categories,
                "image": product.get('image', {}).get('url'),
                "is_configurable": bool(product.get('configurable_options'))
            }
        }

    def _clean_html(self, html: str) -> str:
        from bs4 import BeautifulSoup
        return BeautifulSoup(html or '', 'html.parser').get_text(separator=' ')

B2B Catalog Management

DEVELOPERpython
class MagentoB2BConnector(MagentoGraphQLConnector):
    """Extension for Magento B2B features."""

    def get_customer_specific_prices(
        self,
        customer_group_id: int,
        product_ids: List[int]
    ) -> Dict[int, float]:
        """Retrieve prices specific to customer group."""
        query = f"""
        {{
            products(filter: {{id: {{in: {product_ids}}}}}) {{
                items {{
                    id
                    sku
                    price_tiers {{
                        customer_group_id
                        qty
                        final_price {{
                            value
                            currency
                        }}
                    }}
                }}
            }}
        }}
        """

        response = requests.post(
            self.graphql_url,
            json={'query': query},
            headers=self.headers
        )

        prices = {}
        for item in response.json()['data']['products']['items']:
            # Find price for this customer group
            for tier in item.get('price_tiers', []):
                if tier['customer_group_id'] == customer_group_id:
                    prices[item['id']] = tier['final_price']['value']
                    break

        return prices

    def get_company_catalog(self, company_id: int) -> List[Dict]:
        """Retrieve catalog assigned to a company."""
        # Logic specific to B2B module
        query = f"""
        {{
            company(id: {company_id}) {{
                shared_catalog {{
                    id
                    name
                    products {{
                        items {{
                            id
                            sku
                            name
                        }}
                    }}
                }}
            }}
        }}
        """
        # Implementation according to your B2B configuration
        pass

Magento System Prompt

DEVELOPERpython
MAGENTO_ASSISTANT_PROMPT = """You are the product assistant for {store_name}, a {b2b_or_b2c} store on Magento.

## Your role
Help customers navigate our catalog of {product_count} products
and find exactly what they need.

## Catalog specifics
- Configurable products with multiple options
- {categories_count} hierarchical categories
- Detailed technical attributes
{b2b_context}

## Strict rules
1. Base your answers ONLY on the provided catalog
2. For configurable products, specify available options
3. Always indicate availability (stock)
4. For B2B prices, mention they may vary by customer account

## Product response format
**[Product name]**
SKU: {sku}
Price: From $XX {b2b_price_mention}
Availability: In stock / X options available
[Relevant features]
[View product]({url})

## When customer searches for technical product
1. Ask clarifying questions about specifications
2. Suggest 2-3 options matching the need
3. Explain differences between options
4. Suggest compatible accessories if relevant

## Available catalog
{context}

## Conversation history
{history}

## Customer question
{query}
"""

# Additional B2B context
B2B_CONTEXT = """
- Pricing by customer group and quantity
- Restricted catalog by company possible
- Quotes available for large quantities
"""

Advanced Magento Use Cases

Technical Attribute Navigation

DEVELOPERpython
async def handle_technical_search(query: str, context: dict):
    """Search by technical specifications."""
    # Extract attributes from query
    extracted = await extract_technical_attributes(query)

    # Build filters
    filters = {}
    for attr, value in extracted.items():
        if attr in SEARCHABLE_ATTRIBUTES:
            filters[attr] = value

    # Search with filters
    products = await retriever.search(
        query=query,
        filters=filters,
        top_k=10
    )

    if not products:
        # Suggest alternatives
        return await suggest_alternatives(extracted)

    return format_technical_results(products, extracted)

async def extract_technical_attributes(query: str) -> dict:
    """Extract technical attributes from a query."""
    prompt = f"""
    Extract technical specifications from this query:
    "{query}"

    Possible attributes: {AVAILABLE_ATTRIBUTES}

    Respond in JSON with identified attributes.
    Example: {{"diameter": "DN50", "material": "stainless", "pressure": "PN16"}}
    """

    response = await llm.generate(prompt, temperature=0)
    return json.loads(response)

Guided Product Configuration

DEVELOPERpython
class ProductConfigurator:
    """Guide configuration of complex products."""

    async def start_configuration(self, product_id: str) -> dict:
        """Start a configuration session."""
        product = await self.get_configurable_product(product_id)

        if not product.get('configurable_options'):
            return {"error": "This product is not configurable"}

        return {
            "product": product['name'],
            "steps": [
                {
                    "attribute": opt['attribute_code'],
                    "label": opt['label'],
                    "options": [v['label'] for v in opt['values']]
                }
                for opt in product['configurable_options']
            ],
            "current_step": 0
        }

    async def select_option(
        self,
        session: dict,
        selection: str
    ) -> dict:
        """Process an option selection."""
        current_step = session['current_step']
        steps = session['steps']

        # Validate selection
        current_options = steps[current_step]['options']
        if selection not in current_options:
            similar = self._find_similar_option(selection, current_options)
            if similar:
                return {
                    "clarification": f"Did you mean '{similar}'?",
                    "options": current_options
                }
            return {
                "error": f"Option not available. Choices: {', '.join(current_options)}"
            }

        # Record selection
        session['selections'] = session.get('selections', {})
        session['selections'][steps[current_step]['attribute']] = selection

        # Move to next step or finalize
        if current_step + 1 < len(steps):
            session['current_step'] += 1
            next_step = steps[session['current_step']]

            # Filter available options based on previous selections
            available = await self._get_available_options(
                session['product_id'],
                session['selections'],
                next_step['attribute']
            )

            return {
                "confirmed": f"{steps[current_step]['label']}: {selection}",
                "next_question": f"Which {next_step['label']} would you like?",
                "available_options": available
            }
        else:
            # Configuration complete
            variant = await self._find_variant(
                session['product_id'],
                session['selections']
            )

            return {
                "complete": True,
                "configuration": session['selections'],
                "variant_sku": variant['sku'],
                "price": variant['price'],
                "stock": variant['stock_status'],
                "message": f"Configuration complete! The product {variant['name']} "
                          f"is available at ${variant['price']}."
            }

Multi-Store Search

DEVELOPERpython
async def search_across_stores(query: str, user_context: dict):
    """Search across multiple Magento store views."""
    user_store = user_context.get('store_view', 'default')
    user_language = user_context.get('language', 'en')

    # Search in user's store
    primary_results = await retriever.search(
        query=query,
        filters={"store_view": user_store},
        top_k=10
    )

    # If few results, extend search
    if len(primary_results) < 3:
        all_results = await retriever.search(
            query=query,
            top_k=10
        )

        # Indicate products from other stores
        for result in all_results:
            if result not in primary_results:
                result['from_other_store'] = True
                primary_results.append(result)

    return primary_results

Dynamic B2B Pricing

DEVELOPERpython
async def get_customer_pricing(
    product_ids: List[str],
    customer_context: dict
) -> Dict[str, dict]:
    """Retrieve personalized pricing for B2B customer."""
    customer_group = customer_context.get('group_id')
    company_id = customer_context.get('company_id')
    requested_qty = customer_context.get('quantity', 1)

    pricing = {}

    for product_id in product_ids:
        base_price = await get_base_price(product_id)

        # Customer group price
        group_price = await get_group_price(product_id, customer_group)

        # Quantity price (tier pricing)
        tier_price = await get_tier_price(product_id, requested_qty)

        # Contract price (if applicable)
        contract_price = None
        if company_id:
            contract_price = await get_contract_price(product_id, company_id)

        # Determine best price
        applicable_prices = [p for p in [base_price, group_price, tier_price, contract_price] if p]
        best_price = min(applicable_prices) if applicable_prices else base_price

        pricing[product_id] = {
            "base_price": base_price,
            "your_price": best_price,
            "discount_percent": round((1 - best_price / base_price) * 100, 1) if base_price else 0,
            "tier_info": f"Price for {requested_qty}+ units" if tier_price else None
        }

    return pricing

Magento Integration

Magento 2 Module

DEVELOPERphp
<?php
// app/code/Ailog/ChatWidget/view/frontend/templates/widget.phtml

/** @var \Ailog\ChatWidget\Block\Widget $block */
$helper = $this->helper(\Ailog\ChatWidget\Helper\Data::class);
$customerSession = $this->helper(\Magento\Customer\Model\Session::class);
?>

<script type="text/x-magento-init">
{
    "*": {
        "Ailog_ChatWidget/js/widget": {
            "channelId": "<?= $block->escapeJs($helper->getChannelId()) ?>",
            "context": {
                "platform": "magento2",
                "storeView": "<?= $block->escapeJs($helper->getStoreViewCode()) ?>",
                "currency": "<?= $block->escapeJs($helper->getCurrencyCode()) ?>",
                "customerGroup": <?= $customerSession->isLoggedIn() ? $customerSession->getCustomerGroupId() : 'null' ?>,
                "isB2B": <?= $helper->isB2BEnabled() ? 'true' : 'false' ?>,
                "currentProduct": <?= $block->getCurrentProductJson() ?>,
                "currentCategory": <?= $block->getCurrentCategoryJson() ?>
            }
        }
    }
}
</script>

REST API Extension

DEVELOPERphp
<?php
// app/code/Ailog/ChatWidget/Api/CatalogExportInterface.php

namespace Ailog\ChatWidget\Api;

interface CatalogExportInterface
{
    /**
     * Export catalog for RAG indexing
     *
     * @param int $storeId
     * @param int $page
     * @param int $pageSize
     * @return \Ailog\ChatWidget\Api\Data\ProductExportInterface[]
     */
    public function export(int $storeId, int $page = 1, int $pageSize = 100): array;

    /**
     * Get product with customer-specific pricing
     *
     * @param int $productId
     * @param int $customerId
     * @param int $qty
     * @return \Ailog\ChatWidget\Api\Data\ProductPricingInterface
     */
    public function getCustomerPricing(int $productId, int $customerId, int $qty = 1);
}

Measuring Performance

Magento-Specific KPIs

Related Resources

• E-commerce AI Chatbot - E-commerce pillar guide
• Dynamic E-commerce FAQ - Contextual answers
• AI Upsell and Cross-sell - Advanced recommendations
• Introduction to RAG - Fundamentals

---

Deploy on Magento with Ailog

Complex Magento catalogs require an adapted solution. Ailog offers:

• Native Magento connector: GraphQL and REST API
• B2B management: Per-customer pricing, shared catalogs
• Configurable products: Guided option navigation
• Multi-store: Multiple store view support
• French hosting: GDPR compliance

Try Ailog on your Magento store and simplify navigation through your complex catalogs.