Intercom + RAG: Next-Generation Support Chatbot

TL;DR

RAG + Intercom integration creates a chatbot capable of natural, contextual conversations powered by your knowledge base. Unlike rigid scenario-based chatbots, it understands complex questions and provides personalized responses. This guide covers conversational architecture, Intercom API integration, and strategies to maximize automation while preserving the human experience.

Why RAG + Intercom?

Limitations of Classic Intercom Bots

Intercom's Resolution Bots and Custom Bots work on decision trees:

Benefits of RAG Chatbot

An Intercom chatbot augmented with RAG offers:

• Natural understanding: No need for exact keywords
• Contextual responses: Based on conversation history
• Automatic updates: Synchronized with your Help Center articles
• Personalization: Adapted to user profile and history
• Intelligent escalation: Relevant transfer when necessary

RAG Chatbot Architecture

Overview

┌─────────────────────────────────────────────────────────────┐
│                     Intercom Messenger                       │
└─────────────────────────┬───────────────────────────────────┘
                          │ Webhook
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                    RAG Middleware                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │
│  │    User     │  │   Vector    │  │  LLM Generation     │ │
│  │   Context   │──│  Retrieval  │──│  + Instructions     │ │
│  └─────────────┘  └─────────────┘  └─────────────────────┘ │
└─────────────────────────┬───────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────────┐
│              Intercom API (Reply/Assign)                     │
└─────────────────────────────────────────────────────────────┘

Integration 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:
    """Verifies Intercom webhook signature."""
    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):
    """
    Entry point for all 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":
        # New conversation
        return await handle_new_conversation(payload["data"])
    elif topic == "conversation.user.replied":
        # User message
        return await handle_user_message(payload["data"])
    elif topic == "conversation_part.tag.created":
        # Tag added (for routing)
        return await handle_tag_added(payload["data"])

    return {"status": "ignored"}

Conversation Management

User Message Handler

DEVELOPERpython
class ConversationHandler:
    """
    Manages Intercom conversations with 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:
        """
        Processes a user message and generates a response.
        """
        # 1. Get conversation history
        history = await self._get_conversation_history(conversation_id)

        # 2. Build complete context
        context = await self._build_context(user, history, message)

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

        # 4. Decide 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. Search KB
        documents = await self.rag.search(
            query=self._build_search_query(message, history),
            filter=self._build_filter(user, intent),
            top_k=5
        )

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

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

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

    async def _get_conversation_history(
        self,
        conversation_id: str
    ) -> list:
        """
        Retrieves formatted conversation history.
        """
        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:
        """
        Builds complete user context.
        """
        # Intercom user data
        user_data = await self.intercom.get_user(user["id"])

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

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

        # Previous conversations
        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
        }

Intent Classification

DEVELOPERpython
class IntentClassifier:
    """
    Classifies user message intent.
    """

    INTENTS = {
        "faq_question": {"action": "answer", "priority": "low"},
        "technical_question": {"action": "answer", "priority": "medium"},
        "urgent_problem": {"action": "escalate", "priority": "high"},
        "human_request": {"action": "escalate", "priority": "high"},
        "feedback": {"action": "collect", "priority": "low"},
        "incomplete_info": {"action": "collect_info", "priority": "medium"}
    }

    async def classify(
        self,
        message: str,
        history: list
    ) -> dict:
        """
        Classifies intent with conversational context.
        """
        prompt = f"""
        Analyze this message in the conversation context.

        History (last 5 messages):
        {self._format_history(history[-5:])}

        Current message: {message}

        Classify intent among:
        - faq_question: Common question with KB answer
        - technical_question: Technical question requiring documentation
        - urgent_problem: Blocking problem requiring human intervention
        - human_request: User explicitly requests a human
        - feedback: Experience feedback or suggestion
        - incomplete_info: Need more information to answer

        Respond 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
        }

Intercom Help Center Synchronization

Importing Articles

DEVELOPERpython
class IntercomKBSync:
    """
    Synchronizes Intercom Help Center with 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:
        """
        Synchronizes all published articles.
        """
        stats = {"synced": 0, "skipped": 0, "errors": 0}

        async with httpx.AsyncClient() as client:
            # Get all collections
            collections = await self._get_collections(client)

            for collection in collections:
                # Get sections for each collection
                sections = await self._get_sections(client, collection["id"])

                for section in sections:
                    # Get 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
    ):
        """
        Indexes an article in the RAG system.
        """
        # Extract text content from HTML
        clean_content = self._html_to_text(article["body"])

        # Rich metadata for filtering
        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", [])
        }

        # Index with chunking if article is 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", [])

