Slack Bot RAG : Recherche intelligente dans vos conversations

Slack est devenu le systeme nerveux central de nombreuses entreprises. Des milliers de messages echanges chaque jour contiennent decisions, solutions techniques, contextes de projets et connaissances tacites. Mais cette richesse reste largement inaccessible : la recherche native de Slack ne comprend pas le contexte et noie les utilisateurs dans des resultats non pertinents.

Un bot RAG connecte a Slack transforme votre historique de conversations en assistant intelligent. Il comprend vos questions en langage naturel, retrouve les discussions pertinentes, et synthetise des reponses avec liens vers les messages originaux.

Pourquoi indexer Slack avec le RAG

Le probleme de la memoire d'entreprise

Les conversations Slack contiennent une connaissance precieuse mais ephemere :

• Decisions prises : "On a decide de partir sur la techno X parce que..."
• Solutions trouvees : "J'ai resolu le bug en faisant Y"
• Contexte metier : "Le client Z prefere qu'on procede ainsi"
• Processus informels : "En general, pour ca, on contacte Marie"

Limites de la recherche Slack

Ce qu'apporte le RAG

Architecture Slack + RAG

┌─────────────────────────────────────────────────────────────────────────┐
│                       Architecture Slack + RAG                           │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│   SLACK                       EXTRACTION                  PROCESSING    │
│   ┌──────────────┐           ┌──────────────┐          ┌────────────┐  │
│   │   Channels   │──────────▶│   Slack API  │─────────▶│  Parsing   │  │
│   │              │           │              │          │            │  │
│   │  - Public    │           │  /messages   │          │  Threads   │  │
│   │  - Private   │           │  /threads    │          │  Mentions  │  │
│   │  - DMs       │           │  /users      │          │  Links     │  │
│   └──────────────┘           └──────────────┘          └──────┬─────┘  │
│                                                               │        │
│   ┌──────────────┐                                     ┌──────┴─────┐  │
│   │   Events     │────────────────────────────────────▶│  Chunking  │  │
│   │   Socket     │                                     │  Threads   │  │
│   └──────────────┘                                     └──────┬─────┘  │
│                                                               │        │
│                                                        ┌──────┴─────┐  │
│   VECTOR DB                                            │ Embeddings │  │
│   ┌──────────────┐                                     │  BGE-M3    │  │
│   │    Qdrant    │◀────────────────────────────────────┴────────────┘  │
│   │              │                                                     │
│   │  HNSW Index  │                                                     │
│   └──────┬───────┘                                                     │
│          │                                                             │
│   SLACK BOT                                                            │
│   ┌──────┴───────┐     ┌─────────────┐     ┌──────────────┐           │
│   │   @ailog     │────▶│  Retrieval  │────▶│   Reranker   │           │
│   │   Question   │     │   Top-30    │     │    Top-5     │           │
│   └──────────────┘     └─────────────┘     └──────┬───────┘           │
│                                                    │                   │
│                        ┌──────────────┐    ┌──────┴───────┐           │
│                        │   Reponse    │◀───│     LLM      │           │
│                        │   Slack msg  │    │  GPT-4/Claude│           │
│                        └──────────────┘    └──────────────┘           │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

Connecteur Slack complet

DEVELOPERpython
from slack_sdk import WebClient
from slack_sdk.socket_mode import SocketModeClient
from slack_sdk.socket_mode.request import SocketModeRequest
from slack_sdk.socket_mode.response import SocketModeResponse
import hashlib
from datetime import datetime
from typing import List, Optional

