Chatbot E‑Commerce KI : Conversionen mit RAG steigern

KI-Chatbots revolutionieren den E‑Commerce. Ein Assistent, der Ihre Produkte versteht, Kunden führt und Fragen sofort beantwortet, kann Ihre Conversion-Rate deutlich verbessern. Dieser Leitfaden zeigt, wie Sie einen leistungsfähigen RAG-Chatbot für Ihren Onlineshop bereitstellen.

Warum ein RAG-Chatbot für den E‑Commerce?

Die Grenzen traditioneller Chatbots

Regelbasierte Chatbots (if/then) oder statische FAQs stoßen schnell an ihre Grenzen:

• Rigidität: Verstehen nur die vorgesehenen Formulierungen
• Hoher Wartungsaufwand: Jede neue Frage erfordert eine Regel
• Frustrierende Erfahrung: „Ich habe Ihre Frage nicht verstanden“ in Endlosschleife
• Kein Produktkontext: Können nicht das passende Produkt empfehlen

Der Vorteil von RAG

Ein RAG-Chatbot, der mit Ihrem Produktkatalog verbunden ist:

Nachgewiesener Business-Impact

Shops, die KI-Chatbots einsetzen, berichten von:

• +15–30% Conversion-Rate bei engagierten Besuchern
• -40% Support-Tickets für Pre-Sales-Fragen
• +25% durchschnittlicher Warenkorbwert durch Empfehlungen
• 24/7 Verfügbarkeit ohne zusätzliche Kosten

Architektur eines RAG E‑Commerce Chatbots

┌─────────────────────────────────────────────────────────────┐
│                    SOURCES DE DONNÉES                        │
├─────────────┬─────────────┬─────────────┬──────────────────┤
│  Catalogue  │    FAQ      │  Politique  │   Historique     │
│  Produits   │  Support    │   Retour    │    Client        │
└──────┬──────┴──────┬──────┴──────┬──────┴────────┬─────────┘
       │             │             │               │
       └─────────────┴──────┬──────┴───────────────┘
                            ▼
                   ┌─────────────────┐
                   │   Indexation    │
                   │  (Embeddings)   │
                   └────────┬────────┘
                            ▼
                   ┌─────────────────┐
                   │  Base Vecto.    │
                   │   (Qdrant)      │
                   └────────┬────────┘
                            │
┌───────────────────────────┴───────────────────────────────┐
│                        RUNTIME                              │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────┐    ┌───────────┐    ┌──────────┐   ┌────────┐ │
│  │ Widget  │───▶│ Retriever │───▶│   LLM    │──▶│Response│ │
│  │  Chat   │    │           │    │(Génération)│   │        │ │
│  └─────────┘    └───────────┘    └──────────┘   └────────┘ │
└─────────────────────────────────────────────────────────────┘

Produktkatalog synchronisieren

Shopify-Connector

DEVELOPERpython
import shopify
from datetime import datetime

class ShopifyConnector:
    def __init__(self, shop_url: str, access_token: str):
        shopify.ShopifyResource.set_site(shop_url)
        shopify.ShopifyResource.set_headers({
            'X-Shopify-Access-Token': access_token
        })

    def get_all_products(self) -> list[dict]:
        """
        Récupère tous les produits avec leurs variantes
        """
        products = []
        page = shopify.Product.find(limit=250)

        while page:
            for product in page:
                products.append(self._format_product(product))

            # Paginierung
            if len(page) < 250:
                break
            page = shopify.Product.find(limit=250, since_id=page[-1].id)

        return products

    def _format_product(self, product) -> dict:
        """
        Formate un produit pour l'indexation RAG
        """
        # Eine ausführliche Beschreibung für das RAG erstellen
        description_parts = [
            f"Produit : {product.title}",
            f"Description : {self._clean_html(product.body_html)}",
            f"Catégorie : {product.product_type}",
            f"Marque : {product.vendor}",
            f"Tags : {', '.join(product.tags.split(', '))}",
        ]

        # Varianten hinzufügen
        for variant in product.variants:
            variant_info = f"Variante : {variant.title}"
            if variant.price:
                variant_info += f" - Prix : {variant.price}€"
            if variant.inventory_quantity:
                variant_info += f" - Stock : {variant.inventory_quantity}"
            description_parts.append(variant_info)

        return {
            "id": f"product_{product.id}",
            "title": product.title,
            "content": "\n".join(description_parts),
            "metadata": {
                "type": "product",
                "product_id": product.id,
                "price_min": min(v.price for v in product.variants),
                "price_max": max(v.price for v in product.variants),
                "category": product.product_type,
                "in_stock": any(v.inventory_quantity > 0 for v in product.variants),
                "url": f"/products/{product.handle}",
                "image": product.images[0].src if product.images else None,
                "updated_at": product.updated_at
            }
        }

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