Conversational Response Generation

Conversational Prompt Template

DEVELOPERpython
INTERCOM_RAG_PROMPT = """You are the virtual assistant for {company_name} in Intercom.

USER CONTEXT:
- Name: {user_name}
- Customer since: {user_since}
- Segment: {user_segment}
- Attributes: {custom_attributes}

CONVERSATION HISTORY:
{conversation_history}

RELEVANT DOCUMENTS:
{retrieved_documents}

USER MESSAGE:
{user_message}

INSTRUCTIONS:
1. Respond in a conversational and natural way
2. Base your response ONLY on the provided documents
3. If you can't answer, offer to transfer to a human
4. Personalize based on user context
5. Use first name if available
6. Suggest links to relevant articles
7. If the problem seems complex, ask for clarification

STYLE:
- Tone: {tone} (friendly/professional)
- Length: Concise but complete
- Format: Short paragraphs, lists if necessary

RESPONSE:
"""

async def generate_conversational_response(
    user: dict,
    history: list,
    documents: list,
    message: str,
    config: dict
) -> str:
    """
    Generates a conversational response adapted to context.
    """
    prompt = INTERCOM_RAG_PROMPT.format(
        company_name=config.get("company_name", "Our company"),
        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", "friendly")
    )

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

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

Formatting for Intercom

DEVELOPERpython
def format_for_intercom(response: str, documents: list) -> dict:
    """
    Formats response for Intercom Messenger API.
    """
    # Build response components
    components = []

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

    # Add article links if relevant
    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\nHelpful articles:"
            })
            for link in article_links:
                components.append({
                    "type": "button",
                    "label": link["title"],
                    "action": {
                        "type": "url",
                        "url": link["url"]
                    }
                })

    return {"components": components}

Escalation to Human

Detecting Required Escalation

DEVELOPERpython
class EscalationManager:
    """
    Manages escalation to human agents.
    """

    ESCALATION_TRIGGERS = [
        "talk to someone",
        "human agent",
        "real person",
        "this doesn't help",
        "urgent problem",
        "refund",
        "cancel my account"
    ]

    async def should_escalate(
        self,
        message: str,
        history: list,
        rag_confidence: float
    ) -> tuple[bool, str]:
        """
        Determines if escalation is needed.
        """
        # 1. Explicit triggers
        for trigger in self.ESCALATION_TRIGGERS:
            if trigger.lower() in message.lower():
                return True, f"trigger_explicit: {trigger}"

        # 2. RAG confidence too low
        if rag_confidence < 0.4:
            return True, "low_confidence"

        # 3. Conversation too long without resolution
        if len(history) > 10:
            return True, "conversation_too_long"

        # 4. Repeated negative 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:
        """
        Performs escalation to a human.
        """
        # Prepare context for agent
        handoff_note = self._prepare_handoff_note(context, reason)

        # Assign to right team based on context
        team = self._determine_team(context, reason)

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

        # Transition message for user
        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:
        """
        Prepares a context note for the human agent.
        """
        return f"""
        === Transfer Context ===

        Reason: {reason}

        Conversation summary:
        {context.get('history_summary', 'Not available')}

        Last question:
        {context.get('last_message', '')}

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

        User profile:
        - Segments: {', '.join(context.get('user', {}).get('segments', []))}
        - Customer since: {context.get('user', {}).get('created_at', 'N/A')}
        """

Metrics and Optimization

Conversation Tracking

DEVELOPERpython
class ConversationMetrics:
    """
    Collects chatbot performance metrics.
    """

    async def track_conversation(self, conversation_id: str) -> dict:
        """
        Analyzes a completed conversation.
        """
        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)
        }

        # Auto-resolution rate
        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 with Ailog

Ailog simplifies Intercom integration:

DEVELOPERpython
from ailog import AilogClient

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

# Complete configuration
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

RAG + Intercom integration transforms your chatbot into a truly intelligent assistant. Conversations become natural, contextual, and effective. Start with Help Center synchronization, deploy the bot in agent suggestion mode, then enable full automation once metrics are validated.

Additional Resources

• Intelligent Escalation - When to transfer to a human
• Zendesk + RAG - Zendesk alternative
• Automatic Ticket Classification - Intelligent routing
• RAG for Customer Support - Pillar guide

---

Ready for a next-generation Intercom chatbot? Try Ailog - Intercom integration in a few clicks, intelligent conversations from day one.