class SlackConnector:
    def __init__(self, bot_token: str, user_token: str = None):
        """
        Initialise le connecteur Slack.

        Args:
            bot_token: Token du bot (xoxb-...)
            user_token: Token utilisateur pour l'historique complet (xoxp-...)
        """
        self.bot_client = WebClient(token=bot_token)
        self.user_client = WebClient(token=user_token) if user_token else None
        self.client = self.user_client or self.bot_client

    def get_channels(self, include_private: bool = False) -> List[dict]:
        """Recupere la liste des canaux accessibles."""
        channels = []
        cursor = None

        while True:
            result = self.client.conversations_list(
                types="public_channel,private_channel" if include_private else "public_channel",
                cursor=cursor,
                limit=200
            )

            for channel in result['channels']:
                channels.append({
                    'id': channel['id'],
                    'name': channel['name'],
                    'is_private': channel.get('is_private', False),
                    'member_count': channel.get('num_members', 0),
                    'topic': channel.get('topic', {}).get('value', ''),
                    'purpose': channel.get('purpose', {}).get('value', '')
                })

            cursor = result.get('response_metadata', {}).get('next_cursor')
            if not cursor:
                break

        return channels

    def get_channel_messages(
        self,
        channel_id: str,
        oldest: float = None,
        include_threads: bool = True
    ) -> List[dict]:
        """
        Recupere les messages d'un canal.

        Args:
            channel_id: ID du canal
            oldest: Timestamp minimum (pour sync incrementale)
            include_threads: Inclure les reponses de threads
        """
        messages = []
        cursor = None

        while True:
            params = {
                'channel': channel_id,
                'cursor': cursor,
                'limit': 200
            }
            if oldest:
                params['oldest'] = oldest

            result = self.client.conversations_history(**params)

            for msg in result['messages']:
                # Ignorer les messages systeme
                if msg.get('subtype') in ['channel_join', 'channel_leave', 'bot_message']:
                    continue

                formatted = self._format_message(msg, channel_id)
                if formatted:
                    messages.append(formatted)

                # Recuperer les threads
                if include_threads and msg.get('reply_count', 0) > 0:
                    thread_msgs = self._get_thread_messages(channel_id, msg['ts'])
                    messages.extend(thread_msgs)

            cursor = result.get('response_metadata', {}).get('next_cursor')
            if not cursor:
                break

        return messages

    def _get_thread_messages(self, channel_id: str, thread_ts: str) -> List[dict]:
        """Recupere les messages d'un thread."""
        messages = []
        cursor = None

        while True:
            result = self.client.conversations_replies(
                channel=channel_id,
                ts=thread_ts,
                cursor=cursor,
                limit=200
            )

            for msg in result['messages'][1:]:  # Skip le message parent
                formatted = self._format_message(msg, channel_id, thread_ts)
                if formatted:
                    messages.append(formatted)

            cursor = result.get('response_metadata', {}).get('next_cursor')
            if not cursor:
                break

        return messages

    def _format_message(
        self,
        msg: dict,
        channel_id: str,
        thread_ts: str = None
    ) -> Optional[dict]:
        """Formate un message Slack pour le RAG."""
        text = msg.get('text', '')
        if not text or len(text) < 10:
            return None

        # Resoudre les mentions
        text = self._resolve_mentions(text)

        # Resoudre les liens
        text = self._resolve_links(text)

        # Obtenir info utilisateur
        user_info = self._get_user_info(msg.get('user', ''))

        # Generer ID unique
        msg_id = f"slack_{channel_id}_{msg['ts']}"
        content_hash = hashlib.md5(text.encode()).hexdigest()

        # Construire le permalink
        permalink = f"https://slack.com/archives/{channel_id}/p{msg['ts'].replace('.', '')}"

        return {
            'id': msg_id,
            'title': f"Message de {user_info['name']} dans #{self._get_channel_name(channel_id)}",
            'content': text,
            'metadata': {
                'source': 'slack',
                'source_type': 'message',
                'channel_id': channel_id,
                'channel_name': self._get_channel_name(channel_id),
                'thread_ts': thread_ts,
                'message_ts': msg['ts'],
                'user_id': msg.get('user'),
                'user_name': user_info['name'],
                'user_display_name': user_info['display_name'],
                'timestamp': datetime.fromtimestamp(float(msg['ts'])).isoformat(),
                'permalink': permalink,
                'reactions': self._format_reactions(msg.get('reactions', [])),
                'has_attachments': len(msg.get('files', [])) > 0,
                'content_hash': content_hash
            }
        }

    def _resolve_mentions(self, text: str) -> str:
        """Remplace les IDs utilisateur par les noms."""
        import re
        mentions = re.findall(r'<@([A-Z0-9]+)>', text)
        for user_id in mentions:
            user_info = self._get_user_info(user_id)
            text = text.replace(f'<@{user_id}>', f'@{user_info["display_name"]}')
        return text

    def _resolve_links(self, text: str) -> str:
        """Convertit les liens Slack en Markdown."""
        import re
        # <url|label> -> [label](url)
        text = re.sub(r'<(https?://[^|>]+)\|([^>]+)>', r'[\2](\1)', text)
        # <url> -> url
        text = re.sub(r'<(https?://[^>]+)>', r'\1', text)
        return text

    def _get_user_info(self, user_id: str) -> dict:
        """Cache et recupere les infos utilisateur."""
        if not hasattr(self, '_user_cache'):
            self._user_cache = {}

        if user_id not in self._user_cache:
            try:
                result = self.client.users_info(user=user_id)
                user = result['user']
                self._user_cache[user_id] = {
                    'name': user.get('real_name', user.get('name', 'Unknown')),
                    'display_name': user.get('profile', {}).get('display_name') or user.get('name', 'Unknown')
                }
            except:
                self._user_cache[user_id] = {'name': 'Unknown', 'display_name': 'Unknown'}

        return self._user_cache[user_id]

    def _get_channel_name(self, channel_id: str) -> str:
        """Cache et recupere le nom du canal."""
        if not hasattr(self, '_channel_cache'):
            self._channel_cache = {}

        if channel_id not in self._channel_cache:
            try:
                result = self.client.conversations_info(channel=channel_id)
                self._channel_cache[channel_id] = result['channel']['name']
            except:
                self._channel_cache[channel_id] = channel_id

        return self._channel_cache[channel_id]

    def _format_reactions(self, reactions: list) -> List[dict]:
        """Formate les reactions."""
        return [
            {'emoji': r['name'], 'count': r['count']}
            for r in reactions
        ]