WooCommerce-Connector

DEVELOPERpython
from woocommerce import API

class WooCommerceConnector:
    def __init__(self, url: str, consumer_key: str, consumer_secret: str):
        self.wcapi = API(
            url=url,
            consumer_key=consumer_key,
            consumer_secret=consumer_secret,
            version="wc/v3"
        )

    def get_all_products(self) -> list[dict]:
        """
        Récupère tous les produits WooCommerce
        """
        products = []
        page = 1

        while True:
            response = self.wcapi.get("products", params={
                "per_page": 100,
                "page": page,
                "status": "publish"
            })

            batch = response.json()
            if not batch:
                break

            for product in batch:
                products.append(self._format_product(product))

            page += 1

        return products

    def _format_product(self, product: dict) -> dict:
        """
        Formate un produit WooCommerce pour le RAG
        """
        description_parts = [
            f"Produit : {product['name']}",
            f"Description : {self._clean_html(product.get('description', ''))}",
            f"Description courte : {self._clean_html(product.get('short_description', ''))}",
        ]

        # Kategorien
        categories = [cat['name'] for cat in product.get('categories', [])]
        if categories:
            description_parts.append(f"Catégories : {', '.join(categories)}")

        # Attribute
        for attr in product.get('attributes', []):
            description_parts.append(f"{attr['name']} : {', '.join(attr['options'])}")

        # Preise
        if product.get('sale_price'):
            description_parts.append(f"Prix : {product['sale_price']}€ (au lieu de {product['regular_price']}€)")
        else:
            description_parts.append(f"Prix : {product.get('price', 'N/A')}€")

        return {
            "id": f"product_{product['id']}",
            "title": product['name'],
            "content": "\n".join(description_parts),
            "metadata": {
                "type": "product",
                "product_id": product['id'],
                "price": float(product.get('price', 0)),
                "categories": categories,
                "in_stock": product.get('stock_status') == 'instock',
                "url": product.get('permalink', ''),
                "image": product['images'][0]['src'] if product.get('images') else None
            }
        }

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

Automatische Synchronisation

DEVELOPERpython
import asyncio
from datetime import datetime, timedelta

class ProductSyncService:
    def __init__(self, connector, indexer, sync_interval_hours: int = 6):
        self.connector = connector
        self.indexer = indexer
        self.sync_interval = timedelta(hours=sync_interval_hours)
        self.last_sync = None

    async def start_sync_loop(self):
        """
        Lance la synchronisation périodique
        """
        while True:
            try:
                await self.sync()
            except Exception as e:
                print(f"Erreur sync : {e}")

            await asyncio.sleep(self.sync_interval.total_seconds())

    async def sync(self):
        """
        Synchronise les produits avec la base vectorielle
        """
        print(f"[{datetime.now()}] Début synchronisation produits...")

        # Alle Produkte abrufen
        products = self.connector.get_all_products()

        # Indexieren
        await self.indexer.index_documents(
            documents=products,
            collection="products"
        )

        self.last_sync = datetime.now()
        print(f"[{datetime.now()}] {len(products)} produits synchronisés")

    async def sync_single_product(self, product_id: str):
        """
        Synchronise un seul produit (webhook)
        """
        product = self.connector.get_product(product_id)
        if product:
            await self.indexer.update_document(
                document=product,
                collection="products"
            )

