app.py

# ===================================================================================================
# Note: chainlit==0.7.500 <————was | is————> chainlit==2.0.603 / Demo Daysoff kundeservice 𝘼𝙄 support
# title: 𝘿𝙖𝙮𝙨𝙤𝙛𝙛 𝘼𝙨𝙨𝙞𝙨𝙩𝙖𝙣𝙩-API-v2.0.603| custom starters | API error handling
# file: app.py
# ===================================================================================================

import os
import re
import json
import asyncio
import requests

from pathlib import Path
from datetime import datetime
from dotenv import load_dotenv

import chainlit as cl
from concurrent.futures import ThreadPoolExecutor

from chainlit import user_session
from chainlit.session import WebsocketSession

from langchain import hub
from langchain_openai import OpenAI
from langchain.chains import LLMChain
from langchain_core.prompts import PromptTemplate
from langchain.memory.buffer import ConversationBufferMemory


load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
auth_token = os.environ.get("DAYSOFF_API_TOKEN")

API_URL = "https://aivisions.no/data/daysoff/api/v1/booking/"

daysoff_assistant_template = """
You are a customer support assistant for Daysoff ('Daysoff Kundeservice AI Support') who helps users retrieve booking information based on their bookingnummer.
You should concisely use the term ’bookingnummer’.
By default, you respond using Norwegian.
Provide a conversational answer.
This way you directly address the user's question in a manner that reflects the professionalism and warmth of a a female customer support representative archetype.
============================
Chat History: {chat_history}
Question: {question}
============================
Answer in Markdown:
"""

daysoff_assistant_prompt = PromptTemplate(
    input_variables=["chat_history", "question"],
    template=daysoff_assistant_template,
)


class APIConnectionError(Exception):
    """Raised when API connection fails"""
    pass

class APIResponseError(Exception):
    """Raised when API returns invalid response"""
    pass

class BookingNotFoundError(Exception):
    """Raised when booking ID is not found"""
    pass

async def async_post_request(url, headers, data):
    try:
        response = await asyncio.to_thread(requests.post, url, headers=headers, json=data)
        response.raise_for_status()
        return response
    except requests.ConnectionError:
        raise APIConnectionError("Failed to connect to booking service")
    except requests.Timeout:
        raise APIConnectionError("Request timed out")
    except requests.RequestException as e:
        raise APIResponseError(f"API request failed: {str(e)}")

        
@cl.set_starters
async def set_starters():
    return [
        cl.Starter(
            label="Booking ID request",
            message="Kan du gi meg info om en reservasjon?",
            icon="/public/booking_id.svg",
            ),
        cl.Starter(
            label="Metric Space Self-Identity Framework",
            message="Explain the Metric Space Self-Identity Framework like I'm six years old.",
            icon="/public/learn.svg",
            ),
        cl.Starter(
            label="Python script for daily email reports",
            message="Write a script to automate sending daily email reports in Python, and walk me through how I would set it up.",
            icon="/public/terminal.svg",
            ),
        cl.Starter(
            label="Morning routine ideation",
            message="Can you help me create a personalized Yoga/pranayama/meditation morning routine that would help increase my productivity throughout the day? Start by asking me about my current habits and what activities energize me in the morning.",
            icon="/public/idea.svg",
            )
        ]

#------?-----@cl.Step
#------?-----@cl.Context

@cl.on_chat_start
async def main():
    res = await cl.AskUserMessage(content="Hva er navnet ditt?", timeout=10).send()
    if res:
        await cl.Message(
            content=f"Aha, så du heter: {res['output']}",
        ).send()
  
    cl.user_session.set("socket_auth", True)
    cl.user_session.set("max_retries", 3)
    cl.user_session.set("connection_attempts", 0)
    
    llm = OpenAI(
        model="gpt-3.5-turbo-instruct",
        temperature=0.7,
        openai_api_key=OPENAI_API_KEY,
        max_tokens=2048,
        top_p=0.9,
        frequency_penalty=0.1,
        presence_penalty=0.1,
    )

    conversation_memory = ConversationBufferMemory(
        memory_key="chat_history",
        max_len=30,
        return_messages=True
    )

    llm_chain = LLMChain(
        llm=llm,
        prompt=daysoff_assistant_prompt,
        memory=conversation_memory,
    )

    cl.user_session.set("llm_chain", llm_chain)
    
    #asyncio.create_task(keep_alive())
  