class SlackThreadAggregator:
    """Agregue les threads en documents coherents."""

    def __init__(self, connector: SlackConnector):
        self.connector = connector

    def aggregate_threads(self, messages: List[dict]) -> List[dict]:
        """
        Regroupe les messages de threads en documents uniques.

        Un thread complet devient un seul document pour meilleur contexte.
        """
        threads = {}
        standalone = []

        for msg in messages:
            thread_ts = msg['metadata'].get('thread_ts')

            if thread_ts:
                if thread_ts not in threads:
                    threads[thread_ts] = []
                threads[thread_ts].append(msg)
            else:
                standalone.append(msg)

        # Agreger chaque thread
        aggregated = []
        for thread_ts, thread_msgs in threads.items():
            # Trier par timestamp
            thread_msgs.sort(key=lambda m: m['metadata']['message_ts'])

            # Construire le contenu agrege
            content_parts = []
            for msg in thread_msgs:
                user = msg['metadata']['user_display_name']
                content_parts.append(f"**{user}**: {msg['content']}")

            aggregated_doc = {
                'id': f"slack_thread_{thread_msgs[0]['metadata']['channel_id']}_{thread_ts}",
                'title': f"Thread dans #{thread_msgs[0]['metadata']['channel_name']}",
                'content': '\n\n'.join(content_parts),
                'metadata': {
                    'source': 'slack',
                    'source_type': 'thread',
                    'channel_id': thread_msgs[0]['metadata']['channel_id'],
                    'channel_name': thread_msgs[0]['metadata']['channel_name'],
                    'thread_ts': thread_ts,
                    'message_count': len(thread_msgs),
                    'participants': list(set(m['metadata']['user_display_name'] for m in thread_msgs)),
                    'permalink': thread_msgs[0]['metadata']['permalink'],
                    'timestamp': thread_msgs[0]['metadata']['timestamp']
                }
            }
            aggregated.append(aggregated_doc)

        return standalone + aggregated

Bot Slack interactif

DEVELOPERpython
from slack_sdk.socket_mode import SocketModeClient
from slack_sdk.socket_mode.request import SocketModeRequest
from slack_sdk.socket_mode.response import SocketModeResponse

