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.

Warum RAG + Intercom?

Die Grenzen klassischer Intercom-Bots

Die Resolution Bots und Custom Bots von Intercom funktionieren auf Basis von Entscheidungsbäumen:

Vorteile des RAG-Chatbots

Ein durch RAG erweiterter Intercom-Chatbot bietet:

• Natürliches Verständnis: Keine exakten Schlüsselwörter erforderlich
• Kontextbezogene Antworten: Basierend auf dem Gesprächsverlauf
• Automatische Aktualisierung: Synchronisiert mit Ihren Help-Center-Artikeln
• Personalisierung: Angepasst an das Nutzerprofil und dessen Verlauf
• Intelligente Eskalation: Gezielte Weiterleitung bei Bedarf

Architektur des RAG-Chatbots

Übersicht

┌─────────────────────────────────────────────────────────────┐
│                     Intercom Messenger                       │
└─────────────────────────┬───────────────────────────────────┘
                          │ Webhook
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                    RAG Middleware                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │
│  │  Kontext    │  │  Retrieval  │  │  LLM-Generierung    │ │
│  │  Nutzer     │──│  Vektor     │──│  + Anweisungen      │ │
│  └─────────────┘  └─────────────┘  └─────────────────────┘ │
└─────────────────────────┬───────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────────┐
│              Intercom API (Reply/Assign)                     │
└─────────────────────────────────────────────────────────────┘

Integration über 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 Unterhaltung
        return await handle_new_conversation(payload["data"])
    elif topic == "conversation.user.replied":
        # Nachricht des Nutzers
        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"}

Verwaltung der Unterhaltungen

Handler für Nutzernachrichten

DEVELOPERpython
class ConversationHandler:
    """
    Verwaltet Intercom-Unterhaltungen mit 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:
        """
        Verarbeitet eine Nutzernachricht und generiert eine Antwort.
        """
        # 1. Gesprächsverlauf abrufen
        history = await self._get_conversation_history(conversation_id)

        # 2. Vollständigen Kontext aufbauen
        context = await self._build_context(user, history, message)

        # 3. Absicht bestimmen
        intent = await self._classify_intent(message, history)

        # 4. Aktion entscheiden
        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. In der Wissensdatenbank suchen
        documents = await self.rag.search(
            query=self._build_search_query(message, history),
            filter=self._build_filter(user, intent),
            top_k=5
        )

        # 6. Antwort generieren
        response = await self._generate_response(
            message=message,
            history=history,
            documents=documents,
            user_context=context
        )

        # 7. Über Intercom senden
        await self._send_reply(conversation_id, response)

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

    async def _get_conversation_history(
        self,
        conversation_id: str
    ) -> list:
        """
        Ruft den formatierten Gesprächsverlauf ab.
        """
        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:
        """
        Baut den vollständigen Nutzerkontext auf.
        """
        # Intercom-Nutzerdaten
        user_data = await self.intercom.get_user(user["id"])

        # Segmente und Tags
        segments = user_data.get("segments", [])
        tags = user_data.get("tags", [])

        # Benutzerdefinierte Attribute
        custom_attrs = user_data.get("custom_attributes", {})

        # Vorherige Unterhaltungen
        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
        }

Absichtsklassifizierung

DEVELOPERpython
class IntentClassifier:
    """
    Klassifiziert die Absicht der Nutzernachricht.
    """

    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:
        """
        Klassifiziert die Absicht unter Berücksichtigung des Gesprächskontexts.
        """
        prompt = f"""
        Analysiere diese Nachricht im Kontext der Unterhaltung.

        Verlauf (letzte 5 Nachrichten):
        {self._format_history(history[-5:])}

        Aktuelle Nachricht: {message}

        Klassifiziere die Absicht als eine von:
        - question_faq: Häufige Frage mit Antwort in der Wissensdatenbank
        - question_technique: Technische Frage, die Dokumentation erfordert
        - probleme_urgent: Blockierendes Problem, das menschliches Eingreifen erfordert
        - demande_humain: Der Nutzer verlangt explizit einen Menschen
        - feedback: Erfahrungsbericht oder Vorschlag
        - information_incomplete: Weitere Informationen zur Beantwortung nötig

        Antworte in 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
        }

Synchronisierung des Intercom Help Centers

Import der Artikel

DEVELOPERpython
class IntercomKBSync:
    """
    Synchronisiert das Intercom Help Center mit dem RAG-System.
    """

    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.14"
        }
        self.rag = rag_client

    async def sync_all_articles(self) -> dict:
        """
        Synchronisiert alle veröffentlichten Artikel.
        """
        stats = {"synced": 0, "skipped": 0, "errors": 0}

        async with httpx.AsyncClient() as client:
            # Alle Collections abrufen
            collections = await self._get_collections(client)

            for collection in collections:
                # Sections jeder Collection abrufen
                sections = await self._get_sections(client, collection["id"])

                for section in sections:
                    # Artikel abrufen
                    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
    ):
        """
        Indexiert einen Artikel im RAG-System.
        """
        # Textinhalt aus dem HTML extrahieren
        clean_content = self._html_to_text(article["body"])

        # Umfangreiche Metadaten für das Filtern
        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", [])
        }

        # Mit Chunking indexieren, falls Artikel lang ist
        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", [])

Generierung konversationeller Antworten

Konversationelle Prompt-Vorlage