async def long_running_task(message_content: str):
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as pool:
        llm_chain = cl.user_session.get("llm_chain")
        return await loop.run_in_executor(
            pool,
            lambda: llm_chain.invoke({
                "question": message_content,
                "chat_history": ""
            })
        )
    
@cl.on_message
async def handle_message(message: cl.Message):

    try:
        if not cl.user_session.get("socket_auth"):
            await cl.Message(content="Session authentication failed ————> refresh the page.").send()
            return
            
        user_message = message.content
        llm_chain = cl.user_session.get("llm_chain")
        #languages = cl.user_session.get("languages")
    
        if sys.version_info >= (3, 11):
            from asyncio import timeout
        else:
            from async_timeout import timeout
    
        booking_pattern = r'\b[A-Z]{6}\d{6}\b'
        match = re.search(booking_pattern, user_message)
    
        if match:
            bestillingskode = match.group()
            headers = {
                "Authorization": auth_token,
                "Content-Type": "application/json"
            }
            payload = {"booking_id": bestillingskode}
    
            try:
                
                response = await async_post_request(API_URL, headers, payload)
                response.raise_for_status()
                booking_data = response.json()
    
                if not booking_data:
                    raise BookingNotFoundError("Ingen booking informasjon ble funnet.")
                
                if "error" in booking_data:
                    raise APIResponseError(booking_data["error"])
    
                #table = (
                    #"| 𝑭𝒊𝒆𝒍𝒅           | 𝗜𝗻𝗳𝗼                                      |\n"
                    #"|:----------------|:-------------------------------------------|\n"
                    #f"| 𝙊𝙧𝙙𝙚𝙧 𝙉𝙪𝙢𝙗𝙚𝙧   | {booking_data.get('order_number', 'N/A')}        |\n"
                    #f"| 𝘼𝙘𝙘𝙚𝙥𝙩𝙚𝙙       | {booking_data.get('accepted', 'N/A')}           |\n"
                    #f"| 𝙂𝙪𝙚𝙨𝙩𝙨         | {booking_data.get('guests', 'N/A')}             |\n"
                    #f"| 𝙏𝙞𝙢𝙚 𝙁𝙧𝙤𝙢     | {booking_data.get('time_from', 'N/A')}          |\n"
                    #f"| 𝙏𝙞𝙢𝙚 𝙏𝙤       | {booking_data.get('time_to', 'N/A')}            |\n"
                    #f"| 𝘾𝙖𝙣𝙘𝙚𝙡𝙡𝙚𝙙     | {'Yes' if booking_data.get('cancelled', 0) else 'No'} |\n"
                    #f"| 𝘿𝙚𝙨𝙘𝙧𝙞𝙥𝙩𝙞𝙤𝙣   | {booking_data.get('description', 'N/A')}        |\n"
                    #f"| 𝘼𝙧𝙧𝙞𝙫𝙖𝙡       | {booking_data.get('arrival', 'N/A')}            |\n"
                    #f"| 𝘿𝙚𝙥𝙖𝙧𝙩𝙪𝙧𝙚     | {booking_data.get('departure', 'N/A')}          |\n"
                    #f"| 𝘽𝙚𝙙𝙙𝙞𝙣𝙜       | {booking_data.get('bedding', 'N/A')}            |\n"
                    #f"| 𝙇𝙤𝙘𝙖𝙩𝙞𝙤𝙣      | Lat: {booking_data.get('lat', 'N/A')}, Long: {booking_data.get('long', 'N/A')} |\n"
                    #f"| 𝙉𝙖𝙢𝙚          | {booking_data.get('name', 'N/A')}                |\n"
                    #f"| 𝘾𝙤𝙣𝙩𝙖𝙘𝙩 𝙉𝙖𝙢𝙚 | {booking_data.get('contactname', 'N/A')}         |\n"
                    #f"| 𝘾𝙤𝙣𝙩𝙖𝙘𝙩 𝙀𝙢𝙖𝙞𝙡| {booking_data.get('contactemail', 'N/A')}        |\n"
                    #f"| 𝘾𝙤𝙣𝙩𝙖𝙘𝙩 𝙋𝙝𝙤𝙣𝙚| {booking_data.get('contactphone', 'N/A')}        |\n"
                    #f"| 𝘾𝙤𝙪𝙣𝙩𝙧𝙮 𝘾𝙤𝙙𝙚 | {booking_data.get('country_code', 'N/A')}        |\n"
                #)
                #combined_message = f"### Informasjon om booking:\n\n{table}"
                #await cl.Message(content=combined_message).send()
    
                table = (
                    "| 𝑭𝒊𝒆𝒍𝒅      | 𝗜𝗻𝗳𝗼                |\n"
                    "|:-----------|:---------------------|\n"
                    f"| 𝙱𝚎𝚜𝚝𝚒𝚕𝚕𝚒𝚗𝚐𝚜𝚔𝚘𝚍𝚎 | {booking_data.get('booking_id', 'N/A')} |\n"
                    f"| 𝙁𝙪𝙡𝙡 𝙉𝙖𝙢𝙚  | {booking_data.get('full_name', 'N/A')} |\n"
                    f"| 𝘼𝙢𝙤𝙪𝙣𝙩     | {booking_data.get('amount', 0)} kr |\n"
                    f"| 𝘾𝙝𝙚𝙘𝙠-𝙞𝙣   | {booking_data.get('checkin', 'N/A')} |\n"
                    f"| 𝘾𝙝𝙚𝙘𝙠-𝙤𝙪𝙩  | {booking_data.get('checkout', 'N/A')} |\n"
                    f"| 𝘼𝙙𝙙𝙧𝙚𝙨𝙨    | {booking_data.get('address', 'N/A')} |\n"
                    f"| 𝙐𝙨𝙚𝙧 𝙄𝘿    | {booking_data.get('user_id', 0)} |\n"
                    f"| 𝙄𝙣𝙛𝙤 𝙏𝙚𝙭𝙩  | {booking_data.get('infotext', 'N/A')} |\n"
                    f"| 𝙄𝙣𝙘𝙡𝙪𝙙𝙚𝙙   | {booking_data.get('included', 'N/A')} |"
                )
    
                combined_message = f"### Informasjon om booking:\n\n{table}"
                await cl.Message(content=combined_message).send()
    
            except (APIConnectionError, APIResponseError, BookingNotFoundError) as e:
                error_messages = {
                    APIConnectionError: "Kunne ikke koble til bookingsystemet. Prøv igjen senere.",
                    APIResponseError: "Det oppstod en feil ved henting av bookingdata.",
                    BookingNotFoundError: "Ingen informasjon ble funnet med dette bookingnummeret."
                }
                await cl.Message(content=f"❌ {error_messages[type(e)]}\n\nPrøv igjen, kanskje du feilstavet eller glemte ett siffer eller en bokstav?\n\nHvis du ser denne feilmedlingen gjentatte ganger, vennligst kontakt kundeservice@daysoff.no så de kan hjelpe deg.").send()
                return None
    
        else:
            try:
                response = await long_running_task(message.content)
                await cl.Message(content=response["text"]).send()
            except Exception as e:
                await cl.Message(content=f"Error: {str(e)}").send()
    except Exception as e:
                await cl.Message(content=f"Whot?! eh..rr..or: {str(e)}").send()