class SlackRAGBot:
    def __init__(
        self,
        app_token: str,
        bot_token: str,
        retriever,
        llm
    ):
        """
        Initialise le bot RAG Slack.

        Args:
            app_token: Token Socket Mode (xapp-...)
            bot_token: Token bot (xoxb-...)
            retriever: Instance du retriever RAG
            llm: Instance du LLM
        """
        self.client = WebClient(token=bot_token)
        self.socket = SocketModeClient(
            app_token=app_token,
            web_client=self.client
        )
        self.retriever = retriever
        self.llm = llm

    def start(self):
        """Demarre le bot en mode socket."""
        self.socket.socket_mode_request_listeners.append(self._handle_request)
        self.socket.connect()
        print("Bot Slack connecte")

    def _handle_request(self, client: SocketModeClient, req: SocketModeRequest):
        """Gere les evenements Slack."""
        if req.type == "events_api":
            event = req.payload.get("event", {})

            # Mention du bot
            if event.get("type") == "app_mention":
                self._handle_mention(event)

            # Message direct
            elif event.get("type") == "message" and event.get("channel_type") == "im":
                self._handle_dm(event)

        # Acknowledge
        response = SocketModeResponse(envelope_id=req.envelope_id)
        client.send_socket_mode_response(response)

    def _handle_mention(self, event: dict):
        """Gere les mentions @bot."""
        channel = event["channel"]
        thread_ts = event.get("thread_ts", event["ts"])
        text = event["text"]

        # Nettoyer la mention
        import re
        question = re.sub(r'<@[A-Z0-9]+>\s*', '', text).strip()

        if not question:
            self.client.chat_postMessage(
                channel=channel,
                thread_ts=thread_ts,
                text="Comment puis-je vous aider ? Posez-moi une question sur nos conversations."
            )
            return

        # Recherche RAG
        response = self._generate_response(question)

        # Repondre
        self.client.chat_postMessage(
            channel=channel,
            thread_ts=thread_ts,
            blocks=self._format_response_blocks(response),
            text=response['answer']
        )

    def _handle_dm(self, event: dict):
        """Gere les messages directs."""
        channel = event["channel"]
        question = event["text"]

        if event.get("bot_id"):
            return  # Ignorer les messages du bot

        response = self._generate_response(question)

        self.client.chat_postMessage(
            channel=channel,
            blocks=self._format_response_blocks(response),
            text=response['answer']
        )

    def _generate_response(self, question: str) -> dict:
        """Genere une reponse RAG."""
        # Recherche
        docs = self.retriever.search(question, top_k=5)

        # Construction du contexte
        context = "\n\n---\n\n".join([
            f"**{doc['metadata']['channel_name']}** ({doc['metadata']['timestamp'][:10]}):\n{doc['content'][:500]}"
            for doc in docs
        ])

        # Generation
        prompt = f"""Tu es un assistant qui aide a retrouver des informations dans l'historique Slack.

Contexte des conversations:
{context}

Question: {question}

Reponds de maniere concise en citant les canaux sources. Si l'info n'est pas dans le contexte, dis-le."""

        answer = self.llm.generate(prompt)

        return {
            'answer': answer,
            'sources': docs
        }

    def _format_response_blocks(self, response: dict) -> list:
        """Formate la reponse en blocs Slack."""
        blocks = [
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": response['answer']}
            }
        ]

        if response['sources']:
            source_links = []
            for doc in response['sources'][:3]:
                channel = doc['metadata']['channel_name']
                link = doc['metadata']['permalink']
                source_links.append(f"<{link}|#{channel}>")

            blocks.append({
                "type": "context",
                "elements": [
                    {"type": "mrkdwn", "text": f"Sources: {' | '.join(source_links)}"}
                ]
            })

        return blocks

Cas d'usage

1. Retrouver une decision

Question : "@ailog Pourquoi on a choisi PostgreSQL plutot que MongoDB ?"

Reponse :

D'apres les discussions dans #architecture (janvier 2025), l'equipe a choisi PostgreSQL pour :

• Besoin de transactions ACID pour les paiements
• Equipe plus experimentee sur SQL
• Meilleure integration avec notre stack Python

Sources: #architecture | #tech-decisions

2. Trouver une solution technique

Question : "@ailog Comment on resout le probleme de rate limiting sur l'API Stripe ?"

Reponse :

Thomas a partage une solution dans #payments (mars 2025) :

1. Implementer un queue avec Redis
2. Utiliser exponential backoff
3. Cacher les reponses frequentes

Sources: #payments

3. Retrouver un contexte projet

Question : "@ailog Quel est le contexte du projet Alpha ?"

Reponse :

Projet Alpha lance en Q4 2025 :

• Objectif : Refonte du dashboard client
• PM : Marie (voir #proj-alpha)
• Stack : React + GraphQL
• Deadline : fin mars 2026

Sources: #proj-alpha | #announcements

Bonnes pratiques

Canaux a indexer

Gestion de la confidentialite

• Ne jamais indexer les DMs sans consentement
• Respecter les canaux prives (opt-in explicite)
• Anonymiser si necessaire
• Purger les donnees sensibles

Ressources complementaires

• Base de connaissances entreprise - Guide pilier
• Notion + RAG - Pour la documentation
• Confluence + RAG - Pour Atlassian
• Introduction au RAG - Fondamentaux

Deployer un bot Slack avec Ailog

Transformez vos conversations Slack en base de connaissances. Ailog offre :

• Connecteur Slack natif : Installation en 1 clic
• Indexation selective : Choisissez les canaux
• Bot interactif : @mention ou DM
• Respect confidentialite : Canaux prives opt-in
• Hebergement France : RGPD natif

Testez Ailog gratuitement et deployez votre bot Slack RAG en 10 minutes.