Intercom + RAG: Chatbot-Support der nächsten Generation

TL;DR

Die RAG-Integration + Intercom erstellt einen Chatbot, der natürliche und kontextbezogene Gespräche führen kann und von Ihrer Wissensdatenbank gespeist wird. Im Gegensatz zu starren Szenario-Bots versteht er komplexe Fragen und liefert personalisierte Antworten. Dieser Leitfaden behandelt die conversationelle Architektur, die Intercom-API-Integration und Strategien zur Maximierung der Automatisierung bei gleichzeitiger Wahrung der menschlichen Erfahrung.

Pourquoi RAG + Intercom ?

Les limites des bots Intercom classiques

Les Resolution Bots et Custom Bots d'Intercom fonctionnent sur des arbres de décision :

Avantages du chatbot RAG

Un chatbot Intercom augmenté par RAG offre :

• Compréhension naturelle : Pas besoin de mots-clés exacts
• Réponses contextuelles : Basées sur l'historique conversation
• Mise à jour automatique : Synchronisé avec vos articles Help Center
• Personnalisation : Adapté au profil utilisateur et à son historique
• Escalade intelligente : Transfert pertinent quand nécessaire

Architecture du chatbot RAG

Vue d'ensemble

┌─────────────────────────────────────────────────────────────┐
│                     Intercom Messenger                       │
└─────────────────────────┬───────────────────────────────────┘
                          │ Webhook
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                    RAG Middleware                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │
│  │  Contexte   │  │  Retrieval  │  │  Génération LLM     │ │
│  │  Utilisateur│──│  Vectoriel  │──│  + Instructions     │ │
│  └─────────────┘  └─────────────┘  └─────────────────────┘ │
└─────────────────────────┬───────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────────┐
│              Intercom API (Reply/Assign)                     │
└─────────────────────────────────────────────────────────────┘

Intégration via Custom App

DEVELOPERpython
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
import hmac
import hashlib

app = FastAPI()

class IntercomWebhook(BaseModel):
    type: str
    topic: str
    data: dict

def verify_intercom_signature(request: Request, body: bytes) -> bool:
    """Verifiziert die Signatur des Intercom-Webhooks."""
    signature = request.headers.get("X-Hub-Signature")
    if not signature:
        return False

    expected = hmac.new(
        INTERCOM_CLIENT_SECRET.encode(),
        body,
        hashlib.sha1
    ).hexdigest()

    return hmac.compare_digest(f"sha1={expected}", signature)

@app.post("/intercom/webhook")
async def handle_intercom_webhook(request: Request):
    """
    Einstiegspunkt für alle Intercom-Webhooks.
    """
    body = await request.body()

    if not verify_intercom_signature(request, body):
        raise HTTPException(status_code=401, detail="Invalid signature")

    payload = await request.json()
    topic = payload.get("topic")

    if topic == "conversation.user.created":
        # Neue Konversation
        return await handle_new_conversation(payload["data"])
    elif topic == "conversation.user.replied":
        # Benutzer-Nachricht
        return await handle_user_message(payload["data"])
    elif topic == "conversation_part.tag.created":
        # Tag hinzugefügt (für Routing)
        return await handle_tag_added(payload["data"])

    return {"status": "ignored"}

Gestion des conversations

Handler de message utilisateur

DEVELOPERpython
class ConversationHandler:
    """
    Gère les conversations Intercom avec RAG.
    """

    def __init__(self, rag_client, intercom_client):
        self.rag = rag_client
        self.intercom = intercom_client

    async def handle_message(
        self,
        conversation_id: str,
        message: str,
        user: dict
    ) -> dict:
        """
        Traite un message utilisateur et génère une réponse.
        """
        # 1. Récupérer l'historique de conversation
        history = await self._get_conversation_history(conversation_id)

        # 2. Construire le contexte complet
        context = await self._build_context(user, history, message)

        # 3. Déterminer l'intention
        intent = await self._classify_intent(message, history)

        # 4. Décider de l'action
        if intent["action"] == "escalate":
            return await self._escalate_to_human(conversation_id, intent)

        if intent["action"] == "collect_info":
            return await self._ask_clarification(conversation_id, intent)

        # 5. Rechercher dans la KB
        documents = await self.rag.search(
            query=self._build_search_query(message, history),
            filter=self._build_filter(user, intent),
            top_k=5
        )

        # 6. Générer la réponse
        response = await self._generate_response(
            message=message,
            history=history,
            documents=documents,
            user_context=context
        )

        # 7. Envoyer via Intercom
        await self._send_reply(conversation_id, response)

        return {"status": "replied", "response": response}

    async def _get_conversation_history(
        self,
        conversation_id: str
    ) -> list:
        """
        Récupère l'historique formaté de la conversation.
        """
        conversation = await self.intercom.get_conversation(conversation_id)

        history = []
        for part in conversation["conversation_parts"]["conversation_parts"]:
            role = "user" if part["author"]["type"] == "user" else "assistant"
            history.append({
                "role": role,
                "content": self._extract_text(part["body"]),
                "timestamp": part["created_at"]
            })

        return history

    async def _build_context(
        self,
        user: dict,
        history: list,
        current_message: str
    ) -> dict:
        """
        Construit le contexte utilisateur complet.
        """
        # Données utilisateur Intercom
        user_data = await self.intercom.get_user(user["id"])

        # Segments et tags
        segments = user_data.get("segments", [])
        tags = user_data.get("tags", [])

        # Custom attributes
        custom_attrs = user_data.get("custom_attributes", {})

        # Conversations précédentes
        previous_convos = await self._get_previous_conversations(user["id"])

        return {
            "user": {
                "name": user_data.get("name"),
                "email": user_data.get("email"),
                "created_at": user_data.get("created_at"),
                "last_seen_at": user_data.get("last_seen_at"),
                "session_count": user_data.get("session_count"),
                "segments": segments,
                "tags": tags,
                "custom_attributes": custom_attrs
            },
            "history_summary": self._summarize_history(history),
            "previous_issues": self._extract_previous_issues(previous_convos),
            "current_message": current_message
        }

