import json import requests from pydantic import BaseModel, Extra, Field from typing import List, Optional, Union from datetime import datetime, timedelta from langchain.callbacks.manager import CallbackManager from langchain.chains.openai_functions import create_structured_output_runnable from langchain_core.runnables import Runnable from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate from langchain.schema import SystemMessage, AIMessage, HumanMessage, BaseMessage from langchain.prompts.chat import BaseMessagePromptTemplate from app.configs import settings from app.schemas.chat import ChatAdditonal from . import MultiArgsTool from app.utils.logger import CustomLogger logger = CustomLogger.get_logger() class BookRoomToolSchema(BaseModel): from_date: datetime = Field(description="Start date of booking") to_date: datetime = Field(description="End date of booking") adult: str = Field(description="Count of adult") child: str = Field(description="Count of child") room_name: str = Field(description="Room name that user want to Book") def __init__(self,from_date:datetime, *,to_date:datetime = None, adult:int = 0, child:int = 0, room_name:str = '') -> None: self.from_date = from_date self.to_date = to_date self.adult = adult self.child = child self.room_name = room_name def __setattr__(self, name, value) -> None: self.__dict__[name] = value class BookRoomSchema(BaseModel): answer: str = Field(description="Answer to user question with details of rooms ") followUpQuestions:List[str] = Field(description="Generate answers for ur latest answer that you generated now consisting of 2-3 words only. Examples could include 'Book Hotel name.','Thankyou.', 'Need more details?', or 'Need help?'") class Config: extra = Extra.allow # Allow extra properties arbitrary_types_allowed = True # SEARCH_ROOM_PROMPT = """You are a helpful AI assistant trained to do booking of Room "Bay View Hotel". Use the following pieces of context of rooms details to answer the question. If you don't find the answer in the context, politely explain that you don't know and suggest rephrasing the question, don't try to make up an answer. Answer in the language of the user question. # If you found multiple rooms provide user list of available room and ask for which room should to book. otherwise told them room is added cart. please do checkout process to confirm reservation. Don't make answer too much large for user gave you view in concise way (Follow law of least action). # # {context} # # Format your helpful response as "markdown".""" # def create_search_tool_prompt( # extra_prompt_messages: Optional[List[BaseMessage]] = None, # )-> ChatPromptTemplate: # _prompts = extra_prompt_messages or [] # messages: List[Union[BaseMessage, BaseMessagePromptTemplate]] = [ # *_prompts, # SystemMessagePromptTemplate.from_template(SEARCH_ROOM_PROMPT), # HumanMessagePromptTemplate.from_template("{question}") # ] # return ChatPromptTemplate(messages=messages) class BookRoomTool: def __init__(self, question: str, llm: Runnable, history: List[Union[HumanMessage, AIMessage, SystemMessage]] = None) -> None: self.llm = llm self.question = question self.chat_hisory = history self.hotel_url = settings.LS_API_URL self.hotel_current_verion = settings.HOTEL_CURRENT_VERSION self.sub_hotel_id = settings.SUB_HOTEL_ID def __call__(self): return MultiArgsTool( name="book_hotel_room", description="""Use this tool to Book a room or add to cart only if you have the information which room user want to book and don'assume any room_name itself .If room_name is missing then use search_or_check_hotel_rooms_availabilities tool before it for getting room_name . """, # Use this tool to Book any room or add to cart only if you have the information regarding the members count and checkin checkout dates, room name (use search_or_check_hotel_room_availabilities tool for list of room name) ask it first before booking func=self._book_rooms, args_schema=BookRoomToolSchema, return_direct=True ) def _book_rooms(self, callbacks: CallbackManager = None, from_date = None, to_date = None, adult=0, child=0, room_name= "", **kwargs): return BookRoomSchema( answer= room_name +" is added to you cart.", followUpQuestions= ["Need Help?","Book Now."], ) # chain = create_structured_output_runnable( # llm=self.llm, # output_schema=BookRoomSchema, # prompt=create_search_tool_prompt(extra_prompt_messages=self.chat_hisory) # ) # rooms_list = self._get_rooms(from_date, to_date, adult, child, room_filters, **kwargs) # context = "No Rooms Available to Book" # additional = None # if len(rooms_list) == 1: # context = f"ROOM is Add to Cart Details: {rooms_list[0]}" # additional = ChatAdditonal( # actions=["change-url", "scrollTo", "add-to-cart"], # roomInfo=rooms_list[0], # entities={ # "room_stay_from": from_date, # "room_stay_to": to_date, # "adults" : adult, # "child": child, # "promo": "" # }, # url="/checkout", # scrollToElement=".room-listing-block" # ) # elif len(rooms_list) > 1: # context = rooms_list.__str__() # additional = ChatAdditonal( # actions=["change-url", "scrollTo"], # # createDiv= "SHOW-ROOM-LIST", # # roomList=rooms_list, # url="/Suites", # scrollToElement=".room-listing-block" # ) # result: BookRoomSchema = chain.invoke(input={'question': self.question, "context": context}, config={"callbacks":callbacks}) # result.additional = additional # return result # def _get_rooms(self, from_date = None, to_date = None, adult=0, child=0, room_filters=None, **kwargs) -> list: # try: # response = requests.get( # self.hotel_url + "/frontend/Home/get_hotel_rooms", # headers={"current_version": self.hotel_current_verion} # ) # rooms_data = json.loads(response.content.decode('utf-8-sig')) # if response.status_code != 200 or rooms_data['status'] == 'error': # raise Exception("Error Occur in While Room Fetching") # rooms_list = [] # if room_filters: # rooms_list = [item for item in rooms_data['data'] if BookRoomTool.is_has(item, room_filters)] # rooms_list = [ # { # "roomName": data['room_display_name'], # "roomImage": data['thumbnailImage'], # "bundleRoom": data['bundleRoom'], # "occupancy": data['occupancy'], # "bed": data['bed'], # "room_type_code": data['room_type_code'], # "room_type_id": data["room_type_id"] # } # for data in rooms_data['data'] # ][:3] # if not len(rooms_list): # raise Exception("No Room Found While Filtering") # room_ids = [ data['room_type_id'] for data in rooms_list ] # response = requests.post( # self.hotel_url + "/frontend/Searchresults/getRoomsRatesInfo", # json={ # "adults": adult, # "child": child, # "room_stay_from": from_date.strftime("%Y-%m-%d") if isinstance(from_date, datetime) else datetime.now().strftime("%Y-%m-%d"), # "room_stay_to": to_date.strftime("%Y-%m-%d") if isinstance(to_date, datetime) else (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d"), # "called_for_suites": 'true', # "roomIds": room_ids, # "promo": '' # }, # headers={"current_version": self.hotel_current_verion} # ) # if response.status_code != 200 or rooms_data['status'] == 'error': # raise Exception("Error Occur in While Room Rates Fetching") # rooms_rate_data = json.loads(response.content.decode('utf-8-sig')) # room_rate = [data['rate'] for data in rooms_rate_data['results'].values()] # for room, rate in zip(rooms_list, room_rate): # room['rate'] = rate # return rooms_list # except Exception as e: # logger.error(f"Failed to upload file: {str(e)}") # return [] # @staticmethod # def is_has(item, entities): # return any(entity.lower() in item['room_display_name'].lower() for entity in entities)