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

Architektur Slack + RAG

┌─────────────────────────────────────────────────────────────────────────┐
│                       Architektur 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   │           │
│   │   Frage      │     │   Top-30    │     │    Top-5     │           │
│   └──────────────┘     └─────────────┘     └──────┬───────┘           │
│                                                    │                   │
│                        ┌──────────────┐    ┌──────┴───────┐           │
│                        │   Antwort    │◀───│     LLM      │           │
│                        │   Slack-Msg  │    │  GPT-4/Claude│           │
│                        └──────────────┘    └──────────────┘           │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

Vollständiger Slack-Connector

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: Bot-Token (xoxb-...)
            user_token: Benutzer-Token für den vollständigen Verlauf (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: Kanal-ID
            oldest: Minimaler Timestamp (für inkrementelle Synchronisierung)
            include_threads: Thread-Antworten einschließen
        """
        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"Nachricht von {user_info['name']} in #{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 in #{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: Socket-Mode-Token (xapp-...)
            bot_token: Bot-Token (xoxb-...)
            retriever: Instanz des RAG-Retrievers
            llm: LLM-Instanz
        """
        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("Slack-Bot verbunden")

    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="Wie kann ich Ihnen helfen? Stellen Sie mir eine Frage zu unseren Konversationen."
            )
            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"""Du bist ein Assistent, der hilft, Informationen im Slack-Verlauf zu finden.

Konversationskontext:
{context}

Frage: {question}

Antworte präzise und nenne die Quellkanäle. Wenn die Information nicht im Kontext enthalten ist, sag es."""

        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"Quellen: {' | '.join(source_links)}"}
                ]
            })

        return blocks

Anwendungsfälle

1. Eine Entscheidung wiederfinden

Frage: "@ailog Warum haben wir uns für PostgreSQL statt MongoDB entschieden?"

Antwort:

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

Frage: "@ailog Wie lösen wir das Rate-Limiting-Problem bei der Stripe-API?"

Antwort:

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

Frage: "@ailog Was ist der Kontext des Projekts Alpha?"

Antwort:

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

• Wissensdatenbank für Unternehmen - Leitfaden-Übersicht
• Notion + RAG - Für die Dokumentation
• Confluence + RAG - Für Atlassian
• Einführung in RAG - Grundlagen

Einen Slack-Bot mit Ailog bereitstellen

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

• Nativer Slack-Connector: Installation mit einem Klick
• Selektive Indexierung: Wählen Sie die Kanäle aus
• Interaktiver Bot: @Erwähnung oder DM
• Vertraulichkeit gewahrt: Private Kanäle per Opt-in
• Hosting in Frankreich: DSGVO-nativ

Testen Sie Ailog kostenlos und stellen Sie Ihren Slack-RAG-Bot in 10 Minuten bereit.