Classification d'intention

DEVELOPERpython
class IntentClassifier:
    """
    Classifie l'intention du message utilisateur.
    """

    INTENTS = {
        "question_faq": {"action": "answer", "priority": "low"},
        "question_technique": {"action": "answer", "priority": "medium"},
        "probleme_urgent": {"action": "escalate", "priority": "high"},
        "demande_humain": {"action": "escalate", "priority": "high"},
        "feedback": {"action": "collect", "priority": "low"},
        "information_incomplete": {"action": "collect_info", "priority": "medium"}
    }

    async def classify(
        self,
        message: str,
        history: list
    ) -> dict:
        """
        Classifie l'intention avec contexte conversationnel.
        """
        prompt = f"""
        Analyse ce message dans le contexte de la conversation.

        Historique (derniers 5 messages):
        {self._format_history(history[-5:])}

        Message actuel: {message}

        Classifie l'intention parmi:
        - question_faq: Question courante avec réponse dans la KB
        - question_technique: Question technique nécessitant documentation
        - probleme_urgent: Problème bloquant nécessitant intervention humaine
        - demande_humain: L'utilisateur demande explicitement un humain
        - feedback: Retour d'expérience ou suggestion
        - information_incomplete: Besoin de plus d'informations pour répondre

        Réponds en JSON: {{"intent": "...", "confidence": 0.X, "reason": "..."}}
        """

        result = await self.llm.generate(prompt, temperature=0.1)
        parsed = json.loads(result)

        intent_config = self.INTENTS.get(parsed["intent"], {"action": "answer"})

        return {
            "intent": parsed["intent"],
            "confidence": parsed["confidence"],
            "reason": parsed["reason"],
            **intent_config
        }

Synchronisation Help Center Intercom

Import des articles

