Slack Bot RAG : Intelligente Recherche in Ihren Konversationen

Slack ist zum zentralen Nervensystem vieler Unternehmen geworden. Tausende täglich ausgetauschte Nachrichten enthalten Entscheidungen, technische Lösungen, Projektkontexte und implizites Wissen. Doch dieser Reichtum bleibt weitgehend unzugänglich: die native Slack-Suche versteht den Kontext nicht und überfordert Benutzer mit irrelevanten Ergebnissen.

Ein RAG-Bot, der an Slack angebunden ist, verwandelt Ihr Gesprächsarchiv in einen intelligenten Assistenten. Er versteht Ihre Fragen in natürlicher Sprache, findet relevante Diskussionen und synthetisiert Antworten mit Links zu den Originalnachrichten.

Warum Slack mit RAG indexieren

Das Problem des Unternehmensgedächtnisses

Slack-Konversationen enthalten wertvolles, aber flüchtiges Wissen:

• Getroffene Entscheidungen: "Wir haben uns entschieden, Tech X zu verwenden, weil..."
• Gefundene Lösungen: "Ich habe den Bug gelöst, indem ich Y gemacht habe"
• Geschäftskontext: "Kunde Z bevorzugt, dass wir so vorgehen"
• Informelle Prozesse: "Normalerweise kontaktiert man dafür Marie"

Grenzen der Slack-Suche

Was RAG bringt

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):
        """
        Initialisiert den Slack-Connector.

        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]:
        """Gibt die Liste der zugänglichen Kanäle zurück."""
        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]:
        """
        Ruft die Nachrichten eines Channels ab.

        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']:
                # Systemnachrichten ignorieren
                if msg.get('subtype') in ['channel_join', 'channel_leave', 'bot_message']:
                    continue

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

                # Threads abrufen
                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]:
        """Ruft die Nachrichten eines Threads ab."""
        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:]:  # Parent-Nachricht überspringen
                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]:
        """Formatiert eine Slack-Nachricht für RAG."""
        text = msg.get('text', '')
        if not text or len(text) < 10:
            return None

        # Erwähnungen auflösen
        text = self._resolve_mentions(text)

        # Links auflösen
        text = self._resolve_links(text)

        # Benutzerinfo abrufen
        user_info = self._get_user_info(msg.get('user', ''))

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

        # Permalink erstellen
        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:
        """Ersetzt Benutzer-IDs durch Namen."""
        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:
        """Konvertiert Slack-Links in 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:
        """Cached und ruft Benutzerinformationen ab."""
        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:
        """Cached und ruft den Channel-Namen ab."""
        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]:
        """Formatiert die Reaktionen."""
        return [
            {'emoji': r['name'], 'count': r['count']}
            for r in reactions
        ]


class SlackThreadAggregator:
    """Aggregiert Threads zu zusammenhängenden Dokumenten."""

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

    def aggregate_threads(self, messages: List[dict]) -> List[dict]:
        """
        Gruppiert Thread-Nachrichten zu einzelnen Dokumenten.

        Ein kompletter Thread wird zu einem einzigen Dokument für besseren Kontext.
        """
        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)

        # Jeden Thread zusammenfassen
        aggregated = []
        for thread_ts, thread_msgs in threads.items():
            # Nach Timestamp sortieren
            thread_msgs.sort(key=lambda m: m['metadata']['message_ts'])

            # Aggregierten Inhalt erstellen
            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

Interaktiver Slack-Bot

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
    ):
        """
        Initialisiert den RAG-Slack-Bot.

        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):
        """Startet den Bot im Socket-Modus."""
        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):
        """Verarbeitet Slack-Ereignisse."""
        if req.type == "events_api":
            event = req.payload.get("event", {})

            # Bot-Erwähnung
            if event.get("type") == "app_mention":
                self._handle_mention(event)

            # Direkte Nachricht
            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):
        """Verarbeitet @bot-Erwähnungen."""
        channel = event["channel"]
        thread_ts = event.get("thread_ts", event["ts"])
        text = event["text"]

        # Erwähnung bereinigen
        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

        # RAG-Suche
        response = self._generate_response(question)

        # Antworten
        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):
        """Verarbeitet Direktnachrichten."""
        channel = event["channel"]
        question = event["text"]

        if event.get("bot_id"):
            return  # Nachrichten vom Bot ignorieren

        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:
        """Erzeugt eine RAG-Antwort."""
        # Suche
        docs = self.retriever.search(question, top_k=5)

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

        # Generierung
        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:
        """Formatiert die Antwort in Slack-Blocks."""
        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

Anwendungsfälle

1. Eine Entscheidung wiederfinden

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

Reponse :

Laut den Diskussionen in #architecture (Januar 2025) hat das Team PostgreSQL gewählt wegen:

• Bedarf an ACID-Transaktionen für Zahlungen
• Team hat mehr Erfahrung mit SQL
• Bessere Integration in unseren Python-Stack

Quellen: #architecture | #tech-decisions

2. Eine technische Lösung finden

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

Reponse :

Thomas hat eine Lösung in #payments geteilt (März 2025):

1. Eine Queue mit Redis implementieren
2. Exponentielles Backoff verwenden
3. Häufige Antworten cachen

Quellen: #payments

3. Projektkontext wiederfinden

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

Reponse :

Projekt Alpha gestartet in Q4 2025:

• Ziel: Überarbeitung des Kunden-Dashboards
• PM: Marie (siehe #proj-alpha)
• Stack: React + GraphQL
• Deadline: Ende März 2026

Quellen: #proj-alpha | #announcements

Best Practices

Zu indexierende Kanäle

Umgang mit Vertraulichkeit

• DMs niemals ohne Zustimmung indexieren
• Private Channels respektieren (explizites Opt-in)
• Bei Bedarf anonymisieren
• Sensible Daten löschen

Weiterführende Ressourcen

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

Einen Slack-Bot mit Ailog bereitstellen

Verwandeln Sie Ihre Slack-Konversationen in eine Wissensdatenbank. Ailog bietet:

• 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.