Optimierte E‑Commerce-Prompts

Verkaufsassistent-Prompt

DEVELOPERpython
SALES_ASSISTANT_PROMPT = """Tu es l'assistant de vente IA de {boutique_name}, une boutique spécialisée dans {categorie}.

OBJECTIF PRINCIPAL :
Aider les visiteurs à trouver le produit parfait et les accompagner jusqu'à l'achat.

RÈGLES STRICTES :
1. Base tes recommandations UNIQUEMENT sur les produits du catalogue fourni
2. Ne jamais inventer de produits, prix ou caractéristiques
3. Si un produit n'est pas en stock, le mentionner et proposer des alternatives
4. Toujours inclure le lien vers la fiche produit

COMPORTEMENT :
- Pose des questions pour comprendre le besoin
- Propose 2-3 produits maximum par réponse
- Mets en avant les promotions actuelles
- Rassure sur la politique de retour si hésitation détectée

FORMAT RECOMMANDATION :
Pour chaque produit recommandé, utilise ce format :
**[Nom du produit]**
Prix : XX€
[Description courte de 1-2 phrases expliquant pourquoi ce produit correspond]
👉 [Voir le produit](URL)

PRODUITS DISPONIBLES :
{context}

CONVERSATION :
{history}

QUESTION CLIENT :
{question}
"""

After‑Sales Support-Prompt

DEVELOPERpython
SUPPORT_PROMPT = """Tu es l'assistant support de {boutique_name}.

OBJECTIF :
Aider les clients avec leurs commandes, retours et questions après-achat.

RÈGLES :
1. Réponds uniquement à partir des politiques et FAQ fournies
2. Pour toute question sur une commande spécifique, demande le numéro de commande
3. Si le problème nécessite une intervention humaine, oriente vers le formulaire de contact
4. Reste empathique face aux frustrations

INFORMATIONS DISPONIBLES :
{context}

QUESTION :
{question}
"""

Intent-Erkennung und Routing

DEVELOPERpython
from enum import Enum

class CustomerIntent(Enum):
    PRODUCT_SEARCH = "recherche_produit"
    PRODUCT_COMPARISON = "comparaison"
    PRICE_QUESTION = "question_prix"
    STOCK_CHECK = "verification_stock"
    ORDER_STATUS = "suivi_commande"
    RETURN_REQUEST = "demande_retour"
    GENERAL_SUPPORT = "support_general"
    CHITCHAT = "conversation"

class IntentDetector:
    def __init__(self, llm):
        self.llm = llm

    async def detect(self, message: str, history: list = None) -> CustomerIntent:
        """
        Détecte l'intention du client
        """
        prompt = f"""
        Analyse ce message client et détermine son intention principale.

        Message : {message}

        Intentions possibles :
        - recherche_produit : Le client cherche un produit
        - comparaison : Le client compare des produits
        - question_prix : Question sur un prix ou promotion
        - verification_stock : Vérifie la disponibilité
        - suivi_commande : Demande le statut d'une commande
        - demande_retour : Veut retourner un produit
        - support_general : Question support autre
        - conversation : Simple conversation/salutation

        Réponds uniquement avec l'intention (un seul mot).
        """

        response = await self.llm.generate(prompt, temperature=0)
        intent_str = response.strip().lower()

        try:
            return CustomerIntent(intent_str)
        except ValueError:
            return CustomerIntent.GENERAL_SUPPORT