DEVELOPERpython
class IntercomKBSync:
    """
    Synchronise le Help Center Intercom avec le RAG.
    """

    def __init__(self, access_token: str, rag_client):
        self.base_url = "https://api.intercom.io"
        self.headers = {
            "Authorization": f"Bearer {access_token}",
            "Accept": "application/json",
            "Intercom-Version": "2.10"
        }
        self.rag = rag_client

    async def sync_all_articles(self) -> dict:
        """
        Synchronise tous les articles publiés.
        """
        stats = {"synced": 0, "skipped": 0, "errors": 0}

        async with httpx.AsyncClient() as client:
            # Récupérer toutes les collections
            collections = await self._get_collections(client)

            for collection in collections:
                # Récupérer les sections de chaque collection
                sections = await self._get_sections(client, collection["id"])

                for section in sections:
                    # Récupérer les articles
                    articles = await self._get_articles(client, section["id"])

                    for article in articles:
                        if article["state"] == "published":
                            try:
                                await self._index_article(
                                    article, collection, section
                                )
                                stats["synced"] += 1
                            except Exception as e:
                                stats["errors"] += 1
                        else:
                            stats["skipped"] += 1

        return stats

    async def _index_article(
        self,
        article: dict,
        collection: dict,
        section: dict
    ):
        """
        Indexe un article dans le système RAG.
        """
        # Extraire le contenu textuel du HTML
        clean_content = self._html_to_text(article["body"])

        # Métadonnées riches pour le filtrage
        metadata = {
            "source": "intercom_help_center",
            "article_id": article["id"],
            "title": article["title"],
            "description": article.get("description", ""),
            "url": article["url"],
            "collection": collection["name"],
            "collection_id": collection["id"],
            "section": section["name"],
            "section_id": section["id"],
            "author_id": article.get("author_id"),
            "created_at": article["created_at"],
            "updated_at": article["updated_at"],
            "parent_ids": article.get("parent_ids", [])
        }

        # Indexer avec chunking si article long
        if len(clean_content) > 2000:
            await self._index_with_chunks(
                content=f"# {article['title']}\n\n{clean_content}",
                metadata=metadata,
                doc_id=f"intercom_article_{article['id']}"
            )
        else:
            await self.rag.index_document(
                content=f"# {article['title']}\n\n{clean_content}",
                metadata=metadata,
                doc_id=f"intercom_article_{article['id']}"
            )

    async def _get_collections(self, client: httpx.AsyncClient) -> list:
        response = await client.get(
            f"{self.base_url}/help_center/collections",
            headers=self.headers
        )
        return response.json().get("data", [])

    async def _get_sections(
        self,
        client: httpx.AsyncClient,
        collection_id: str
    ) -> list:
        response = await client.get(
            f"{self.base_url}/help_center/collections/{collection_id}/sections",
            headers=self.headers
        )
        return response.json().get("data", [])

    async def _get_articles(
        self,
        client: httpx.AsyncClient,
        section_id: str
    ) -> list:
        response = await client.get(
            f"{self.base_url}/help_center/sections/{section_id}/articles",
            headers=self.headers
        )
        return response.json().get("data", [])

Génération de réponses conversationnelles

Template de prompt conversationnel

DEVELOPERpython
INTERCOM_RAG_PROMPT = """Tu es l'assistant virtuel de {company_name} dans Intercom.

CONTEXTE UTILISATEUR:
- Nom: {user_name}
- Client depuis: {user_since}
- Segment: {user_segment}
- Attributs: {custom_attributes}

HISTORIQUE DE CONVERSATION:
{conversation_history}

DOCUMENTS PERTINENTS:
{retrieved_documents}

MESSAGE UTILISATEUR:
{user_message}

INSTRUCTIONS:
1. Réponds de manière conversationnelle et naturelle
2. Base ta réponse UNIQUEMENT sur les documents fournis
3. Si tu ne peux pas répondre, propose de transférer à un humain
4. Personnalise selon le contexte utilisateur
5. Utilise le prénom si disponible
6. Propose des liens vers les articles pertinents
7. Si le problème semble complexe, demande des précisions

STYLE:
- Ton: {tone} (amical/professionnel)
- Longueur: Concise mais complète
- Format: Paragraphes courts, listes si nécessaire

RÉPONSE:
"""

async def generate_conversational_response(
    user: dict,
    history: list,
    documents: list,
    message: str,
    config: dict
) -> str:
    """
    Génère une réponse conversationnelle adaptée au contexte.
    """
    prompt = INTERCOM_RAG_PROMPT.format(
        company_name=config.get("company_name", "Notre entreprise"),
        user_name=user.get("name", ""),
        user_since=format_date(user.get("created_at")),
        user_segment=", ".join(user.get("segments", [])),
        custom_attributes=json.dumps(user.get("custom_attributes", {})),
        conversation_history=format_history(history),
        retrieved_documents=format_documents(documents),
        user_message=message,
        tone=config.get("tone", "amical")
    )

    response = await llm.generate(prompt, temperature=0.4)

    # Post-processing pour le format Intercom
    return format_for_intercom(response, documents)

Formatage pour Intercom

DEVELOPERpython
def format_for_intercom(response: str, documents: list) -> dict:
    """
    Formate la réponse pour l'API Intercom Messenger.
    """
    # Construire les composants de la réponse
    components = []

    # Texte principal
    components.append({
        "type": "text",
        "text": response
    })

    # Ajouter des liens vers les articles si pertinents
    if documents:
        article_links = []
        for doc in documents[:3]:
            if doc.score > 0.7:
                article_links.append({
                    "title": doc.metadata["title"],
                    "url": doc.metadata["url"]
                })

        if article_links:
            components.append({
                "type": "text",
                "text": "\n\nArticles utiles :"
            })
            for link in article_links:
                components.append({
                    "type": "button",
                    "label": link["title"],
                    "action": {
                        "type": "url",
                        "url": link["url"]
                    }
                })

    return {"components": components}

Escalade vers humain

Détection d'escalade nécessaire

