Claudette in Slack

claudette_in_slack.py

import os
from slack_sdk.web import WebClient
from slack_sdk.socket_mode import SocketModeClient
from slack_sdk.socket_mode.response import SocketModeResponse
from slack_sdk.socket_mode.request import SocketModeRequest
import dotenv
import logging
from src.utils import (
    get_thread,
    apply_chat_template,
)
from claudette import Chat, models, contents

logging.basicConfig(level=logging.INFO)
dotenv.load_dotenv()


customers = {
    "C1": dict(
        name="Alice Johnson",
        email="alice@example.com",
        phone="123-456-7890",
        games=["G1", "G2", "G3"],
    ),
    "C2": dict(
        name="Bob Smith",
        email="bob@example.com",
        phone="987-654-3210",
        games=["G4", "G5"],
    ),
}

games = {
    "G1": dict(
        id="G1",
        name="Shadow Realms",
        release_date="2023-03-15",
        description="Navigate enchanted forests and haunted castles.",
        status="Shipped",
    ),
    "G2": dict(
        id="G2",
        name="Solar Winds",
        release_date="2023-07-22",
        description="Explore space with stunning visuals and alien planets.",
        status="Shipped",
    ),
    "G3": dict(
        id="G3",
        name="Mystic Legends",
        release_date="2023-11-10",
        description="Epic fantasy RPG with beautiful landscapes.",
        status="Shipped",
    ),
    "G4": dict(
        id="G4",
        name="Cyber Revolution",
        release_date="2024-02-28",
        description="Dystopian future with advanced technology and cyber warfare.",
        status="Shipped",
    ),
    "G5": dict(
        id="G5",
        name="Desert Storm",
        release_date="2024-05-05",
        description="Tactical shooter in a war-torn desert.",
        status="Processing",
    ),
}


def get_customer_info(
    customer_id: str,  # ID of the customer
):  # Customer's name, email, phone number, and list of games
    "Retrieves a customer's information and their orders based on the customer ID"
    print(f"- Retrieving customer {customer_id}")
    return customers.get(customer_id, "Customer not found")


def get_game_details(
    game_id: str,  # ID of the game
):  # Game's ID, name, release date, description & status
    "Retrieves the details of a specific game based on the game ID"
    print(f"- Retrieving game {game_id}")
    return games.get(game_id, "Game not found")


def return_game(
    game_id: str,  # ID of the order to cancel
) -> bool:  # True if the return is successful
    "Returns a game to the cmpany based on game ID."
    print(f"- Returning game {game_id}")
    if game_id not in games:
        return False
    games[game_id]["status"] = "Returned"
    return True


_client = WebClient(token=os.environ["SLACK_BOT_TOKEN"])
BOT_USER_ID = _client.auth_test()["user_id"]

# Initialize SocketModeClient with an app-level token + WebClient
client = SocketModeClient(
    app_token=os.environ.get("SLACK_APP_TOKEN"),
    web_client=WebClient(token=os.environ.get("SLACK_BOT_TOKEN")),
)
CHANNEL_ID = os.environ["LISTEN_CHANNEL_ID"]  # blog

chat = Chat(
    model=models[1],
    sp="You are Claudia. You never tell users what tools you use.",
    tools=[get_customer_info, get_game_details, return_game],
)


def process(client: SocketModeClient, req: SocketModeRequest):
    print(req.payload)
    if req.type == "events_api" or req.type == "event_callback":
        response = SocketModeResponse(envelope_id=req.envelope_id)
        client.send_socket_mode_response(response)
        if (
            req.payload["event"]["type"] == "message"
            and req.payload["event"].get("subtype") is None
            and "bot_profile" not in req.payload["event"].keys()
        ):
            thread_ts = req.payload["event"]["ts"]
            if "thread_ts" in req.payload["event"].keys():
                thread_ts = req.payload["event"]["thread_ts"]
            text = req.payload["event"]["text"]
            r = chat.toolloop(text, maxtok=200)
            response = _client.chat_postMessage(
                channel=CHANNEL_ID, text=contents(r), thread_ts=thread_ts
            )


# Add a new listener to receive messages from Slack
client.socket_mode_request_listeners.append(process)

# Establish a WebSocket connection to the Socket Mode servers
client.connect()

# Just not to stop this process
from threading import Event

Event().wait()