class EcommerceRAGRouter:
    def __init__(self, intent_detector, retrievers: dict, generators: dict):
        self.intent_detector = intent_detector
        self.retrievers = retrievers
        self.generators = generators

    async def route(self, message: str, history: list = None) -> dict:
        """
        Route la requête vers le bon retriever et générateur
        """
        intent = await self.intent_detector.detect(message, history)

        # Mapping intention -> configuration
        config = {
            CustomerIntent.PRODUCT_SEARCH: {
                "retriever": "products",
                "generator": "sales",
                "filters": {"in_stock": True}
            },
            CustomerIntent.PRODUCT_COMPARISON: {
                "retriever": "products",
                "generator": "sales",
                "top_k": 10
            },
            CustomerIntent.PRICE_QUESTION: {
                "retriever": "products",
                "generator": "sales"
            },
            CustomerIntent.STOCK_CHECK: {
                "retriever": "products",
                "generator": "sales"
            },
            CustomerIntent.ORDER_STATUS: {
                "retriever": "support",
                "generator": "support",
                "require_order_number": True
            },
            CustomerIntent.RETURN_REQUEST: {
                "retriever": "policies",
                "generator": "support"
            },
            CustomerIntent.GENERAL_SUPPORT: {
                "retriever": "support",
                "generator": "support"
            },
            CustomerIntent.CHITCHAT: {
                "retriever": None,
                "generator": "chitchat"
            }
        }

        return {
            "intent": intent,
            "config": config.get(intent, config[CustomerIntent.GENERAL_SUPPORT])
        }

Intelligente Produktempfehlungen

Empfehlung per Ähnlichkeit

DEVELOPERpython
class ProductRecommender:
    def __init__(self, retriever):
        self.retriever = retriever

    async def similar_products(
        self,
        product_id: str,
        exclude_ids: list[str] = None,
        limit: int = 4
    ) -> list[dict]:
        """
        Trouve des produits similaires
        """
        # Quellprodukt abrufen
        source_product = await self.retriever.get_by_id(product_id)
        if not source_product:
            return []

        # Nach Ähnlichkeit suchen
        results = await self.retriever.search(
            query=source_product["content"],
            top_k=limit + len(exclude_ids or []) + 1,
            filters={"type": "product", "in_stock": True}
        )

        # Filtern
        exclude = set(exclude_ids or []) | {product_id}
        recommendations = [
            r for r in results
            if r["metadata"]["product_id"] not in exclude
        ][:limit]

        return recommendations

    async def complementary_products(
        self,
        cart_items: list[str],
        limit: int = 3
    ) -> list[dict]:
        """
        Suggère des produits complémentaires (cross-sell)
        """
        # Abfrage basierend auf dem Warenkorb erstellen
        cart_products = []
        for product_id in cart_items:
            product = await self.retriever.get_by_id(product_id)
            if product:
                cart_products.append(product["title"])

        if not cart_products:
            return []

        # Ergänzende Produkte suchen
        query = f"Produits qui vont bien avec : {', '.join(cart_products)}"

        results = await self.retriever.search(
            query=query,
            top_k=limit + len(cart_items),
            filters={"type": "product", "in_stock": True}
        )

        # Produkte ausschließen, die bereits im Warenkorb sind
        return [r for r in results if r["metadata"]["product_id"] not in cart_items][:limit]

Kontextbasierte Empfehlung mit LLM

DEVELOPERpython
async def generate_recommendation(
    self,
    customer_query: str,
    customer_context: dict,
    retrieved_products: list[dict]
) -> str:
    """
    Génère une recommandation personnalisée
    """
    prompt = f"""
    Tu dois recommander des produits à un client.

    CONTEXTE CLIENT :
    - Requête : {customer_query}
    - Budget indiqué : {customer_context.get('budget', 'Non précisé')}
    - Historique achats : {customer_context.get('purchase_history', 'Nouveau client')}

    PRODUITS DISPONIBLES :
    {self._format_products(retrieved_products)}

    CONSIGNES :
    1. Choisis 2-3 produits maximum
    2. Explique pourquoi chaque produit correspond
    3. Adapte ton discours au budget si indiqué
    4. Propose une alternative moins chère si budget limité

    RECOMMANDATION :
    """

    return await self.llm.generate(prompt, temperature=0.3)

Warenkorb-Management und Upsell

Erkennung von Warenkorbabbrüchen