DEVELOPERpython
class EscalationManager:
    """
    Gère l'escalade vers les agents humains.
    """

    ESCALATION_TRIGGERS = [
        "parler à quelqu'un",
        "agent humain",
        "vrai personne",
        "ça ne m'aide pas",
        "problème urgent",
        "remboursement",
        "annuler mon compte"
    ]

    async def should_escalate(
        self,
        message: str,
        history: list,
        rag_confidence: float
    ) -> tuple[bool, str]:
        """
        Détermine si une escalade est nécessaire.
        """
        # 1. Déclencheurs explicites
        for trigger in self.ESCALATION_TRIGGERS:
            if trigger.lower() in message.lower():
                return True, f"trigger_explicit: {trigger}"

        # 2. Confiance RAG trop basse
        if rag_confidence < 0.4:
            return True, "low_confidence"

        # 3. Conversation trop longue sans résolution
        if len(history) > 10:
            return True, "conversation_too_long"

        # 4. Sentiment négatif répété
        if await self._detect_frustration(history):
            return True, "user_frustrated"

        return False, None

    async def escalate(
        self,
        conversation_id: str,
        reason: str,
        context: dict
    ) -> dict:
        """
        Effectue l'escalade vers un humain.
        """
        # Préparer le contexte pour l'agent
        handoff_note = self._prepare_handoff_note(context, reason)

        # Assigner à la bonne équipe selon le contexte
        team = self._determine_team(context, reason)

        # Via l'API Intercom
        await self.intercom.assign_conversation(
            conversation_id=conversation_id,
            assignee_id=team["inbox_id"],
            body=handoff_note
        )

        # Message de transition pour l'utilisateur
        transition_message = self._get_transition_message(team, reason)
        await self.intercom.reply(
            conversation_id=conversation_id,
            body=transition_message,
            message_type="comment"
        )

        return {
            "escalated": True,
            "team": team["name"],
            "reason": reason
        }

    def _prepare_handoff_note(self, context: dict, reason: str) -> str:
        """
        Prépare une note de contexte pour l'agent humain.
        """
        return f"""
        === Contexte du transfert ===

        Raison: {reason}

        Résumé de la conversation:
        {context.get('history_summary', 'Non disponible')}

        Dernière question:
        {context.get('last_message', '')}

        Documents consultés:
        {self._format_consulted_docs(context.get('documents', []))}

        Profil utilisateur:
        - Segments: {', '.join(context.get('user', {}).get('segments', []))}
        - Client depuis: {context.get('user', {}).get('created_at', 'N/A')}
        """

Métriques et optimisation

Tracking des conversations

DEVELOPERpython
class ConversationMetrics:
    """
    Collecte les métriques de performance du chatbot.
    """

    async def track_conversation(self, conversation_id: str) -> dict:
        """
        Analyse une conversation terminée.
        """
        conversation = await self.intercom.get_conversation(conversation_id)

        parts = conversation["conversation_parts"]["conversation_parts"]

        metrics = {
            "conversation_id": conversation_id,
            "total_messages": len(parts),
            "bot_messages": len([p for p in parts if p["author"]["type"] == "bot"]),
            "human_messages": len([p for p in parts if p["author"]["type"] == "admin"]),
            "user_messages": len([p for p in parts if p["author"]["type"] == "user"]),
            "escalated": any(p["author"]["type"] == "admin" for p in parts),
            "resolution_time": self._calculate_resolution_time(conversation),
            "csat_rating": conversation.get("sla_applied", {}).get("csat"),
            "first_response_time": self._calculate_first_response_time(conversation)
        }

        # Taux de résolution automatique
        if not metrics["escalated"]:
            metrics["auto_resolved"] = True
            metrics["auto_resolution_confidence"] = await self._get_avg_confidence(
                conversation_id
            )
        else:
            metrics["auto_resolved"] = False
            metrics["escalation_point"] = self._find_escalation_point(parts)

        return metrics

Intégration avec Ailog

Ailog simplifie l'intégration Intercom :

DEVELOPERpython
from ailog import AilogClient

client = AilogClient(api_key="your-key")

# Configuration complète
client.integrations.intercom.connect(
    access_token="intercom-token",
    sync_help_center=True,
    enable_messenger_bot=True,
    escalation_rules={
        "low_confidence_threshold": 0.4,
        "max_turns_before_escalate": 10,
        "sentiment_monitoring": True
    }
)

Conclusion

L'intégration RAG + Intercom transforme votre chatbot en assistant véritablement intelligent. Les conversations deviennent naturelles, contextuelles et efficaces. Commencez par la synchronisation du Help Center, déployez le bot en mode suggestion aux agents, puis activez l'automatisation complète une fois les métriques validées.

Ressources complémentaires

• Escalade intelligente - Quand transférer à un humain
• Zendesk + RAG - Alternative Zendesk
• Classification automatique des tickets - Routage intelligent
• RAG pour le support client - Guide pilier