app.py

# ===========================================
# title: daysoff-assistant-API-v2 | API error handling | no custom starters
# file: app.py
# NOTE: chainlit==0.7.500
# ===========================================

import json
import asyncio
import os
import re
import requests
from dotenv import load_dotenv
import chainlit as cl
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’. Maintain a friendly and professional tone, **reflecting the warmth of a female customer support 
representative archetype.** By default, you answer in **Norwegian**.
============================
Chat History: {chat_history}
Question: {question}
============================
Answer:
"""
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.on_chat_start
def setup_multiple_chains():
    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)

@cl.on_message
async def handle_message(message: cl.Message):
    user_message = message.content
    llm_chain = cl.user_session.get("llm_chain")

    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("No booking data returned")
            
            if "error" in booking_data:
                raise APIResponseError(booking_data["error"])

            # --markdown_table
            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 booking 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 ta kontakt [email protected]").send()
            return None
        except requests.exceptions.RequestException as e:
            await cl.Message(content="En uventet feil oppstod. Vennligst kontakt [email protected]").send()
            return None

    else:
        try:
            response = await llm_chain.ainvoke({
                "question": user_message,
                "chat_history": ""
            }, callbacks=[cl.AsyncLangchainCallbackHandler()])
            await cl.Message(content=response["text"]).send()
        except Exception as e:
            await cl.Message(content=f"Error: {str(e)}").send()