DEVELOPERpython
class CartAbandonmentDetector:
    def __init__(self, session_timeout_minutes: int = 30):
        self.timeout = timedelta(minutes=session_timeout_minutes)
        self.active_carts = {}

    def track_activity(self, session_id: str, cart_items: list):
        """
        Suit l'activité du panier
        """
        self.active_carts[session_id] = {
            "items": cart_items,
            "last_activity": datetime.now(),
            "notified": False
        }

    def get_abandoned_carts(self) -> list[dict]:
        """
        Identifie les paniers abandonnés
        """
        abandoned = []
        now = datetime.now()

        for session_id, cart in self.active_carts.items():
            if (now - cart["last_activity"] > self.timeout
                and not cart["notified"]
                and len(cart["items"]) > 0):
                abandoned.append({
                    "session_id": session_id,
                    "items": cart["items"],
                    "abandoned_at": cart["last_activity"]
                })

        return abandoned

class CartRecoveryBot:
    def __init__(self, recommender, llm):
        self.recommender = recommender
        self.llm = llm

    async def generate_recovery_message(
        self,
        cart_items: list[dict],
        customer_name: str = None
    ) -> str:
        """
        Génère un message de récupération de panier
        """
        items_text = "\n".join([
            f"- {item['title']} ({item['price']}€)"
            for item in cart_items
        ])

        prompt = f"""
        Un client a abandonné son panier. Génère un message amical pour l'encourager à finaliser.

        Articles dans le panier :
        {items_text}

        Le message doit :
        - Être chaleureux mais pas insistant
        - Rappeler les articles
        - Mentionner qu'on peut aider si question
        - Proposer de l'aide pour choisir

        {"Commence par 'Bonjour " + customer_name + "'" if customer_name else ""}
        """

        return await self.llm.generate(prompt, temperature=0.7)

Optimiertes Chat-Widget

Widget-Konfiguration

DEVELOPERjavascript
// Konfiguration des E-Commerce-Widgets
const chatConfig = {
  // Erscheinungsbild
  theme: {
    primaryColor: '#2563eb',
    position: 'bottom-right',
    bubbleText: 'Besoin d\'aide ?'
  },

  // Proaktives Verhalten
  proactive: {
    enabled: true,
    triggers: [
      {
        event: 'time_on_page',
        value: 60, // secondes
        message: 'Vous cherchez quelque chose en particulier ? Je peux vous aider !'
      },
      {
        event: 'page_type',
        value: 'product',
        delay: 30,
        message: 'Des questions sur ce produit ? Je suis là pour vous aider.'
      },
      {
        event: 'cart_value',
        value: 100,
        message: 'Saviez-vous que la livraison est gratuite dès 150€ ?'
      }
    ]
  },

  // Kontextbezogene Schnellantworten
  quickReplies: {
    homepage: [
      'Quelles sont vos meilleures ventes ?',
      'Y a-t-il des promotions ?',
      'Comment fonctionne la livraison ?'
    ],
    product: [
      'Ce produit est-il disponible ?',
      'Quels sont les délais de livraison ?',
      'Puis-je retourner ce produit ?'
    ],
    cart: [
      'Comment utiliser un code promo ?',
      'Quels sont les frais de livraison ?',
      'Puis-je modifier ma commande ?'
    ]
  }
};

Conversion-Tracking

DEVELOPERpython
class ChatAnalytics:
    def __init__(self, analytics_client):
        self.client = analytics_client

    def track_conversation(
        self,
        session_id: str,
        messages: list,
        outcome: str,
        metadata: dict = None
    ):
        """
        Track une conversation pour analyse
        """
        self.client.track({
            "event": "chat_conversation",
            "session_id": session_id,
            "message_count": len(messages),
            "outcome": outcome,  # "purchase", "support_ticket", "abandoned"
            "duration_seconds": self._calculate_duration(messages),
            "products_discussed": metadata.get("products", []),
            "intent_detected": metadata.get("intent"),
            "timestamp": datetime.now().isoformat()
        })

    def track_chat_to_purchase(
        self,
        session_id: str,
        order_value: float,
        products: list
    ):
        """
        Track une conversion chat -> achat
        """
        self.client.track({
            "event": "chat_conversion",
            "session_id": session_id,
            "order_value": order_value,
            "products": products,
            "attribution": "chat_assisted"
        })

    def get_chat_roi_metrics(self, period_days: int = 30) -> dict:
        """
        Calcule le ROI du chat
        """
        data = self.client.query(f"""
            SELECT
                COUNT(DISTINCT session_id) as conversations,
                COUNT(DISTINCT CASE WHEN outcome = 'purchase' THEN session_id END) as conversions,
                SUM(CASE WHEN outcome = 'purchase' THEN order_value ELSE 0 END) as revenue,
                AVG(message_count) as avg_messages
            FROM chat_events
            WHERE timestamp > NOW() - INTERVAL '{period_days} days'
        """)

        return {
            "total_conversations": data["conversations"],
            "conversions": data["conversions"],
            "conversion_rate": data["conversions"] / data["conversations"] * 100,
            "revenue_attributed": data["revenue"],
            "avg_messages_per_conversation": data["avg_messages"]
        }

