# app.py import re import os import chainlit as cl from typing import List from pathlib import Path from dotenv import load_dotenv from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain.prompts import ChatPromptTemplate from langchain.schema.runnable import Runnable, RunnablePassthrough, RunnableConfig from langchain.schema import StrOutputParser from langchain_community.document_loaders import ( PyMuPDFLoader, ) from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.vectorstores.chroma import Chroma from langchain.indexes import SQLRecordManager, index from langchain.schema import Document from langchain.callbacks.base import BaseCallbackHandler from langchain.document_loaders import UnstructuredWordDocumentLoader, UnstructuredHTMLLoader, CSVLoader from api_data import booking_agent_system # ==================================================================================== # general queries use the retriever and prompt context # booking queries are intercepted and processed separately, bypassing retriever chain # ==================================================================================== # --------------------------------=== environment ===------------------------------- load_dotenv() OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") auth_token = os.environ.get("DAYSOFF_API_TOKEN") # --------------------------------=== globals ===----------------------------------- chunk_size = 1024 chunk_overlap = 50 embeddings_model = OpenAIEmbeddings() PDF_STORAGE_PATH = "./pdfs" DOCS_STORAGE_PATH = "./data" # --------------------------------=== model ===------------------------------------- model = ChatOpenAI(model_name="gpt-4", temperature=0.5, streaming=True) # ----------------------------=== vectorstore setup ===----------------------------- def process_documents(pdf_storage_path: str, docs_storage_path: str): docs = [] # --type: List[Document] text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100) pdf_directory = Path(pdf_storage_path) for pdf_path in pdf_directory.glob("*.pdf"): loader = PyMuPDFLoader(str(pdf_path)) documents = loader.load() docs += text_splitter.split_documents(documents) for doc_path in Path(docs_storage_path).glob("*"): if doc_path.suffix.lower() in [".docx", ".html", ".csv"]: if doc_path.suffix.lower() == ".docx": loader = UnstructuredWordDocumentLoader(str(doc_path)) documents = loader.load() elif doc_path.suffix.lower() == ".html": loader = UnstructuredHTMLLoader(str(doc_path)) documents = loader.load() elif doc_path.suffix.lower() == ".csv": loader = CSVLoader(str(doc_path)) documents = loader.load() processed_documents = [] # --———> post-process/remove empty Info_Url line for doc in documents: lines = doc.page_content.split("\n") new_lines = [] for line in lines: if line.startswith("Info_Url:"): content = line.split(":", 1)[1].strip() if content: # --———> include only if not empty new_lines.append(line) else: new_lines.append(line) doc.page_content = "\n".join(new_lines) processed_documents.append(doc) documents = processed_documents docs += text_splitter.split_documents(documents) doc_search = Chroma.from_documents(docs, embeddings_model) namespace = "chromadb/datasphere" record_manager = SQLRecordManager( namespace, db_url="sqlite:///record_manager_cache.sql" ) record_manager.create_schema() index_result = index( docs, record_manager, doc_search, cleanup="incremental", source_id_key="source", ) print(f"Indexing stats: {index_result}") return doc_search doc_search = process_documents(PDF_STORAGE_PATH, DOCS_STORAGE_PATH) # ----------------------------=== @cl.set_starters ===------------------------------ # 𝘽𝙤𝙤𝙠𝙞𝙣𝙜 𝙞𝙣𝙛𝙤𝙧𝙢𝙖𝙨𝙟𝙤𝙣, 𝘿𝙖𝙮𝙨𝙤𝙛𝙛 @cl.set_starters async def set_starters(): return [ cl.Starter( label="𝙁𝘼𝙌 𝙛𝙤𝙧 𝙖𝙣𝙨𝙖𝙩𝙩𝙚", message="Hva er spørsmål og svar dere ofte får fra ansatte i bedrifter med DaysOff firmahytteordning?", icon="/public/faq-1.svg", ), cl.Starter( label="𝙁𝘼𝙌 𝙛𝙤𝙧 𝙪𝙩𝙡𝙚𝙞𝙚𝙧𝙚", message="Hva er spørsmål og svar dere får fra utleiere?", icon="/public/faq-2.svg", ), cl.Starter( label="𝙋𝙚𝙧𝙨𝙤𝙣𝙫𝙚𝙧𝙣", message="Hvilke spørsmål får dere vanligvis om personvernspolicyen?", icon="/public/terminal.svg", ), cl.Starter( label="𝙱𝚘𝚘𝚔𝚒𝚗𝚐 𝚒𝚗𝚏𝚘𝚛𝚖𝚊𝚜𝚓𝚘𝚗", message="Halla, du! Ryktet sier du kan fiske opp info for et bookingnr.?", icon="/public/booking_id.svg", ), cl.Starter( label="𝘿𝙖𝙮𝙨ᴏꜰꜰ", message="Gi en kort oppsummering av hva daysoff.no dreier seg om", icon="/public/daysoff.svg", ), cl.Starter( label="𝗔𝗜 𝘒𝘶𝘯𝘥𝘦𝘴𝘦𝘳𝘷𝘪𝘤𝘦..", message="Hva er dette og hvem er du?", icon="/public/metric-space.svg", ) ] # ----------------------------=== @cl.on_chat_start ===------------------------------ @cl.on_chat_start async def main(): # ----------------------------=== system-instruct ===------------------------------ template = """ ## Daysoff Kundeservice AI Support You are a customer support assistant for Daysoff. ## Assistant behaviour - languages: Norwegian (default), English, Polish, Latin, Spanish and Korean. - response prefix: consistently adhere to not adding prefix ’Answer:’ or ’Svar:’ to your response - human support: ```do not refer users to kundeservice@daysoff.no arbitrarily. Only give out this contact information if there is a query you absolutely cannot handle yourself or if user insists on talking to human support``` - communication archetype (default): empathetic professional with feminine resonance - style: focus on emotionally resonant storytelling that builds strong connections with users, inspired by industry-leading content creators like Jon Morrow, Seth Godin, and Neil Patel - emojis policy: use when appropriate for better engagement and clarity - assistant name: ‘Agrippina’, inspired by Julia Agrippina (15-59 AD) for her remarkable organizational and administrative abilities. - fun fact: there are 6,227,020,800 possible anagram combinations to evaluate for ’Julia Agrippina’ ## Assistant tasks # Handle queries about booking information: - concisely use the term ’bookingnummer’ - always format booking-related answers using **markdown tables for clarity**. - ```help user with details in their booking information: example 1: User: "Kan jeg sjekke inn tidlig?", if you do not have the bookingnumber already, you should ask user for bookingnr, retrive the booking information and inform about the related check-in time. example 2: User: "Hvor mange gjester er på denne bookingen?", if you do not have the bookingnumber already, you should ask user for bookingnr, retrive the booking information and inform about the related number of guests. (etc.) ``` # Q&A with Daysoff Kundeservice AI Support - Daysoff, general info: brand, firmahytte ordensregler, verticals, link to website:https://www.daysoff.no - Daysoff, social media links: [@daysoffnow] Instagram, [facebook.com/daysoff.no] Facebook, [linkedin.com/company/daysoff] Linkedin, [@DaysOffNow] Twitter/X # Frequently Asked Questions If user query is about FAQs, display FAQ accordingly. > Notes: inform user to copy and paste the question from the currently displayed table they like answered. "FAQ for Ansatte": ```Place the following questions in a markdown table: |# 𝙁𝘼𝙌 𝙛𝙤𝙧 𝙖𝙣𝙨𝙖𝙩𝙩𝙚| |:----------------| |--- |``` Hvordan registrerer jeg meg som bruker?, Når får jeg leieinstruks for min bestilling? Informasjon om nøkler etc.?, Det står barneseng og barnestol under fasiliteter, må dette forhåndsbestilles?, Kan jeg ta med hund eller katt?, Jeg har lagt inn en bestilling hva skjer videre?, Jeg har bestilt firmahytte, men kan ikke reise. Kan jeg endre navn på bestillingen til min kollega eller familiemedlem som vil reise i stedet for meg?", "Kan jeg avbestille min reservasjon?, Jeg har bestilt utvask. Hva må jeg gjøre i tillegg til dette?, Jeg er medlem og eier en hytte! Kan jeg bli utleier i DaysOff?, Bestille opphold? "FAQ for Utleiere": ```Place the following questions in a markdown table: |# 𝙁𝘼𝙌 𝙛𝙤𝙧 𝙪𝙩𝙡𝙚𝙞𝙚𝙧𝙚| |:----------------| |--- |``` Hva er betingelser for utleie?, Hvor lang tid har jeg på å bekrefte en bestilling?, Hvilke kanselleringsregler gjelder?, Hvem er kundene deres?", Kan jeg legge inn rabatterte priser for å lage egne kampanjer?, Når mottar jeg betaling for leie?", Jeg fikk en e-post om ny bestilling, men jeg finner den ikke i systemet?, Hvordan registrerer jeg opptatte perioder i kalenderen?, Jeg leier ut i andre kanaler. Hvordan kan jeg synkronisere kalenderne? "Personvernspolicy FAQ": ```Place the following questions in a markdown table: |# 𝙋𝙚𝙧𝙨𝙤𝙣𝙫𝙚𝙧𝙣𝙨𝙥𝙤𝙡𝙞𝙘𝙮 𝙁𝘼𝙌| |:----------------| |--- |``` Hvilke personlige opplysninger samler dere inn?, Kan dere motta personlig informasjon fra tredjepart?, Hvordan bruker dere mine personlige opplysninger?, Med hvem deler dere mine personlige opplysninger?, Adferdsmessig annonsering?, Hvordan reagerer dere på «Spor ikke» forespørsler?, Hva er mine rettigheter?, Hvordan beskytter dere dataene mine?, Hvilke data brudd prosedyrer har dere på plass?, Hvem i deres team har tilgang til mine data?, Hva er policyendringer?" ## Use the following context to interact with user: ===================== {context} ===================== Question: {question} """ prompt = ChatPromptTemplate.from_template(template) # ------------------------------=== retriever ===----------------------------------- 𝙋𝙚𝙧𝙨𝙤𝙣𝙫𝙚𝙧𝙣𝙨𝙥𝙤𝙡𝙞𝙘𝙮 def format_docs(docs): return "\n\n".join([d.page_content for d in docs]) retriever = doc_search.as_retriever() runnable = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | model | StrOutputParser() ) cl.user_session.set("runnable", runnable) # ----------------------------=== @cl.on_message ===------------------------------ @cl.on_message async def incoming(message: cl.Message): booking_pattern = r'\b[A-Z]{6}\d{6}\b' if re.search(booking_pattern, message.content): booking_msg = cl.Message(content="") await booking_agent_system(message, booking_msg) return # --———> if no booking number/ooking handling, back to here: runnable = cl.user_session.get("runnable") # --type: Runnable msg = cl.Message(content="") class PostMessageHandler(BaseCallbackHandler): def __init__(self, msg: cl.Message): BaseCallbackHandler.__init__(self) self.msg = msg def on_llm_end(self, response, *, run_id, parent_run_id, **kwargs): pass async for chunk in runnable.astream( message.content, config=RunnableConfig(callbacks=[PostMessageHandler(msg)]), ): await msg.stream_token(chunk) await msg.send()