Skip to content

Enabling usage of groq's api keys #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Abhi2april wants to merge 7 commits into
base: main
Choose a base branch
from
Open

Enabling usage of groq's api keys #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c08721c
Delete listeners/__init__.py
Abhi2april Feb 28, 2025
63955d2
Delete listeners/assistant.py
Abhi2april Feb 28, 2025
59d2437
Delete listeners/llm_caller.py
Abhi2april Feb 28, 2025
652f238
Add files via upload
Abhi2april Feb 28, 2025
171d446
Add files via upload
Abhi2april Feb 28, 2025
cff848e
Update README.md
Abhi2april Feb 28, 2025
dcfa67c
Update README.md
Abhi2april Feb 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
# App Agent & Assistant Template (Bolt for Python)

# EthicAll For Your Slack workspace moderation --> By Abhishek Singh
You Don't need any more to pay for you OpenAI's API keys-> Just use your free Groq's Inference Tokens and change to your favourite model to work in backend from [here](https://console.groq.com/docs/models) and copy it to listeners/llm_caller.py
This Bolt for Python template demonstrates how to build [Agents & Assistants](https://api.slack.com/docs/apps/ai) in Slack.

## Setup
Expand Down Expand Up @@ -29,14 +29,13 @@ Before you can run the app, you'll need to store some environment variables.
export SLACK_BOT_TOKEN=<your-bot-token>
export SLACK_APP_TOKEN=<your-app-token>
# This sample uses OpenAI's API by default, but you can switch to any other solution!
export OPENAI_API_KEY=<your-openai-api-key>
export GROQ_API_KEY=<your-groq-api-key>
```

### Setup Your Local Project
```zsh
# Clone this project onto your machine
git clone https://github.com/slack-samples/bolt-python-assistant-template.git

git clone https://github.com/Abhi2april/bolt-python-assistant-template
# Change into this project directory
cd bolt-python-assistant-template

Expand Down
15 changes: 5 additions & 10 deletions listeners/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,5 @@
from .assistant import assistant


def register_listeners(app):
# Using assistant middleware is the recommended way.
app.assistant(assistant)

# The following event listeners demonstrate how to implement the same on your own.
# from listeners import events
# events.register(app)
from .assistant import assistant


def register_listeners(app):
app.assistant(assistant)
218 changes: 109 additions & 109 deletions listeners/assistant.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,109 +1,109 @@
import logging
from typing import List, Dict
from slack_bolt import Assistant, BoltContext, Say, SetSuggestedPrompts, SetStatus
from slack_bolt.context.get_thread_context import GetThreadContext
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

from .llm_caller import call_llm

# Refer to https://tools.slack.dev/bolt-python/concepts/assistant/ for more details
assistant = Assistant()


# This listener is invoked when a human user opened an assistant thread
@assistant.thread_started
def start_assistant_thread(
say: Say,
get_thread_context: GetThreadContext,
set_suggested_prompts: SetSuggestedPrompts,
logger: logging.Logger,
):
try:
say("How can I help you?")

prompts: List[Dict[str, str]] = [
{
"title": "What does Slack stand for?",
"message": "Slack, a business communication service, was named after an acronym. Can you guess what it stands for?",
},
{
"title": "Write a draft announcement",
"message": "Can you write a draft announcement about a new feature my team just released? It must include how impactful it is.",
},
{
"title": "Suggest names for my Slack app",
"message": "Can you suggest a few names for my Slack app? The app helps my teammates better organize information and plan priorities and action items.",
},
]

thread_context = get_thread_context()
if thread_context is not None and thread_context.channel_id is not None:
summarize_channel = {
"title": "Summarize the referred channel",
"message": "Can you generate a brief summary of the referred channel?",
}
prompts.append(summarize_channel)

set_suggested_prompts(prompts=prompts)
except Exception as e:
logger.exception(f"Failed to handle an assistant_thread_started event: {e}", e)
say(f":warning: Something went wrong! ({e})")


# This listener is invoked when the human user sends a reply in the assistant thread
@assistant.user_message
def respond_in_assistant_thread(
payload: dict,
logger: logging.Logger,
context: BoltContext,
set_status: SetStatus,
get_thread_context: GetThreadContext,
client: WebClient,
say: Say,
):
try:
user_message = payload["text"]
set_status("is typing...")

if user_message == "Can you generate a brief summary of the referred channel?":
# the logic here requires the additional bot scopes:
# channels:join, channels:history, groups:history
thread_context = get_thread_context()
referred_channel_id = thread_context.get("channel_id")
try:
channel_history = client.conversations_history(channel=referred_channel_id, limit=50)
except SlackApiError as e:
if e.response["error"] == "not_in_channel":
# If this app's bot user is not in the public channel,
# we'll try joining the channel and then calling the same API again
client.conversations_join(channel=referred_channel_id)
channel_history = client.conversations_history(channel=referred_channel_id, limit=50)
else:
raise e

prompt = f"Can you generate a brief summary of these messages in a Slack channel <#{referred_channel_id}>?\n\n"
for message in reversed(channel_history.get("messages")):
if message.get("user") is not None:
prompt += f"\n<@{message['user']}> says: {message['text']}\n"
messages_in_thread = [{"role": "user", "content": prompt}]
returned_message = call_llm(messages_in_thread)
say(returned_message)
return

replies = client.conversations_replies(
channel=context.channel_id,
ts=context.thread_ts,
oldest=context.thread_ts,
limit=10,
)
messages_in_thread: List[Dict[str, str]] = []
for message in replies["messages"]:
role = "user" if message.get("bot_id") is None else "assistant"
messages_in_thread.append({"role": role, "content": message["text"]})
returned_message = call_llm(messages_in_thread)
say(returned_message)

except Exception as e:
logger.exception(f"Failed to handle a user message event: {e}")
say(f":warning: Something went wrong! ({e})")
import logging
from typing import List, Dict
from slack_bolt import Assistant, BoltContext, Say, SetSuggestedPrompts, SetStatus
from slack_bolt.context.get_thread_context import GetThreadContext
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from .llm_caller import call_llm
# Refer to https://tools.slack.dev/bolt-python/concepts/assistant/ for more details
assistant = Assistant()
# This listener is invoked when a human user opened an assistant thread
@assistant.thread_started
def start_assistant_thread(
say: Say,
get_thread_context: GetThreadContext,
set_suggested_prompts: SetSuggestedPrompts,
logger: logging.Logger,
):
try:
say("How can I help you?")
prompts: List[Dict[str, str]] = [
{
"title": "What does Slack stand for?",
"message": "Slack, a business communication service, was named after an acronym. Can you guess what it stands for?",
},
{
"title": "Write a draft announcement",
"message": "Can you write a draft announcement about a new feature my team just released? It must include how impactful it is.",
},
{
"title": "Suggest names for my Slack app",
"message": "Can you suggest a few names for my Slack app? The app helps my teammates better organize information and plan priorities and action items.",
},
]
thread_context = get_thread_context()
if thread_context is not None and thread_context.channel_id is not None:
summarize_channel = {
"title": "Summarize the referred channel",
"message": "Can you generate a brief summary of the referred channel?",
}
prompts.append(summarize_channel)
set_suggested_prompts(prompts=prompts)
except Exception as e:
logger.exception(f"Failed to handle an assistant_thread_started event: {e}", e)
say(f":warning: Something went wrong! ({e})")
# This listener is invoked when the human user sends a reply in the assistant thread
@assistant.user_message
def respond_in_assistant_thread(
payload: dict,
logger: logging.Logger,
context: BoltContext,
set_status: SetStatus,
get_thread_context: GetThreadContext,
client: WebClient,
say: Say,
):
try:
user_message = payload["text"]
set_status("is typing...")
if user_message == "Can you generate a brief summary of the referred channel?":
# the logic here requires the additional bot scopes:
# channels:join, channels:history, groups:history
thread_context = get_thread_context()
referred_channel_id = thread_context.get("channel_id")
try:
channel_history = client.conversations_history(channel=referred_channel_id, limit=50)
except SlackApiError as e:
if e.response["error"] == "not_in_channel":
# If this app's bot user is not in the public channel,
# we'll try joining the channel and then calling the same API again
client.conversations_join(channel=referred_channel_id)
channel_history = client.conversations_history(channel=referred_channel_id, limit=50)
else:
raise e
prompt = f"Can you generate a brief summary of these messages in a Slack channel <#{referred_channel_id}>?\n\n"
for message in reversed(channel_history.get("messages")):
if message.get("user") is not None:
prompt += f"\n<@{message['user']}> says: {message['text']}\n"
messages_in_thread = [{"role": "user", "content": prompt}]
returned_message = call_llm(messages_in_thread)
say(returned_message)
return
replies = client.conversations_replies(
channel=context.channel_id,
ts=context.thread_ts,
oldest=context.thread_ts,
limit=10,
)
messages_in_thread: List[Dict[str, str]] = []
for message in replies["messages"]:
role = "user" if message.get("bot_id") is None else "assistant"
messages_in_thread.append({"role": role, "content": message["text"]})
returned_message = call_llm(messages_in_thread)
say(returned_message)
except Exception as e:
logger.exception(f"Failed to handle a user message event: {e}")
say(f":warning: Something went wrong! ({e})")
76 changes: 43 additions & 33 deletions listeners/events/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,33 +1,43 @@
# This sample app repository contains event listener code to help developers understand what's happening under the hood.
# We recommend using assistant middleware instead of these event listeners.
# For more details, refer to https://tools.slack.dev/bolt-python/concepts/assistant/.

from typing import Dict, Any

from slack_bolt import App
from slack_bolt.request.payload_utils import is_event

from .assistant_thread_started import start_thread_with_suggested_prompts
from .asssistant_thread_context_changed import save_new_thread_context
from .user_message import respond_to_user_message


def register(app: App):
app.event("assistant_thread_started")(start_thread_with_suggested_prompts)
app.event("assistant_thread_context_changed")(save_new_thread_context)
app.event("message", matchers=[is_user_message_event_in_assistant_thread])(respond_to_user_message)
app.event("message", matchers=[is_message_event_in_assistant_thread])(just_ack)


def is_message_event_in_assistant_thread(body: Dict[str, Any]) -> bool:
if is_event(body):
return body["event"]["type"] == "message" and body["event"].get("channel_type") == "im"
return False


def is_user_message_event_in_assistant_thread(body: Dict[str, Any]) -> bool:
return is_message_event_in_assistant_thread(body) and body["event"].get("subtype") in (None, "file_share")


def just_ack():
pass
from typing import Dict, Any

from slack_bolt import App
from slack_bolt.request.payload_utils import is_event
from .app_mention import respond_to_mention
from .assistant_thread_started import start_thread_with_suggested_prompts
from .asssistant_thread_context_changed import save_new_thread_context

def register(app: App):
app.event("assistant_thread_started")(start_thread_with_suggested_prompts)
app.event("assistant_thread_context_changed")(save_new_thread_context)
app.event("message", matchers=[is_user_message_event_in_assistant_thread])(respond_to_user_message)
app.event("message", matchers=[is_message_event_in_assistant_thread])(just_ack)
app.event("message")(handle_app_mention)

def handle_app_mention(payload: dict, client, logger):
event = payload.get("event", {})
channel_id = event.get("channel")
thread_ts = event.get("ts")
user = event.get("user")

try:
client.chat_postMessage(
channel=channel_id,
thread_ts=thread_ts,
text=f"Hi <@{user}>, how can I help you?"
)
except Exception as e:
logger.exception(f"Failed to handle app_mention event: {e}")



def is_message_event_in_assistant_thread(body: Dict[str, Any]) -> bool:
if is_event(body):
return body["event"]["type"] == "message" and body["event"].get("channel_type") == "im"
return False


def is_user_message_event_in_assistant_thread(body: Dict[str, Any]) -> bool:
return is_message_event_in_assistant_thread(body) and body["event"].get("subtype") in (None, "file_share")

def just_ack():
pass
Loading