DEVELOPERpython
INTERCOM_RAG_PROMPT = """Du bist der virtuelle Assistent von {company_name} in Intercom.

NUTZERKONTEXT:
- Name: {user_name}
- Kunde seit: {user_since}
- Segment: {user_segment}
- Attribute: {custom_attributes}

GESPRÄCHSVERLAUF:
{conversation_history}

RELEVANTE DOKUMENTE:
{retrieved_documents}

NUTZERNACHRICHT:
{user_message}

ANWEISUNGEN:
1. Antworte konversationell und natürlich
2. Stütze deine Antwort AUSSCHLIESSLICH auf die bereitgestellten Dokumente
3. Wenn du nicht antworten kannst, schlage eine Weiterleitung an einen Menschen vor
4. Personalisiere entsprechend dem Nutzerkontext
5. Verwende den Vornamen, falls verfügbar
6. Schlage Links zu relevanten Artikeln vor
7. Bei komplexem Problem frage nach Präzisierungen

STIL:
- Ton: {tone} (freundlich/professionell)
- Länge: Prägnant, aber vollständig
- Format: Kurze Absätze, Listen bei Bedarf

ANTWORT:
"""

async def generate_conversational_response(
    user: dict,
    history: list,
    documents: list,
    message: str,
    config: dict
) -> str:
    """
    Generiert eine an den Kontext angepasste konversationelle Antwort.
    """
    prompt = INTERCOM_RAG_PROMPT.format(
        company_name=config.get("company_name", "Unser Unternehmen"),
        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)

    # Nachbearbeitung für das Intercom-Format
    return format_for_intercom(response, documents)

Formatierung für Intercom

DEVELOPERpython
def format_for_intercom(response: str, documents: list) -> dict:
    """
    Formatiert die Antwort für die Intercom Messenger API.
    """
    # Komponenten der Antwort aufbauen
    components = []

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

    # Links zu relevanten Artikeln hinzufügen
    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\nNützliche Artikel:"
            })
            for link in article_links:
                components.append({
                    "type": "button",
                    "label": link["title"],
                    "action": {
                        "type": "url",
                        "url": link["url"]
                    }
                })

    return {"components": components}

Eskalation an einen Menschen

Erkennung notwendiger Eskalation

DEVELOPERpython
class EscalationManager:
    """
    Verwaltet die Eskalation an menschliche Mitarbeiter.
    """

    ESCALATION_TRIGGERS = [
        "mit jemandem sprechen",
        "menschlicher mitarbeiter",
        "echte person",
        "das hilft mir nicht",
        "dringendes problem",
        "rückerstattung",
        "mein konto kündigen"
    ]

    async def should_escalate(
        self,
        message: str,
        history: list,
        rag_confidence: float
    ) -> tuple[bool, str]:
        """
        Bestimmt, ob eine Eskalation notwendig ist.
        """
        # 1. Explizite Auslöser
        for trigger in self.ESCALATION_TRIGGERS:
            if trigger.lower() in message.lower():
                return True, f"trigger_explicit: {trigger}"

        # 2. RAG-Konfidenz zu niedrig
        if rag_confidence < 0.4:
            return True, "low_confidence"

        # 3. Unterhaltung zu lang ohne Lösung
        if len(history) > 10:
            return True, "conversation_too_long"

        # 4. Wiederholt negatives Sentiment
        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:
        """
        Führt die Eskalation an einen Menschen durch.
        """
        # Kontext für den Agenten vorbereiten
        handoff_note = self._prepare_handoff_note(context, reason)

        # Dem passenden Team je nach Kontext zuweisen
        team = self._determine_team(context, reason)

        # Über die Intercom-API
        await self.intercom.assign_conversation(
            conversation_id=conversation_id,
            assignee_id=team["inbox_id"],
            body=handoff_note
        )

        # Übergangsnachricht für den Nutzer
        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:
        """
        Bereitet eine Kontextnotiz für den menschlichen Agenten vor.
        """
        return f"""
        === Kontext der Weiterleitung ===

        Grund: {reason}

        Zusammenfassung der Unterhaltung:
        {context.get('history_summary', 'Nicht verfügbar')}

        Letzte Frage:
        {context.get('last_message', '')}

        Konsultierte Dokumente:
        {self._format_consulted_docs(context.get('documents', []))}

        Nutzerprofil:
        - Segmente: {', '.join(context.get('user', {}).get('segments', []))}
        - Kunde seit: {context.get('user', {}).get('created_at', 'N/A')}
        """

Metriken und Optimierung

Tracking der Unterhaltungen

DEVELOPERpython
class ConversationMetrics:
    """
    Sammelt Performance-Metriken des Chatbots.
    """

    async def track_conversation(self, conversation_id: str) -> dict:
        """
        Analysiert eine abgeschlossene Unterhaltung.
        """
        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)
        }

        # Automatische Lösungsquote
        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

Integration mit Ailog

Ailog vereinfacht die Intercom-Integration:

DEVELOPERpython
from ailog import AilogClient

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

# Vollständige Konfiguration
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
    }
)

Fazit

Die RAG + Intercom-Integration verwandelt Ihren Chatbot in einen wirklich intelligenten Assistenten. Gespräche werden natürlich, kontextbezogen und effizient. Beginnen Sie mit der Help-Center-Synchronisation, stellen Sie den Bot im Vorschlagsmodus für Agenten bereit und aktivieren Sie dann die vollständige Automatisierung, sobald die Metriken validiert sind.

Ergänzende Ressourcen

• Intelligente Eskalation - Wann an einen Menschen weiterleiten
• Zendesk + RAG - Zendesk-Alternative
• Automatische Ticket-Klassifizierung - Intelligentes Routing
• RAG für den Kundensupport - Hauptleitfaden