Slack Bot to respond very quickly to acknowledge seeing the question (#544)

backend/danswer/bots/slack/constants.py

@@ -1,2 +1,3 @@
 LIKE_BLOCK_ACTION_ID = "feedback-like"
 DISLIKE_BLOCK_ACTION_ID = "feedback-dislike"
+SLACK_CHANNEL_ID = "channel_id"

backend/danswer/bots/slack/handlers/handle_message.py

@@ -1,4 +1,5 @@
 import logging
+from typing import cast
 
 from retry import retry
 from slack_sdk import WebClient
@@ -7,56 +8,60 @@ from sqlalchemy.orm import Session
 from danswer.bots.slack.blocks import build_documents_blocks
 from danswer.bots.slack.blocks import build_qa_response_blocks
 from danswer.bots.slack.blocks import get_restate_blocks
-from danswer.bots.slack.config import get_slack_bot_config_for_channel
+from danswer.bots.slack.constants import SLACK_CHANNEL_ID
+from danswer.bots.slack.models import SlackMessageInfo
+from danswer.bots.slack.utils import ChannelIdAdapter
 from danswer.bots.slack.utils import fetch_userids_from_emails
-from danswer.bots.slack.utils import get_channel_name_from_id
 from danswer.bots.slack.utils import respond_in_thread
 from danswer.configs.app_configs import DANSWER_BOT_ANSWER_GENERATION_TIMEOUT
 from danswer.configs.app_configs import DANSWER_BOT_DISABLE_DOCS_ONLY_ANSWER
 from danswer.configs.app_configs import DANSWER_BOT_DISPLAY_ERROR_MSGS
 from danswer.configs.app_configs import DANSWER_BOT_NUM_RETRIES
-from danswer.configs.app_configs import DANSWER_BOT_RESPOND_EVERY_CHANNEL
 from danswer.configs.app_configs import DOCUMENT_INDEX_NAME
 from danswer.configs.app_configs import ENABLE_DANSWERBOT_REFLEXION
 from danswer.configs.constants import DOCUMENT_SETS
 from danswer.db.engine import get_sqlalchemy_engine
+from danswer.db.models import SlackBotConfig
 from danswer.direct_qa.answer_question import answer_qa_query
 from danswer.server.models import QAResponse
 from danswer.server.models import QuestionRequest
+from danswer.utils.logger import setup_logger
+
+logger_base = setup_logger()
 
 
 def handle_message(
-    msg: str,
-    channel: str,
-    message_ts_to_respond_to: str | None,
-    sender_id: str | None,
+    message_info: SlackMessageInfo,
+    channel_config: SlackBotConfig | None,
     client: WebClient,
-    logger: logging.Logger,
-    skip_filters: bool = False,
-    is_bot_msg: bool = False,
     num_retries: int = DANSWER_BOT_NUM_RETRIES,
     answer_generation_timeout: int = DANSWER_BOT_ANSWER_GENERATION_TIMEOUT,
     should_respond_with_error_msgs: bool = DANSWER_BOT_DISPLAY_ERROR_MSGS,
     disable_docs_only_answer: bool = DANSWER_BOT_DISABLE_DOCS_ONLY_ANSWER,
-    respond_every_channel: bool = DANSWER_BOT_RESPOND_EVERY_CHANNEL,
-) -> None:
-    engine = get_sqlalchemy_engine()
-    with Session(engine) as db_session:
-        channel_name = get_channel_name_from_id(client=client, channel_id=channel)
-        slack_bot_config = get_slack_bot_config_for_channel(
-            channel_name=channel_name, db_session=db_session
+) -> bool:
+    """Potentially respond to the user message depending on filters and if an answer was generated
+
+    Returns True if need to respond with an additional message to the user(s) after this
+    function is finished. True indicates an unexpected failure that needs to be communicated
+    Query thrown out by filters due to config does not count as a failure that should be notified
+    Danswer failing to answer/retrieve docs does count and should be notified
+    """
+    msg = message_info.msg_content
+    channel = message_info.channel_to_respond
+    message_ts_to_respond_to = message_info.msg_to_respond
+    sender_id = message_info.sender
+    bipass_filters = message_info.bipass_filters
+    is_bot_msg = message_info.is_bot_msg
+
+    logger = cast(
+        logging.Logger,
+        ChannelIdAdapter(logger_base, extra={SLACK_CHANNEL_ID: channel}),
     )
-        if slack_bot_config is None and not respond_every_channel:
-            logger.info(
-                "Skipping message since the channel is not configured to use DanswerBot"
-            )
-            return
 
     document_set_names: list[str] | None = None
-        if slack_bot_config and slack_bot_config.persona:
+    if channel_config and channel_config.persona:
         document_set_names = [
-                document_set.name
-                for document_set in slack_bot_config.persona.document_sets
+            document_set.name for document_set in channel_config.persona.document_sets
         ]
 
     reflexion = ENABLE_DANSWERBOT_REFLEXION
@@ -65,9 +70,9 @@ def handle_message(
     send_to: list[str] | None = None
     respond_tag_only = False
     respond_team_member_list = None
-        if slack_bot_config and slack_bot_config.channel_config:
-            channel_conf = slack_bot_config.channel_config
-            if not skip_filters and "answer_filters" in channel_conf:
+    if channel_config and channel_config.channel_config:
+        channel_conf = channel_config.channel_config
+        if not bipass_filters and "answer_filters" in channel_conf:
             reflexion = "well_answered_postfilter" in channel_conf["answer_filters"]
 
             if (
@@ -77,7 +82,7 @@ def handle_message(
                 logger.info(
                     "Skipping message since it does not contain a question mark"
                 )
-                    return
+                return False
 
         logger.info(
             "Found slack bot config for channel. Restricting bot to use document "
@@ -86,17 +91,14 @@ def handle_message(
         )
 
         respond_tag_only = channel_conf.get("respond_tag_only") or False
-            respond_team_member_list = (
-                channel_conf.get("respond_team_member_list") or None
-            )
+        respond_team_member_list = channel_conf.get("respond_team_member_list") or None
 
-        # `skip_filters=True` -> this is a tag, so we *should* respond
-        if respond_tag_only and not skip_filters:
+    if respond_tag_only and not bipass_filters:
         logger.info(
             "Skipping message since the channel is configured such that "
             "DanswerBot only responds to tags"
         )
-            return
+        return False
 
     if respond_team_member_list:
         send_to = fetch_userids_from_emails(respond_team_member_list, client)
@@ -163,7 +165,7 @@ def handle_message(
                 text=f"Encountered exception when trying to answer: \n\n```{e}```",
                 thread_ts=message_ts_to_respond_to,
             )
-        return
+        return True
 
     if answer.eval_res_valid is False:
         logger.info(
@@ -171,7 +173,7 @@ def handle_message(
         )
         if answer.answer:
             logger.debug(answer.answer)
-        return
+        return True
 
     if not answer.top_ranked_docs:
         logger.error(f"Unable to answer question: '{msg}' - no documents found")
@@ -185,14 +187,14 @@ def handle_message(
                 text="Found no documents when trying to answer. Did you index any documents?",
                 thread_ts=message_ts_to_respond_to,
             )
-        return
+        return True
 
     if not answer.answer and disable_docs_only_answer:
         logger.info(
             "Unable to find answer - not responding since the "
             "`DANSWER_BOT_DISABLE_DOCS_ONLY_ANSWER` env variable is set"
         )
-        return
+        return True
 
     # convert raw response into "nicely" formatted Slack message
 
@@ -233,8 +235,10 @@ def handle_message(
                 thread_ts=message_ts_to_respond_to,
             )
 
+        return False
+
     except Exception:
         logger.exception(
             f"Unable to process message - could not respond in slack in {num_retries} attempts"
         )
-        return
+        return True

backend/danswer/bots/slack/listener.py

@@ -1,46 +1,40 @@
-import logging
 import re
 import time
-from collections.abc import MutableMapping
 from typing import Any
 from typing import cast
 
 from slack_sdk import WebClient
+from slack_sdk.errors import SlackApiError
 from slack_sdk.socket_mode import SocketModeClient
 from slack_sdk.socket_mode.request import SocketModeRequest
 from slack_sdk.socket_mode.response import SocketModeResponse
+from sqlalchemy.orm import Session
 
+from danswer.bots.slack.config import get_slack_bot_config_for_channel
+from danswer.bots.slack.constants import SLACK_CHANNEL_ID
 from danswer.bots.slack.handlers.handle_feedback import handle_slack_feedback
 from danswer.bots.slack.handlers.handle_message import handle_message
+from danswer.bots.slack.models import SlackMessageInfo
 from danswer.bots.slack.tokens import fetch_tokens
+from danswer.bots.slack.utils import ChannelIdAdapter
 from danswer.bots.slack.utils import decompose_block_id
+from danswer.bots.slack.utils import get_channel_name_from_id
+from danswer.bots.slack.utils import respond_in_thread
+from danswer.configs.app_configs import DANSWER_BOT_RESPOND_EVERY_CHANNEL
+from danswer.configs.app_configs import DANSWER_REACT_EMOJI
+from danswer.connectors.slack.utils import make_slack_api_rate_limited
+from danswer.db.engine import get_sqlalchemy_engine
 from danswer.dynamic_configs.interface import ConfigNotFoundError
 from danswer.utils.logger import setup_logger
 
+
 logger = setup_logger()
 
 
-_CHANNEL_ID = "channel_id"
-
-
 class MissingTokensException(Exception):
     pass
 
 
-class _ChannelIdAdapter(logging.LoggerAdapter):
-    """This is used to add the channel ID to all log messages
-    emitted in this file"""
-
-    def process(
-        self, msg: str, kwargs: MutableMapping[str, Any]
-    ) -> tuple[str, MutableMapping[str, Any]]:
-        channel_id = self.extra.get(_CHANNEL_ID) if self.extra else None
-        if channel_id:
-            return f"[Channel ID: {channel_id}] {msg}", kwargs
-        else:
-            return msg, kwargs
-
-
 def _get_socket_client() -> SocketModeClient:
     # For more info on how to set this up, checkout the docs:
     # https://docs.danswer.dev/slack_bot_setup
@@ -55,47 +49,44 @@ def _get_socket_client() -> SocketModeClient:
     )
 
 
-def _process_slack_event(client: SocketModeClient, req: SocketModeRequest) -> None:
-    logger.info(f"Received Slack request of type: '{req.type}'")
-
+def prefilter_requests(req: SocketModeRequest, client: SocketModeClient) -> bool:
+    """True to keep going, False to ignore this Slack request"""
     if req.type == "events_api":
-        # Acknowledge the request immediately
-        response = SocketModeResponse(envelope_id=req.envelope_id)
-        client.send_socket_mode_response(response)
-
         # Verify channel is valid
         event = cast(dict[str, Any], req.payload.get("event", {}))
+        msg = cast(str | None, event.get("text"))
         channel = cast(str | None, event.get("channel"))
-        channel_specific_logger = _ChannelIdAdapter(
-            logger, extra={_CHANNEL_ID: channel}
+        channel_specific_logger = ChannelIdAdapter(
+            logger, extra={SLACK_CHANNEL_ID: channel}
         )
-        # this should never happen, but we can't continue without a channel since
+
+        # This should never happen, but we can't continue without a channel since
         # we can't send a response without it
         if not channel:
             channel_specific_logger.error("Found message without channel - skipping")
-            return
+            return False
 
-        event = cast(dict[str, Any], req.payload.get("event", {}))
+        if not msg:
+            channel_specific_logger.error("Cannot respond to empty message - skipping")
+            return False
 
-        # Ensure that the message is a new message + of expected type
+        # Ensure that the message is a new message of expected type
         event_type = event.get("type")
         if event_type not in ["app_mention", "message"]:
             channel_specific_logger.info(
                 f"Ignoring non-message event of type '{event_type}' for channel '{channel}'"
             )
-            return
+            return False
 
-        # Don't insert Danswer thoughts if there's already a long conversation
-        # Or if a bunch of blocks already came through from responding to the @DanswerBot tag
-        if len(event.get("blocks", [])) > 10:
-            channel_specific_logger.debug(
-                "Ignoring message because thread is already long or has been answered to."
-            )
-            return
+        if event_type == "message":
+            bot_tag_id = client.web_client.auth_test().get("user_id")
+            if bot_tag_id and bot_tag_id in msg:
+                # Let the tag flow handle this case, don't reply twice
+                return False
 
         if event.get("bot_profile"):
             channel_specific_logger.info("Ignoring message from bot")
-            return
+            return False
 
         # Ignore things like channel_join, channel_leave, etc.
         # NOTE: "file_share" is just a message with a file attachment, so we
@@ -105,86 +96,45 @@ def _process_slack_event(client: SocketModeClient, req: SocketModeRequest) -> No
             channel_specific_logger.info(
                 f"Ignoring message with subtype '{message_subtype}' since is is a special message type"
             )
-            return
+            return False
 
         message_ts = event.get("ts")
         thread_ts = event.get("thread_ts")
         # Pick the root of the thread (if a thread exists)
-        message_ts_to_respond_to = cast(str, thread_ts or message_ts)
         if thread_ts and message_ts != thread_ts:
             channel_specific_logger.info(
                 "Skipping message since it is not the root of a thread"
             )
-            return
+            return False
 
         msg = cast(str, event.get("text", ""))
         if not msg:
             channel_specific_logger.error("Unable to process empty message")
-            return
-
-        tagged = event_type == "app_mention"
-        if tagged:
-            logger.info("User tagged DanswerBot")
-            msg = re.sub(r"<@\w+>\s", "", msg)
-
-        # TODO: message should be enqueued and processed elsewhere,
-        # but doing it here for now for simplicity
-        handle_message(
-            msg=msg,
-            channel=channel,
-            message_ts_to_respond_to=message_ts_to_respond_to,
-            sender_id=event.get("user") or None,
-            client=client.web_client,
-            skip_filters=tagged,
-            logger=cast(logging.Logger, channel_specific_logger),
-        )
-        channel_specific_logger.info(
-            f"Successfully processed message with ts: '{message_ts}'"
-        )
+            return False
 
     if req.type == "slash_commands":
-        # Acknowledge the request immediately
-        response = SocketModeResponse(envelope_id=req.envelope_id)
-        client.send_socket_mode_response(response)
-
         # Verify that there's an associated channel
         channel = req.payload.get("channel_id")
-        channel_specific_logger = _ChannelIdAdapter(
-            logger, extra={_CHANNEL_ID: channel}
+        channel_specific_logger = ChannelIdAdapter(
+            logger, extra={SLACK_CHANNEL_ID: channel}
         )
         if not channel:
             channel_specific_logger.error(
                 "Received DanswerBot command without channel - skipping"
             )
-            return
+            return False
 
-        msg = req.payload.get("text", "")
         sender = req.payload.get("user_id")
         if not sender:
-            raise ValueError(
+            channel_specific_logger.error(
                 "Cannot respond to DanswerBot command without sender to respond to."
             )
+            return False
 
-        handle_message(
-            msg=msg,
-            channel=channel,
-            message_ts_to_respond_to=None,
-            sender_id=sender,
-            client=client.web_client,
-            skip_filters=True,
-            is_bot_msg=True,
-            logger=cast(logging.Logger, channel_specific_logger),
-        )
-        channel_specific_logger.info(
-            f"Successfully processed DanswerBot request in channel: {req.payload.get('channel_name')}"
-        )
+    return True
 
-    # Handle button clicks
-    if req.type == "interactive" and req.payload.get("type") == "block_actions":
-        # Acknowledge the request immediately
-        response = SocketModeResponse(envelope_id=req.envelope_id)
-        client.send_socket_mode_response(response)
 
+def process_feedback(req: SocketModeRequest, client: SocketModeClient) -> None:
     actions = req.payload.get("actions")
     if not actions:
         logger.error("Unable to process block actions - no actions found")
@@ -210,9 +160,157 @@ def _process_slack_event(client: SocketModeClient, req: SocketModeRequest) -> No
     logger.info(f"Successfully handled QA feedback for event: {query_event_id}")
 
 
-def process_slack_event(client: SocketModeClient, req: SocketModeRequest) -> None:
+def build_request_details(
+    req: SocketModeRequest, client: SocketModeClient
+) -> SlackMessageInfo:
+    if req.type == "events_api":
+        event = cast(dict[str, Any], req.payload["event"])
+        msg = cast(str, event["text"])
+        channel = cast(str, event["channel"])
+        tagged = event.get("type") == "app_mention"
+        message_ts = event.get("ts")
+        thread_ts = event.get("thread_ts")
+        if tagged:
+            logger.info("User tagged DanswerBot")
+            bot_tag_id = client.web_client.auth_test().get("user_id")
+            msg = re.sub(rf"<@{bot_tag_id}>\s", "", msg)
+
+        return SlackMessageInfo(
+            msg_content=msg,
+            channel_to_respond=channel,
+            msg_to_respond=cast(str, thread_ts or message_ts),
+            sender=event.get("user") or None,
+            bipass_filters=tagged,
+            is_bot_msg=False,
+        )
+
+    elif req.type == "slash_commands":
+        channel = req.payload["channel_id"]
+        msg = req.payload["text"]
+        sender = req.payload["user_id"]
+
+        return SlackMessageInfo(
+            msg_content=msg,
+            channel_to_respond=channel,
+            msg_to_respond=None,
+            sender=sender,
+            bipass_filters=True,
+            is_bot_msg=True,
+        )
+
+    raise RuntimeError("Programming fault, this should never happen.")
+
+
+def send_msg_ack_to_user(details: SlackMessageInfo, client: SocketModeClient) -> None:
+    if details.is_bot_msg and details.sender:
+        respond_in_thread(
+            client=client.web_client,
+            channel=details.channel_to_respond,
+            thread_ts=details.msg_to_respond,
+            receiver_ids=[details.sender],
+            text="Hi, we're evaluating your query :face_with_monocle:",
+        )
+        return
+
+    slack_call = make_slack_api_rate_limited(client.web_client.reactions_add)
+    slack_call(
+        name=DANSWER_REACT_EMOJI,
+        channel=details.channel_to_respond,
+        timestamp=details.msg_to_respond,
+    )
+
+
+def remove_react(details: SlackMessageInfo, client: SocketModeClient) -> None:
+    if details.is_bot_msg:
+        return
+
+    slack_call = make_slack_api_rate_limited(client.web_client.reactions_remove)
+    slack_call(
+        name=DANSWER_REACT_EMOJI,
+        channel=details.channel_to_respond,
+        timestamp=details.msg_to_respond,
+    )
+
+
+def apologize_for_fail(
+    details: SlackMessageInfo,
+    client: SocketModeClient,
+) -> None:
+    respond_in_thread(
+        client=client.web_client,
+        channel=details.channel_to_respond,
+        thread_ts=details.msg_to_respond,
+        text="Sorry, we weren't able to find anything relevant :cold_sweat:",
+    )
+
+
+def process_message(
+    req: SocketModeRequest,
+    client: SocketModeClient,
+    respond_every_channel: bool = DANSWER_BOT_RESPOND_EVERY_CHANNEL,
+) -> None:
+    logger.info(f"Received Slack request of type: '{req.type}'")
+
+    # Throw out requests that can't or shouldn't be handled
+    if not prefilter_requests(req, client):
+        return
+
+    details = build_request_details(req, client)
+    channel = details.channel_to_respond
+    channel_name = get_channel_name_from_id(
+        client=client.web_client, channel_id=channel
+    )
+
+    engine = get_sqlalchemy_engine()
+    with Session(engine) as db_session:
+        slack_bot_config = get_slack_bot_config_for_channel(
+            channel_name=channel_name, db_session=db_session
+        )
+
+    # Be careful about this default, don't want to accidentally spam every channel
+    if slack_bot_config is None and not respond_every_channel:
+        logger.info(
+            "Skipping message since the channel is not configured to use DanswerBot"
+        )
+        return
+
     try:
-        _process_slack_event(client=client, req=req)
+        send_msg_ack_to_user(details, client)
+    except SlackApiError as e:
+        logger.error(f"Was not able to react to user message due to: {e}")
+
+    failed = handle_message(
+        message_info=details,
+        channel_config=slack_bot_config,
+        client=client.web_client,
+    )
+
+    # Skipping answering due to pre-filtering is not considered a failure
+    if failed:
+        apologize_for_fail(details, client)
+
+    try:
+        remove_react(details, client)
+    except SlackApiError as e:
+        logger.error(f"Failed to remove Reaction due to: {e}")
+
+
+def acknowledge_message(req: SocketModeRequest, client: SocketModeClient) -> None:
+    response = SocketModeResponse(envelope_id=req.envelope_id)
+    client.send_socket_mode_response(response)
+
+
+def process_slack_event(client: SocketModeClient, req: SocketModeRequest) -> None:
+    # Always respond right away, if Slack doesn't receive these frequently enough
+    # it will assume the Bot is DEAD!!! :(
+    acknowledge_message(req, client)
+
+    try:
+        if req.type == "interactive" and req.payload.get("type") == "block_actions":
+            return process_feedback(req, client)
+
+        elif req.type == "events_api" or req.type == "slash_commands":
+            return process_message(req, client)
     except Exception:
         logger.exception("Failed to process slack event")
 

backend/danswer/bots/slack/models.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+
+
+class SlackMessageInfo(BaseModel):
+    msg_content: str
+    channel_to_respond: str
+    msg_to_respond: str | None
+    sender: str | None
+    bipass_filters: bool
+    is_bot_msg: bool

backend/danswer/bots/slack/utils.py

@@ -2,6 +2,7 @@ import logging
 import random
 import re
 import string
+from collections.abc import MutableMapping
 from typing import Any
 from typing import cast
 
@@ -10,6 +11,7 @@ from slack_sdk import WebClient
 from slack_sdk.models.blocks import Block
 from slack_sdk.models.metadata import Metadata
 
+from danswer.bots.slack.constants import SLACK_CHANNEL_ID
 from danswer.bots.slack.tokens import fetch_tokens
 from danswer.configs.app_configs import DANSWER_BOT_NUM_RETRIES
 from danswer.configs.constants import ID_SEPARATOR
@@ -21,6 +23,20 @@ from danswer.utils.text_processing import replace_whitespaces_w_space
 logger = setup_logger()
 
 
+class ChannelIdAdapter(logging.LoggerAdapter):
+    """This is used to add the channel ID to all log messages
+    emitted in this file"""
+
+    def process(
+        self, msg: str, kwargs: MutableMapping[str, Any]
+    ) -> tuple[str, MutableMapping[str, Any]]:
+        channel_id = self.extra.get(SLACK_CHANNEL_ID) if self.extra else None
+        if channel_id:
+            return f"[Channel ID: {channel_id}] {msg}", kwargs
+        else:
+            return msg, kwargs
+
+
 def get_web_client() -> WebClient:
     slack_tokens = fetch_tokens()
     return WebClient(token=slack_tokens.bot_token)

backend/danswer/configs/app_configs.py

@@ -230,6 +230,7 @@ DANSWER_BOT_DISPLAY_ERROR_MSGS = os.environ.get(
 DANSWER_BOT_DISABLE_DOCS_ONLY_ANSWER = os.environ.get(
     "DANSWER_BOT_DISABLE_DOCS_ONLY_ANSWER", ""
 ).lower() not in ["false", ""]
+DANSWER_REACT_EMOJI = os.environ.get("DANSWER_REACT_EMOJI") or "eyes"
 
 # Default is only respond in channels that are included by a slack config set in the UI
 DANSWER_BOT_RESPOND_EVERY_CHANNEL = (