Best Practices für E‑Commerce

1. Reichhaltige Antworten mit Produktkarten

Empfehlungen für Produkte immer visuell formatieren:

DEVELOPERpython
def format_product_card(product: dict) -> str:
    """
    Formate un produit en carte visuelle (Markdown)
    """
    price_text = f"{product['price']}€"
    if product.get('original_price'):
        price_text = f"~~{product['original_price']}€~~ **{product['price']}€**"

    stock_badge = "En stock" if product['in_stock'] else "Rupture de stock"

    return f"""
**{product['title']}**
{price_text} | {stock_badge}

{product['short_description']}

[Voir le produit]({product['url']})
"""

2. Umgang mit Lagerausfällen

DEVELOPERpython
async def handle_out_of_stock(self, product: dict, query: str) -> str:
    """
    Gère intelligemment les ruptures de stock
    """
    # Alternativen finden
    alternatives = await self.recommender.similar_products(
        product_id=product["id"],
        filters={"in_stock": True},
        limit=3
    )

    if alternatives:
        alt_text = "\n".join([format_product_card(p) for p in alternatives])
        return f"""
Malheureusement, **{product['title']}** n'est plus disponible actuellement.

Voici des alternatives similaires que vous pourriez aimer :

{alt_text}

Souhaitez-vous être notifié quand le produit sera de retour en stock ?
"""
    else:
        return f"""
Malheureusement, **{product['title']}** n'est plus disponible actuellement et nous n'avons pas d'alternative similaire.

Souhaitez-vous être notifié quand le produit sera de retour en stock ?
"""

3. Eskalation an menschlichen Support

DEVELOPERpython
def should_escalate(self, conversation: list, last_response: dict) -> bool:
    """
    Détermine si une escalade vers un humain est nécessaire
    """
    escalation_triggers = [
        # Wiederholte negative Stimmung
        len([m for m in conversation[-3:] if m.get("sentiment") == "negative"]) >= 2,

        # Explizite Anfrage
        any(kw in conversation[-1]["text"].lower() for kw in [
            "parler à quelqu'un", "humain", "conseiller", "téléphone"
        ]),

        # Geringes Modellvertrauen
        last_response.get("confidence", 1) < 0.5,

        # Zu lange Unterhaltung ohne Lösung
        len(conversation) > 10 and not last_response.get("resolved")
    ]

    return any(escalation_triggers)

Erfolg messen

Zu verfolgende KPIs

Pour aller plus loin

• Fondamentaux du Retrieval - Optimiser la recherche produit
• Support Client IA - Réduire les tickets support
• Introduction au RAG - Comprendre les bases

---

Starten Sie Ihren E‑Commerce-Chatbot mit Ailog

Die Bereitstellung eines leistungsfähigen RAG-Chatbots für Ihren Shop kann Monate an Entwicklung erfordern. Mit Ailog starten Sie in wenigen Stunden:

• Connecteurs natifs : Shopify, WooCommerce, PrestaShop, Magento
• Synchronisation automatique du catalogue produit
• Widget personnalisable qui s'intègre à votre charte graphique
• Analytics conversationnels pour mesurer le ROI
• Hébergement France conforme RGPD

Testez Ailog gratuitement et boostez vos conversions dès aujourd'hui.